diff --git a/moli-benchmark/moli_benchmark/wpt_cross/case_set.py b/moli-benchmark/moli_benchmark/wpt_cross/case_set.py index 6214837c6b..f2f192dc8a 100644 --- a/moli-benchmark/moli_benchmark/wpt_cross/case_set.py +++ b/moli-benchmark/moli_benchmark/wpt_cross/case_set.py @@ -815,10 +815,36 @@ def _empty_location_handler_reference_patterns(directory: str) -> tuple[re.Patte return tuple(patterns) +@lru_cache(maxsize=None) +def _service_worker_script_handler_reference_patterns(directory: str) -> tuple[re.Pattern[str], ...]: + references = [] + resources = tuple( + "service-workers/service-worker/resources/" + name + for name in ( + "redirect.py", "update-worker.py", "update-worker-from-file.py", + "update-during-installation-worker.py", + "import-scripts-version.py", "import-scripts-get.py", "import-scripts-echo.py", + "subdir/import-scripts-echo.py", "scope2/import-scripts-echo.py", + ) + ) + ("service-workers/service-worker/ServiceWorkerGlobalScope/resources/update-worker.py",) + for resource in resources: + relative = posixpath.relpath(resource, directory) + references.extend(("/" + resource, relative, "./" + relative)) + return tuple( + re.compile( + rf"(? tuple[re.Pattern[str], ...]: supported: tuple[re.Pattern[str], ...] = () + if rel is not None: + supported += _service_worker_script_handler_reference_patterns(posixpath.dirname(rel) or ".") if rel is not None and rel.startswith("fetch/api/"): supported += _empty_location_handler_reference_patterns(posixpath.dirname(rel)) if rel is not None: diff --git a/moli-benchmark/moli_benchmark/wpt_cross/server.py b/moli-benchmark/moli_benchmark/wpt_cross/server.py index 5b48ccecbf..403e178588 100644 --- a/moli-benchmark/moli_benchmark/wpt_cross/server.py +++ b/moli-benchmark/moli_benchmark/wpt_cross/server.py @@ -29,6 +29,10 @@ import http.client import json import math import mimetypes +import os +import random +from datetime import datetime + import re import socket import struct @@ -121,6 +125,20 @@ NAVIGATION_SECOND_VISIT_PATH = ( "/navigation-api/navigation-methods/return-value/resources/" "204-205-download-on-second-visit.py" ) +SERVICE_WORKER_SCRIPT_RESOURCE_PATHS = { + "/service-workers/service-worker/resources/redirect.py", + "/service-workers/service-worker/resources/update-worker.py", + "/service-workers/service-worker/resources/update-worker-from-file.py", + "/service-workers/service-worker/resources/update-during-installation-worker.py", + "/service-workers/service-worker/ServiceWorkerGlobalScope/resources/update-worker.py", + "/service-workers/service-worker/resources/import-scripts-version.py", + "/service-workers/service-worker/resources/import-scripts-get.py", + "/service-workers/service-worker/resources/import-scripts-echo.py", + "/service-workers/service-worker/resources/subdir/import-scripts-echo.py", + "/service-workers/service-worker/resources/scope2/import-scripts-echo.py", +} + + BENCH_TIMEOUT_MULTIPLIER_QUERY = "__moli_bench_timeout_multiplier" BENCH_REPORT_BRIDGE_SRC_RE = re.compile( rb"(?P\bsrc\s*=\s*)(?P['\"])" @@ -1720,6 +1738,9 @@ def _make_handler( return if self._serve_empty_location_resource(emit_body=self.command != "HEAD"): return + if unquote(urlsplit(self.path).path) in SERVICE_WORKER_SCRIPT_RESOURCE_PATHS: + self._serve_service_worker_script_resource() + return if self._serve_xhr_response_resource(): return parsed = urlparse(self.path) @@ -1754,6 +1775,9 @@ def _make_handler( return if self._serve_empty_location_resource(emit_body=self.command != "HEAD"): return + if unquote(urlsplit(self.path).path) in SERVICE_WORKER_SCRIPT_RESOURCE_PATHS: + self._serve_service_worker_script_resource() + return if self._serve_xhr_response_resource(): return parsed = urlparse(self.path) @@ -1809,6 +1833,9 @@ def _make_handler( return if self._serve_empty_location_resource(emit_body=self.command != "HEAD"): return + if unquote(urlsplit(self.path).path) in SERVICE_WORKER_SCRIPT_RESOURCE_PATHS: + self._serve_service_worker_script_resource() + return if self._serve_xhr_response_resource(): return parsed = urlparse(self.path) @@ -1847,6 +1874,9 @@ def _make_handler( } | XHR_BODY_RESOURCE_PATHS: self._serve_xhr_response_resource() return + if unquote(urlsplit(self.path).path) in SERVICE_WORKER_SCRIPT_RESOURCE_PATHS: + self._serve_service_worker_script_resource() + return parsed = urlparse(self.path) if unquote(parsed.path) == NAVIGATION_SECOND_VISIT_PATH: self._serve_navigation_second_visit() @@ -2061,6 +2091,9 @@ def _make_handler( return if self._serve_empty_location_resource(emit_body=emit_body): return + if unquote(urlsplit(self.path).path) in SERVICE_WORKER_SCRIPT_RESOURCE_PATHS: + self._serve_service_worker_script_resource() + return if self._serve_xhr_response_resource(emit_body=emit_body): return parsed = urlparse(self.path) @@ -2549,6 +2582,126 @@ def _make_handler( emit_body=emit_body, cache_control=None, ) + def _serve_service_worker_script_resource(self) -> None: + if not self._consume_request_body(): + return + parsed = urlsplit(self.path) + path = unquote(parsed.path) + params = parse_qs(parsed.query, keep_blank_values=True, encoding="latin-1") + headers: list[tuple[str, str]] = [] + status = 200 + body = b"" + try: + if path.endswith("/redirect.py"): + status = int(params.get("Status", ["302"])[0]) + headers.append(("Location", params["Redirect"][0])) + if "ACAOrigin" in params: + headers.extend( + ("Access-Control-Allow-Origin", value) + for value in params["ACAOrigin"][0].split(",") + ) + for suffix in ("Headers", "Methods", "Credentials"): + if "ACA" + suffix in params: + headers.append(( + "Access-Control-Allow-" + suffix, + params["ACA" + suffix][0], + )) + if "ACEHeaders" in params: + headers.append(("Access-Control-Expose-Headers", params["ACEHeaders"][0])) + else: + headers = [ + ("Cache-Control", "no-cache, must-revalidate"), + ("Pragma", "no-cache"), + ("Content-Type", "application/javascript"), + ] + if path.endswith("/update-worker-from-file.py"): + count = fetch_stash.increment(params["Key"][0], path=parsed.path) + if count > 2: + self.send_error(500, "Unknown update worker state") + return + filename = os.fsdecode(params["First" if count == 1 else "Second"][0].encode("latin-1")) + source = ((wpt_root / path.lstrip("/")).parent / filename).resolve() + source.relative_to(wpt_root.resolve()) + body = source.read_bytes() + elif path.endswith("/ServiceWorkerGlobalScope/resources/update-worker.py"): + headers = [ + ("Cache-Control", "max-age: 0"), + ("Content-Type", "application/javascript"), + ] + source = (wpt_root / path.lstrip("/")).with_suffix(".js") + script = source.read_text(encoding="utf-8") + body = f"// {time.time()}\n{script}".encode("utf-8") + elif path.endswith("/update-during-installation-worker.py"): + headers = [ + ("Content-Type", "application/javascript"), + ("Cache-Control", "max-age=0"), + ] + body = ( + f"// {random.random()}\n" + "importScripts('update-during-installation-worker.js');" + ).encode("ascii") + elif path.endswith("/import-scripts-version.py"): + # Match the upstream delay so update checks see new bytes. + if stopping.wait(0.1): + self.close_connection = True + return + version = (datetime.now() - datetime(1970, 1, 1)).total_seconds() + body = f'version = "{version}";\n'.encode("ascii") + elif path.endswith("/import-scripts-get.py"): + body = ('%s = "%s";\n' % ( + params["output"][0], params["msg"][0], + )).encode("latin-1") + elif path.endswith("/import-scripts-echo.py"): + directory = path.rsplit("/", 2)[-2] + suffix = f" ({directory}/)" if directory in ("subdir", "scope2") else "" + body = ('echo_output = "%s%s";\n' % ( + params["msg"][0], suffix, + )).encode("latin-1") + else: + mode = params["Mode"][0] + count = fetch_stash.increment(params["Key"][0], path=parsed.path) + extra_body = "" + if count == 2: + if mode == "bad_mime_type": + headers[-1] = ("Content-Type", "text/html") + elif mode == "not_found": + status = 404 + headers = [("Content-Type", "text/plain")] + elif mode == "redirect": + status = 301 + location = unquote(params.get("Redirect", ["empty.js"])[0]) + headers.append(("Location", location)) + elif mode == "syntax_error": + extra_body = "badsyntax(isbad;" + elif mode == "throw_install": + extra_body = ( + "addEventListener('install', function(e) { " + "throw new Error('boom'); });" + ) + if status == 404: + body = b"Page not found" + elif status == 301: + body = f"/* {count} */".encode("ascii") + else: + body = f"/* {count} */ {extra_body}".encode("utf-8") + if not 100 <= status <= 599: + raise ValueError("invalid response status") + for _, value in headers: + if "\r" in value or "\n" in value: + raise ValueError("invalid response header") + value.encode("latin-1") + except OSError: + self.send_error(500) + return + except (KeyError, ValueError, UnicodeError): + self.send_error(400) + return + self._send_bytes( + None, body, emit_body=self.command != "HEAD", + extra_headers=headers, status_code=status, cache_control=None, + ) + + def _serve_xhr_delay(self, query: str, *, emit_body: bool) -> None: delay_seconds = _wpt_delay_seconds(query) if delay_seconds is None: @@ -2600,6 +2753,8 @@ def _make_handler( return self._serve_navigation_second_visit if name.startswith("do_") and unquote(urlparse(self.path).path) == FETCH_EMPTY_LOCATION_PATH: return self._serve_empty_location_resource + if name.startswith("do_") and unquote(urlsplit(self.path).path) in SERVICE_WORKER_SCRIPT_RESOURCE_PATHS: + return self._serve_service_worker_script_resource if name.startswith("do_") and unquote(urlparse(self.path).path) in XHR_RESPONSE_RESOURCE_PATHS: return self._serve_xhr_response_resource if name.startswith("do_") and unquote(urlparse(self.path).path) in FETCH_RANGE_RESOURCE_PATHS: diff --git a/moli-benchmark/tests/test_wpt_service_worker_script_server.py b/moli-benchmark/tests/test_wpt_service_worker_script_server.py new file mode 100644 index 0000000000..19a1ecf8f0 --- /dev/null +++ b/moli-benchmark/tests/test_wpt_service_worker_script_server.py @@ -0,0 +1,346 @@ +from __future__ import annotations + +import re +import tempfile +import time +import unittest +import uuid +from concurrent.futures import ThreadPoolExecutor +from contextlib import ExitStack +from datetime import datetime +from http.client import HTTPConnection +from pathlib import Path +from urllib.parse import urlencode +from unittest.mock import patch + +from moli_benchmark.wpt_cross.case_set import enumerate_cases +from moli_benchmark.wpt_cross.server import WptFixtureServer + + +DIRECTORY = "service-workers/service-worker" +RESOURCES = "/" + DIRECTORY + "/resources/" + + +class ServiceWorkerScriptFixtureTests(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() + for name in ("testharness.js", "testharnessreport.js"): + (self.root / "resources" / name).write_text("// harness") + resources = self.root / DIRECTORY / "resources" + resources.mkdir(parents=True) + for name in ( + "redirect.py", "update-worker.py", "update-worker-from-file.py", "import-scripts-version.py", + "update-during-installation-worker.py", + "import-scripts-get.py", "import-scripts-echo.py", + "subdir/import-scripts-echo.py", "scope2/import-scripts-echo.py", + ): + (resources / name).parent.mkdir(exist_ok=True) + (resources / name).write_text("# Python source must not be sent as a script") + global_resources = self.root / DIRECTORY / "ServiceWorkerGlobalScope/resources" + global_resources.mkdir(parents=True) + (global_resources / "update-worker.py").write_text("# Python source must not be sent as a script") + (global_resources / "update-worker.js").write_bytes(b"// caf\xc3\xa9\r\nself.ready = true;\r\n") + self.stack.enter_context(patch( + "moli_benchmark.wpt_cross.server._global_ipv6_address", return_value=None + )) + + def server(self) -> WptFixtureServer: + return self.stack.enter_context(WptFixtureServer(self.root)) + + def request( + self, port: int, resource: str, query: str = "", *, method: str = "GET", + ) -> tuple[int, list[tuple[str, str]], bytes]: + connection = HTTPConnection("127.0.0.1", port, timeout=5) + try: + path = resource if resource.startswith("/") else RESOURCES + resource + connection.request(method, path + "?" + query) + response = connection.getresponse() + return ( + response.status, + [(name.lower(), value) for name, value in response.getheaders()], + response.read(), + ) + finally: + connection.close() + + def test_redirect_preserves_status_location_and_repeated_cors_headers(self) -> None: + server = self.server() + query = urlencode([ + ("Redirect", "../target.js?one=1&two=%20"), + ("Redirect", "ignored.js"), + ("Status", "307"), + ("ACAOrigin", "https://one.test,https://two.test"), + ("ACAHeaders", "X-One, X-Two"), + ("ACAMethods", "GET, OPTIONS"), + ("ACACredentials", "true"), + ("ACEHeaders", "X-Result"), + ]) + for method in ("GET", "HEAD", "POST", "OPTIONS", "PUT", "YO", "CUSTOM"): + with self.subTest(method=method): + status, headers, body = self.request(server.port, "redirect.py", query, method=method) + self.assertEqual(status, 307) + self.assertEqual(body, b"") + self.assertEqual(dict(headers)["location"], "../target.js?one=1&two=%20") + self.assertEqual( + [value for name, value in headers if name == "access-control-allow-origin"], + ["https://one.test", "https://two.test"], + ) + for name, value in ( + ("access-control-allow-headers", "X-One, X-Two"), + ("access-control-allow-methods", "GET, OPTIONS"), + ("access-control-allow-credentials", "true"), + ("access-control-expose-headers", "X-Result"), + ): + self.assertEqual(dict(headers)[name], value) + self.assertNotIn("content-type", dict(headers)) + self.assertNotIn("cache-control", dict(headers)) + status, headers, _ = self.request(server.port, "%72edirect.py", "Redirect=%FF") + self.assertEqual(status, 302) + self.assertEqual(dict(headers)["location"], "\xff") + + def test_update_modes_only_change_the_second_response(self) -> None: + server = self.server() + modes = ( + ("normal", 200, "application/javascript", b"/* 2 */ "), + ("bad_mime_type", 200, "text/html", b"/* 2 */ "), + ("not_found", 404, "text/plain", b"Page not found"), + ("redirect", 301, "application/javascript", b"/* 2 */"), + ("syntax_error", 200, "application/javascript", b"/* 2 */ badsyntax(isbad;"), + ("throw_install", 200, "application/javascript", b"/* 2 */ addEventListener('install', function(e) { throw new Error('boom'); });"), + ("unknown", 200, "application/javascript", b"/* 2 */ "), + ) + for mode, expected_status, expected_type, expected_body in modes: + query = urlencode({"Key": str(uuid.uuid4()), "Mode": mode}) + with self.subTest(mode=mode): + self.assertEqual(self.request(server.port, "update-worker.py", query)[2], b"/* 1 */ ") + status, headers, body = self.request(server.port, "update-worker.py", query) + self.assertEqual((status, dict(headers)["content-type"], body), + (expected_status, expected_type, expected_body)) + if mode == "redirect": + self.assertEqual(dict(headers)["location"], "empty.js") + if mode == "not_found": + self.assertNotIn("cache-control", dict(headers)) + else: + self.assertEqual(dict(headers)["cache-control"], "no-cache, must-revalidate") + self.assertEqual(dict(headers)["pragma"], "no-cache") + self.assertEqual(self.request(server.port, "update-worker.py", query)[2], b"/* 3 */ ") + + def test_update_state_is_shared_across_origins_and_modes_but_not_servers(self) -> None: + first, second = self.server(), self.server() + key = str(uuid.uuid4()) + first.fetch_stash.put(key, 50) + first_query = urlencode({"Key": key, "Mode": "normal"}) + self.assertEqual(self.request(first.port, "update-worker.py", first_query)[2], b"/* 1 */ ") + # Equivalent UUID spelling, first repeated parameter, and a second + # percent decode of Redirect all follow the upstream handler. + redirect_query = urlencode([ + ("Key", "{" + key.upper() + "}"), ("Mode", "redirect"), + ("Mode", "not_found"), ("Redirect", "target.js?one=1%26two=2"), + ]) + status, headers, body = self.request(first.alternate_port, "update-worker.py", redirect_query) + self.assertEqual((status, body), (301, b"/* 2 */")) + self.assertEqual(dict(headers)["location"], "target.js?one=1&two=2") + self.assertEqual(self.request(second.port, "update-worker.py", first_query)[2], b"/* 1 */ ") + self.assertEqual(self.request(first.port, "update-worker.py", first_query)[2], b"/* 3 */ ") + self.assertEqual(first.fetch_stash.take(key), 50) + + def test_concurrent_updates_keep_every_visit(self) -> None: + server = self.server() + query = urlencode({"Key": str(uuid.uuid4()), "Mode": "normal"}) + def visit(index: int) -> int: + port = server.port if index % 2 else server.alternate_port + status, _, body = self.request(port, "update-worker.py", query) + self.assertEqual(status, 200) + match = re.fullmatch(rb"/\* (\d+) \*/ ", body) + self.assertIsNotNone(match) + return int(match[1]) + with ThreadPoolExecutor(max_workers=6) as pool: + self.assertEqual(sorted(pool.map(visit, range(18))), list(range(1, 19))) + + def test_head_runs_the_update_handler_without_emitting_its_body(self) -> None: + server = self.server() + query = urlencode({"Key": str(uuid.uuid4()), "Mode": "not_found"}) + for expected_status, expected_length in ((200, 8), (404, 14), (200, 8)): + status, headers, body = self.request(server.port, "update-worker.py", query, method="HEAD") + self.assertEqual(status, expected_status) + self.assertEqual(dict(headers)["content-length"], str(expected_length)) + self.assertEqual(body, b"") + + def test_imported_version_changes_with_revalidation_headers(self) -> None: + server = self.server() + versions = [] + for port in (server.port, server.alternate_port, server.port): + before = (datetime.now() - datetime(1970, 1, 1)).total_seconds() + status, headers, body = self.request(port, "import-scripts-version.py") + after = (datetime.now() - datetime(1970, 1, 1)).total_seconds() + self.assertEqual(status, 200) + self.assertEqual(dict(headers)["content-type"], "application/javascript") + self.assertEqual(dict(headers)["cache-control"], "no-cache, must-revalidate") + self.assertEqual(dict(headers)["pragma"], "no-cache") + match = re.fullmatch(rb'version = "([0-9.]+)";\n', body) + self.assertIsNotNone(match) + versions.append(float(match[1])) + self.assertLessEqual(before, versions[-1]) + self.assertLessEqual(versions[-1], after) + self.assertLess(versions[0], versions[1]) + self.assertLess(versions[1], versions[2]) + + def test_update_from_file_preserves_bytes_and_shares_only_its_own_stash(self) -> None: + first, second = self.server(), self.server() + resources = self.root / DIRECTORY / "resources" + before, after = b"// before\r\n\xff\n", b"// after\x00\n" + (resources / "café.js").write_bytes(before) + (resources / "after.js").write_bytes(after) + key = str(uuid.uuid4()) + query = urlencode([ + ("Key", key), ("First", "café.js"), ("First", "ignored.js"), + ("Second", "after.js"), + ]) + for port, expected in ( + (first.port, before), + (second.port, before), + (first.alternate_port, after), + ): + status, headers, body = self.request(port, "update-worker-from-file.py", query) + self.assertEqual((status, body), (200, expected)) + self.assertEqual(dict(headers)["content-type"], "application/javascript") + self.assertEqual(dict(headers)["cache-control"], "no-cache, must-revalidate") + self.assertEqual(dict(headers)["pragma"], "no-cache") + self.assertEqual(self.request(first.port, "update-worker-from-file.py", query)[0], 500) + self.assertEqual( + self.request(first.port, "update-worker.py", urlencode({"Key": key, "Mode": "normal"}))[2], + b"/* 1 */ ", + ) + head_query = urlencode({"Key": str(uuid.uuid4()), "First": "café.js", "Second": "after.js"}) + status, headers, body = self.request(first.port, "update-worker-from-file.py", head_query, method="HEAD") + self.assertEqual((status, dict(headers)["content-length"], body), (200, str(len(before)), b"")) + self.assertEqual(self.request(first.port, "update-worker-from-file.py", head_query, method="POST")[2], after) + failure_query = urlencode({"Key": str(uuid.uuid4()), "First": "missing.js", "Second": "after.js"}) + self.assertEqual(self.request(first.port, "update-worker-from-file.py", failure_query)[0], 500) + self.assertEqual(self.request(first.port, "update-worker-from-file.py", failure_query)[2], after) + + def test_global_scope_update_script_uses_timestamp_and_original_text(self) -> None: + server = self.server() + resource = "/" + DIRECTORY + "/ServiceWorkerGlobalScope/resources/update-worker.py" + script = "// café\nself.ready = true;\n".encode("utf-8") + versions = [] + for method in ("GET", "POST", "OPTIONS", "HEAD"): + before = time.time() + status, headers, body = self.request(server.port, resource, method=method) + after = time.time() + self.assertEqual(status, 200) + self.assertEqual(dict(headers)["cache-control"], "max-age: 0") + self.assertEqual(dict(headers)["content-type"], "application/javascript") + self.assertNotIn("pragma", dict(headers)) + if method == "HEAD": + self.assertEqual(body, b"") + self.assertGreater(int(dict(headers)["content-length"]), len(script)) + continue + stamp, source = body.split(b"\n", 1) + self.assertEqual(source, script) + self.assertTrue(stamp.startswith(b"// ")) + versions.append(float(stamp[3:])) + self.assertLessEqual(before, versions[-1]) + self.assertLessEqual(versions[-1], after) + self.assertEqual(versions, sorted(set(versions))) + + def test_installation_update_script_changes_without_rewriting_its_import(self) -> None: + server = self.server() + versions = [] + for method in ("GET", "POST", "OPTIONS", "HEAD"): + status, headers, body = self.request(server.port, "update-during-installation-worker.py", method=method) + self.assertEqual(status, 200) + self.assertEqual(dict(headers)["content-type"], "application/javascript") + self.assertEqual(dict(headers)["cache-control"], "max-age=0") + self.assertNotIn("pragma", dict(headers)) + if method == "HEAD": + self.assertEqual(body, b"") + continue + version, script = body.split(b"\n", 1) + self.assertEqual(script, b"importScripts('update-during-installation-worker.js');") + self.assertTrue(version.startswith(b"// ")) + versions.append(float(version[3:])) + self.assertGreaterEqual(versions[-1], 0) + self.assertLess(versions[-1], 1) + self.assertEqual(len(set(versions)), 3) + + def test_imported_assignments_preserve_raw_parameters_and_directory(self) -> None: + server = self.server() + for resource, query, expected in ( + ("import-scripts-get.py", "output=echo1&msg=test1", b'echo1 = "test1";\n'), + ("import-scripts-get.py", "output=x&output=y&msg=%FF%22&msg=ignored", b'x = "\xff\"";\n'), + ("import-scripts-echo.py", "msg=top+level", b'echo_output = "top level";\n'), + ("import-scripts-echo.py", "msg=", b'echo_output = "";\n'), + ("subdir/import-scripts-echo.py", "msg=install", b'echo_output = "install (subdir/)";\n'), + ("scope2/import-scripts-echo.py", "msg=message", b'echo_output = "message (scope2/)";\n'), + ): + for method in ("GET", "HEAD", "POST", "OPTIONS"): + with self.subTest(resource=resource, query=query, method=method): + status, headers, body = self.request(server.port, resource, query, method=method) + self.assertEqual(status, 200) + self.assertEqual(body, b"" if method == "HEAD" else expected) + self.assertEqual(dict(headers)["content-length"], str(len(expected))) + self.assertEqual(dict(headers)["content-type"], "application/javascript") + self.assertEqual(dict(headers)["cache-control"], "no-cache, must-revalidate") + self.assertEqual(dict(headers)["pragma"], "no-cache") + + def test_malformed_parameters_are_not_served_as_javascript(self) -> None: + server = self.server() + for resource, query in ( + ("redirect.py", ""), + ("redirect.py", "Redirect=x&Status=invalid"), + ("redirect.py", "Redirect=x&Status=999"), + ("redirect.py", "Redirect=%0D%0AX-Injected:yes"), + ("redirect.py", "Redirect=x&ACAOrigin=%0Ainjected"), + ("update-worker.py", "Mode=normal"), + ("update-worker.py", "Key=invalid&Mode=normal"), + ("update-worker.py", "Key=" + str(uuid.uuid4())), + ("update-worker-from-file.py", "First=x&Second=y"), + ("update-worker-from-file.py", urlencode({"Key": str(uuid.uuid4()), "First": "../../../../outside.js"})), + ("import-scripts-get.py", "output=x"), + ("import-scripts-get.py", "msg=x"), + ("import-scripts-echo.py", ""), + ): + with self.subTest(resource=resource, query=query): + self.assertEqual(self.request(server.port, resource, query)[0], 400) + + def test_discovery_only_accepts_supported_handler_locations(self) -> None: + cases = { + "redirect-relative.html": ("resources/redirect.py", True), + "update-absolute.html": (RESOURCES + "update-worker.py", True), + "update-from-file.html": ("resources/update-worker-from-file.py", True), + "update-installing.html": ("resources/update-during-installation-worker.py", True), + "ServiceWorkerGlobalScope/update.html": ("resources/update-worker.py", True), + "ServiceWorkerGlobalScope/update-wrong.html": ("resources/update-worker-from-file.py", False), + "sub/version-relative.html": ("../resources/import-scripts-version.py", True), + "sub/redirect-dot-relative.html": ("./../resources/redirect.py", True), + "sub/redirect-wrong-relative.html": ("resources/redirect.py", False), + "redirect-suffix.html": ("resources/redirect.py.extra", False), + "redirect-other.html": ("/unrelated/resources/redirect.py", False), + "get.html": ("resources/import-scripts-get.py", True), + "echo.html": ("resources/import-scripts-echo.py", True), + "echo-subdir.html": ("resources/subdir/import-scripts-echo.py", True), + "echo-scope2.html": (RESOURCES + "scope2/import-scripts-echo.py", True), + "echo-wrong.html": ("resources/wrong/import-scripts-echo.py", False), + "unknown.html": ("resources/unknown.py", False), + } + for name, (reference, _) in cases.items(): + path = self.root / DIRECTORY / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + '' + '' + f'' + ) + discovered = enumerate_cases(self.root, dir_prefixes=(DIRECTORY,)) + self.assertEqual( + sorted(case.case_path for case in discovered), + sorted(DIRECTORY + "/" + name for name, (_, allowed) in cases.items() if allowed), + ) + + +if __name__ == "__main__": + unittest.main()