fix(wpt): serve XHR PUT and wildcard preflight fixtures

Model the XHR PUT CORS grant and wildcard request-header preflight resources. Preserve method-specific grants, raw PUT bytes, missing-Origin failures and responses without Content-Type; cover exact paths and discovery.

moli-benchmark/moli_benchmark/wpt_cross/case_set.py

@@ -853,7 +853,9 @@ def _xhr_response_handler_reference_patterns(directory: str) -> tuple[re.Pattern
                  "inspect-headers.py", "echo-headers.py", "content.py",
                  "echo-content-type.py", "win-1252-xml.py", "win-1252-html.py",
                  "invalid-utf8-html.py", "shift-jis-html.py", "img-utf8-html.py",
-                 "empty-div-utf8-html.py"):
+                 "empty-div-utf8-html.py",
+                 "access-control-basic-put-allow.py",
+                 "access-control-preflight-request-allow-headers-returns-star.py"):
         resource = f"xhr/resources/{name}"
         relative = posixpath.relpath(resource, directory)
         references.extend(("/" + resource, relative, "./" + relative))

moli-benchmark/moli_benchmark/wpt_cross/server.py

@@ -94,6 +94,8 @@ XHR_RESPONSE_RESOURCE_PATHS = {
     *XHR_BODY_RESOURCE_PATHS,
     "/xhr/resources/inspect-headers.py",
     "/xhr/resources/echo-headers.py",
+    "/xhr/resources/access-control-basic-put-allow.py",
+    "/xhr/resources/access-control-preflight-request-allow-headers-returns-star.py",
     "/xhr/resources/corsenabled.py",
     "/xhr/resources/status.py",
     "/xhr/resources/last-modified.py",
@@ -1844,6 +1846,8 @@ def _make_handler(
                 return
             if unquote(urlparse(self.path).path) in {
                 "/xhr/resources/inspect-headers.py", "/xhr/resources/echo-headers.py",
+                "/xhr/resources/access-control-basic-put-allow.py",
+                "/xhr/resources/access-control-preflight-request-allow-headers-returns-star.py",
             } | XHR_BODY_RESOURCE_PATHS:
                 self._serve_xhr_response_resource()
                 return
@@ -2727,6 +2731,49 @@ def _make_handler(
                 ),
             )
 
+        def _serve_xhr_preflight_fixture(self, parsed, *, emit_body: bool) -> None:
+            status, headers, body = 200, [], b""
+            try:
+                if unquote(parsed.path) == "/xhr/resources/access-control-basic-put-allow.py":
+                    headers = [("Content-Type", "text/plain")]
+                    if self.command in {"OPTIONS", "PUT"}:
+                        origin = self.headers.get("Origin")
+                        if origin is None:
+                            raise ValueError("upstream handler requires Origin")
+                        headers.extend([
+                            ("Access-Control-Allow-Credentials", "true"),
+                            ("Access-Control-Allow-Origin", origin),
+                        ])
+                        if self.command == "OPTIONS":
+                            headers.append(("Access-Control-Allow-Methods", "PUT"))
+                        else:
+                            request_body = self._read_content_length_request_body()
+                            if request_body is None:
+                                return
+                            body = b"PASS: Cross-domain access allowed.\n" + request_body
+                    else:
+                        body = b"Wrong method: " + self.command.encode("latin-1")
+                elif self.command == "OPTIONS":
+                    headers = [
+                        ("Access-Control-Allow-Origin", "*"),
+                        ("Access-Control-Allow-Headers", "*"),
+                    ]
+                elif self.command == "GET":
+                    headers = [("Access-Control-Allow-Origin", "*")]
+                    if self.headers.get("X-Test"):
+                        headers.append(("Content-Type", "text/plain"))
+                        body = b"PASS"
+                    else:
+                        status = 400
+            except ValueError:
+                self.send_error(500)
+                return
+            self.close_connection = True
+            headers.append(("Connection", "close"))
+            self._send_bytes(None, body, emit_body=emit_body, extra_headers=headers,
+                             status_code=status, cache_control=None)
+
+
         def _serve_xhr_response_resource(self, *, emit_body: bool = True) -> bool:
             parsed = urlparse(self.path)
             path = unquote(parsed.path)
@@ -2735,6 +2782,12 @@ def _make_handler(
             if path in XHR_BODY_RESOURCE_PATHS:
                 self._serve_xhr_body_resource(parsed, emit_body=emit_body)
                 return True
+            if path in {
+                "/xhr/resources/access-control-basic-put-allow.py",
+                "/xhr/resources/access-control-preflight-request-allow-headers-returns-star.py",
+            }:
+                self._serve_xhr_preflight_fixture(parsed, emit_body=emit_body)
+                return True
             if path == "/xhr/resources/corsenabled.py":
                 self._serve_xhr_cors_echo(parsed, emit_body=emit_body)
                 return True

moli-benchmark/tests/test_wpt_xhr_cors_fixture.py

@@ -114,3 +114,126 @@ class XhrCorsFixtureTests(unittest.TestCase):
         self.assertEqual([case.case_path for case in selected], [
             "xhr/absolute.html", "xhr/nested/relative.html", "xhr/relative.html",
         ])
+
+    def test_put_preflight_and_upload_preserve_origin_and_bytes(self) -> None:
+        server = self.stack.enter_context(WptFixtureServer(self.root))
+        for origin in ("https://caller.test", "null", ""):
+            with self.subTest(origin=origin):
+                status, headers, body = self.request(
+                    server, "OPTIONS", headers={"Origin": origin}, resource=PUT_RESOURCE
+                )
+                self.assertEqual((status, body), (200, b""))
+                self.assertEqual(headers["Content-Type"], "text/plain")
+                self.assertEqual(headers["Access-Control-Allow-Origin"], origin)
+                self.assertEqual(headers["Access-Control-Allow-Credentials"], "true")
+                self.assertEqual(headers["Access-Control-Allow-Methods"], "PUT")
+                self.assertIsNone(headers["Access-Control-Allow-Headers"])
+                payload = b"PUT \x00\xff body"
+                status, headers, body = self.request(
+                    server, "PUT", body=payload, headers={"Origin": origin},
+                    resource=PUT_RESOURCE,
+                )
+                self.assertEqual((status, body),
+                                 (200, b"PASS: Cross-domain access allowed.\n" + payload))
+                self.assertEqual(headers["Content-Type"], "text/plain")
+                self.assertEqual(headers["Access-Control-Allow-Origin"], origin)
+                self.assertEqual(headers["Access-Control-Allow-Credentials"], "true")
+                self.assertIsNone(headers["Access-Control-Allow-Methods"])
+
+
+    def test_put_fixture_missing_origin_matches_upstream_failure(self) -> None:
+        server = self.stack.enter_context(WptFixtureServer(self.root))
+        for method in ("OPTIONS", "PUT"):
+            with self.subTest(method=method):
+                status, headers, _ = self.request(server, method, resource=PUT_RESOURCE)
+                self.assertEqual(status, 500)
+                self.assertIsNone(headers["Access-Control-Allow-Origin"])
+
+
+    def test_put_fixture_reports_other_methods_without_cors_headers(self) -> None:
+        server = self.stack.enter_context(WptFixtureServer(self.root))
+        for method in ("GET", "HEAD", "POST", "DELETE", "FOO"):
+            with self.subTest(method=method):
+                status, headers, body = self.request(server, method, resource=PUT_RESOURCE)
+                expected = b"Wrong method: " + method.encode("ascii")
+                self.assertEqual((status, body),
+                                 (200, b"" if method == "HEAD" else expected))
+                self.assertEqual(headers["Content-Length"], str(len(expected)))
+                self.assertIsNone(headers["Access-Control-Allow-Origin"])
+
+
+    def test_wildcard_preflight_and_actual_request_have_distinct_headers(self) -> None:
+        server = self.stack.enter_context(WptFixtureServer(self.root))
+        status, headers, body = self.request(server, "OPTIONS", resource=STAR_RESOURCE)
+        self.assertEqual((status, body), (200, b""))
+        self.assertEqual(headers["Access-Control-Allow-Origin"], "*")
+        self.assertEqual(headers["Access-Control-Allow-Headers"], "*")
+        self.assertIsNone(headers["Access-Control-Allow-Methods"])
+        self.assertIsNone(headers["Access-Control-Allow-Credentials"])
+        self.assertIsNone(headers["Content-Type"])
+        for value in ("foobar", "0"):
+            with self.subTest(value=value):
+                status, headers, body = self.request(
+                    server, headers={"x-test": value}, resource=STAR_RESOURCE
+                )
+                self.assertEqual((status, body), (200, b"PASS"))
+                self.assertEqual(headers["Access-Control-Allow-Origin"], "*")
+                self.assertEqual(headers["Content-Type"], "text/plain")
+                self.assertIsNone(headers["Access-Control-Allow-Headers"])
+
+
+    def test_wildcard_fixture_rejects_missing_or_empty_test_header(self) -> None:
+        server = self.stack.enter_context(WptFixtureServer(self.root))
+        for request_headers in ({}, {"X-Test": ""}):
+            with self.subTest(headers=request_headers):
+                status, headers, body = self.request(
+                    server, headers=request_headers, resource=STAR_RESOURCE
+                )
+                self.assertEqual((status, body), (400, b""))
+                self.assertEqual(headers["Access-Control-Allow-Origin"], "*")
+                self.assertIsNone(headers["Content-Type"])
+
+
+    def test_wildcard_fixture_other_methods_have_no_cors_grants(self) -> None:
+        server = self.stack.enter_context(WptFixtureServer(self.root))
+        for method in ("HEAD", "POST", "PUT", "DELETE", "FOO"):
+            with self.subTest(method=method):
+                status, headers, body = self.request(
+                    server, method, headers={"X-Test": "foobar"}, resource=STAR_RESOURCE
+                )
+                self.assertEqual((status, body), (200, b""))
+                self.assertIsNone(headers["Access-Control-Allow-Origin"])
+                self.assertIsNone(headers["Content-Type"])
+
+
+    def test_preflight_fixtures_require_exact_resource_paths(self) -> None:
+        server = self.stack.enter_context(WptFixtureServer(self.root))
+        for resource in (PUT_RESOURCE, STAR_RESOURCE):
+            for wrong in (resource + "2", resource + ".js", "/wrong" + resource):
+                with self.subTest(path=wrong):
+                    status, _, _ = self.request(server, resource=wrong)
+                    self.assertEqual(status, 404)
+
+
+    def test_preflight_case_selection_rejects_unsupported_dependencies(self) -> None:
+        for resource in (PUT_RESOURCE, STAR_RESOURCE):
+            name = Path(resource).name
+            sources = {
+                f"xhr/absolute-{name}.html": f"fetch('{resource}');",
+                f"xhr/relative-{name}.html": f"fetch('resources/{name}');",
+                f"xhr/nested/relative-{name}.html": f"fetch('../resources/{name}?x=1');",
+                f"xhr/unknown-{name}.html": f"fetch('{resource}'); fetch('unknown.py');",
+                f"xhr/wrong-{name}.html": f"fetch('{name}');",
+            }
+            for path, script in sources.items():
+                target = self.root / path
+                target.parent.mkdir(parents=True, exist_ok=True)
+                target.write_text('<script src="/resources/testharness.js"></script><script>'
+                                  + script + '</script>')
+        selected = enumerate_cases(self.root, dir_prefixes=("xhr",))
+        expected = sorted(
+            f"xhr/{prefix}{Path(resource).name}.html"
+            for resource in (PUT_RESOURCE, STAR_RESOURCE)
+            for prefix in ("absolute-", "relative-", "nested/relative-")
+        )
+        self.assertEqual([case.case_path for case in selected], expected)