Storm Dragon
101 lines
3.4 KiB
Python
101 lines
3.4 KiB
Python
import os
|
|
import select
|
|
import sys
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
|
|
|
from cthulhu import hardwarefactory
|
|
from cthulhu import settings
|
|
|
|
|
|
def read_available(fd, expectedLength, timeout=1.0):
|
|
deadline = time.monotonic() + timeout
|
|
data = b""
|
|
while len(data) < expectedLength and time.monotonic() < deadline:
|
|
readable, _, _ = select.select([fd], [], [], 0.05)
|
|
if readable:
|
|
data += os.read(fd, 1024)
|
|
return data
|
|
|
|
|
|
class HardwareFactoryRegressionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self._oldDevice = settings.hardwareSpeechDevice
|
|
self._oldBaudRate = settings.hardwareSpeechBaudRate
|
|
hardwarefactory.SpeechServer.shutdownActiveServers()
|
|
|
|
def tearDown(self):
|
|
hardwarefactory.SpeechServer.shutdownActiveServers()
|
|
settings.hardwareSpeechDevice = self._oldDevice
|
|
settings.hardwareSpeechBaudRate = self._oldBaudRate
|
|
|
|
def test_lists_explicit_synth_choices_without_opening_serial_device(self):
|
|
settings.hardwareSpeechDevice = ""
|
|
|
|
servers = hardwarefactory.SpeechServer.getSpeechServers()
|
|
|
|
self.assertEqual(
|
|
["litetalk", "doubletalk", "tripletalk", "dectalk"],
|
|
[server.getInfo()[1] for server in servers],
|
|
)
|
|
self.assertEqual({}, hardwarefactory.SpeechServer._active_servers)
|
|
self.assertTrue(all(server._driver is None for server in servers))
|
|
|
|
def test_failed_initialization_is_not_cached(self):
|
|
settings.hardwareSpeechDevice = ""
|
|
|
|
self.assertIsNone(
|
|
hardwarefactory.SpeechServer.getSpeechServer(["LiteTalk", "litetalk"])
|
|
)
|
|
self.assertEqual({}, hardwarefactory.SpeechServer._active_servers)
|
|
|
|
masterFd, slaveFd = os.openpty()
|
|
try:
|
|
settings.hardwareSpeechDevice = os.ttyname(slaveFd)
|
|
server = hardwarefactory.SpeechServer.getSpeechServer(
|
|
["LiteTalk", "litetalk"]
|
|
)
|
|
|
|
self.assertIsNotNone(server)
|
|
self.assertIsNotNone(server._driver)
|
|
self.assertIs(
|
|
server,
|
|
hardwarefactory.SpeechServer._active_servers.get("litetalk"),
|
|
)
|
|
finally:
|
|
os.close(masterFd)
|
|
os.close(slaveFd)
|
|
|
|
def test_explicit_synth_choices_write_expected_serial_bytes(self):
|
|
expectedBytes = {
|
|
"litetalk": b"Alias\r",
|
|
"doubletalk": b"Alias\r",
|
|
"tripletalk": b"Alias\r",
|
|
"dectalk": b"Alias\x01",
|
|
}
|
|
|
|
for synthId, expected in expectedBytes.items():
|
|
with self.subTest(synthId=synthId):
|
|
hardwarefactory.SpeechServer.shutdownActiveServers()
|
|
masterFd, slaveFd = os.openpty()
|
|
try:
|
|
settings.hardwareSpeechDevice = os.ttyname(slaveFd)
|
|
server = hardwarefactory.SpeechServer.getSpeechServer(
|
|
["", synthId]
|
|
)
|
|
|
|
self.assertIsNotNone(server)
|
|
server.speak("Alias", interrupt=False)
|
|
self.assertEqual(expected, read_available(masterFd, len(expected)))
|
|
finally:
|
|
hardwarefactory.SpeechServer.shutdownActiveServers()
|
|
os.close(masterFd)
|
|
os.close(slaveFd)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|