fenrir/tests/unit/test_dectalk_usb_driver.py

import threading
import time
from unittest.mock import Mock

import pytest

from fenrirscreenreader.speechDriver import dectalkUsbDriver


class FakeUsbDevice:
    def __init__(self):
        self.writes = []

    def get_active_configuration(self):
        return object()

    def write(self, endpoint, data, timeout=None):
        payload = bytes(data)
        self.writes.append((endpoint, payload, timeout))
        return len(payload)


class FakeUsbCore:
    USBError = OSError

    class NoBackendError(ValueError):
        pass

    def __init__(self, device, find_error=None):
        self.device = device
        self.find_error = find_error
        self.find_calls = []

    def find(self, **kwargs):
        self.find_calls.append(kwargs)
        if self.find_error:
            raise self.find_error
        return self.device


class FakeUsbUtil:
    def __init__(
        self,
        claim_error=None,
        release_error=None,
        dispose_error=None,
    ):
        self.claim_error = claim_error
        self.release_error = release_error
        self.dispose_error = dispose_error
        self.claimed = []
        self.released = []
        self.disposed = []

    def claim_interface(self, device, interface):
        if self.claim_error:
            raise self.claim_error
        self.claimed.append((device, interface))

    def release_interface(self, device, interface):
        if self.release_error:
            raise self.release_error
        self.released.append((device, interface))

    def dispose_resources(self, device):
        if self.dispose_error:
            raise self.dispose_error
        self.disposed.append(device)


class BlockingUsbDevice(FakeUsbDevice):
    def __init__(self):
        super().__init__()
        self.normal_write_started = threading.Event()
        self.release_normal_write = threading.Event()
        self.immediate_write_finished = threading.Event()

    def write(self, endpoint, data, timeout=None):
        if endpoint == 0x02:
            self.normal_write_started.set()
            self.release_normal_write.wait(timeout=1.0)
        result = super().write(endpoint, data, timeout=timeout)
        if endpoint == 0x04:
            self.immediate_write_finished.set()
        return result


class CancelDuringSpeechUsbDevice(FakeUsbDevice):
    def __init__(self):
        super().__init__()
        self.first_normal_write_started = threading.Event()
        self.release_first_normal_write = threading.Event()

    def write(self, endpoint, data, timeout=None):
        if endpoint == 0x02 and not self.first_normal_write_started.is_set():
            self.first_normal_write_started.set()
            self.release_first_normal_write.wait(timeout=1.0)
        return super().write(endpoint, data, timeout=timeout)


def build_environment():
    return {
        "runtime": {
            "SettingsManager": Mock(),
            "DebugManager": Mock(),
        }
    }


def initialized_driver(monkeypatch, clear_startup_writes=True):
    device = FakeUsbDevice()
    usb_core = FakeUsbCore(device)
    usb_util = FakeUsbUtil()
    speech_driver = dectalkUsbDriver.driver()
    monkeypatch.setattr(
        speech_driver,
        "_load_usb_modules",
        lambda: (usb_core, usb_util),
    )
    speech_driver.initialize(build_environment())
    if clear_startup_writes:
        device.writes.clear()
    return speech_driver, device, usb_core, usb_util


def wait_for_writes(device, count, timeout=1.0):
    deadline = time.monotonic() + timeout
    while len(device.writes) < count and time.monotonic() < deadline:
        time.sleep(0.01)


def test_dectalk_usb_driver_finds_and_claims_native_device(monkeypatch):
    speech_driver, device, usb_core, usb_util = initialized_driver(monkeypatch)
    try:
        assert usb_core.find_calls == [
            {"idVendor": 0x0DD0, "idProduct": 0x2001}
        ]
        assert usb_util.claimed == [(device, 0)]
        assert speech_driver._is_initialized
    finally:
        speech_driver.shutdown()

    assert usb_util.released == [(device, 0)]
    assert usb_util.disposed == [device]


def test_dectalk_usb_driver_sets_low_buffer_timeout_on_initialize(monkeypatch):
    speech_driver, device, _, _ = initialized_driver(
        monkeypatch,
        clear_startup_writes=False,
    )
    try:
        assert device.writes[0] == (0x02, b"[:timeout 1]", 1000)
    finally:
        speech_driver.shutdown()


def test_dectalk_usb_driver_writes_commands_and_speech(monkeypatch):
    speech_driver, device, _, _ = initialized_driver(monkeypatch)
    try:
        speech_driver.set_rate(1.0)
        speech_driver.set_pitch(1.0)
        speech_driver.set_volume(0.5)
        speech_driver.speak("Hello\nworld")
        wait_for_writes(device, 4)

        assert device.writes == [
            (0x02, b"[:ra 650]", 1000),
            (0x02, b"[:dv ap 350]", 1000),
            (0x02, b"[:dv g5 73]", 1000),
            (0x02, b"Hello world\x0b", 1000),
        ]
    finally:
        speech_driver.shutdown()


def test_dectalk_usb_driver_skips_redundant_setting_writes(monkeypatch):
    speech_driver, device, _, _ = initialized_driver(monkeypatch)
    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)

        assert device.writes == [
            (0x02, b"[:ra 180]", 1000),
            (0x02, b"[:dv ap 122]", 1000),
            (0x02, b"[:dv g5 86]", 1000),
        ]
    finally:
        speech_driver.shutdown()


def test_dectalk_usb_driver_chunks_normal_writes(monkeypatch):
    speech_driver, device, _, _ = initialized_driver(monkeypatch)
    try:
        speech_driver._write_bytes(b"x" * 70, "speech")
        assert device.writes == [
            (0x02, b"x" * 64, 1000),
            (0x02, b"x" * 6, 1000),
        ]
    finally:
        speech_driver.shutdown()


def test_dectalk_usb_driver_uses_immediate_cancel_endpoint(monkeypatch):
    speech_driver, device, _, _ = initialized_driver(monkeypatch)
    try:
        speech_driver.cancel()
        assert device.writes == [(0x04, b"\x06", 1000)]
    finally:
        speech_driver.shutdown()


def test_dectalk_usb_cancel_bypasses_blocked_normal_write(monkeypatch):
    device = BlockingUsbDevice()
    usb_core = FakeUsbCore(device)
    usb_util = FakeUsbUtil()
    speech_driver = dectalkUsbDriver.driver()
    monkeypatch.setattr(
        speech_driver,
        "_load_usb_modules",
        lambda: (usb_core, usb_util),
    )
    speech_driver.initialize(build_environment())
    device.writes.clear()
    normal_thread = threading.Thread(
        target=speech_driver._write_bytes,
        args=(b"normal speech", "speech"),
    )
    cancel_thread = threading.Thread(target=speech_driver.cancel)
    try:
        normal_thread.start()
        assert device.normal_write_started.wait(timeout=0.2)
        cancel_thread.start()
        assert device.immediate_write_finished.wait(timeout=0.2)
    finally:
        device.release_normal_write.set()
        normal_thread.join(timeout=1.0)
        cancel_thread.join(timeout=1.0)
        speech_driver.shutdown()


def test_dectalk_usb_cancel_stops_current_speech_upload(monkeypatch):
    device = CancelDuringSpeechUsbDevice()
    usb_core = FakeUsbCore(device)
    usb_util = FakeUsbUtil()
    speech_driver = dectalkUsbDriver.driver()
    monkeypatch.setattr(
        speech_driver,
        "_load_usb_modules",
        lambda: (usb_core, usb_util),
    )
    speech_driver.initialize(build_environment())
    device.writes.clear()
    device.first_normal_write_started.clear()
    normal_thread = threading.Thread(
        target=speech_driver._write_bytes,
        args=(b"x" * 192, "speech"),
    )
    try:
        normal_thread.start()
        assert device.first_normal_write_started.wait(timeout=0.2)
        speech_driver.cancel()
    finally:
        device.release_first_normal_write.set()
        normal_thread.join(timeout=1.0)
        speech_driver.shutdown()

    normal_writes = [
        payload for endpoint, payload, _ in device.writes if endpoint == 0x02
    ]
    assert normal_writes == [b"x" * 64]


def test_dectalk_usb_driver_reports_missing_pyusb(monkeypatch):
    speech_driver = dectalkUsbDriver.driver()

    def missing_pyusb():
        raise ImportError("No module named usb")

    monkeypatch.setattr(speech_driver, "_load_usb_modules", missing_pyusb)

    with pytest.raises(RuntimeError, match="PyUSB"):
        speech_driver.initialize(build_environment())


def test_dectalk_usb_driver_reports_missing_libusb_backend(monkeypatch):
    speech_driver = dectalkUsbDriver.driver()
    backend_error = FakeUsbCore.NoBackendError("No backend available")
    usb_core = FakeUsbCore(None, find_error=backend_error)
    monkeypatch.setattr(
        speech_driver,
        "_load_usb_modules",
        lambda: (usb_core, FakeUsbUtil()),
    )

    with pytest.raises(RuntimeError, match="discovery failed.*backend"):
        speech_driver.initialize(build_environment())


def test_dectalk_usb_driver_reports_missing_device(monkeypatch):
    speech_driver = dectalkUsbDriver.driver()
    usb_core = FakeUsbCore(None)
    usb_util = FakeUsbUtil()
    monkeypatch.setattr(
        speech_driver,
        "_load_usb_modules",
        lambda: (usb_core, usb_util),
    )

    with pytest.raises(RuntimeError, match="0dd0:2001"):
        speech_driver.initialize(build_environment())


def test_dectalk_usb_driver_reports_interface_permission_error(monkeypatch):
    speech_driver = dectalkUsbDriver.driver()
    device = FakeUsbDevice()
    usb_core = FakeUsbCore(device)
    usb_util = FakeUsbUtil(claim_error=PermissionError("access denied"))
    monkeypatch.setattr(
        speech_driver,
        "_load_usb_modules",
        lambda: (usb_core, usb_util),
    )

    with pytest.raises(RuntimeError, match="open failed.*access denied"):
        speech_driver.initialize(build_environment())


def test_dectalk_usb_shutdown_survives_cleanup_errors(monkeypatch):
    device = FakeUsbDevice()
    usb_core = FakeUsbCore(device)
    usb_util = FakeUsbUtil(
        release_error=OSError("release failed"),
        dispose_error=OSError("dispose failed"),
    )
    speech_driver = dectalkUsbDriver.driver()
    monkeypatch.setattr(
        speech_driver,
        "_load_usb_modules",
        lambda: (usb_core, usb_util),
    )
    speech_driver.initialize(build_environment())

    speech_driver.shutdown()

    assert not speech_driver._is_initialized
    assert speech_driver.usb_device is None