test(wpt): substitute request headers in fixtures

moli-benchmark/moli_benchmark/wpt_cross/server.py

@@ -35,6 +35,7 @@ import sys
 import threading
 import time
 import uuid
+from collections.abc import Mapping
 from html import escape as html_escape
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
@@ -798,6 +799,10 @@ _TRICKLE_PIPE_RE = re.compile(r"(?:^|[|,])trickle\(d([0-9]+(?:\.[0-9]+)?)\)(?:$|
 _HEADER_PIPE_RE = re.compile(r"^header\(([^,()]+),([^()]*)\)$")
 _STATUS_PIPE_RE = re.compile(r"^status\(([0-9]{3})\)$")
 _GET_TEMPLATE_RE = re.compile(rb"\{\{GET\[([^\]\r\n]+)\]\}\}")
+_HEADER_OR_DEFAULT_TEMPLATE_RE = re.compile(
+    rb"\{\{header_or_default\(\s*([^,()]+?)\s*,\s*([^()]*)\)\}\}",
+    re.IGNORECASE,
+)
 _UUID_TEMPLATE_RE = re.compile(rb"\{\{\$([A-Za-z_][A-Za-z0-9_]*):uuid\(\)\}\}")
 _ID_TEMPLATE_RE = re.compile(rb"\{\{\$([A-Za-z_][A-Za-z0-9_]*)\}\}")
 _HTTP_TOKEN_RE = re.compile(r"^[!#$%&'*+\-.^_`|~0-9A-Za-z]+$")
@@ -1272,6 +1277,7 @@ def _static_response_headers(
     request_path: str = "/",
     request_hostname: str = "localhost",
     primary_hostname: str | None = None,
+    request_headers: Mapping[str, str] | None = None,
 ) -> list[tuple[str, str]]:
     headers = _apply_header_operations(
         _sidecar_response_headers(file_path),
@@ -1292,6 +1298,7 @@ def _static_response_headers(
                 request_path=request_path,
                 request_hostname=request_hostname,
                 primary_hostname=primary_hostname,
+                request_headers=request_headers,
                 template_ids=template_ids,
             ).decode("latin-1"),
         )
@@ -1337,6 +1344,7 @@ def _substitute_wpt_template_variables(
     request_path: str = "/",
     request_hostname: str = "localhost",
     primary_hostname: str | None = None,
+    request_headers: Mapping[str, str] | None = None,
     template_ids: dict[bytes, bytes] | None = None,
 ) -> bytes:
     if alternate_port is None:
@@ -1507,6 +1515,17 @@ def _substitute_wpt_template_variables(
     }
     for marker, value in replacements.items():
         body = body.replace(marker, value)
+    normalized_request_headers = {
+        name.lower(): value for name, value in (request_headers or {}).items()
+    }
+
+    def replace_header_or_default(match: re.Match[bytes]) -> bytes:
+        name = match.group(1).decode("ascii", errors="ignore").strip().lower()
+        default = match.group(2).decode("utf-8", errors="replace").strip()
+        value = normalized_request_headers.get(name, default)
+        return html_escape(value, quote=True).encode("utf-8")
+
+    body = _HEADER_OR_DEFAULT_TEMPLATE_RE.sub(replace_header_or_default, body)
     if template_ids is None:
         template_ids = {}
 
@@ -2143,6 +2162,7 @@ def _make_handler(
                     request_path=path,
                     request_hostname=_host_header_hostname(self.headers.get("Host")),
                     primary_hostname=primary_hostname,
+                    request_headers=self.headers,
                 )
             static_header_context = {
                 "port": int(
@@ -2187,6 +2207,7 @@ def _make_handler(
                         _host_header_hostname(self.headers.get("Host")),
                     )
                 ),
+                "request_headers": self.headers,
             }
 
             def static_headers() -> list[tuple[str, str]]:

moli-benchmark/tests/test_wpt_cross.py

@@ -5223,6 +5223,25 @@ test(() => {}, "ok");
             b' var missing = "";',
         )
 
+    def test_fixture_server_substitutes_request_header_or_default(self) -> None:
+        body = (
+            b"present={{header_or_default(Referer, missing)}}"
+            b" absent={{header_or_default(X-Absent, <missing>)}}"
+            b" empty={{header_or_default(X-Empty, )}}"
+        )
+
+        self.assertEqual(
+            _substitute_wpt_template_variables(
+                body,
+                port=12345,
+                request_headers={
+                    "referer": "https://example.test/path?a=1&b=2",
+                },
+            ),
+            b"present=https://example.test/path?a=1&amp;b=2"
+            b" absent=&lt;missing&gt; empty=",
+        )
+
     def test_fixture_server_substitutes_template_variables_in_sidecar_headers(self) -> None:
         with tempfile.TemporaryDirectory() as root:
             fixture = Path(root) / "frame-ancestors.sub.html"