fix(wpt): model Fetch status fixture responses

Serve fetch/api/resources/status.py with upstream status, reason, method
header and byte-preserving query content across supported request methods.
Keep duplicate parameter precedence and explicit empty values intact.

Validate fixture parameters and 32 HTTP status/method combinations. The
focused CLI comparison gains 16 UTF-8 subtests with the same engine binary;
null response bodies and BOM handling remain separate engine defects.
This commit is contained in:
ldm0
2026-09-14 16:27:32 +08:00
parent b5dcb333df
commit 9f6e97047c
2 changed files with 78 additions and 0 deletions
@@ -1064,6 +1064,17 @@ def _response_content_type_and_extra_headers(
return merged_content_type, merged_extra_headers
def _fetch_status_response(query: str) -> tuple[int, str, str, bytes]:
"""Model fetch/api/resources/status.py without decoding its byte payload."""
params = parse_qs(query, keep_blank_values=True, encoding="latin-1")
return (
int(params.get("code", ["200"])[0]),
params.get("text", ["OMG"])[0],
params.get("type", [""])[0],
params.get("content", [""])[0].encode("latin-1"),
)
def _wasm_webapi_status_code(query: str) -> int | None:
"""Return the status for WPT's wasm/webapi/status.py fixture.
@@ -2072,6 +2083,9 @@ def _make_handler(
emit_body=emit_body,
)
return
if path == "/fetch/api/resources/status.py":
self._serve_fetch_status(parsed.query, emit_body=emit_body)
return
if path == "/fetch/api/resources/inspect-headers.py":
self._send_bytes(
"text/plain",
@@ -2731,6 +2745,21 @@ def _make_handler(
except (BrokenPipeError, ConnectionResetError):
return
def _serve_fetch_status(self, query: str, *, emit_body: bool) -> None:
try:
status, text, content_type, body = _fetch_status_response(query)
except ValueError:
self.send_error(500)
return
self._send_bytes(
content_type,
body,
emit_body=emit_body,
status_code=status,
status_text=text,
extra_headers=[("X-Request-Method", self.command)],
)
def _send_bytes(
self,
content_type: str,
+49
View File
@@ -126,6 +126,7 @@ from moli_benchmark.wpt_cross.server import (
_host_header_hostname,
_headers_include,
_inspect_headers_response_headers,
_fetch_status_response,
_normalize_harness_case_key,
_nosniff_javascript_response,
_needs_wpt_template_substitution,
@@ -4343,6 +4344,54 @@ test(() => {}, "ok");
response.read(), b"FAIL" if method == "GET" else b""
)
def test_fixture_server_parses_fetch_status_parameters_as_bytes(self) -> None:
self.assertEqual(_fetch_status_response(""), (200, "OMG", "", b""))
self.assertEqual(
_fetch_status_response(
"code=201&code=404&text=&text=ignored&type=text%2Fplain"
"&content=%FF%FE%00%2B+end&content=ignored"
),
(201, "", "text/plain", b"\xff\xfe\x00+ end"),
)
def test_fixture_server_models_fetch_status_for_supported_methods(self) -> None:
with tempfile.TemporaryDirectory() as root:
root_path = Path(root)
(root_path / "resources").mkdir()
(root_path / "resources" / "testharness.js").write_text(
"// testharness", encoding="utf-8"
)
with WptFixtureServer(root_path) as server:
for method in ["GET", "HEAD", "POST", "OPTIONS", "PUT", "PATCH", "DELETE", "YO"]:
for status in [200, 204, 205, 304]:
with self.subTest(method=method, status=status):
connection = HTTPConnection("127.0.0.1", server.port, timeout=2)
try:
connection.request(
method,
"/fetch/api/resources/status.py"
f"?code={status}&text=Custom%20%FF"
"&type=text%2Fplain%3Bcharset%3DUTF-16"
"&content=%FF%FE%00%2B+end",
body=b"ignored" if method not in {"GET", "HEAD"} else None,
)
response = connection.getresponse()
self.assertEqual(response.status, status)
self.assertEqual(response.reason, "Custom \xff")
self.assertEqual(response.headers["X-Request-Method"], method)
self.assertEqual(
response.headers["Content-Type"],
"text/plain;charset=UTF-16",
)
self.assertEqual(response.headers["Content-Length"], "8")
self.assertEqual(
response.read(),
b"" if method == "HEAD" or status in {204, 304}
else b"\xff\xfe\x00+ end",
)
finally:
connection.close()
def test_fixture_server_models_fetch_inspect_headers_handler(self) -> None:
self.assertEqual(
_inspect_headers_response_headers(