fenrir/tests/unit/test_hardware_speech_drivers.py

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