55 lines
2.2 KiB
Python
55 lines
2.2 KiB
Python
"""Exercise real audio/session integration with deterministic provider responses."""
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
|
|
import numpy as np
|
|
|
|
from blurt_client import TranscriptionConfig
|
|
from blurt_session import RecordingSession, TranscriptionScheduler
|
|
|
|
|
|
class PipelineTests(unittest.TestCase):
|
|
def test_real_audio_uploads_before_stop_and_seals_final_speech(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
requested = threading.Event()
|
|
calls = []
|
|
completions = []
|
|
|
|
def transcribe(audio, config):
|
|
calls.append((len(audio), config))
|
|
requested.set()
|
|
return f"phrase {len(calls)}"
|
|
|
|
scheduler = TranscriptionScheduler(transcribe)
|
|
config = TranscriptionConfig("test", "en", "openai", "fake")
|
|
session = RecordingSession(48000, 1, config, directory, scheduler,
|
|
on_complete=completions.append)
|
|
try:
|
|
tone = (0.05 * np.sin(2 * np.pi * 220 * np.arange(48000) / 48000)).astype(np.float32)
|
|
for _ in range(9):
|
|
self.assertTrue(session.feed(tone[:, None]))
|
|
for _ in range(2):
|
|
self.assertTrue(session.feed(np.zeros((48000, 1), dtype=np.float32)))
|
|
self.assertTrue(requested.wait(5), "real pause detector did not upload during capture")
|
|
self.assertIsNone(session.result)
|
|
self.assertEqual(completions, [])
|
|
self.assertTrue(session.feed(tone[:, None]))
|
|
session.stop()
|
|
result = session.join(5)
|
|
self.assertIsNotNone(result)
|
|
self.assertTrue(result.complete, result.errors)
|
|
self.assertEqual(result.text, "phrase 1 phrase 2")
|
|
self.assertEqual(sum(length for length, _ in calls), 12 * 16000)
|
|
self.assertTrue(all(snapshot is config for _, snapshot in calls))
|
|
session.acknowledge_delivery()
|
|
self.assertFalse(session.path.exists())
|
|
finally:
|
|
session.stop()
|
|
session.join(5)
|
|
scheduler.shutdown()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|