From d8a0faa3fa552ca07c30e33e93a5e7a481eb4818 Mon Sep 17 00:00:00 2001 From: ldm0 Date: Sat, 12 Sep 2026 15:18:31 +0800 Subject: [PATCH] fix(wpt): model Fetch redirect fixtures Serve Fetch redirect.py with upstream Origin and credentials responses, preflight handling, query inheritance, counters, delays and form status codes. Recognize its relative, absolute and RESOURCES_DIR references during discovery. Cover the HTTP behavior and discovery with regressions. Verify the fixture against upstream wptserve, including byte parsing and unread request bodies. --- .../moli_benchmark/wpt_cross/case_set.py | 11 +- .../moli_benchmark/wpt_cross/server.py | 129 +++++++++- .../tests/test_wpt_fetch_redirect_server.py | 235 ++++++++++++++++++ 3 files changed, 364 insertions(+), 11 deletions(-) create mode 100644 moli-benchmark/tests/test_wpt_fetch_redirect_server.py diff --git a/moli-benchmark/moli_benchmark/wpt_cross/case_set.py b/moli-benchmark/moli_benchmark/wpt_cross/case_set.py index c90de1d023..fc5d5cac0d 100644 --- a/moli-benchmark/moli_benchmark/wpt_cross/case_set.py +++ b/moli-benchmark/moli_benchmark/wpt_cross/case_set.py @@ -753,8 +753,8 @@ def _script_content_type_handler_reference_patterns(directory: str) -> tuple[re. @lru_cache(maxsize=None) -def _fetch_preflight_handler_reference_patterns(directory: str) -> tuple[re.Pattern[str], ...]: - names = ("preflight.py", "clean-stash.py", "inspect-headers.py") +def _fetch_resource_handler_reference_patterns(directory: str) -> tuple[re.Pattern[str], ...]: + names = ("preflight.py", "clean-stash.py", "inspect-headers.py", "redirect.py") references = [] for name in names: resource = "fetch/api/resources/" + name @@ -774,6 +774,11 @@ def _fetch_preflight_handler_reference_patterns(directory: str) -> tuple[re.Patt + "|".join(re.escape(name) for name in names) + rf"){WPTSERVE_HANDLER_TRAILING_BOUNDARY}" )) + patterns.append(re.compile( + r"`\$\{\s*RESOURCES_DIR\s*\}(?:" + + "|".join(re.escape(name) for name in names) + + rf"){WPTSERVE_HANDLER_TRAILING_BOUNDARY}" + )) return tuple(patterns) @@ -893,7 +898,7 @@ def _supported_wptserve_handler_references( if rel is not None and rel.rsplit("/", 1)[0] == "fetch/range": supported += SUPPORTED_FETCH_RANGE_WPTSERVE_HANDLER_PATTERNS if rel is not None and rel.startswith("fetch/api/"): - supported += _fetch_preflight_handler_reference_patterns(posixpath.dirname(rel)) + supported += _fetch_resource_handler_reference_patterns(posixpath.dirname(rel)) if rel is not None and rel.startswith("wasm/webapi/"): supported += SUPPORTED_WASM_WEBAPI_WPTSERVE_HANDLER_PATTERNS 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 a9061c14e3..549a35a009 100644 --- a/moli-benchmark/moli_benchmark/wpt_cross/server.py +++ b/moli-benchmark/moli_benchmark/wpt_cross/server.py @@ -45,7 +45,7 @@ from html import escape as html_escape from email.utils import formatdate from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path -from urllib.parse import parse_qs, parse_qsl, quote, unquote, urlparse, urlsplit, urlunsplit +from urllib.parse import parse_qs, parse_qsl, quote, unquote, urlencode, urlparse, urlsplit, urlunsplit from .pipes import WptPipeError, parse_pipe_commands @@ -151,6 +151,7 @@ FETCH_PREFLIGHT_RESOURCE_PATHS = { "/fetch/api/resources/preflight.py", "/fetch/api/resources/clean-stash.py", } +FETCH_REDIRECT_RESOURCE_PATH = "/fetch/api/resources/redirect.py" BENCH_TIMEOUT_MULTIPLIER_QUERY = "__moli_bench_timeout_multiplier" FORM_ECHO_PATH = "/html/semantics/forms/form-submission-0/form-echo.py" FORM_SUBMISSION_PATH = ( @@ -1340,6 +1341,40 @@ def _redirect_fixture_response(query: str) -> tuple[int, str] | None: return status, location +def _fetch_redirect_form_status(method: str, content_type: str | None, body: bytes) -> str | None: + """Read the first redirect_status field like wptserve's CGI form parser.""" + if content_type is None: + content_type = "application/x-www-form-urlencoded" if method == "POST" else "text/plain" + media_type = content_type.partition(";")[0].strip() + if media_type == "application/x-www-form-urlencoded": + params = parse_qs(body.decode("latin-1"), keep_blank_values=True, encoding="latin-1") + return params.get("redirect_status", [None])[0] + if media_type.startswith("multipart/"): + message = BytesParser(policy=policy.HTTP).parsebytes( + f"Content-Type: {content_type}\r\n\r\n".encode("latin-1") + body + ) + boundary = message.get_boundary() + if boundary is None or re.fullmatch(r"[ -~]{0,200}[!-~]", boundary) is None: + raise ValueError("Invalid multipart boundary") + for part in message.iter_parts(): + if part.get_param("name", header="content-disposition") == "redirect_status": + if part.get_filename(): + raise ValueError("A file is not a redirect status") + # CGI leaves transfer encodings untouched and treats form + # values as isomorphic bytes, regardless of part charset. + del part["Content-Transfer-Encoding"] + payload = part.get_payload(decode=True) + if payload is None: + raise ValueError("A multipart value is not a redirect status") + return payload.decode("latin-1") + return None + if body: + # Upstream FieldStorage's non-form binary read raises TypeError when + # writing a nonempty upload into its text buffer. + raise ValueError("Unsupported non-form upload") + return None + + def _content_security_policy_resource_response() -> tuple[bytes, list[tuple[str, str]]]: """Return the minimal CSP resource.py fixture used by worker CSP WPT.""" @@ -1954,7 +1989,7 @@ def _make_handler( return self._serve_fetch_resource_method if path in XHR_RESOURCE_PATHS: return self._serve_xhr_method - if path in FETCH_PREFLIGHT_RESOURCE_PATHS: + if path in FETCH_PREFLIGHT_RESOURCE_PATHS or path == FETCH_REDIRECT_RESOURCE_PATH: return self._serve_fetch_resource_method raise AttributeError(name) @@ -1986,7 +2021,7 @@ def _make_handler( self._serve_navigation_second_visit() return if path in FETCH_ABORT_RESOURCE_PATHS | FETCH_RANGE_RESOURCE_PATHS | FETCH_PREFLIGHT_RESOURCE_PATHS | { - "/fetch/api/resources/status.py", "/fetch/api/resources/trickle.py" + "/fetch/api/resources/status.py", "/fetch/api/resources/trickle.py", FETCH_REDIRECT_RESOURCE_PATH }: self._serve_fetch_resource_method() return @@ -2023,7 +2058,7 @@ def _make_handler( self._serve_navigation_second_visit() return if path in FETCH_ABORT_RESOURCE_PATHS | FETCH_RANGE_RESOURCE_PATHS | FETCH_PREFLIGHT_RESOURCE_PATHS | { - "/fetch/api/resources/status.py", "/fetch/api/resources/trickle.py" + "/fetch/api/resources/status.py", "/fetch/api/resources/trickle.py", FETCH_REDIRECT_RESOURCE_PATH }: self._serve_fetch_resource_method() return @@ -2103,6 +2138,9 @@ def _make_handler( if path in FETCH_RANGE_RESOURCE_PATHS: self._serve_fetch_range_resource(path, parsed.query, emit_body=self.command != "HEAD") return + if unquote(parsed.path) == FETCH_REDIRECT_RESOURCE_PATH: + self._serve_fetch_redirect_resource(parsed.query, emit_body=self.command != "HEAD") + return if unquote(parsed.path) in FETCH_PREFLIGHT_RESOURCE_PATHS: self._serve_fetch_preflight_resource( unquote(parsed.path), parsed.query, emit_body=self.command != "HEAD" @@ -2143,7 +2181,7 @@ def _make_handler( self._serve_navigation_second_visit() return if unquote(parsed.path) in FETCH_ABORT_RESOURCE_PATHS | FETCH_RANGE_RESOURCE_PATHS | FETCH_PREFLIGHT_RESOURCE_PATHS | { - "/fetch/api/resources/status.py", "/fetch/api/resources/trickle.py" + "/fetch/api/resources/status.py", "/fetch/api/resources/trickle.py", FETCH_REDIRECT_RESOURCE_PATH }: self._serve_fetch_resource_method() return @@ -2374,6 +2412,9 @@ def _make_handler( if path == LINK_STYLESHEET_COUNTER_PATH: self._serve_link_stylesheet_counter(parsed.query, emit_body=emit_body) return + if path == FETCH_REDIRECT_RESOURCE_PATH: + self._serve_fetch_redirect_resource(parsed.query, emit_body=emit_body) + return if path in FETCH_PREFLIGHT_RESOURCE_PATHS: self._serve_fetch_preflight_resource(path, parsed.query, emit_body=emit_body) return @@ -2512,7 +2553,6 @@ def _make_handler( ) return if path in { - "/fetch/api/resources/redirect.py", "/common/redirect.py", "/common/redirect-opt-in.py", }: @@ -2827,8 +2867,8 @@ 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: + def _read_content_length_request_body(self, *, ignore_transfer_encoding: bool = False) -> bytes | None: + if not ignore_transfer_encoding and self.headers.get("Transfer-Encoding") is not None: self._reject_request_body(400) return None length_str = self.headers.get("Content-Length") @@ -3413,6 +3453,79 @@ def _make_handler( except OSError: pass # Upstream ends its stream when a write reports disconnect. + def _serve_fetch_redirect_resource(self, query: str, *, emit_body: bool) -> None: + connection_headers = [] + if (self.headers.get("Transfer-Encoding") is not None + or self.headers.get("Content-Length", "0").strip() not in {"", "0"}): + # A query status or an ordinary OPTIONS response never reads + # the upload in upstream redirect.py. Do not wait for EOF. + self.close_connection = True + connection_headers.append(("Connection", "close")) + params = parse_qs(query, keep_blank_values=True, encoding="latin-1") + stash_path = urlsplit(self.path).path + headers = [*connection_headers, ("Content-Type", "text/plain"), ("Pragma", "no-cache")] + if "Origin" in self.headers: + headers.extend([ + ("Access-Control-Allow-Origin", self.headers.get("Origin", "")), + ("Access-Control-Allow-Credentials", "true"), + ]) + else: + headers.append(("Access-Control-Allow-Origin", "*")) + token = params.get("token", [None])[0] + data = {"count": 0, "preflight": "0"} + try: + if "token" in params: + data = fetch_stash.take(token, path=stash_path) or data + if self.command == "OPTIONS": + if "allow_headers" in params: + headers.append(("Access-Control-Allow-Headers", params["allow_headers"][0])) + data["preflight"] = "1" + if "redirect_preflight" not in params: + if token: + fetch_stash.put(token, data, path=stash_path) + self._send_bytes(None, b"", emit_body=emit_body, extra_headers=headers, + cache_control="no-cache") + return + + status = 302 + if "redirect_status" in params: + status = int(params["redirect_status"][0].encode("latin-1")) + elif self.command not in {"GET", "HEAD"}: + # wptserve's CGI input is bounded by Content-Length even + # when Transfer-Encoding is also present. + body = self._read_content_length_request_body(ignore_transfer_encoding=True) + if body is None: + return + form_status = _fetch_redirect_form_status( + self.command, self.headers.get("Content-Type"), body + ) + if form_status is not None: + status = int(form_status.encode("latin-1")) + data["count"] += 1 + if "location" in params: + location = params["location"][0] + if "simple" not in params and urlparse(location).scheme in {"", "http", "https"}: + location += "&" if "?" in location else "?" + location += urlencode({name: values[0] for name, values in params.items()}) + location += "&count=" + str(data["count"]) + headers.append(("Location", location)) + if "redirect_referrerpolicy" in params: + headers.append(("Referrer-Policy", params["redirect_referrerpolicy"][0])) + if "delay" in params: + time.sleep(float(params["delay"][0].encode("latin-1")) / 1000) + if token: + fetch_stash.put(token, data, path=stash_path) + if "max_count" in params and data["count"] > int(params["max_count"][0].encode("latin-1")): + # Upstream returns a plain body instead of its tuple; + # none of the redirect/CORS headers survive that return. + self._send_bytes(None, str(data["count"] - 1).encode(), emit_body=emit_body, + extra_headers=connection_headers, cache_control=None) + return + self._send_bytes(None, b"", emit_body=emit_body, extra_headers=headers, + status_code=status, cache_control="no-cache") + except (KeyError, ValueError, TypeError, OverflowError): + self.send_error(500) + def _serve_fetch_preflight_resource( self, path: str, query: str, *, emit_body: bool ) -> None: diff --git a/moli-benchmark/tests/test_wpt_fetch_redirect_server.py b/moli-benchmark/tests/test_wpt_fetch_redirect_server.py new file mode 100644 index 0000000000..4e90045d1e --- /dev/null +++ b/moli-benchmark/tests/test_wpt_fetch_redirect_server.py @@ -0,0 +1,235 @@ +from __future__ import annotations + +import tempfile +import unittest +import uuid +from concurrent.futures import ThreadPoolExecutor +from contextlib import ExitStack +from http.client import HTTPConnection +from pathlib import Path +from threading import Event +from unittest.mock import patch +from urllib.parse import urlencode + +from moli_benchmark.wpt_cross.case_set import enumerate_cases +from moli_benchmark.wpt_cross.server import WptFixtureServer + + +REDIRECT = "/fetch/api/resources/redirect.py" + + +class FetchRedirectFixtureTests(unittest.TestCase): + def setUp(self): + 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") + 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="", *, method="GET", headers=None, body=None, port=None, path=REDIRECT): + connection = HTTPConnection("127.0.0.1", port or self.server.port, timeout=2) + try: + connection.request(method, path + "?" + query, body, headers or {}) + response = connection.getresponse() + return response.status, response.headers, response.read() + finally: + connection.close() + + def test_standard_and_extension_methods_echo_the_request_origin(self): + query = "redirect_status=303&location=%2Ftarget" + for method in ("GET", "HEAD", "POST", "PUT", "PATCH", "patcH", "DELETE", "YO", "chicken", "*"): + with self.subTest(method=method): + status, headers, body = self.request(query, method=method, body=b"payload", headers={ + "Origin": "https://caller.test", + }) + self.assertEqual((status, body), (303, b"")) + self.assertEqual(headers["Content-Type"], "text/plain") + self.assertEqual(headers.get_all("Cache-Control"), ["no-cache"]) + self.assertEqual(headers["Pragma"], "no-cache") + self.assertEqual(headers["Access-Control-Allow-Origin"], "https://caller.test") + self.assertEqual(headers["Access-Control-Allow-Credentials"], "true") + self.assertEqual(headers["Location"], "/target?redirect_status=303&location=%2Ftarget&count=1") + for origin in (None, "", "null", "\xff"): + with self.subTest(origin=origin): + _, headers, _ = self.request(query, headers={} if origin is None else {"Origin": origin}) + self.assertEqual(headers["Access-Control-Allow-Origin"], "*" if origin is None else origin) + self.assertEqual(headers.get("Access-Control-Allow-Credentials"), None if origin is None else "true") + + def test_location_inherits_first_query_values_and_preserves_existing_syntax(self): + query = "location=%2Ftarget%3Fexisting%3D1&location=%2Fignored&a=x+y&a=ignored&count=20" + self.assertEqual(self.request(query)[1]["Location"], + "/target?existing=1&location=%2Ftarget%3Fexisting%3D1&a=x+y&count=20&count=1") + for simple in ("", "false", "1"): + with self.subTest(simple=simple): + self.assertEqual(self.request(query + "&simple=" + simple)[1]["Location"], "/target?existing=1") + self.assertIsNone(self.request()[1].get("Location")) + self.assertEqual(self.request("location=")[1]["Location"], "?location=&count=1") + self.assertEqual(self.request("location=&simple")[1]["Location"], "") + self.assertEqual(self.request("location=%2Ftarget%23fragment")[1]["Location"], + "/target#fragment?location=%2Ftarget%23fragment&count=1") + self.assertEqual(self.request("location=%2Ftarget&raw=%FF")[1]["Location"], + "/target?location=%2Ftarget&raw=%C3%BF&count=1") + + def test_non_http_locations_skip_parameter_inheritance(self): + for location in ("data:text/plain,hello", "blob:https://a.test/id", "about:blank", "file:///tmp/example"): + with self.subTest(location=location): + self.assertEqual(self.request(urlencode({"location": location}))[1]["Location"], location) + location = "http://user:password@a.test/path" + self.assertEqual(self.request(urlencode({"location": location, "simple": ""}))[1]["Location"], location) + + def test_options_only_redirect_when_the_flag_is_present(self): + query = "redirect_status=307&location=%2Ftarget&allow_headers=X-One%2C+x-two&allow_headers=ignored" + status, headers, body = self.request(query, method="OPTIONS", headers={"Origin": "https://caller.test"}) + self.assertEqual((status, body), (200, b"")) + self.assertIsNone(headers.get("Location")) + self.assertEqual(headers["Access-Control-Allow-Headers"], "X-One, x-two") + self.assertIsNone(headers.get("Access-Control-Allow-Methods")) + self.assertEqual(headers["Access-Control-Allow-Origin"], "https://caller.test") + self.assertEqual(headers["Access-Control-Allow-Credentials"], "true") + status, headers, body = self.request(query + "&redirect_preflight=false", method="OPTIONS") + self.assertEqual((status, body), (307, b"")) + self.assertIn("redirect_preflight=false&count=1", headers["Location"]) + + def test_counter_is_shared_between_origins_and_stops_without_redirect_headers(self): + key = str(uuid.uuid4()) + query = "token=" + key + "&location=%2Ftarget&max_count=2" + for count in range(1, 5): + with self.subTest(count=count): + status, headers, body = self.request(query, port=self.server.port if count % 2 else self.server.alternate_port) + if count <= 2: + self.assertEqual((status, body), (302, b"")) + self.assertTrue(headers["Location"].endswith("&count=" + str(count))) + else: + self.assertEqual((status, body), (200, str(count - 1).encode())) + for name in ("Location", "Content-Type", "Cache-Control", "Pragma", "Access-Control-Allow-Origin"): + self.assertIsNone(headers.get(name), name) + key = str(uuid.uuid4()) + query = "token=" + key + "&location=%2Ftarget&max_count=0" + self.assertEqual(self.request(query, method="OPTIONS")[0], 200) + self.assertEqual(self.request(query)[2], b"0") + + def test_counter_uses_the_complete_raw_request_path(self): + key = str(uuid.uuid4()) + query = "token=" + key + "&location=%2Ftarget" + encoded = REDIRECT.replace("redirect.py", "%72edirect.py") + for count in (1, 2): + for path in (REDIRECT, encoded): + with self.subTest(path=path, count=count): + self.assertTrue(self.request(query, path=path)[1]["Location"].endswith("&count=" + str(count))) + self.request("token=" + key, path="/fetch/api/resources/clean-stash.py") + self.assertTrue(self.request(query)[1]["Location"].endswith("&count=3")) + + def test_status_uses_query_before_urlencoded_and_multipart_form_fields(self): + form = b"redirect_status=307&redirect_status=301" + for method in ("POST", "PUT"): + with self.subTest(method=method): + status, _, _ = self.request("location=%2Ftarget", method=method, body=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}) + self.assertEqual(status, 307) + self.assertEqual(self.request("location=%2Ftarget", method="POST", body=form)[0], 307) + self.assertEqual(self.request("location=%2Ftarget&redirect_status=301&redirect_status=308", + method="POST", body=form)[0], 301) + self.assertEqual(self.request("location=%2Ftarget", body=form, + headers={"Content-Type": "application/x-www-form-urlencoded"})[0], 302) + body = b'--boundary\r\nContent-Disposition: form-data; name="redirect_status"\r\n\r\n308\r\n--boundary--\r\n' + self.assertEqual(self.request("location=%2Ftarget", method="POST", body=body, + headers={"Content-Type": "multipart/form-data; boundary=boundary"})[0], 308) + self.assertEqual(self.request("location=%2Ftarget", method="POST", body=form, + headers={"Content-Type": "text/plain"})[0], 500) + for content_type in ("", "APPLICATION/X-WWW-FORM-URLENCODED"): + with self.subTest(content_type=content_type): + self.assertEqual(self.request("location=%2Ftarget", method="POST", body=form, + headers={"Content-Type": content_type})[0], 500) + + def test_non_redirect_statuses_and_referrer_policy_are_preserved(self): + for code in (200, 201, 204, 299, 301, 302, 303, 304, 307, 308, 399, 400, 418, 500, 599): + with self.subTest(code=code): + status, headers, _ = self.request("redirect_status=" + str(code) + "&location=%2Ftarget") + self.assertEqual(status, code) + self.assertIsNotNone(headers.get("Location")) + _, headers, _ = self.request("location=%2Ftarget&redirect_referrerpolicy=no-referrer&redirect_referrerpolicy=unsafe-url") + self.assertEqual(headers["Referrer-Policy"], "no-referrer") + for query in ("redirect_status=bad", "redirect_status=", "token=", "token=bad", "delay=bad", "delay=-1", + "redirect_status=%A0302%A0", "delay=%A01%A0", + "token=" + str(uuid.uuid4()) + "&max_count=%A01%A0"): + with self.subTest(query=query): + self.assertEqual(self.request(query)[0], 500) + + def test_multipart_status_uses_raw_first_field_bytes(self): + for parts, expected in [ + ([(b'', b'307'), (b'', b'308')], 307), + ([(b'Content-Transfer-Encoding: base64\r\n', b'MzA3')], 500), + ([(b'Content-Type: text/plain; charset=utf-8\r\n', b'\xa0307\xa0')], 500), + ]: + with self.subTest(parts=parts): + body = b''.join( + b'--boundary\r\nContent-Disposition: form-data; name="redirect_status"\r\n' + + headers + b'\r\n' + value + b'\r\n' for headers, value in parts + ) + b'--boundary--\r\n' + self.assertEqual(self.request("location=%2Ftarget", method="POST", body=body, + headers={"Content-Type": "multipart/form-data; boundary=boundary"})[0], expected) + body = b'--boundary\r\nContent-Disposition: form-data; name="redirect_status"; filename="status.txt"\r\n\r\n307\r\n--boundary--\r\n' + self.assertEqual(self.request("location=%2Ftarget", method="POST", body=body, + headers={"Content-Type": "multipart/form-data; boundary=boundary"})[0], 500) + + def test_delay_precedes_response_headers(self): + entered, release = Event(), Event() + + def delay(seconds): + self.assertEqual(seconds, 0.025) + entered.set() + release.wait(2) + + with patch("moli_benchmark.wpt_cross.server.time.sleep", side_effect=delay), ThreadPoolExecutor() as executor: + future = executor.submit(self.request, "location=%2Ftarget&delay=25") + try: + self.assertTrue(entered.wait(2), "the redirect handler must delay its response") + self.assertFalse(future.done(), "headers must not arrive before the delay completes") + finally: + release.set() + self.assertEqual(future.result(timeout=2)[0], 302) + + def test_handlers_respond_before_unused_uploads_finish(self): + for method in ("GET", "POST", "OPTIONS"): + for framing in (("Content-Length", "1000000"), ("Transfer-Encoding", "chunked")): + with self.subTest(method=method, framing=framing): + connection = HTTPConnection("127.0.0.1", self.server.port, timeout=2) + try: + connection.putrequest(method, REDIRECT + "?redirect_status=302&location=%2Ftarget") + connection.putheader(*framing) + connection.endheaders() + response = connection.getresponse() + self.assertEqual((response.status, response.read()), (200 if method == "OPTIONS" else 302, b"")) + self.assertEqual(response.headers["Connection"], "close") + finally: + connection.close() + + def test_case_selection_recognizes_fetch_redirect_references(self): + sources = { + "absolute": "fetch('/fetch/api/resources/redirect.py');", + "concat": 'fetch(RESOURCES_DIR + "redirect.py");', + "relative": "fetch('../resources/redirect.py');", + "template": "fetch(`${RESOURCES_DIR}redirect.py?token=${token}`); fetch(`${RESOURCES_DIR}clean-stash.py`);", + "bare": "fetch('redirect.py');", + "prefix": "fetch('/wrong/fetch/api/resources/redirect.py');", + "suffix": "fetch('../resources/redirect.py2');", + "unknown": "fetch('../resources/redirect.py'); fetch('../resources/unknown.py');", + } + for name, source in sources.items(): + path = self.root / f"fetch/api/redirect/{name}.any.js" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("// META: global=window,worker\n" + source) + selected = enumerate_cases(self.root, dir_prefixes=("fetch",), any_js_global="both") + self.assertEqual([case.case_path for case in selected], [ + f"fetch/api/redirect/{name}.any.js?moli-wpt-any={realm}" + for name in ("absolute", "concat", "relative", "template") + for realm in ("dedicatedworker", "window") + ]) + + +if __name__ == "__main__": + unittest.main()