Storm Dragon
151 lines
4.3 KiB
Python
151 lines
4.3 KiB
Python
import importlib.util
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from fenrirscreenreader.core.outputManager import OutputManager
|
|
|
|
|
|
def build_output_manager():
|
|
settings_manager = Mock()
|
|
settings_manager.get_setting_as_bool.return_value = True
|
|
settings_manager.get_setting_as_float.return_value = 1.0
|
|
sound_driver = Mock()
|
|
speech_driver = Mock()
|
|
output_manager = OutputManager()
|
|
output_manager.env = {
|
|
"soundIcons": {
|
|
"ACCEPT": "/tmp/Accept.wav",
|
|
"ERRORSCREEN": "/tmp/ErrorScreen.wav",
|
|
},
|
|
"runtime": {
|
|
"SettingsManager": settings_manager,
|
|
"SoundDriver": sound_driver,
|
|
"SpeechDriver": speech_driver,
|
|
"DebugManager": Mock(write_debug_out=Mock()),
|
|
},
|
|
}
|
|
return output_manager, sound_driver, speech_driver
|
|
|
|
|
|
def load_key_interrupt_module():
|
|
module_path = (
|
|
Path(__file__).resolve().parents[2]
|
|
/ "src"
|
|
/ "fenrirscreenreader"
|
|
/ "commands"
|
|
/ "onKeyInput"
|
|
/ "10000-shut_up.py"
|
|
)
|
|
spec = importlib.util.spec_from_file_location(
|
|
"fenrir_key_interrupt", module_path
|
|
)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_present_text_allows_sound_only_feedback():
|
|
output_manager, sound_driver, _speech_driver = build_output_manager()
|
|
|
|
output_manager.present_text("", sound_icon="Accept", interrupt=False)
|
|
|
|
sound_driver.play_sound_file.assert_called_once_with(
|
|
"/tmp/Accept.wav", False
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_play_sound_supports_error_alias():
|
|
output_manager, sound_driver, _speech_driver = build_output_manager()
|
|
|
|
assert output_manager.play_sound("Error") is True
|
|
|
|
sound_driver.play_sound_file.assert_called_once_with(
|
|
"/tmp/ErrorScreen.wav", True
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_interrupt_output_async_does_not_block_on_slow_cancel():
|
|
output_manager, _sound_driver, speech_driver = build_output_manager()
|
|
interrupt_started = threading.Event()
|
|
release_interrupt = threading.Event()
|
|
|
|
def slow_cancel():
|
|
interrupt_started.set()
|
|
release_interrupt.wait(timeout=1.0)
|
|
|
|
speech_driver.cancel.side_effect = slow_cancel
|
|
|
|
start_time = time.monotonic()
|
|
output_manager.interrupt_output_async()
|
|
elapsed = time.monotonic() - start_time
|
|
|
|
try:
|
|
assert interrupt_started.wait(timeout=0.2)
|
|
assert elapsed < 0.2
|
|
output_manager.interrupt_output_async()
|
|
assert speech_driver.cancel.call_count == 1
|
|
finally:
|
|
release_interrupt.set()
|
|
output_manager.interrupt_thread.join(timeout=1.0)
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_interrupt_output_waits_only_briefly_for_slow_cancel():
|
|
output_manager, _sound_driver, speech_driver = build_output_manager()
|
|
interrupt_started = threading.Event()
|
|
release_interrupt = threading.Event()
|
|
|
|
def slow_cancel():
|
|
interrupt_started.set()
|
|
release_interrupt.wait(timeout=1.0)
|
|
|
|
speech_driver.cancel.side_effect = slow_cancel
|
|
|
|
start_time = time.monotonic()
|
|
output_manager.interrupt_output()
|
|
elapsed = time.monotonic() - start_time
|
|
|
|
try:
|
|
assert interrupt_started.wait(timeout=0.2)
|
|
assert elapsed < 0.2
|
|
output_manager.interrupt_output()
|
|
assert speech_driver.cancel.call_count == 1
|
|
finally:
|
|
release_interrupt.set()
|
|
output_manager.interrupt_thread.join(timeout=1.0)
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_key_interrupt_command_uses_nonblocking_interrupt():
|
|
module = load_key_interrupt_module()
|
|
settings_manager = Mock()
|
|
settings_manager.get_setting_as_bool.return_value = True
|
|
settings_manager.get_setting.return_value = ""
|
|
output_manager = Mock()
|
|
env = {
|
|
"input": {
|
|
"curr_input": ["KEY_A"],
|
|
"prev_input": [],
|
|
},
|
|
"runtime": {
|
|
"InputManager": Mock(no_key_pressed=Mock(return_value=False)),
|
|
"OutputManager": output_manager,
|
|
"ScreenManager": Mock(is_screen_change=Mock(return_value=False)),
|
|
"SettingsManager": settings_manager,
|
|
},
|
|
}
|
|
command = module.command()
|
|
command.initialize(env)
|
|
|
|
command.run()
|
|
|
|
output_manager.interrupt_output_async.assert_called_once_with()
|
|
output_manager.interrupt_output.assert_not_called()
|