test_timeline_plugin.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """VID_01, VID_07–VID_14 — timeline plugin integration tests.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import pathlib |
| 7 | import shutil |
| 8 | |
| 9 | import pytest |
| 10 | |
| 11 | from muse.core.schema import DomainSchema |
| 12 | from muse.core.types import JsonObject, content_hash |
| 13 | from muse.domain import MuseDomainPlugin, SnapshotManifest |
| 14 | from muse.plugins.registry import registered_domains, resolve_plugin_by_domain |
| 15 | from muse.plugins.timeline._otio_bridge import BridgeError, load_otio_json, parse_otio_to_project |
| 16 | from muse.plugins.timeline.entity import ( |
| 17 | Clip, |
| 18 | Gap, |
| 19 | assign_clip_occurrences, |
| 20 | assign_identities, |
| 21 | clip_identity, |
| 22 | project_content_id, |
| 23 | sequence_identity, |
| 24 | track_identity, |
| 25 | ) |
| 26 | from muse.plugins.timeline.manifest import build_timeline_manifest |
| 27 | from muse.plugins.timeline.plugin import TimelinePlugin, plugin |
| 28 | from muse.plugins.timeline.rational_time import RationalRate, RationalTime, TimeRange |
| 29 | |
| 30 | FIXTURES = pathlib.Path(__file__).parent / "fixtures" / "timeline" |
| 31 | GOLDEN = FIXTURES / "golden" |
| 32 | |
| 33 | |
| 34 | @pytest.fixture() |
| 35 | def timeline_plugin() -> TimelinePlugin: |
| 36 | return TimelinePlugin() |
| 37 | |
| 38 | |
| 39 | @pytest.fixture() |
| 40 | def fixture_repo(tmp_path: pathlib.Path) -> pathlib.Path: |
| 41 | repo = tmp_path / "repo" |
| 42 | repo.mkdir() |
| 43 | media = repo / "media" |
| 44 | media.mkdir() |
| 45 | for name in ("shot_a.mov", "shot_b.mov", "shot_c.mov", "same_shot.mov", "v1.mov", "v2.mov", "a1.wav", "ntsc.mov", "df.mov", "nested_ref.mov"): |
| 46 | (media / name).write_bytes(b"stub") |
| 47 | shutil.copy(FIXTURES / "single_track.otio", repo / "single_track.otio") |
| 48 | return repo |
| 49 | |
| 50 | |
| 51 | class TestVID01Protocol: |
| 52 | def test_plugin_satisfies_protocol(self) -> None: |
| 53 | assert isinstance(plugin, MuseDomainPlugin) |
| 54 | |
| 55 | def test_schema_valid(self, timeline_plugin: TimelinePlugin) -> None: |
| 56 | schema: DomainSchema = timeline_plugin.schema() |
| 57 | assert schema["domain"] == "timeline" |
| 58 | assert schema["merge_mode"] == "three_way" |
| 59 | assert len(schema["dimensions"]) > 0 |
| 60 | |
| 61 | def test_registered_active(self) -> None: |
| 62 | assert "timeline" in registered_domains() |
| 63 | p = resolve_plugin_by_domain("timeline") |
| 64 | assert isinstance(p, TimelinePlugin) |
| 65 | |
| 66 | |
| 67 | class TestVID07OtioBridge: |
| 68 | def test_single_track_parse(self, fixture_repo: pathlib.Path) -> None: |
| 69 | raw = load_otio_json(fixture_repo / "single_track.otio") |
| 70 | project = parse_otio_to_project(raw, fixture_repo / "single_track.otio", fixture_repo) |
| 71 | assert project.project_key == "single_track" |
| 72 | assert len(project.sequences) == 1 |
| 73 | track = project.sequences[0].video_tracks[0] |
| 74 | clips = [i for i in track.items if isinstance(i, Clip)] |
| 75 | gaps = [i for i in track.items if isinstance(i, Gap)] |
| 76 | assert len(clips) == 3 |
| 77 | assert len(gaps) == 1 |
| 78 | assert clips[0].source_range.start_time.rate == RationalRate(24, 1) |
| 79 | |
| 80 | def test_unknown_schema_version_raises(self) -> None: |
| 81 | bad = { |
| 82 | "OTIO_SCHEMA": "Timeline.999", |
| 83 | "name": "bad", |
| 84 | "tracks": {"OTIO_SCHEMA": "Stack.1", "children": []}, |
| 85 | } |
| 86 | with pytest.raises(BridgeError): |
| 87 | parse_otio_to_project(bad, pathlib.Path("bad.otio"), None) |
| 88 | |
| 89 | |
| 90 | class TestVID08Determinism: |
| 91 | def test_snapshot_twice_identical(self, fixture_repo: pathlib.Path) -> None: |
| 92 | a = plugin.snapshot(fixture_repo) |
| 93 | b = plugin.snapshot(fixture_repo) |
| 94 | assert content_hash(a) == content_hash(b) |
| 95 | |
| 96 | def test_reformatted_identical_hash(self, fixture_repo: pathlib.Path) -> None: |
| 97 | shutil.copy(FIXTURES / "single_track_reformatted.otio", fixture_repo / "reformatted.otio") |
| 98 | base = plugin.snapshot(fixture_repo) |
| 99 | only_reformatted = fixture_repo / "only_ref" |
| 100 | only_reformatted.mkdir() |
| 101 | (only_reformatted / "media").mkdir() |
| 102 | for name in ("shot_a.mov", "shot_b.mov", "shot_c.mov"): |
| 103 | (only_reformatted / "media" / name).write_bytes(b"stub") |
| 104 | shutil.copy(FIXTURES / "single_track.otio", only_reformatted / "single_track.otio") |
| 105 | snap_a = plugin.snapshot(only_reformatted) |
| 106 | shutil.copy(FIXTURES / "single_track_reformatted.otio", only_reformatted / "single_track.otio") |
| 107 | snap_b = plugin.snapshot(only_reformatted) |
| 108 | assert snap_a["files"]["single_track.otio"] == snap_b["files"]["single_track.otio"] |
| 109 | |
| 110 | |
| 111 | class TestVID09ClipIdentityExcludesTimelineStart: |
| 112 | def test_widen_gap_preserves_clip_ids(self, fixture_repo: pathlib.Path) -> None: |
| 113 | raw = load_otio_json(fixture_repo / "single_track.otio") |
| 114 | project = parse_otio_to_project(raw, fixture_repo / "single_track.otio", fixture_repo) |
| 115 | before = assign_identities(project) |
| 116 | clip_ids_before = [ic.clip_id for ic in before.sequences[0].tracks[0].clips] |
| 117 | |
| 118 | track = project.sequences[0].video_tracks[0] |
| 119 | items = list(track.items) |
| 120 | gap = items[1] |
| 121 | assert isinstance(gap, Gap) |
| 122 | wider = Gap( |
| 123 | duration=RationalTime(gap.duration.value + 48, gap.duration.rate), |
| 124 | metadata=gap.metadata, |
| 125 | ) |
| 126 | items[1] = wider |
| 127 | from dataclasses import replace |
| 128 | |
| 129 | new_track = replace(track, items=tuple(items)) |
| 130 | new_seq = replace( |
| 131 | project.sequences[0], |
| 132 | video_tracks=(new_track,), |
| 133 | ) |
| 134 | new_project = replace(project, sequences=(new_seq,)) |
| 135 | after = assign_identities(new_project) |
| 136 | clip_ids_after = [ic.clip_id for ic in after.sequences[0].tracks[0].clips] |
| 137 | assert clip_ids_before == clip_ids_after |
| 138 | |
| 139 | def test_trim_source_changes_clip_id(self, fixture_repo: pathlib.Path) -> None: |
| 140 | raw = load_otio_json(fixture_repo / "single_track.otio") |
| 141 | project = parse_otio_to_project(raw, fixture_repo / "single_track.otio", fixture_repo) |
| 142 | before = assign_identities(project) |
| 143 | cid_before = before.sequences[0].tracks[0].clips[0].clip_id |
| 144 | |
| 145 | from dataclasses import replace |
| 146 | |
| 147 | clip = project.sequences[0].video_tracks[0].items[0] |
| 148 | assert isinstance(clip, Clip) |
| 149 | new_sr = TimeRange( |
| 150 | clip.source_range.start_time, |
| 151 | RationalTime(clip.source_range.duration.value + 10, clip.source_range.duration.rate), |
| 152 | ) |
| 153 | new_clip = replace(clip, source_range=new_sr) |
| 154 | items = list(project.sequences[0].video_tracks[0].items) |
| 155 | items[0] = new_clip |
| 156 | new_track = replace(project.sequences[0].video_tracks[0], items=tuple(items)) |
| 157 | new_seq = replace(project.sequences[0], video_tracks=(new_track,)) |
| 158 | new_project = replace(project, sequences=(new_seq,)) |
| 159 | after = assign_identities(new_project) |
| 160 | assert after.sequences[0].tracks[0].clips[0].clip_id != cid_before |
| 161 | |
| 162 | |
| 163 | class TestVID10TrackIdentity: |
| 164 | def test_reorder_swaps_track_ids(self, fixture_repo: pathlib.Path) -> None: |
| 165 | raw = load_otio_json(FIXTURES / "multi_track.otio") |
| 166 | project = parse_otio_to_project(raw, FIXTURES / "multi_track.otio", fixture_repo) |
| 167 | seq = project.sequences[0] |
| 168 | seq_id = sequence_identity(project.project_key, seq.name, seq.ordinal) |
| 169 | t0_id = track_identity(seq_id, "video", 0) |
| 170 | t1_id = track_identity(seq_id, "video", 1) |
| 171 | |
| 172 | from dataclasses import replace |
| 173 | |
| 174 | vtracks = list(seq.video_tracks) |
| 175 | vtracks[0], vtracks[1] = vtracks[1], vtracks[0] |
| 176 | vtracks[0] = replace(vtracks[0], ordinal=0) |
| 177 | vtracks[1] = replace(vtracks[1], ordinal=1) |
| 178 | reordered = replace(seq, video_tracks=tuple(vtracks)) |
| 179 | assert track_identity(seq_id, "video", 0) == t0_id |
| 180 | assert track_identity(seq_id, "video", 1) == t1_id |
| 181 | assert vtracks[0].name == "V2" |
| 182 | assert vtracks[1].name == "V1" |
| 183 | id_v1_after = track_identity(seq_id, "video", vtracks[1].ordinal) |
| 184 | id_v2_after = track_identity(seq_id, "video", vtracks[0].ordinal) |
| 185 | assert t0_id == id_v2_after |
| 186 | assert t1_id == id_v1_after |
| 187 | |
| 188 | def test_edit_clip_preserves_track_id(self, fixture_repo: pathlib.Path) -> None: |
| 189 | raw = load_otio_json(FIXTURES / "multi_track.otio") |
| 190 | project = parse_otio_to_project(raw, FIXTURES / "multi_track.otio", fixture_repo) |
| 191 | seq = project.sequences[0] |
| 192 | tid = track_identity( |
| 193 | sequence_identity(project.project_key, seq.name, seq.ordinal), |
| 194 | "video", |
| 195 | 0, |
| 196 | ) |
| 197 | from dataclasses import replace |
| 198 | |
| 199 | clip = seq.video_tracks[0].items[0] |
| 200 | assert isinstance(clip, Clip) |
| 201 | new_clip = replace(clip, name="renamed") |
| 202 | items = list(seq.video_tracks[0].items) |
| 203 | items[0] = new_clip |
| 204 | new_track = replace(seq.video_tracks[0], items=tuple(items)) |
| 205 | new_seq = replace(seq, video_tracks=(new_track, seq.video_tracks[1])) |
| 206 | new_project = replace(project, sequences=(new_seq,)) |
| 207 | tid_after = track_identity( |
| 208 | sequence_identity(new_project.project_key, new_seq.name, new_seq.ordinal), |
| 209 | "video", |
| 210 | 0, |
| 211 | ) |
| 212 | assert tid == tid_after |
| 213 | |
| 214 | |
| 215 | class TestVID11PositionalTiebreaker: |
| 216 | def test_identical_shots_distinct_ids(self, fixture_repo: pathlib.Path) -> None: |
| 217 | shutil.copy(FIXTURES / "identical_shots.otio", fixture_repo / "identical_shots.otio") |
| 218 | raw = load_otio_json(fixture_repo / "identical_shots.otio") |
| 219 | project = parse_otio_to_project(raw, fixture_repo / "identical_shots.otio", fixture_repo) |
| 220 | identified = assign_identities(project) |
| 221 | clips = identified.sequences[0].tracks[0].clips |
| 222 | assert len(clips) == 2 |
| 223 | assert clips[0].clip_id != clips[1].clip_id |
| 224 | assert clips[0].occurrence == 0 |
| 225 | assert clips[1].occurrence == 1 |
| 226 | |
| 227 | def test_stable_across_resnapshot(self, fixture_repo: pathlib.Path) -> None: |
| 228 | shutil.copy(FIXTURES / "identical_shots.otio", fixture_repo / "identical_shots.otio") |
| 229 | a = plugin.snapshot(fixture_repo) |
| 230 | b = plugin.snapshot(fixture_repo) |
| 231 | assert a == b |
| 232 | |
| 233 | def test_trim_leaves_identical_set(self, fixture_repo: pathlib.Path) -> None: |
| 234 | raw = load_otio_json(FIXTURES / "identical_shots.otio") |
| 235 | project = parse_otio_to_project(raw, FIXTURES / "identical_shots.otio", fixture_repo) |
| 236 | track = project.sequences[0].video_tracks[0] |
| 237 | tid = track_identity( |
| 238 | sequence_identity(project.project_key, project.sequences[0].name, 0), |
| 239 | "video", |
| 240 | 0, |
| 241 | ) |
| 242 | clip = track.items[1] |
| 243 | assert isinstance(clip, Clip) |
| 244 | from dataclasses import replace |
| 245 | |
| 246 | new_sr = TimeRange( |
| 247 | RationalTime(10, clip.source_range.start_time.rate), |
| 248 | clip.source_range.duration, |
| 249 | ) |
| 250 | new_clip = replace(clip, source_range=new_sr) |
| 251 | items = list(track.items) |
| 252 | items[1] = new_clip |
| 253 | occ = assign_clip_occurrences(tid, tuple(items)) |
| 254 | assert occ[1][1] == 0 |
| 255 | |
| 256 | |
| 257 | def _build_manifest(repo: pathlib.Path, otio_name: str) -> JsonObject: |
| 258 | snap = plugin.snapshot(repo) |
| 259 | raw = load_otio_json(repo / otio_name) |
| 260 | project = parse_otio_to_project(raw, repo / otio_name, repo) |
| 261 | identified = assign_identities(project) |
| 262 | manifest = build_timeline_manifest( |
| 263 | snap["files"], |
| 264 | {otio_name: identified}, |
| 265 | {otio_name: project_content_id(project)}, |
| 266 | snapshot_id="golden", |
| 267 | ) |
| 268 | return {"files": snap["files"], "manifest": manifest} |
| 269 | |
| 270 | |
| 271 | class TestVID12GoldenSingleTrack: |
| 272 | def test_matches_golden(self, fixture_repo: pathlib.Path) -> None: |
| 273 | built = _build_manifest(fixture_repo, "single_track.otio") |
| 274 | golden_path = GOLDEN / "single_track.manifest.json" |
| 275 | if not golden_path.exists(): |
| 276 | golden_path.parent.mkdir(parents=True, exist_ok=True) |
| 277 | golden_path.write_text(json.dumps(built, indent=2, sort_keys=True) + "\n") |
| 278 | golden = json.loads(golden_path.read_text()) |
| 279 | assert built["files"] == golden["files"] |
| 280 | |
| 281 | |
| 282 | class TestVID13GoldenMultiTrack: |
| 283 | def test_matches_golden(self, fixture_repo: pathlib.Path) -> None: |
| 284 | shutil.copy(FIXTURES / "multi_track.otio", fixture_repo / "multi_track.otio") |
| 285 | built = _build_manifest(fixture_repo, "multi_track.otio") |
| 286 | golden_path = GOLDEN / "multi_track.manifest.json" |
| 287 | if not golden_path.exists(): |
| 288 | golden_path.parent.mkdir(parents=True, exist_ok=True) |
| 289 | golden_path.write_text(json.dumps(built, indent=2, sort_keys=True) + "\n") |
| 290 | golden = json.loads(golden_path.read_text()) |
| 291 | assert built["files"] == golden["files"] |
| 292 | |
| 293 | |
| 294 | class TestVID14Security: |
| 295 | def test_offline_media_deterministic(self, tmp_path: pathlib.Path) -> None: |
| 296 | repo = tmp_path / "offline" |
| 297 | repo.mkdir() |
| 298 | shutil.copy(FIXTURES / "offline_media.otio", repo / "offline_media.otio") |
| 299 | a = plugin.snapshot(repo) |
| 300 | b = plugin.snapshot(repo) |
| 301 | assert a == b |
| 302 | raw = load_otio_json(repo / "offline_media.otio") |
| 303 | project = parse_otio_to_project(raw, repo / "offline_media.otio", repo) |
| 304 | ref = project.media_pool[0].references[0][1] |
| 305 | assert ref.is_offline |
| 306 | |
| 307 | def test_path_traversal_rejected(self) -> None: |
| 308 | bad = { |
| 309 | "OTIO_SCHEMA": "Timeline.1", |
| 310 | "name": "bad", |
| 311 | "metadata": {"muse.project_key": "bad"}, |
| 312 | "tracks": { |
| 313 | "OTIO_SCHEMA": "Stack.1", |
| 314 | "children": [ |
| 315 | { |
| 316 | "OTIO_SCHEMA": "Track.1", |
| 317 | "kind": "Video", |
| 318 | "children": [ |
| 319 | { |
| 320 | "OTIO_SCHEMA": "Clip.2", |
| 321 | "name": "x", |
| 322 | "media_reference": { |
| 323 | "OTIO_SCHEMA": "ExternalReference.1", |
| 324 | "target_url": "../etc/passwd", |
| 325 | }, |
| 326 | "source_range": { |
| 327 | "OTIO_SCHEMA": "TimeRange.1", |
| 328 | "start_time": {"OTIO_SCHEMA": "RationalTime.1", "rate": 24, "value": 0}, |
| 329 | "duration": {"OTIO_SCHEMA": "RationalTime.1", "rate": 24, "value": 1}, |
| 330 | }, |
| 331 | } |
| 332 | ], |
| 333 | } |
| 334 | ], |
| 335 | }, |
| 336 | } |
| 337 | with pytest.raises(BridgeError): |
| 338 | parse_otio_to_project(bad, pathlib.Path("bad.otio"), None) |
| 339 | |
| 340 | def test_no_media_bytes_in_objects(self, fixture_repo: pathlib.Path, tmp_path: pathlib.Path) -> None: |
| 341 | muse_repo = tmp_path / "muse_repo" |
| 342 | muse_repo.mkdir() |
| 343 | shutil.copytree(fixture_repo / "media", muse_repo / "media") |
| 344 | shutil.copy(fixture_repo / "single_track.otio", muse_repo / "single_track.otio") |
| 345 | from muse.core.paths import objects_dir |
| 346 | |
| 347 | objects_dir(muse_repo).mkdir(parents=True, exist_ok=True) |
| 348 | snap = plugin.snapshot(muse_repo) |
| 349 | for digest in snap["files"].values(): |
| 350 | blob_path = objects_dir(muse_repo) / digest.split(":", 1)[1] |
| 351 | if blob_path.exists(): |
| 352 | data = blob_path.read_bytes() |
| 353 | assert b"stub" not in data |