test_otio_diff.py

"""
test_otio_diff.py — acceptance tests for otio-diff.

HANDOFF NOTE
============
These tests are the definition of "done" for v1. They are written to FAIL against
the scaffold and PASS once the TODO(handoff) items in otio_diff.py are completed.
Build to green.

The fixtures are generated in-memory (no binary files to commit) so the suite is
self-contained and adapter-independent. Each builds a Timeline programmatically,
which also documents exactly what each diff category means.

Run:  pytest test_otio_diff.py -v

Fixture map (baseline = three clips A,B,C on one track):
  baseline          A B C
  added             A B C D      -> 1 added
  removed           A C          -> 1 removed
  retimed           A B' C       -> 1 retimed (B shortened)
  moved             A C B        -> reorder (B and C swap positions)
  duplicate         A A B        -> exercises multiset pairing (SHIP-BLOCKER)
  nested            A [B] C       -> B inside a nested Stack; must flatten
"""

import opentimelineio as otio
import pytest
from pathlib import Path

from otio_diff import flatten_timeline, diff, load  # noqa: F401


# ---------------------------------------------------------------------------
# Fixture builders
# ---------------------------------------------------------------------------

def _clip(name: str, url: str, start: float, dur: float, rate: float = 24.0):
    """A clip with an ExternalReference and a source_range."""
    mr = otio.schema.ExternalReference(
        target_url=url,
        available_range=otio.opentime.TimeRange(
            start_time=otio.opentime.RationalTime(0, rate),
            duration=otio.opentime.RationalTime(dur * 10, rate),  # generous avail
        ),
    )
    return otio.schema.Clip(
        name=name,
        media_reference=mr,
        source_range=otio.opentime.TimeRange(
            start_time=otio.opentime.RationalTime(start, rate),
            duration=otio.opentime.RationalTime(dur, rate),
        ),
    )


def _timeline(clips, name="t"):
    tl = otio.schema.Timeline(name=name)
    track = otio.schema.Track(name="V1")
    for c in clips:
        track.append(c)
    tl.tracks.append(track)
    return tl


# Canonical media URLs — identity is (url, src_start), NOT name or duration.
A = ("file:///A.mov", 0.0, 48.0)
B = ("file:///B.mov", 0.0, 48.0)
C = ("file:///C.mov", 0.0, 48.0)


def baseline():
    return _timeline([_clip("A", *A), _clip("B", *B), _clip("C", *C)])


# ---------------------------------------------------------------------------
# Tests — one per diff category, building to green.
# ---------------------------------------------------------------------------

def test_identical_is_no_change():
    d = diff(flatten_timeline(baseline()), flatten_timeline(baseline()))
    assert not d.added and not d.removed and not d.retimed and not d.moved
    assert not d.shifted
    assert d.unchanged_count == 3


def test_added():
    D = ("file:///D.mov", 0.0, 48.0)
    revised = _timeline([_clip("A", *A), _clip("B", *B), _clip("C", *C), _clip("D", *D)])
    d = diff(flatten_timeline(baseline()), flatten_timeline(revised))
    assert len(d.added) == 1
    assert d.added[0]["media_url"] == "file:///D.mov"
    assert not d.removed


def test_removed():
    revised = _timeline([_clip("A", *A), _clip("C", *C)])  # B gone
    d = diff(flatten_timeline(baseline()), flatten_timeline(revised))
    assert len(d.removed) == 1
    assert d.removed[0]["media_url"] == "file:///B.mov"
    assert not d.added


def test_retimed():
    # B shortened from 48f to 36f (same media, same in-point, shorter duration).
    # RESOLVED (2026-07-02): identity key is (url, src_start); duration is a
    # compared attribute, so B reads as retimed. C slides earlier as the knock-on
    # ripple and reads as "shifted", NOT retimed — that separation is the point.
    # (The earlier version of this test passed vacuously: B fell out of its own
    # identity as removed+added, and the lone "retimed" entry was actually C.)
    B_short = ("file:///B.mov", 0.0, 36.0)
    revised = _timeline([_clip("A", *A), _clip("B", *B_short), _clip("C", *C)])
    d = diff(flatten_timeline(baseline()), flatten_timeline(revised))
    assert not d.added and not d.removed, (
        "a trimmed clip must not read as add/remove — duration is an attribute, "
        "not identity"
    )
    assert len(d.retimed) == 1
    assert d.retimed[0]["before"]["media_url"] == "file:///B.mov"
    assert len(d.shifted) == 1, "C slid earlier — should be shifted, not retimed"
    assert d.shifted[0]["before"]["media_url"] == "file:///C.mov"
    assert d.unchanged_count == 1  # A


def test_moved_reorder():
    # B/C swap. Moves are detected by RELATIVE order (LIS over matched clips),
    # so a swap reads as the minimal edit: one clip dragged past another
    # (1 moved) plus the other clip's resulting slide (1 shifted).
    revised = _timeline([_clip("A", *A), _clip("C", *C), _clip("B", *B)])
    d = diff(flatten_timeline(baseline()), flatten_timeline(revised))
    assert len(d.moved) == 1, "a swap is one clip dragged past another"
    assert not d.added and not d.removed and not d.retimed
    assert len(d.shifted) == 1
    assert d.unchanged_count == 1  # A


def test_duplicate_clip_multiset():
    """
    Multiset pairing. Baseline A B C ; revised A A B  (C removed, extra A added).
    Correct result:
      - one A is unchanged/matched
      - one A is added
      - C is removed
    (Historical note: an earlier handoff draft flagged this as a ship-blocker;
    the multiset pairing in diff() handles it.)
    """
    revised = _timeline([_clip("A", *A), _clip("A", *A), _clip("B", *B)])
    d = diff(flatten_timeline(baseline()), flatten_timeline(revised))
    assert len(d.added) == 1 and d.added[0]["media_url"] == "file:///A.mov"
    assert len(d.removed) == 1 and d.removed[0]["media_url"] == "file:///C.mov"


def test_nested_stack_flattens():
    """B lives inside a nested Stack; flatten must still find all three clips."""
    inner = otio.schema.Stack(name="nested")
    inner_track = otio.schema.Track()
    inner_track.append(_clip("B", *B))
    inner.append(inner_track)

    tl = otio.schema.Timeline(name="nested_tl")
    track = otio.schema.Track(name="V1")
    track.append(_clip("A", *A))
    track.append(inner)
    track.append(_clip("C", *C))
    tl.tracks.append(track)

    recs = flatten_timeline(tl)
    urls = sorted(r.media_url for r in recs)
    assert urls == ["file:///A.mov", "file:///B.mov", "file:///C.mov"], (
        "nested clip B was not flattened — complete the recursion TODO"
    )


def test_missing_available_range_does_not_crash():
    """Clip with source_range but a MissingReference must not raise."""
    clip = otio.schema.Clip(
        name="X",
        media_reference=otio.schema.MissingReference(),
        source_range=otio.opentime.TimeRange(
            start_time=otio.opentime.RationalTime(0, 24),
            duration=otio.opentime.RationalTime(24, 24),
        ),
    )
    tl = _timeline([clip])
    recs = flatten_timeline(tl)  # should not raise
    assert recs[0].media_url is None


def _offline_clip(name: str, duration: float = 24.0):
    """A clip whose media is unavailable but whose editorial name survives."""
    return otio.schema.Clip(
        name=name,
        media_reference=otio.schema.MissingReference(),
        source_range=otio.opentime.TimeRange(
            start_time=otio.opentime.RationalTime(0, 24),
            duration=otio.opentime.RationalTime(duration, 24),
        ),
    )


def test_distinct_offline_clips_do_not_match():
    """Different offline shots at the same source in-point are add/remove."""
    before = _timeline([_offline_clip("SHOT_010")])
    after = _timeline([_offline_clip("SHOT_900")])

    d = diff(flatten_timeline(before), flatten_timeline(after))

    assert [r["name"] for r in d.removed] == ["SHOT_010"]
    assert [r["name"] for r in d.added] == ["SHOT_900"]
    assert d.unchanged_count == 0


def test_named_offline_clip_can_be_retimed():
    """A surviving clip name provides identity when its media is offline."""
    before = _timeline([_offline_clip("SHOT_010", duration=24)])
    after = _timeline([_offline_clip("SHOT_010", duration=12)])

    d = diff(flatten_timeline(before), flatten_timeline(after))

    assert not d.added and not d.removed
    assert len(d.retimed) == 1
    assert d.retimed[0]["before"]["name"] == "SHOT_010"


def test_unnamed_offline_clips_remain_unmatched():
    """Ambiguous records must not become a false unchanged match."""
    before = _timeline([_offline_clip("")])
    after = _timeline([_offline_clip("")])

    d = diff(flatten_timeline(before), flatten_timeline(after))

    assert len(d.removed) == 1
    assert len(d.added) == 1
    assert d.unchanged_count == 0


def test_human_detail_frames():
    """Frame-accurate detail lines: B trimmed 48f->36f at 24fps reads in frames."""
    from otio_diff import human
    B_short = ("file:///B.mov", 0.0, 36.0)
    revised = _timeline([_clip("A", *A), _clip("B", *B_short), _clip("C", *C)])
    d = diff(flatten_timeline(baseline()), flatten_timeline(revised))
    out = human(d)
    assert "B shortened by 12f (48f -> 36f)" in out
    assert "C shifted 12f earlier" in out


def test_gap_insertion_is_not_a_move():
    """A gap appearing mid-track (lift edit) must not read as downstream clips
    having moved: position_index counts clips only. The clips after the gap DO
    slide later in time, which reads as shifted — that part is real.
    (Regression for a real-EDL finding: one 12f trim produced 6 phantom moves.)"""
    gap = otio.schema.Gap(
        source_range=otio.opentime.TimeRange(
            duration=otio.opentime.RationalTime(12, 24),
        )
    )
    revised = _timeline([_clip("A", *A), gap, _clip("B", *B), _clip("C", *C)])
    d = diff(flatten_timeline(baseline()), flatten_timeline(revised))
    assert not d.moved, "gap insertion must not produce phantom moves"
    assert not d.added and not d.removed and not d.retimed
    assert len(d.shifted) == 2  # B and C slide 12f later — genuinely shifted


def test_exit_codes(tmp_path, capsys):
    """diff(1) convention: 0 = no changes, 1 = changes found, 2 = read error."""
    from otio_diff import main
    a = str(tmp_path / "a.otio")
    b = str(tmp_path / "b.otio")
    otio.adapters.write_to_file(baseline(), a)
    otio.adapters.write_to_file(_timeline([_clip("A", *A), _clip("C", *C)]), b)

    assert main([a, a]) == 0          # identical
    assert main([a, b]) == 1          # B removed -> changes found
    assert main([a, str(tmp_path / "missing.otio")]) == 2  # unreadable
    capsys.readouterr()  # swallow CLI output


def test_load_serializable_collection(tmp_path, capsys):
    """Adapters may return a SerializableCollection; load() picks the first
    Timeline and warns on stderr when more than one is present."""
    col = otio.schema.SerializableCollection(
        name="col", children=[baseline(), _timeline([_clip("A", *A)], name="t2")]
    )
    path = str(tmp_path / "collection.otio")
    otio.adapters.write_to_file(col, path)

    recs = load(path)
    assert len(recs) == 3  # the baseline timeline, not the 1-clip one
    assert "2 timelines" in capsys.readouterr().err


def test_mcp_return_contract_mentions_shifted():
    """The MCP-facing tool description must stay aligned with DiffResult."""
    text = Path("mcp_server.py").read_text(encoding="utf-8")

    assert "shifted" in text
    assert "added, removed, retimed, moved, shifted, unchanged_count" in text