"""The boat-cam media map, and the rule that there is only ONE of it. Two producers find a boat's clips in genuinely different ways — a database read plus a signed URL, or `clips.json` over REST — and must publish the identical shape, because one renderer consumes it. When that shape was written out twice, the two did not disagree so much as diverge in existence: the REST producer had no media path at all for months, so the mode a render PC, a partner and a guest stream all run published an empty map while the phones filmed normally. Every tile drew "(clip unavailable for this window)" and nothing said why. The last test here is the one that matters: the same clips through both producers, asserted byte identical. It is what a third producer gets for free and what a fourth cannot opt out of. """ from __future__ import annotations import sys from pathlib import Path _LIVE = Path(__file__).resolve().parents[2] / "live" sys.path.insert(0, str(_LIVE)) from media_map import media_map, media_rows # noqa: E402 def test_rows_are_seconds_from_the_track_start_in_clip_order(): rows = media_rows([(10_000, "c.mp4"), (4_000, "a.mp4"), (7_000, "b.mp4")], 4_000) assert [r["tSec"] for r in rows] == [0, 3, 6] assert [r["file"] for r in rows] == ["a.mp4", "b.mp4", "c.mp4"] assert all(r["video"] is True for r in rows) def test_a_clips_position_on_the_timeline_keeps_its_milliseconds(): """The renderer seeks into a clip by `cursor − tSec`, so a whole-second floor here put the picture up to a second from the instant its own burned-in clock claimed — and by a different amount per boat, since each camera's clips start on their own fraction of a second. Two tiles at the same moment then disagreed by up to a second, for ever. The filename carries the millisecond and the device burns it into the frame; this was the one place it was discarded.""" assert media_rows([(1_000_500, "a.mp4")], 1_000_000)[0]["tSec"] == 0.5 assert media_rows([(1_000_120, "a.mp4")], 1_000_000)[0]["tSec"] == 0.12 def test_two_clips_half_a_second_apart_do_not_land_on_the_same_second(): rows = media_rows([(1_000_000, "a.mp4"), (1_000_500, "b.mp4")], 1_000_000) assert [r["tSec"] for r in rows] == [0.0, 0.5] def test_a_clip_that_predates_the_track_lands_at_the_start_rather_than_before_it(): """A device buffers while it has no signal and uploads on reconnect, so a clip stamped just before the race began is ordinary. Clamping keeps it at the head of the timeline; a negative `tSec` would put it somewhere the renderer's cursor never reaches.""" assert media_rows([(9_000, "early.mp4")], 10_000)[0]["tSec"] == 0 def test_an_unusable_clip_is_dropped_never_defaulted(): """`tSec: 0` for a clip whose timing could not be read would stack every such clip onto the gun — the tile would then play the wrong footage, confidently, at the most-watched moment of the race. Absent is the honest answer.""" rows = media_rows([(None, "x.mp4"), (1_000, None), (1_000, ""), (True, "b.mp4"), (2_000, "ok.mp4")], 0) assert [r["file"] for r in rows] == ["ok.mp4"], "a row with no usable start or URL survived" def test_a_boat_with_no_usable_clip_is_omitted_not_emptied(): """Absent and empty mean different things to the tile — "no footage is known for this hull" versus "there is footage and none of it covers now" — and it tells the viewer a different thing in each case. The map must not blur them.""" m = media_map({"b1": [(0, "a.mp4")], "b2": [], "b3": [(None, None)]}, 0) assert set(m) == {"b1"} # --- the rule: both producers, one shape ---------------------------------------------------------- def test_the_store_producer_and_the_rest_producer_agree_exactly(): """The same clips, through both media paths, must come out identical. This is the test the REST producer's absence would have failed on the day it was written, and it is the reason the mapping now lives in one module: the two differ in how they FIND clips, and in nothing else.""" import serve_v3_live as S from rest_source import RestBackend track_start = 1_000_000 starts = [1_010_000, 1_004_000, 1_007_000] class _Seg: def __init__(self, start_ms: int, path: str) -> None: self.start_ms, self.gcs_path = start_ms, path class _Store: def slots_of_boat(self, _boat): return ["slot:1"] # The bounds mirror the real stores' signature (`store.py`, `postgres_store.py`). A fake # that accepts less than the real thing cannot fail when a caller starts passing more. def video_segments_for(self, _slot, *, since_ms=None, until_ms=None): return [_Seg(s, f"gs://bucket/{s}.mp4") for s in starts if (since_ms is None or s >= since_ms) and (until_ms is None or s <= until_ms)] from_store = S.media_from_store( _Store(), ["b1"], track_start, clip_url=lambda seg: f"https://signed.example/{Path(seg.gcs_path).name}") src = RestBackend("https://signed.example", on_log=lambda _m: None) src.get = lambda _path, **_q: {"clips": [ {"startMs": s, "url": f"https://signed.example/{s}.mp4"} for s in starts]} from_rest = src.media("race-1", ["b1"], track_start) assert from_store == from_rest, "the two producers publish different media maps" assert [r["tSec"] for r in from_store["b1"]] == [4, 7, 10] # --- the two faults that only show up minutes into a race ------------------------------------------ def test_the_store_producer_reads_this_races_window_not_the_whole_day(): """The tick runs once a second and the segment table grows all day, so an unbounded read makes the producer's own cadence decay as the day goes on. That cadence is the frontier every renderer presents against — a slow producer is the condition the presentation clock exists to absorb, and there is no reason to manufacture it. The window is the SAME one the REST producer already asks `clips.json` for, so the two still find the same clips.""" import serve_v3_live as S from media_map import PRE_RACE_CLIP_MARGIN_MS asked: list = [] track_start = 5_000_000 class _Store: def slots_of_boat(self, _boat): return ["slot:1"] def video_segments_for(self, _slot, *, since_ms=None, until_ms=None): asked.append((since_ms, until_ms)) return [] until = track_start + 1_200_000 S.media_from_store(_Store(), ["b1"], track_start, clip_url=lambda _s: "x.mp4", since_ms=max(0, track_start - PRE_RACE_CLIP_MARGIN_MS), until_ms=until) assert asked == [(track_start - PRE_RACE_CLIP_MARGIN_MS, until)], \ "the store producer read outside this race's clip window" # BOTH ends must reach the QUERY. Bounding only the near one leaves it open forwards, and a # camera keeps filming long after the race it was pointed at: measured on `seatrial-r2`, a # 19-minute race fetched 9843 rows to keep 228 — two minutes of a fourteen-second export. asked.clear() S.media_from_store(_Store(), ["b1"], track_start, clip_url=lambda _s: "x.mp4", since_ms=1_000) assert asked == [(1_000, None)], "a live caller has no far end yet and must be unaffected" def test_the_rest_producer_asks_for_the_whole_race_not_the_last_few_seconds(): """`clips.json` defaults to a 20-SECOND window — right for a guest viewer polling for the newest clip, wrong for a producer building a race's timeline. Left at the default the map held a rolling ~20 s, so its clip count plateaued; and because the feed was republished only when that count GREW, the renderer kept the last map it was handed while the presentation cursor walked off the end of it. Measured: the tiles played for a minute or two and then went dark for the rest of the race.""" from rest_source import PRE_RACE_CLIP_MARGIN_MS, RestBackend asked: list[dict] = [] src = RestBackend("http://h", on_log=lambda _m: None) src.get = lambda _path, **q: asked.append(q) or {"clips": []} track_start = 1_000_000_000 src.media("race-1", ["b1"], track_start) assert asked and "since" in asked[0], "no `since` — the read falls back to a 20 s window" assert asked[0]["since"] == track_start - PRE_RACE_CLIP_MARGIN_MS, \ "the window must cover the race, with room for clips buffered before the track began" def test_the_feed_is_republished_when_the_clip_set_CHANGES_not_when_it_grows(): """A clip set does not only grow: the read plane answers over a window and a local uploader sweeps past its retention, so an hour in the map is a rolling set of flat size and moving contents. Comparing counts with `>` stops firing at the plateau and the picture freezes on a map whose clips are all in the past.""" from serve_v3_live import _media_key a = {"b1": [{"tSec": 10, "file": "a.mp4", "video": True}, {"tSec": 12, "file": "b.mp4", "video": True}]} slid = {"b1": [{"tSec": 12, "file": "b.mp4", "video": True}, {"tSec": 14, "file": "c.mp4", "video": True}]} assert len(a["b1"]) == len(slid["b1"]), "the premise: same size, different clips" assert _media_key(a) != _media_key(slid), \ "a window that slid without changing size would never be republished" assert _media_key(a) == _media_key({"b1": list(a["b1"])}), "an unchanged map must not republish" # --- the two line cameras are channels of their own ----------------------------------------------- def test_the_vessel_cameras_are_named_by_ROLE_and_built_from_the_account_name(): """A race-committee vessel can never be a `cameraSlot`: that is a position in the racing fleet (A..D, derived from the teams), and a vessel has no team, no sail number and no place in the order. The picture reaches it by role. The slot is built from the account NAME the QR was minted with, never from an owner column — those are `users.id`s, and comparing the two matches nothing without failing.""" from event_meta import overlay_event_config, vessel_camera_slots cfg = {"devices": {"vessels": {"start_stbd": "A", "finish_stbd": "B"}}} assert vessel_camera_slots(cfg, "trwc2026") == {"start": "trwc2026:buoy:A", "finish": "trwc2026:buoy:B"} meta = overlay_event_config({"x": 1}, cfg, owner="trwc2026") assert meta["vesselCameras"]["start"] == "trwc2026:buoy:A" assert meta["vesselRoles"] == ["start_stbd", "finish_stbd"] mirrored_finish = overlay_event_config( {}, {"devices": {"vessels": {"start_stbd": "A", "finish_port": "B"}}}, owner="trwc2026", ) assert mirrored_finish["vesselRoles"] == ["start_stbd", "finish_port"] def test_an_event_with_no_vessels_publishes_no_vessel_cameras(): """A club Saturday whose line runs between two buoys has no vessel cameras — which is absence, not a camera that is switched off, and the picture must be able to tell those apart.""" from event_meta import overlay_event_config, vessel_camera_slots assert vessel_camera_slots({"devices": {}}, "gsys") == {} assert "vesselCameras" not in overlay_event_config({}, {"devices": {}}, owner="gsys") assert "vesselRoles" not in overlay_event_config({}, {"devices": {}}, owner="gsys") # …and with no owner there is no slot to build, so nothing is guessed. assert vessel_camera_slots({"devices": {"vessels": {"start_stbd": "A"}}}, None) == {} def test_the_owner_is_read_off_a_boat_ref_not_an_owner_column(): """`trwc2026:boat:7` → `trwc2026`. A bare ref with no prefix yields None rather than itself, because `7` is not an account and a slot built from it would belong to nobody.""" from serve_v3_live import _owner_of assert _owner_of(["trwc2026:boat:7", "trwc2026:boat:8"]) == "trwc2026" assert _owner_of(["7"]) is None assert _owner_of([]) is None def test_both_producers_derive_the_owner_identically(): """The two `_owner_of`s are deliberate copies (one module must not import the other), so they are asserted equal here rather than trusted to stay so.""" import rest_source import serve_v3_live for boats in (["trwc2026:boat:7"], ["7"], [], ["gsys:boat:2", "gsys:boat:3"]): assert serve_v3_live._owner_of(boats) == rest_source._owner_of(boats) def test_a_line_cameras_footage_is_bounded_to_the_race_it_is_shown_under(): """The two LINE cameras are bound to the COURSE, not to a race — they film every race run on it. Which makes the clamping above, which is right for a hull, wrong for them. A race-committee vessel's clips from the PREVIOUS race fall inside the generous pre-race margin, and because `tSec` is clamped at zero they do not merely appear on the timeline, they stack at its start and play at the gun: the line camera shows a start that is not the one on screen. So a vessel's map is bounded to this race's own window. Nothing is dropped from the record — every clip is still stored, and a replay of the neighbouring race plays it in its own place.""" gun, end = 100_000, 400_000 clips = [(40_000, "previous-race.mp4"), # the race before — inside the 10-min margin (150_000, "this-race-a.mp4"), (300_000, "this-race-b.mp4"), (450_000, "next-race.mp4")] # the race after unbounded = media_rows(clips, gun) assert [r["file"] for r in unbounded] == [ "previous-race.mp4", "this-race-a.mp4", "this-race-b.mp4", "next-race.mp4"] assert unbounded[0]["tSec"] == 0, "the neighbouring race's clip lands ON the gun — the defect" bounded = media_rows(clips, gun, from_ms=gun, until_ms=end) assert [r["file"] for r in bounded] == ["this-race-a.mp4", "this-race-b.mp4"] assert [r["tSec"] for r in bounded] == [50.0, 200.0] def test_a_running_race_bounds_only_its_past_because_it_has_no_later_race_yet(): """`until` is None while the race is still sailing. There is no next race to exclude, and a bound guessed from the clock would cut the finish off the very tile watching the finish line.""" rows = media_rows([(90_000, "before.mp4"), (500_000, "live.mp4")], 100_000, from_ms=100_000, until_ms=None) assert [r["file"] for r in rows] == ["live.mp4"] def test_the_bound_is_presentation_only_and_a_hull_never_gets_it(): """A hull is entered in ONE race, so its clips are that race's by construction and the margin only recovers what it buffered while offline. Bounding a hull would throw those away.""" early = media_rows([(95_000, "buffered.mp4")], 100_000) assert [r["tSec"] for r in early] == [0], "a hull's buffered pre-start clip must survive" def test_the_store_producers_camera_channel_is_actually_WIRED(): """`gcs_clip_url` is the production `clip_url` — and for the life of the store path it had no caller, so a producer reading the database published an empty media map. Every piece was present and correct: `gcs_clip_url` signed a clip, `run()` accepted a `clip_url`, `broadcast_following` accepted one, `_OnAirRace` used it. Nothing connected them. Both ends of a wire can pass their own tests while the wire does not exist, which is why this asserts the CALL GRAPH rather than any one function: the symptom is a tile drawing "(clip unavailable for this window)" while the phones film normally and the clips land — the same silent-camera defect this module's own docstring records for the REST producer, arriving from the other side.""" import inspect import re import serve_v3_live src = inspect.getsource(serve_v3_live) # Matched as a PATTERN, not a literal, because the argument is legitimately guarded: # `clip_url=None if camera_media_disabled else gcs_clip_url` (the deliberate camera kill switch # from "withdraw the left panel and suppress camera media"). The literal string this used to # assert stopped appearing the day that guard landed, and the test then failed for a whole # release while the wire it protects was perfectly intact — which is its own kind of silence. # What still must hold is the call graph: `gcs_clip_url` reaches `run()` AS `clip_url`. A bare # `clip_url=None`, or the argument dropped entirely, fails this exactly as before. assert re.search(r"clip_url=[^,]*gcs_clip_url", src), ( "the live CLI must hand the production clip_url to run() — without it the store-backed " "producer has no camera channel at all") run_src = inspect.getsource(serve_v3_live.run) following = run_src.split("broadcast_following(")[1].split(")")[0] assert "clip_url=clip_url" in following, ( "run() must forward clip_url into broadcast_following — a following producer is the one " "that airs a whole race day, and it is the path that had no cameras") def test_an_empty_roster_silently_costs_the_two_line_cameras(): """WHY the following path must pass THIS race's roster and not the outer `boats`. The vessel slots are built from the account NAME, which is read off a boat ref. The following path was passing the outer `boats`, which is `[]` for its entire life — it is only assigned on the pinned-`--instance` path. From `[]` the owner is None and the two LINE cameras vanish from every race a following producer airs. Nothing raises: a camera nobody declared looks exactly like a camera nobody pointed anywhere, which is why this is asserted as a VALUE rather than left to be noticed on air.""" from event_meta import overlay_event_config from serve_v3_live import _owner_of cfg = {"devices": {"vessels": {"start_stbd": "A", "finish_stbd": "B"}}} roster = ["trwc-sim:boat:1", "trwc-sim:boat:2"] assert _owner_of(roster) == "trwc-sim" assert _owner_of([]) is None, "an empty roster names no account — the trap" with_roster = overlay_event_config({}, cfg, owner=_owner_of(roster)) assert with_roster["vesselCameras"] == {"start": "trwc-sim:buoy:A", "finish": "trwc-sim:buoy:B"} without = overlay_event_config({}, cfg, owner=_owner_of([])) assert "vesselCameras" not in without, ( "the config declares two vessels; an empty roster dropped them without a word") def test_camera_display_rotation_is_event_data_and_only_accepts_quarter_turns(): """A live mounting correction must travel through the generic event seam without teaching the renderer an event's hull ids, and a typo must not put an arbitrary slant on air.""" from event_meta import overlay_event_config cfg = {"presentation": {"cameraDisplayRotationDegByRef": { "fleet:boat:11": 180, "fleet:buoy:A": 270, "fleet:buoy:B": "90", "fleet:boat:bad": 45, "fleet:boat:nonnumeric": "left", }}} meta = overlay_event_config({}, cfg, owner="fleet") assert meta["cameraDisplayRotationDegByRef"] == { "fleet:boat:11": 180, "fleet:buoy:A": -90, "fleet:buoy:B": 90, } def test_line_camera_overlay_is_event_data_with_bounded_detector_controls(): """Physical camera calibration is event data; only bounded renderer capabilities reach air.""" from event_meta import overlay_event_config cfg = {"presentation": {"lineCameraOverlay": { "enabled": True, "sampleFps": 20, "maxHoldMs": 5_000, "roles": { "start": {"targetRole": "start_port", "detector": "blue", "anchor": [0.5, 1], "search": [0.04, 0.08, 0.92, 0.82], "targetBox": [0.01, 0.01, 0.08, 0.10], "targetSize": [0.04, 0.05], "minConfidence": 0.78, "minPixels": 7}, "finish": {"targetRole": "finish_stbd", "detector": "darklight", "anchor": [0.5, 1], "search": [0.5, 0.5, 0.3, 0.2], "targetSize": [0.07, 0.08]}, }, }}} meta = overlay_event_config({}, cfg) assert meta["lineCameraOverlay"] == { "enabled": True, "sampleFps": 6.0, "maxHoldMs": 750, "roles": {"start": {"targetRole": "start_port", "detector": "blue", "anchor": [0.5, 1.0], "search": [0.04, 0.08, 0.92, 0.82], "targetBox": [0.01, 0.01, 0.08, 0.10], "targetSize": [0.04, 0.05], "minConfidence": 0.78, "minPixels": 7}, "finish": {"targetRole": "finish_stbd", "detector": "darklight", "anchor": [0.5, 1.0], "search": [0.5, 0.5, 0.3, 0.2], "targetSize": [0.07, 0.08]}}, } assert "lineCameraOverlay" not in overlay_event_config( {}, {"presentation": {"lineCameraOverlay": {"enabled": True, "roles": {}}}}) def test_the_following_presentation_passes_this_races_roster(): """…and the wiring itself, since the value test above cannot see which variable reached it.""" import inspect import serve_v3_live src = inspect.getsource(serve_v3_live) assert "boats=b)" in src, ( "the following presentation must pass THIS race's roster to apply_event_config, not the " "outer `boats` (which is [] on that path)")