"""The layout as a file a partner can import, generated from the one we run ourselves. Structural checks only — whether OBS *accepts* the file is a thing only OBS can answer, and that verification belongs to the first import on a partner's box. What is pinned here is that nothing is lost or invented on the way out: every scene, every source, every URL and every transform. """ from __future__ import annotations import json import sys from pathlib import Path import pytest _LIVE = Path(__file__).resolve().parents[2] / "live" if str(_LIVE) not in sys.path: sys.path.insert(0, str(_LIVE)) import obs_export as EX # noqa: E402 PRODUCER = "https://trap-v3-producer-example.run.app" @pytest.fixture(scope="module") def layout() -> dict: return json.loads(EX.LAYOUT.read_text(encoding="utf-8")) @pytest.fixture(scope="module") def variables(layout) -> dict: return EX.partner_vars(layout, {"producer": PRODUCER, "event": "trwc-2026"}) @pytest.fixture(scope="module") def doc(layout, variables) -> dict: """What a partner is handed: everything they can reach. The venue wall's panels come from the race office and are not theirs to render, so that scene is left out.""" return EX.scene_collection(layout, variables, name="ViewRegatta", skip=("Video wall",)) def _scenes(doc): return {s["name"]: s for s in doc["sources"] if s["id"] == "scene"} def _inputs(doc): return {s["name"]: s for s in doc["sources"] if s["id"] != "scene"} # --- nothing is lost on the way out -------------------------------------------------------------- def _exportable(layout) -> list[dict]: """The scenes a partner is actually handed — everything except the ones that play a file held on our own broadcast machine (see `test_a_machine_local_scene_is_dropped_whole_and_named`).""" dropped = EX.machine_local_scenes(layout) return [s for s in layout["scenes"] if s["name"] not in dropped] def test_every_scene_and_source_survives(layout, variables): doc = EX.scene_collection(layout, variables) scenes = _exportable(layout) assert set(_scenes(doc)) == {s["name"] for s in scenes} named = {it["source"] for s in scenes for it in s["items"] if it.get("source")} assert set(_inputs(doc)) == named assert [s["name"] for s in doc["scene_order"]] == [s["name"] for s in scenes] def test_the_programme_scene_is_the_one_on_air(layout, doc): assert doc["current_program_scene"] == layout["program"] == "Broadcast 3D" def test_every_transform_is_carried_across(layout, variables): doc = EX.scene_collection(layout, variables) scenes = _scenes(doc) for s in _exportable(layout): items = scenes[s["name"]]["settings"]["items"] assert len(items) == len(s["items"]) for src, out in zip(s["items"], items, strict=True): t = src.get("transform") or {} assert out["name"] == src["source"] assert out["pos"] == {"x": float(t.get("positionX", 0)), "y": float(t.get("positionY", 0))} assert out["bounds"] == {"x": float(t.get("boundsWidth", 0)), "y": float(t.get("boundsHeight", 0))} assert out["bounds_type"] == EX._BOUNDS[t["boundsType"]] def test_a_machine_local_scene_is_dropped_whole_and_named(layout, variables): """A source that plays a FILE cannot be handed over: the material lives where it is produced. It has to be the whole SCENE, not just the source, and the export has to SAY so. `_input_source` writes a browser source and nothing else, so a media source forced through it arrives as a browser source with no URL — a permanently black rectangle wearing a name the partner cannot fix, because what it wants is a folder on our machine. And a scene list one short is a fact you were told or a fault you discover; the sheet is where it gets told.""" dropped = EX.machine_local_scenes(layout) assert dropped.get("Break roll") == ["break:roll"] assert "Break line-up" not in dropped # a pure browser source: a partner can render it doc = EX.scene_collection(layout, variables) assert "Break roll" not in _scenes(doc) assert "break:roll" not in _inputs(doc) # Nothing was left behind wearing browser defaults with nowhere to point. assert all(s["settings"].get("url") for s in _inputs(doc).values()) sheet = EX.setup_sheet(layout, variables) assert "Left out of this export" in sheet assert "Break roll" in sheet and "break:roll" in sheet def test_sources_are_shared_between_scenes_not_duplicated(layout, variables): """The wall borrows the programme page and each boat camera. A second copy would be a second feed connection for an identical page — the thing the layout is arranged to avoid.""" doc = EX.scene_collection(layout, variables) wall = _scenes(doc)["Video wall"]["settings"]["items"] borrowed = {i["name"] for i in wall} assert "programme" in borrowed and "boatcam:slot:A" in borrowed assert len(_inputs(doc)) < sum(len(s["items"]) for s in layout["scenes"]), "shared, not copied" names = [s["name"] for s in doc["sources"]] assert len(names) == len(set(names)), "a source is declared exactly once" # --- the URLs a partner actually gets ------------------------------------------------------------- def test_every_url_points_at_the_partners_producer(doc): for src in _inputs(doc).values(): url = src["settings"].get("url", "") if url: assert url.startswith(PRODUCER), url assert "${" not in url, f"an unsubstituted variable reached a partner: {url}" def test_the_cameras_ask_for_a_slot_not_a_boat(doc): cams = {n: s["settings"]["url"] for n, s in _inputs(doc).items() if n.startswith("boatcam:slot:")} assert len(cams) == 4 for letter, url in zip("ABCD", [cams[f"boatcam:slot:{c}"] for c in "ABCD"], strict=True): assert f"slot={letter}" in url assert "boat=" not in url, "a hull id here would go stale at the first rotation" def test_a_partner_gets_one_origin_because_cloud_run_is_http2(layout): """Our rehearsal rig splits `camorigin` and `mediabase` onto a second origin to stay under the ~6-connections-per-origin limit of HTTP/1.1. A partner on Cloud Run has HTTP/2, where the limit does not apply, and their clips arrive as signed GCS URLs — already another origin.""" v = EX.partner_vars(layout, {"producer": PRODUCER}) assert v["camorigin"] == PRODUCER, "not the rehearsal rig's 127.0.0.1" assert v["mediabase"] == "" assert v["readbase"] == EX.PRODUCTION_READ_BASE doc = EX.scene_collection(layout, v, skip=("Video wall",)) for src in _inputs(doc).values(): assert "127.0.0.1" not in src["settings"].get("url", "") def test_an_unset_variable_removes_its_parameter_rather_than_leaving_it_empty(doc): """`?mediaBase=` is not the same as no `mediaBase`: the page would read the empty string and use it instead of falling back to the default a partner on Cloud Run wants.""" for src in _inputs(doc).values(): url = src["settings"].get("url", "") assert "mediaBase=" not in url, url assert not url.endswith(("?", "&", "=")) assert EX.clean_url("https://x/p?a=1&b=&c=2") == "https://x/p?a=1&c=2" assert EX.clean_url("https://x/p?a=") == "https://x/p" assert EX.clean_url("https://x/p") == "https://x/p" def test_an_override_still_wins(layout): v = EX.partner_vars(layout, {"producer": PRODUCER, "camorigin": "https://cams.example", "readbase": "https://read.staging.example"}) assert v["camorigin"] == "https://cams.example" assert v["readbase"] == "https://read.staging.example" # --- the settings that fail silently when wrong --------------------------------------------------- def test_pages_are_not_shut_down_or_reloaded_on_a_cut(doc): """A page that reloads on every cut spends a second black and drops its feed connection.""" for src in _inputs(doc).values(): assert src["settings"]["shutdown"] is False assert src["settings"]["restart_when_active"] is False def test_only_the_programme_page_routes_its_audio(doc): """All continuing programme sound comes out of the produced picture. A boat camera routing audio too would duplicate that complete mix.""" routed = [n for n, s in _inputs(doc).items() if s["settings"].get("reroute_audio")] assert routed == ["programme"] def test_complete_programme_audio_remains_active_under_every_telemetry_camera(layout): """Changing among our own pictures must never deactivate the programme-audio source. Borrowing the same input as the bottom scene item keeps commentary, countdowns, ambience and every other existing programme sound alive; the opaque camera above it replaces only the picture. Break scenes intentionally remain separate. """ racing_pictures = { "Race cam", "Boat cam A", "Boat cam B", "Boat cam C", "Boat cam D", "Start vessel", "Finish vessel", } by_name = {scene["name"]: scene for scene in layout["scenes"]} for name in racing_pictures: assert by_name[name]["items"][0]["source"] == "programme", name assert "settings" not in by_name[name]["items"][0], "reuse the one source; never clone its voice" for name in ("Break line-up", "Break roll"): assert all(item.get("source") != "programme" for item in by_name[name]["items"]), name def test_a_routed_page_is_also_allowed_to_SPEAK(doc): """Routing a page's audio into the mixer does nothing if the page never opens its mouth. A `feed=live` page gates its whole commentary scheduler on a gesture that unlocks the voice (`app.js` `commentaryReady`), and OBS's embedded Chromium never delivers one — so the stream went out with a perfect picture and no sound at all (Ulf 2026-08-11) while every OBS-side check read clean. `trustedAutoplayAudio=1` is the page's declaration that its renderer was launched with autoplay permission, which OBS's CEF is. The two settings are one decision, so they are asserted together: a source that routes audio must be allowed to produce some.""" for name, s in _inputs(doc).items(): if s["settings"].get("reroute_audio"): assert "trustedAutoplayAudio=1" in s["settings"]["url"], ( f"{name} routes audio into the mixer but is not permitted to speak") def test_a_page_that_is_not_routed_never_claims_autoplay_audio(doc): """The other half. A boat-cam page granted the same permission would speak the SAME commentary into the room's default output — audible in the hall, absent from the stream, and duplicated on any box that captures desktop audio.""" for name, s in _inputs(doc).items(): if not s["settings"].get("reroute_audio"): assert "trustedAutoplayAudio" not in s["settings"].get("url", ""), ( f"{name} claims autoplay audio but does not route it to the mixer") # --- leaving a scene out -------------------------------------------------------------------------- def test_a_skipped_scene_takes_its_orphaned_sources_with_it(layout, variables): """A partner who does not drive the venue wall should not be handed its panels — nor a dangling source referencing a race-office URL they have no access to.""" doc = EX.scene_collection(layout, variables, skip=("Video wall",)) assert "Video wall" not in _scenes(doc) assert "panel:results" not in _inputs(doc) assert "programme" in _inputs(doc), "still used by Broadcast 3D" def test_skipping_the_programme_scene_repoints_what_is_on_air(layout, variables): doc = EX.scene_collection(layout, variables, skip=("Broadcast 3D",)) assert doc["current_program_scene"] in _scenes(doc) # --- the setup sheet ------------------------------------------------------------------------------ def test_the_setup_sheet_states_the_canvas_and_every_url(layout, variables): sheet = EX.setup_sheet(layout, variables) assert "1920×1080" in sheet for name, src in _inputs(EX.scene_collection(layout, variables)).items(): assert f"`{name}`" in sheet if src["settings"].get("url"): assert src["settings"]["url"] in sheet assert "${" not in sheet assert "Only the scene on programme renders" in sheet def test_the_cli_writes_both_files(tmp_path): out, sheet = tmp_path / "c.json", tmp_path / "s.md" rc = EX.main(["--var", f"producer={PRODUCER}", "--var", "event=trwc-2026", "--out", str(out), "--sheet", str(sheet), "--skip-scene", "Video wall"]) assert rc == 0 doc = json.loads(out.read_text(encoding="utf-8")) assert doc["name"] and doc["sources"] and doc["scene_order"] assert "Video wall" not in {s["name"] for s in doc["sources"]} assert sheet.read_text(encoding="utf-8").startswith("# ViewRegatta") def test_a_collection_that_would_ship_a_localhost_url_is_refused(tmp_path): """THE MOST LIKELY WAY THIS GOES WRONG. The layout's defaults are our rehearsal rig's, so a forgotten `--var` is the ordinary mistake — and on a partner's machine it fails as a permanently blank browser source with nothing in any log. Refused at generation time instead.""" out = tmp_path / "c.json" with pytest.raises(SystemExit) as e: # no --var at all: every URL is localhost EX.main(["--out", str(out)]) assert "point at this machine" in str(e.value) assert "--skip-scene" in str(e.value), "and it says how to resolve it" assert not out.exists(), "nothing is written when it would be broken on arrival" # The race-office panels are the trap: `producer` is supplied, but `burgee` still is not. with pytest.raises(SystemExit) as e2: EX.main(["--var", f"producer={PRODUCER}", "--out", str(out)]) assert "panel:results" in str(e2.value) # Either resolve it or leave the scene out — both work. assert EX.main(["--var", f"producer={PRODUCER}", "--out", str(out), "--skip-scene", "Video wall"]) == 0 assert EX.main(["--var", f"producer={PRODUCER}", "--var", "burgee=https://burgee.example", "--var", "event=trwc-2026", "--out", str(out)]) == 0 def test_a_credential_in_a_url_is_refused_with_no_way_to_override(layout, tmp_path): """This file is e-mailed, zipped and forwarded, and `obs_scenes dump` writes URLs back into a layout meant to be committed — so a token in a query string reaches an inbox and a repository at once, and neither can be un-sent. There is deliberately no `--allow` for it.""" v = EX.partner_vars(layout, {"producer": f"{PRODUCER}/x?readToken=s3cret"}) doc = EX.scene_collection(layout, v, skip=("Video wall",)) assert EX.secret_urls(doc), "the probe sees it" out = tmp_path / "c.json" with pytest.raises(SystemExit) as e: EX.main(["--var", f"producer={PRODUCER}/x?readToken=s3cret", "--out", str(out), "--skip-scene", "Video wall", "--allow-localhost"]) assert "looks like a credential" in str(e.value) assert not out.exists() def test_the_credential_check_reads_parameter_names_not_prose(doc): """`?token=…` is a credential; a path containing the word is not. A guard that cried wolf on every URL would be turned off.""" assert EX.secret_urls(doc) == [], "the real layout is clean" assert EX.secret_urls({"sources": [{"name": "s", "settings": {"url": "https://x/tokens/list"}}]}) == [] # camelCase matters: `readToken` is what our own read token would be called in a URL, and a # substring matcher that missed it would be worse than none at all. for param in ("token", "readToken", "api_key", "apiKey", "signature", "VR_READ_TOKEN", "sig"): found = EX.secret_urls({"sources": [{"name": "s", "settings": {"url": f"https://x/p?{param}=v"}}]}) assert found, param for benign in ("slots", "mediaBase", "feed", "slot", "world", "lang"): clean = EX.secret_urls({"sources": [{"name": "s", "settings": {"url": f"https://x/p?{benign}=v"}}]}) assert clean == [], benign def test_a_local_rig_can_still_ask_for_localhost(tmp_path): out = tmp_path / "c.json" assert EX.main(["--out", str(out), "--allow-localhost"]) == 0 assert EX.local_urls(json.loads(out.read_text(encoding="utf-8"))), "our own rig, deliberately" if __name__ == "__main__": raise SystemExit(pytest.main([__file__, "-q"]))