"""Where UTC zero is inside the local recording — the one fact that cannot be recovered afterwards.

Nothing in the picture is machine-readable — not the broadcast's own stamp, and no longer the clock a
device burns into its footage either — so a race is only findable inside a day-long recording if the
anchor was written down while OBS was still running. These tests cover what that write has to
survive: a recording started by hand, a file split mid-day, OBS going away and coming back.
"""
import sys
from pathlib import Path

_LIVE = Path(__file__).resolve().parents[2] / "live"
sys.path.insert(0, str(_LIVE))

from recording_anchor import (  # noqa: E402
    anchor_for,
    read_samples,
    run,
    sample,
    stream_anchor_for,
)


def test_a_minute_with_no_recording_is_not_a_measurement():
    """None, not a row of nulls: a reader that has to skip null rows will one day forget to."""
    assert sample({"recording": False, "recordDurationMs": 0}, now_ms=1_000) is None
    assert sample({"recording": True}, now_ms=1_000) is None, "recording but no duration is not usable"


def test_the_anchor_is_now_minus_how_far_in_obs_is():
    got = sample({"recording": True, "recordDurationMs": 90_000, "recordPath": "d/a.mp4"},
                 now_ms=1_000_000)
    assert got["anchorUtcMs"] == 910_000
    assert got["path"] == "d/a.mp4"


def test_the_anchor_is_the_median_so_one_slow_round_trip_cannot_move_it():
    """Every sample carries the jitter of one call to OBS. The median discards the outlier without a
    threshold anyone would have to justify."""
    rows = [{"anchorUtcMs": 1_000, "path": "a.mp4"},
            {"anchorUtcMs": 1_050, "path": "a.mp4"},
            {"anchorUtcMs": 9_999, "path": "a.mp4"},          # the machine stalled for a second
            {"anchorUtcMs": 1_020, "path": "a.mp4"},
            {"anchorUtcMs": 1_010, "path": "a.mp4"}]
    assert anchor_for(rows) == 1_020


def test_each_split_file_gets_its_own_zero():
    """OBS splits the recording on a timer and every new segment restarts at zero. Averaging across a
    split would place every race in the day somewhere between two files."""
    rows = [{"anchorUtcMs": 1_000, "path": "a.mp4"},
            {"anchorUtcMs": 1_000, "path": "a.mp4"},
            {"anchorUtcMs": 5_000, "path": "b.mp4"},
            {"anchorUtcMs": 5_010, "path": "b.mp4"}]
    assert anchor_for(rows, "a.mp4") == 1_000
    assert anchor_for(rows, "b.mp4") == 5_010
    assert anchor_for(rows) == 5_010, "with no file named, the one being written now"


def test_obs_going_away_does_not_end_the_run(tmp_path):
    """The sampler outlives an OBS restart: the samples on either side of one are both good, and a
    process that quit on the first refused connection would have thrown away the second half of a
    race day."""
    calls = {"n": 0}

    def flaky():
        calls["n"] += 1
        if calls["n"] == 2:
            raise ConnectionRefusedError("obs-websocket is not up")
        return {"recording": True, "recordDurationMs": calls["n"] * 1_000, "recordPath": "a.mp4"}

    out = tmp_path / "anchor.jsonl"
    written = run(out, interval_s=0, status_fn=flaky, now_ms_fn=lambda: 1_000_000,
                  sleep_fn=lambda _s: None, max_samples=4, on_log=lambda _m: None)
    assert written == 3, "three good minutes, one unreachable — and the run carried on"
    assert len(read_samples(out)) == 3


# --- the OUTGOING STREAM, which has a different zero and belongs to somebody else ----------------

def test_the_stream_gets_its_own_anchor_from_the_same_reading():
    """A partner recording our feed starts at the moment our STREAM opened, not our file. Both
    anchors come out of one call to OBS, so this costs nothing extra at run time."""
    got = sample({"recording": True, "recordDurationMs": 90_000, "recordPath": "d/a.mp4",
                  "streaming": True, "streamDurationMs": 120_000}, now_ms=1_000_000)
    assert got["anchorUtcMs"] == 910_000, "the file's zero"
    assert got["streamAnchorUtcMs"] == 880_000, "the stream's zero — earlier, it was started first"


def test_streaming_without_recording_is_still_a_measurement():
    """The two outputs are independent. A minute of one is a measurement of that one; only a minute
    of NEITHER is nothing."""
    got = sample({"recording": False, "streaming": True, "streamDurationMs": 5_000}, now_ms=9_000)
    assert got is not None
    assert got["streamAnchorUtcMs"] == 4_000
    assert "anchorUtcMs" not in got, "no recording, so no claim about a file"
    assert sample({"recording": False, "streaming": False}, now_ms=9_000) is None


def test_a_stream_only_row_does_not_break_the_recording_reader():
    """`anchor_for` filters on the field it needs, so rows describing only the stream are skipped
    rather than mistaken for a recording with no anchor."""
    rows = [{"tUtcMs": 1, "streamAnchorUtcMs": 500, "streamDurationMs": 100},
            {"tUtcMs": 2, "anchorUtcMs": 1_000, "path": "a.mp4"},
            {"tUtcMs": 3, "streamAnchorUtcMs": 505, "streamDurationMs": 200}]
    assert anchor_for(rows) == 1_000


def test_a_restarted_stream_starts_a_new_zero():
    """THE CASE THE FILENAME SOLVES FOR THE RECORDING AND NOTHING SOLVES FOR THE STREAM. A stream
    that stops and starts again restarts its elapsed time, so a duration going DOWN is the session
    boundary. Averaging across it would place the whole day between two sessions."""
    rows = [{"tUtcMs": 10, "streamDurationMs": 8_000, "streamAnchorUtcMs": 2},
            {"tUtcMs": 20, "streamDurationMs": 18_000, "streamAnchorUtcMs": 2},
            {"tUtcMs": 30, "streamDurationMs": 2_000, "streamAnchorUtcMs": 9_000},   # restarted
            {"tUtcMs": 40, "streamDurationMs": 12_000, "streamAnchorUtcMs": 9_010}]
    assert stream_anchor_for(rows) == 9_010, "the CURRENT session's zero, not a blend of both"


def test_the_stream_anchor_is_the_median_within_its_session():
    rows = [{"tUtcMs": t, "streamDurationMs": t * 1_000, "streamAnchorUtcMs": a}
            for t, a in ((1, 1_000), (2, 1_050), (3, 9_999), (4, 1_020), (5, 1_010))]
    assert stream_anchor_for(rows) == 1_020


def test_no_stream_samples_is_none_not_a_guess():
    assert stream_anchor_for([]) is None
    assert stream_anchor_for([{"tUtcMs": 1, "anchorUtcMs": 5, "path": "a.mp4"}]) is None


def test_a_stream_only_minute_does_not_crash_the_sampler(tmp_path):
    """REGRESSION. The run loop announced each new recording path off `row['anchorUtcMs']`. Once a
    row could describe the stream and not the recording, a stream-only minute had neither key — and
    the sampler would have died on a KeyError partway through a race day."""
    statuses = [
        {"recording": True, "recordDurationMs": 1_000, "recordPath": "a.mp4",
         "streaming": True, "streamDurationMs": 5_000},
        {"recording": False, "streaming": True, "streamDurationMs": 6_000},   # stopped recording
        {"recording": True, "recordDurationMs": 1_000, "recordPath": "b.mp4",
         "streaming": True, "streamDurationMs": 7_000},
    ]
    out = tmp_path / "anchor.jsonl"
    written = run(out, interval_s=0, status_fn=lambda: statuses.pop(0),
                  now_ms_fn=lambda: 1_000_000, sleep_fn=lambda _s: None,
                  max_samples=3, on_log=lambda _m: None)
    assert written == 3
    rows = read_samples(out)
    assert anchor_for(rows, "b.mp4") == 999_000
    assert stream_anchor_for(rows) == 994_000


def test_a_truncated_last_line_is_dropped_not_raised(tmp_path):
    """The log is append-only and the run can be killed mid-write. Its tail is the least valuable
    part of it — losing one sample must never cost the other several hundred."""
    out = tmp_path / "anchor.jsonl"
    out.write_text('{"anchorUtcMs": 1000, "path": "a.mp4"}\n{"anchorUtcMs": 10',
                   encoding="utf-8")
    assert anchor_for(read_samples(out)) == 1_000