fix(wpt): serve raw form echo fixture

moli-benchmark/moli_benchmark/wpt_cross/server.py

@@ -122,6 +122,7 @@ NAVIGATION_SECOND_VISIT_PATH = (
     "204-205-download-on-second-visit.py"
 )
 BENCH_TIMEOUT_MULTIPLIER_QUERY = "__moli_bench_timeout_multiplier"
+FORM_ECHO_PATH = "/html/semantics/forms/form-submission-0/form-echo.py"
 BENCH_REPORT_BRIDGE_SRC_RE = re.compile(
     rb"(?P<prefix>\bsrc\s*=\s*)(?P<quote>['\"])"
     rb"/resources/testharnessreport\.js(?P=quote)",
@@ -1170,6 +1171,10 @@ def _workers_url_encoding_response(query: str) -> bytes:
     return b"PASS" if value == "å" else b"FAIL"
 
 
+def _form_echo_response(body: bytes) -> bytes:
+    return b" ".join(f"{byte:02x}".encode("ascii") for byte in body)
+
+
 def _redirect_fixture_response(query: str) -> tuple[int, str] | None:
     """Return the shared redirect response used by static WPT fixture handlers."""
 
@@ -1852,6 +1857,15 @@ def _make_handler(
             }:
                 self._serve_fetch_resource_method()
                 return
+            if path == FORM_ECHO_PATH:
+                raw = self._read_content_length_request_body()
+                if raw is not None:
+                    self._send_bytes(
+                        "text/plain",
+                        _form_echo_response(raw),
+                        emit_body=True,
+                    )
+                return
             if path == "/xhr/resources/delay.py":
                 if self._consume_request_body():
                     self._serve_xhr_delay(parsed.query, emit_body=True)
@@ -2200,6 +2214,9 @@ def _make_handler(
             if path == JSON_THEN_JS_PATH:
                 self._serve_json_then_js(parsed.query, emit_body=emit_body)
                 return
+            if path == FORM_ECHO_PATH:
+                self._send_bytes("text/plain", b"", emit_body=emit_body)
+                return
             if path == "/xhr/resources/delay.py":
                 self._serve_xhr_delay(parsed.query, emit_body=emit_body)
                 return
@@ -2566,6 +2583,34 @@ def _make_handler(
                 return self._reject_request_body(413)
             return self._discard_request_body_bytes(length)
 
+        def _read_content_length_request_body(self) -> bytes | None:
+            if self.headers.get("Transfer-Encoding") is not None:
+                self._reject_request_body(400)
+                return None
+            length_str = self.headers.get("Content-Length")
+            if length_str is None:
+                return b""
+            try:
+                length = int(length_str)
+            except ValueError:
+                self._reject_request_body(400)
+                return None
+            if length < 0:
+                self._reject_request_body(400)
+                return None
+            if length > MAX_REQUEST_BODY_BYTES:
+                self._reject_request_body(413)
+                return None
+            try:
+                raw = self.rfile.read(length)
+            except (BrokenPipeError, ConnectionResetError, OSError):
+                self.close_connection = True
+                return None
+            if len(raw) != length:
+                self.close_connection = True
+                return None
+            return raw
+
         def _consume_chunked_request_body(self) -> bool:
             total = 0
             try:

moli-benchmark/tests/test_wpt_cross.py

@@ -4113,6 +4113,44 @@ test(() => {}, "ok");
             ],
         )
 
+    def test_fixture_server_models_form_echo_handler(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:
+                url = (
+                    f"{server.base_url}/html/semantics/forms/"
+                    "form-submission-0/form-echo.py"
+                )
+                responses = []
+                for method, body in (
+                    ("GET", None),
+                    ("HEAD", None),
+                    ("POST", b"\x00\x09\x7f\x80\xff"),
+                ):
+                    request = Request(url, data=body, method=method)
+                    with urlopen(request, timeout=2) as response:
+                        responses.append(
+                            (
+                                method,
+                                response.status,
+                                response.read(),
+                                response.headers["Content-Type"],
+                            )
+                        )
+
+        self.assertEqual(
+            responses,
+            [
+                ("GET", 200, b"", "text/plain"),
+                ("HEAD", 200, b"", "text/plain"),
+                ("POST", 200, b"00 09 7f 80 ff", "text/plain"),
+            ],
+        )
+
     def test_fixture_server_models_fetch_inspect_headers_handler(self) -> None:
         self.assertEqual(
             _inspect_headers_response_headers(