Storm Dragon
130 lines
4.9 KiB
Python
130 lines
4.9 KiB
Python
import sys
|
|
import types
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
|
|
|
input_event_manager_stub = types.ModuleType("cthulhu.input_event_manager")
|
|
input_event_manager_stub.get_manager = mock.Mock(return_value=mock.Mock())
|
|
sys.modules["cthulhu.input_event_manager"] = input_event_manager_stub
|
|
|
|
from cthulhu.plugin_system_manager import PluginSystemManager
|
|
from cthulhu import cthulhu_state
|
|
from cthulhu import speech
|
|
from cthulhu.plugins.CthulhuRemote.connection_info import ConnectionInfo, ConnectionMode
|
|
from cthulhu.plugins.CthulhuRemote.local_machine import LocalMachine
|
|
from cthulhu.plugins.CthulhuRemote.plugin import CthulhuRemote
|
|
from cthulhu.plugins.CthulhuRemote.protocol import RemoteMessageType
|
|
from cthulhu.plugins.CthulhuRemote.serializer import JSONSerializer
|
|
|
|
|
|
class CthulhuRemotePluginTests(unittest.TestCase):
|
|
def test_connection_info_accepts_nvda_remote_style_fields(self):
|
|
info = ConnectionInfo.from_url("cthulhuremote://example.com:1234?key=abc&mode=slave")
|
|
|
|
self.assertEqual(info.hostname, "example.com")
|
|
self.assertEqual(info.port, 1234)
|
|
self.assertEqual(info.key, "abc")
|
|
self.assertEqual(info.mode, ConnectionMode.SLAVE)
|
|
self.assertEqual(
|
|
info.get_url_to_connect(),
|
|
"cthulhuremote://example.com:1234?key=abc&mode=master",
|
|
)
|
|
|
|
def test_json_serializer_uses_remote_message_type_values(self):
|
|
serializer = JSONSerializer()
|
|
|
|
payload = serializer.deserialize(serializer.serialize(RemoteMessageType.join, channel="abc"))
|
|
|
|
self.assertEqual(payload["type"], "join")
|
|
self.assertEqual(payload["channel"], "abc")
|
|
|
|
def test_local_machine_maps_common_windows_vk_codes_to_keyvals(self):
|
|
machine = LocalMachine(lambda message: None)
|
|
|
|
self.assertEqual(machine._resolve_keyval(0x41, None), machine._resolve_keyval(None, "a"))
|
|
self.assertEqual(machine._resolve_keyval(0x70, None), machine._resolve_keyval(None, "F1"))
|
|
self.assertIsNone(machine._resolve_keyval(0xFF, None))
|
|
|
|
def test_speech_monitor_callbacks_are_additive(self):
|
|
primary = mock.Mock()
|
|
listener = mock.Mock()
|
|
speech.set_monitor_callbacks(writeText=primary)
|
|
speech.add_monitor_callback(listener)
|
|
self.addCleanup(speech.set_monitor_callbacks, None)
|
|
self.addCleanup(speech.remove_monitor_callback, listener)
|
|
|
|
speech._write_to_monitor("status")
|
|
|
|
primary.assert_called_once_with("status")
|
|
listener.assert_called_once_with("status")
|
|
|
|
def test_remote_speech_does_not_echo_to_monitor_callbacks(self):
|
|
listener = mock.Mock()
|
|
speech.add_monitor_callback(listener)
|
|
self.addCleanup(speech.remove_monitor_callback, listener)
|
|
|
|
with (
|
|
mock.patch.object(speech, "_speechserver", None),
|
|
mock.patch.object(speech.speech_history, "add"),
|
|
mock.patch.object(cthulhu_state, "activeScript", None),
|
|
):
|
|
LocalMachine(lambda message: None).speak(["remote", "status"])
|
|
|
|
listener.assert_not_called()
|
|
|
|
def test_slave_mode_forwards_local_speech_to_relay(self):
|
|
plugin = CthulhuRemote()
|
|
plugin._connectionInfo = ConnectionInfo(
|
|
hostname="example.com",
|
|
port=1234,
|
|
key="abc",
|
|
mode=ConnectionMode.SLAVE,
|
|
)
|
|
plugin._transport = mock.Mock()
|
|
plugin._transport.connected = True
|
|
|
|
plugin._send_local_speech("focused button")
|
|
|
|
plugin._transport.send.assert_called_once_with(
|
|
RemoteMessageType.speak,
|
|
sequence=["focused button"],
|
|
)
|
|
|
|
def test_master_mode_does_not_forward_local_speech_to_relay(self):
|
|
plugin = CthulhuRemote()
|
|
plugin._connectionInfo = ConnectionInfo(
|
|
hostname="example.com",
|
|
port=1234,
|
|
key="abc",
|
|
mode=ConnectionMode.MASTER,
|
|
)
|
|
plugin._transport = mock.Mock()
|
|
plugin._transport.connected = True
|
|
|
|
plugin._send_local_speech("local status")
|
|
|
|
plugin._transport.send.assert_not_called()
|
|
|
|
@mock.patch("cthulhu.plugin_system_manager.dbus_service.get_remote_controller")
|
|
def test_plugin_manager_can_load_cthulhu_remote(self, remote_controller):
|
|
remote_controller.return_value = mock.Mock()
|
|
app = mock.Mock()
|
|
app.getSignalManager.return_value = mock.Mock()
|
|
app.getAPIHelper.return_value = None
|
|
manager = PluginSystemManager(app)
|
|
manager.rescanPlugins()
|
|
plugin_info = manager._resolve_plugin_info("CthulhuRemote")
|
|
|
|
self.assertIsNotNone(plugin_info)
|
|
self.assertTrue(manager.loadPlugin(plugin_info))
|
|
|
|
self.assertEqual(plugin_info.instance.__class__.__name__, "CthulhuRemote")
|
|
self.assertTrue(manager.unloadPlugin(plugin_info))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|