| from __future__ import annotations |
|
|
| import io |
| import struct |
| import sys |
| import tempfile |
| import unittest |
| import wave |
| from pathlib import Path |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) |
|
|
| from turn_detection.data import AudioPayloadError, exact_audio_sha256, inspect_audio |
|
|
|
|
| def wav_bytes(*, frames: int = 160, sample_rate: int = 16_000) -> bytes: |
| output = io.BytesIO() |
| with wave.open(output, "wb") as handle: |
| handle.setnchannels(1) |
| handle.setsampwidth(2) |
| handle.setframerate(sample_rate) |
| handle.writeframes(struct.pack("<h", 100) * frames) |
| return output.getvalue() |
|
|
|
|
| def minimal_flac_streaminfo(*, sample_rate: int, channels: int, bits: int, frames: int) -> bytes: |
| streaminfo = bytearray(34) |
| packed = (sample_rate << 44) | ((channels - 1) << 41) | ((bits - 1) << 36) | frames |
| streaminfo[10:18] = packed.to_bytes(8, "big") |
| return b"fLaC" + bytes([0x80]) + len(streaminfo).to_bytes(3, "big") + bytes(streaminfo) |
|
|
|
|
| class AudioInspectionTests(unittest.TestCase): |
| def test_wav_metadata_and_exact_hash(self) -> None: |
| payload = wav_bytes(frames=320) |
| info = inspect_audio({"bytes": payload, "path": "virtual.wav"}) |
| self.assertEqual(info.format, "wav") |
| self.assertEqual(info.sample_rate, 16_000) |
| self.assertEqual(info.num_channels, 1) |
| self.assertEqual(info.num_frames, 320) |
| self.assertAlmostEqual(info.duration_seconds or 0, 0.02) |
| self.assertEqual(info.sha256, exact_audio_sha256(payload)) |
|
|
| def test_embedded_bytes_are_preferred_over_virtual_path(self) -> None: |
| payload = wav_bytes() |
| info = inspect_audio({"bytes": payload, "path": "does-not-exist.flac"}) |
| self.assertEqual(info.format, "wav") |
| self.assertEqual(info.path, "does-not-exist.flac") |
|
|
| def test_relative_path_resolution(self) -> None: |
| with tempfile.TemporaryDirectory() as temporary: |
| path = Path(temporary) / "sample.wav" |
| path.write_bytes(wav_bytes()) |
| from_path = inspect_audio("sample.wav", base_dir=temporary) |
| from_bytes = inspect_audio(path.read_bytes()) |
| self.assertEqual(from_path.sha256, from_bytes.sha256) |
|
|
| def test_flac_streaminfo_without_decoder(self) -> None: |
| payload = minimal_flac_streaminfo(sample_rate=16_000, channels=2, bits=24, frames=32_000) |
| info = inspect_audio(payload) |
| self.assertEqual(info.format, "flac") |
| self.assertEqual(info.sample_rate, 16_000) |
| self.assertEqual(info.num_channels, 2) |
| self.assertEqual(info.bits_per_sample, 24) |
| self.assertEqual(info.duration_seconds, 2.0) |
|
|
| def test_missing_path_is_actionable(self) -> None: |
| with self.assertRaises(AudioPayloadError): |
| inspect_audio({"path": "missing.wav"}) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
