[音频 Transport]：完成本机音频抽象与回放测试通道，包含内存 Transport、文件回放、可选 sounddevice 适配和环形缓冲

tests/test_transport.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from owner_voice_pet.models import AudioFrame, AudioSegment, ErrorCode
+from owner_voice_pet.transport import (
+    AudioRingBuffer,
+    FileReplayTransport,
+    MemoryAudioTransport,
+    SoundDeviceAudioTransport,
+)
+
+
+def frame(idx: int, timestamp_ms: int, metadata: dict[str, object] | None = None) -> AudioFrame:
+    return AudioFrame(
+        pcm=idx.to_bytes(2, "little", signed=False),
+        sample_rate=16000,
+        channels=1,
+        timestamp_ms=timestamp_ms,
+        frame_id=idx,
+        metadata=metadata or {},
+    )
+
+
+class TransportTests(unittest.TestCase):
+    def test_memory_transport_replays_frames_and_captures_playback(self) -> None:
+        transport = MemoryAudioTransport([frame(1, 0), frame(2, 20)])
+        transport.start_input()
+        self.assertEqual([f.frame_id for f in transport.read_frames(10)], [1])
+        self.assertEqual([f.frame_id for f in transport.read_frames(10)], [2])
+        self.assertEqual(transport.read_frames(10), [])
+
+        result = transport.play_pcm(AudioSegment(b"\x00\x00", 16000, 1, 0, 100))
+        self.assertTrue(result.played)
+        self.assertEqual(len(transport.played_segments), 1)
+
+    def test_memory_transport_reports_missing_output(self) -> None:
+        transport = MemoryAudioTransport(output_available=False)
+        result = transport.play_pcm(AudioSegment(b"\x00\x00", 16000, 1, 0, 100))
+        self.assertFalse(result.played)
+        self.assertEqual(result.error.code, ErrorCode.AUDIO_OUTPUT_DEVICE_MISSING)
+
+    def test_file_replay_roundtrip(self) -> None:
+        frames = [frame(1, 0, {"wake": True}), frame(2, 20, {"speech": True})]
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "fixture.jsonl"
+            FileReplayTransport.write_jsonl(path, frames)
+            transport = FileReplayTransport.from_jsonl(path)
+            transport.start_input()
+            self.assertTrue(transport.read_frames(10)[0].metadata["wake"])
+            self.assertTrue(transport.read_frames(10)[0].metadata["speech"])
+
+    def test_ring_buffer_trims_old_frames_and_requires_order(self) -> None:
+        buffer = AudioRingBuffer(max_duration_ms=40)
+        buffer.extend([frame(1, 0), frame(2, 20), frame(3, 50)])
+        self.assertEqual([f.frame_id for f in buffer.frames()], [2, 3])
+        with self.assertRaises(ValueError):
+            buffer.append(frame(4, 10))
+
+    def test_sounddevice_transport_is_safe_without_optional_dependency(self) -> None:
+        transport = SoundDeviceAudioTransport()
+        health = transport.health()
+        self.assertIsInstance(health.input_available, bool)
+        self.assertIsInstance(health.output_available, bool)
+
+
+if __name__ == "__main__":
+    unittest.main()