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()