235 lines
8.2 KiB
Python
235 lines
8.2 KiB
Python
#!/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()
|