84 lines
3.2 KiB
Python
84 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from yovision_brain.config import parse_input_config
|
|
from yovision_brain.input import CancellationToken, InputError, LocalFileInput, SyntheticInput, build_input_source
|
|
|
|
|
|
def synthetic_config(seed: int = 9, frame_count: int = 3):
|
|
return parse_input_config(
|
|
{
|
|
"schema": "brain.internal.input/v1",
|
|
"logical_device_id": "camera-lab-01",
|
|
"profile": {"id": "main", "width": 3, "height": 2, "fps": 4},
|
|
"source": {"kind": "synthetic", "seed": seed, "frame_count": frame_count},
|
|
}
|
|
)
|
|
|
|
|
|
def local_config(path: Path, *, chunk_size: int = 4):
|
|
return parse_input_config(
|
|
{
|
|
"schema": "brain.internal.input/v1",
|
|
"logical_device_id": "local-video-01",
|
|
"profile": {"id": "archive", "width": 1920, "height": 1080, "fps": 25},
|
|
"source": {"kind": "local_file", "path": str(path), "chunk_size": chunk_size},
|
|
}
|
|
)
|
|
|
|
|
|
def test_synthetic_input_is_deterministic_and_carries_metadata() -> None:
|
|
first = list(SyntheticInput(synthetic_config()).packets())
|
|
second = list(build_input_source(synthetic_config()).packets())
|
|
different = list(SyntheticInput(synthetic_config(seed=10)).packets())
|
|
|
|
assert first == second
|
|
assert [packet.sequence for packet in first] == [0, 1, 2]
|
|
assert [packet.timestamp_ns for packet in first] == [0, 250_000_000, 500_000_000]
|
|
assert first[0].logical_device_id == "camera-lab-01"
|
|
assert first[0].profile_id == "main"
|
|
assert first[0].media_format == "rgb24"
|
|
assert len(first[0].payload) == 3 * 2 * 3
|
|
assert first[0].payload != different[0].payload
|
|
|
|
|
|
def test_synthetic_input_honors_cancellation() -> None:
|
|
token = CancellationToken()
|
|
packets = SyntheticInput(synthetic_config(frame_count=10)).packets(token)
|
|
assert next(packets).sequence == 0
|
|
token.cancel()
|
|
assert list(packets) == []
|
|
|
|
|
|
def test_local_file_input_reads_chunks_and_finishes_at_eof(tmp_path: Path) -> None:
|
|
video = tmp_path / "anonymous-fixture.bin"
|
|
video.write_bytes(b"abcdefghij")
|
|
source = LocalFileInput(local_config(video))
|
|
packets = list(source.packets())
|
|
|
|
assert source.source_label == "anonymous-fixture.bin"
|
|
assert [packet.payload for packet in packets] == [b"abcd", b"efgh", b"ij"]
|
|
assert [packet.sequence for packet in packets] == [0, 1, 2]
|
|
assert all(packet.timestamp_ns is None for packet in packets)
|
|
assert all(packet.media_format == "container-bytes" for packet in packets)
|
|
|
|
|
|
def test_local_file_input_honors_cancellation(tmp_path: Path) -> None:
|
|
video = tmp_path / "anonymous-fixture.bin"
|
|
video.write_bytes(b"abcdefghij")
|
|
token = CancellationToken()
|
|
packets = LocalFileInput(local_config(video)).packets(token)
|
|
assert next(packets).payload == b"abcd"
|
|
token.cancel()
|
|
assert list(packets) == []
|
|
|
|
|
|
def test_missing_file_error_is_actionable_without_absolute_path(tmp_path: Path) -> None:
|
|
missing = tmp_path / "missing-video.mp4"
|
|
with pytest.raises(InputError, match="local video source not found: missing-video.mp4") as caught:
|
|
list(LocalFileInput(local_config(missing)).packets())
|
|
assert str(tmp_path) not in str(caught.value)
|