"""HTTP surface + signaling socket tests. No browser required.""" from __future__ import annotations import json import re import shutil import subprocess from pathlib import Path import pytest from fastapi.testclient import TestClient from starlette.websockets import WebSocketDisconnect from mirror.access import parse_cidrs from mirror.app import create_app from mirror.config import Settings STATIC = Path(__file__).resolve().parent.parent / "src/mirror/static" def gated(*allow, client_address="203.0.113.9", **kwargs): """A client speaking as ``client_address`` to a server restricted to ``allow``.""" settings = Settings(allow_cidrs=parse_cidrs(",".join(allow)), **kwargs) return TestClient(create_app(settings), client=(client_address, 44444)) EXTERNAL_REF = re.compile(r"""(?:src|href)\s*=\s*["']https?://""", re.IGNORECASE) def handshake(socket, role, name=""): socket.send_json({"type": "hello", "role": role, "name": name}) message = socket.receive_json() assert message["type"] == "welcome", message return message["peerId"] def wait_for(socket, kind): """Read frames until one has ``type == kind``. The roster legitimately pushes ``stream_updated`` frames whenever the audience changes, so tests must not assume a one-message-per-action cadence. Any ``error`` frame seen on the way is a failure worth surfacing. """ seen = [] while True: message = socket.receive_json() if message.get("type") == kind: return message assert message.get("type") != "error", ( f"unexpected error while waiting for {kind}: {message}" ) seen.append(message) assert len(seen) < 25, f"never saw {kind!r}; got {seen}" def wait_for_update_with(socket, **stream_fields): """`wait_for` for `stream_updated` frames, matched on the stream payload. `stream_updated` is pushed for audience changes too, so the test must keep reading until the payload itself says what is being awaited (e.g. the audio flag actually flipped).""" seen = [] while True: message = socket.receive_json() if message.get("type") == "stream_updated" and all( message["stream"].get(key) == value for key, value in stream_fields.items() ): return message["stream"] assert message.get("type") != "error", f"unexpected error: {message}" seen.append(message) assert len(seen) < 25, f"never saw stream_updated with {stream_fields}; got {seen}" class TestPages: @pytest.mark.parametrize( ("path", "marker"), [ ("/", "Watch everything"), ("/mirror", "Share my screen"), ("/share", "Share my screen"), ("/watch", "tile-template"), ], ) def test_pages_render(self, client, path, marker): response = client.get(path) assert response.status_code == 200 assert response.headers["content-type"].startswith("text/html") assert marker in response.text def test_pages_are_not_cacheable(self, client): assert client.get("/watch").headers["cache-control"] == "no-store" def test_the_two_pages_use_the_media_apis_they_claim(self): share = (STATIC / "share.js").read_text() watch = (STATIC / "watch.js").read_text() assert "getDisplayMedia" in share assert "RTCPeerConnection" in share and "RTCPeerConnection" in watch assert "new WebSocket" in (STATIC / "net.js").read_text() def test_nothing_is_loaded_from_a_cdn(self): """The no-CDN rule is a hard product requirement, so assert it.""" offenders = [] for file in sorted(STATIC.iterdir()): text = file.read_text() if EXTERNAL_REF.search(text): offenders.append(file.name) if "@import" in text or "fonts.googleapis" in text: offenders.append(f"{file.name} (@import)") assert offenders == [] @pytest.mark.parametrize("asset", ["app.css", "net.js", "share.js", "watch.js"]) def test_static_assets_are_served(self, client, asset): assert client.get(f"/static/{asset}").status_code == 200 def test_unknown_page_is_a_404(self, client): assert client.get("/nope").status_code == 404 class TestClientScripts: """Guards for the hand-written ES modules: they must parse, and the ids they reach for must exist. Both of these have already bitten once.""" NODE = shutil.which("node") @pytest.mark.skipif(NODE is None, reason="node is not installed") @pytest.mark.parametrize("script", ["net.js", "share.js", "watch.js"]) def test_scripts_parse_as_es_modules(self, script, tmp_path): copy = tmp_path / "subject.mjs" copy.write_text((STATIC / script).read_text()) result = subprocess.run([self.NODE, "--check", str(copy)], capture_output=True, text=True) assert result.returncode == 0, f"{script}: {result.stderr}" @pytest.mark.parametrize( ("script", "page"), [("share.js", "mirror.html"), ("watch.js", "watch.html")], ) def test_every_element_the_script_reaches_for_exists(self, script, page): html = (STATIC / page).read_text() wanted = set(re.findall(r"""\$\(\"#([A-Za-z0-9_-]+)\"\)""", (STATIC / script).read_text())) missing = {name for name in wanted if f'id="{name}"' not in html} assert not missing, f"{script} looks up {sorted(missing)} which {page} does not render" def test_both_pages_load_the_stylesheet_and_their_module(self): for page, script in (("mirror.html", "share.js"), ("watch.html", "watch.js")): html = (STATIC / page).read_text() assert 'href="/static/app.css"' in html assert f'src="/static/{script}"' in html assert 'type="module"' in html def test_share_never_asks_for_an_exact_capture_size(self): """A hard size breaks pickers: `exact` fails or hides picker entries on displays/DPRs the browser cannot honour. The size request must stay a preference the browser can decline while still capturing.""" source = (STATIC / "share.js").read_text() assert not re.search(r"\bexact\s*:", source), ( "share.js must not use an `exact` constraint key" ) def test_server_and_client_agree_on_the_bitrate_default(self): """`/api/config` advertises the server's per-leg cap; a browser that cannot reach it falls back to the hardcoded literal in `net.js`. If the two drift, offline clients silently get a different picture quality and nothing fails. One number, two places, one test that notices.""" server = int( re.search( r"DEFAULT_MAX_BITRATE_BPS\s*=\s*(\d[\d_]*)", (STATIC.parent / "config.py").read_text(), ) .group(1) .replace("_", "") ) client = int(re.search(r"maxBitrateBps:\s*(\d+)", (STATIC / "net.js").read_text()).group(1)) assert server == client, ( f"the server default ({server}) and the net.js fallback ({client}) " "have drifted: a client that cannot reach /api/config would use " "a different bitrate silently" ) def test_sender_quality_is_reapplied_on_negotiation_and_connection(self): """`setParameters` is per-negotiation: a re-offer (a surface switch, a viewer that re-joined) resets the encodings. The cap must therefore be re-applied after `setLocalDescription` *and* when the leg becomes `connected`. Both call sites are asserted so a future edit cannot drop one and silently soften the wall.""" source = (STATIC / "share.js").read_text() assert source.count('"maintain-resolution"') == 1, ( "the degradation preference must be a single, named decision" ) assert "async function applySenderQuality" in source def section(start: str, stop: str) -> str: begin = source.index(start) return source[begin : source.index(stop, begin)] negotiate = section("async function negotiate(viewerId)", "async function onSignal") assert "applySenderQuality" in negotiate, ( "the cap is not re-applied after setLocalDescription" ) assert negotiate.index("setLocalDescription") < negotiate.index("applySenderQuality"), ( "the cap is applied before the encodings exist" ) connected = section( "pc.onconnectionstatechange = () => {", "for (const track of state.stream.getTracks())", ) assert re.search( r'connectionState\s*===\s*"connected".*applySenderQuality', connected, re.S ), "the cap is not re-applied when the leg becomes connected" def test_pick_surface_rethrows_cancellation_and_retries_only_no_audio_errors(self): """The audio-off retry must be precise. A cancelled picker (NotAllowedError) is a refusal — in Chrome it is also what unticking "Share tab audio" produces — and must be *rethrown*: retrying would silently re-open the picker and share a surface the person meant not to. Only the errors that genuinely mean "this source has no audio" (a failed audio *constraint*, not a failed capture) may drop the audio requirement; a bare retry in a catch-all would hide real capture failures and turn refusals into silent shares.""" source = (STATIC / "share.js").read_text() # pickSurface and its named-error set live between these markers; the # first `catch` in that span is pickSurface's own. begin = source.index("async function pickSurface") section = source[begin : source.index("function publish", begin)] catch = section[section.index("catch") :] assert "NotAllowedError" in catch, "the catch block must name the refusal error" assert re.search(r"NotAllowedError[\s\S]*?throw\s+err", catch), ( "a cancelled picker must be rethrown, not retried" ) for name in ( "NotSupportedError", "OverconstrainedError", "NotFoundError", "TypeError", ): assert name in section, f"the audio-off retry must name {name}" assert "return await navigator.mediaDevices.getDisplayMedia" not in catch, ( "the audio-off retry must be conditional on the named errors, " "not a bare retry in a catch-all" ) assert re.search(r"audiolessError\(err\)\s*\)\s*throw\s+err", section), ( "an error the named set does not cover must surface, not be retried" ) def test_no_bare_publish_frame_wipes_the_roster(self): """The wire's `publish` is a full-metadata frame. A bare `{ type: "publish" }` frame parses with the model's defaults and would wipe width/height/audio off the roster on every surface switch (the mute/unmute republish path) — the panel and the wall would disagree from a single event. The full frame is the only publish share.js may send.""" source = (STATIC / "share.js").read_text() assert source.count('type: "publish"') == 1, ( "share.js must send exactly one publish frame (the full one); " "a bare { type: 'publish' } frame would reset the metadata" ) def test_the_two_pages_share_no_element_ids_but_the_shared_chrome(self): """Both pages consume one stylesheet, so a duplicated id means a colliding rule. This is not hypothetical: a `#stage` on both pages once made the share page's preview inherit `display: none` from the watch page's overlay. """ shared = {"net", "notice", "toast"} ids = { page: set(re.findall(r'id="([A-Za-z0-9_-]+)"', (STATIC / page).read_text())) for page in ("mirror.html", "watch.html", "index.html") } assert ids["mirror.html"] & ids["watch.html"] == shared assert ids["index.html"] & ids["mirror.html"] <= shared | {"livebar"} for page, found in ids.items(): assert len(found) == len(set(found)), f"{page} declares an id twice" def test_videos_are_muted_and_inline_so_autoplay_is_not_blocked(self): for page in ("mirror.html", "watch.html"): for tag in re.findall(r"]*>", (STATIC / page).read_text()): assert "muted" in tag, f"{page}: {tag} would be blocked by autoplay policy" assert "playsinline" in tag, f"{page}: {tag} would hijack iOS into fullscreen" class TestStylesheet: def test_hidden_always_means_gone(self): """Every component that sets its own `display` outranks the UA's `[hidden]`. Without this rule the share page shows the LIVE badge and the Stop button while nothing is being shared, and every wall tile offers an Exit button it is not on stage for. """ css = (STATIC / "app.css").read_text() assert re.search(r"\[hidden\]\s*\{[^}]*display:\s*none\s*!important", css), ( "app.css must declare a global `[hidden] { display: none !important }`" ) # Components toggled with `hidden` in JS. All but `.placeholder` set # their own `display`, which would outrank the UA's `[hidden]` without # the global rule; `.placeholder` is guarded because its block is the # escape hatch that keeps its content inside the stage. for selector in (".btn", ".livetag", ".icon-btn", ".toast", ".net", ".placeholder"): assert selector in css, f"{selector} is toggled by `hidden`; it lives in app.css" def test_no_rule_reintroduces_a_conflicting_display_for_hidden(self): """Every rule whose selector mentions `[hidden]` must actually hide.""" css = re.sub(r"/\*.*?\*/", "", (STATIC / "app.css").read_text(), flags=re.S) offenders = [ selector.strip() for selector, body in re.findall(r"([^{}]+)\{([^{}]*)\}", css) if "[hidden]" in selector and not re.search(r"display\s*:\s*none", body) ] assert offenders == [] def test_a_hidden_control_is_not_revealed_only_by_hover(self): """Kiosk rule: a control hidden by default must be reachable without a pointer. The tile's controls (`.actions`) start at `opacity: 0`. If the only rule that raises them to full opacity were a `:hover` rule, a mouseless kiosk would have no mute/expand controls at all. So while any `.actions` rule hides them, the stylesheet must also contain a `.actions` reveal rule with a non-hover trigger — keyboard focus or the selection state. """ css = re.sub(r"/\*.*?\*/", "", (STATIC / "app.css").read_text(), flags=re.S) actions_rules = [ (selector.strip(), body) for selector, body in re.findall(r"([^{}]+)\{([^{}]*)\}", css) if ".actions" in selector ] hiding = [ selector for selector, body in actions_rules if re.search(r"opacity\s*:\s*0(?![0-9])", body) ] non_hover_reveals = [ selector for selector, body in actions_rules if re.search(r"opacity\s*:\s*1(?![0-9])", body) and re.search(r":focus-visible|:focus-within|data-selected", selector) ] if hiding: assert non_hover_reveals, ( f".actions starts hidden ({sorted(hiding)}) but no rule reveals it " "without a pointer — a mouseless kiosk would have no tile controls" ) def _rules(self) -> list[tuple[str, str]]: r"""app.css as ``(selector, body)`` pairs, comments stripped. The flat ``[^{}]+\{[^{}]*\}`` pass treats an ``@media`` header as a selector and the first child's header as its body; the child rules that follow still come back as their own pairs, which is all these guards need — every assertion is about a top-level block (``.stage``, ``.stage > *``) or about properties that must sit in the *same* block (``overflow``/``aspect-ratio``/``min-height`` together). """ css = re.sub(r"/\*.*?\*/", "", (STATIC / "app.css").read_text(), flags=re.S) return [(sel.strip(), body) for sel, body in re.findall(r"([^{}]+)\{([^{}]*)\}", css)] def test_the_share_stage_is_one_cell(self): """The stage's children overlap on purpose. In plain flow an idle ``