moli/moli-benchmark/tests/test_wpt_fetch_range_server.py

from __future__ import annotations

import json
import tempfile
import unittest
import uuid
from contextlib import ExitStack
from http.client import HTTPConnection, IncompleteRead
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


RESOURCE_DIR = "/fetch/range/resources/"


class FetchRangeFixtureTests(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
        ))
        self.server = self.stack.enter_context(WptFixtureServer(self.root))

    def request(self, resource="long-wav.py", *, method="GET", headers=(), query="", port=None):
        connection = HTTPConnection("127.0.0.1", port or self.server.port, timeout=3)
        self.stack.callback(connection.close)
        connection.putrequest(method, RESOURCE_DIR + resource + "?" + query,
                              skip_accept_encoding=True)
        for name, value in headers:
            connection.putheader(name, value)
        connection.endheaders()
        response = connection.getresponse()
        self.stack.callback(response.close)
        return response

    def take(self, key, *, port=None):
        response = self.request("stash-take.py", query=urlencode({"key": key}), port=port)
        self.assertEqual(response.status, 200)
        self.assertEqual(response.getheader("Content-Type"), "application/json")
        self.assertIsNone(response.getheader("Access-Control-Allow-Origin"))
        self.assertIsNone(response.getheader("Cache-Control"))
        return json.loads(response.read())

    def test_closed_ranges_preserve_wav_bytes_and_response_metadata(self):
        response = self.request(headers=[("Range", "bytes=0-10"), ("Origin", "http://other.test")])
        self.assertEqual(response.status, 206)
        self.assertEqual(response.getheader("Content-Type"), "audio/wav")
        self.assertEqual(response.getheader("Content-Length"), "11")
        self.assertEqual(response.getheader("Content-Range"), "bytes 0-10/2400000")
        self.assertEqual(response.getheader("Accept-Ranges"), "bytes")
        self.assertEqual(response.getheader("Cache-Control"), "no-cache")
        self.assertEqual(response.getheader("Access-Control-Allow-Origin"), "http://other.test")
        self.assertEqual(response.read(), b"RIFF\x24\x9f\x24\0WAV")
        response = self.request(headers=[("Range", "bytes=100-103")])
        self.assertEqual(response.status, 206)
        self.assertEqual(response.read(), b"\0" * 4)

    def test_open_ranges_and_zero_end_match_upstream(self):
        for value, length, content_range in [
            ("bytes=11-", "2399989", "bytes 11-2399999/2400000"),
            ("bytes=0-0", "2400000", "bytes 0-2399999/2400000"),
            ("bytes=0000000000011-0000000000111", "101", "bytes 11-111/2400000"),
        ]:
            with self.subTest(value=value):
                response = self.request(method="HEAD", headers=[("Range", value)])
                self.assertEqual(response.status, 206)
                self.assertEqual(response.getheader("Content-Length"), length)
                self.assertEqual(response.getheader("Content-Range"), content_range)
                self.assertEqual(response.read(), b"")

    def test_invalid_units_empty_and_multiple_ranges_produce_full_response(self):
        for value in ["", "foo", "foo=0-10", "bytes=0-1, 4-5", "BYTES=0-10"]:
            with self.subTest(value=value):
                response = self.request(method="HEAD", headers=[("Range", value)])
                self.assertEqual(response.status, 200)
                self.assertEqual(response.getheader("Content-Length"), "2400000")
                self.assertIsNone(response.getheader("Content-Range"))
        response = self.request(method="HEAD", headers=[("Range", "bytes=0-1"), ("range", "4-5")])
        self.assertEqual(response.status, 200)

    def test_stash_preserves_header_bytes_duplicates_and_first_query_value(self):
        key, ignored = str(uuid.uuid4()), str(uuid.uuid4())
        self.server.fetch_stash.put(key, "separate abort namespace")
        query = urlencode({"accept-encoding-key": key, "range-received-key": ignored})
        response = self.request(method="HEAD", query=query + "&accept-encoding-key=" + ignored,
                                headers=[("Range", "bytes=0-10"), ("Accept-Encoding", "gzip"),
                                         ("accept-encoding", "\xff")])
        self.assertEqual(response.status, 206)
        self.assertEqual(self.take(key.upper().replace("-", ""), port=self.server.alternate_port), "gzip, \xff")
        self.assertIsNone(self.take(key))
        self.assertEqual(self.take(ignored), "range-header-received")
        self.assertEqual(self.server.fetch_stash.take(key), "separate abort namespace")

    def test_stash_overwrites_encoding_and_only_records_nonempty_range(self):
        key, range_key = str(uuid.uuid4()), str(uuid.uuid4())
        query = urlencode({"accept-encoding-key": key, "range-received-key": range_key})
        self.request(method="HEAD", query=query, headers=[("Range", "bytes=1-2"), ("Accept-Encoding", "gzip")])
        self.request(method="HEAD", query=query, headers=[("Range", ""), ("Accept-Encoding", "identity")])
        self.assertEqual(self.take(key), "identity")
        self.assertEqual(self.take(range_key), "range-header-received")
        self.request(method="HEAD", query=query)
        self.assertEqual(self.take(key), "")
        self.assertIsNone(self.take(range_key))

    def test_preflight_rejection_does_not_record_headers(self):
        key = str(uuid.uuid4())
        response = self.request(method="OPTIONS", query=urlencode({"accept-encoding-key": key}),
                                headers=[("Origin", "http://other.test"), ("Accept-Encoding", "gzip")])
        self.assertEqual(response.status, 404)
        self.assertEqual(response.read(), b"Preflight not accepted")
        self.assertIsNone(response.getheader("Access-Control-Allow-Origin"))
        self.assertIsNone(self.take(key))

    def test_invalid_keys_and_suffix_range_keep_upstream_errors(self):
        for resource, query, headers in [
            ("stash-take.py", "", []),
            ("stash-take.py", "key=not-a-uuid", []),
            ("long-wav.py", "accept-encoding-key=not-a-uuid", []),
            ("long-wav.py", "", [("Range", "bytes=-10")]),
        ]:
            with self.subTest(resource=resource, query=query, headers=headers):
                self.assertEqual(self.request(resource, query=query, headers=headers).status, 500)

    def test_streaming_and_shutdown_do_not_buffer_full_wav(self):
        with WptFixtureServer(self.root) as server:
            response = self.request(port=server.port)
            self.assertEqual(response.status, 200)
            self.assertEqual(response.getheader("Content-Length"), "2400000")
            prefix = response.read(8044)
            self.assertEqual(prefix[:4], b"RIFF")
            self.assertEqual(prefix[8:16], b"WAVEfmt ")
            self.assertEqual(prefix[36:40], b"data")
            self.assertEqual(prefix[44:], b"\0" * 8000)
        # Keep the client open: server shutdown must interrupt the slow stream.
        with self.assertRaises(IncompleteRead):
            response.read()

    def test_non_get_methods_use_the_same_resource_handlers(self):
        for method in ("POST", "PUT", "PATCH", "DELETE", "CUSTOM"):
            with self.subTest(method=method):
                key = str(uuid.uuid4())
                response = self.request(method=method, headers=[("Range", "bytes=0-10")],
                                        query=urlencode({"range-received-key": key}))
                self.assertEqual((response.status, response.read()), (206, b"RIFF\x24\x9f\x24\0WAV"))
                response = self.request("stash-take.py", method=method, query=urlencode({"key": key}))
                self.assertEqual(json.loads(response.read()), "range-header-received")
                self.assertIsNone(self.take(key))

    def test_case_selection_only_allows_supported_range_resource_paths(self):
        bodies = {
            "fetch/range/general.any.js": "fetch('resources/long-wav.py');fetch('resources/stash-take.py')",
            "fetch/range/absolute.any.js": "fetch('/fetch/range/resources/stash-take.py')",
            "fetch/range/dotted.any.js": "fetch('./resources/long-wav.py')",
            "fetch/range/unknown.any.js": "fetch('resources/other.py')",
            "fetch/range/suffix.any.js": "fetch('resources/long-wav.py2')",
            "fetch/range/parent.any.js": "fetch('../resources/long-wav.py')",
            "fetch/other/general.any.js": "fetch('resources/long-wav.py')",
        }
        for path, body in bodies.items():
            target = self.root / path
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_text("// META: global=window,worker\n" + body)
        cases = enumerate_cases(self.root, dir_prefixes=("fetch",), any_js_global="both")
        self.assertEqual([case.case_path for case in cases], [
            f"fetch/range/{name}.any.js?moli-wpt-any={realm}"
            for name in ["absolute", "dotted", "general"]
            for realm in ["dedicatedworker", "window"]
        ])