142 lines
5.0 KiB
Python
142 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from k1link.compute import prepare_camera_compute_job, validate_camera_compute_job
|
|
from k1link.sessions import SessionIntegrityError, inspect_recorded_media_epoch
|
|
from k1link.web.camera_archive import CameraArchiveWriter
|
|
|
|
|
|
def _recorded_h264_fixture() -> tuple[bytes, bytes]:
|
|
def box(box_type: bytes, payload: bytes = b"") -> bytes:
|
|
return (8 + len(payload)).to_bytes(4, "big") + box_type + payload
|
|
|
|
def full_box(box_type: bytes, payload: bytes = b"", *, flags: int = 0) -> bytes:
|
|
return box(box_type, bytes([0]) + flags.to_bytes(3, "big") + payload)
|
|
|
|
track_id = 1
|
|
sample_duration = 500
|
|
tkhd = full_box(
|
|
b"tkhd",
|
|
b"\x00" * 8 + track_id.to_bytes(4, "big") + b"\x00" * 4,
|
|
)
|
|
mdhd = full_box(
|
|
b"mdhd",
|
|
b"\x00" * 8 + (1_000).to_bytes(4, "big") + b"\x00" * 4,
|
|
)
|
|
hdlr = full_box(b"hdlr", b"\x00" * 4 + b"vide")
|
|
trak = box(b"trak", tkhd + box(b"mdia", mdhd + hdlr))
|
|
trex = full_box(
|
|
b"trex",
|
|
track_id.to_bytes(4, "big")
|
|
+ (1).to_bytes(4, "big")
|
|
+ sample_duration.to_bytes(4, "big")
|
|
+ b"\x00" * 8,
|
|
)
|
|
init = box(b"ftyp", b"isom") + box(
|
|
b"moov",
|
|
trak + box(b"mvex", trex) + box(b"avcC", b"\x01\x64\x00\x28"),
|
|
)
|
|
tfhd = full_box(b"tfhd", track_id.to_bytes(4, "big"), flags=0x020000)
|
|
tfdt = full_box(b"tfdt", (0).to_bytes(4, "big"))
|
|
trun = full_box(b"trun", (1).to_bytes(4, "big"))
|
|
fragment = box(b"moof", box(b"traf", tfhd + tfdt + trun)) + box(b"mdat", b"frame")
|
|
return init, fragment
|
|
|
|
|
|
def _camera_epoch(tmp_path: Path) -> tuple[Path, bytes]:
|
|
session = tmp_path / "session-1"
|
|
session.mkdir()
|
|
init, fragment = _recorded_h264_fixture()
|
|
writer = CameraArchiveWriter(session, "sensor.camera.left", 1)
|
|
writer.append("init", init, host_epoch_ns=1_000_000_000, host_monotonic_ns=2_000_000_000)
|
|
writer.append(
|
|
"media",
|
|
fragment,
|
|
host_epoch_ns=1_500_000_000,
|
|
host_monotonic_ns=2_500_000_000,
|
|
)
|
|
writer.close()
|
|
return session, fragment
|
|
|
|
|
|
def test_camera_compute_job_is_content_addressed_and_idempotent(tmp_path: Path) -> None:
|
|
session, fragment = _camera_epoch(tmp_path)
|
|
source_epoch = session / "media" / "sensor.camera.left" / "epoch-1"
|
|
source_identity = {
|
|
path.relative_to(source_epoch).as_posix(): hashlib.sha256(path.read_bytes()).hexdigest()
|
|
for path in source_epoch.rglob("*")
|
|
if path.is_file()
|
|
}
|
|
|
|
job = prepare_camera_compute_job(
|
|
session_root=session,
|
|
source_id="sensor.camera.left",
|
|
codec_epoch=1,
|
|
origin_epoch_ns=1_000_000_000,
|
|
origin_monotonic_ns=2_000_000_000,
|
|
output_root=tmp_path / "jobs",
|
|
)
|
|
repeated = prepare_camera_compute_job(
|
|
session_root=session,
|
|
source_id="sensor.camera.left",
|
|
codec_epoch=1,
|
|
origin_epoch_ns=1_000_000_000,
|
|
origin_monotonic_ns=2_000_000_000,
|
|
output_root=tmp_path / "jobs",
|
|
)
|
|
|
|
assert repeated == job
|
|
assert job.job_id == f"recorded-camera-{job.input_sha256[:24]}"
|
|
assert job.segment_count == 1
|
|
assert job.timeline_start_seconds == 0.0
|
|
assert job.timeline_end_seconds == 0.5
|
|
copied = (
|
|
job.job_root / "input" / "camera" / "sensor.camera.left" / "epoch-1" / "segments" / "1.m4s"
|
|
)
|
|
assert copied.read_bytes() == fragment
|
|
assert validate_camera_compute_job(job.job_root) == job
|
|
assert source_identity == {
|
|
path.relative_to(source_epoch).as_posix(): hashlib.sha256(path.read_bytes()).hexdigest()
|
|
for path in source_epoch.rglob("*")
|
|
if path.is_file()
|
|
}
|
|
|
|
|
|
def test_camera_compute_job_rejects_changed_published_payload(tmp_path: Path) -> None:
|
|
session, _ = _camera_epoch(tmp_path)
|
|
job = prepare_camera_compute_job(
|
|
session_root=session,
|
|
source_id="sensor.camera.left",
|
|
codec_epoch=1,
|
|
origin_epoch_ns=1_000_000_000,
|
|
origin_monotonic_ns=2_000_000_000,
|
|
output_root=tmp_path / "jobs",
|
|
)
|
|
segment = next(job.job_root.glob("input/camera/*/epoch-1/segments/1.m4s"))
|
|
segment.write_bytes(b"changed")
|
|
|
|
with pytest.raises(SessionIntegrityError, match="payload identity changed"):
|
|
validate_camera_compute_job(job.job_root)
|
|
|
|
|
|
def test_epoch_inspection_checks_summary_aggregate_digests(tmp_path: Path) -> None:
|
|
session, _ = _camera_epoch(tmp_path)
|
|
epoch = session / "media" / "sensor.camera.left" / "epoch-1"
|
|
summary_path = epoch / "summary.json"
|
|
summary = json.loads(summary_path.read_text(encoding="utf-8"))
|
|
summary["index_sha256"] = "0" * 64
|
|
summary_path.write_text(json.dumps(summary), encoding="utf-8")
|
|
|
|
with pytest.raises(SessionIntegrityError, match="index digest changed"):
|
|
inspect_recorded_media_epoch(
|
|
epoch,
|
|
expected_source_name="sensor.camera.left",
|
|
origin_epoch_ns=1_000_000_000,
|
|
origin_monotonic_ns=2_000_000_000,
|
|
)
|