File size: 2,888 Bytes
35d483e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 | 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 # noqa: E402
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()
|