feat: 实现 Brain 合成与本地视频输入 (#11)

Brain/tests/input/test_sources.py

@@ -0,0 +1,83 @@
+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)