fix(wpt-cross): model inspect-headers fixture

moli-benchmark/moli_benchmark/wpt_cross/server.py

@@ -1191,6 +1191,61 @@ def _workers_modules_export_on_load_script_response() -> tuple[bytes, list[tuple
     )
 
 
+def _inspect_headers_response_headers(
+    query: str,
+    request_headers: list[tuple[str, str]],
+) -> list[tuple[str, str]]:
+    """Model fetch/api/resources/inspect-headers.py's response headers."""
+
+    params = parse_qsl(query, keep_blank_values=True)
+    checked_headers_value = next(
+        (value for name, value in params if name == "headers"),
+        None,
+    )
+    checked_headers = (
+        checked_headers_value.split("|") if checked_headers_value is not None else []
+    )
+    request_headers_by_name: dict[str, str] = {}
+    for name, value in request_headers:
+        request_headers_by_name.setdefault(name.lower(), value)
+
+    response_headers = []
+    for name in checked_headers:
+        value = request_headers_by_name.get(name.lower())
+        if value is not None:
+            response_headers.append((f"x-request-{name}", value))
+
+    if any(name == "cors" for name, _ in params):
+        response_headers.extend(
+            [
+                (
+                    "Access-Control-Allow-Origin",
+                    request_headers_by_name.get("origin", "*"),
+                ),
+                ("Access-Control-Allow-Credentials", "true"),
+                ("Access-Control-Allow-Methods", "GET, POST, HEAD"),
+                (
+                    "Access-Control-Expose-Headers",
+                    ", ".join(f"x-request-{name}" for name in checked_headers),
+                ),
+            ]
+        )
+        allow_headers = next(
+            (value for name, value in params if name == "allow_headers"),
+            None,
+        )
+        response_headers.append(
+            (
+                "Access-Control-Allow-Headers",
+                allow_headers
+                if allow_headers is not None
+                else ", ".join(name for name, _ in request_headers),
+            )
+        )
+
+    return response_headers
+
+
 def _url_host_literal(hostname: str) -> str:
     if hostname.startswith("[") and hostname.endswith("]"):
         return hostname
@@ -2022,6 +2077,18 @@ def _make_handler(
                     emit_body=emit_body,
                 )
                 return
+            if path == "/fetch/api/resources/inspect-headers.py":
+                self._send_bytes(
+                    "text/plain",
+                    b"",
+                    emit_body=emit_body,
+                    extra_headers=_inspect_headers_response_headers(
+                        parsed.query,
+                        list(self.headers.items()),
+                    ),
+                    status_code=pipe_status_code or 200,
+                )
+                return
             if path == (
                 "/html/semantics/scripting-1/the-script-element/module/"
                 "resources/delayed-modulescript.py"

moli-benchmark/tests/test_wpt_cross.py

@@ -125,6 +125,7 @@ from moli_benchmark.wpt_cross.server import (
     _host_header_hostname,
     _headers_include,
     _normalize_harness_case_key,
+    _inspect_headers_response_headers,
     _needs_wpt_template_substitution,
     _legacy_wpt_resource_alias,
     _pipe_response_header_operations,
@@ -4111,6 +4112,68 @@ test(() => {}, "ok");
             ],
         )
 
+    def test_fixture_server_models_fetch_inspect_headers_handler(self) -> None:
+        self.assertEqual(
+            _inspect_headers_response_headers(
+                "headers=referer%7Corigin%7Cmissing&cors&allow_headers=x-test",
+                [
+                    ("Referer", "http://source.test/path"),
+                    ("Origin", "http://origin.test"),
+                    ("X-Test", "value"),
+                ],
+            ),
+            [
+                ("x-request-referer", "http://source.test/path"),
+                ("x-request-origin", "http://origin.test"),
+                ("Access-Control-Allow-Origin", "http://origin.test"),
+                ("Access-Control-Allow-Credentials", "true"),
+                ("Access-Control-Allow-Methods", "GET, POST, HEAD"),
+                (
+                    "Access-Control-Expose-Headers",
+                    "x-request-referer, x-request-origin, x-request-missing",
+                ),
+                ("Access-Control-Allow-Headers", "x-test"),
+            ],
+        )
+
+        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}/fetch/api/resources/"
+                    "inspect-headers.py?headers=referer%7Corigin&cors"
+                )
+                request = Request(
+                    url,
+                    headers={
+                        "Referer": "http://source.test/path",
+                        "Origin": "http://origin.test",
+                    },
+                )
+                with urlopen(request, timeout=2) as response:
+                    self.assertEqual(response.status, 200)
+                    self.assertEqual(response.read(), b"")
+                    self.assertEqual(
+                        response.headers["x-request-referer"],
+                        "http://source.test/path",
+                    )
+                    self.assertEqual(
+                        response.headers["x-request-origin"],
+                        "http://origin.test",
+                    )
+                    self.assertEqual(
+                        response.headers["Access-Control-Allow-Origin"],
+                        "http://origin.test",
+                    )
+                    self.assertEqual(
+                        response.headers["Access-Control-Expose-Headers"],
+                        "x-request-referer, x-request-origin",
+                    )
+
     def test_fixture_server_models_xhr_delay_py_methods(self) -> None:
         with tempfile.TemporaryDirectory() as root:
             root_path = Path(root)