Storm Dragon
277 lines
8.3 KiB
Python
277 lines
8.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()),
|
|
"TextManager": Mock(
|
|
replace_head_lines=Mock(side_effect=lambda text: text)
|
|
),
|
|
"PunctuationManager": Mock(
|
|
proceed_punctuation=Mock(
|
|
side_effect=lambda text, _ignore_punctuation: text
|
|
)
|
|
),
|
|
},
|
|
}
|
|
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_present_text_sound_icon_with_interrupt_cancels_speech():
|
|
output_manager, sound_driver, speech_driver = build_output_manager()
|
|
|
|
output_manager.present_text(
|
|
"end of screen", sound_icon="Accept", interrupt=True
|
|
)
|
|
|
|
speech_driver.cancel.assert_called_once_with()
|
|
sound_driver.play_sound_file.assert_called_once_with(
|
|
"/tmp/Accept.wav", True
|
|
)
|
|
speech_driver.speak.assert_not_called()
|
|
|
|
|
|
@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_cancel_speech_skips_when_speech_driver_is_busy():
|
|
output_manager, _sound_driver, speech_driver = build_output_manager()
|
|
output_manager.speech_driver_lock_timeout = 0.01
|
|
output_manager.speech_driver_lock.acquire()
|
|
|
|
try:
|
|
start_time = time.monotonic()
|
|
output_manager.cancel_speech()
|
|
elapsed = time.monotonic() - start_time
|
|
finally:
|
|
output_manager.speech_driver_lock.release()
|
|
|
|
assert elapsed < 0.2
|
|
speech_driver.cancel.assert_not_called()
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_speak_text_drops_speech_when_cancel_holds_driver_lock():
|
|
output_manager, _sound_driver, speech_driver = build_output_manager()
|
|
output_manager.speech_driver_lock_timeout = 0.01
|
|
output_manager.speech_driver_lock.acquire()
|
|
|
|
try:
|
|
start_time = time.monotonic()
|
|
output_manager.speak_text("hello", interrupt=False, flush=False)
|
|
elapsed = time.monotonic() - start_time
|
|
finally:
|
|
output_manager.speech_driver_lock.release()
|
|
|
|
assert elapsed < 0.2
|
|
speech_driver.speak.assert_not_called()
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_present_text_records_speech_history_when_enabled():
|
|
output_manager, _sound_driver, speech_driver = build_output_manager()
|
|
speech_history_manager = Mock(add_text=Mock())
|
|
output_manager.env["runtime"]["SpeechHistoryManager"] = (
|
|
speech_history_manager
|
|
)
|
|
|
|
output_manager.present_text("hello history", interrupt=False)
|
|
|
|
speech_history_manager.add_text.assert_called_once_with("hello history")
|
|
speech_driver.speak.assert_called_once()
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_present_text_does_not_record_when_speech_disabled():
|
|
output_manager, _sound_driver, speech_driver = build_output_manager()
|
|
speech_history_manager = Mock(add_text=Mock())
|
|
output_manager.env["runtime"]["SpeechHistoryManager"] = (
|
|
speech_history_manager
|
|
)
|
|
|
|
def _get_setting_as_bool(section, setting):
|
|
if (section, setting) == ("speech", "enabled"):
|
|
return False
|
|
return True
|
|
|
|
output_manager.env["runtime"][
|
|
"SettingsManager"
|
|
].get_setting_as_bool.side_effect = _get_setting_as_bool
|
|
|
|
output_manager.present_text("hello history", interrupt=False)
|
|
|
|
speech_history_manager.add_text.assert_not_called()
|
|
speech_driver.speak.assert_not_called()
|
|
|
|
|
|
@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()
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_key_interrupt_command_ignores_fenrir_shortcuts():
|
|
module = load_key_interrupt_module()
|
|
settings_manager = Mock()
|
|
settings_manager.get_setting_as_bool.return_value = True
|
|
settings_manager.get_setting.return_value = ""
|
|
input_manager = Mock(
|
|
no_key_pressed=Mock(return_value=False),
|
|
get_curr_shortcut=Mock(return_value=[1, ["KEY_KP9"]]),
|
|
get_command_for_shortcut=Mock(return_value="REVIEW_NEXT_LINE"),
|
|
)
|
|
output_manager = Mock()
|
|
env = {
|
|
"input": {
|
|
"curr_input": ["KEY_KP9"],
|
|
"prev_input": [],
|
|
},
|
|
"runtime": {
|
|
"InputManager": input_manager,
|
|
"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_not_called()
|