mirror of
https://github.com/lexmount/moli.git
synced 2026-10-05 16:00:54 +00:00
fix(wpt): serve service worker import and update fixtures
Model service worker script redirects, versioned imports and update response sequences, including file-based and installation updates. Preserve request methods, cache metadata, raw script bytes and per-resource state shared across fixture origins.
This commit is contained in:
ldm0
@@ -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"(?<![A-Za-z0-9_./-]){re.escape(reference)}"
|
||||
rf"{WPTSERVE_HANDLER_TRAILING_BOUNDARY}"
|
||||
)
|
||||
for reference in references
|
||||
)
|
||||
|
||||
|
||||
def _supported_wptserve_handler_references(
|
||||
rel: str | None,
|
||||
) -> 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:
|
||||
|
||||
@@ -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<prefix>\bsrc\s*=\s*)(?P<quote>['\"])"
|
||||
@@ -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:
|
||||
|
||||
@@ -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(
|
||||
'<!doctype html><script src="/resources/testharness.js"></script>'
|
||||
'<script src="/resources/testharnessreport.js"></script>'
|
||||
f'<script src="{reference}?Key=ignored"></script>'
|
||||
)
|
||||
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()
|
||||
Reference in New Issue
Block a user