#!/usr/bin/env python3
"""Regressionstests für den Multipart-Parser im Router.

Testet quoted/unquoted Boundaries, Open-WebUI-artige Requests,
beliebige Feldreihenfolge und binäre Payloads.

Aufruf: python3 dev/test_multipart.py
"""
import sys
from email.parser import BytesParser
from email.policy import compat32


def parse_multipart(data: bytes, content_type: str) -> tuple[bytes, str, dict]:
    """Identische Logik wie router/ai_profile_router.py::_parse_multipart."""
    raw = (f"Content-Type: {content_type}\r\n\r\n").encode("utf-8") + data
    msg = BytesParser(policy=compat32).parsebytes(raw)
    if not msg.is_multipart():
        raise ValueError("Kein multipart/form-data")
    file_data = b""
    filename = ""
    fields = {}
    for part in msg.get_payload():
        disposition = part.get("Content-Disposition", "")
        name = None
        part_filename = None
        for kv in disposition.split(";"):
            kv = kv.strip()
            if kv.startswith("name="):
                name = kv[len("name="):].strip('"')
            elif kv.startswith("filename="):
                part_filename = kv[len("filename="):].strip('"')
        if name is None:
            continue
        payload = part.get_payload(decode=True)
        if payload is None:
            payload = b""
        if part_filename is not None:
            file_data = payload
            filename = part_filename or ""
        else:
            fields[name] = payload.decode("utf-8", errors="replace")
    return file_data, filename, fields


def build(body_parts: list[tuple[str, str | bytes, str | None]],
          boundary: str, quoted: bool = False) -> tuple[bytes, str]:
    """Baut einen Multipart-Body. body_parts: (name, value, filename)."""
    b = f'"{boundary}"' if quoted else boundary
    ct = f"multipart/form-data; boundary={b}"
    parts = []
    for name, value, filename in body_parts:
        if isinstance(value, bytes):
            header = (f'Content-Disposition: form-data; name="{name}"')
            if filename:
                header += f'; filename="{filename}"'
            header += f'\r\nContent-Type: application/octet-stream\r\n\r\n'
            parts.append(header.encode() + value + b"\r\n")
        else:
            parts.append(
                f'Content-Disposition: form-data; name="{name}"\r\n\r\n{value}\r\n'
                .encode()
            )
    sep = f"--{boundary}\r\n".encode()
    body = sep + sep.join(parts) + (f"--{boundary}--\r\n").encode()
    return body, ct


def test_quoted_boundary():
    """Quoted boundary (boundary=\"...\")."""
    boundary = "----WebKitFormBoundary7MA4YWxkTrZu0gW"
    webm = b"\x1a\x45\xdf\xa3" + b"\x00\x01\x02\x03\xff\xfe\xfd" * 50
    body, ct = build(
        [("model", "qwen3-asr", None), ("file", webm, "t.webm")],
        boundary, quoted=True,
    )
    fd, fn, fl = parse_multipart(body, ct)
    assert fd == webm, "file_data mismatch"
    assert fn == "t.webm", f"filename mismatch: {fn!r}"
    assert fl["model"] == "qwen3-asr", f"model mismatch: {fl!r}"
    print("  quoted boundary: OK")


def test_unquoted_boundary():
    """Unquoted boundary (boundary=...)."""
    boundary = "AaB03x"
    body, ct = build(
        [("file", b"RIFF", "a.wav"), ("language", "de", None)],
        boundary, quoted=False,
    )
    fd, fn, fl = parse_multipart(body, ct)
    assert fd == b"RIFF", "file_data mismatch"
    assert fn == "a.wav", f"filename mismatch: {fn!r}"
    assert fl["language"] == "de", f"language mismatch: {fl!r}"
    print("  unquoted boundary: OK")


def test_openwebui_style():
    """Open-WebUI-artiger Request (quoted, extra Header, File vor model)."""
    boundary = "----formdata-boundary"
    webm = b"\x1a\x45\xdf\xa3" + b"\xde\xad\xbe\xef" * 30
    ct = f'multipart/form-data; boundary="{boundary}"'
    body = (
        f"--{boundary}\r\n"
        f'Content-Disposition: form-data; name="file"; filename="rec.webm"\r\n'
        f"Content-Type: audio/webm; codecs=opus\r\n"
        f"\r\n"
    ).encode() + webm + (
        f"\r\n--{boundary}\r\n"
        f'Content-Disposition: form-data; name="model"\r\n'
        f"\r\n"
        f"qwen3-asr\r\n"
        f"--{boundary}\r\n"
        f'Content-Disposition: form-data; name="temperature"\r\n'
        f"\r\n"
        f"0.0\r\n"
        f"--{boundary}--\r\n"
    ).encode()
    fd, fn, fl = parse_multipart(body, ct)
    assert fd == webm, "file_data mismatch"
    assert fn == "rec.webm", f"filename mismatch: {fn!r}"
    assert fl["model"] == "qwen3-asr", f"model mismatch: {fl!r}"
    assert fl["temperature"] == "0.0", f"temperature mismatch: {fl!r}"
    print("  Open-WebUI-artig: OK")


def test_file_before_model():
    """File-Feld vor model-Feld."""
    boundary = "boundary123"
    body, ct = build(
        [("file", b"DATA", "f.wav"), ("model", "qwen3-asr", None)],
        boundary, quoted=False,
    )
    fd, fn, fl = parse_multipart(body, ct)
    assert fd == b"DATA", "file_data mismatch"
    assert fn == "f.wav", f"filename mismatch: {fn!r}"
    assert fl["model"] == "qwen3-asr", f"model mismatch: {fl!r}"
    print("  File vor model: OK")


def test_file_after_model():
    """model-Feld vor File-Feld."""
    boundary = "boundary456"
    body, ct = build(
        [("model", "qwen3-asr", None), ("file", b"DATA", "g.wav")],
        boundary, quoted=False,
    )
    fd, fn, fl = parse_multipart(body, ct)
    assert fd == b"DATA", "file_data mismatch"
    assert fn == "g.wav", f"filename mismatch: {fn!r}"
    assert fl["model"] == "qwen3-asr", f"model mismatch: {fl!r}"
    print("  File nach model: OK")


def test_webm_binary_payload():
    """WebM-Binärpayload (alle 256 Byte-Werte)."""
    boundary = "boundary789"
    webm = bytes(range(256)) * 10  # 2560 bytes
    body, ct = build(
        [("file", webm, "bin.webm")],
        boundary, quoted=False,
    )
    fd, fn, fl = parse_multipart(body, ct)
    assert fd == webm, f"file_data mismatch ({len(fd)} vs {len(webm)})"
    assert fn == "bin.webm", f"filename mismatch: {fn!r}"
    print("  WebM-Binärpayload: OK")


def test_extra_headers_ignored():
    """Zusätzliche Multipart-Header werden ignoriert."""
    boundary = "boundaryExtra"
    webm = b"\x1a\x45\xdf\xa3" + b"\x00" * 100
    ct = f'multipart/form-data; boundary="{boundary}"'
    body = (
        f"--{boundary}\r\n"
        f'Content-Disposition: form-data; name="file"; filename="x.webm"\r\n'
        f"Content-Type: audio/webm\r\n"
        f"X-Custom-Header: ignored\r\n"
        f"Content-Transfer-Encoding: binary\r\n"
        f"\r\n"
    ).encode() + webm + (
        f"\r\n--{boundary}\r\n"
        f'Content-Disposition: form-data; name="model"\r\n'
        f"\r\n"
        f"qwen3-asr\r\n"
        f"--{boundary}--\r\n"
    ).encode()
    fd, fn, fl = parse_multipart(body, ct)
    assert fd == webm, "file_data mismatch"
    assert fn == "x.webm", f"filename mismatch: {fn!r}"
    assert fl["model"] == "qwen3-asr", f"model mismatch: {fl!r}"
    print("  Extra-Header ignoriert: OK")


def test_all_fields():
    """Alle unterstützten Felder (model, language, prompt, response_format, temperature)."""
    boundary = "boundaryAll"
    body, ct = build(
        [
            ("file", b"AUDIO", "a.webm"),
            ("model", "qwen3-asr", None),
            ("language", "de", None),
            ("prompt", "Kontext", None),
            ("response_format", "verbose_json", None),
            ("temperature", "0.5", None),
        ],
        boundary, quoted=True,
    )
    fd, fn, fl = parse_multipart(body, ct)
    assert fd == b"AUDIO", "file_data mismatch"
    assert fn == "a.webm", f"filename mismatch: {fn!r}"
    assert fl["model"] == "qwen3-asr", f"model mismatch: {fl!r}"
    assert fl["language"] == "de", f"language mismatch: {fl!r}"
    assert fl["prompt"] == "Kontext", f"prompt mismatch: {fl!r}"
    assert fl["response_format"] == "verbose_json", f"response_format mismatch: {fl!r}"
    assert fl["temperature"] == "0.5", f"temperature mismatch: {fl!r}"
    print("  Alle Felder: OK")


def main():
    print("Multipart-Parser Regressionstests:")
    test_quoted_boundary()
    test_unquoted_boundary()
    test_openwebui_style()
    test_file_before_model()
    test_file_after_model()
    test_webm_binary_payload()
    test_extra_headers_ignored()
    test_all_fields()
    print("\nAlle Multipart-Parser-Tests bestanden!")


if __name__ == "__main__":
    main()