test(wpt): support programmed history traversal responses

moli-benchmark/moli_benchmark/wpt_cross/case_set.py

@@ -758,6 +758,22 @@ def _fetch_resource_handler_reference_patterns(directory: str) -> tuple[re.Patte
     return tuple(patterns)
 
 
+@lru_cache(maxsize=None)
+def _navigation_handler_reference_patterns(directory: str) -> tuple[re.Pattern[str], ...]:
+    resource = (
+        "navigation-api/navigation-methods/return-value/resources/"
+        "204-205-download-on-second-visit.py"
+    )
+    relative = posixpath.relpath(resource, directory)
+    return tuple(
+        re.compile(
+            rf"(?<![A-Za-z0-9_./-]){re.escape(reference)}"
+            rf"{WPTSERVE_HANDLER_TRAILING_BOUNDARY}"
+        )
+        for reference in ("/" + resource, relative, "./" + relative)
+    )
+
+
 @lru_cache(maxsize=None)
 def _json_module_handler_reference_patterns(directory: str) -> tuple[re.Pattern[str], ...]:
     references = []
@@ -882,6 +898,7 @@ def _supported_wptserve_handler_references(
     if rel is not None and rel.startswith("fetch/api/"):
         supported += _empty_location_handler_reference_patterns(posixpath.dirname(rel))
     if rel is not None:
+        supported += _navigation_handler_reference_patterns(posixpath.dirname(rel) or ".")
         supported += _json_module_handler_reference_patterns(posixpath.dirname(rel) or ".")
     if rel is not None:
         supported += _script_handler_reference_patterns(posixpath.dirname(rel) or ".")

moli-benchmark/moli_benchmark/wpt_cross/server.py

@@ -163,6 +163,10 @@ JSON_THEN_JS_PATH = "/html/semantics/scripting-1/the-script-element/serve-json-t
 JSON_LOAD_ERROR_PATH = (
     "/html/semantics/scripting-1/the-script-element/json-module/load-error-events.py"
 )
+NAVIGATION_SECOND_VISIT_PATH = (
+    "/navigation-api/navigation-methods/return-value/resources/"
+    "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"
 FORM_SUBMISSION_PATH = (
@@ -1620,6 +1624,8 @@ def _make_handler(
                 path = unquote(urlsplit(getattr(self, "path", "")).path)
                 if path == FETCH_EMPTY_LOCATION_PATH:
                     return self._serve_empty_location_resource
+                if path == NAVIGATION_SECOND_VISIT_PATH:
+                    return self._serve_navigation_second_visit
                 if path in SERVICE_WORKER_SCRIPT_RESOURCE_PATHS:
                     return self._serve_service_worker_script_resource
                 if path in XHR_RESOURCE_PATHS:
@@ -1649,6 +1655,9 @@ def _make_handler(
                 return
             parsed = urlparse(self.path)
             path = unquote(parsed.path)
+            if path == NAVIGATION_SECOND_VISIT_PATH:
+                self._serve_navigation_second_visit()
+                return
             if path in SERVICE_WORKER_SCRIPT_RESOURCE_PATHS:
                 self._serve_service_worker_script_resource()
                 return
@@ -1682,6 +1691,9 @@ def _make_handler(
                 return
             parsed = urlparse(self.path)
             path = unquote(parsed.path)
+            if path == NAVIGATION_SECOND_VISIT_PATH:
+                self._serve_navigation_second_visit()
+                return
             if path in SERVICE_WORKER_SCRIPT_RESOURCE_PATHS:
                 self._serve_service_worker_script_resource()
                 return
@@ -1754,6 +1766,9 @@ def _make_handler(
                 return
             parsed = urlparse(self.path)
             path = unquote(parsed.path)
+            if path == NAVIGATION_SECOND_VISIT_PATH:
+                self._serve_navigation_second_visit()
+                return
             if path in FETCH_RANGE_RESOURCE_PATHS:
                 self._serve_fetch_range_resource(path, parsed.query, emit_body=self.command != "HEAD")
                 return
@@ -1797,6 +1812,9 @@ def _make_handler(
             if self._serve_xhr_resource(emit_body=True):
                 return
             parsed = urlparse(self.path)
+            if unquote(parsed.path) == NAVIGATION_SECOND_VISIT_PATH:
+                self._serve_navigation_second_visit()
+                return
             if unquote(parsed.path) in SERVICE_WORKER_SCRIPT_RESOURCE_PATHS:
                 self._serve_service_worker_script_resource()
                 return
@@ -1938,6 +1956,9 @@ def _make_handler(
                 return
             parsed = urlparse(self.path)
             path = unquote(parsed.path)
+            if path == NAVIGATION_SECOND_VISIT_PATH:
+                self._serve_navigation_second_visit()
+                return
             if path in FETCH_RANGE_RESOURCE_PATHS:
                 self._serve_fetch_range_resource(path, parsed.query, emit_body=emit_body)
                 return
@@ -3124,6 +3145,37 @@ def _make_handler(
             except OSError:
                 pass  # Upstream ends its stream when a write reports disconnect.
 
+        def _serve_navigation_second_visit(self) -> None:
+            if not self._consume_request_body():
+                return
+            params = parse_qs(urlsplit(self.path).query, keep_blank_values=True, encoding="latin-1")
+            stash_path = NAVIGATION_SECOND_VISIT_PATH.rsplit("/", 1)[0] + "/"
+            status, content_type, body = 400, None, b""
+            headers: list[tuple[str, str]] = []
+            cache_control = None
+            try:
+                key = params["id"][0]
+                if self.command == "POST":
+                    fetch_stash.put(key, params["action"][0], path=stash_path)
+                    status = 204
+                elif self.command == "GET":
+                    action = fetch_stash.take(key, path=stash_path)
+                    if action is None:
+                        status, content_type, body = 200, "text/html", b"initial page"
+                        cache_control = "no-store"
+                    elif action in ("204", "205"):
+                        status = int(action)
+                    elif action == "download":
+                        status, content_type, body = 200, "text/plain", b"some text to download"
+                        headers.append(("Content-Disposition", "attachment"))
+            except (KeyError, ValueError):
+                self.send_error(500)
+                return
+            self._send_bytes(
+                content_type, body, emit_body=self.command != "HEAD",
+                status_code=status, extra_headers=headers, cache_control=cache_control,
+            )
+
         def _serve_fetch_abort_resource(
             self, path: str, query: str, *, emit_body: bool
         ) -> None:

moli-benchmark/tests/test_wpt_navigation_response_server.py

@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+import uuid
+from contextlib import ExitStack
+from http.client import HTTPConnection
+from pathlib import Path
+from unittest.mock import patch
+
+from moli_benchmark.wpt_cross.case_set import enumerate_cases
+from moli_benchmark.wpt_cross.server import NAVIGATION_SECOND_VISIT_PATH, WptFixtureServer
+
+
+CASE_DIRECTORY = "navigation-api/navigation-methods/return-value"
+
+
+class NavigationResponseFixtureTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.stack = ExitStack()
+        self.addCleanup(self.stack.close)
+        self.root = Path(self.stack.enter_context(tempfile.TemporaryDirectory()))
+        (self.root / "resources").mkdir()
+        (self.root / "resources/testharness.js").write_text("// testharness")
+        resource = self.root / NAVIGATION_SECOND_VISIT_PATH.lstrip("/")
+        resource.parent.mkdir(parents=True)
+        resource.write_text("# Python source must not be served")
+        self.stack.enter_context(patch(
+            "moli_benchmark.wpt_cross.server._global_ipv6_address", return_value=None,
+        ))
+        self.server = self.stack.enter_context(WptFixtureServer(self.root))
+
+    def request(self, query: str, *, method: str = "GET", port: int | None = None,
+                path: str = NAVIGATION_SECOND_VISIT_PATH):
+        connection = HTTPConnection("127.0.0.1", port or self.server.port, timeout=2)
+        try:
+            connection.request(method, path + "?" + query,
+                               body=b"ignored request body" if method == "POST" else None)
+            response = connection.getresponse()
+            return response.status, response.headers, response.read()
+        finally:
+            connection.close()
+
+    def assert_initial_page(self, query: str, *, port: int | None = None) -> None:
+        status, headers, body = self.request(query, port=port)
+        self.assertEqual((status, headers["Content-Type"], body),
+                         (200, "text/html", b"initial page"))
+        self.assertEqual(headers["Cache-Control"], "no-store")
+        self.assertIsNone(headers["Content-Disposition"])
+
+    def test_post_programs_one_get_response(self) -> None:
+        for action in ("204", "205", "download"):
+            with self.subTest(action=action):
+                key = uuid.uuid4()
+                query = f"id={key}"
+                self.assert_initial_page(query)
+                status, headers, body = self.request(query + f"&action={action}", method="POST")
+                self.assertEqual((status, body), (204, b""))
+                self.assertIsNone(headers["Content-Type"])
+                self.assertIsNone(headers["Cache-Control"])
+                status, headers, body = self.request(query)
+                if action == "download":
+                    self.assertEqual((status, headers["Content-Type"], body),
+                                     (200, "text/plain", b"some text to download"))
+                    self.assertEqual(headers["Content-Disposition"], "attachment")
+                else:
+                    self.assertEqual((status, body), (int(action), b""))
+                    self.assertIsNone(headers["Content-Type"])
+                self.assertIsNone(headers["Cache-Control"])
+                self.assert_initial_page(query)
+
+    def test_state_is_shared_across_origins_but_isolated_by_key_server_and_path(self) -> None:
+        key = uuid.uuid4()
+        query = f"id={key}"
+        self.assertEqual(self.request(query + "&action=205", method="POST")[0], 204)
+        self.assert_initial_page(f"id={uuid.uuid4()}")
+        other = self.stack.enter_context(WptFixtureServer(self.root))
+        self.assert_initial_page(query, port=other.port)
+        status, _, body = self.request(f"key={key}", path="/fetch/api/resources/stash-take.py")
+        self.assertEqual((status, body), (200, b"null"))
+        status, _, _ = self.request(f"id={str(key).upper()}", port=self.server.alternate_port)
+        self.assertEqual(status, 205)
+        self.assert_initial_page(f"id={key.hex}")
+
+    def test_unsupported_methods_do_not_consume_the_programmed_response(self) -> None:
+        query = f"id={uuid.uuid4()}"
+        self.assertEqual(self.request(query + "&action=204", method="POST")[0], 204)
+        for method in ("HEAD", "OPTIONS", "PUT", "PATCH", "DELETE", "YO", "CUSTOM"):
+            with self.subTest(method=method):
+                status, _, body = self.request(query, method=method)
+                self.assertEqual((status, body), (400, b""))
+        self.assertEqual(self.request(query)[0], 204)
+        self.assert_initial_page(query)
+
+    def test_unknown_actions_are_consumed_and_duplicate_posts_do_not_overwrite(self) -> None:
+        query = f"id={uuid.uuid4()}"
+        for action in ("unknown", ""):
+            self.assertEqual(self.request(query + f"&action={action}", method="POST")[0], 204)
+            self.assertEqual(self.request(query)[0], 400)
+            self.assert_initial_page(query)
+        self.assertEqual(self.request(query + "&action=205&action=204", method="POST")[0], 204)
+        self.assertEqual(self.request(query + "&action=204", method="POST")[0], 500)
+        self.assertEqual(self.request(query + "&id=ignored")[0], 205)
+
+    def test_invalid_parameters_report_errors(self) -> None:
+        for query, method in (("", "GET"), ("id=", "GET"), ("id=not-a-uuid", "GET"),
+                              (f"id={uuid.uuid4()}", "POST"), ("id=%FF&action=204", "POST")):
+            with self.subTest(query=query, method=method):
+                self.assertEqual(self.request(query, method=method)[0], 500)
+
+    def test_discovery_accepts_only_supported_resource_locations(self) -> None:
+        resource = "resources/204-205-download-on-second-visit.py"
+        cases = {
+            "relative.html": (resource, True),
+            "dot.html": ("./" + resource, True),
+            "absolute.html": (NAVIGATION_SECOND_VISIT_PATH, True),
+            "wrong-directory.html": ("../" + resource, False),
+            "basename.html": (resource.rsplit("/", 1)[1], False),
+            "suffix.html": (resource + ".extra", False),
+        }
+        for name, (reference, _) in cases.items():
+            (self.root / CASE_DIRECTORY / name).write_text(
+                '<script src="/resources/testharness.js"></script>'
+                '<script src="/resources/testharnessreport.js"></script>'
+                f'<script>fetch(`{reference}?id=${{id}}`);</script>'
+            )
+        discovered = enumerate_cases(self.root, dir_prefixes=(CASE_DIRECTORY,))
+        self.assertEqual(
+            sorted(case.case_path for case in discovered),
+            sorted(CASE_DIRECTORY + "/" + name for name, (_, allowed) in cases.items() if allowed),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()