"""The OBS layout file, and the command that makes OBS match it. `obs_scenes` had no tests, which was survivable while every source it could build was a browser source and the file it builds was checked by looking at OBS. It stopped being survivable when the layout gained a source that plays a FILE: the module's create path was written for one kind, and a kind it cannot build is not reported as unbuildable — it falls through to "not on this machine, cell left empty" and the scene comes up short with a line in a log nobody is reading at that moment. So what is pinned here is the seam rather than the wiring: every kind the shipped layout names is one `apply` can actually create, a borrowed source is never created empty, and `preflight` fails on the targets that go black while staying quiet about the one that is designed to fall through. No OBS instance. `FakeOBS` is an in-memory obs-websocket, and its only job is to record what `apply` asked for. """ from __future__ import annotations from copy import deepcopy import json import sys from pathlib import Path from types import SimpleNamespace import pytest _LIVE = Path(__file__).resolve().parents[2] / "live" if str(_LIVE) not in sys.path: sys.path.insert(0, str(_LIVE)) import obs_scenes as OS # noqa: E402 from director_obs import OBSController # noqa: E402 LAYOUT = _LIVE / "obs" / "stream_1080.json" @pytest.fixture(scope="module") def layout() -> dict: return json.loads(LAYOUT.read_text(encoding="utf-8")) # --- an in-memory OBS ---------------------------------------------------------------------------- class FakeOBS: """Just enough of obs-websocket v5 for `apply` and `dump` to run against.""" def __init__(self, *, scenes: dict[str, list[str]] | None = None, inputs: dict[str, dict] | None = None, program: str = "", profile: str = "Untitled", collection: str = "Streaming 1080p", canvas: tuple[int, int] = (1920, 1080)): self.scenes: dict[str, list[dict]] = {} self.inputs: dict[str, dict] = dict(inputs or {}) self.program, self.profile, self.collection, self.canvas = program, profile, collection, canvas self._next_id = 1 self.log: list[tuple] = [] for scene, sources in (scenes or {}).items(): self.scenes[scene] = [] for source in sources: self._place(scene, source) # reads --------------------------------------------------------------------------------------- def get_video_settings(self): return SimpleNamespace(base_width=self.canvas[0], base_height=self.canvas[1], fps_numerator=30, fps_denominator=1) def get_profile_list(self): return SimpleNamespace(current_profile_name=self.profile, profiles=[self.profile, "Spare"]) def get_scene_collection_list(self): return SimpleNamespace(current_scene_collection_name=self.collection, scene_collections=[self.collection]) def get_scene_list(self): # OBS lists top-first; obs_scenes reverses it so a file reads bottom-first. return SimpleNamespace(scenes=[{"sceneName": n} for n in reversed(list(self.scenes))]) def get_scene_item_list(self, scene): if scene not in self.scenes: raise RuntimeError(f"no scene {scene!r}") return SimpleNamespace(scene_items=[dict(it) for it in reversed(self.scenes[scene])]) def get_input_list(self): return SimpleNamespace(inputs=[{"inputName": n} for n in self.inputs]) def send(self, _request): # director_obs tries the raw request first raise RuntimeError("the fake has no raw send") def get_input_settings(self, name): if name not in self.inputs: raise RuntimeError(f"no input {name!r}") return SimpleNamespace(input_settings=dict(self.inputs[name]["settings"])) def get_current_program_scene(self): return SimpleNamespace(current_program_scene_name=self.program, scene_name=self.program) def get_scene_item_transform(self, scene, item_id): return SimpleNamespace(scene_item_transform={}) # writes -------------------------------------------------------------------------------------- def _place(self, scene, source, enabled=True) -> int: item_id = self._next_id self._next_id += 1 self.scenes.setdefault(scene, []).append( {"sourceName": source, "sceneItemId": item_id, "sceneItemEnabled": bool(enabled), "inputKind": (self.inputs.get(source) or {}).get("kind", "browser_source")}) return item_id def create_scene(self, name): self.scenes.setdefault(name, []) self.log.append(("create_scene", name)) def create_input(self, scene, name, kind, settings, enabled): self.inputs[name] = {"kind": kind, "settings": dict(settings)} self._place(scene, name, enabled) self.log.append(("create_input", scene, name, kind, dict(settings))) def create_scene_item(self, scene, source, enabled=True): if source not in self.inputs: # What OBS does: you cannot place a source that does not exist on this machine. raise RuntimeError(f"no input named {source!r}") item_id = self._place(scene, source, enabled) self.log.append(("create_scene_item", scene, source)) return SimpleNamespace(scene_item_id=item_id) def set_input_settings(self, name, settings, overlay=True): cur = self.inputs.setdefault(name, {"kind": "browser_source", "settings": {}}) cur["settings"] = ({**cur["settings"], **settings} if overlay else dict(settings)) self.log.append(("set_input_settings", name, dict(settings))) def set_scene_item_transform(self, scene, item_id, transform): self.log.append(("set_transform", scene, item_id)) def set_scene_item_index(self, scene, item_id, index): self.log.append(("set_index", scene, item_id, index)) def set_scene_item_enabled(self, scene, item_id, enabled): self.log.append(("set_enabled", scene, item_id, bool(enabled))) def remove_scene_item(self, scene, item_id): self.scenes[scene] = [it for it in self.scenes[scene] if it["sceneItemId"] != item_id] self.log.append(("remove_scene_item", scene, item_id)) def set_current_program_scene(self, name): self.program = name self.log.append(("set_current_program_scene", name)) def _apply(doc, fake=None, **kw): fake = fake or FakeOBS() acted = OS.apply(OBSController(client=fake), doc, log=lambda *_a: None, **kw) return fake, acted def _created(fake, name): return next((c for c in fake.log if c[0] == "create_input" and c[2] == name), None) # --- the shipped layout -------------------------------------------------------------------------- def test_every_kind_the_layout_names_is_one_apply_can_build(layout): """The check the break roll needed and nobody had. `apply` creates an input only for a kind it knows the settings of. A kind outside that table is not refused — it falls through to `create_scene_item`, which raises because the input does not exist, and the failure is swallowed as one log line saying the cell was left empty. On a race morning that is a scene that comes up one source short and says so nowhere anyone is looking.""" for scene in layout["scenes"]: for item in scene["items"]: if not (item.get("settings") or {}): continue # borrowed — covered by the next test kind = item.get("kind", "browser_source") assert kind in OS._INPUT_KEYS, ( f"{scene['name']} / {item['source']}: `apply` cannot create a {kind!r}") def test_the_settings_a_layout_carries_are_ones_the_module_would_keep(layout): """A key `dump` does not record is a key that disappears on the next round-trip, so writing one into the file by hand produces a layout that quietly stops describing OBS.""" for scene in layout["scenes"]: for item in scene["items"]: settings = item.get("settings") or {} if not settings: continue known = set(OS._INPUT_KEYS[item.get("kind", "browser_source")]) assert set(settings) <= known, ( f"{scene['name']} / {item['source']}: {sorted(set(settings) - known)} would be " f"dropped by `dump`") def test_shipped_camera_scenes_preserve_the_complete_programme_audio_bus(layout): policy = layout["audio_continuity"] assert policy["source"] == "programme" assert set(policy["preserve_in_scenes"]) == { "Race cam", "Boat cam A", "Boat cam B", "Boat cam C", "Boat cam D", "Start vessel", "Finish vessel", # Not a camera, and deliberately in the set anyway: the series card is a GRAPHIC laid over a # race that is still being talked about, so cutting to it must not cut the commentary. The # two break scenes are the other case — a break owns its own sound. "Standings", } OS.validate_audio_continuity(layout) def test_apply_refuses_a_camera_scene_that_drops_the_programme_audio_bus(layout): broken = deepcopy(layout) race_cam = next(s for s in broken["scenes"] if s["name"] == "Race cam") race_cam["items"] = [item for item in race_cam["items"] if item["source"] != "programme"] with pytest.raises(ValueError, match="must contain 'programme' exactly once"): _apply(broken) def test_apply_refuses_a_programme_audio_bus_that_restarts_on_a_camera_cut(layout): broken = deepcopy(layout) programme = next( item for scene in broken["scenes"] for item in scene["items"] if item["source"] == "programme" and item.get("settings") ) programme["settings"]["restart_when_active"] = True with pytest.raises(ValueError, match="must not restart on scene activation"): _apply(broken) def test_apply_refuses_duplicate_audio_from_the_camera_picture(layout): broken = deepcopy(layout) race_cam = next(s for s in broken["scenes"] if s["name"] == "Race cam") panel = next(item for item in race_cam["items"] if item["source"] == "racecam:panel") panel["settings"]["reroute_audio"] = True with pytest.raises(ValueError, match="must not route duplicate audio"): _apply(broken) def test_a_borrowed_source_is_defined_before_it_is_borrowed(layout): """A scene item is a PLACEMENT; a source's settings live on its first appearance in the file. List the borrowing scene first and the definition can no longer claim the name — the video wall established the idiom, and the break roll's bed is the second use of it.""" defined: set[str] = set() for scene in layout["scenes"]: for item in scene["items"]: name = item["source"] if item.get("settings"): assert name not in defined, f"{name} is defined twice; the second one wins silently" defined.add(name) else: assert name in defined, ( f"{scene['name']} borrows {name!r} before anything defines it") def test_the_break_scenes_are_one_page_and_one_file(layout): scenes = {s["name"]: s for s in layout["scenes"]} assert "Break line-up" in scenes and "Break roll" in scenes # One line-up PAGE, placed twice — a second copy would be a second feed connection. beds = [it for s in layout["scenes"] for it in s["items"] if it["source"] == "break:bed"] assert len(beds) == 2 assert sum(1 for it in beds if it.get("settings")) == 1 roll = next(it for it in scenes["Break roll"]["items"] if it["source"] == "break:roll") assert roll["kind"] == "ffmpeg_source" # The three that make a file source safe on air rather than merely working. assert roll["settings"]["close_when_inactive"] is True # the syncer can replace the file assert roll["settings"]["clear_on_media_end"] is True # it ends on the bed, not a frozen frame assert roll["settings"]["looping"] is False # a repeating roll says the day stopped # The bed is UNDER the roll: item order is bottom-first, so the bed must come first. order = [it["source"] for it in scenes["Break roll"]["items"]] assert order == ["break:bed", "break:roll"] def test_the_emergency_scenes_have_no_network_or_browser_dependency(layout): scenes = {s["name"]: s for s in layout["scenes"]} card_scene = scenes["Technical interruption"] cached_scene = scenes["Emergency last video"] card = card_scene["items"][0] assert card["source"] == "emergency:technical" assert card["kind"] == "ffmpeg_source" assert card["settings"]["local_file"] == "${emergencydir}/technical_hold.mp4" assert card["settings"]["looping"] is True assert card["settings"]["close_when_inactive"] is False # already decoded when the emergency cut lands assert not any(it.get("kind") == "browser_source" for it in card_scene["items"]) order = [it["source"] for it in cached_scene["items"]] assert order == ["emergency:technical", "emergency:last-good"] cached = cached_scene["items"][1] assert cached["kind"] == "ffmpeg_source" assert cached["settings"]["close_when_inactive"] is True # cache writer can replace it off-air assert cached["settings"]["clear_on_media_end"] is True # the card beneath always wins over black def test_emergency_only_manifest_cannot_rewrite_normal_production_sources(): manifest = Path(__file__).resolve().parents[2] / "live" / "obs" / "emergency_1080.json" doc = json.loads(manifest.read_text(encoding="utf-8")) assert [scene["name"] for scene in doc["scenes"]] == [ "Technical interruption", "Emergency last video", ] assert "program" not in doc items = [item for scene in doc["scenes"] for item in scene["items"]] assert {item["source"] for item in items} == {"emergency:technical", "emergency:last-good"} assert not any(item.get("kind") == "browser_source" for item in items) def test_the_standings_only_manifest_can_be_applied_to_an_on_air_collection(): """The additive file exists because the full layout cannot be applied to a live OBS. Measured against a streaming box: a dry run of `stream_1080.json` reported twelve source rewrites, `programme` among them — the deployed producer's URL carried a resolution floor and a camera-rotation set the file does not. So this file must name ONE scene, must not be able to move the canvas, must not cut the programme, and must never carry settings for the source that holds the picture and the voice. """ manifest = Path(__file__).resolve().parents[2] / "live" / "obs" / "standings_1080.json" doc = json.loads(manifest.read_text(encoding="utf-8")) assert [scene["name"] for scene in doc["scenes"]] == ["Standings"] assert "profile" not in doc, "the canvas belongs to the profile; this must not move it" assert "program" not in doc, "it adds a scene, it never cuts to one" items = doc["scenes"][0]["items"] assert [item["source"] for item in items] == ["programme", "results:standings"] assert "settings" not in items[0], "programme is BORROWED — never re-pointed from here" # And it describes the same scene as the full layout, or an operator would get a different card # depending on which file was applied. full = json.loads(LAYOUT.read_text(encoding="utf-8")) same = next(s for s in full["scenes"] if s["name"] == "Standings") assert [i["source"] for i in same["items"]] == [i["source"] for i in items] assert same["items"][1]["settings"] == items[1]["settings"] assert full["vars"]["standingsbase"] == doc["vars"]["standingsbase"] def test_refresh_reloads_only_the_sources_the_file_defines(): """`--refresh` must never reload a BORROWED source. A borrowed item is another scene's source placed here, so reloading it reloads it everywhere. On the additive standings manifest the borrowed source is `programme` — the one carrying the picture and the commentary that are on air — and a refresh meant for a results card would have blacked the programme for the length of a page load. It also stops the full layout reloading one page once per scene that borrows it. """ refreshed: list[str] = [] class RecordingController(OBSController): def refresh_browser_input(self, source): # noqa: ANN001 refreshed.append(source) fake = FakeOBS(scenes={"Broadcast 3D": ["programme"]}, inputs={"programme": {"kind": "browser_source", "settings": {"url": "http://x/index.html"}}}, program="Broadcast 3D") doc = { "collection": "Streaming 1080p", "canvas": [1920, 1080], "scenes": [{"name": "Standings", "items": [ {"source": "programme", "enabled": True}, {"source": "results:standings", "kind": "browser_source", "enabled": True, "settings": {"url": "http://x/core/web/standings.html"}}, ]}], } OS.apply(RecordingController(client=fake), doc, refresh=True, log=lambda *_a: None) assert refreshed == ["results:standings"], "the borrowed programme must not be reloaded" def test_partial_manifest_restores_program_if_obs_selects_a_new_scene(): class SelectingOBS(FakeOBS): def create_scene(self, name): super().create_scene(name) self.program = name fake = SelectingOBS(scenes={"Broadcast 3D": []}, program="Broadcast 3D") doc = { "collection": "Streaming 1080p", "canvas": [1920, 1080], "scenes": [{"name": "Technical interruption", "items": []}], } fake, acted = _apply(doc, fake) assert fake.program == "Broadcast 3D" assert "restore program scene -> 'Broadcast 3D'" in acted # --- apply --------------------------------------------------------------------------------------- DOC_MEDIA = { "vars": {"breakdir": "/vr/break"}, "scenes": [{"name": "Break roll", "items": [ {"source": "break:roll", "kind": "ffmpeg_source", "enabled": True, "settings": {"local_file": "${breakdir}/break_roll.mp4", "is_local_file": True, "looping": False, "clear_on_media_end": True}}, ]}], } def test_apply_creates_a_media_source_with_its_path_expanded(): fake, acted = _apply(DOC_MEDIA) call = _created(fake, "break:roll") assert call is not None, "the media source was never created" _, scene, _, kind, settings = call assert (scene, kind) == ("Break roll", "ffmpeg_source") assert settings["local_file"] == "/vr/break/break_roll.mp4" # A file has no viewport. Width/height are the browser's, and handing them to a media source is # a setting OBS ignores sitting in a file that claims to describe the layout. assert "width" not in settings and "height" not in settings assert any("create media source" in a for a in acted) def test_apply_still_merges_the_canvas_into_a_browser_source(): doc = {"vars": {"producer": "http://p"}, "canvas": [1920, 1080], "scenes": [{"name": "S", "items": [ {"source": "page", "kind": "browser_source", "settings": {"url": "${producer}/x.html"}}]}]} fake, _ = _apply(doc) _, _, _, _, settings = _created(fake, "page") assert settings == {"width": 1920, "height": 1080, "url": "http://p/x.html"} def test_a_borrowed_item_is_never_created_as_an_empty_source(): """An item with no settings is a borrowing. Creating it here would make a source pointing at nothing AND leave the real definition unable to claim the name — a black rectangle wearing a name that looks right.""" doc = {"scenes": [{"name": "Break roll", "items": [ {"source": "break:bed", "enabled": True, "transform": {"positionX": 0}}]}]} fake, _ = _apply(doc) assert _created(fake, "break:bed") is None assert "break:bed" not in fake.inputs def test_a_media_sources_settings_are_written_when_they_differ(): fake = FakeOBS(scenes={"Break roll": ["break:roll"]}, inputs={"break:roll": {"kind": "ffmpeg_source", "settings": {"local_file": "/old/break_roll.mp4"}}}) _apply(DOC_MEDIA, fake) wrote = [c for c in fake.log if c[0] == "set_input_settings"] assert wrote and wrote[0][2]["local_file"] == "/vr/break/break_roll.mp4" def test_an_unchanged_media_source_is_left_alone(): """`--dry-run` reporting a change for every source on every run is the same as reporting nothing: an operator cannot tell a real edit from the tool restating the file.""" fake = FakeOBS(scenes={"Break roll": ["break:roll"]}, inputs={"break:roll": {"kind": "ffmpeg_source", "settings": { "local_file": "/vr/break/break_roll.mp4", "is_local_file": True, "looping": False, "clear_on_media_end": True}}}) _, acted = _apply(DOC_MEDIA, fake) assert not [c for c in fake.log if c[0] == "set_input_settings"] assert acted == [] def test_dump_records_a_media_source_and_puts_the_variable_back(): fake = FakeOBS(scenes={"Break roll": ["break:roll"]}, inputs={"break:roll": {"kind": "ffmpeg_source", "settings": { "local_file": "/vr/break/break_roll.mp4", "looping": False, "some_key_we_do_not_manage": 7}}}, program="Break roll") doc = OS.dump(OBSController(client=fake), {"breakdir": "/vr/break"}) item = doc["scenes"][0]["items"][0] assert item["kind"] == "ffmpeg_source" assert item["settings"]["local_file"] == "${breakdir}/break_roll.mp4" assert "some_key_we_do_not_manage" not in item["settings"] # --- preflight ----------------------------------------------------------------------------------- def _doc_one_file(path: str, *, bed: bool) -> dict: items = [] if bed: items.append({"source": "break:bed", "enabled": True}) items.append({"source": "break:roll", "kind": "ffmpeg_source", "settings": {"local_file": path}}) return {"scenes": [{"name": "Break roll", "items": items}]} def test_preflight_accepts_a_file_that_is_there(tmp_path: Path): f = tmp_path / "break_roll.mp4" f.write_bytes(b"not really an mp4, but it is not empty") assert OS.preflight(_doc_one_file(str(f), bed=False), {}, log=lambda *_a: None) == [] def test_preflight_refuses_a_missing_file_with_nothing_beneath_it(tmp_path: Path): """The mistyped `--var breakdir=` case. Nothing under it, so it is a black rectangle.""" missing = str(tmp_path / "gone.mp4") assert OS.preflight(_doc_one_file(missing, bed=False), {}, log=lambda *_a: None) == [missing] def test_preflight_lets_a_missing_file_pass_when_something_is_beneath_it(tmp_path: Path): """The ordinary state of the break roll before the day's first clip is approved. The scene falls through to its bed, so refusing here would let an empty break folder block the boat cameras.""" said: list[str] = [] bad = OS.preflight(_doc_one_file(str(tmp_path / "gone.mp4"), bed=True), {}, log=said.append) assert bad == [] assert any("beneath it" in line for line in said), "it must still SAY the roll is not there" def test_preflight_refuses_a_file_that_is_there_but_empty(tmp_path: Path): empty = tmp_path / "break_roll.mp4" empty.touch() assert OS.preflight(_doc_one_file(str(empty), bed=False), {}, log=lambda *_a: None) == [str(empty)] def test_preflight_expands_the_variable_before_looking(tmp_path: Path): doc = _doc_one_file("${breakdir}/break_roll.mp4", bed=False) (tmp_path / "break_roll.mp4").write_bytes(b"x") assert OS.preflight(doc, {"breakdir": str(tmp_path)}, log=lambda *_a: None) == [] def test_preflight_checks_a_vlc_playlists_entries_too(tmp_path: Path): """Not what this layout uses, but a machine that has one is dumped and re-applied by the same command, and an absent folder is the same black rectangle.""" doc = {"scenes": [{"name": "S", "items": [ {"source": "playlist", "kind": "vlc_source", "settings": {"playlist": [{"value": str(tmp_path / "nope")}]}}]}]} assert OS.preflight(doc, {}, log=lambda *_a: None) == [str(tmp_path / "nope")] def test_target_names_what_a_source_points_at(): assert OS._target({"url": "http://p/x"}) == "http://p/x" assert OS._target({"local_file": "/vr/break/roll.mp4"}) == "/vr/break/roll.mp4" assert OS._target({"playlist": [{"value": "/vr/break"}]}) == "/vr/break" assert OS._target({}) == ""