starred/moli
1
0
Fork 0
mirror of https://github.com/lexmount/moli.git synced 2026-10-05 08:00:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(wpt): serve XHR body and encoded document fixtures

Browse Source 
Model XHR binary body and Content-Type echo responses plus six encoded document resources. Preserve raw request metadata and encoding bytes, avoid waiting for unused uploads, and discover only supported resource references. Match upstream cache and absent-body response headers.

Validation: 584 Python tests passed with 32 workers; 270 HTTP responses matched official wptserve for status, body, and 15 response fields. New regression tests fail on the unmodified main baseline.

Source-commit: 5389231812
This commit is contained in:
ldm0
2026-09-23 14:19:42 +08:00
parent 71c46d6606
commit cb337baa6c
3 changed files with 245 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
moli-benchmark/moli_benchmark/wpt_cross/case_set.py
+4 -1
View File
@@ -836,7 +836,10 @@ def _supported_wptserve_handler_references(
def _xhr_response_handler_reference_patterns(directory: str) -> tuple[re.Pattern[str], ...]:
references = []
for name in ("status.py", "last-modified.py", "corsenabled.py",
"inspect-headers.py", "echo-headers.py"):
"inspect-headers.py", "echo-headers.py", "content.py",
"echo-content-type.py", "win-1252-xml.py", "win-1252-html.py",
"invalid-utf8-html.py", "shift-jis-html.py", "img-utf8-html.py",
"empty-div-utf8-html.py"):
resource = f"xhr/resources/{name}"
relative = posixpath.relpath(resource, directory)
references.extend(("/" + resource, relative, "./" + relative))
moli-benchmark/moli_benchmark/wpt_cross/server.py
+68 -2
View File
@@ -76,7 +76,22 @@ http.client._MAXHEADERS = 512
DEFAULT_TESTHARNESS_TIMEOUT_SECONDS = 10.0
MAX_REQUEST_BODY_BYTES = 16 * 1024 * 1024
MAX_REQUEST_BODY_LINE_BYTES = 64 * 1024
XHR_DOCUMENT_FIXTURES = {
"/xhr/resources/win-1252-xml.py": ("application/xml;charset=windows-1252", b"<\xff/>"),
# The upstream handler returns a Unicode string, encoded by wptserve as UTF-8.
"/xhr/resources/win-1252-html.py": ("text/html;charset=windows-1252", b"\xc3\xbf"),
"/xhr/resources/invalid-utf8-html.py": ("text/html;charset=utf-8", b"\xff"),
"/xhr/resources/shift-jis-html.py": ("text/html;charset=shift-jis", b"\x83e\x83X\x83g"),
"/xhr/resources/img-utf8-html.py": ("text/html;charset=utf-8", b"<img>foo"),
"/xhr/resources/empty-div-utf8-html.py": ("text/html;charset=utf-8", b"<!DOCTYPE html><div></div>"),
}
XHR_BODY_RESOURCE_PATHS = {
"/xhr/resources/content.py",
"/xhr/resources/echo-content-type.py",
*XHR_DOCUMENT_FIXTURES,
}
XHR_RESPONSE_RESOURCE_PATHS = {
*XHR_BODY_RESOURCE_PATHS,
"/xhr/resources/inspect-headers.py",
"/xhr/resources/echo-headers.py",
"/xhr/resources/corsenabled.py",
@@ -1819,7 +1834,7 @@ def _make_handler(
return
if unquote(urlparse(self.path).path) in {
"/xhr/resources/inspect-headers.py", "/xhr/resources/echo-headers.py",
}:
} | XHR_BODY_RESOURCE_PATHS:
self._serve_xhr_response_resource()
return
parsed = urlparse(self.path)
@@ -2600,11 +2615,61 @@ def _make_handler(
)
return True
def _serve_xhr_body_resource(self, parsed, *, emit_body: bool) -> None:
path = unquote(parsed.path)
upload_consumed = False
try:
if path == "/xhr/resources/content.py":
params = parse_qs(parsed.query, keep_blank_values=True, encoding="latin-1")
if "content" in params:
body = params["content"][0].encode("latin-1")
else:
body = self._read_content_length_request_body()
if body is None:
return
upload_consumed = True
content_type = "text/plain"
if "response_charset_label" in params:
content_type += ";charset=" + params["response_charset_label"][0]
headers = [
("Content-Type", content_type),
("X-Request-Method", self.command),
("X-Request-Query", parsed.query or "NO"),
("X-Request-Content-Length", self.headers.get("Content-Length", "NO")),
("X-Request-Content-Type", self.headers.get("Content-Type", "NO")),
]
elif path == "/xhr/resources/echo-content-type.py":
headers = [("Content-Type", "text/plain")]
body = self.headers.get("Content-Type", "").encode("latin-1")
else:
content_type, body = XHR_DOCUMENT_FIXTURES[path]
headers = [("Content-Type", content_type)]
except (ValueError, KeyError, OverflowError, OSError):
self.send_error(500)
return
# Unused uploads must not delay an early response or be parsed as
# another request. Consumed uploads can finish normally.
if path == "/xhr/resources/echo-content-type.py" or not upload_consumed and (
self.headers.get("Transfer-Encoding") is not None
or self.headers.get("Content-Length", "0").strip() not in {"", "0"}
):
self.close_connection = True
headers.append(("Connection", "close"))
self._send_bytes(
None, body, emit_body=emit_body, extra_headers=headers, cache_control=None,
auto_content_length=(
path != "/xhr/resources/echo-content-type.py" or "Content-Type" in self.headers
),
)
def _serve_xhr_response_resource(self, *, emit_body: bool = True) -> bool:
parsed = urlparse(self.path)
path = unquote(parsed.path)
if path not in XHR_RESPONSE_RESOURCE_PATHS:
return False
if path in XHR_BODY_RESOURCE_PATHS:
self._serve_xhr_body_resource(parsed, emit_body=emit_body)
return True
if path == "/xhr/resources/corsenabled.py":
self._serve_xhr_cors_echo(parsed, emit_body=emit_body)
return True
@@ -2935,6 +3000,7 @@ def _make_handler(
status_code: int = 200,
status_text: str | None = None,
cache_control: str | None = "no-store",
auto_content_length: bool = True,
) -> None:
content_type, extra_headers = _response_content_type_and_extra_headers(
content_type,
@@ -2944,7 +3010,7 @@ def _make_handler(
header_block = _static_response_header_block(content_type, extra_headers)
for name, value in header_block:
self.send_header(name, value)
if not _headers_include(header_block, "Content-Length"):
if auto_content_length and not _headers_include(header_block, "Content-Length"):
self.send_header("Content-Length", str(len(body)))
if cache_control is not None:
self.send_header("Cache-Control", cache_control)
moli-benchmark/tests/test_wpt_xhr_body_server.py
+173
View File
@@ -0,0 +1,173 @@
from __future__ import annotations
import tempfile
import unittest
from contextlib import ExitStack
from http.client import HTTPConnection
from pathlib import Path
from unittest.mock import patch
from moli_benchmark.wpt_cross.case_set import enumerate_cases
from moli_benchmark.wpt_cross.server import WptFixtureServer
class XhrBodyFixtureTests(unittest.TestCase):
def setUp(self) -> None:
self.stack = ExitStack()
self.addCleanup(self.stack.close)
self.root = Path(self.stack.enter_context(tempfile.TemporaryDirectory()))
(self.root / "resources").mkdir()
(self.root / "resources/testharness.js").write_text("// testharness")
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, path: str, *, method: str = "GET",
headers: tuple[tuple[str, str], ...] = (), body: bytes = b"",
) -> tuple[int, dict[str, str], bytes]:
connection = HTTPConnection("127.0.0.1", port, timeout=2)
try:
connection.putrequest(method, "/xhr/resources/" + path)
for name, value in headers:
connection.putheader(name, value)
connection.endheaders(body)
response = connection.getresponse()
return (
response.status,
{name.lower(): value for name, value in response.getheaders()},
response.read(),
)
finally:
connection.close()
def test_content_echoes_binary_upload_and_request_metadata_for_all_methods(self) -> None:
server = self.server()
payload = b"\x00\xff\xc3\xa9\r\n"
for method in ("GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "YO", "FOO"):
with self.subTest(method=method):
status, headers, body = self.request(
server.port, "content.py?raw=%FF+%20", method=method,
headers=(("Content-Length", str(len(payload))),
("Content-Type", "application/custom; charset=ascii")),
body=payload,
)
self.assertEqual(status, 200)
self.assertEqual(body, b"" if method == "HEAD" else payload)
self.assertEqual(headers["content-length"], str(len(payload)))
self.assertEqual(headers["content-type"], "text/plain")
self.assertNotIn("cache-control", headers)
self.assertEqual(headers["x-request-method"], method)
self.assertEqual(headers["x-request-query"], "raw=%FF+%20")
self.assertEqual(headers["x-request-content-length"], str(len(payload)))
self.assertEqual(headers["x-request-content-type"], "application/custom; charset=ascii")
def test_content_distinguishes_missing_and_empty_headers(self) -> None:
server = self.server()
for request_headers, expected in (
((), "NO"),
((("Content-Type", ""),), ""),
((("content-TYPE", "caf\xe9 \t"),), "caf\xe9 \t"),
((("Content-Type", "first"), ("content-type", "second")), "first"),
):
with self.subTest(headers=request_headers):
status, headers, body = self.request(server.port, "content.py", headers=request_headers)
self.assertEqual((status, body), (200, b""))
self.assertEqual(headers["x-request-query"], "NO")
self.assertEqual(headers["x-request-content-length"], "NO")
self.assertEqual(headers["x-request-content-type"], expected)
def test_content_query_override_uses_first_byte_values_and_preserves_charset_label(self) -> None:
server = self.server()
for query, expected_type, expected_body in (
("content=%00%FF+a&content=ignored&response_charset_label=windows-1252", "text/plain;charset=windows-1252", b"\x00\xff a"),
("content=&content=ignored&response_charset_label=&response_charset_label=UTF-8", "text/plain;charset=", b""),
("content=x&response_charset_label=UTF-8&response_charset_label=ignored", "text/plain;charset=UTF-8", b"x"),
):
with self.subTest(query=query):
status, headers, body = self.request(server.port, "content.py?" + query)
self.assertEqual((status, body), (200, expected_body))
self.assertEqual(headers["content-type"], expected_type)
self.assertEqual(headers["x-request-query"], query)
def test_content_reads_upload_before_reusing_connection(self) -> None:
server = self.server()
connection = HTTPConnection("127.0.0.1", server.port, timeout=2)
self.addCleanup(connection.close)
for payload in (b"first", b"second"):
connection.request("POST", "/xhr/resources/content.py", payload)
response = connection.getresponse()
self.assertEqual(response.status, 200)
self.assertEqual(response.read(), payload)
def test_echo_content_type_preserves_header_value_and_closes_connection(self) -> None:
server = self.server()
for value in (None, "", "Application/JSON; charset=UTF-8", "caf\xe9 \t"):
with self.subTest(value=value):
status, headers, body = self.request(
server.port, "echo-content-type.py", method="POST",
headers=() if value is None else (("Content-Type", value),),
)
self.assertEqual((status, body), (200, (value or "").encode("latin-1")))
self.assertEqual(headers["content-type"], "text/plain")
self.assertEqual(headers["connection"], "close")
self.assertNotIn("cache-control", headers)
self.assertEqual(
headers.get("content-length"),
None if value is None else str(len(value.encode("latin-1"))),
)
def test_early_responses_do_not_wait_for_overridden_or_unused_body(self) -> None:
server = self.server()
for path, expected in (
("content.py?content=override", b"override"),
("echo-content-type.py", b"application/test"),
("empty-div-utf8-html.py", b"<!DOCTYPE html><div></div>"),
):
with self.subTest(path=path):
status, headers, body = self.request(
server.port, path, method="POST",
headers=(("Content-Length", "1000000"), ("Content-Type", "application/test")),
)
self.assertEqual((status, body), (200, expected))
self.assertEqual(headers["connection"], "close")
def test_document_fixtures_preserve_encoding_labels_and_exact_bytes(self) -> None:
server = self.server()
for path, expected_type, expected in (
("win-1252-xml.py", "application/xml;charset=windows-1252", b"<\xff/>"),
("win-1252-html.py", "text/html;charset=windows-1252", b"\xc3\xbf"),
("invalid-utf8-html.py", "text/html;charset=utf-8", b"\xff"),
("shift-jis-html.py", "text/html;charset=shift-jis", "テスト".encode("shift-jis")),
("img-utf8-html.py", "text/html;charset=utf-8", b"<img>foo"),
("empty-div-utf8-html.py", "text/html;charset=utf-8", b"<!DOCTYPE html><div></div>"),
):
for method in ("GET", "HEAD"):
with self.subTest(path=path, method=method):
status, headers, body = self.request(server.port, path, method=method)
self.assertEqual((status, body), (200, b"" if method == "HEAD" else expected))
self.assertEqual(headers["content-type"], expected_type)
self.assertEqual(headers["content-length"], str(len(expected)))
self.assertNotIn("cache-control", headers)
def test_case_selection_recognizes_body_fixtures_without_matching_unhandled_paths(self) -> None:
sources = {
"xhr/content.window.js": "fetch('resources/content.py');",
"xhr/echo.window.js": "fetch('/xhr/resources/echo-content-type.py');",
"xhr/nested/doc.window.js": "fetch('../resources/shift-jis-html.py');",
"xhr/unknown.window.js": "fetch('resources/content.py'); fetch('resources/unknown.py');",
"xhr/suffix.window.js": "fetch('resources/content.py2');",
"xhr/wrong.window.js": "fetch('/other/resources/echo-content-type.py');",
}
for path, source in sources.items():
target = self.root / path
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(source)
self.assertEqual([case.case_path for case in enumerate_cases(self.root, dir_prefixes=("xhr",))], [
"xhr/content.window.js?moli-wpt-script=window",
"xhr/echo.window.js?moli-wpt-script=window",
"xhr/nested/doc.window.js?moli-wpt-script=window",
])