mirror of
https://github.com/lexmount/moli.git
synced 2026-09-28 08:01:37 +00:00
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.
This commit is contained in:
ldm0
@@ -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:
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user