diff --git a/moler/device/atremote.py b/moler/device/atremote.py index c8216e408..487df6834 100644 --- a/moler/device/atremote.py +++ b/moler/device/atremote.py @@ -5,13 +5,11 @@ - be the state machine that controls which commands may run in given state """ -__author__ = 'Grzegorz Latuszek' -__copyright__ = 'Copyright (C) 2020, Nokia' -__email__ = 'grzegorz.latuszek@nokia.com' +__author__ = 'Grzegorz Latuszek, Marcin Usielski' +__copyright__ = 'Copyright (C) 2020-2024, Nokia' +__email__ = 'grzegorz.latuszek@nokia.com, marcin.usielski@nokia.com' from moler.device.textualdevice import TextualDevice -# from moler.device.proxy_pc import ProxyPc # TODO: allow jumping towards AT_REMOTE via proxy-pc -from moler.device.unixlocal import UnixLocal from moler.device.unixremote import UnixRemote from moler.helpers import call_base_class_method_with_same_name, mark_to_call_base_class_method_with_same_name from moler.cmd.at.genericat import GenericAtCommand @@ -32,21 +30,21 @@ class AtRemote(UnixRemote): DEVICE_CLASS: moler.device.atremote.AtRemote CONNECTION_HOPS: UNIX_LOCAL: + UNIX_REMOTE: + execute_command: ssh # default value + command_params: + expected_prompt: unix_remote_prompt + host: host_ip + login: login + password: password UNIX_REMOTE: - execute_command: ssh # default value + AT_REMOTE: + execute_command: plink_serial # default value command_params: - expected_prompt: unix_remote_prompt - host: host_ip - login: login - password: password - UNIX_REMOTE: + serial_devname: 'COM5' AT_REMOTE: - execute_command: plink_serial # default value - command_params: - serial_devname: 'COM5' - AT_REMOTE: - UNIX_REMOTE: - execute_command: ctrl_c # default value + UNIX_REMOTE: + execute_command: ctrl_c # default value """ at_remote = "AT_REMOTE" @@ -58,8 +56,8 @@ def __init__(self, sm_params, name=None, io_connection=None, io_type=None, varia :param sm_params: dict with parameters of state machine for device :param name: name of device :param io_connection: External-IO connection having embedded moler-connection - :param io_type: type of connection - tcp, udp, ssh, telnet, ... - :param variant: connection implementation variant, ex. 'threaded', 'twisted', 'asyncio', ... + :param io_type: type of connection - tcp, udp, ssh, telnet, .. + :param variant: connection implementation variant, ex. 'threaded', 'twisted', 'asyncio', .. (if not given then default one is taken) :param io_constructor_kwargs: additional parameter into constructor of selected connection type (if not given then default one is taken) @@ -82,7 +80,7 @@ def _get_default_sm_configuration_without_proxy_pc(self): """ config = { # TODO: shell we use direct-string names of config dicts? change simplicity vs readability TextualDevice.connection_hops: { - UnixRemote.unix_remote: { # from + AtRemote.unix_remote: { # from AtRemote.at_remote: { # to "execute_command": "plink_serial", "command_params": { # with parameters @@ -94,7 +92,7 @@ def _get_default_sm_configuration_without_proxy_pc(self): }, }, AtRemote.at_remote: { # from - UnixRemote.unix_remote: { # to + AtRemote.unix_remote: { # to "execute_command": "ctrl_c", # using command "command_params": { # with parameters "expected_prompt": 'remote_prompt', # overwritten in _configure_state_machine @@ -114,7 +112,7 @@ def _prepare_transitions_without_proxy_pc(self): :return: transitions without proxy_pc state. """ transitions = { - UnixRemote.unix_remote: { + AtRemote.unix_remote: { AtRemote.at_remote: { "action": [ "_execute_command_to_change_state" @@ -122,7 +120,64 @@ def _prepare_transitions_without_proxy_pc(self): } }, AtRemote.at_remote: { - UnixRemote.unix_remote: { + AtRemote.unix_remote: { + "action": [ + "_execute_command_to_change_state" + ], + } + }, + } + return transitions + + @mark_to_call_base_class_method_with_same_name + def _get_default_sm_configuration_with_proxy_pc(self): + """ + Return State Machine default configuration without proxy_pc state. + :return: default sm configuration without proxy_pc state. + """ + config = { + TextualDevice.connection_hops: { + AtRemote.unix_remote: { # from + AtRemote.at_remote: { # to + "execute_command": "plink_serial", + "command_params": { # with parameters + "target_newline": "\r\n" + }, + "required_command_params": [ + "serial_devname" + ] + }, + }, + AtRemote.at_remote: { # from + AtRemote.unix_remote: { # to + "execute_command": "ctrl_c", # using command + "command_params": { # with parameters + "expected_prompt": 'remote_prompt', # overwritten in _configure_state_machine + }, + "required_command_params": [ + ] + }, + }, + } + } + return config + + @mark_to_call_base_class_method_with_same_name + def _prepare_transitions_with_proxy_pc(self): + """ + Prepare transitions to change states without proxy_pc state. + :return: transitions without proxy_pc state. + """ + transitions = { + AtRemote.unix_remote: { + AtRemote.at_remote: { + "action": [ + "_execute_command_to_change_state" + ], + } + }, + AtRemote.at_remote: { + AtRemote.unix_remote: { "action": [ "_execute_command_to_change_state" ], @@ -138,7 +193,22 @@ def _prepare_state_prompts_without_proxy_pc(self): :return: textual prompt for each state without proxy_pc state. """ hops_config = self._configurations[TextualDevice.connection_hops] - serial_devname = hops_config[UnixRemote.unix_remote][AtRemote.at_remote]["command_params"]["serial_devname"] + serial_devname = hops_config[AtRemote.unix_remote][AtRemote.at_remote]["command_params"]["serial_devname"] + proxy_prompt = f"{serial_devname}> port READY" + at_cmds_prompt = GenericAtCommand._re_default_at_prompt.pattern # pylint: disable=protected-access + state_prompts = { + AtRemote.at_remote: f"{proxy_prompt}|{at_cmds_prompt}" + } + return state_prompts + + @mark_to_call_base_class_method_with_same_name + def _prepare_state_prompts_with_proxy_pc(self): + """ + Prepare textual prompt for each state for State Machine without proxy_pc state. + :return: textual prompt for each state without proxy_pc state. + """ + hops_config = self._configurations[TextualDevice.connection_hops] + serial_devname = hops_config[AtRemote.unix_remote][AtRemote.at_remote]["command_params"]["serial_devname"] proxy_prompt = f"{serial_devname}> port READY" at_cmds_prompt = GenericAtCommand._re_default_at_prompt.pattern # pylint: disable=protected-access state_prompts = { @@ -153,7 +223,20 @@ def _prepare_newline_chars_without_proxy_pc(self): :return: newline char for each state without proxy_pc state. """ hops_config = self._configurations[TextualDevice.connection_hops] - hops_2_at_remote_config = hops_config[UnixRemote.unix_remote][AtRemote.at_remote] + hops_2_at_remote_config = hops_config[AtRemote.unix_remote][AtRemote.at_remote] + newline_chars = { + AtRemote.at_remote: hops_2_at_remote_config["command_params"]["target_newline"], + } + return newline_chars + + @mark_to_call_base_class_method_with_same_name + def _prepare_newline_chars_with_proxy_pc(self): + """ + Prepare newline char for each state for State Machine without proxy_pc state. + :return: newline char for each state without proxy_pc state. + """ + hops_config = self._configurations[TextualDevice.connection_hops] + hops_2_at_remote_config = hops_config[AtRemote.unix_remote][AtRemote.at_remote] newline_chars = { AtRemote.at_remote: hops_2_at_remote_config["command_params"]["target_newline"], } @@ -167,37 +250,90 @@ def _prepare_state_hops_without_proxy_pc(self): """ state_hops = { TextualDevice.not_connected: { - UnixLocal.unix_local_root: UnixLocal.unix_local, - UnixRemote.unix_remote: UnixLocal.unix_local, - UnixRemote.unix_remote_root: UnixLocal.unix_local, - AtRemote.at_remote: UnixLocal.unix_local, - }, - UnixLocal.unix_local: { - UnixRemote.unix_remote_root: UnixRemote.unix_remote, - AtRemote.at_remote: UnixRemote.unix_remote, - }, - UnixLocal.unix_local_root: { - TextualDevice.not_connected: UnixLocal.unix_local, - UnixRemote.unix_remote: UnixLocal.unix_local, - UnixRemote.unix_remote_root: UnixLocal.unix_local, - AtRemote.at_remote: UnixLocal.unix_local, - }, - UnixRemote.unix_remote: { - TextualDevice.not_connected: UnixLocal.unix_local, - UnixLocal.unix_local_root: UnixLocal.unix_local, - }, - UnixRemote.unix_remote_root: { - TextualDevice.not_connected: UnixRemote.unix_remote, - UnixLocal.unix_local: UnixRemote.unix_remote, - UnixLocal.unix_local_root: UnixRemote.unix_remote, - AtRemote.at_remote: UnixRemote.unix_remote, + AtRemote.unix_local_root: AtRemote.unix_local, + AtRemote.unix_remote: AtRemote.unix_local, + AtRemote.unix_remote_root: AtRemote.unix_local, + AtRemote.at_remote: AtRemote.unix_local, + }, + AtRemote.unix_local: { + AtRemote.unix_remote_root: AtRemote.unix_remote, + AtRemote.at_remote: AtRemote.unix_remote, + }, + AtRemote.unix_local_root: { + TextualDevice.not_connected: AtRemote.unix_local, + AtRemote.unix_remote: AtRemote.unix_local, + AtRemote.unix_remote_root: AtRemote.unix_local, + AtRemote.at_remote: AtRemote.unix_local, + }, + AtRemote.unix_remote: { + TextualDevice.not_connected: AtRemote.unix_local, + AtRemote.unix_local_root: AtRemote.unix_local, + }, + AtRemote.unix_remote_root: { + TextualDevice.not_connected: AtRemote.unix_remote, + AtRemote.unix_local: AtRemote.unix_remote, + AtRemote.unix_local_root: AtRemote.unix_remote, + AtRemote.at_remote: AtRemote.unix_remote, + }, + AtRemote.at_remote: { + TextualDevice.not_connected: AtRemote.unix_remote, + AtRemote.unix_local: AtRemote.unix_remote, + AtRemote.unix_local_root: AtRemote.unix_remote, + AtRemote.unix_remote_root: AtRemote.unix_remote, + }, + } + return state_hops + + @mark_to_call_base_class_method_with_same_name + def _prepare_state_hops_with_proxy_pc(self): + """ + Prepare non direct transitions for each state for State Machine without proxy_pc state. + :return: non direct transitions for each state without proxy_pc state. + """ + state_hops = { + TextualDevice.not_connected: { + AtRemote.unix_local_root: AtRemote.unix_local, + AtRemote.proxy_pc: AtRemote.unix_local, + AtRemote.unix_remote: AtRemote.unix_local, + AtRemote.unix_remote_root: AtRemote.unix_local, + AtRemote.at_remote: AtRemote.unix_local, + }, + AtRemote.unix_local: { + AtRemote.unix_remote_root: AtRemote.proxy_pc, + AtRemote.at_remote: AtRemote.proxy_pc, + }, + AtRemote.unix_local_root: { + TextualDevice.not_connected: AtRemote.unix_local, + AtRemote.proxy_pc: AtRemote.unix_local, + AtRemote.unix_remote: AtRemote.unix_local, + AtRemote.unix_remote_root: AtRemote.unix_local, + AtRemote.at_remote: AtRemote.unix_local, + }, + AtRemote.unix_remote: { + AtRemote.unix_local: AtRemote.proxy_pc, + TextualDevice.not_connected: AtRemote.proxy_pc, + AtRemote.unix_local_root: AtRemote.proxy_pc, + }, + AtRemote.unix_remote_root: { + TextualDevice.not_connected: AtRemote.unix_remote, + AtRemote.proxy_pc: AtRemote.unix_remote, + AtRemote.unix_local: AtRemote.unix_remote, + AtRemote.unix_local_root: AtRemote.unix_remote, + AtRemote.at_remote: AtRemote.unix_remote, }, AtRemote.at_remote: { - TextualDevice.not_connected: UnixRemote.unix_remote, - UnixLocal.unix_local: UnixRemote.unix_remote, - UnixLocal.unix_local_root: UnixRemote.unix_remote, - UnixRemote.unix_remote_root: UnixRemote.unix_remote, + TextualDevice.not_connected: AtRemote.unix_remote, + AtRemote.proxy_pc: AtRemote.unix_remote, + AtRemote.unix_local: AtRemote.unix_remote, + AtRemote.unix_local_root: AtRemote.unix_remote, + AtRemote.unix_remote_root: AtRemote.unix_remote, }, + AtRemote.proxy_pc: { + AtRemote.unix_remote_root: AtRemote.unix_remote, + AtRemote.at_remote: AtRemote.unix_remote, + AtRemote.not_connected: AtRemote.unix_remote, + AtRemote.unix_local_root: AtRemote.unix_local, + } } return state_hops @@ -211,9 +347,9 @@ def _configure_state_machine(self, sm_params): # copy prompt for AT_REMOTE/ctrl_c from UNIX_REMOTE_ROOT/exit hops_config = self._configurations[TextualDevice.connection_hops] - remote_ux_root_exit_params = hops_config[UnixRemote.unix_remote_root][UnixRemote.unix_remote]["command_params"] + remote_ux_root_exit_params = hops_config[AtRemote.unix_remote_root][AtRemote.unix_remote]["command_params"] remote_ux_prompt = remote_ux_root_exit_params["expected_prompt"] - hops_config[AtRemote.at_remote][UnixRemote.unix_remote]["command_params"]["expected_prompt"] = remote_ux_prompt + hops_config[AtRemote.at_remote][AtRemote.unix_remote]["command_params"]["expected_prompt"] = remote_ux_prompt def _get_packages_for_state(self, state, observer): """ @@ -230,7 +366,7 @@ def _get_packages_for_state(self, state, observer): TextualDevice.events: ['moler.events.shared']} if available: return available[observer] - elif state == UnixRemote.unix_remote: # this is unix extended with plink_serial command + elif state == AtRemote.unix_remote: # this is unix extended with plink_serial command if observer == TextualDevice.cmds: available.append('moler.cmd.at.plink_serial') available.append('moler.cmd.at.cu') diff --git a/test/device/test_SM_at_remote.py b/test/device/test_SM_at_remote.py index bdf60543c..a920b43f6 100644 --- a/test/device/test_SM_at_remote.py +++ b/test/device/test_SM_at_remote.py @@ -1,6 +1,6 @@ -__author__ = 'Grzegorz Latuszek' -__copyright__ = 'Copyright (C) 2020, Nokia' -__email__ = 'grzegorz.latuszek@nokia.com' +__author__ = 'Grzegorz Latuszek, Marcin Usielski' +__copyright__ = 'Copyright (C) 2020-2024, Nokia' +__email__ = 'grzegorz.latuszek@nokia.com, marcin.usielski@nokia.com' import pytest @@ -14,6 +14,13 @@ def test_at_remote_device(device_connection, at_remote_output): iterate_over_device_states(device=at_remote) +def test_at_remote_device_proxy_pc(device_connection, at_remote_output_proxy_pc): + at_remote = get_device(name="AT_REMOTE_PROXY_PC", connection=device_connection, device_output=at_remote_output_proxy_pc, + test_file_path=__file__) + + iterate_over_device_states(device=at_remote) + + @pytest.fixture def at_remote_output(): plink_cmd_string = 'plink -serial COM5 |& awk \'BEGIN {print "COM5> port READY"} {print} END {print "^C"}\'' @@ -39,3 +46,34 @@ def at_remote_output(): } return output + + +@pytest.fixture +def at_remote_output_proxy_pc(): + plink_cmd_string = 'plink -serial COM5 |& awk \'BEGIN {print "COM5> port READY"} {print} END {print "^C"}\'' + output = { + "UNIX_LOCAL": { + 'TERM=xterm-mono ssh -l proxy_pc_login -o ServerAliveInterval=7 -o ServerAliveCountMax=2 proxy_pc_host': 'proxy_pc#', + 'su': 'local_root_prompt' + }, + "PROXY_PC": { + 'TERM=xterm-mono ssh -l remote_login -o ServerAliveInterval=7 -o ServerAliveCountMax=2 remote_host': 'remote#', + 'exit': 'moler_bash#', + }, + "UNIX_LOCAL_ROOT": { + 'exit': 'moler_bash#' + }, + "UNIX_REMOTE": { + 'exit': 'proxy_pc#', + 'su': 'remote_root_prompt', + plink_cmd_string: 'COM5>' + }, + "AT_REMOTE": { + '\x03': '^C\nremote#', + }, + "UNIX_REMOTE_ROOT": { + 'exit': 'remote#', + }, + } + + return output diff --git a/test/resources/device_config.yml b/test/resources/device_config.yml index 97b7bc64b..fee161a2a 100644 --- a/test/resources/device_config.yml +++ b/test/resources/device_config.yml @@ -385,6 +385,34 @@ DEVICES: command_params: serial_devname: 'COM5' + AT_REMOTE_PROXY_PC: + DEVICE_CLASS: moler.device.atremote.AtRemote + INITIAL_STATE: UNIX_LOCAL + CONNECTION_HOPS: + UNIX_LOCAL: + PROXY_PC: + execute_command: ssh + command_params: + expected_prompt: "proxy_pc#" + host: proxy_pc_host + login: proxy_pc_login + password: password + set_timeout: null + PROXY_PC: + UNIX_REMOTE: + execute_command: ssh # default value + command_params: + expected_prompt: 'remote#' + host: remote_host + login: remote_login + password: passwd4remote + set_timeout: null + UNIX_REMOTE: + AT_REMOTE: + execute_command: plink_serial # default value + command_params: + serial_devname: 'COM5' + ADB_REMOTE: DEVICE_CLASS: moler.device.adbremote.AdbRemote INITIAL_STATE: UNIX_LOCAL