Storm Dragon
79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
import sys
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
|
|
|
from cthulhu import piperfactory
|
|
from cthulhu import speechserver
|
|
|
|
|
|
class PiperFactoryRateMappingTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.server = piperfactory.SpeechServer.__new__(piperfactory.SpeechServer)
|
|
|
|
def test_default_rate_maps_to_native_piper_speed(self):
|
|
self.assertEqual(1.0, self.server._mapRate(50))
|
|
|
|
def test_rate_scale_uses_full_cthulhu_range(self):
|
|
self.assertEqual(2.0, self.server._mapRate(0))
|
|
self.assertEqual(0.25, self.server._mapRate(100))
|
|
|
|
def test_high_screen_reader_rate_is_substantially_faster(self):
|
|
self.assertAlmostEqual(0.415, self.server._mapRate(89), places=3)
|
|
|
|
def test_rate_values_are_clamped(self):
|
|
self.assertEqual(2.0, self.server._mapRate(-1))
|
|
self.assertEqual(0.25, self.server._mapRate(101))
|
|
|
|
def _make_say_all_server(self):
|
|
server = piperfactory.SpeechServer.__new__(piperfactory.SpeechServer)
|
|
server._lock = threading.Lock()
|
|
server._stopEvent = threading.Event()
|
|
server._speakGeneration = 1
|
|
server._currentFuture = None
|
|
server._executor = mock.Mock()
|
|
server._audioPlayer = None
|
|
return server
|
|
|
|
def _make_say_all_context(self, utterance):
|
|
return speechserver.SayAllContext(
|
|
mock.Mock(),
|
|
utterance,
|
|
0,
|
|
len(utterance),
|
|
)
|
|
|
|
def test_stop_invalidates_pending_say_all_worker(self):
|
|
server = self._make_say_all_server()
|
|
iterator = iter([[self._make_say_all_context("first"), None]])
|
|
callback = mock.Mock()
|
|
|
|
server.stop()
|
|
result = server._sayAllWorker(iterator, callback, 1)
|
|
|
|
self.assertFalse(result)
|
|
callback.assert_not_called()
|
|
server._executor.submit.assert_not_called()
|
|
|
|
def test_stale_say_all_completion_does_not_advance(self):
|
|
server = self._make_say_all_server()
|
|
first = self._make_say_all_context("first")
|
|
second = self._make_say_all_context("second")
|
|
iterator = iter([[first, None], [second, None]])
|
|
callback = mock.Mock()
|
|
|
|
server._sayAllWorker(iterator, callback, 1)
|
|
onComplete = server._executor.submit.call_args.args[3]
|
|
server.stop()
|
|
result = onComplete()
|
|
|
|
self.assertFalse(result)
|
|
self.assertEqual(1, server._executor.submit.call_count)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|