429 lines
17 KiB
Python
429 lines
17 KiB
Python
"""Tests for src/clip.py — stdlib only, no network, no ffmpeg."""
|
|
|
|
import os
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
import src.clip as clip
|
|
from src.clip import ClipError, find_clips, _path_start
|
|
|
|
|
|
def dt(value: str) -> datetime:
|
|
return datetime.fromisoformat(value)
|
|
|
|
|
|
def fake_probe(durations):
|
|
"""probe(path) -> duration; records calls on probe.calls."""
|
|
calls = []
|
|
|
|
def probe(path):
|
|
calls.append(path)
|
|
return durations.get(path, 0.0)
|
|
|
|
probe.calls = calls
|
|
return probe
|
|
|
|
|
|
# ---------------------------------------------------------------- path parsing
|
|
|
|
def test_path_start_valid_path():
|
|
assert _path_start("2026-09-26/09-51-40.mp4") == datetime(2026, 9, 26, 9, 51, 40)
|
|
assert _path_start("/2026-09-26/09-51-40.mp4") == datetime(2026, 9, 26, 9, 51, 40)
|
|
|
|
|
|
def test_path_start_skips_unparseable_paths():
|
|
assert _path_start("2026-09-26/09-51-40.avi") is None
|
|
assert _path_start("09-51-40.mp4") is None
|
|
assert _path_start("2026-9-26/09-51-40.mp4") is None
|
|
assert _path_start("") is None
|
|
assert _path_start(None) is None
|
|
|
|
|
|
# ------------------------------------------------------------------ find_clips
|
|
|
|
def test_find_clips_only_overlapping_clip():
|
|
movies = [
|
|
{"path": "2026-09-26/09-00-00.mp4", "start": dt("2026-09-26 09:00:00")},
|
|
{"path": "2026-09-26/09-20-00.mp4", "start": dt("2026-09-26 09:20:00")},
|
|
]
|
|
probe = fake_probe({
|
|
"2026-09-26/09-00-00.mp4": 300.0,
|
|
"2026-09-26/09-20-00.mp4": 60.0,
|
|
})
|
|
# B(09:20, 60s): window starts 10s into B and ends 50s into B.
|
|
got = find_clips(movies, dt("2026-09-26 09:20:10"), dt("2026-09-26 09:20:50"), probe)
|
|
assert [c["path"] for c in got] == ["2026-09-26/09-20-00.mp4"]
|
|
assert got[0]["off0"] == 10.0
|
|
assert got[0]["off1"] == 50.0
|
|
assert got[0]["duration"] == 60.0
|
|
|
|
|
|
def test_find_clips_window_touching_clip_end_excluded():
|
|
# B(09:20, 60s) ends exactly 09:21; window 09:21→09:25 overlaps by 0s.
|
|
movies = [
|
|
{"path": "2026-09-26/09-00-00.mp4", "start": dt("2026-09-26 09:00:00")},
|
|
{"path": "2026-09-26/09-20-00.mp4", "start": dt("2026-09-26 09:20:00")},
|
|
]
|
|
probe = fake_probe({
|
|
"2026-09-26/09-00-00.mp4": 300.0,
|
|
"2026-09-26/09-20-00.mp4": 60.0,
|
|
})
|
|
got = find_clips(movies, dt("2026-09-26 09:21:00"), dt("2026-09-26 09:25:00"), probe)
|
|
assert got == []
|
|
|
|
|
|
def test_find_clips_window_inside_one_clip():
|
|
movies = [{"path": "2026-09-26/10-00-00.mp4", "start": dt("2026-09-26 10:00:00")}]
|
|
probe = fake_probe({"2026-09-26/10-00-00.mp4": 600.0})
|
|
got = find_clips(movies, dt("2026-09-26 10:02:00"), dt("2026-09-26 10:04:00"), probe)
|
|
assert len(got) == 1
|
|
assert got[0]["off0"] == 120.0
|
|
assert got[0]["off1"] == 240.0
|
|
|
|
|
|
def test_find_clips_window_spanning_two_clips_in_order():
|
|
movies = [
|
|
{"path": "2026-09-26/09-04-00.mp4", "start": dt("2026-09-26 09:04:00")},
|
|
{"path": "2026-09-26/09-00-00.mp4", "start": dt("2026-09-26 09:00:00")},
|
|
]
|
|
probe = fake_probe({
|
|
"2026-09-26/09-00-00.mp4": 300.0,
|
|
"2026-09-26/09-04-00.mp4": 600.0,
|
|
})
|
|
got = find_clips(movies, dt("2026-09-26 09:03:00"), dt("2026-09-26 09:05:30"), probe)
|
|
assert [c["path"] for c in got] == [
|
|
"2026-09-26/09-00-00.mp4",
|
|
"2026-09-26/09-04-00.mp4",
|
|
]
|
|
assert (got[0]["off0"], got[0]["off1"]) == (180.0, 300.0)
|
|
assert (got[1]["off0"], got[1]["off1"]) == (0.0, 90.0)
|
|
|
|
|
|
def test_find_clips_no_overlap_empty():
|
|
movies = [{"path": "2026-09-26/10-35-00.mp4", "start": dt("2026-09-26 10:35:00")}]
|
|
probe = fake_probe({"2026-09-26/10-35-00.mp4": 60.0})
|
|
got = find_clips(movies, dt("2026-09-26 12:00:00"), dt("2026-09-26 12:01:00"), probe)
|
|
assert got == []
|
|
|
|
|
|
def test_find_clips_probe_zero_duration_skipped():
|
|
movies = [{"path": "2026-09-26/11-59-00.mp4", "start": dt("2026-09-26 11:59:00")}]
|
|
probe = fake_probe({})
|
|
assert probe("2026-09-26/11-59-00.mp4") == 0.0
|
|
got = find_clips(movies, dt("2026-09-26 12:00:00"), dt("2026-09-26 12:01:00"),
|
|
fake_probe({"2026-09-26/11-59-00.mp4": 0.0}))
|
|
assert got == []
|
|
|
|
|
|
def test_find_clips_probe_error_treated_as_zero():
|
|
def broken_probe(path):
|
|
raise RuntimeError("boom")
|
|
|
|
movies = [{"path": "2026-09-26/11-59-00.mp4", "start": dt("2026-09-26 11:59:00")}]
|
|
got = find_clips(movies, dt("2026-09-26 12:00:00"), dt("2026-09-26 12:01:00"),
|
|
broken_probe)
|
|
assert got == []
|
|
|
|
|
|
def test_find_clips_midnight_crossing_window():
|
|
movies = [
|
|
{"path": "/2026-09-26/23-58-00.mp4", "start": dt("2026-09-26 23:58:00")},
|
|
{"path": "/2026-09-27/00-00-30.mp4", "start": dt("2026-09-27 00:00:30")},
|
|
]
|
|
probe = fake_probe({
|
|
"/2026-09-26/23-58-00.mp4": 300.0,
|
|
"/2026-09-27/00-00-30.mp4": 300.0,
|
|
})
|
|
got = find_clips(movies, dt("2026-09-26 23:59:00"), dt("2026-09-27 00:01:00"), probe)
|
|
assert [c["path"] for c in got] == [
|
|
"/2026-09-26/23-58-00.mp4",
|
|
"/2026-09-27/00-00-30.mp4",
|
|
]
|
|
assert (got[1]["off0"], got[1]["off1"]) == (0.0, 30.0)
|
|
|
|
|
|
def test_find_clips_old_clip_not_probed():
|
|
start, end = dt("2026-09-26 12:00:00"), dt("2026-09-26 12:10:00")
|
|
old = {"path": "old.mp4", "start": start - clip._CANDIDATE_BACK - clip.timedelta(seconds=1)}
|
|
edge = {"path": "edge.mp4", "start": start - clip._CANDIDATE_BACK}
|
|
inside = {"path": "inside.mp4", "start": dt("2026-09-26 11:59:00")}
|
|
probe = fake_probe({
|
|
"old.mp4": 3600.0,
|
|
"edge.mp4": 3600.0,
|
|
"inside.mp4": 60.0,
|
|
})
|
|
find_clips([old, edge, inside], start, end, probe)
|
|
assert "old.mp4" not in probe.calls
|
|
assert "edge.mp4" in probe.calls
|
|
assert "inside.mp4" in probe.calls
|
|
|
|
|
|
# ----------------------------------------------------------------- clip_batch
|
|
|
|
BASE = "http://192.168.192.253:8765"
|
|
CLIP_A = "/2026-09-26/09-00-00.mp4"
|
|
CLIP_B = "/2026-09-26/09-04-00.mp4"
|
|
START_ISO = "2026-09-26T09:03:00"
|
|
END_ISO = "2026-09-26T09:05:30"
|
|
|
|
|
|
def test_clip_batch_no_recording_raises(monkeypatch, tmp_path):
|
|
monkeypatch.setattr(clip, "list_movies", lambda *a, **k: [])
|
|
monkeypatch.setattr(clip, "find_clips", lambda *a, **k: [])
|
|
with pytest.raises(ClipError, match="Tidak ada rekaman motionEye"):
|
|
clip.clip_batch(BASE, 1, START_ISO, END_ISO, str(tmp_path / "out.mp4"))
|
|
|
|
|
|
def fake_run_factory(calls, captured):
|
|
def fake_run(args, timeout):
|
|
args = list(args)
|
|
calls.append((args, timeout))
|
|
if "concat" in args:
|
|
list_file = args[args.index("-i") + 1]
|
|
with open(list_file) as handle:
|
|
captured["concat_list"] = handle.read()
|
|
target = args[-1]
|
|
os.makedirs(os.path.dirname(target), exist_ok=True)
|
|
with open(target, "wb") as handle:
|
|
handle.write(b"piece-bytes")
|
|
return fake_run
|
|
|
|
|
|
def patch_batch_env(monkeypatch, movies, duration=600.0):
|
|
monkeypatch.setattr(clip, "list_movies", lambda *a, **k: list(movies))
|
|
monkeypatch.setattr(clip, "probe_duration", lambda url, timeout=30: duration)
|
|
|
|
|
|
def test_clip_batch_single_piece_argv(monkeypatch, tmp_path):
|
|
movies = [{"path": CLIP_A, "start": dt("2026-09-26 09:00:00")}]
|
|
patch_batch_env(monkeypatch, movies, duration=600.0)
|
|
calls, captured = [], {}
|
|
monkeypatch.setattr(clip, "_run_ffmpeg", fake_run_factory(calls, captured))
|
|
out = tmp_path / "sub" / "batch.mp4"
|
|
clip.clip_batch(BASE, 1, "2026-09-26T09:00:10", "2026-09-26T09:00:50", str(out),
|
|
pad_seconds=0, align_osd=False)
|
|
assert len(calls) == 1
|
|
args, timeout = calls[0]
|
|
assert timeout == 180
|
|
piece = args[-1]
|
|
assert args[:-1] == [
|
|
"ffmpeg", "-y",
|
|
"-ss", "10.000",
|
|
"-to", "50.000",
|
|
"-i", BASE + "/movie/1/download/2026-09-26/09-00-00.mp4",
|
|
"-c", "copy",
|
|
]
|
|
assert piece.endswith(".mp4")
|
|
# temp dir created next to out_path's dir
|
|
assert os.path.dirname(piece).startswith(str(tmp_path / "sub") + os.sep)
|
|
assert out.exists() and out.read_bytes() == b"piece-bytes"
|
|
assert not os.path.exists(os.path.dirname(piece)) # temp dir cleaned up
|
|
|
|
|
|
def test_clip_batch_two_pieces_trim_and_concat(monkeypatch, tmp_path):
|
|
movies = [
|
|
{"path": CLIP_B, "start": dt("2026-09-26 09:04:00")},
|
|
{"path": CLIP_A, "start": dt("2026-09-26 09:00:00")},
|
|
]
|
|
patch_batch_env(monkeypatch, movies, duration=600.0)
|
|
calls, captured = [], {}
|
|
monkeypatch.setattr(clip, "_run_ffmpeg", fake_run_factory(calls, captured))
|
|
out = tmp_path / "batch.mp4"
|
|
clip.clip_batch(BASE, 1, START_ISO, END_ISO, str(out), timeout=60,
|
|
pad_seconds=0, align_osd=False)
|
|
assert len(calls) == 3
|
|
(piece_a, timeout_a), (piece_b, _), (concat_args, timeout_b) = calls
|
|
assert timeout_a == timeout_b == 60
|
|
assert piece_a[:8] == [
|
|
"ffmpeg", "-y", "-ss", "180.000", "-to", "330.000", "-i",
|
|
BASE + "/movie/1/download/2026-09-26/09-00-00.mp4",
|
|
]
|
|
assert piece_a[8:10] == ["-c", "copy"]
|
|
assert piece_b[:8] == [
|
|
"ffmpeg", "-y", "-ss", "0.000", "-to", "90.000", "-i",
|
|
BASE + "/movie/1/download/2026-09-26/09-04-00.mp4",
|
|
]
|
|
assert piece_b[8:10] == ["-c", "copy"]
|
|
assert concat_args == [
|
|
"ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i",
|
|
concat_args[concat_args.index("-i") + 1],
|
|
"-c", "copy", str(out),
|
|
]
|
|
lines = captured["concat_list"].splitlines()
|
|
assert len(lines) == 2
|
|
assert all(line.startswith("file '") and line.endswith(".mp4'") for line in lines)
|
|
assert lines[0].endswith("piece_000.mp4'")
|
|
assert lines[1].endswith("piece_001.mp4'")
|
|
assert out.exists()
|
|
|
|
|
|
def test_clip_batch_ffmpeg_failure_raises_cliperror(monkeypatch, tmp_path):
|
|
movies = [{"path": CLIP_A, "start": dt("2026-09-26 09:00:00")}]
|
|
patch_batch_env(monkeypatch, movies, duration=600.0)
|
|
|
|
def failing_run(args, timeout):
|
|
raise ClipError("ffmpeg gagal: ...Connection refused")
|
|
|
|
monkeypatch.setattr(clip, "_run_ffmpeg", failing_run)
|
|
with pytest.raises(ClipError, match="Connection refused"):
|
|
clip.clip_batch(BASE, 1, "2026-09-26T09:00:10", "2026-09-26T09:00:50",
|
|
str(tmp_path / "out.mp4"))
|
|
|
|
|
|
def test_clip_batch_bad_iso_raises_cliperror(tmp_path):
|
|
with pytest.raises(ClipError):
|
|
clip.clip_batch(BASE, 1, "bukan-tanggal", "2026-09-26T10:00:00",
|
|
str(tmp_path / "out.mp4"))
|
|
|
|
|
|
# ------------------------------------------------------- pad + OSD alignment
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_osd_cache():
|
|
clip._OSD_CACHE.clear()
|
|
yield
|
|
clip._OSD_CACHE.clear()
|
|
|
|
|
|
def _one_movie(monkeypatch, start_iso, duration):
|
|
movies = [{"path": CLIP_A, "start": dt("2026-09-26 09:00:00")}]
|
|
patch_batch_env(monkeypatch, movies, duration=duration)
|
|
|
|
|
|
def test_pad_extends_selection_window(monkeypatch, tmp_path):
|
|
# clip 09:00:00 + 5 s ends exactly at batch start → pad 0 finds nothing
|
|
movies = [{"path": CLIP_A, "start": dt("2026-09-26 09:00:00")}]
|
|
patch_batch_env(monkeypatch, movies, duration=5.0)
|
|
calls, captured = [], {}
|
|
monkeypatch.setattr(clip, "_run_ffmpeg", fake_run_factory(calls, captured))
|
|
with pytest.raises(ClipError, match="Tidak ada rekaman"):
|
|
clip.clip_batch(BASE, 1, "2026-09-26T09:00:05", "2026-09-26T09:00:10",
|
|
str(tmp_path / "a.mp4"), pad_seconds=0, align_osd=False)
|
|
clip.clip_batch(BASE, 1, "2026-09-26T09:00:05", "2026-09-26T09:00:10",
|
|
str(tmp_path / "b.mp4"), pad_seconds=3, align_osd=False)
|
|
args = calls[0][0]
|
|
assert args[args.index("-ss") + 1] == "2.000" # window_start 09:00:02
|
|
assert args[args.index("-to") + 1] == "5.000" # clip ends 09:00:05
|
|
|
|
|
|
def test_osd_alignment_corrects_start_offset(monkeypatch, tmp_path):
|
|
_one_movie(monkeypatch, None, 600.0)
|
|
calls, captured = [], {}
|
|
monkeypatch.setattr(clip, "_run_ffmpeg", fake_run_factory(calls, captured))
|
|
# batch 09:01:10→09:02:00 (pad 0): filename off0=70, off1=120.
|
|
# OSD at media 70 reads desired+14 → shift to 56; at 56 reads desired.
|
|
reads = {
|
|
70: dt("2026-09-26 09:01:24"),
|
|
56: dt("2026-09-26 09:01:10"),
|
|
120: dt("2026-09-26 09:02:00"),
|
|
}
|
|
|
|
def fake_osd(url, media_offset, timeout=45):
|
|
return reads.get(int(round(media_offset)))
|
|
|
|
monkeypatch.setattr(clip, "osd_time_at", fake_osd)
|
|
out = tmp_path / "out.mp4"
|
|
clip.clip_batch(BASE, 1, "2026-09-26T09:01:10", "2026-09-26T09:02:00",
|
|
str(out), pad_seconds=0, align_osd=True)
|
|
args = calls[0][0]
|
|
assert args[args.index("-ss") + 1] == "56.000"
|
|
assert args[args.index("-to") + 1] == "120.000"
|
|
|
|
|
|
def test_align_disabled_never_probes_osd(monkeypatch, tmp_path):
|
|
_one_movie(monkeypatch, None, 600.0)
|
|
calls, captured = [], {}
|
|
monkeypatch.setattr(clip, "_run_ffmpeg", fake_run_factory(calls, captured))
|
|
|
|
def boom(*a, **k):
|
|
raise AssertionError("osd_time_at must not be called")
|
|
|
|
monkeypatch.setattr(clip, "osd_time_at", boom)
|
|
clip.clip_batch(BASE, 1, "2026-09-26T09:00:10", "2026-09-26T09:00:50",
|
|
str(tmp_path / "o.mp4"), pad_seconds=0, align_osd=False)
|
|
args = calls[0][0]
|
|
assert args[args.index("-ss") + 1] == "10.000"
|
|
|
|
|
|
def test_osd_failure_keeps_filename_offsets(monkeypatch, tmp_path):
|
|
_one_movie(monkeypatch, None, 600.0)
|
|
calls, captured = [], {}
|
|
monkeypatch.setattr(clip, "_run_ffmpeg", fake_run_factory(calls, captured))
|
|
monkeypatch.setattr(clip, "osd_time_at", lambda *a, **k: None)
|
|
clip.clip_batch(BASE, 1, "2026-09-26T09:00:10", "2026-09-26T09:00:50",
|
|
str(tmp_path / "o.mp4"), pad_seconds=0, align_osd=True)
|
|
args = calls[0][0]
|
|
assert args[args.index("-ss") + 1] == "10.000"
|
|
assert args[args.index("-to") + 1] == "50.000"
|
|
|
|
|
|
def test_osd_alignment_probe_iteration_cap(monkeypatch, tmp_path):
|
|
_one_movie(monkeypatch, None, 600.0)
|
|
calls, captured = [], {}
|
|
monkeypatch.setattr(clip, "_run_ffmpeg", fake_run_factory(calls, captured))
|
|
probes = []
|
|
|
|
def always_off(url, media_offset, timeout=45):
|
|
probes.append(int(round(media_offset)))
|
|
return dt("2026-09-26 09:01:24") # always +14 vs desired 09:01:10
|
|
|
|
monkeypatch.setattr(clip, "osd_time_at", always_off)
|
|
clip.clip_batch(BASE, 1, "2026-09-26T09:01:10", "2026-09-26T09:02:00",
|
|
str(tmp_path / "o.mp4"), pad_seconds=0, align_osd=True)
|
|
# start align: at most 2 probes; end align: 2 more (same cap)
|
|
assert len(probes) <= 4
|
|
start_probes = [p for p in probes if p in (70, 56, 42)]
|
|
assert len(start_probes) <= 2
|
|
args = calls[0][0]
|
|
# after two corrections 70 -> 56 -> 42
|
|
assert args[args.index("-ss") + 1] == "42.000"
|
|
|
|
|
|
def test_osd_date_mismatch_corrects_once_then_stops(monkeypatch, tmp_path):
|
|
# batch spans New Year's Eve; OSD read lands on the wrong calendar day
|
|
movies = [{"path": "/2026-12-31/23-00-00.mp4",
|
|
"start": dt("2026-12-31 23:00:00")}]
|
|
monkeypatch.setattr(clip, "list_movies", lambda *a, **k: list(movies))
|
|
monkeypatch.setattr(clip, "probe_duration", lambda url, timeout=30: 7200.0)
|
|
calls, captured = [], {}
|
|
monkeypatch.setattr(clip, "_run_ffmpeg", fake_run_factory(calls, captured))
|
|
probes = []
|
|
|
|
def wrong_day(url, media_offset, timeout=45):
|
|
off = int(round(media_offset))
|
|
probes.append(off)
|
|
if off == 1924:
|
|
return dt("2027-01-01 00:32:04") # +1 h vs desired Dec31 23:32:04
|
|
if off == 0:
|
|
return dt("2027-01-01 00:31:50") # still Jan 1 after correction
|
|
return None # end-offset probes do nothing
|
|
|
|
monkeypatch.setattr(clip, "osd_time_at", wrong_day)
|
|
clip.clip_batch(BASE, 1, "2026-12-31T23:32:04", "2026-12-31T23:35:00",
|
|
str(tmp_path / "o.mp4"), pad_seconds=0, align_osd=True)
|
|
# start align: probe at 1924 → corrects (clamped to 0), probe at 0 has
|
|
# wrong date again → rejected by date check (2 probes max).
|
|
assert probes[:2] == [1924, 0]
|
|
args = calls[0][0]
|
|
assert args[args.index("-ss") + 1] == "0.000"
|
|
assert args[args.index("-to") + 1] == "2100.000" # end offset untouched
|
|
|
|
|
|
def test_alignment_span_collapse_falls_back(monkeypatch, tmp_path):
|
|
_one_movie(monkeypatch, None, 600.0)
|
|
calls, captured = [], {}
|
|
monkeypatch.setattr(clip, "_run_ffmpeg", fake_run_factory(calls, captured))
|
|
|
|
def collapse(url, media_offset, timeout=45):
|
|
# readings that would push off1 before off0
|
|
return dt("2026-09-26 09:02:59") # end desired 09:00:50 → huge shift
|
|
|
|
monkeypatch.setattr(clip, "osd_time_at", collapse)
|
|
clip.clip_batch(BASE, 1, "2026-09-26T09:00:10", "2026-09-26T09:00:50",
|
|
str(tmp_path / "o.mp4"), pad_seconds=0, align_osd=True)
|
|
args = calls[0][0]
|
|
# fallback keeps filename-based offsets
|
|
assert args[args.index("-ss") + 1] == "10.000"
|
|
assert args[args.index("-to") + 1] == "50.000"
|