import os import select import termios import time from unittest.mock import ANY from unittest.mock import Mock import pytest from fenrirscreenreader.speechDriver import dectalkDriver from fenrirscreenreader.speechDriver import doubletalkDriver from fenrirscreenreader.speechDriver import litetalkDriver from fenrirscreenreader.speechDriver import tripletalkDriver def build_environment(device): settings_manager = Mock() settings_manager.get_setting.side_effect = ( lambda section, setting: device if (section, setting) == ("speech", "hardware_device") else "" ) settings_manager.get_setting_as_int.side_effect = ( lambda section, setting: 9600 if (section, setting) == ("speech", "hardware_baud_rate") else 0 ) return { "runtime": { "SettingsManager": settings_manager, "DebugManager": Mock(), } } def read_available(fd, expected_length, timeout=1.0): deadline = time.monotonic() + timeout data = b"" while len(data) < expected_length and time.monotonic() < deadline: readable, _, _ = select.select([fd], [], [], 0.05) if readable: data += os.read(fd, 1024) return data @pytest.fixture def serial_pair(): master_fd, slave_fd = os.openpty() try: yield master_fd, os.ttyname(slave_fd) finally: os.close(master_fd) os.close(slave_fd) def initialized_driver(driver_class, serial_pair): master_fd, slave_name = serial_pair speech_driver = driver_class.driver() speech_driver.initialize(build_environment(slave_name)) assert speech_driver._is_initialized return speech_driver, master_fd def test_dectalk_driver_speaks_printable_text(serial_pair): speech_driver, master_fd = initialized_driver(dectalkDriver, serial_pair) try: speech_driver.speak("Hello\nworld ☃") assert read_available(master_fd, 13) == b"Hello world \x0b" finally: speech_driver.shutdown() def test_dectalk_driver_neutralizes_inline_commands_in_speech( serial_pair, ): speech_driver, master_fd = initialized_driver(dectalkDriver, serial_pair) try: speech_driver.speak("status [:dv g5 10] done") expected = b"status [ :dv g5 10] done\x0b" assert read_available(master_fd, len(expected)) == expected finally: speech_driver.shutdown() def test_dectalk_driver_writes_settings_and_cancel(serial_pair): speech_driver, master_fd = initialized_driver(dectalkDriver, serial_pair) try: speech_driver.set_rate(1.0) speech_driver.set_pitch(1.0) speech_driver.set_volume(0.5) speech_driver.cancel() expected = b"[:ra 650][:dv ap 350][:dv g5 73]\x03" assert read_available(master_fd, len(expected)) == expected finally: speech_driver.shutdown() def test_dectalk_driver_skips_redundant_setting_commands(serial_pair): speech_driver, master_fd = initialized_driver(dectalkDriver, serial_pair) try: speech_driver.set_rate(0.5) speech_driver.set_pitch(0.5) speech_driver.set_volume(1.0) speech_driver.set_rate(0.5) speech_driver.set_pitch(0.5) speech_driver.set_volume(1.0) expected = b"[:ra 180][:dv ap 122][:dv g5 86]" assert read_available(master_fd, len(expected)) == expected assert read_available(master_fd, 1, timeout=0.2) == b"" finally: speech_driver.shutdown() def test_dectalk_driver_uses_native_defaults_at_fenrir_midpoint(serial_pair): speech_driver, master_fd = initialized_driver(dectalkDriver, serial_pair) try: speech_driver.set_rate(0.5) speech_driver.set_pitch(0.5) expected = b"[:ra 180][:dv ap 122]" assert read_available(master_fd, len(expected)) == expected finally: speech_driver.shutdown() def test_dectalk_driver_enables_xon_xoff_flow_control(serial_pair): speech_driver, _ = initialized_driver(dectalkDriver, serial_pair) try: input_flags = termios.tcgetattr(speech_driver.serial_port)[0] assert input_flags & termios.IXON assert not input_flags & termios.IXOFF assert not input_flags & termios.IXANY finally: speech_driver.shutdown() def test_litetalk_driver_speaks_printable_text(serial_pair): speech_driver, master_fd = initialized_driver(litetalkDriver, serial_pair) try: speech_driver.speak("Ready") assert read_available(master_fd, 6) == b"Ready\r" finally: speech_driver.shutdown() def test_litetalk_driver_writes_settings_and_cancel(serial_pair): speech_driver, master_fd = initialized_driver(litetalkDriver, serial_pair) try: speech_driver.set_rate(1.0) speech_driver.set_pitch(0.0) speech_driver.set_volume(0.5) speech_driver.cancel() assert read_available(master_fd, 10) == b"\x019S\x010P\x014V\x18" finally: speech_driver.shutdown() def test_configured_device_supports_classic_serial(serial_pair): master_fd, slave_name = serial_pair speech_driver = litetalkDriver.driver() speech_driver.initialize(build_environment(slave_name)) try: assert speech_driver.device == slave_name speech_driver.speak("Serial") assert read_available(master_fd, 7) == b"Serial\r" finally: speech_driver.shutdown() def test_configured_device_strips_inline_comment(serial_pair): master_fd, slave_name = serial_pair device_setting = f"{slave_name} # built-in serial port" speech_driver = litetalkDriver.driver() speech_driver.initialize(build_environment(device_setting)) try: assert speech_driver.device == slave_name speech_driver.speak("Specific") assert read_available(master_fd, 9) == b"Specific\r" finally: speech_driver.shutdown() def test_auto_device_is_rejected(): speech_driver = litetalkDriver.driver() with pytest.raises(RuntimeError, match="hardware speech device"): speech_driver.initialize(build_environment("auto")) debug_manager = speech_driver.env["runtime"]["DebugManager"] debug_manager.write_debug_out.assert_called_with( "Hardware speech requires an explicit serial device", ANY, on_any_level=True, ) def test_hardware_driver_retries_partial_serial_writes(monkeypatch): speech_driver = litetalkDriver.driver() speech_driver.env = build_environment("/dev/ttyUSB0") speech_driver.serial_port = 12 written_chunks = [] def fake_write(port, data): assert port == 12 chunk = data[:2] written_chunks.append(chunk) return len(chunk) monkeypatch.setattr( "fenrirscreenreader.speechDriver.hardwareSerialDriver.os.write", fake_write, ) speech_driver._write_bytes(b"abcdef", "speech") assert written_chunks == [b"ab", b"cd", b"ef"] def test_hardware_driver_retries_setting_after_failed_write(monkeypatch): speech_driver = dectalkDriver.driver() speech_driver._is_initialized = True write_results = [False, True] written_commands = [] def fake_write(data, description): written_commands.append((data, description)) return write_results.pop(0) monkeypatch.setattr(speech_driver, "_write_bytes", fake_write) speech_driver.set_rate(0.5) speech_driver.set_rate(0.5) assert written_commands == [ (b"[:ra 180]", "rate"), (b"[:ra 180]", "rate"), ] @pytest.mark.parametrize("driver_class", [doubletalkDriver, tripletalkDriver]) def test_litetalk_compatible_alias_drivers(driver_class, serial_pair): speech_driver, master_fd = initialized_driver(driver_class, serial_pair) try: speech_driver.speak("Alias") speech_driver.set_rate(1.0) assert read_available(master_fd, 10) == b"\x019SAlias\r" finally: speech_driver.shutdown() def test_hardware_driver_ignores_empty_and_non_string_text(serial_pair): speech_driver, master_fd = initialized_driver(dectalkDriver, serial_pair) try: speech_driver.speak("") speech_driver.speak(None) assert read_available(master_fd, 1, timeout=0.2) == b"" finally: speech_driver.shutdown()