From 64a6f4226a496144efbe4b575e265be2edb4f078 Mon Sep 17 00:00:00 2001 From: Marcin Usielski <35992110+marcin-usielski@users.noreply.github.com> Date: Tue, 5 Nov 2024 13:15:48 +0100 Subject: [PATCH] ProxyPC - Juniper (#549) * ProxyPC - Juniper * style * fail_under due to new devices3 * pdu_aten3 and scpi3 * style * no tests for adb_remote3 * no tests for adb_remote3 * no tests for adb_remote3 * timeout * adbremote * AdbRemote * style * AtRemote * pro * moler/device/adbremote3.py * dev * AdbRemote3 * style * no more tests for unix remote3 * atremote3 * Coverage on Python 3,11 * pytest-cov * timeout * f * timeout coverage * no coverage * review - JuniperGeneric3 * style * Python 3.13 --- .github/workflows/ci.yaml | 12 +- moler/cmd/unix/bzip2.py | 1 - moler/device/adbremote.py | 299 ++++++++++++++++--- moler/device/adbremote3.py | 433 ++++++++++++++++++++++++++++ moler/device/atremote.py | 248 ++++++++++++---- moler/device/atremote3.py | 282 ++++++++++++++++++ moler/device/juniper_ex3.py | 59 ++++ moler/device/junipergeneric.py | 9 +- moler/device/junipergeneric3.py | 235 +++++++++++++++ moler/device/pdu_aten.py | 2 +- moler/device/pdu_aten3.py | 237 +++++++++++++++ moler/device/proxy_pc3.py | 299 +++++++++++++++++++ moler/device/scpi.py | 1 + moler/device/scpi3.py | 242 ++++++++++++++++ moler/device/textualdevice.py | 2 +- moler/device/unixremote.py | 4 + moler/device/unixremote3.py | 74 +---- moler/helpers.py | 23 ++ test/device/test_SM_adb_remote.py | 74 ++++- test/device/test_SM_at_remote.py | 68 ++++- test/device/test_SM_juniper_ex.py | 32 +- test/device/test_SM_pdu_aten.py | 29 +- test/device/test_SM_scpi.py | 32 +- test/device/test_SM_unix_remote3.py | 195 ------------- test/resources/device_config.yml | 264 +++++++++++++++++ test/test_helpers.py | 114 ++++++++ 26 files changed, 2865 insertions(+), 405 deletions(-) create mode 100644 moler/device/adbremote3.py create mode 100644 moler/device/atremote3.py create mode 100644 moler/device/juniper_ex3.py create mode 100644 moler/device/junipergeneric3.py create mode 100644 moler/device/pdu_aten3.py create mode 100644 moler/device/proxy_pc3.py create mode 100644 moler/device/scpi3.py delete mode 100644 test/device/test_SM_unix_remote3.py diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index fe9e0dd7e..b3f78cad7 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -4,15 +4,15 @@ on: [push] jobs: test: - timeout-minutes: 10 + timeout-minutes: 15 runs-on: ${{ matrix.os }} strategy: matrix: os: [ubuntu-latest] - python-version: [3.7, 3.8, 3.9, "3.10", 3.11, 3.12] + python-version: [3.7, 3.8, 3.9, "3.10", 3.11, 3.12, 3.13] env: MOLER_DEBUG_THREADS: True - PYTHON_COVERAGE: '3.12' + PYTHON_COVERAGE: '3.15' steps: - uses: actions/checkout@v2 - name: Set up Python ${{ matrix.os }} ${{ matrix.python-version }} @@ -48,13 +48,13 @@ jobs: adb devices - name: Test with pytest if: ${{ matrix.python-version != env.PYTHON_COVERAGE }} - timeout-minutes: 9 + timeout-minutes: 14 run: | python -m pytest -vvvsss test/ # python -m pytest -c py3pytest.ini -vvvsss test/ - name: Test with pytest and coverage if: ${{ matrix.python-version == env.PYTHON_COVERAGE }} - timeout-minutes: 9 + timeout-minutes: 49 run: | pip list python --version @@ -107,7 +107,7 @@ jobs: # pydocstyle --count moler - name: Check PyLint run: | - pylint moler --fail-under=9.4 --rcfile=./pylint.cfg || (exit $(($? % 4))) + pylint moler --fail-under=8.4 --rcfile=./pylint.cfg || (exit $(($? % 4))) - name: Check flake8 run: | flake8 moler --count --select=E9,F63,F7,F82 --show-source --statistics diff --git a/moler/cmd/unix/bzip2.py b/moler/cmd/unix/bzip2.py index 1dcc980a6..1da914827 100644 --- a/moler/cmd/unix/bzip2.py +++ b/moler/cmd/unix/bzip2.py @@ -47,7 +47,6 @@ def build_command_string(self): return cmd def on_new_line(self, line, is_full_line): - print(f"on_new_line: {is_full_line}-> '{line}'") try: if is_full_line: self._parse_error_via_output_line(line) diff --git a/moler/device/adbremote.py b/moler/device/adbremote.py index 2b0ada00b..5dd8ef10e 100644 --- a/moler/device/adbremote.py +++ b/moler/device/adbremote.py @@ -5,15 +5,13 @@ - 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' import logging from moler.device.textualdevice import TextualDevice -# from moler.device.proxy_pc import ProxyPc # TODO: allow jumping towards ADB_REMOTE via proxy-pc -from moler.device.unixlocal import UnixLocal from moler.device.unixremote import UnixRemote from moler.cmd.adb.adb_shell import AdbShell from moler.helpers import call_base_class_method_with_same_name, mark_to_call_base_class_method_with_same_name @@ -86,7 +84,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 + AdbRemote.unix_remote: { # from AdbRemote.adb_shell: { # to "execute_command": "adb_shell", "command_params": { # with parameters @@ -99,7 +97,7 @@ def _get_default_sm_configuration_without_proxy_pc(self): }, }, AdbRemote.adb_shell: { # from - UnixRemote.unix_remote: { # to + AdbRemote.unix_remote: { # to "execute_command": "exit", # using command "command_params": { # with parameters "expected_prompt": r'remote_user_prompt', # overwritten in _configure_state_machine() @@ -135,6 +133,99 @@ def _get_default_sm_configuration_without_proxy_pc(self): } return config + @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 = { # TODO: shell we use direct-string names of config dicts? change simplicity vs readability + TextualDevice.connection_hops: { + AdbRemote.unix_remote: { # from + AdbRemote.adb_shell: { # to + "execute_command": "adb_shell", + "command_params": { # with parameters + "target_newline": "\n", + "prompt_from_serial_number": True, + }, + "required_command_params": [ + "serial_number", + ] + }, + }, + AdbRemote.adb_shell: { # from + AdbRemote.unix_remote: { # to + "execute_command": "exit", # using command + "command_params": { # with parameters + "expected_prompt": r'remote_user_prompt', # overwritten in _configure_state_machine() + "target_newline": "\n", + "allowed_newline_after_prompt": True, + }, + "required_command_params": [ + ] + }, + AdbRemote.adb_shell_root: { # to + "execute_command": "su", # using command + "command_params": { # with parameters + "password": "provide_root_password_in_cfg", # if su requires passwd and not given in cfg + "expected_prompt": None, # overwritten in _prepare_state_prompts...() + "target_newline": None, # overwritten in _prepare_newline_chars_...() + }, + "required_command_params": [ + ] + }, + }, + AdbRemote.adb_shell_root: { # from + AdbRemote.adb_shell: { # to + "execute_command": "exit", # using command + "command_params": { # with parameters + "expected_prompt": r'adb_prompt', # overwritten in _configure_state_machine() + "target_newline": "\n" + }, + "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 = { + AdbRemote.unix_remote: { + AdbRemote.adb_shell: { + "action": [ + "_execute_command_to_change_state" + ], + } + }, + AdbRemote.adb_shell: { + AdbRemote.unix_remote: { + "action": [ + "_execute_command_to_change_state" + ], + }, + AdbRemote.adb_shell_root: { + "action": [ + "_execute_command_to_change_state" + ], + } + }, + AdbRemote.adb_shell_root: { + AdbRemote.adb_shell: { + "action": [ + "_execute_command_to_change_state" + ], + }, + }, + } + return transitions + @mark_to_call_base_class_method_with_same_name def _prepare_transitions_without_proxy_pc(self): """ @@ -142,7 +233,7 @@ def _prepare_transitions_without_proxy_pc(self): :return: transitions without proxy_pc state. """ transitions = { - UnixRemote.unix_remote: { + AdbRemote.unix_remote: { AdbRemote.adb_shell: { "action": [ "_execute_command_to_change_state" @@ -150,7 +241,7 @@ def _prepare_transitions_without_proxy_pc(self): } }, AdbRemote.adb_shell: { - UnixRemote.unix_remote: { + AdbRemote.unix_remote: { "action": [ "_execute_command_to_change_state" ], @@ -178,7 +269,34 @@ 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] - cfg_ux2adb = hops_config[UnixRemote.unix_remote][AdbRemote.adb_shell] + cfg_ux2adb = hops_config[AdbRemote.unix_remote][AdbRemote.adb_shell] + cfg_adb2adbroot = hops_config[AdbRemote.adb_shell][AdbRemote.adb_shell_root] + adb_shell_cmd_params = cfg_ux2adb["command_params"] + adb_shell_prompt = self._get_adb_shell_prompt(adb_shell_cmd_params) + adb_shell_root_prompt = cfg_adb2adbroot["command_params"]["expected_prompt"] + if adb_shell_root_prompt is None: + if adb_shell_prompt.endswith("$"): + adb_shell_root_prompt = f"{adb_shell_prompt[:-1]}#" + else: + consequence = f"Won't be able to detect {AdbRemote.adb_shell_root} state" + fix = f"Please provide configuration with 'expected_prompt' for {AdbRemote.adb_shell_root} state" + self._log(logging.WARNING, f"Unknown prompt for {AdbRemote.adb_shell_root} state. {consequence}. {fix}.") + adb_shell_root_prompt = "Unknown_adb_root_prompt" + + state_prompts = { + AdbRemote.adb_shell: adb_shell_prompt, + AdbRemote.adb_shell_root: adb_shell_root_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] + cfg_ux2adb = hops_config[AdbRemote.unix_remote][AdbRemote.adb_shell] cfg_adb2adbroot = hops_config[AdbRemote.adb_shell][AdbRemote.adb_shell_root] adb_shell_cmd_params = cfg_ux2adb["command_params"] adb_shell_prompt = self._get_adb_shell_prompt(adb_shell_cmd_params) @@ -206,7 +324,7 @@ def _serial_number(self): :return: serial_number. """ hops_config = self._configurations[TextualDevice.connection_hops] - cfg_ux2adb = hops_config[UnixRemote.unix_remote][AdbRemote.adb_shell] + cfg_ux2adb = hops_config[AdbRemote.unix_remote][AdbRemote.adb_shell] serial_number = cfg_ux2adb["command_params"]["serial_number"] return serial_number @@ -226,7 +344,27 @@ 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] - cfg_ux2adb = hops_config[UnixRemote.unix_remote][AdbRemote.adb_shell] + cfg_ux2adb = hops_config[AdbRemote.unix_remote][AdbRemote.adb_shell] + cfg_adb2adbroot = hops_config[AdbRemote.adb_shell][AdbRemote.adb_shell_root] + adb_shell_newline = cfg_ux2adb["command_params"]["target_newline"] + adb_shell_root_newline = cfg_adb2adbroot["command_params"]["target_newline"] + if adb_shell_root_newline is None: + adb_shell_root_newline = adb_shell_newline # we are on same machine just changing to root + + newline_chars = { + AdbRemote.adb_shell: adb_shell_newline, + AdbRemote.adb_shell_root: adb_shell_root_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] + cfg_ux2adb = hops_config[AdbRemote.unix_remote][AdbRemote.adb_shell] cfg_adb2adbroot = hops_config[AdbRemote.adb_shell][AdbRemote.adb_shell_root] adb_shell_newline = cfg_ux2adb["command_params"]["target_newline"] adb_shell_root_newline = cfg_adb2adbroot["command_params"]["target_newline"] @@ -247,49 +385,111 @@ 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, - AdbRemote.adb_shell: UnixLocal.unix_local, - AdbRemote.adb_shell_root: UnixLocal.unix_local, + AdbRemote.unix_local_root: AdbRemote.unix_local, + AdbRemote.unix_remote: AdbRemote.unix_local, + AdbRemote.unix_remote_root: AdbRemote.unix_local, + AdbRemote.adb_shell: AdbRemote.unix_local, + AdbRemote.adb_shell_root: AdbRemote.unix_local, + }, + AdbRemote.unix_local: { + AdbRemote.unix_remote_root: AdbRemote.unix_remote, + AdbRemote.adb_shell: AdbRemote.unix_remote, + AdbRemote.adb_shell_root: AdbRemote.unix_remote, + }, + AdbRemote.unix_local_root: { + TextualDevice.not_connected: AdbRemote.unix_local, + AdbRemote.unix_remote: AdbRemote.unix_local, + AdbRemote.unix_remote_root: AdbRemote.unix_local, + AdbRemote.adb_shell: AdbRemote.unix_local, + AdbRemote.adb_shell_root: AdbRemote.unix_local, + }, + AdbRemote.unix_remote: { + AdbRemote.adb_shell_root: AdbRemote.adb_shell, + }, + AdbRemote.unix_remote_root: { + TextualDevice.not_connected: AdbRemote.unix_remote, + AdbRemote.unix_local: AdbRemote.unix_remote, + AdbRemote.unix_local_root: AdbRemote.unix_remote, + AdbRemote.adb_shell: AdbRemote.unix_remote, + AdbRemote.adb_shell_root: AdbRemote.unix_remote, + }, + AdbRemote.adb_shell: { + TextualDevice.not_connected: AdbRemote.unix_remote, + AdbRemote.unix_local: AdbRemote.unix_remote, + AdbRemote.unix_local_root: AdbRemote.unix_remote, + AdbRemote.unix_remote_root: AdbRemote.unix_remote, + }, + AdbRemote.adb_shell_root: { + TextualDevice.not_connected: AdbRemote.adb_shell, + AdbRemote.unix_local: AdbRemote.adb_shell, + AdbRemote.unix_local_root: AdbRemote.adb_shell, + AdbRemote.unix_remote: AdbRemote.adb_shell, + AdbRemote.unix_remote_root: AdbRemote.adb_shell, + }, + } + 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: { + AdbRemote.unix_local_root: AdbRemote.unix_local, + AdbRemote.proxy_pc: AdbRemote.unix_local, + AdbRemote.unix_remote: AdbRemote.unix_local, + AdbRemote.unix_remote_root: AdbRemote.unix_local, + AdbRemote.adb_shell: AdbRemote.unix_local, + AdbRemote.adb_shell_root: AdbRemote.unix_local, }, - UnixLocal.unix_local: { - UnixRemote.unix_remote_root: UnixRemote.unix_remote, - AdbRemote.adb_shell: UnixRemote.unix_remote, - AdbRemote.adb_shell_root: UnixRemote.unix_remote, + AdbRemote.unix_local: { + AdbRemote.unix_remote_root: AdbRemote.proxy_pc, + AdbRemote.adb_shell: AdbRemote.proxy_pc, + AdbRemote.adb_shell_root: AdbRemote.proxy_pc, }, - UnixLocal.unix_local_root: { - TextualDevice.not_connected: UnixLocal.unix_local, - UnixRemote.unix_remote: UnixLocal.unix_local, - UnixRemote.unix_remote_root: UnixLocal.unix_local, - AdbRemote.adb_shell: UnixLocal.unix_local, - AdbRemote.adb_shell_root: UnixLocal.unix_local, + AdbRemote.unix_local_root: { + TextualDevice.not_connected: AdbRemote.unix_local, + AdbRemote.proxy_pc: AdbRemote.unix_local, + AdbRemote.unix_remote: AdbRemote.unix_local, + AdbRemote.unix_remote_root: AdbRemote.unix_local, + AdbRemote.adb_shell: AdbRemote.unix_local, + AdbRemote.adb_shell_root: AdbRemote.unix_local, }, - UnixRemote.unix_remote: { - TextualDevice.not_connected: UnixLocal.unix_local, - UnixLocal.unix_local_root: UnixLocal.unix_local, + AdbRemote.unix_remote: { AdbRemote.adb_shell_root: AdbRemote.adb_shell, }, - UnixRemote.unix_remote_root: { - TextualDevice.not_connected: UnixRemote.unix_remote, - UnixLocal.unix_local: UnixRemote.unix_remote, - UnixLocal.unix_local_root: UnixRemote.unix_remote, - AdbRemote.adb_shell: UnixRemote.unix_remote, - AdbRemote.adb_shell_root: UnixRemote.unix_remote, + AdbRemote.unix_remote_root: { + TextualDevice.not_connected: AdbRemote.unix_remote, + AdbRemote.proxy_pc: AdbRemote.unix_remote, + AdbRemote.unix_local: AdbRemote.unix_remote, + AdbRemote.unix_local_root: AdbRemote.unix_remote, + AdbRemote.adb_shell: AdbRemote.unix_remote, + AdbRemote.adb_shell_root: AdbRemote.unix_remote, }, AdbRemote.adb_shell: { - 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: AdbRemote.unix_remote, + AdbRemote.proxy_pc: AdbRemote.unix_remote, + AdbRemote.unix_local: AdbRemote.unix_remote, + AdbRemote.unix_local_root: AdbRemote.unix_remote, + AdbRemote.unix_remote_root: AdbRemote.unix_remote, }, AdbRemote.adb_shell_root: { TextualDevice.not_connected: AdbRemote.adb_shell, - UnixLocal.unix_local: AdbRemote.adb_shell, - UnixLocal.unix_local_root: AdbRemote.adb_shell, - UnixRemote.unix_remote: AdbRemote.adb_shell, - UnixRemote.unix_remote_root: AdbRemote.adb_shell, + AdbRemote.proxy_pc: AdbRemote.adb_shell, + AdbRemote.unix_local: AdbRemote.adb_shell, + AdbRemote.unix_local_root: AdbRemote.adb_shell, + AdbRemote.unix_remote: AdbRemote.adb_shell, + AdbRemote.unix_remote_root: AdbRemote.adb_shell, }, + AdbRemote.proxy_pc: { + AdbRemote.unix_remote_root: AdbRemote.unix_remote, + AdbRemote.adb_shell: AdbRemote.unix_remote, + AdbRemote.adb_shell_root: AdbRemote.unix_remote, + AdbRemote.not_connected: AdbRemote.unix_local, + AdbRemote.unix_local_root: AdbRemote.unix_local, + } } return state_hops @@ -304,13 +504,16 @@ def _configure_state_machine(self, sm_params): hops_config = self._configurations[TextualDevice.connection_hops] # copy prompt for ADB_SHELL/exit from UNIX_LOCAL/ssh - cfg_uxloc2ux = hops_config[UnixLocal.unix_local][UnixRemote.unix_remote] - cfg_adb2ux = hops_config[AdbRemote.adb_shell][UnixRemote.unix_remote] + if self._use_proxy_pc: + cfg_uxloc2ux = hops_config[AdbRemote.proxy_pc][AdbRemote.unix_remote] + else: + cfg_uxloc2ux = hops_config[AdbRemote.unix_local][AdbRemote.unix_remote] + cfg_adb2ux = hops_config[AdbRemote.adb_shell][AdbRemote.unix_remote] remote_ux_prompt = cfg_uxloc2ux["command_params"]["expected_prompt"] cfg_adb2ux["command_params"]["expected_prompt"] = remote_ux_prompt # copy prompt for ADB_SHELL_ROOT/exit from UNIX_REMOTE/adb shell - cfg_ux2adb = hops_config[UnixRemote.unix_remote][AdbRemote.adb_shell] + cfg_ux2adb = hops_config[AdbRemote.unix_remote][AdbRemote.adb_shell] cfg_adbroot2adb = hops_config[AdbRemote.adb_shell_root][AdbRemote.adb_shell] adb_shell_prompt = self._get_adb_shell_prompt(cfg_ux2adb["command_params"]) cfg_adbroot2adb["command_params"]["expected_prompt"] = adb_shell_prompt @@ -337,7 +540,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 adb commands + elif state == AdbRemote.unix_remote: # this is unix extended with adb commands if observer == TextualDevice.cmds: available.append('moler.cmd.adb') diff --git a/moler/device/adbremote3.py b/moler/device/adbremote3.py new file mode 100644 index 000000000..f5975ab3c --- /dev/null +++ b/moler/device/adbremote3.py @@ -0,0 +1,433 @@ +# -*- coding: utf-8 -*- +""" +Moler's device has 2 main responsibilities: +- be the factory that returns commands of that device +- be the state machine that controls which commands may run in given state +""" + +__author__ = 'Grzegorz Latuszek, Marcin Usielski' +__copyright__ = 'Copyright (C) 2020-2024, Nokia' +__email__ = 'grzegorz.latuszek@nokia.com, marcin.usielski@nokia.com' + +import logging + +from moler.device.unixremote3 import UnixRemote3 +from moler.cmd.adb.adb_shell import AdbShell +from moler.helpers import call_base_class_method_with_same_name, mark_to_call_base_class_method_with_same_name + + +@call_base_class_method_with_same_name +class AdbRemote3(UnixRemote3): + r""" + AdbRemote device class. + + + :: + + + Example of device in yaml configuration file: + -without PROXY_PC: + ADB_1: + DEVICE_CLASS: moler.device.adbremote3.AdbRemote3 + 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: + ADB_SHELL: + execute_command: adb_shell # default value; default command is: adb shell + command_params: + serial_number: 'f57e6b7d' # to create: adb -s f57e6b7d shell + expected_prompt: 'shell@adbhost: $' + ADB_SHELL: + UNIX_REMOTE: + execute_command: exit # default value + """ + + adb_shell = "ADB_SHELL" + adb_shell_root = "ADB_SHELL_ROOT" + + def __init__(self, sm_params, name=None, io_connection=None, io_type=None, variant=None, io_constructor_kwargs=None, + initial_state=None, lazy_cmds_events=False): + """ + Create ADB device communicating over io_connection + :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', ... + (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) + :param initial_state: name of initial state. State machine tries to enter this state just after creation. + :param lazy_cmds_events: set False to load all commands and events when device is initialized, set True to load + commands and events when they are required for the first time. + """ + initial_state = initial_state if initial_state is not None else AdbRemote3.adb_shell + super(AdbRemote3, self).__init__(name=name, io_connection=io_connection, + io_type=io_type, variant=variant, + io_constructor_kwargs=io_constructor_kwargs, + sm_params=sm_params, initial_state=initial_state, + lazy_cmds_events=lazy_cmds_events) + + @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 = { + AdbRemote3.connection_hops: { + AdbRemote3.unix_remote: { # from + AdbRemote3.adb_shell: { # to + "execute_command": "adb_shell", + "command_params": { # with parameters + "target_newline": "\n", + "prompt_from_serial_number": True, + }, + "required_command_params": [ + "serial_number", + ] + }, + }, + AdbRemote3.adb_shell: { # from + AdbRemote3.unix_remote: { # to + "execute_command": "exit", # using command + "command_params": { # with parameters + "expected_prompt": r'remote_user_prompt', # overwritten in _configure_state_machine() + "target_newline": "\n", + "allowed_newline_after_prompt": True, + }, + "required_command_params": [ + ] + }, + AdbRemote3.adb_shell_root: { # to + "execute_command": "su", # using command + "command_params": { # with parameters + "password": "provide_root_password_in_cfg", # if su requires passwd and not given in cfg + "expected_prompt": None, # overwritten in _prepare_state_prompts...() + "target_newline": None, # overwritten in _prepare_newline_chars_...() + }, + "required_command_params": [ + ] + }, + }, + AdbRemote3.adb_shell_root: { # from + AdbRemote3.adb_shell: { # to + "execute_command": "exit", # using command + "command_params": { # with parameters + "expected_prompt": r'adb_prompt', # overwritten in _configure_state_machine() + "target_newline": "\n" + }, + "required_command_params": [ + ] + }, + }, + } + } + return config + + def _overwrite_prompts(self): + """ + Overwrite prompts for some states to easily configure the SM. + """ + super(AdbRemote3, self)._overwrite_prompts() + hops_config = self._configurations[AdbRemote3.connection_hops] + cfg_ux2adb = hops_config[AdbRemote3.unix_remote][AdbRemote3.adb_shell] + cfg_adb2adbroot = hops_config[AdbRemote3.adb_shell][AdbRemote3.adb_shell_root] + adb_shell_cmd_params = cfg_ux2adb["command_params"] + adb_shell_prompt = self._get_adb_shell_prompt(adb_shell_cmd_params) + adb_shell_root_prompt = cfg_adb2adbroot["command_params"]["expected_prompt"] + if adb_shell_root_prompt is None: + if adb_shell_prompt.endswith("$"): + adb_shell_root_prompt = f"{adb_shell_prompt[:-1]}#" + else: + consequence = f"Won't be able to detect {AdbRemote3.adb_shell_root} state" + fix = f"Please provide configuration with 'expected_prompt' for {AdbRemote3.adb_shell_root} state" + self._log(logging.WARNING, f"Unknown prompt for {AdbRemote3.adb_shell_root} state. {consequence}. {fix}.") + adb_shell_root_prompt = "Unknown_adb_root_prompt" + + if self._use_proxy_pc: + cfg_uxloc2ux = hops_config[AdbRemote3.proxy_pc][AdbRemote3.unix_remote] + else: + cfg_uxloc2ux = hops_config[AdbRemote3.unix_local][AdbRemote3.unix_remote] + cfg_adb2ux = hops_config[AdbRemote3.adb_shell][AdbRemote3.unix_remote] + remote_ux_prompt = cfg_uxloc2ux["command_params"]["expected_prompt"] + cfg_adb2ux["command_params"]["expected_prompt"] = remote_ux_prompt + + # copy prompt for ADB_SHELL_ROOT/exit from UNIX_REMOTE/adb shell + cfg_adbroot2adb = hops_config[AdbRemote3.adb_shell_root][AdbRemote3.adb_shell] + cfg_adbroot2adb["command_params"]["expected_prompt"] = adb_shell_prompt + cfg_adb2adbroot["command_params"]["expected_prompt"] = adb_shell_root_prompt + + @mark_to_call_base_class_method_with_same_name + def _prepare_state_prompts_without_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[AdbRemote3.connection_hops] + cfg_ux2adb = hops_config[AdbRemote3.unix_remote][AdbRemote3.adb_shell] + cfg_adb2adbroot = hops_config[AdbRemote3.adb_shell][AdbRemote3.adb_shell_root] + adb_shell_cmd_params = cfg_ux2adb["command_params"] + adb_shell_prompt = self._get_adb_shell_prompt(adb_shell_cmd_params) + adb_shell_root_prompt = cfg_adb2adbroot["command_params"]["expected_prompt"] + if adb_shell_root_prompt is None: + if adb_shell_prompt.endswith("$"): + adb_shell_root_prompt = f"{adb_shell_prompt[:-1]}#" + else: + consequence = f"Won't be able to detect {AdbRemote3.adb_shell_root} state" + fix = f"Please provide configuration with 'expected_prompt' for {AdbRemote3.adb_shell_root} state" + self._log(logging.WARNING, f"Unknown prompt for {AdbRemote3.adb_shell_root} state. {consequence}. {fix}.") + adb_shell_root_prompt = "Unknown_adb_root_prompt" + + state_prompts = { + AdbRemote3.adb_shell: adb_shell_prompt, + AdbRemote3.adb_shell_root: adb_shell_root_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[AdbRemote3.connection_hops] + cfg_ux2adb = hops_config[AdbRemote3.unix_remote][AdbRemote3.adb_shell] + cfg_adb2adbroot = hops_config[AdbRemote3.adb_shell][AdbRemote3.adb_shell_root] + adb_shell_cmd_params = cfg_ux2adb["command_params"] + adb_shell_prompt = self._get_adb_shell_prompt(adb_shell_cmd_params) + adb_shell_root_prompt = cfg_adb2adbroot["command_params"]["expected_prompt"] + if adb_shell_root_prompt is None: + if adb_shell_prompt.endswith("$"): + adb_shell_root_prompt = f"{adb_shell_prompt[:-1]}#" + else: + consequence = f"Won't be able to detect {AdbRemote3.adb_shell_root} state" + fix = f"Please provide configuration with 'expected_prompt' for {AdbRemote3.adb_shell_root} state" + self._log(logging.WARNING, f"Unknown prompt for {AdbRemote3.adb_shell_root} state. {consequence}. {fix}.") + adb_shell_root_prompt = "Unknown_adb_root_prompt" + + state_prompts = { + AdbRemote3.adb_shell: adb_shell_prompt, + AdbRemote3.adb_shell_root: adb_shell_root_prompt, + } + return state_prompts + + @property + def _serial_number(self): + """ + Retrieve serial_number based on required parameter of state machine. + + :return: serial_number. + """ + hops_config = self._configurations[AdbRemote3.connection_hops] + cfg_ux2adb = hops_config[AdbRemote3.unix_remote][AdbRemote3.adb_shell] + serial_number = cfg_ux2adb["command_params"]["serial_number"] + return serial_number + + def _get_adb_shell_prompt(self, adb_shell_cmd_params): + adb_shell_prompt = None + if 'expected_prompt' in adb_shell_cmd_params: + adb_shell_prompt = adb_shell_cmd_params["expected_prompt"] + if not adb_shell_prompt: + # adb_shell@f57e6b77 $ + adb_shell_prompt = AdbShell.re_generated_prompt.format(self._serial_number) # pylint-disable-line: consider-using-f-string + return adb_shell_prompt + + @mark_to_call_base_class_method_with_same_name + def _prepare_newline_chars_without_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[AdbRemote3.connection_hops] + cfg_ux2adb = hops_config[AdbRemote3.unix_remote][AdbRemote3.adb_shell] + cfg_adb2adbroot = hops_config[AdbRemote3.adb_shell][AdbRemote3.adb_shell_root] + adb_shell_newline = cfg_ux2adb["command_params"]["target_newline"] + adb_shell_root_newline = cfg_adb2adbroot["command_params"]["target_newline"] + if adb_shell_root_newline is None: + adb_shell_root_newline = adb_shell_newline # we are on same machine just changing to root + + newline_chars = { + AdbRemote3.adb_shell: adb_shell_newline, + AdbRemote3.adb_shell_root: adb_shell_root_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[AdbRemote3.connection_hops] + cfg_ux2adb = hops_config[AdbRemote3.unix_remote][AdbRemote3.adb_shell] + cfg_adb2adbroot = hops_config[AdbRemote3.adb_shell][AdbRemote3.adb_shell_root] + adb_shell_newline = cfg_ux2adb["command_params"]["target_newline"] + adb_shell_root_newline = cfg_adb2adbroot["command_params"]["target_newline"] + if adb_shell_root_newline is None: + adb_shell_root_newline = adb_shell_newline # we are on same machine just changing to root + + newline_chars = { + AdbRemote3.adb_shell: adb_shell_newline, + AdbRemote3.adb_shell_root: adb_shell_root_newline, + } + return newline_chars + + @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 = { + AdbRemote3.not_connected: { + AdbRemote3.unix_local_root: AdbRemote3.unix_local, + AdbRemote3.proxy_pc: AdbRemote3.unix_local, + AdbRemote3.unix_remote: AdbRemote3.unix_local, + AdbRemote3.unix_remote_root: AdbRemote3.unix_local, + AdbRemote3.adb_shell: AdbRemote3.unix_local, + AdbRemote3.adb_shell_root: AdbRemote3.unix_local, + }, + AdbRemote3.unix_local: { + AdbRemote3.unix_remote_root: AdbRemote3.proxy_pc, + AdbRemote3.adb_shell: AdbRemote3.proxy_pc, + AdbRemote3.adb_shell_root: AdbRemote3.proxy_pc, + }, + AdbRemote3.unix_local_root: { + AdbRemote3.not_connected: AdbRemote3.unix_local, + AdbRemote3.proxy_pc: AdbRemote3.unix_local, + AdbRemote3.unix_remote: AdbRemote3.unix_local, + AdbRemote3.unix_remote_root: AdbRemote3.unix_local, + AdbRemote3.adb_shell: AdbRemote3.unix_local, + AdbRemote3.adb_shell_root: AdbRemote3.unix_local, + }, + AdbRemote3.unix_remote: { + AdbRemote3.unix_local: AdbRemote3.proxy_pc, + AdbRemote3.unix_local_root: AdbRemote3.proxy_pc, + AdbRemote3.adb_shell_root: AdbRemote3.adb_shell, + }, + AdbRemote3.unix_remote_root: { + AdbRemote3.not_connected: AdbRemote3.unix_remote, + AdbRemote3.proxy_pc: AdbRemote3.unix_remote, + AdbRemote3.unix_local: AdbRemote3.unix_remote, + AdbRemote3.unix_local_root: AdbRemote3.unix_remote, + AdbRemote3.adb_shell: AdbRemote3.unix_remote, + AdbRemote3.adb_shell_root: AdbRemote3.unix_remote, + }, + AdbRemote3.adb_shell: { + AdbRemote3.not_connected: AdbRemote3.unix_remote, + AdbRemote3.proxy_pc: AdbRemote3.unix_remote, + AdbRemote3.unix_local: AdbRemote3.unix_remote, + AdbRemote3.unix_local_root: AdbRemote3.unix_remote, + AdbRemote3.unix_remote_root: AdbRemote3.unix_remote, + }, + AdbRemote3.adb_shell_root: { + AdbRemote3.not_connected: AdbRemote3.adb_shell, + AdbRemote3.proxy_pc: AdbRemote3.adb_shell, + AdbRemote3.unix_local: AdbRemote3.adb_shell, + AdbRemote3.unix_local_root: AdbRemote3.adb_shell, + AdbRemote3.unix_remote: AdbRemote3.adb_shell, + AdbRemote3.unix_remote_root: AdbRemote3.adb_shell, + }, + AdbRemote3.proxy_pc: { + AdbRemote3.adb_shell: AdbRemote3.unix_remote, + AdbRemote3.adb_shell_root: AdbRemote3.unix_remote, + } + } + return state_hops + + def _configure_state_machine(self, sm_params): + """ + Configure device State Machine. + :param sm_params: dict with parameters of state machine for device. + :return: None. + """ + super(AdbRemote3, self)._configure_state_machine(sm_params) + + hops_config = self._configurations[AdbRemote3.connection_hops] + + # copy prompt for ADB_SHELL/exit from UNIX_LOCAL/ssh + if self._use_proxy_pc: + cfg_uxloc2ux = hops_config[AdbRemote3.proxy_pc][AdbRemote3.unix_remote] + else: + cfg_uxloc2ux = hops_config[AdbRemote3.unix_local][AdbRemote3.unix_remote] + cfg_adb2ux = hops_config[AdbRemote3.adb_shell][AdbRemote3.unix_remote] + remote_ux_prompt = cfg_uxloc2ux["command_params"]["expected_prompt"] + cfg_adb2ux["command_params"]["expected_prompt"] = remote_ux_prompt + + # copy prompt for ADB_SHELL_ROOT/exit from UNIX_REMOTE/adb shell + cfg_ux2adb = hops_config[AdbRemote3.unix_remote][AdbRemote3.adb_shell] + cfg_adbroot2adb = hops_config[AdbRemote3.adb_shell_root][AdbRemote3.adb_shell] + adb_shell_prompt = self._get_adb_shell_prompt(cfg_ux2adb["command_params"]) + cfg_adbroot2adb["command_params"]["expected_prompt"] = adb_shell_prompt + + cfg_adb2adbroot = hops_config[AdbRemote3.adb_shell][AdbRemote3.adb_shell_root] + adb_shell_root_prompt = cfg_adb2adbroot["command_params"]["expected_prompt"] + if adb_shell_root_prompt is None: + if adb_shell_prompt.endswith("$"): + adb_shell_root_prompt = f"{adb_shell_prompt[:-1]}#" + cfg_adb2adbroot["command_params"]["expected_prompt"] = adb_shell_root_prompt + + def _get_packages_for_state(self, state, observer): + """ + Get available packages containing cmds and events for each state. + :param state: device state. + :param observer: observer type, available: cmd, events + :return: available cmds or events for specific device state. + """ + available = super(AdbRemote3, self)._get_packages_for_state(state, observer) + + if not available: + if (state == AdbRemote3.adb_shell) or (state == AdbRemote3.adb_shell_root): + available = {AdbRemote3.cmds: ['moler.cmd.unix'], + AdbRemote3.events: ['moler.events.shared']} + if available: + return available[observer] + elif state == AdbRemote3.unix_remote: # this is unix extended with adb commands + if observer == AdbRemote3.cmds: + available.append('moler.cmd.adb') + + return available + + @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 = { + AdbRemote3.unix_remote: { + AdbRemote3.adb_shell: { + "action": [ + "_execute_command_to_change_state" + ], + } + }, + AdbRemote3.adb_shell: { + AdbRemote3.unix_remote: { + "action": [ + "_execute_command_to_change_state" + ], + }, + AdbRemote3.adb_shell_root: { + "action": [ + "_execute_command_to_change_state" + ], + } + }, + AdbRemote3.adb_shell_root: { + AdbRemote3.adb_shell: { + "action": [ + "_execute_command_to_change_state" + ], + }, + }, + } + return transitions diff --git a/moler/device/atremote.py b/moler/device/atremote.py index c8216e408..2ad44d19f 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_local, + 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/moler/device/atremote3.py b/moler/device/atremote3.py new file mode 100644 index 000000000..bccc116cd --- /dev/null +++ b/moler/device/atremote3.py @@ -0,0 +1,282 @@ +# -*- coding: utf-8 -*- +""" +Moler's device has 2 main responsibilities: +- be the factory that returns commands of that device +- be the state machine that controls which commands may run in given state +""" + +__author__ = 'Grzegorz Latuszek, Marcin Usielski' +__copyright__ = 'Copyright (C) 2020-2024, Nokia' +__email__ = 'grzegorz.latuszek@nokia.com, marcin.usielski@nokia.com' + +from moler.device.unixremote3 import UnixRemote3 +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 + + +@call_base_class_method_with_same_name +class AtRemote3(UnixRemote3): + r""" + AtRemote device class. + + + :: + + + Example of device in yaml configuration file: + -without PROXY_PC: + AT_1: + DEVICE_CLASS: moler.device.atremote3.AtRemote3 + 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: + AT_REMOTE: + execute_command: plink_serial # default value + command_params: + serial_devname: 'COM5' + AT_REMOTE: + UNIX_REMOTE: + execute_command: ctrl_c # default value + """ + + at_remote = "AT_REMOTE" + + def __init__(self, sm_params, name=None, io_connection=None, io_type=None, variant=None, io_constructor_kwargs=None, + initial_state=None, lazy_cmds_events=False): + """ + Create AT device communicating over io_connection + :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', .. + (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) + :param initial_state: name of initial state. State machine tries to enter this state just after creation. + :param lazy_cmds_events: set False to load all commands and events when device is initialized, set True to load + commands and events when they are required for the first time. + """ + initial_state = initial_state if initial_state is not None else AtRemote3.at_remote + super(AtRemote3, self).__init__(name=name, io_connection=io_connection, + io_type=io_type, variant=variant, + io_constructor_kwargs=io_constructor_kwargs, + sm_params=sm_params, initial_state=initial_state, + lazy_cmds_events=lazy_cmds_events) + + def _overwrite_prompts(self): + super()._overwrite_prompts() + # copy prompt for AT_REMOTE/ctrl_c from UNIX_REMOTE_ROOT/exit + hops_config = self._configurations[AtRemote3.connection_hops] + remote_ux_root_exit_params = hops_config[AtRemote3.unix_remote_root][AtRemote3.unix_remote]["command_params"] + remote_ux_prompt = remote_ux_root_exit_params["expected_prompt"] + hops_config[AtRemote3.at_remote][AtRemote3.unix_remote]["command_params"]["expected_prompt"] = remote_ux_prompt + + @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 = { + AtRemote3.connection_hops: { + AtRemote3.unix_remote: { # from + AtRemote3.at_remote: { # to + "execute_command": "plink_serial", + "command_params": { # with parameters + "target_newline": "\r\n" + }, + "required_command_params": [ + "serial_devname" + ] + }, + }, + AtRemote3.at_remote: { # from + AtRemote3.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 = { + AtRemote3.unix_remote: { + AtRemote3.at_remote: { + "action": [ + "_execute_command_to_change_state" + ], + } + }, + AtRemote3.at_remote: { + AtRemote3.unix_remote: { + "action": [ + "_execute_command_to_change_state" + ], + } + }, + } + return transitions + + @mark_to_call_base_class_method_with_same_name + def _prepare_state_prompts_without_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[AtRemote3.connection_hops] + serial_devname = hops_config[AtRemote3.unix_remote][AtRemote3.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 = { + AtRemote3.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[AtRemote3.connection_hops] + serial_devname = hops_config[AtRemote3.unix_remote][AtRemote3.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 = { + AtRemote3.at_remote: f"{proxy_prompt}|{at_cmds_prompt}" + } + return state_prompts + + @mark_to_call_base_class_method_with_same_name + def _prepare_newline_chars_without_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[AtRemote3.connection_hops] + hops_2_at_remote_config = hops_config[AtRemote3.unix_remote][AtRemote3.at_remote] + newline_chars = { + AtRemote3.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[AtRemote3.connection_hops] + hops_2_at_remote_config = hops_config[AtRemote3.unix_remote][AtRemote3.at_remote] + newline_chars = { + AtRemote3.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_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 = { + AtRemote3.not_connected: { + AtRemote3.unix_local_root: AtRemote3.unix_local, + AtRemote3.proxy_pc: AtRemote3.unix_local, + AtRemote3.unix_remote: AtRemote3.unix_local, + AtRemote3.unix_remote_root: AtRemote3.unix_local, + AtRemote3.at_remote: AtRemote3.unix_local, + }, + AtRemote3.unix_local: { + AtRemote3.unix_remote_root: AtRemote3.proxy_pc, + AtRemote3.at_remote: AtRemote3.proxy_pc, + }, + AtRemote3.unix_local_root: { + AtRemote3.not_connected: AtRemote3.unix_local, + AtRemote3.proxy_pc: AtRemote3.unix_local, + AtRemote3.unix_remote: AtRemote3.unix_local, + AtRemote3.unix_remote_root: AtRemote3.unix_local, + AtRemote3.at_remote: AtRemote3.unix_local, + }, + AtRemote3.unix_remote: { + AtRemote3.unix_local: AtRemote3.proxy_pc, + AtRemote3.not_connected: AtRemote3.proxy_pc, + AtRemote3.unix_local_root: AtRemote3.proxy_pc, + }, + AtRemote3.unix_remote_root: { + AtRemote3.not_connected: AtRemote3.unix_remote, + AtRemote3.proxy_pc: AtRemote3.unix_remote, + AtRemote3.unix_local: AtRemote3.unix_remote, + AtRemote3.unix_local_root: AtRemote3.unix_remote, + AtRemote3.at_remote: AtRemote3.unix_remote, + }, + AtRemote3.at_remote: { + AtRemote3.not_connected: AtRemote3.unix_remote, + AtRemote3.proxy_pc: AtRemote3.unix_remote, + AtRemote3.unix_local: AtRemote3.unix_remote, + AtRemote3.unix_local_root: AtRemote3.unix_remote, + AtRemote3.unix_remote_root: AtRemote3.unix_remote, + }, + AtRemote3.proxy_pc: { + AtRemote3.unix_remote_root: AtRemote3.unix_remote, + AtRemote3.at_remote: AtRemote3.unix_remote, + AtRemote3.not_connected: AtRemote3.unix_local, + AtRemote3.unix_local_root: AtRemote3.unix_local, + } + } + return state_hops + + def _configure_state_machine(self, sm_params): + """ + Configure device State Machine. + :param sm_params: dict with parameters of state machine for device. + :return: None. + """ + super(AtRemote3, self)._configure_state_machine(sm_params) + + # copy prompt for AT_REMOTE/ctrl_c from UNIX_REMOTE_ROOT/exit + hops_config = self._configurations[AtRemote3.connection_hops] + remote_ux_root_exit_params = hops_config[AtRemote3.unix_remote_root][AtRemote3.unix_remote]["command_params"] + remote_ux_prompt = remote_ux_root_exit_params["expected_prompt"] + hops_config[AtRemote3.at_remote][AtRemote3.unix_remote]["command_params"]["expected_prompt"] = remote_ux_prompt + + def _get_packages_for_state(self, state, observer): + """ + Get available packages containing cmds and events for each state. + :param state: device state. + :param observer: observer type, available: cmd, events + :return: available cmds or events for specific device state. + """ + available = super(AtRemote3, self)._get_packages_for_state(state, observer) + + if not available: + if state == AtRemote3.at_remote: + available = {AtRemote3.cmds: ['moler.cmd.at', 'moler.cmd.unix.ctrl_c'], + AtRemote3.events: ['moler.events.shared']} + if available: + return available[observer] + elif state == AtRemote3.unix_remote: # this is unix extended with plink_serial command + if observer == AtRemote3.cmds: + available.append('moler.cmd.at.plink_serial') + available.append('moler.cmd.at.cu') + + return available diff --git a/moler/device/juniper_ex3.py b/moler/device/juniper_ex3.py new file mode 100644 index 000000000..0aab2e814 --- /dev/null +++ b/moler/device/juniper_ex3.py @@ -0,0 +1,59 @@ +# -*- coding: utf-8 -*- +""" +JuniperEX module. +""" + +__author__ = 'Sylwester Golonka, Jakub Kupiec, Marcin Usielski' +__copyright__ = 'Copyright (C) 2019-2024, Nokia' +__email__ = 'sylwester.golonka@nokia.com, jakub.kupiec@nokia.com, marcin.usielski@nokia.com' + +from moler.device.junipergeneric3 import JuniperGeneric3 +from moler.helpers import call_base_class_method_with_same_name + + +@call_base_class_method_with_same_name +class JuniperEX3(JuniperGeneric3): + r""" + Juniperex device class. + + + :: + + + Example of device in yaml configuration file: + - with PROXY_PC: + JUNIPER_EX_PROXY_PC: + DEVICE_CLASS: moler.device.juniper_ex3.JuniperEX3 + CONNECTION_HOPS: + PROXY_PC: + CLI: + execute_command: ssh + command_params: + host: cli_host + login: cli_login + password: password + UNIX_LOCAL: + PROXY_PC: + execute_command: ssh + command_params: + expected_prompt: "proxy_pc#" + host: proxy_pc_host + login: proxy_pc_login + password: password + CLI: + PROXY_PC: + execute_command: exit + command_params: + expected_prompt: "proxy_pc#" + - without PROXY_PC: + JUNIPER_EX: + DEVICE_CLASS: moler.device.juniper_ex3.JuniperEX3 + CONNECTION_HOPS: + UNIX_LOCAL: + CLI: + execute_command: ssh # default value + command_params: + host: cli_host + login: cli_login + password: password + """ diff --git a/moler/device/junipergeneric.py b/moler/device/junipergeneric.py index 05f4a71d1..9b055b80c 100644 --- a/moler/device/junipergeneric.py +++ b/moler/device/junipergeneric.py @@ -10,13 +10,13 @@ import logging from abc import ABCMeta from six import add_metaclass -from moler.device.proxy_pc import ProxyPc +from moler.device.proxy_pc3 import ProxyPc3 from moler.helpers import call_base_class_method_with_same_name, mark_to_call_base_class_method_with_same_name @call_base_class_method_with_same_name @add_metaclass(ABCMeta) -class JuniperGeneric(ProxyPc): +class JuniperGeneric(ProxyPc3): """Junipergeneric device class.""" cli = "CLI" @@ -124,8 +124,10 @@ def _get_default_sm_configuration_without_proxy_pc(self): JuniperGeneric.unix_local: { # to "execute_command": "exit", # using command "command_params": { # with parameters - "expected_prompt": r'^moler_bash#' + "expected_prompt": r'^moler_bash#', + "target_newline": "\n", }, + "required_command_params": [], }, JuniperGeneric.configure: { "execute_command": "configure", @@ -313,6 +315,7 @@ def _prepare_state_hops_without_proxy_pc(self): JuniperGeneric.configure: JuniperGeneric.cli, }, JuniperGeneric.unix_local_root: { + JuniperGeneric.not_connected: JuniperGeneric.unix_local, JuniperGeneric.cli: JuniperGeneric.unix_local, JuniperGeneric.configure: JuniperGeneric.unix_local, }, diff --git a/moler/device/junipergeneric3.py b/moler/device/junipergeneric3.py new file mode 100644 index 000000000..c14bef336 --- /dev/null +++ b/moler/device/junipergeneric3.py @@ -0,0 +1,235 @@ +# -*- coding: utf-8 -*- +""" +Juniper Generic module. +""" + +__author__ = 'Sylwester Golonka, Jakub Kupiec, Marcin Usielski' +__copyright__ = 'Copyright (C) 2019-2024, Nokia' +__email__ = 'sylwester.golonka@nokia.com, jakub.kupiec@nokia.com, marcin.usielski@nokia.com' + +import logging +from moler.device.proxy_pc3 import ProxyPc3 +from moler.helpers import call_base_class_method_with_same_name, mark_to_call_base_class_method_with_same_name + + +# Do not create object directly. Use subclass instead - JuniperEX. +@call_base_class_method_with_same_name +class JuniperGeneric3(ProxyPc3): + """Junipergeneric device class.""" + + cli = "CLI" + configure = "CONFIGURE" + + def __init__(self, sm_params, name=None, io_connection=None, io_type=None, variant=None, + io_constructor_kwargs=None, initial_state=None, lazy_cmds_events=False): + """ + Create unix device communicating over io_connection. + + :param sm_params: params with machine state description. + :param name: name of device. + :param io_connection: External-IO connection having embedded moler-connection + :param io_type: External-IO connection connection type + :param variant: External-IO connection variant + :param io_constructor_kwargs: additional parameters for constructor of selected io_type + :param initial_state: Initial state for device + :param lazy_cmds_events: set False to load all commands and events when device is initialized, set True to load + commands and events when they are required for the first time. + """ + sm_params = sm_params.copy() + initial_state = initial_state if initial_state is not None else JuniperGeneric3.cli + super(JuniperGeneric3, self).__init__(sm_params=sm_params, name=name, io_connection=io_connection, + io_type=io_type, variant=variant, + io_constructor_kwargs=io_constructor_kwargs, + initial_state=initial_state, lazy_cmds_events=lazy_cmds_events) + self.logger = logging.getLogger('moler.juniper') + + @mark_to_call_base_class_method_with_same_name + def _get_default_sm_configuration_with_proxy_pc(self): + """ + Return State Machine default configuration with proxy_pc state. + + :return: default sm configuration with proxy_pc state. + """ + config = { + JuniperGeneric3.connection_hops: { + JuniperGeneric3.proxy_pc: { # from + JuniperGeneric3.cli: { # to + "execute_command": "ssh", # using command + "command_params": { # with parameters + "set_timeout": None, + "expected_prompt": "^admin@switch>" + }, + "required_command_params": [ + "host", + "login", + "password", + ] + } + }, + JuniperGeneric3.cli: { # from + JuniperGeneric3.proxy_pc: { # to + "execute_command": "exit", # using command + "command_params": { # with parameters + }, + "required_command_params": [ + "expected_prompt" + ] + }, + JuniperGeneric3.configure: { + "execute_command": "configure", + "command_params": { + "expected_prompt": "^admin@switch#" + } + } + }, + JuniperGeneric3.configure: { # from + JuniperGeneric3.cli: { # to + "execute_command": "exit_configure", # using command + "command_params": { # with parameters + "expected_prompt": "^admin@switch>" + } + }, + } + + } + } + return config + + @mark_to_call_base_class_method_with_same_name + def _prepare_transitions_with_proxy_pc(self): + """ + Prepare transitions to change states with proxy_pc state. + + :return: transitions with proxy_pc state. + """ + + transitions = { + JuniperGeneric3.proxy_pc: { + JuniperGeneric3.cli: { + "action": [ + "_execute_command_to_change_state" + ], + }, + }, + JuniperGeneric3.cli: { + + JuniperGeneric3.proxy_pc: { + "action": [ + "_execute_command_to_change_state" + ], + }, + JuniperGeneric3.configure: { + "action": [ + "_execute_command_to_change_state" + ], + }, + }, + JuniperGeneric3.configure: { + JuniperGeneric3.cli: { + "action": [ + "_execute_command_to_change_state" + ], + } + } + } + return transitions + + @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 with proxy_pc state. + + :return: textual prompt for each state with proxy_pc state. + """ + state_prompts = { + JuniperGeneric3.cli: + self._configurations[JuniperGeneric3.connection_hops][JuniperGeneric3.proxy_pc][ + JuniperGeneric3.cli][ + "command_params"]["expected_prompt"], + JuniperGeneric3.configure: + self._configurations[JuniperGeneric3.connection_hops][JuniperGeneric3.cli][JuniperGeneric3.configure][ + "command_params"]["expected_prompt"] + } + return state_prompts + + @mark_to_call_base_class_method_with_same_name + def _prepare_state_prompts_without_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. + """ + state_prompts = { + JuniperGeneric3.cli: + self._configurations[JuniperGeneric3.connection_hops][JuniperGeneric3.unix_local][JuniperGeneric3.cli][ + "command_params"]["expected_prompt"], + JuniperGeneric3.configure: + self._configurations[JuniperGeneric3.connection_hops][JuniperGeneric3.cli][JuniperGeneric3.configure][ + "command_params"]["expected_prompt"] + } + return state_prompts + + @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 with proxy_pc state. + + :return: non direct transitions for each state with proxy_pc state. + """ + state_hops = { + JuniperGeneric3.not_connected: { + JuniperGeneric3.cli: JuniperGeneric3.unix_local, + JuniperGeneric3.configure: JuniperGeneric3.unix_local, + }, + JuniperGeneric3.cli: { + JuniperGeneric3.not_connected: JuniperGeneric3.proxy_pc, + JuniperGeneric3.unix_local: JuniperGeneric3.proxy_pc, + JuniperGeneric3.unix_local_root: JuniperGeneric3.proxy_pc, + }, + JuniperGeneric3.configure: { + JuniperGeneric3.unix_local: JuniperGeneric3.cli, + JuniperGeneric3.proxy_pc: JuniperGeneric3.cli, + JuniperGeneric3.not_connected: JuniperGeneric3.cli, + JuniperGeneric3.unix_local_root: JuniperGeneric3.cli + }, + JuniperGeneric3.unix_local: { + JuniperGeneric3.cli: JuniperGeneric3.proxy_pc, + JuniperGeneric3.configure: JuniperGeneric3.proxy_pc, + }, + JuniperGeneric3.unix_local_root: { + JuniperGeneric3.not_connected: JuniperGeneric3.unix_local, + JuniperGeneric3.cli: JuniperGeneric3.unix_local, + JuniperGeneric3.configure: JuniperGeneric3.unix_local, + }, + JuniperGeneric3.proxy_pc: { + JuniperGeneric3.not_connected: JuniperGeneric3.unix_local, + JuniperGeneric3.configure: JuniperGeneric3.cli, + } + } + return state_hops + + def _get_packages_for_state(self, state, observer): + """ + Get available packages contain cmds and events for each state. + + :param state: device state. + :param observer: observer type, available: cmd, events + :return: available cmds or events for specific device state. + """ + available = super(JuniperGeneric3, self)._get_packages_for_state(state, observer) + + if not available: + if state == JuniperGeneric3.cli: + available = { + JuniperGeneric3.cmds: ['moler.cmd.unix', 'moler.cmd.juniper.cli', 'moler.cmd.juniper_ex.cli'], + JuniperGeneric3.events: ['moler.events.unix', 'moler.events.juniper', 'moler.events.juniper_ex']} + elif state == JuniperGeneric3.configure: + available = { + JuniperGeneric3.cmds: ['moler.events.unix', 'moler.cmd.juniper.configure', + 'moler.cmd.juniper_ex.configure'], + JuniperGeneric3.events: ['moler.events.unix', 'moler.events.juniper', 'moler.events.juniper_ex']} + + if available: + return available[observer] + + return available diff --git a/moler/device/pdu_aten.py b/moler/device/pdu_aten.py index 5ba54fd6e..ae411e5f0 100644 --- a/moler/device/pdu_aten.py +++ b/moler/device/pdu_aten.py @@ -138,7 +138,7 @@ def _get_default_sm_configuration_without_proxy_pc(self): "target_newline": "\r\n", "login": "teladmin", "password": "telpwd", - "encrypt_password": False, + "encrypt_password": True, "send_enter_after_connection": False, "cmds_before_establish_connection": ['unset binary'], }, diff --git a/moler/device/pdu_aten3.py b/moler/device/pdu_aten3.py new file mode 100644 index 000000000..a60a149bb --- /dev/null +++ b/moler/device/pdu_aten3.py @@ -0,0 +1,237 @@ +# -*- coding: utf-8 -*- +""" +PduAten device class. +""" + +__author__ = 'Marcin Usielski' +__copyright__ = 'Copyright (C) 2024, Nokia' +__email__ = 'marcin.usielski@nokia.com' + +import logging + +from moler.device.proxy_pc3 import ProxyPc3 +from moler.helpers import call_base_class_method_with_same_name, mark_to_call_base_class_method_with_same_name + + +@call_base_class_method_with_same_name +class PduAten3(ProxyPc3): + r""" + PDU Aten device class. + + + :: + + + Example of device in yaml configuration file: + - with PROXY_PC: + PDU_1: + DEVICE_CLASS: moler.device.pdu_aten.PduAten + CONNECTION_HOPS: + PROXY_PC: + PDU: + execute_command: telnet # default value + command_params: + host: 10.0.0.1 + PDU: + PROXY_PC: + execute_command: exit_telnet # default value + command_params: + expected_prompt: proxy_pc.*> + UNIX_LOCAL: + PROXY_PC: + execute_command: ssh # default value + command_params: + expected_prompt: proxy_pc.*> + host: 10.0.0.2 + login: user + password: password + -without PROXY_PC: + PDU_1: + DEVICE_CLASS: moler.device.pdu_aten.PduAten + CONNECTION_HOPS: + UNIX_LOCAL: + PDU: + execute_command: telnet # default value + command_params: + host: 10.0.0.1 + """ + + pdu = "PDU" + + def __init__(self, sm_params, name=None, io_connection=None, io_type=None, variant=None, io_constructor_kwargs=None, + initial_state=None, lazy_cmds_events=False): + """ + Create PDU device communicating over io_connection. + + :param sm_params: params with machine state description. + :param name: name of device. + :param io_connection: External-IO connection having embedded moler-connection. + :param io_type: External-IO connection type + :param variant: External-IO connection variant + :param io_constructor_kwargs: additional parameters for constructor of selected io_type + :param initial_state: Initial state for device + :param lazy_cmds_events: set False to load all commands and events when device is initialized, set True to load + commands and events when they are required for the first time. + """ + sm_params = sm_params.copy() + initial_state = initial_state if initial_state is not None else PduAten3.pdu + super(PduAten3, self).__init__(sm_params=sm_params, name=name, io_connection=io_connection, + io_type=io_type, variant=variant, io_constructor_kwargs=io_constructor_kwargs, + initial_state=initial_state, lazy_cmds_events=lazy_cmds_events) + self.logger = logging.getLogger('moler.pdu') + + @mark_to_call_base_class_method_with_same_name + def _get_default_sm_configuration_with_proxy_pc(self): + """ + Return State Machine default configuration with proxy_pc state. + :return: default sm configuration with proxy_pc state. + """ + config = { + PduAten3.connection_hops: { + PduAten3.proxy_pc: { # from + PduAten3.pdu: { # to + "execute_command": "telnet", # using command + "command_params": { # with parameters + "expected_prompt": r'^>', + "set_timeout": None, + "target_newline": "\r\n", + "login": "teladmin", + "password": "telpwd", + "encrypt_password": True, + "send_enter_after_connection": False, + "cmds_before_establish_connection": ['unset binary'], + }, + "required_command_params": [ + "host", + ] + }, + }, + PduAten3.pdu: { # from + PduAten3.proxy_pc: { # to + "execute_command": "exit_telnet", # using command + "command_params": { # with parameters + "target_newline": "\n" + }, + "required_command_params": [ + "expected_prompt" + ] + }, + }, + } + } + return config + + @mark_to_call_base_class_method_with_same_name + def _prepare_transitions_with_proxy_pc(self): + transitions = { + PduAten3.pdu: { + PduAten3.proxy_pc: { + "action": [ + "_execute_command_to_change_state" + ], + }, + }, + PduAten3.proxy_pc: { + PduAten3.pdu: { + "action": [ + "_execute_command_to_change_state" + ], + }, + }, + } + return transitions + + @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 with proxy_pc state. + :return: textual prompt for each state with proxy_pc state. + """ + state_prompts = { + PduAten3.pdu: + self._configurations[PduAten3.connection_hops][PduAten3.proxy_pc][PduAten3.pdu][ + "command_params"]["expected_prompt"], + } + return state_prompts + + @mark_to_call_base_class_method_with_same_name + def _prepare_state_prompts_without_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. + """ + state_prompts = { + PduAten3.pdu: + self._configurations[PduAten3.connection_hops][PduAten3.unix_local][PduAten3.pdu][ + "command_params"]["expected_prompt"], + } + return state_prompts + + @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 with proxy_pc state. + :return: non direct transitions for each state with proxy_pc state. + """ + state_hops = { + PduAten3.not_connected: { + PduAten3.pdu: PduAten3.unix_local, + }, + PduAten3.pdu: { + PduAten3.not_connected: PduAten3.proxy_pc, + PduAten3.unix_local: PduAten3.proxy_pc, + PduAten3.unix_local_root: PduAten3.proxy_pc + }, + PduAten3.unix_local: { + PduAten3.pdu: PduAten3.proxy_pc, + }, + PduAten3.unix_local_root: { + PduAten3.not_connected: PduAten3.unix_local, + PduAten3.pdu: PduAten3.unix_local, + }, + } + return state_hops + + def _get_packages_for_state(self, state, observer): + """ + Get available packages contain cmds and events for each state. + :param state: device state. + :param observer: observer type, available: cmd, events + :return: available cmds or events for specific device state. + """ + available = super(PduAten3, self)._get_packages_for_state(state, observer) + + if not available: + if state == PduAten3.pdu: + available = {PduAten3.cmds: ['moler.cmd.pdu_aten.pdu'], + PduAten3.events: ['moler.events.unix', 'moler.events.pdu_aten']} + if available: + return available[observer] + + return available + + @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 with proxy_pc state. + :return: newline char for each state with proxy_pc state. + """ + newline_chars = { + PduAten3.pdu: + self._configurations[PduAten3.connection_hops][PduAten3.proxy_pc][PduAten3.pdu][ + "command_params"]["target_newline"], + } + return newline_chars + + @mark_to_call_base_class_method_with_same_name + def _prepare_newline_chars_without_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. + """ + newline_chars = { + PduAten3.pdu: + self._configurations[PduAten3.connection_hops][PduAten3.unix_local][PduAten3.pdu][ + "command_params"]["target_newline"], + } + return newline_chars diff --git a/moler/device/proxy_pc3.py b/moler/device/proxy_pc3.py new file mode 100644 index 000000000..78b9d0bb4 --- /dev/null +++ b/moler/device/proxy_pc3.py @@ -0,0 +1,299 @@ +# -*- coding: utf-8 -*- +""" +Moler's device has 2 main responsibilities: +- be the factory that returns commands of that device +- be the state machine that controls which commands may run in given state +""" + +__author__ = 'Marcin Usielski' +__copyright__ = 'Copyright (C) 2024, Nokia' +__email__ = 'marcin.usielski@nokia.com' +import six +import abc +import logging + +from moler.device.unixlocal import UnixLocal +from moler.helpers import remove_state_from_sm, remove_state_hops_from_sm + + +@six.add_metaclass(abc.ABCMeta) +class ProxyPc3(UnixLocal): + proxy_pc = "PROXY_PC" + + def __init__(self, sm_params, name=None, io_connection=None, io_type=None, variant=None, io_constructor_kwargs=None, + initial_state=None, lazy_cmds_events=False): + """ + Create Unix device communicating over io_connection + :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', ... + (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) + :param initial_state: name of initial state. State machine tries to enter this state just after creation. + :param lazy_cmds_events: set False to load all commands and events when device is initialized, set True to load + commands and events when they are required for the first time. + """ + initial_state = initial_state if initial_state is not None else ProxyPc3.unix_local + self._use_proxy_pc = self._is_proxy_pc_in_sm_params(sm_params, ProxyPc3.proxy_pc) + super(ProxyPc3, self).__init__(name=name, io_connection=io_connection, + io_type=io_type, variant=variant, + io_constructor_kwargs=io_constructor_kwargs, + sm_params=sm_params, initial_state=initial_state, + lazy_cmds_events=lazy_cmds_events) + self._log(level=logging.WARNING, msg="Experimental device. May be deleted at any moment. Please don't use it in your scripts.") + + def _prepare_sm_data(self, sm_params): + self._prepare_dicts_for_sm(sm_params=sm_params) + + self._prepare_newline_chars() + self._send_transitions_to_sm(self._stored_transitions) + + def _prepare_transitions(self): + """ + Prepare transitions to change states. + :return: None. + """ + + stored_is_proxy_pc = self._use_proxy_pc + self._use_proxy_pc = True + super(ProxyPc3, self)._prepare_transitions() + self._use_proxy_pc = stored_is_proxy_pc + transitions = self._prepare_transitions_with_proxy_pc() + self._add_transitions(transitions=transitions) + + def _prepare_dicts_for_sm(self, sm_params): + """ + Prepare transitions to change states. + :return: None. + """ + + self._prepare_transitions() + transitions = self._stored_transitions + state_hops = self._prepare_state_hops_with_proxy_pc() + + default_sm_configurations = self._get_default_sm_configuration() + + if not self._use_proxy_pc: + (connection_hops, transitions) = remove_state_from_sm( + source_sm=default_sm_configurations[ProxyPc3.connection_hops], + source_transitions=transitions, + state_to_remove=ProxyPc3.proxy_pc, + ) + state_hops = remove_state_hops_from_sm( + source_hops=state_hops, state_to_remove=ProxyPc3.proxy_pc + ) + default_sm_configurations[ProxyPc3.connection_hops] = connection_hops + + self._stored_transitions = transitions + self._update_dict(self._state_hops, state_hops) + + self._configurations = self._prepare_sm_configuration( + default_sm_configurations, sm_params + ) + self._overwrite_prompts() + self._validate_device_configuration() + self._prepare_state_prompts() + + def _overwrite_prompts(self) -> None: + """ + Method to overwrite prompts in commands. + :return: None + """ + + def _get_default_sm_configuration(self): + """ + Create State Machine default configuration. + :return: default sm configuration. + """ + config = super(ProxyPc3, self)._get_default_sm_configuration() + default_config = self._get_default_sm_configuration_with_proxy_pc() + + self._update_dict(config, default_config) + return config + + def _get_default_sm_configuration_with_proxy_pc(self): + """ + Return State Machine default configuration with proxy_pc state. + :return: default sm configuration with proxy_pc state. + """ + config = { + ProxyPc3.connection_hops: { + ProxyPc3.unix_local: { # from + ProxyPc3.proxy_pc: { # to + "execute_command": "ssh", # using command + "command_params": { # with parameters + "target_newline": "\n" + }, + "required_command_params": [ + "host", + "login", + "password", + "expected_prompt" + ] + }, + }, + ProxyPc3.proxy_pc: { # from + ProxyPc3.unix_local: { # to + "execute_command": "exit", # using command + "command_params": { # with parameters + "target_newline": "\n", + "expected_prompt": r'^moler_bash#', + }, + "required_command_params": [ + ] + } + }, + } + } + return config + + def _prepare_transitions(self): + """ + Prepare transitions to change states. + :return: None. + """ + + stored_is_proxy_pc = self._use_proxy_pc + self._use_proxy_pc = True + super(ProxyPc3, self)._prepare_transitions() + self._use_proxy_pc = stored_is_proxy_pc + transitions = self._prepare_transitions_with_proxy_pc() + self._add_transitions(transitions=transitions) + + def _prepare_transitions_with_proxy_pc(self): + """ + Prepare transitions to change states with proxy_pc state. + :return: transitions with proxy_pc state. + """ + transitions = { + ProxyPc3.unix_local: { + ProxyPc3.proxy_pc: { + "action": [ + "_execute_command_to_change_state" + ], + } + }, + ProxyPc3.proxy_pc: { + ProxyPc3.unix_local: { + "action": [ + "_execute_command_to_change_state" + ], + }, + }, + } + return transitions + + def _prepare_state_prompts(self): + """ + Prepare textual prompt for each state. + :return: None. + """ + super(ProxyPc3, self)._prepare_state_prompts() + + if self._use_proxy_pc: + state_prompts = self._prepare_state_prompts_with_proxy_pc() + else: + state_prompts = self._prepare_state_prompts_without_proxy_pc() + + self._update_dict(self._state_prompts, state_prompts) + + def _prepare_state_prompts_with_proxy_pc(self): + """ + Prepare textual prompt for each state for State Machine with proxy_pc state. + :return: textual prompt for each state with proxy_pc state. + """ + state_prompts = { + ProxyPc3.proxy_pc: + self._configurations[ProxyPc3.connection_hops][ProxyPc3.unix_local][ProxyPc3.proxy_pc][ + "command_params"]["expected_prompt"], + ProxyPc3.unix_local: + self._configurations[ProxyPc3.connection_hops][ProxyPc3.proxy_pc][ProxyPc3.unix_local][ + "command_params"]["expected_prompt"], + } + return state_prompts + + def _prepare_state_prompts_without_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. + """ + state_prompts = {} + return state_prompts + + def _prepare_newline_chars(self): + """ + Prepare newline char for each state. + :return: None. + """ + super(ProxyPc3, self)._prepare_newline_chars() + + if self._use_proxy_pc: + newline_chars = self._prepare_newline_chars_with_proxy_pc() + else: + newline_chars = self._prepare_newline_chars_without_proxy_pc() + + self._update_dict(self._newline_chars, newline_chars) + + def _prepare_newline_chars_with_proxy_pc(self): + """ + Prepare newline char for each state for State Machine with proxy_pc state. + :return: newline char for each state with proxy_pc state. + """ + newline_chars = { + ProxyPc3.proxy_pc: + self._configurations[ProxyPc3.connection_hops][ProxyPc3.unix_local][ProxyPc3.proxy_pc][ + "command_params"]["target_newline"], + ProxyPc3.unix_local: + self._configurations[ProxyPc3.connection_hops][ProxyPc3.proxy_pc][ProxyPc3.unix_local][ + "command_params"]["target_newline"], + } + return newline_chars + + def _prepare_newline_chars_without_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. + """ + newline_chars = {} + return newline_chars + + def _prepare_state_hops_with_proxy_pc(self): + """ + Prepare non direct transitions for each state for State Machine with proxy_pc state. + :return: non direct transitions for each state with proxy_pc state. + """ + state_hops = { + UnixLocal.not_connected: { + ProxyPc3.proxy_pc: ProxyPc3.unix_local, + }, + UnixLocal.unix_local_root: { + ProxyPc3.proxy_pc: ProxyPc3.unix_local, + ProxyPc3.not_connected: ProxyPc3.unix_local, + }, + ProxyPc3.proxy_pc: { + ProxyPc3.not_connected: ProxyPc3.unix_local, + ProxyPc3.unix_local_root: ProxyPc3.unix_local, + }, + } + return state_hops + + def _get_packages_for_state(self, state, observer): + """ + Get available packages contain cmds and events for each state. + :param state: device state. + :param observer: observer type, available: cmd, events + :return: available cmds or events for specific device state. + """ + available = super(ProxyPc3, self)._get_packages_for_state(state, observer) + + if not available: + if state == ProxyPc3.proxy_pc: + available = {UnixLocal.cmds: ['moler.cmd.unix'], + UnixLocal.events: ['moler.events.shared', 'moler.events.unix']} + if available: + return available[observer] + + return available diff --git a/moler/device/scpi.py b/moler/device/scpi.py index 6e4505c5b..38a07ff20 100644 --- a/moler/device/scpi.py +++ b/moler/device/scpi.py @@ -267,6 +267,7 @@ def _prepare_state_hops_without_proxy_pc(self): state_hops = { Scpi.not_connected: { Scpi.scpi: Scpi.unix_local, + Scpi.unix_local_root: Scpi.unix_local, }, Scpi.scpi: { Scpi.not_connected: Scpi.unix_local, diff --git a/moler/device/scpi3.py b/moler/device/scpi3.py new file mode 100644 index 000000000..145e4a4a3 --- /dev/null +++ b/moler/device/scpi3.py @@ -0,0 +1,242 @@ +# -*- coding: utf-8 -*- +""" +SCPI device class +""" + +__author__ = 'Marcin Usielski, Marcin Szlapa' +__copyright__ = 'Copyright (C) 2019-2024, Nokia' +__email__ = 'marcin.usielski@nokia.com, marcin.szlapa@nokia.com' + +import logging + +from moler.device.proxy_pc3 import ProxyPc3 +from moler.helpers import call_base_class_method_with_same_name, mark_to_call_base_class_method_with_same_name + + +@call_base_class_method_with_same_name +class Scpi3(ProxyPc3): + r""" + Scpi device class. + + + :: + + + Example of device in yaml configuration file: + - with PROXY_PC: + SCPI_1: + DEVICE_CLASS: moler.device.scpi.Scpi + CONNECTION_HOPS: + PROXY_PC: + SCPI: + execute_command: telnet # default value + command_params: + expected_prompt: SCPI> + host: 10.0.0.1 + port: 99999 + SCPI: + PROXY_PC: + execute_command: exit_telnet # default value + command_params: + expected_prompt: proxy_pc.*> + UNIX_LOCAL: + PROXY_PC: + execute_command: ssh # default value + command_params: + expected_prompt: proxy_pc.*> + host: 10.0.0.2 + login: user + password: password + -without PROXY_PC: + SCPI_1: + DEVICE_CLASS: moler.device.scpi.Scpi + CONNECTION_HOPS: + UNIX_LOCAL: + SCPI: + execute_command: telnet # default value + command_params: + expected_prompt: SCPI> + host: 10.0.0.1 + port: 99999 + """ + + scpi = "SCPI" + + def __init__(self, sm_params, name=None, io_connection=None, io_type=None, variant=None, + io_constructor_kwargs=None, initial_state=None, lazy_cmds_events=False): + """ + Create SCPI device communicating over io_connection. + + :param sm_params: params with machine state description. + :param name: name of device. + :param io_connection: External-IO connection having embedded moler-connection. + :param io_type: External-IO connection type + :param variant: External-IO connection variant + :param io_constructor_kwargs: additional parameters for constructor of selected io_type + :param initial_state: Initial state for device + :param lazy_cmds_events: set False to load all commands and events when device is initialized, set True to load + commands and events when they are required for the first time. + """ + sm_params = sm_params.copy() + initial_state = initial_state if initial_state is not None else Scpi3.scpi + super(Scpi3, self).__init__(sm_params=sm_params, name=name, io_connection=io_connection, io_type=io_type, + variant=variant, io_constructor_kwargs=io_constructor_kwargs, + initial_state=initial_state, lazy_cmds_events=lazy_cmds_events) + self.logger = logging.getLogger('moler.scpi') + + @mark_to_call_base_class_method_with_same_name + def _get_default_sm_configuration_with_proxy_pc(self): + """ + Return State Machine default configuration with proxy_pc state. + :return: default sm configuration with proxy_pc state. + """ + config = { + Scpi3.connection_hops: { + Scpi3.proxy_pc: { # from + Scpi3.scpi: { # to + "execute_command": "telnet", # using command + "command_params": { # with parameters + "expected_prompt": r'SCPI>', + "set_timeout": None, + "target_newline": "\r\n", + }, + "required_command_params": [ + "host", + "port", + ] + }, + }, + Scpi3.scpi: { # from + Scpi3.proxy_pc: { # to + "execute_command": "exit_telnet", # using command + "command_params": { # with parameters + "target_newline": "\n" + }, + "required_command_params": [ + "expected_prompt" + ] + }, + }, + } + } + return config + + @mark_to_call_base_class_method_with_same_name + def _prepare_transitions_with_proxy_pc(self): + transitions = { + Scpi3.scpi: { + Scpi3.proxy_pc: { + "action": [ + "_execute_command_to_change_state" + ], + }, + }, + Scpi3.proxy_pc: { + Scpi3.scpi: { + "action": [ + "_execute_command_to_change_state" + ], + }, + }, + } + return transitions + + @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 with proxy_pc state. + :return: textual prompt for each state with proxy_pc state. + """ + state_prompts = { + Scpi3.scpi: + self._configurations[Scpi3.connection_hops][Scpi3.proxy_pc][Scpi3.scpi][ + "command_params"]["expected_prompt"], + } + return state_prompts + + @mark_to_call_base_class_method_with_same_name + def _prepare_state_prompts_without_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. + """ + state_prompts = { + Scpi3.scpi: + self._configurations[Scpi3.connection_hops][Scpi3.unix_local][Scpi3.scpi][ + "command_params"]["expected_prompt"], + } + return state_prompts + + @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 with proxy_pc state. + :return: non direct transitions for each state with proxy_pc state. + """ + state_hops = { + Scpi3.not_connected: { + Scpi3.scpi: Scpi3.unix_local, + Scpi3.proxy_pc: Scpi3.unix_local, + Scpi3.unix_local_root: Scpi3.unix_local, + }, + Scpi3.scpi: { + Scpi3.not_connected: Scpi3.proxy_pc, + Scpi3.unix_local: Scpi3.proxy_pc, + Scpi3.unix_local_root: Scpi3.proxy_pc + }, + Scpi3.unix_local: { + Scpi3.scpi: Scpi3.proxy_pc, + }, + Scpi3.unix_local_root: { + Scpi3.scpi: Scpi3.unix_local, + Scpi3.not_connected: Scpi3.unix_local, + }, + Scpi3.proxy_pc: { + Scpi3.not_connected: Scpi3.unix_local, + } + } + return state_hops + + def _get_packages_for_state(self, state, observer): + """ + Get available packages contain cmds and events for each state. + :param state: device state. + :param observer: observer type, available: cmd, events + :return: available cmds or events for specific device state. + """ + available = super(Scpi3, self)._get_packages_for_state(state, observer) + + if not available: + if state == Scpi3.scpi: + available = {Scpi3.cmds: ['moler.cmd.scpi.scpi'], + Scpi3.events: ['moler.events.unix', 'moler.events.scpi']} + if available: + return available[observer] + + return available + + @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 with proxy_pc state. + :return: newline char for each state with proxy_pc state. + """ + newline_chars = { + Scpi3.scpi: + self._configurations[Scpi3.connection_hops][Scpi3.proxy_pc][Scpi3.scpi][ + "command_params"]["target_newline"], + } + return newline_chars + + @mark_to_call_base_class_method_with_same_name + def _prepare_newline_chars_without_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. + """ + newline_chars = { + Scpi3.scpi: + self._configurations[Scpi3.connection_hops][Scpi3.unix_local][Scpi3.scpi][ + "command_params"]["target_newline"], + } + return newline_chars diff --git a/moler/device/textualdevice.py b/moler/device/textualdevice.py index 242446132..575456f15 100644 --- a/moler/device/textualdevice.py +++ b/moler/device/textualdevice.py @@ -159,6 +159,7 @@ def _prepare_sm_data(self, sm_params): self._prepare_transitions() self._prepare_state_hops() self._configure_state_machine(sm_params) + self._validate_device_configuration() self._prepare_newline_chars() self._send_transitions_to_sm(self._stored_transitions) @@ -1153,7 +1154,6 @@ def _configure_state_machine(self, sm_params): self._configurations = self._prepare_sm_configuration( default_sm_configurations, sm_params ) - self._validate_device_configuration() self._prepare_state_prompts() def _prepare_sm_configuration(self, default_sm_configurations, sm_params): diff --git a/moler/device/unixremote.py b/moler/device/unixremote.py index f729888d2..a4ba40ef8 100644 --- a/moler/device/unixremote.py +++ b/moler/device/unixremote.py @@ -435,6 +435,10 @@ def _overwrite_prompts(self): Overwrite prompts for some states to easily configure the SM. """ if self._use_proxy_pc: + self._configurations[UnixRemote.connection_hops][UnixRemote.unix_remote][UnixRemote.proxy_pc][ + "command_params"]["expected_prompt"] = \ + self._configurations[UnixRemote.connection_hops][UnixRemote.unix_local][UnixRemote.proxy_pc][ + "command_params"]["expected_prompt"] self._configurations[UnixRemote.connection_hops][UnixRemote.unix_remote_root][UnixRemote.unix_remote][ "command_params"]["expected_prompt"] = \ self._configurations[UnixRemote.connection_hops][UnixRemote.proxy_pc][UnixRemote.unix_remote][ diff --git a/moler/device/unixremote3.py b/moler/device/unixremote3.py index 76018e76b..239971277 100644 --- a/moler/device/unixremote3.py +++ b/moler/device/unixremote3.py @@ -11,16 +11,15 @@ import logging -from moler.device.proxy_pc import ProxyPc +from moler.device.proxy_pc3 import ProxyPc3 from moler.helpers import ( call_base_class_method_with_same_name, mark_to_call_base_class_method_with_same_name, - remove_state_from_sm, remove_state_hops_from_sm ) @call_base_class_method_with_same_name -class UnixRemote3(ProxyPc): +class UnixRemote3(ProxyPc3): r""" UnixRemote3 device class. @@ -113,74 +112,17 @@ def __init__( ) self._log(level=logging.WARNING, msg="Experimental device. May be deleted at any moment. Please don't use it in your scripts.") - def _prepare_sm_data(self, sm_params): - self._prepare_dicts_for_sm(sm_params=sm_params) - - self._prepare_newline_chars() - self._send_transitions_to_sm(self._stored_transitions) - - def _prepare_transitions(self): - """ - Prepare transitions to change states. - :return: None. - """ - - stored_is_proxy_pc = self._use_proxy_pc - self._use_proxy_pc = True - super(UnixRemote3, self)._prepare_transitions() - self._use_proxy_pc = stored_is_proxy_pc - transitions = self._prepare_transitions_with_proxy_pc() - self._add_transitions(transitions=transitions) - - def _prepare_dicts_for_sm(self, sm_params): - """ - Prepare transitions to change states. - :return: None. - """ - - self._prepare_transitions() - transitions = self._stored_transitions - state_hops = self._prepare_state_hops_with_proxy_pc() - - default_sm_configurations = self._get_default_sm_configuration() - - if not self._use_proxy_pc: - (connection_hops, transitions) = remove_state_from_sm( - source_sm=default_sm_configurations[UnixRemote3.connection_hops], - source_transitions=transitions, - state_to_remove=UnixRemote3.proxy_pc, - ) - state_hops = remove_state_hops_from_sm( - source_hops=state_hops, state_to_remove=UnixRemote3.proxy_pc - ) - default_sm_configurations[UnixRemote3.connection_hops] = connection_hops - - self._stored_transitions = transitions - self._update_dict(self._state_hops, state_hops) - - self._configurations = self._prepare_sm_configuration( - default_sm_configurations, sm_params - ) - self._overwrite_prompts() - self._validate_device_configuration() - self._prepare_state_prompts() - - def _get_default_sm_configuration(self): - """ - Create State Machine default configuration. - :return: default sm configuration. - """ - config = super(ProxyPc, self)._get_default_sm_configuration() - default_config = self._get_default_sm_configuration_with_proxy_pc() - - self._update_dict(config, default_config) - return config - def _overwrite_prompts(self): """ Overwrite prompts for some states to easily configure the SM. """ + super(UnixRemote3, self)._overwrite_prompts() if self._use_proxy_pc: + self._configurations[UnixRemote3.connection_hops][UnixRemote3.unix_remote][UnixRemote3.proxy_pc][ + "command_params"]["expected_prompt"] = \ + self._configurations[UnixRemote3.connection_hops][UnixRemote3.unix_local][UnixRemote3.proxy_pc][ + "command_params"]["expected_prompt"] + self._configurations[UnixRemote3.connection_hops][UnixRemote3.unix_remote_root][UnixRemote3.unix_remote][ "command_params"]["expected_prompt"] = \ self._configurations[UnixRemote3.connection_hops][UnixRemote3.proxy_pc][UnixRemote3.unix_remote][ diff --git a/moler/helpers.py b/moler/helpers.py index a5d0e53c3..beeddc75f 100644 --- a/moler/helpers.py +++ b/moler/helpers.py @@ -19,6 +19,7 @@ from types import FunctionType, MethodType from six import string_types, integer_types +from moler.exceptions import MolerException class ClassProperty(property): @@ -603,7 +604,11 @@ def remove_state_from_sm(source_sm: dict, source_transitions: dict, state_to_rem new_sm[new_from] = {} if new_from not in new_transitions: new_transitions[new_from] = {} + new_sm[new_from][to_state] = copy.deepcopy(source_sm[state_to_remove][to_state]) + if 'execute_command' in source_sm[new_from][state_to_remove]: + new_sm[new_from][to_state]['execute_command'] = source_sm[new_from][state_to_remove]['execute_command'] + if state_to_remove in source_transitions and to_state in source_transitions[state_to_remove]: new_transitions[new_from][to_state] = copy.deepcopy(source_transitions[state_to_remove][to_state]) else: @@ -615,6 +620,8 @@ def remove_state_from_sm(source_sm: dict, source_transitions: dict, state_to_rem _delete_state(sm=new_sm, state_to_remove=state_to_remove) _delete_state(sm=new_transitions, state_to_remove=state_to_remove) + _delete_empty_states(new_sm) + _delete_empty_states(new_transitions) return (new_sm, new_transitions) @@ -633,6 +640,18 @@ def _delete_state(sm: dict, state_to_remove: str) -> None: del sm[from_state][state_to_remove] +def _delete_empty_states(sm: dict) -> None: + """ + Delete empty states from a state machine dict (in place). + :param sm: dict with state machine + :return: None + """ + states = list(sm.keys()) + for state in states: + if sm[state] is None or not sm[state]: + del sm[state] + + def remove_state_hops_from_sm(source_hops: dict, state_to_remove: str) -> dict: """ Remove a state from a state machine dict. @@ -648,6 +667,9 @@ def remove_state_hops_from_sm(source_hops: dict, state_to_remove: str) -> dict: direct_state = item[dest_state] if direct_state == state_to_remove: if state_to_remove in source_hops and dest_state in source_hops[state_to_remove]: + if source_hops[state_to_remove][dest_state] == from_state: + msg = f"Found cycle from '{from_state}' to '{dest_state}' via '{source_hops[state_to_remove][dest_state]}'. Please verify state hops: {source_hops}" + raise MolerException(msg) new_hops[from_state][dest_state] = source_hops[state_to_remove][dest_state] else: del new_hops[from_state][dest_state] @@ -659,4 +681,5 @@ def remove_state_hops_from_sm(source_hops: dict, state_to_remove: str) -> dict: if state_to_remove in new_hops: del new_hops[state_to_remove] + _delete_empty_states(new_hops) return new_hops diff --git a/test/device/test_SM_adb_remote.py b/test/device/test_SM_adb_remote.py index a9cc7d43c..a7bba5b69 100644 --- a/test/device/test_SM_adb_remote.py +++ b/test/device/test_SM_adb_remote.py @@ -1,19 +1,48 @@ -__author__ = 'Grzegorz Latuszek' -__copyright__ = 'Copyright (C) 2020-2022, Nokia' +__author__ = 'Grzegorz Latuszek, Marcin Usielski' +__copyright__ = 'Copyright (C) 2020-2024, Nokia' __email__ = 'grzegorz.latuszek@nokia.com, marcin.usielski@nokia.com' import pytest from moler.util.devices_SM import iterate_over_device_states, get_device +from moler.device import DeviceFactory -def test_adb_remote_device(device_connection, adb_remote_output): - adb_remote = get_device(name="ADB_REMOTE", connection=device_connection, device_output=adb_remote_output, +adb_remotes = ['ADB_REMOTE', 'ADB_REMOTE3'] +adb_remotes_proxy_pc = ['ADB_REMOTE_PROXY_PC', 'ADB_REMOTE_PROXY_PC3'] + +@pytest.mark.parametrize("device_name", adb_remotes) +def test_adb_remote_device(device_name, device_connection, adb_remote_output): + adb_remote = get_device(name=device_name, connection=device_connection, device_output=adb_remote_output, + test_file_path=__file__) + + iterate_over_device_states(device=adb_remote) + + +@pytest.mark.parametrize("device_name", adb_remotes_proxy_pc) +def test_adb_remote_device_proxy_pc(device_name, device_connection, adb_remote_output_proxy_pc): + adb_remote = get_device(name=device_name, connection=device_connection, device_output=adb_remote_output_proxy_pc, test_file_path=__file__) iterate_over_device_states(device=adb_remote) +@pytest.mark.parametrize("devices", [adb_remotes_proxy_pc, adb_remotes]) +def test_unix_sm_identity(devices, device_connection, adb_remote_output): + + dev0 = get_device(name=devices[0], connection=device_connection, device_output=adb_remote_output, + test_file_path=__file__) + dev1 = get_device(name=devices[1], connection=device_connection, device_output=adb_remote_output, + test_file_path=__file__) + + assert dev0._state_hops == dev1._state_hops + assert dev0._configurations == dev1._configurations + + assert dev0._stored_transitions == dev1._stored_transitions + assert dev0._state_prompts == dev1._state_prompts + assert dev0._newline_chars == dev1._newline_chars + + @pytest.fixture def adb_remote_output(): output = { @@ -45,3 +74,40 @@ def adb_remote_output(): } return output + + +@pytest.fixture +def adb_remote_output_proxy_pc(): + 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' + }, + "UNIX_LOCAL_ROOT": { + 'exit': 'moler_bash#' + }, + "UNIX_REMOTE": { + 'exit': 'proxy_pc#', + 'su': 'remote_root_prompt', + 'adb -s f57e6b77 shell': 'shell@adbhost:/ $', # adb shell is changing prompt so it triggers following 2 send-responses + '': 'shell@adbhost:/ $', # to allow for self.connection.sendline("") in _send_prompt_set() + 'export PS1="adb_shell@f57e6b77 \\$ "': 'adb_shell@f57e6b77 $' # to allow for self.connection.sendline(self.set_prompt) in _send_prompt_set() + }, + "ADB_SHELL": { + '': 'adb_shell@f57e6b77 $', + 'exit': 'remote#', + 'su': 'adb_shell@f57e6b77 #', + }, + "ADB_SHELL_ROOT": { + 'exit': 'adb_shell@f57e6b77 $', + }, + "UNIX_REMOTE_ROOT": { + 'exit': 'remote#', + }, + "PROXY_PC": { + 'TERM=xterm-mono ssh -l remote_login -o ServerAliveInterval=7 -o ServerAliveCountMax=2 remote_host': 'remote#', + 'exit': 'moler_bash#' + }, + } + + return output diff --git a/test/device/test_SM_at_remote.py b/test/device/test_SM_at_remote.py index bdf60543c..6cbe4fa36 100644 --- a/test/device/test_SM_at_remote.py +++ b/test/device/test_SM_at_remote.py @@ -1,19 +1,46 @@ -__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 from moler.util.devices_SM import iterate_over_device_states, get_device +at_remotes = ["AT_REMOTE", "AT_REMOTE3"] +at_remotes_proxy_pc = ["AT_REMOTE_PROXY_PC", "AT_REMOTE_PROXY_PC3"] -def test_at_remote_device(device_connection, at_remote_output): - at_remote = get_device(name="AT_REMOTE", connection=device_connection, device_output=at_remote_output, + +@pytest.mark.parametrize("device_name", at_remotes) +def test_at_remote_device(device_name, device_connection, at_remote_output): + at_remote = get_device(name=device_name, connection=device_connection, device_output=at_remote_output, + test_file_path=__file__) + + iterate_over_device_states(device=at_remote) + + +@pytest.mark.parametrize("device_name", at_remotes_proxy_pc) +def test_at_remote_device_proxy_pc(device_name, device_connection, at_remote_output_proxy_pc): + at_remote = get_device(name=device_name, connection=device_connection, device_output=at_remote_output_proxy_pc, test_file_path=__file__) iterate_over_device_states(device=at_remote) +@pytest.mark.parametrize("devices", [at_remotes_proxy_pc, at_remotes]) +def test_unix_sm_identity(devices, device_connection, at_remote_output): + + dev0 = get_device(name=devices[0], connection=device_connection, device_output=at_remote_output, + test_file_path=__file__) + dev1 = get_device(name=devices[1], connection=device_connection, device_output=at_remote_output, + test_file_path=__file__) + + assert dev0._state_hops == dev1._state_hops + assert dev0._configurations == dev1._configurations + + assert dev0._stored_transitions == dev1._stored_transitions + assert dev0._state_prompts == dev1._state_prompts + assert dev0._newline_chars == dev1._newline_chars + @pytest.fixture def at_remote_output(): plink_cmd_string = 'plink -serial COM5 |& awk \'BEGIN {print "COM5> port READY"} {print} END {print "^C"}\'' @@ -39,3 +66,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/device/test_SM_juniper_ex.py b/test/device/test_SM_juniper_ex.py index 0a3ea050f..3cbb9b470 100644 --- a/test/device/test_SM_juniper_ex.py +++ b/test/device/test_SM_juniper_ex.py @@ -1,23 +1,41 @@ -__author__ = 'Michal Ernst' -__copyright__ = 'Copyright (C) 2018-2019, Nokia' -__email__ = 'michal.ernst@nokia.com' +__author__ = 'Michal Ernst, Marcin Usielski' +__copyright__ = 'Copyright (C) 2018-2024, Nokia' +__email__ = 'michal.ernst@nokia.com, marcin.usielski@nokia.com' import pytest +from moler.device import DeviceFactory from moler.util.devices_SM import iterate_over_device_states, get_device -def test_juniper_ex_device(device_connection, juniper_ex_output): - juniper_ex = get_device(name="JUNIPER_EX", connection=device_connection, device_output=juniper_ex_output, +junipers = ['JUNIPER_EX', 'JUNIPER_EX3'] +junipers_proxy = ['JUNIPER_EX_PROXY_PC', 'JUNIPER_EX_PROXY_PC3'] + +@pytest.mark.parametrize("device_name", junipers) +def test_juniper_ex_device(device_name, device_connection, juniper_ex_output): + juniper_ex = get_device(name=device_name, connection=device_connection, device_output=juniper_ex_output, test_file_path=__file__) iterate_over_device_states(device=juniper_ex) -def test_juniper_ex_proxy_pc_device(device_connection, juniper_ex_proxy_pc_output): - juniper_ex_proxy_pc = get_device(name="JUNIPER_EX_PROXY_PC", connection=device_connection, +@pytest.mark.parametrize("device_name", junipers_proxy) +def test_juniper_ex_proxy_pc_device(device_name, device_connection, juniper_ex_proxy_pc_output): + juniper_ex_proxy_pc = get_device(name=device_name, connection=device_connection, device_output=juniper_ex_proxy_pc_output, test_file_path=__file__) iterate_over_device_states(device=juniper_ex_proxy_pc) +@pytest.mark.parametrize("devices", [junipers_proxy, junipers]) +def test_unix_sm_identity(devices): + dev0 = DeviceFactory.get_device(name=devices[0]) + dev1 = DeviceFactory.get_device(name=devices[1]) + + assert dev0._stored_transitions == dev1._stored_transitions + assert dev0._state_hops == dev1._state_hops + assert dev0._state_prompts == dev1._state_prompts + assert dev0._configurations == dev1._configurations + assert dev0._newline_chars == dev1._newline_chars + + @pytest.fixture def juniper_ex_output(): output = { diff --git a/test/device/test_SM_pdu_aten.py b/test/device/test_SM_pdu_aten.py index 97e8503e3..29f400e15 100644 --- a/test/device/test_SM_pdu_aten.py +++ b/test/device/test_SM_pdu_aten.py @@ -1,5 +1,5 @@ __author__ = 'Marcin Usielski' -__copyright__ = 'Copyright (C) 2020, Nokia' +__copyright__ = 'Copyright (C) 2020-2024, Nokia' __email__ = 'marcin.usielski@nokia.com' import pytest @@ -7,20 +7,39 @@ from moler.util.devices_SM import iterate_over_device_states, get_device -def test_pdu_device(device_connection, pdu_output): - pdu = get_device(name="PDU", connection=device_connection, device_output=pdu_output, +pdus = ["PDU", "PDU3"] +pdus_proxy = ["PDU_PROXY_PC", "PDU_PROXY_PC3"] + + +@pytest.mark.parametrize("device_name", pdus) +def test_pdu_device(device_name, device_connection, pdu_output): + pdu = get_device(name=device_name, connection=device_connection, device_output=pdu_output, test_file_path=__file__) iterate_over_device_states(device=pdu) -def test_pdu_proxy_pc_device(device_connection, pdu_proxy_pc_output): - pdu = get_device(name="PDU_PROXY_PC", connection=device_connection, +@pytest.mark.parametrize("device_name", pdus_proxy) +def test_pdu_proxy_pc_device(device_name, device_connection, pdu_proxy_pc_output): + pdu = get_device(name=device_name, connection=device_connection, device_output=pdu_proxy_pc_output, test_file_path=__file__) iterate_over_device_states(device=pdu) +@pytest.mark.parametrize("devices", [pdus, pdus_proxy]) +def test_unix_sm_identity(devices): + from moler.device import DeviceFactory + dev0 = DeviceFactory.get_device(name=devices[0]) + dev1 = DeviceFactory.get_device(name=devices[1]) + + assert dev0._stored_transitions == dev1._stored_transitions + assert dev0._state_hops == dev1._state_hops + assert dev0._state_prompts == dev1._state_prompts + assert dev0._configurations == dev1._configurations + assert dev0._newline_chars == dev1._newline_chars + + @pytest.fixture def pdu_output(): output = { diff --git a/test/device/test_SM_scpi.py b/test/device/test_SM_scpi.py index cbb1d79ce..a1aa3f239 100644 --- a/test/device/test_SM_scpi.py +++ b/test/device/test_SM_scpi.py @@ -1,26 +1,44 @@ -__author__ = 'Marcin Szlapa' -__copyright__ = 'Copyright (C) 2019, Nokia' -__email__ = 'marcin.szlapa@nokia.com' +__author__ = 'Marcin Szlapa, Marcin Usielski' +__copyright__ = 'Copyright (C) 2019-2024, Nokia' +__email__ = 'marcin.szlapa@nokia.com, marcin.usielski@nokia.com' import pytest from moler.util.devices_SM import iterate_over_device_states, get_device +scpis = ["SCPI", "SCPI3"] +scpis_proxy = ["SCPI_PROXY_PC", "SCPI_PROXY_PC3"] -def test_scpi_device(device_connection, scpi_output): - scpi = get_device(name="SCPI", connection=device_connection, device_output=scpi_output, + +@pytest.mark.parametrize("device_name", scpis) +def test_scpi_device(device_name, device_connection, scpi_output): + scpi = get_device(name=device_name, connection=device_connection, device_output=scpi_output, test_file_path=__file__) iterate_over_device_states(device=scpi) -def test_scpi_proxy_pc_device(device_connection, scpi_proxy_pc_output): - scpi_proxy_pc = get_device(name="SCPI_PROXY_PC", connection=device_connection, +@pytest.mark.parametrize("device_name", scpis_proxy) +def test_scpi_proxy_pc_device(device_name, device_connection, scpi_proxy_pc_output): + scpi_proxy_pc = get_device(name=device_name, connection=device_connection, device_output=scpi_proxy_pc_output, test_file_path=__file__) iterate_over_device_states(device=scpi_proxy_pc) +@pytest.mark.parametrize("devices", [scpis, scpis_proxy]) +def test_unix_sm_identity(devices): + from moler.device import DeviceFactory + dev0 = DeviceFactory.get_device(name=devices[0]) + dev1 = DeviceFactory.get_device(name=devices[1]) + + assert dev0._stored_transitions == dev1._stored_transitions + assert dev0._state_hops == dev1._state_hops + assert dev0._state_prompts == dev1._state_prompts + assert dev0._configurations == dev1._configurations + assert dev0._newline_chars == dev1._newline_chars + + @pytest.fixture def scpi_output(): output = { diff --git a/test/device/test_SM_unix_remote3.py b/test/device/test_SM_unix_remote3.py deleted file mode 100644 index ec0d84c61..000000000 --- a/test/device/test_SM_unix_remote3.py +++ /dev/null @@ -1,195 +0,0 @@ -__author__ = 'Marcin Usielski' -__copyright__ = 'Copyright (C) 2024, Nokia' -__email__ = 'marcin.usielski@nokia.com' - -import pytest -import time -import os -from moler.util.devices_SM import iterate_over_device_states, get_device -from moler.exceptions import MolerException, DeviceChangeStateFailure -from moler.helpers import copy_dict -from moler.util.moler_test import MolerTest -from moler.device import DeviceFactory -from moler.config import load_config -from moler.exceptions import DeviceFailure - - -def test_unix_remote_device(device_connection, unix_remote_output3): - unix_remote = get_device(name="UNIX_REMOTE3", connection=device_connection, device_output=unix_remote_output3, - test_file_path=__file__) - iterate_over_device_states(device=unix_remote) - assert None is not unix_remote._cmdnames_available_in_state['UNIX_LOCAL_ROOT'] - - -def test_unix_remote_proxy_pc_device(device_connection, unix_remote_proxy_pc_output): - unix_remote_proxy_pc = get_device(name="UNIX_REMOTE3_PROXY_PC", connection=device_connection, - device_output=unix_remote_proxy_pc_output, test_file_path=__file__) - - iterate_over_device_states(device=unix_remote_proxy_pc) - assert None is not unix_remote_proxy_pc._cmdnames_available_in_state['UNIX_LOCAL_ROOT'] - - -def test_unix_remote_proxy_pc_device_multiple_prompts(device_connection, unix_remote_proxy_pc_output): - unix_remote_proxy_pc_changed_output = copy_dict(unix_remote_proxy_pc_output, deep_copy=True) - combined_line = "moler_bash#" - for src_state in unix_remote_proxy_pc_output.keys(): - for cmd_string in unix_remote_proxy_pc_output[src_state].keys(): - combined_line = f"{combined_line} {unix_remote_proxy_pc_output[src_state][cmd_string]}" - for src_state in unix_remote_proxy_pc_changed_output.keys(): - for cmd_string in unix_remote_proxy_pc_changed_output[src_state].keys(): - unix_remote_proxy_pc_changed_output[src_state][cmd_string] = combined_line - - unix_remote_proxy_pc = get_device(name="UNIX_REMOTE3_PROXY_PC", connection=device_connection, - device_output=unix_remote_proxy_pc_changed_output, - test_file_path=__file__) - assert unix_remote_proxy_pc._check_all_prompts_on_line is True - assert unix_remote_proxy_pc._prompts_event.check_against_all_prompts is True - - with pytest.raises(MolerException) as exception: - iterate_over_device_states(device=unix_remote_proxy_pc, max_no_of_threads=0) - assert "More than 1 prompt match the same line" in str(exception.value) - - -def test_unix_remote_proxy_pc_device_goto_state_bg(device_connection, unix_remote_proxy_pc_output): - unix_remote_proxy_pc = get_device(name="UNIX_REMOTE3_PROXY_PC", connection=device_connection, - device_output=unix_remote_proxy_pc_output, test_file_path=__file__) - unix_remote_proxy_pc._goto_state_in_production_mode = True - dst_state = "UNIX_REMOTE_ROOT" - src_state = "UNIX_LOCAL" - unix_remote_proxy_pc.goto_state(state=src_state, sleep_after_changed_state=0) - assert unix_remote_proxy_pc.current_state == src_state - start_time = time.monotonic() - unix_remote_proxy_pc.goto_state_bg(state=dst_state) - assert unix_remote_proxy_pc.current_state != dst_state - while dst_state != unix_remote_proxy_pc.current_state and (time.monotonic() - start_time) < 10: - MolerTest.sleep(0.01) - execution_time_bg = time.monotonic() - start_time - assert unix_remote_proxy_pc.current_state == dst_state - - unix_remote_proxy_pc.goto_state(state=src_state, sleep_after_changed_state=0) - assert unix_remote_proxy_pc.current_state == src_state - start_time = time.monotonic() - unix_remote_proxy_pc.goto_state(state=dst_state, sleep_after_changed_state=0) - execution_time_fg = time.monotonic() - start_time - assert unix_remote_proxy_pc.current_state == dst_state - time_diff = abs(execution_time_bg - execution_time_fg) - assert time_diff < min(execution_time_fg, execution_time_bg) - - -def test_unix_remote_proxy_pc_device_goto_state_bg_and_goto(device_connection, unix_remote_proxy_pc_output): - unix_remote_proxy_pc = get_device(name="UNIX_REMOTE3_PROXY_PC", connection=device_connection, - device_output=unix_remote_proxy_pc_output, test_file_path=__file__) - unix_remote_proxy_pc._goto_state_in_production_mode = True - - dst_state = "UNIX_REMOTE_ROOT" - src_state = "UNIX_LOCAL" - unix_remote_proxy_pc.goto_state(state=src_state, sleep_after_changed_state=0) - assert unix_remote_proxy_pc.current_state == src_state - unix_remote_proxy_pc.goto_state_bg(state=dst_state) - assert unix_remote_proxy_pc.current_state != dst_state - unix_remote_proxy_pc.goto_state(state=dst_state, sleep_after_changed_state=0) - assert unix_remote_proxy_pc.current_state == dst_state - - -def test_unix_remote_proxy_pc_device_goto_state_bg_await(device_connection, unix_remote_proxy_pc_output): - unix_remote_proxy_pc = get_device(name="UNIX_REMOTE3_PROXY_PC", connection=device_connection, - device_output=unix_remote_proxy_pc_output, test_file_path=__file__) - unix_remote_proxy_pc._goto_state_in_production_mode = True - dst_state = "UNIX_REMOTE_ROOT" - src_state = "UNIX_LOCAL" - unix_remote_proxy_pc.goto_state(state=src_state, sleep_after_changed_state=0) - assert unix_remote_proxy_pc.current_state == src_state - unix_remote_proxy_pc.goto_state_bg(state=dst_state) - assert unix_remote_proxy_pc.current_state != dst_state - unix_remote_proxy_pc.await_goto_state() - assert unix_remote_proxy_pc.current_state == dst_state - - -def test_unix_remote_proxy_pc_device_goto_state_bg_await_excption(device_connection, unix_remote_proxy_pc_output): - unix_remote_proxy_pc = get_device(name="UNIX_REMOTE3_PROXY_PC", connection=device_connection, - device_output=unix_remote_proxy_pc_output, test_file_path=__file__) - unix_remote_proxy_pc._goto_state_in_production_mode = True - dst_state = "UNIX_REMOTE_ROOT" - src_state = "UNIX_LOCAL" - unix_remote_proxy_pc.goto_state(state=src_state, sleep_after_changed_state=0) - assert unix_remote_proxy_pc.current_state == src_state - unix_remote_proxy_pc.goto_state_bg(state=dst_state) - assert unix_remote_proxy_pc.current_state != dst_state - with pytest.raises(DeviceChangeStateFailure) as de: - unix_remote_proxy_pc.await_goto_state(timeout=0.001) - assert 'seconds there are still states to go' in str(de.value) - unix_remote_proxy_pc.await_goto_state() - assert unix_remote_proxy_pc.current_state == dst_state - - -def test_unix_remote_device_not_connected(): - dir_path = os.path.dirname(os.path.realpath(__file__)) - load_config(os.path.join(dir_path, os.pardir, os.pardir, 'test', 'resources', 'device_config.yml')) - unix_remote = DeviceFactory.get_device(name="UNIX_REMOTE3_REAL_IO", initial_state="UNIX_LOCAL") - unix_remote.goto_state("UNIX_LOCAL", sleep_after_changed_state=0) - cmd_whoami = unix_remote.get_cmd(cmd_name="whoami") - ret1 = cmd_whoami() - execution = 0 - while execution < 5: - unix_remote.goto_state("NOT_CONNECTED", sleep_after_changed_state=0) - with pytest.raises(DeviceFailure) as ex: - cmd_whoami = unix_remote.get_cmd(cmd_name="whoami") - cmd_whoami() - assert "cmd is unknown for state 'NOT_CONNECTED'" in str(ex) - assert unix_remote.io_connection._terminal is None - assert unix_remote.io_connection.moler_connection.is_open() is False - unix_remote.goto_state("UNIX_LOCAL", sleep_after_changed_state=0) - assert unix_remote.io_connection._terminal is not None - assert unix_remote.io_connection.moler_connection.is_open() is True - cmd_whoami = unix_remote.get_cmd(cmd_name="whoami") - ret2 = cmd_whoami() - assert ret1 == ret2 - execution += 1 - - -@pytest.fixture -def unix_remote_output3(): - output = { - "UNIX_LOCAL": { - 'TERM=xterm-mono ssh -l remote_login -o ServerAliveInterval=7 -o ServerAliveCountMax=2 remote_host': 'remote#', - 'su': 'local_root_prompt' - }, - "UNIX_LOCAL_ROOT": { - 'exit': 'moler_bash#' - }, - "UNIX_REMOTE": { - 'exit': 'moler_bash#', - 'su': 'remote_root_prompt' - }, - "UNIX_REMOTE_ROOT": { - 'exit': 'remote#', - }, - } - - return output - - -@pytest.fixture -def unix_remote_proxy_pc_output(): - 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' - }, - "UNIX_LOCAL_ROOT": { - 'exit': 'moler_bash#' - }, - "UNIX_REMOTE": { - 'exit': 'proxy_pc#', - 'su': 'remote_root_prompt' - }, - "PROXY_PC": { - 'TERM=xterm-mono ssh -l remote_login -o ServerAliveInterval=7 -o ServerAliveCountMax=2 remote_host': 'remote#', - 'exit': 'moler_bash#' - }, - "UNIX_REMOTE_ROOT": { - 'exit': 'remote#', - }, - } - - return output diff --git a/test/resources/device_config.yml b/test/resources/device_config.yml index b9f5640c8..9f391a6bb 100644 --- a/test/resources/device_config.yml +++ b/test/resources/device_config.yml @@ -175,6 +175,44 @@ DEVICES: command_params: expected_prompt: "proxy_pc#" + JUNIPER_EX3: + DEVICE_CLASS: moler.device.juniper_ex3.JuniperEX3 + INITIAL_STATE: UNIX_LOCAL + CONNECTION_HOPS: + UNIX_LOCAL: + CLI: + execute_command: ssh # default value + command_params: + host: cli_host + login: cli_login + password: password + + JUNIPER_EX_PROXY_PC3: + DEVICE_CLASS: moler.device.juniper_ex3.JuniperEX3 + INITIAL_STATE: UNIX_LOCAL + CONNECTION_HOPS: + PROXY_PC: + CLI: + execute_command: ssh + command_params: + host: cli_host + login: cli_login + password: password + 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 + CLI: + PROXY_PC: + execute_command: exit + command_params: + expected_prompt: "proxy_pc#" + SCPI: DEVICE_CLASS: moler.device.scpi.Scpi INITIAL_STATE: UNIX_LOCAL @@ -215,6 +253,46 @@ DEVICES: password: password set_timeout: null + SCPI3: + DEVICE_CLASS: moler.device.scpi3.Scpi3 + INITIAL_STATE: UNIX_LOCAL + CONNECTION_HOPS: + UNIX_LOCAL: + SCPI: + execute_command: telnet # default value + command_params: + expected_prompt: SCPI> + host: 10.0.0.1 + port: 99999 + set_timeout: null + + SCPI_PROXY_PC3: + DEVICE_CLASS: moler.device.scpi3.Scpi3 + INITIAL_STATE: UNIX_LOCAL + CONNECTION_HOPS: + PROXY_PC: + SCPI: + execute_command: telnet # default value + command_params: + expected_prompt: SCPI> + host: 10.0.0.1 + port: 99999 + set_timeout: null + SCPI: + PROXY_PC: + execute_command: exit_telnet # default value + command_params: + expected_prompt: "proxy_pc#" + 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 + PDU: DEVICE_CLASS: moler.device.pdu_aten.PduAten INITIAL_STATE: UNIX_LOCAL @@ -252,6 +330,42 @@ DEVICES: set_timeout: null + PDU3: + DEVICE_CLASS: moler.device.pdu_aten3.PduAten3 + INITIAL_STATE: UNIX_LOCAL + CONNECTION_HOPS: + UNIX_LOCAL: + PDU: + execute_command: telnet # default value + command_params: + host: 10.0.0.1 + cmds_before_establish_connection: null + + PDU_PROXY_PC3: + DEVICE_CLASS: moler.device.pdu_aten3.PduAten3 + INITIAL_STATE: UNIX_LOCAL + CONNECTION_HOPS: + PROXY_PC: + PDU: + execute_command: telnet # default value + command_params: + host: 10.0.0.1 + cmds_before_establish_connection: null + PDU: + PROXY_PC: + execute_command: exit_telnet # default value + command_params: + expected_prompt: "proxy_pc#" + 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 + AT_REMOTE: DEVICE_CLASS: moler.device.atremote.AtRemote INITIAL_STATE: UNIX_LOCAL @@ -271,6 +385,81 @@ 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' + + AT_REMOTE3: + DEVICE_CLASS: moler.device.atremote3.AtRemote3 + INITIAL_STATE: UNIX_LOCAL + CONNECTION_HOPS: + UNIX_LOCAL: + 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' + + AT_REMOTE_PROXY_PC3: + DEVICE_CLASS: moler.device.atremote3.AtRemote3 + 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 @@ -289,3 +478,78 @@ DEVICES: execute_command: adb_shell # default value; default command is: adb shell command_params: serial_number: 'f57e6b77' # to create: adb -s f57e6b77 shell + + ADB_REMOTE_PROXY_PC: + DEVICE_CLASS: moler.device.adbremote.AdbRemote + 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: + ADB_SHELL: + execute_command: adb_shell # default value; default command is: adb shell + command_params: + serial_number: 'f57e6b77' # to create: adb -s f57e6b77 shell + + ADB_REMOTE3: + DEVICE_CLASS: moler.device.adbremote3.AdbRemote3 + INITIAL_STATE: UNIX_LOCAL + CONNECTION_HOPS: + UNIX_LOCAL: + 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: + ADB_SHELL: + execute_command: adb_shell # default value; default command is: adb shell + command_params: + serial_number: 'f57e6b77' # to create: adb -s f57e6b77 shell + + ADB_REMOTE_PROXY_PC3: + DEVICE_CLASS: moler.device.adbremote3.AdbRemote3 + 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: + ADB_SHELL: + execute_command: adb_shell # default value; default command is: adb shell + command_params: + serial_number: 'f57e6b77' # to create: adb -s f57e6b77 shell \ No newline at end of file diff --git a/test/test_helpers.py b/test/test_helpers.py index deb3246d3..40490ca43 100644 --- a/test/test_helpers.py +++ b/test/test_helpers.py @@ -694,3 +694,117 @@ def test_remove_state_hops_from_sm(): current_hops = remove_state_hops_from_sm(source_hops, UnixRemote.proxy_pc) assert expected_hops == current_hops + + +def test_remove_state_hops_from_sm_adb(): + from moler.helpers import remove_state_hops_from_sm + source_hops = { + 'ADB_SHELL': { + 'NOT_CONNECTED': 'UNIX_REMOTE', + 'PROXY_PC': 'UNIX_REMOTE', + 'UNIX_LOCAL': 'UNIX_REMOTE', + 'UNIX_LOCAL_ROOT': 'UNIX_REMOTE', + 'UNIX_REMOTE_ROOT': 'UNIX_REMOTE' + }, + 'ADB_SHELL_ROOT': { + 'NOT_CONNECTED': 'ADB_SHELL', + 'PROXY_PC': 'ADB_SHELL', + 'UNIX_LOCAL': 'ADB_SHELL', + 'UNIX_LOCAL_ROOT': 'ADB_SHELL', + 'UNIX_REMOTE': 'ADB_SHELL', + 'UNIX_REMOTE_ROOT': 'ADB_SHELL' + }, + 'NOT_CONNECTED': { + 'ADB_SHELL': 'UNIX_LOCAL', + 'ADB_SHELL_ROOT': 'UNIX_LOCAL', + 'PROXY_PC': 'UNIX_LOCAL', + 'UNIX_LOCAL_ROOT': 'UNIX_LOCAL', + 'UNIX_REMOTE': 'UNIX_LOCAL', + 'UNIX_REMOTE_ROOT': 'UNIX_LOCAL' + }, + 'PROXY_PC': { + 'ADB_SHELL': 'UNIX_REMOTE', + 'ADB_SHELL_ROOT': 'UNIX_REMOTE', + 'NOT_CONNECTED': 'UNIX_LOCAL', + 'UNIX_LOCAL_ROOT': 'UNIX_LOCAL', + 'UNIX_REMOTE_ROOT': 'UNIX_REMOTE' + }, + 'UNIX_LOCAL': { + 'ADB_SHELL': 'PROXY_PC', + 'ADB_SHELL_ROOT': 'PROXY_PC', + 'UNIX_REMOTE': 'PROXY_PC', + 'UNIX_REMOTE_ROOT': 'PROXY_PC' + }, + 'UNIX_LOCAL_ROOT': { + 'ADB_SHELL': 'UNIX_LOCAL', + 'ADB_SHELL_ROOT': 'UNIX_LOCAL', + 'NOT_CONNECTED': 'UNIX_LOCAL', + 'PROXY_PC': 'UNIX_LOCAL', + 'UNIX_REMOTE': 'UNIX_LOCAL', + 'UNIX_REMOTE_ROOT': 'UNIX_LOCAL' + }, + 'UNIX_REMOTE': { + 'ADB_SHELL_ROOT': 'ADB_SHELL', + 'NOT_CONNECTED': 'PROXY_PC', + 'UNIX_LOCAL': 'PROXY_PC', + 'UNIX_LOCAL_ROOT': 'PROXY_PC' + }, + 'UNIX_REMOTE_ROOT': { + 'ADB_SHELL': 'UNIX_REMOTE', + 'ADB_SHELL_ROOT': 'UNIX_REMOTE', + 'NOT_CONNECTED': 'UNIX_REMOTE', + 'PROXY_PC': 'UNIX_REMOTE', + 'UNIX_LOCAL': 'UNIX_REMOTE', + 'UNIX_LOCAL_ROOT': 'UNIX_REMOTE' + } + } + + expected_hops = { + 'ADB_SHELL': { + 'NOT_CONNECTED': 'UNIX_REMOTE', + 'UNIX_LOCAL': 'UNIX_REMOTE', + 'UNIX_LOCAL_ROOT': 'UNIX_REMOTE', + 'UNIX_REMOTE_ROOT': 'UNIX_REMOTE' + }, + 'ADB_SHELL_ROOT': { + 'NOT_CONNECTED': 'ADB_SHELL', + 'UNIX_LOCAL': 'ADB_SHELL', + 'UNIX_LOCAL_ROOT': 'ADB_SHELL', + 'UNIX_REMOTE': 'ADB_SHELL', + 'UNIX_REMOTE_ROOT': 'ADB_SHELL' + }, + 'NOT_CONNECTED': { + 'ADB_SHELL': 'UNIX_LOCAL', + 'ADB_SHELL_ROOT': 'UNIX_LOCAL', + 'UNIX_LOCAL_ROOT': 'UNIX_LOCAL', + 'UNIX_REMOTE': 'UNIX_LOCAL', + 'UNIX_REMOTE_ROOT': 'UNIX_LOCAL' + }, + 'UNIX_LOCAL': { + 'ADB_SHELL': 'UNIX_REMOTE', + 'ADB_SHELL_ROOT': 'UNIX_REMOTE', + 'UNIX_REMOTE_ROOT': 'UNIX_REMOTE' + }, + 'UNIX_LOCAL_ROOT': { + 'ADB_SHELL': 'UNIX_LOCAL', + 'ADB_SHELL_ROOT': 'UNIX_LOCAL', + 'NOT_CONNECTED': 'UNIX_LOCAL', + 'UNIX_REMOTE': 'UNIX_LOCAL', + 'UNIX_REMOTE_ROOT': 'UNIX_LOCAL' + }, + 'UNIX_REMOTE': { + 'ADB_SHELL_ROOT': 'ADB_SHELL', + 'NOT_CONNECTED': 'UNIX_LOCAL', + 'UNIX_LOCAL_ROOT': 'UNIX_LOCAL' + }, + 'UNIX_REMOTE_ROOT': { + 'ADB_SHELL': 'UNIX_REMOTE', + 'ADB_SHELL_ROOT': 'UNIX_REMOTE', + 'NOT_CONNECTED': 'UNIX_REMOTE', + 'UNIX_LOCAL': 'UNIX_REMOTE', + 'UNIX_LOCAL_ROOT': 'UNIX_REMOTE' + } + } + + current_hops = remove_state_hops_from_sm(source_hops, "PROXY_PC") + assert expected_hops == current_hops