mirror of
https://github.com/lexmount/moli.git
synced 2026-09-28 16:01:39 +00:00
ldm0
ebc016ace3
Serve common/echo.py as HTML with byte-preserving first query values, all-method handling, and ordered response pipes. Recognize exact relative and absolute references during case discovery.
Validation: 582 Python tests passed with 32 workers; 144 HTTP responses matched official wptserve for status, body, and 15 response fields. New regression tests fail on the unmodified main baseline.
Source-commit: 4042e1a11a
171 lines
8.7 KiB
Python
171 lines
8.7 KiB
Python
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 urllib.parse import urlencode
|
|
|
|
from moli_benchmark.wpt_cross.case_set import enumerate_cases
|
|
from moli_benchmark.wpt_cross.server import WptFixtureServer
|
|
|
|
|
|
ECHO_PATH = "/common/echo.py"
|
|
|
|
|
|
class CommonEchoFixtureTests(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.root / "common").mkdir()
|
|
(self.root / "common/echo.py").write_text("# Python source must not be served")
|
|
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: str, *, method: str = "GET", path: str = ECHO_PATH, body=None):
|
|
connection = HTTPConnection("127.0.0.1", self.server.port, timeout=2)
|
|
try:
|
|
connection.request(method, path + "?" + query, body, {"Origin": "https://caller.test"})
|
|
response = connection.getresponse()
|
|
return response.status, response.headers, response.read()
|
|
finally:
|
|
connection.close()
|
|
|
|
def test_content_is_html_and_preserves_query_bytes_and_first_value(self) -> None:
|
|
cases = (
|
|
("content=%3Cscript%3EglobalThis.answer%3D42%3C%2Fscript%3E", b"<script>globalThis.answer=42</script>"),
|
|
("content=first&content=second", b"first"),
|
|
("content=&content=second", b""),
|
|
("content", b""),
|
|
("content=A+B%2BC%26D%3DE", b"A B+C&D=E"),
|
|
("content=%FF%00%C3%A9%FE", b"\xff\x00\xc3\xa9\xfe"),
|
|
("co%6Etent=decoded-key", b"decoded-key"),
|
|
("content=%2520;%invalid", b"%20;%invalid"),
|
|
("content={{host}}", b"{{host}}"),
|
|
)
|
|
for query, expected in cases:
|
|
for method in ("GET", "HEAD"):
|
|
with self.subTest(query=query, method=method):
|
|
status, headers, body = self.request(query, method=method)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(headers.get_all("Content-Type"), ["text/html"])
|
|
self.assertEqual(headers["X-XSS-Protection"], "0")
|
|
self.assertEqual(headers["Content-Length"], str(len(expected)))
|
|
self.assertEqual(body, b"" if method == "HEAD" else expected)
|
|
self.assertIsNone(headers["Access-Control-Allow-Origin"])
|
|
self.assertIsNone(headers["Cache-Control"])
|
|
|
|
def test_all_methods_read_query_content_instead_of_request_body(self) -> None:
|
|
for method in ("POST", "OPTIONS", "PUT", "PATCH", "DELETE", "YO", "chicken"):
|
|
with self.subTest(method=method):
|
|
status, headers, body = self.request(
|
|
"content=query", method=method, body=b"content=upload",
|
|
)
|
|
self.assertEqual((status, body), (200, b"query"))
|
|
self.assertEqual(headers["Content-Type"], "text/html")
|
|
self.assertIsNone(headers["Access-Control-Allow-Origin"])
|
|
|
|
def test_unused_uploads_do_not_delay_the_response(self) -> None:
|
|
for method in ("POST", "PUT", "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, ECHO_PATH + "?content=ready")
|
|
connection.putheader(*framing)
|
|
connection.endheaders()
|
|
response = connection.getresponse()
|
|
self.assertEqual((response.status, response.read()), (200, b"ready"))
|
|
self.assertEqual(response.headers["Connection"], "close")
|
|
finally:
|
|
connection.close()
|
|
|
|
def test_missing_content_is_an_error_and_other_paths_are_not_echo_handlers(self) -> None:
|
|
for method in ("GET", "HEAD", "POST"):
|
|
for query in ("", "Content=wrong-case"):
|
|
with self.subTest(method=method, query=query):
|
|
self.assertEqual(self.request(query, method=method)[0], 500)
|
|
for path in (ECHO_PATH + "2", "/wrong" + ECHO_PATH, "/COMMON/echo.py"):
|
|
with self.subTest(path=path):
|
|
self.assertEqual(self.request("content=not-an-echo", path=path)[0], 404)
|
|
|
|
def test_response_pipes_apply_after_the_handler(self) -> None:
|
|
query = urlencode({
|
|
"content": "{{host}}",
|
|
"pipe": "sub(none)|status(201)|header(Content-Type,text/plain)|header(X-XSS-Protection,1)",
|
|
})
|
|
for method in ("GET", "HEAD", "POST", "chicken"):
|
|
with self.subTest(method=method):
|
|
status, headers, body = self.request(query, method=method)
|
|
self.assertEqual(status, 201)
|
|
self.assertEqual(headers.get_all("Content-Type"), ["text/plain"])
|
|
self.assertEqual(headers.get_all("X-XSS-Protection"), ["1"])
|
|
self.assertEqual(headers["Content-Length"], "9")
|
|
self.assertEqual(body, b"" if method == "HEAD" else b"localhost")
|
|
with patch("moli_benchmark.wpt_cross.server.time.sleep") as sleep:
|
|
status, headers, body = self.request(urlencode({"content": "delayed", "pipe": "trickle(d0.01)"}))
|
|
self.assertEqual((status, body), (200, b"delayed"))
|
|
self.assertIsNone(headers["Content-Length"])
|
|
self.assertEqual(headers["Cache-Control"], "no-cache, no-store, must-revalidate")
|
|
self.assertEqual(headers["Pragma"], "no-cache")
|
|
self.assertEqual(headers["Expires"], "0")
|
|
sleep.assert_called_once_with(0.01)
|
|
for command, expected, pragma, expires in (
|
|
("trickle(d0)|header(Cache-Control,private,true)",
|
|
["no-cache, no-store, must-revalidate", "private"], "no-cache", "0"),
|
|
("header(Cache-Control,private)|trickle(d0)", ["private"], None, None),
|
|
("header(Pragma,no-cache)|trickle(d0)", None, "no-cache", None),
|
|
("header(Expires,0)|trickle(d0)", None, None, "0"),
|
|
):
|
|
with self.subTest(command=command):
|
|
status, headers, body = self.request(urlencode({"content": "ok", "pipe": command}))
|
|
self.assertEqual((status, body), (200, b"ok"))
|
|
self.assertEqual(headers.get_all("Cache-Control"), expected)
|
|
self.assertEqual(headers["Pragma"], pragma)
|
|
self.assertEqual(headers["Expires"], expires)
|
|
for method in ("GET", "POST", "chicken"):
|
|
self.assertEqual(self.request("content=not-served&pipe=unknown", method=method)[0], 500)
|
|
|
|
def test_discovery_recognizes_only_references_to_the_common_echo_handler(self) -> None:
|
|
directory = "html/browsers/echo-tests"
|
|
cases = {
|
|
"absolute.html": (ECHO_PATH, True),
|
|
"relative.html": ("../../../common/echo.py", True),
|
|
"dot-relative.html": ("./../../../common/echo.py", True),
|
|
"bare.html": ("echo.py", False),
|
|
"wrong-relative.html": ("../../common/echo.py", False),
|
|
"wrong-root.html": ("/other/common/echo.py", False),
|
|
"suffix.html": (ECHO_PATH + ".extra", False),
|
|
}
|
|
for name, (reference, _) in cases.items():
|
|
path = self.root / directory / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(
|
|
'<script src="/resources/testharness.js"></script>'
|
|
f'<script>fetch("{reference}?content=hello")</script>'
|
|
)
|
|
(self.root / directory / "unknown.html").write_text(
|
|
'<script src="/resources/testharness.js"></script>'
|
|
f'<script>fetch("{ECHO_PATH}?content=x"); fetch("/unsupported.py")</script>'
|
|
)
|
|
(self.root / directory / "script.window.js").write_text(
|
|
f'const echoURL = content => `{ECHO_PATH}?content=${{encodeURIComponent(content)}}`;'
|
|
)
|
|
discovered = enumerate_cases(self.root, dir_prefixes=(directory,))
|
|
self.assertEqual(
|
|
sorted(case.case_path.split("?")[0] for case in discovered),
|
|
sorted([directory + "/" + name for name, (_, allowed) in cases.items() if allowed]
|
|
+ [directory + "/script.window.js"]),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|