From 9f6e97047cb5a19b73fc45e09377cef08b830c35 Mon Sep 17 00:00:00 2001 From: ldm0 Date: Thu, 10 Sep 2026 16:10:33 +0800 Subject: [PATCH] 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. --- .../moli_benchmark/wpt_cross/server.py | 29 +++++++++++ moli-benchmark/tests/test_wpt_cross.py | 49 +++++++++++++++++++ 2 files changed, 78 insertions(+) diff --git a/moli-benchmark/moli_benchmark/wpt_cross/server.py b/moli-benchmark/moli_benchmark/wpt_cross/server.py index 3250e213b8..7f9756f04d 100644 --- a/moli-benchmark/moli_benchmark/wpt_cross/server.py +++ b/moli-benchmark/moli_benchmark/wpt_cross/server.py @@ -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, diff --git a/moli-benchmark/tests/test_wpt_cross.py b/moli-benchmark/tests/test_wpt_cross.py index f6e0ce8bbd..044c27dada 100644 --- a/moli-benchmark/tests/test_wpt_cross.py +++ b/moli-benchmark/tests/test_wpt_cross.py @@ -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(