from functools import partial import os import tempfile import pytest from plover import system from plover.config import Config, DictionaryConfig from plover.engine import ErroredDictionary, StenoEngine from plover.machine.base import ( STATE_INITIALIZING, STATE_RUNNING, STATE_STOPPED, StenotypeBase, ) from plover.machine.keymap import Keymap from plover.misc import normalize_path from plover.oslayer.controller import Controller from plover.output import Output from plover.registry import Registry from plover.steno_dictionary import StenoDictionaryCollection from plover_build_utils.testing import make_dict from .py37compat import mock class FakeMachine(StenotypeBase): instance = None def __init__(self, options): super().__init__() self.options = options self.is_suppressed = False @classmethod def get_keys(cls): return system.KEYS def start_capture(self): assert FakeMachine.instance is None FakeMachine.instance = self self._initializing() self._ready() def stop_capture(self): FakeMachine.instance = None self._stopped() def set_suppression(self, enabled): self.is_suppressed = enabled class FakeKeyboardEmulation(Output): def send_backspaces(self, b): pass def send_string(self, s): pass def send_key_combination(self, c): pass def set_key_press_delay(self, delay_ms): pass class FakeEngine(StenoEngine): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.events = [] def hook_callback(hook, *args, **kwargs): self.events.append((hook, args, kwargs)) for hook in self.HOOKS: self.hook_connect(hook, partial(hook_callback, hook)) def _in_engine_thread(self): return True def quit(self, code=0): self._same_thread_hook(self._quit, code) def start(self): StenoEngine.start(self) @pytest.fixture def engine(monkeypatch): FakeMachine.instance = None registry = Registry() registry.update() registry.register_plugin("machine", "Fake", FakeMachine) monkeypatch.setattr("plover.config.registry", registry) monkeypatch.setattr("plover.engine.registry", registry) ctrl = mock.MagicMock(spec=Controller) kbd = FakeKeyboardEmulation() cfg_file = tempfile.NamedTemporaryFile( prefix="plover", suffix="config", delete=False ) try: cfg_file.close() cfg = Config(cfg_file.name) cfg["dictionaries"] = [] cfg["machine_type"] = "Fake" cfg["system_keymap"] = [(k, k) for k in system.KEYS] cfg.save() yield FakeEngine(cfg, ctrl, kbd) finally: os.unlink(cfg_file.name) def test_engine_lifecycle(engine): # Config load. assert engine.load_config() assert engine.events == [] # Startup. engine.start() assert engine.events == [ ("machine_state_changed", ("Fake", "initializing"), {}), ("machine_state_changed", ("Fake", "connected"), {}), ("config_changed", (engine.config,), {}), ] assert FakeMachine.instance is not None assert not FakeMachine.instance.is_suppressed assert len(engine._controller.mock_calls) == 1 engine._controller.start.assert_called_once() engine._controller.reset_mock() # Output enabled. engine.events.clear() engine.output = True assert engine.events == [ ("output_changed", (True,), {}), ] assert FakeMachine.instance.is_suppressed # Machine reconnection. engine.events.clear() engine.reset_machine() assert engine.events == [ ("machine_state_changed", ("Fake", STATE_STOPPED), {}), ("machine_state_changed", ("Fake", STATE_INITIALIZING), {}), ("machine_state_changed", ("Fake", STATE_RUNNING), {}), ] assert FakeMachine.instance is not None assert FakeMachine.instance.is_suppressed # No machine reset on keymap change. engine.events.clear() new_keymap = Keymap(system.KEYS, system.KEYS) new_keymap.set_mappings(zip(system.KEYS, reversed(system.KEYS))) config_update = {"system_keymap": new_keymap} assert FakeMachine.instance.keymap != new_keymap engine.config = config_update assert engine.events == [ ("config_changed", (config_update,), {}), ] assert FakeMachine.instance.keymap == new_keymap # Output disabled engine.events.clear() engine.output = False assert engine.events == [ ("output_changed", (False,), {}), ] assert not FakeMachine.instance.is_suppressed # Stopped. engine.events.clear() engine.quit(42) assert engine.join() == 42 assert engine.events == [ ("machine_state_changed", ("Fake", STATE_STOPPED), {}), ("quit", (), {}), ] assert FakeMachine.instance is None assert len(engine._controller.mock_calls) == 1 engine._controller.stop.assert_called_once() def test_loading_dictionaries(tmp_path, engine): def check_loaded_events(actual_events, expected_events): assert len(actual_events) == len(expected_events) for n, event in enumerate(actual_events): event_type, event_args, event_kwargs = event msg = "event %u: %r" % (n, event) assert event_type == "dictionaries_loaded", msg assert event_kwargs == {}, msg assert len(event_args) == 1, msg assert isinstance(event_args[0], StenoDictionaryCollection), msg assert [ (d.path, d.enabled, isinstance(d, ErroredDictionary)) for d in event_args[0].dicts ] == expected_events[n], msg with ( make_dict(tmp_path, b"{}", "json", "valid1") as valid_dict_1, make_dict(tmp_path, b"{}", "json", "valid2") as valid_dict_2, make_dict(tmp_path, b"", "json", "invalid1") as invalid_dict_1, make_dict(tmp_path, b"", "json", "invalid2") as invalid_dict_2, ): valid_dict_1 = normalize_path(str(valid_dict_1)) valid_dict_2 = normalize_path(str(valid_dict_2)) invalid_dict_1 = normalize_path(str(invalid_dict_1)) invalid_dict_2 = normalize_path(str(invalid_dict_2)) engine.start() for new_dictionaries, *expected_events in ( # Load one valid dictionary. [ [ # path, enabled (valid_dict_1, True), ], [ # path, enabled, errored (valid_dict_1, True, False), ], ], # Load another invalid dictionary. [ [ (valid_dict_1, True), (invalid_dict_1, True), ], [ (valid_dict_1, True, False), (invalid_dict_1, True, True), ], ], # Disable first dictionary. [ [ (valid_dict_1, False), (invalid_dict_1, True), ], [ (valid_dict_1, False, False), (invalid_dict_1, True, True), ], ], # Replace invalid dictonary with another invalid one. [ [ (valid_dict_1, False), (invalid_dict_2, True), ], [ (valid_dict_1, False, False), ], [ (valid_dict_1, False, False), (invalid_dict_2, True, True), ], ], ): engine.events.clear() config_update = { "dictionaries": [DictionaryConfig(*d) for d in new_dictionaries] } engine.config = dict(config_update) assert engine.events[0] == ("config_changed", (config_update,), {}) check_loaded_events(engine.events[1:], expected_events) # Simulate an outdated dictionary. engine.events.clear() engine.dictionaries[valid_dict_1].timestamp -= 1 engine.config = {} check_loaded_events( engine.events, [ [ (invalid_dict_2, True, True), ], [ (valid_dict_1, False, False), (invalid_dict_2, True, True), ], ], ) def test_engine_running_state(engine): # Running state must be different # from initial (disabled state). initial_state = engine.translator_state assert engine.load_config() engine.set_output(True) running_state = engine.translator_state assert running_state != initial_state # Disabled state is reset every time # output is disabled. engine.set_output(False) disabled_state = engine.translator_state assert disabled_state != running_state assert disabled_state != initial_state # Running state is kept throughout. engine.set_output(True) assert engine.translator_state == running_state def test_undo_and_clear_empty_translator_state(engine): engine.clear_translator_state(undo=True)