Move auxiliary work off the main event loop
Run voice discovery, voice testing, clipboard I/O, and external scripts in isolated background worker lanes. Deliver results through Fenrir's event queue so manager state remains owned by the main loop.
This commit is contained in:
Storm Dragon
@@ -0,0 +1,303 @@
|
||||
from unittest.mock import Mock
|
||||
|
||||
from fenrirscreenreader.commands.commands import apply_tested_voice
|
||||
from fenrirscreenreader.commands.commands import voice_browser
|
||||
from fenrirscreenreader.commands.commands import voice_browser_safe
|
||||
from fenrirscreenreader.core import dynamicVoiceMenu
|
||||
from fenrirscreenreader.core.quickMenuManager import QuickMenuManager
|
||||
|
||||
|
||||
class FakeDiscoveryManager:
|
||||
def __init__(self):
|
||||
self.cached_modules = None
|
||||
self.cached_voices = {}
|
||||
self.module_callbacks = []
|
||||
self.voice_callbacks = {}
|
||||
self.test_callbacks = []
|
||||
|
||||
def get_cached_modules(self):
|
||||
return self.cached_modules
|
||||
|
||||
def get_cached_voices(self, module):
|
||||
return self.cached_voices.get(module)
|
||||
|
||||
def request_modules(self, callback, refresh=False):
|
||||
self.module_callbacks.append(callback)
|
||||
return len(self.module_callbacks)
|
||||
|
||||
def request_voices(self, module, callback, refresh=False):
|
||||
self.voice_callbacks.setdefault(module, []).append(callback)
|
||||
return len(self.voice_callbacks[module])
|
||||
|
||||
def request_voice_test(self, module, voice, message, callback):
|
||||
self.test_callbacks.append((module, voice, message, callback))
|
||||
return len(self.test_callbacks)
|
||||
|
||||
def complete_modules(self, modules, error=""):
|
||||
self.cached_modules = modules
|
||||
callbacks = self.module_callbacks
|
||||
self.module_callbacks = []
|
||||
for callback in callbacks:
|
||||
callback(list(modules), error)
|
||||
|
||||
def complete_voices(self, module, voices, error=""):
|
||||
self.cached_voices[module] = voices
|
||||
callbacks = self.voice_callbacks.pop(module, [])
|
||||
for callback in callbacks:
|
||||
callback(module, list(voices), error)
|
||||
|
||||
def complete_test(self, succeeded, error=""):
|
||||
_module, _voice, _message, callback = self.test_callbacks.pop(0)
|
||||
callback(succeeded, error)
|
||||
|
||||
|
||||
class RuntimeSettings:
|
||||
def __init__(
|
||||
self,
|
||||
module="rhvoice",
|
||||
voice="alan",
|
||||
language="en-us",
|
||||
driver="speechdDriver",
|
||||
):
|
||||
self.values = {
|
||||
("speech", "driver"): driver,
|
||||
("speech", "module"): module,
|
||||
("speech", "voice"): voice,
|
||||
("speech", "language"): language,
|
||||
}
|
||||
|
||||
def get_setting(self, section, setting):
|
||||
return self.values[(section, setting)]
|
||||
|
||||
def set_setting(self, section, setting, value):
|
||||
self.values[(section, setting)] = value
|
||||
|
||||
|
||||
def create_voice_environment(discovery):
|
||||
return {
|
||||
"runtime": {
|
||||
"DebugManager": Mock(),
|
||||
"OutputManager": Mock(),
|
||||
"SpeechDiscoveryManager": discovery,
|
||||
},
|
||||
"commandBuffer": {},
|
||||
}
|
||||
|
||||
|
||||
def test_safe_browser_records_voice_only_after_successful_test():
|
||||
discovery = FakeDiscoveryManager()
|
||||
environment = create_voice_environment(discovery)
|
||||
browser = voice_browser_safe.command()
|
||||
browser.initialize(environment)
|
||||
|
||||
browser.run()
|
||||
assert "lastTestedVoice" not in environment["commandBuffer"]
|
||||
|
||||
discovery.complete_modules(["rhvoice"])
|
||||
discovery.complete_voices("rhvoice", ["alan", "slt"])
|
||||
assert "lastTestedVoice" not in environment["commandBuffer"]
|
||||
|
||||
discovery.complete_test(True)
|
||||
assert environment["commandBuffer"]["lastTestedModule"] == "rhvoice"
|
||||
assert environment["commandBuffer"]["lastTestedVoice"] == "alan"
|
||||
|
||||
|
||||
def test_safe_browser_does_not_record_failed_test():
|
||||
discovery = FakeDiscoveryManager()
|
||||
environment = create_voice_environment(discovery)
|
||||
browser = voice_browser_safe.command()
|
||||
browser.initialize(environment)
|
||||
|
||||
browser.run()
|
||||
discovery.complete_modules(["rhvoice"])
|
||||
discovery.complete_voices("rhvoice", ["alan"])
|
||||
discovery.complete_test(False, "test timed out")
|
||||
|
||||
assert "lastTestedVoice" not in environment["commandBuffer"]
|
||||
|
||||
|
||||
def test_safe_browser_keeps_voxin_language_separate():
|
||||
discovery = FakeDiscoveryManager()
|
||||
environment = create_voice_environment(discovery)
|
||||
browser = voice_browser_safe.command()
|
||||
browser.initialize(environment)
|
||||
|
||||
browser.run()
|
||||
discovery.complete_modules(["voxin"])
|
||||
discovery.complete_voices("voxin", ["Nathan|en-US"])
|
||||
discovery.complete_test(True)
|
||||
|
||||
assert environment["commandBuffer"]["lastTestedVoice"] == "Nathan"
|
||||
assert environment["commandBuffer"]["lastTestedLanguage"] == "en-US"
|
||||
|
||||
|
||||
def test_interactive_browser_keeps_voxin_language_through_test_and_apply():
|
||||
discovery = FakeDiscoveryManager()
|
||||
environment = create_voice_environment(discovery)
|
||||
environment["runtime"]["SettingsManager"] = Mock(
|
||||
settings={"speech": {}},
|
||||
)
|
||||
browser = voice_browser.command()
|
||||
browser.initialize(environment)
|
||||
browser.modules = ["voxin"]
|
||||
browser.voices = ["Nathan|en-US"]
|
||||
|
||||
browser.test_voice()
|
||||
assert discovery.test_callbacks[0][1] == "Nathan"
|
||||
discovery.complete_test(True)
|
||||
assert environment["commandBuffer"]["lastTestedVoice"] == "Nathan"
|
||||
assert environment["commandBuffer"]["lastTestedLanguage"] == "en-US"
|
||||
|
||||
browser.apply_voice()
|
||||
speech_settings = environment["runtime"]["SettingsManager"].settings[
|
||||
"speech"
|
||||
]
|
||||
assert speech_settings["voice"] == "Nathan"
|
||||
assert speech_settings["language"] == "en-US"
|
||||
|
||||
|
||||
def test_apply_tested_voice_applies_voxin_language():
|
||||
settings = RuntimeSettings()
|
||||
speech_driver = Mock()
|
||||
environment = {
|
||||
"runtime": {
|
||||
"OutputManager": Mock(),
|
||||
"SettingsManager": settings,
|
||||
"SpeechDriver": speech_driver,
|
||||
},
|
||||
"commandBuffer": {
|
||||
"lastTestedModule": "voxin",
|
||||
"lastTestedVoice": "Nathan",
|
||||
"lastTestedLanguage": "en-US",
|
||||
},
|
||||
}
|
||||
command = apply_tested_voice.command()
|
||||
command.initialize(environment)
|
||||
|
||||
command.run()
|
||||
|
||||
assert settings.get_setting("speech", "language") == "en-US"
|
||||
speech_driver.set_language.assert_called_once_with("en-US")
|
||||
|
||||
|
||||
def test_dynamic_voice_command_commits_result_on_main_loop_callback():
|
||||
discovery = FakeDiscoveryManager()
|
||||
environment = create_voice_environment(discovery)
|
||||
voice_command = dynamicVoiceMenu.DynamicVoiceCommand(
|
||||
"rhvoice", "alan", environment
|
||||
)
|
||||
|
||||
voice_command.run()
|
||||
assert environment["commandBuffer"] == {}
|
||||
|
||||
discovery.complete_test(True)
|
||||
assert environment["commandBuffer"]["pendingVoiceModule"] == "rhvoice"
|
||||
assert environment["commandBuffer"]["pendingVoiceVoice"] == "alan"
|
||||
assert environment["commandBuffer"]["voiceTestCompleted"] is True
|
||||
|
||||
|
||||
def test_dynamic_voxin_voice_keeps_language_out_of_voice_name():
|
||||
discovery = FakeDiscoveryManager()
|
||||
environment = create_voice_environment(discovery)
|
||||
voice_command = dynamicVoiceMenu.DynamicVoiceCommand(
|
||||
"voxin", "Nathan|en-US", environment
|
||||
)
|
||||
|
||||
voice_command.run()
|
||||
assert discovery.test_callbacks[0][1] == "Nathan"
|
||||
|
||||
discovery.complete_test(True)
|
||||
assert environment["commandBuffer"]["pendingVoiceVoice"] == "Nathan"
|
||||
assert environment["commandBuffer"]["pendingVoiceLanguage"] == "en-US"
|
||||
|
||||
|
||||
def test_quick_menu_waits_for_discovery_then_applies_module():
|
||||
discovery = FakeDiscoveryManager()
|
||||
settings = RuntimeSettings()
|
||||
output_manager = Mock()
|
||||
manager = QuickMenuManager()
|
||||
manager.env = {
|
||||
"runtime": {
|
||||
"DebugManager": Mock(),
|
||||
"OutputManager": output_manager,
|
||||
"SettingsManager": settings,
|
||||
"SpeechDriver": Mock(),
|
||||
"SpeechDiscoveryManager": discovery,
|
||||
}
|
||||
}
|
||||
|
||||
assert manager.cycle_speech_module("next") is False
|
||||
assert settings.get_setting("speech", "module") == "rhvoice"
|
||||
|
||||
discovery.complete_modules(["rhvoice", "espeak-ng"])
|
||||
discovery.complete_voices("espeak-ng", ["en-us"])
|
||||
|
||||
assert settings.get_setting("speech", "module") == "espeak-ng"
|
||||
assert settings.get_setting("speech", "voice") == "en-us"
|
||||
output_manager.present_text.assert_any_call("espeak-ng", interrupt=True)
|
||||
|
||||
|
||||
def test_quick_menu_deduplicates_voice_request_for_module_cycle():
|
||||
discovery = FakeDiscoveryManager()
|
||||
discovery.cached_modules = ["rhvoice", "espeak-ng"]
|
||||
settings = RuntimeSettings()
|
||||
manager = QuickMenuManager()
|
||||
manager.env = {
|
||||
"runtime": {
|
||||
"DebugManager": Mock(),
|
||||
"OutputManager": Mock(),
|
||||
"SettingsManager": settings,
|
||||
"SpeechDriver": Mock(),
|
||||
"SpeechDiscoveryManager": discovery,
|
||||
}
|
||||
}
|
||||
|
||||
assert manager.cycle_speech_module("next") is False
|
||||
assert manager.cycle_speech_module("next") is False
|
||||
assert len(discovery.voice_callbacks["espeak-ng"]) == 1
|
||||
|
||||
|
||||
def test_dynamic_vmenu_keeps_loading_entry_until_all_results_arrive():
|
||||
discovery = FakeDiscoveryManager()
|
||||
environment = create_voice_environment(discovery)
|
||||
vmenu_manager = Mock()
|
||||
vmenu_manager.env = environment
|
||||
vmenu_manager.menuDict = {}
|
||||
vmenu_manager._voice_menu_generation = 0
|
||||
vmenu_manager.get_active.return_value = False
|
||||
|
||||
dynamicVoiceMenu.add_dynamic_voice_menus(vmenu_manager)
|
||||
assert "Loading voices Action" in vmenu_manager.menuDict[
|
||||
"Voice Browser Menu"
|
||||
]
|
||||
|
||||
discovery.complete_modules(["rhvoice", "espeak-ng"])
|
||||
discovery.complete_voices("rhvoice", ["alan"])
|
||||
assert "Loading voices Action" in vmenu_manager.menuDict[
|
||||
"Voice Browser Menu"
|
||||
]
|
||||
|
||||
discovery.complete_voices("espeak-ng", ["en-us"])
|
||||
voice_menu = vmenu_manager.menuDict["Voice Browser Menu"]
|
||||
assert "rhvoice Menu" in voice_menu
|
||||
assert "espeak-ng Menu" in voice_menu
|
||||
assert "alan Action" in voice_menu["rhvoice Menu"]
|
||||
|
||||
|
||||
def test_dynamic_vmenu_does_not_replace_active_loading_menu():
|
||||
discovery = FakeDiscoveryManager()
|
||||
environment = create_voice_environment(discovery)
|
||||
vmenu_manager = Mock()
|
||||
vmenu_manager.env = environment
|
||||
vmenu_manager.menuDict = {}
|
||||
vmenu_manager._voice_menu_generation = 0
|
||||
vmenu_manager.get_active.return_value = False
|
||||
|
||||
dynamicVoiceMenu.add_dynamic_voice_menus(vmenu_manager)
|
||||
vmenu_manager.get_active.return_value = True
|
||||
discovery.complete_modules(["voxin"])
|
||||
discovery.complete_voices("voxin", ["Nathan|en-US"])
|
||||
|
||||
voice_menu = vmenu_manager.menuDict["Voice Browser Menu"]
|
||||
assert "Loading voices Action" in voice_menu
|
||||
assert vmenu_manager._voice_menu_refresh_pending is True
|
||||
@@ -0,0 +1,109 @@
|
||||
import threading
|
||||
import time
|
||||
from unittest.mock import Mock
|
||||
|
||||
from fenrirscreenreader.core.backgroundTaskManager import BackgroundTaskManager
|
||||
from fenrirscreenreader.core.eventData import FenrirEventType
|
||||
|
||||
|
||||
def wait_for_call(mock, timeout=1.0):
|
||||
deadline = time.monotonic() + timeout
|
||||
while not mock.called and time.monotonic() < deadline:
|
||||
time.sleep(0.01)
|
||||
assert mock.called
|
||||
|
||||
|
||||
def test_worker_returns_result_through_event_queue_before_callback_runs():
|
||||
event_manager = Mock()
|
||||
environment = {
|
||||
"runtime": {
|
||||
"DebugManager": Mock(),
|
||||
"EventManager": event_manager,
|
||||
}
|
||||
}
|
||||
manager = BackgroundTaskManager(worker_count=1)
|
||||
manager.initialize(environment)
|
||||
callback = Mock()
|
||||
main_thread_id = threading.get_ident()
|
||||
|
||||
try:
|
||||
task_id = manager.submit_task(lambda value: value * 2, callback, 21)
|
||||
wait_for_call(event_manager.put_to_event_queue)
|
||||
|
||||
callback.assert_not_called()
|
||||
event_type, result = event_manager.put_to_event_queue.call_args.args
|
||||
assert event_type == FenrirEventType.background_task_result
|
||||
assert result == {
|
||||
"task_id": task_id,
|
||||
"succeeded": True,
|
||||
"value": 42,
|
||||
"error": "",
|
||||
}
|
||||
|
||||
callback.side_effect = lambda _result: setattr(
|
||||
callback, "thread_id", threading.get_ident()
|
||||
)
|
||||
manager.handle_result(result)
|
||||
callback.assert_called_once_with(result)
|
||||
assert callback.thread_id == main_thread_id
|
||||
finally:
|
||||
manager.shutdown()
|
||||
|
||||
|
||||
def test_cancelled_task_result_is_not_delivered():
|
||||
environment = {
|
||||
"runtime": {
|
||||
"DebugManager": Mock(),
|
||||
"EventManager": Mock(),
|
||||
}
|
||||
}
|
||||
manager = BackgroundTaskManager(worker_count=1)
|
||||
manager.initialize(environment)
|
||||
callback = Mock()
|
||||
|
||||
try:
|
||||
task_id = manager.submit_task(lambda: "late", callback)
|
||||
manager.cancel_task(task_id)
|
||||
manager.handle_result(
|
||||
{
|
||||
"task_id": task_id,
|
||||
"succeeded": True,
|
||||
"value": "late",
|
||||
"error": "",
|
||||
}
|
||||
)
|
||||
callback.assert_not_called()
|
||||
finally:
|
||||
manager.shutdown()
|
||||
|
||||
|
||||
def test_external_tasks_cannot_starve_default_tasks():
|
||||
event_manager = Mock()
|
||||
environment = {
|
||||
"runtime": {
|
||||
"DebugManager": Mock(),
|
||||
"EventManager": event_manager,
|
||||
}
|
||||
}
|
||||
manager = BackgroundTaskManager(worker_count=1, external_worker_count=2)
|
||||
manager.initialize(environment)
|
||||
release_external = threading.Event()
|
||||
external_started = [threading.Event(), threading.Event()]
|
||||
|
||||
def block_external(started):
|
||||
started.set()
|
||||
release_external.wait(timeout=2)
|
||||
|
||||
try:
|
||||
for started in external_started:
|
||||
manager.submit_external_task(block_external, Mock(), started)
|
||||
for started in external_started:
|
||||
assert started.wait(timeout=1)
|
||||
|
||||
manager.submit_task(lambda: "voice result", Mock())
|
||||
wait_for_call(event_manager.put_to_event_queue)
|
||||
result = event_manager.put_to_event_queue.call_args.args[1]
|
||||
assert result["value"] == "voice result"
|
||||
finally:
|
||||
release_external.set()
|
||||
manager.shutdown()
|
||||
@@ -0,0 +1,72 @@
|
||||
from unittest.mock import Mock
|
||||
|
||||
from fenrirscreenreader.commands.commands import export_clipboard_to_x
|
||||
from fenrirscreenreader.commands.commands import import_clipboard_from_x
|
||||
|
||||
|
||||
class FakeTaskManager:
|
||||
def __init__(self):
|
||||
self.callback = None
|
||||
self.function = None
|
||||
self.args = None
|
||||
|
||||
def submit_task(self, function, callback, *args):
|
||||
self.function = function
|
||||
self.callback = callback
|
||||
self.args = args
|
||||
return 7
|
||||
|
||||
def cancel_task(self, _task_id):
|
||||
pass
|
||||
|
||||
|
||||
def test_clipboard_import_changes_history_only_after_result_event():
|
||||
task_manager = FakeTaskManager()
|
||||
memory_manager = Mock()
|
||||
environment = {
|
||||
"runtime": {
|
||||
"BackgroundTaskManager": task_manager,
|
||||
"MemoryManager": memory_manager,
|
||||
"OutputManager": Mock(),
|
||||
}
|
||||
}
|
||||
command = import_clipboard_from_x.command()
|
||||
command.initialize(environment)
|
||||
|
||||
command.run()
|
||||
memory_manager.add_value_to_first_index.assert_not_called()
|
||||
|
||||
task_manager.callback(
|
||||
{"task_id": 7, "succeeded": True, "value": "copied text"}
|
||||
)
|
||||
memory_manager.add_value_to_first_index.assert_called_once_with(
|
||||
"clipboardHistory", "copied text"
|
||||
)
|
||||
|
||||
|
||||
def test_clipboard_export_updates_sync_state_only_after_result_event():
|
||||
task_manager = FakeTaskManager()
|
||||
memory_manager = Mock(
|
||||
is_index_list_empty=Mock(return_value=False),
|
||||
get_index_list_element=Mock(return_value="copied text"),
|
||||
)
|
||||
sync_manager = Mock()
|
||||
environment = {
|
||||
"runtime": {
|
||||
"BackgroundTaskManager": task_manager,
|
||||
"ClipboardSyncManager": sync_manager,
|
||||
"DebugManager": Mock(),
|
||||
"MemoryManager": memory_manager,
|
||||
"OutputManager": Mock(),
|
||||
}
|
||||
}
|
||||
command = export_clipboard_to_x.command()
|
||||
command.initialize(environment)
|
||||
|
||||
command.run()
|
||||
sync_manager.mark_written_to_x.assert_not_called()
|
||||
|
||||
task_manager.callback(
|
||||
{"task_id": 7, "succeeded": True, "value": True}
|
||||
)
|
||||
sync_manager.mark_written_to_x.assert_called_once_with("copied text")
|
||||
@@ -3,6 +3,7 @@ from unittest.mock import Mock
|
||||
import pytest
|
||||
|
||||
from fenrirscreenreader.core.clipboardSyncManager import ClipboardSyncManager
|
||||
from fenrirscreenreader.core.clipboardSyncManager import synchronize_clipboards
|
||||
|
||||
|
||||
def build_env(
|
||||
@@ -270,3 +271,87 @@ def test_x_clipboard_paste_exception_is_ignored(monkeypatch):
|
||||
|
||||
env["runtime"]["MemoryManager"].add_value_to_first_index.assert_not_called()
|
||||
env["runtime"]["DebugManager"].write_debug_out.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_runtime_sync_applies_worker_result_on_main_loop():
|
||||
env = build_env(fenrir_text=None)
|
||||
task_manager = Mock()
|
||||
task_manager.submit_task.return_value = 12
|
||||
env["runtime"]["BackgroundTaskManager"] = task_manager
|
||||
manager = ClipboardSyncManager()
|
||||
manager.env = env
|
||||
manager.enabled = True
|
||||
manager.display = ":1"
|
||||
manager.running = True
|
||||
|
||||
manager.handle_sync_event()
|
||||
env["runtime"]["MemoryManager"].add_value_to_first_index.assert_not_called()
|
||||
|
||||
callback = task_manager.submit_task.call_args.args[1]
|
||||
callback(
|
||||
{
|
||||
"task_id": 12,
|
||||
"succeeded": True,
|
||||
"value": {
|
||||
"import_text": "from x",
|
||||
"last_imported_from_x": "from x",
|
||||
"last_observed_fenrir": "from x",
|
||||
"last_observed_x": "from x",
|
||||
},
|
||||
"error": "",
|
||||
}
|
||||
)
|
||||
|
||||
env["runtime"]["MemoryManager"].add_value_to_first_index.assert_called_once_with(
|
||||
"clipboardHistory", "from x"
|
||||
)
|
||||
assert manager.last_imported_from_x == "from x"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_runtime_sync_discards_import_if_fenrir_clipboard_changed():
|
||||
env = build_env(fenrir_text=None)
|
||||
task_manager = Mock()
|
||||
task_manager.submit_task.return_value = 12
|
||||
env["runtime"]["BackgroundTaskManager"] = task_manager
|
||||
manager = ClipboardSyncManager()
|
||||
manager.env = env
|
||||
manager.display = ":1"
|
||||
manager.running = True
|
||||
|
||||
manager.handle_sync_event()
|
||||
env["runtime"]["MemoryManager"].is_index_list_empty.return_value = False
|
||||
env["runtime"]["MemoryManager"].get_index_list_element.return_value = (
|
||||
"new fenrir copy"
|
||||
)
|
||||
callback = task_manager.submit_task.call_args.args[1]
|
||||
callback(
|
||||
{
|
||||
"task_id": 12,
|
||||
"succeeded": True,
|
||||
"value": {
|
||||
"import_text": "stale x text",
|
||||
"last_imported_from_x": "stale x text",
|
||||
"last_observed_fenrir": "stale x text",
|
||||
"last_observed_x": "stale x text",
|
||||
},
|
||||
"error": "",
|
||||
}
|
||||
)
|
||||
|
||||
env["runtime"]["MemoryManager"].add_value_to_first_index.assert_not_called()
|
||||
assert manager.last_imported_from_x is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_sync_worker_returns_changes_without_manager_access(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
"fenrirscreenreader.core.clipboardSyncManager.x_clipboard.read_text",
|
||||
Mock(return_value="from x"),
|
||||
)
|
||||
|
||||
result = synchronize_clipboards(":1", None, None, None)
|
||||
|
||||
assert result["import_text"] == "from x"
|
||||
assert result["last_observed_x"] == "from x"
|
||||
|
||||
@@ -0,0 +1,116 @@
|
||||
from unittest.mock import Mock
|
||||
|
||||
from fenrirscreenreader.core.speechDiscoveryManager import (
|
||||
SpeechDiscoveryManager,
|
||||
query_speechd_voices,
|
||||
)
|
||||
|
||||
|
||||
class FakeBackgroundTaskManager:
|
||||
def __init__(self):
|
||||
self.tasks = []
|
||||
|
||||
def submit_task(self, function, callback, *args, **kwargs):
|
||||
task_id = len(self.tasks) + 1
|
||||
self.tasks.append((task_id, function, callback, args, kwargs))
|
||||
return task_id
|
||||
|
||||
def cancel_task(self, task_id):
|
||||
pass
|
||||
|
||||
def complete(self, task_id, value=None, error=""):
|
||||
_, _function, callback, _args, _kwargs = self.tasks[task_id - 1]
|
||||
callback(
|
||||
{
|
||||
"task_id": task_id,
|
||||
"succeeded": not error,
|
||||
"value": value,
|
||||
"error": error,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def create_manager():
|
||||
task_manager = FakeBackgroundTaskManager()
|
||||
manager = SpeechDiscoveryManager()
|
||||
manager.initialize(
|
||||
{
|
||||
"runtime": {
|
||||
"BackgroundTaskManager": task_manager,
|
||||
"DebugManager": Mock(),
|
||||
}
|
||||
}
|
||||
)
|
||||
return manager, task_manager
|
||||
|
||||
|
||||
def test_module_requests_share_one_worker_and_cache_main_thread_result():
|
||||
manager, task_manager = create_manager()
|
||||
first_callback = Mock()
|
||||
second_callback = Mock()
|
||||
|
||||
first_task_id = manager.request_modules(first_callback)
|
||||
second_task_id = manager.request_modules(second_callback)
|
||||
|
||||
assert first_task_id == second_task_id
|
||||
assert len(task_manager.tasks) == 1
|
||||
|
||||
task_manager.complete(first_task_id, ["rhvoice", "espeak-ng"])
|
||||
|
||||
first_callback.assert_called_once_with(["rhvoice", "espeak-ng"], "")
|
||||
second_callback.assert_called_once_with(["rhvoice", "espeak-ng"], "")
|
||||
|
||||
cached_callback = Mock()
|
||||
assert manager.request_modules(cached_callback) is None
|
||||
cached_callback.assert_called_once_with(["rhvoice", "espeak-ng"], "")
|
||||
assert len(task_manager.tasks) == 1
|
||||
|
||||
|
||||
def test_voice_requests_are_deduplicated_per_module():
|
||||
manager, task_manager = create_manager()
|
||||
rhvoice_callback = Mock()
|
||||
repeated_callback = Mock()
|
||||
espeak_callback = Mock()
|
||||
|
||||
rhvoice_task = manager.request_voices("rhvoice", rhvoice_callback)
|
||||
repeated_task = manager.request_voices("rhvoice", repeated_callback)
|
||||
espeak_task = manager.request_voices("espeak-ng", espeak_callback)
|
||||
|
||||
assert rhvoice_task == repeated_task
|
||||
assert espeak_task != rhvoice_task
|
||||
assert len(task_manager.tasks) == 2
|
||||
|
||||
task_manager.complete(rhvoice_task, ["alan", "slt"])
|
||||
rhvoice_callback.assert_called_once_with("rhvoice", ["alan", "slt"], "")
|
||||
repeated_callback.assert_called_once_with(
|
||||
"rhvoice", ["alan", "slt"], ""
|
||||
)
|
||||
espeak_callback.assert_not_called()
|
||||
|
||||
|
||||
def test_failing_discovery_callback_does_not_hide_result_from_others():
|
||||
manager, task_manager = create_manager()
|
||||
failing_callback = Mock(side_effect=RuntimeError("consumer failed"))
|
||||
second_callback = Mock()
|
||||
|
||||
task_id = manager.request_modules(failing_callback)
|
||||
manager.request_modules(second_callback)
|
||||
task_manager.complete(task_id, ["rhvoice"])
|
||||
|
||||
failing_callback.assert_called_once_with(["rhvoice"], "")
|
||||
second_callback.assert_called_once_with(["rhvoice"], "")
|
||||
manager.env["runtime"]["DebugManager"].write_debug_out.assert_called_once()
|
||||
|
||||
|
||||
def test_voice_query_preserves_multiword_synthesis_voice_names(monkeypatch):
|
||||
voice_list = (
|
||||
" NAME LANGUAGE VARIANT\n"
|
||||
" Perfect Paul en-US none\n"
|
||||
" Big Bob en-US none\n"
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"fenrirscreenreader.core.speechDiscoveryManager.subprocess.run",
|
||||
Mock(return_value=Mock(returncode=0, stdout=voice_list, stderr="")),
|
||||
)
|
||||
|
||||
assert query_speechd_voices("doubletalk") == ["Perfect Paul", "Big Bob"]
|
||||
@@ -2,90 +2,11 @@ import sys
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import Mock
|
||||
|
||||
from fenrirscreenreader.commands.commands import voice_browser
|
||||
from fenrirscreenreader.commands.commands import voice_browser_safe
|
||||
from fenrirscreenreader.core import dynamicVoiceMenu
|
||||
from fenrirscreenreader.core.quickMenuManager import (
|
||||
QuickMenuManager,
|
||||
SpeechHelperMixin,
|
||||
)
|
||||
from fenrirscreenreader.core.quickMenuManager import QuickMenuManager
|
||||
from fenrirscreenreader.speechDriver import speechdDriver
|
||||
from fenrirscreenreader.utils.speechd_utils import get_synthesis_voice_name
|
||||
|
||||
|
||||
VOICE_LIST = (
|
||||
" NAME LANGUAGE VARIANT\n"
|
||||
" Perfect Paul en-US none\n"
|
||||
" Big Bob en-US none\n"
|
||||
)
|
||||
|
||||
|
||||
def completed_voice_list():
|
||||
return SimpleNamespace(returncode=0, stdout=VOICE_LIST)
|
||||
|
||||
|
||||
def test_quick_menu_preserves_multiword_synthesis_voice_names(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
"fenrirscreenreader.core.quickMenuManager.subprocess.run",
|
||||
Mock(return_value=completed_voice_list()),
|
||||
)
|
||||
helper = SpeechHelperMixin()
|
||||
helper.env = {"runtime": {"DebugManager": Mock()}}
|
||||
|
||||
assert helper.get_module_voices("doubletalk") == [
|
||||
"Perfect Paul",
|
||||
"Big Bob",
|
||||
]
|
||||
|
||||
|
||||
def test_safe_voice_browser_preserves_multiword_synthesis_voice_names(
|
||||
monkeypatch,
|
||||
):
|
||||
monkeypatch.setattr(
|
||||
voice_browser_safe.subprocess,
|
||||
"run",
|
||||
Mock(return_value=completed_voice_list()),
|
||||
)
|
||||
browser = voice_browser_safe.command()
|
||||
browser.initialize({"runtime": {}})
|
||||
|
||||
assert browser.get_module_voices_with_timeout("doubletalk") == [
|
||||
"Perfect Paul",
|
||||
"Big Bob",
|
||||
]
|
||||
|
||||
|
||||
def test_interactive_voice_browser_preserves_multiword_synthesis_voice_names(
|
||||
monkeypatch,
|
||||
):
|
||||
monkeypatch.setattr(
|
||||
voice_browser.subprocess,
|
||||
"run",
|
||||
Mock(return_value=completed_voice_list()),
|
||||
)
|
||||
browser = voice_browser.command()
|
||||
|
||||
assert browser.get_module_voices("doubletalk") == [
|
||||
"Perfect Paul",
|
||||
"Big Bob",
|
||||
]
|
||||
|
||||
|
||||
def test_dynamic_voice_menu_preserves_multiword_synthesis_voice_names(
|
||||
monkeypatch,
|
||||
):
|
||||
monkeypatch.setattr(
|
||||
dynamicVoiceMenu.subprocess,
|
||||
"run",
|
||||
Mock(return_value=completed_voice_list()),
|
||||
)
|
||||
|
||||
assert dynamicVoiceMenu.get_module_voices("doubletalk") == [
|
||||
"Perfect Paul",
|
||||
"Big Bob",
|
||||
]
|
||||
|
||||
|
||||
def test_espeak_voice_selection_keeps_language_and_variant_behavior():
|
||||
voice = get_synthesis_voice_name(
|
||||
"espeak-ng",
|
||||
@@ -160,6 +81,12 @@ def test_speechd_driver_keeps_running_when_default_module_query_fails(
|
||||
|
||||
def test_quick_menu_espeak_voice_can_move_away_and_return():
|
||||
settings = RuntimeSettings(module="rhvoice")
|
||||
discovery = Mock()
|
||||
discovery.get_cached_modules.return_value = ["rhvoice", "espeak-ng"]
|
||||
discovery.get_cached_voices.side_effect = lambda module: {
|
||||
"rhvoice": ["alan"],
|
||||
"espeak-ng": ["en-gb", "en-us", "en-us+female2"],
|
||||
}[module]
|
||||
manager = QuickMenuManager()
|
||||
manager.env = {
|
||||
"runtime": {
|
||||
@@ -167,15 +94,9 @@ def test_quick_menu_espeak_voice_can_move_away_and_return():
|
||||
"OutputManager": Mock(),
|
||||
"SettingsManager": settings,
|
||||
"SpeechDriver": Mock(),
|
||||
"SpeechDiscoveryManager": discovery,
|
||||
}
|
||||
}
|
||||
manager._modules_cache = ["rhvoice", "espeak-ng"]
|
||||
manager._cache_timestamp = float("inf")
|
||||
manager._voices_cache["espeak-ng"] = [
|
||||
"en-gb",
|
||||
"en-us",
|
||||
"en-us+female2",
|
||||
]
|
||||
|
||||
assert manager.cycle_speech_module("next") is True
|
||||
assert settings.get_setting("speech", "module") == "espeak-ng"
|
||||
|
||||
@@ -8,26 +8,19 @@ from fenrirscreenreader.commands.commands import subprocess as subprocess_comman
|
||||
@pytest.mark.unit
|
||||
def test_script_command_executes_without_shell(monkeypatch):
|
||||
process = Mock()
|
||||
process.communicate.return_value = (b"done", b"")
|
||||
process.returncode = 0
|
||||
process.communicate.return_value = ("done", "")
|
||||
popen = Mock(return_value=process)
|
||||
monkeypatch.setattr(subprocess_command, "Popen", popen)
|
||||
output_manager = Mock()
|
||||
command = subprocess_command.command()
|
||||
command.initialize(
|
||||
{
|
||||
"general": {"curr_user": "Username"},
|
||||
"runtime": {"OutputManager": output_manager},
|
||||
},
|
||||
"/tmp/script with spaces",
|
||||
)
|
||||
monkeypatch.setattr(subprocess_command.subprocess, "Popen", popen)
|
||||
|
||||
command._thread_run()
|
||||
result = subprocess_command.run_script(
|
||||
"/tmp/script with spaces", "Username"
|
||||
)
|
||||
|
||||
popen.assert_called_once_with(
|
||||
["/tmp/script with spaces", "Username"],
|
||||
stdout=subprocess_command.PIPE,
|
||||
stderr=subprocess_command.PIPE,
|
||||
)
|
||||
output_manager.present_text.assert_called_once_with(
|
||||
"done", sound_icon="", interrupt=False
|
||||
stdout=subprocess_command.subprocess.PIPE,
|
||||
stderr=subprocess_command.subprocess.PIPE,
|
||||
text=True,
|
||||
)
|
||||
assert result == {"return_code": 0, "stdout": "done", "stderr": ""}
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
import importlib
|
||||
from unittest.mock import Mock
|
||||
|
||||
|
||||
subprocess_command = importlib.import_module(
|
||||
"fenrirscreenreader.commands.commands.subprocess"
|
||||
)
|
||||
|
||||
|
||||
def test_external_script_output_is_delivered_by_main_loop_callback(tmp_path):
|
||||
script_path = tmp_path / "helper.sh"
|
||||
script_path.write_text("#!/bin/sh\necho ready\n", encoding="utf-8")
|
||||
script_path.chmod(0o755)
|
||||
task_manager = Mock()
|
||||
task_manager.submit_external_task.return_value = 7
|
||||
output_manager = Mock()
|
||||
environment = {
|
||||
"runtime": {
|
||||
"BackgroundTaskManager": task_manager,
|
||||
"OutputManager": output_manager,
|
||||
},
|
||||
"general": {"curr_user": "Username"},
|
||||
}
|
||||
script_command = subprocess_command.command()
|
||||
script_command.initialize(environment, str(script_path))
|
||||
|
||||
script_command.run()
|
||||
|
||||
output_manager.present_text.assert_not_called()
|
||||
function, callback, path, current_user = (
|
||||
task_manager.submit_external_task.call_args.args
|
||||
)
|
||||
assert function is subprocess_command.run_script
|
||||
assert path == str(script_path)
|
||||
assert current_user == "Username"
|
||||
|
||||
callback(
|
||||
{
|
||||
"task_id": 7,
|
||||
"succeeded": True,
|
||||
"value": {"return_code": 0, "stdout": "ready\n", "stderr": ""},
|
||||
"error": "",
|
||||
}
|
||||
)
|
||||
output_manager.present_text.assert_called_once_with(
|
||||
"ready\n", sound_icon="", interrupt=False
|
||||
)
|
||||
Reference in New Issue
Block a user