30e6acea70
Route live recording through ffmpeg MP4 segments and the 5.15 batch subprocess, remove simulated RTSP chain 2, purge expired slices on startup and hourly, and expose TTL settings to the demo client. Co-authored-by: Cursor <cursoragent@cursor.com>
38 lines
1.1 KiB
Python
38 lines
1.1 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from app.services.video.rtsp_segment_cleanup import (
|
|
default_rtsp_segments_root,
|
|
purge_expired_rtsp_segments,
|
|
)
|
|
|
|
|
|
def test_purge_expired_rtsp_segments_removes_old_slice(tmp_path: Path) -> None:
|
|
root = default_rtsp_segments_root(base_dir=tmp_path / "logs")
|
|
mp4 = root / "123456" / "or-cam-03" / "slice_0000.mp4"
|
|
mp4.parent.mkdir(parents=True)
|
|
mp4.write_bytes(b"slice")
|
|
old = time.time() - (25 * 3600)
|
|
os.utime(mp4, (old, old))
|
|
|
|
removed = purge_expired_rtsp_segments(root, ttl_hours=24.0)
|
|
|
|
assert removed == 1
|
|
assert not mp4.exists()
|
|
assert not (root / "123456" / "or-cam-03").exists()
|
|
|
|
|
|
def test_purge_expired_rtsp_segments_keeps_recent_slice(tmp_path: Path) -> None:
|
|
root = default_rtsp_segments_root(base_dir=tmp_path / "logs")
|
|
mp4 = root / "123456" / "or-cam-03" / "slice_0001.mp4"
|
|
mp4.parent.mkdir(parents=True)
|
|
mp4.write_bytes(b"slice")
|
|
|
|
removed = purge_expired_rtsp_segments(root, ttl_hours=24.0)
|
|
|
|
assert removed == 0
|
|
assert mp4.is_file()
|