93 lines
3.9 KiB
Python
93 lines
3.9 KiB
Python
"""Audio cancellation/deduplication checks without speakers or Piper."""
|
|
import copy
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
import teddy_speech as speech
|
|
|
|
def state(key="welcome", run="run1"):
|
|
return {"run_id": run, "teddy": {"online": True, "muted": False, "takeover": False,
|
|
"current": {"id": key, "started": 0, "transcript": "Hello player"}}}
|
|
|
|
class SpeechTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.data = patch.object(speech, "DATA", Path(self.temp.name))
|
|
self.data.start()
|
|
self.player = Mock()
|
|
self.follower = speech.Follower(self.player)
|
|
self.job = Mock()
|
|
self.job.poll.return_value = None
|
|
self.spawn = patch.object(speech.subprocess, "Popen", return_value=self.job).start()
|
|
def tearDown(self):
|
|
patch.stopall()
|
|
self.temp.cleanup()
|
|
def test_poll_does_not_repeat_and_stale_generation_cannot_play(self):
|
|
self.follower.update(state())
|
|
self.follower.update(state())
|
|
self.assertEqual(self.spawn.call_count, 1)
|
|
self.follower.update(None)
|
|
self.job.terminate.assert_called_once()
|
|
self.player.play.assert_not_called()
|
|
self.assertEqual(speech.STATUS["status"], "disconnected")
|
|
def test_mute_failure_takeover_reset_and_replacement_cancel(self):
|
|
for gate in ("muted", "takeover", "online", "current", "reset", "replace"):
|
|
with self.subTest(gate=gate):
|
|
self.follower.current = None
|
|
self.follower.job = None
|
|
self.job.reset_mock()
|
|
self.job.poll.return_value = None
|
|
self.follower.update(state())
|
|
changed = copy.deepcopy(state())
|
|
if gate == "reset":
|
|
changed = {"run_id": "run2", "teddy": {"current": None}}
|
|
elif gate == "replace":
|
|
changed = state("new")
|
|
else:
|
|
changed["teddy"][gate] = False if gate == "online" else (None if gate == "current" else True)
|
|
self.follower.update(changed)
|
|
self.job.terminate.assert_called_once()
|
|
self.player.play.assert_not_called()
|
|
def test_completed_generation_plays_once(self):
|
|
self.follower.update(state())
|
|
with speech.wave.open(str(self.follower.output), "wb") as wav:
|
|
wav.setparams((1, 2, 22050, 0, "NONE", "not compressed"))
|
|
wav.writeframes(bytes(44100))
|
|
self.job.poll.return_value = 0
|
|
self.job.returncode = 0
|
|
self.follower.update(state())
|
|
self.follower.update(state())
|
|
self.player.play.assert_called_once()
|
|
self.assertEqual(speech.STATUS["status"], "playing")
|
|
def test_resident_worker_cancelled_result_cannot_play_after_replacement(self):
|
|
from chatterbox_client import Job
|
|
engine = Mock()
|
|
engine.submit.side_effect = lambda text, output: Job(text, output)
|
|
follower = speech.Follower(self.player, engine)
|
|
follower.update(state())
|
|
first = follower.job
|
|
follower.update(state("replacement"))
|
|
second = follower.job
|
|
self.assertTrue(first.cancelled.is_set())
|
|
first.returncode = 0 # Old GPU work finished after replacement.
|
|
follower.update(state("replacement"))
|
|
self.player.play.assert_not_called()
|
|
self.assertIs(follower.job, second)
|
|
engine.close.assert_not_called()
|
|
follower.update(None)
|
|
self.assertTrue(second.cancelled.is_set())
|
|
self.assertEqual(speech.AUDIO, {})
|
|
|
|
def test_synthesis_failure_keeps_caption_only(self):
|
|
self.follower.update(state())
|
|
self.job.poll.return_value = 1
|
|
self.job.returncode = 1
|
|
self.follower.update(state())
|
|
self.assertEqual(speech.STATUS["status"], "error")
|
|
self.player.play.assert_not_called()
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|