From a40c5f075e316a7431f12eca7948ea42574f43c5 Mon Sep 17 00:00:00 2001 From: Marcin Usielski <35992110+marcin-usielski@users.noreply.github.com> Date: Mon, 21 Oct 2024 11:24:05 +0200 Subject: [PATCH] Remove state from state machine dict - UnixRemote conditionally deletes PROXY_PC (#547) * Remove state from state machine dict * upload-artifact@v4 * fix * delte transitions * tuple annotations * start dev unixremote3 * dev * dev * style * rd * tests * test * t2 * fix formattig * warning * warning * review#2 * after review --- moler/device/textualdevice.py | 18 +- moler/device/unixlocal.py | 1 - moler/device/unixremote.py | 74 ++--- moler/device/unixremote3.py | 413 ++++++++++++++++++++++++++++ moler/helpers.py | 86 ++++++ moler/util/devices_SM.py | 1 + test/device/test_SM_unix_remote.py | 57 ++-- test/device/test_SM_unix_remote3.py | 195 +++++++++++++ test/resources/device_config.yml | 56 ++++ test/test_helpers.py | 266 ++++++++++++++++++ 10 files changed, 1110 insertions(+), 57 deletions(-) create mode 100644 moler/device/unixremote3.py create mode 100644 test/device/test_SM_unix_remote3.py diff --git a/moler/device/textualdevice.py b/moler/device/textualdevice.py index 9fe967f3c..242446132 100644 --- a/moler/device/textualdevice.py +++ b/moler/device/textualdevice.py @@ -123,10 +123,8 @@ def __init__( self.logger = logging.getLogger(f"moler.connection.{self.name}") self.configure_logger(name=self.name, propagate=False) - self._prepare_transitions() - self._prepare_state_hops() - self._configure_state_machine(sm_params) - self._prepare_newline_chars() + self._stored_transitions = {} + self._prepare_sm_data(sm_params=sm_params) # TODO: Need test to ensure above sentence for all connection self.io_connection.notify( @@ -157,6 +155,13 @@ def __init__( # same line. self._sleep_after_state_change = 0.5 + def _prepare_sm_data(self, sm_params): + self._prepare_transitions() + self._prepare_state_hops() + self._configure_state_machine(sm_params) + self._prepare_newline_chars() + self._send_transitions_to_sm(self._stored_transitions) + def set_all_prompts_on_line(self, value=True): """ Set True to check all prompts on line. False to interrupt after 1st prompt (default). @@ -993,7 +998,10 @@ def _collect_events_for_state(self, state): return events - def _add_transitions(self, transitions): + def _add_transitions(self, transitions: dict): + self._update_dict(self._stored_transitions, transitions) + + def _send_transitions_to_sm(self, transitions: dict) -> None: for source_state in transitions.keys(): for dest_state in transitions[source_state].keys(): self._update_SM_states(dest_state) diff --git a/moler/device/unixlocal.py b/moler/device/unixlocal.py index d906347e3..25132e184 100644 --- a/moler/device/unixlocal.py +++ b/moler/device/unixlocal.py @@ -242,7 +242,6 @@ def _execute_command_to_change_state(self, source_state, dest_state, timeout=-1) command_name = configurations["execute_command"] command_params = configurations["command_params"] - command_timeout = self.calc_timeout_for_command(timeout, command_params) command_params_without_timeout = self._parameters_without_timeout( parameters=command_params diff --git a/moler/device/unixremote.py b/moler/device/unixremote.py index e6053b2dd..f729888d2 100644 --- a/moler/device/unixremote.py +++ b/moler/device/unixremote.py @@ -6,7 +6,7 @@ """ __author__ = 'Grzegorz Latuszek, Marcin Usielski, Michal Ernst' -__copyright__ = 'Copyright (C) 2018-2019, Nokia' +__copyright__ = 'Copyright (C) 2018-2024, Nokia' __email__ = 'grzegorz.latuszek@nokia.com, marcin.usielski@nokia.com, michal.ernst@nokia.com' from moler.device.proxy_pc import ProxyPc @@ -25,41 +25,41 @@ class UnixRemote(ProxyPc): Example of device in yaml configuration file: - with PROXY_PC: UNIX_1: - DEVICE_CLASS: moler.device.unixremote.UnixRemote - CONNECTION_HOPS: - PROXY_PC: - UNIX_REMOTE: - execute_command: ssh # default value - command_params: - expected_prompt: unix_remote_prompt - host: host_ip - login: login - password: password - UNIX_REMOTE: - PROXY_PC: - execute_command: exit # default value - command_params: - expected_prompt: proxy_pc_prompt - UNIX_LOCAL: - PROXY_PC: - execute_command: ssh # default value - command_params: - expected_prompt: proxy_pc_prompt - host: host_ip - login: login - password: password + DEVICE_CLASS: moler.device.unixremote.UnixRemote + CONNECTION_HOPS: + PROXY_PC: + UNIX_REMOTE: + execute_command: ssh # default value + command_params: + expected_prompt: unix_remote_prompt + host: host_ip + login: login + password: password + UNIX_REMOTE: + PROXY_PC: + execute_command: exit # default value + command_params: + expected_prompt: proxy_pc_prompt + UNIX_LOCAL: + PROXY_PC: + execute_command: ssh # default value + command_params: + expected_prompt: proxy_pc_prompt + host: host_ip + login: login + password: password -without PROXY_PC: UNIX_1: - DEVICE_CLASS: moler.device.unixremote.UnixRemote - 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 + DEVICE_CLASS: moler.device.unixremote.UnixRemote + 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 """ @@ -417,7 +417,6 @@ def _prepare_state_hops_without_proxy_pc(self): UnixRemote.not_connected: UnixRemote.unix_remote, UnixRemote.unix_local: UnixRemote.unix_remote, UnixRemote.unix_local_root: UnixRemote.unix_remote, - UnixRemote.proxy_pc: UnixRemote.unix_remote, } } return state_hops @@ -429,7 +428,12 @@ def _configure_state_machine(self, sm_params): :return: None. """ super(UnixRemote, self)._configure_state_machine(sm_params) + self._overwrite_prompts() + 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_root][UnixRemote.unix_remote][ "command_params"]["expected_prompt"] = \ diff --git a/moler/device/unixremote3.py b/moler/device/unixremote3.py new file mode 100644 index 000000000..76018e76b --- /dev/null +++ b/moler/device/unixremote3.py @@ -0,0 +1,413 @@ +# -*- 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 logging +from moler.device.proxy_pc import ProxyPc +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): + r""" + UnixRemote3 device class. + + + :: + + + Example of device in yaml configuration file: + - with PROXY_PC: + UNIX_1: + DEVICE_CLASS: moler.device.UnixRemote3.UnixRemote3 + CONNECTION_HOPS: + PROXY_PC: + UNIX_REMOTE: + execute_command: ssh # default value + command_params: + expected_prompt: unix_remote_prompt + host: host_ip + login: login + password: password + UNIX_REMOTE: + PROXY_PC: + execute_command: exit # default value + command_params: + expected_prompt: proxy_pc_prompt + UNIX_LOCAL: + PROXY_PC: + execute_command: ssh # default value + command_params: + expected_prompt: proxy_pc_prompt + host: host_ip + login: login + password: password + -without PROXY_PC: + UNIX_1: + DEVICE_CLASS: moler.device.UnixRemote3.UnixRemote3 + 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 = "UNIX_REMOTE" + unix_remote_root = "UNIX_REMOTE_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 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 UnixRemote3.unix_remote + ) + super(UnixRemote3, 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(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. + """ + if self._use_proxy_pc: + 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][ + "command_params"]["expected_prompt"] + else: + self._configurations[UnixRemote3.connection_hops][UnixRemote3.unix_remote_root][UnixRemote3.unix_remote][ + "command_params"]["expected_prompt"] = \ + self._configurations[UnixRemote3.connection_hops][UnixRemote3.unix_local][UnixRemote3.unix_remote][ + "command_params"]["expected_prompt"] + + @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 = { + UnixRemote3.connection_hops: { + UnixRemote3.proxy_pc: { # from + UnixRemote3.unix_remote: { # to + "execute_command": "ssh", # using command + "command_params": { # with parameters + "target_newline": "\n" + }, + "required_command_params": [ + "host", + "login", + "password", + "expected_prompt", + ], + }, + }, + UnixRemote3.unix_remote: { # from + UnixRemote3.proxy_pc: { # to + "execute_command": "exit", # using command + "command_params": { # with parameters + "target_newline": "\n" + }, + "required_command_params": ["expected_prompt"], + }, + UnixRemote3.unix_remote_root: { # to + "execute_command": "su", # using command + "command_params": { # with parameters + "password": "root_password", + "expected_prompt": r"remote_root_prompt", + "target_newline": "\n", + }, + "required_command_params": [], + }, + }, + UnixRemote3.unix_remote_root: { # from + UnixRemote3.unix_remote: { # to + "execute_command": "exit", # using command + "command_params": { # with parameters + "target_newline": "\n", + "expected_prompt": r"remote_user_prompt", + }, + "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 with proxy_pc state. + :return: transitions with proxy_pc state. + """ + transitions = { + UnixRemote3.proxy_pc: { + UnixRemote3.unix_remote: { + "action": ["_execute_command_to_change_state"], + } + }, + UnixRemote3.unix_remote: { + UnixRemote3.proxy_pc: { + "action": ["_execute_command_to_change_state"], + }, + UnixRemote3.unix_remote_root: { + "action": ["_execute_command_to_change_state"], + }, + }, + UnixRemote3.unix_remote_root: { + UnixRemote3.unix_remote: { + "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 = { + UnixRemote3.unix_remote: self._configurations[UnixRemote3.connection_hops][ + UnixRemote3.proxy_pc + ][UnixRemote3.unix_remote]["command_params"]["expected_prompt"], + UnixRemote3.unix_remote_root: self._configurations[ + UnixRemote3.connection_hops + ][UnixRemote3.unix_remote][UnixRemote3.unix_remote_root]["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 = { + UnixRemote3.unix_remote: self._configurations[UnixRemote3.connection_hops][ + UnixRemote3.unix_local + ][UnixRemote3.unix_remote]["command_params"]["expected_prompt"], + UnixRemote3.unix_remote_root: self._configurations[ + UnixRemote3.connection_hops + ][UnixRemote3.unix_remote][UnixRemote3.unix_remote_root]["command_params"][ + "expected_prompt" + ], + UnixRemote3.unix_local: self._configurations[UnixRemote3.connection_hops][ + UnixRemote3.unix_remote + ][UnixRemote3.unix_local]["command_params"]["expected_prompt"], + } + return state_prompts + + @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 = { + UnixRemote3.unix_remote: self._configurations[UnixRemote3.connection_hops][ + UnixRemote3.proxy_pc + ][UnixRemote3.unix_remote]["command_params"]["target_newline"], + UnixRemote3.unix_remote_root: self._configurations[ + UnixRemote3.connection_hops + ][UnixRemote3.unix_remote][UnixRemote3.unix_remote_root]["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 = { + UnixRemote3.unix_remote: self._configurations[UnixRemote3.connection_hops][ + UnixRemote3.unix_local + ][UnixRemote3.unix_remote]["command_params"]["target_newline"], + UnixRemote3.unix_local: self._configurations[UnixRemote3.connection_hops][ + UnixRemote3.unix_remote + ][UnixRemote3.unix_local]["command_params"]["target_newline"], + UnixRemote3.unix_remote_root: self._configurations[ + UnixRemote3.connection_hops + ][UnixRemote3.unix_remote][UnixRemote3.unix_remote_root]["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 with proxy_pc state. + :return: non direct transitions for each state with proxy_pc state. + """ + state_hops = { + UnixRemote3.not_connected: { + UnixRemote3.unix_remote: UnixRemote3.unix_local, + UnixRemote3.proxy_pc: UnixRemote3.unix_local, + UnixRemote3.unix_local_root: UnixRemote3.unix_local, + UnixRemote3.unix_remote_root: UnixRemote3.unix_local, + }, + UnixRemote3.unix_remote: { + UnixRemote3.not_connected: UnixRemote3.proxy_pc, + UnixRemote3.unix_local: UnixRemote3.proxy_pc, + UnixRemote3.unix_local_root: UnixRemote3.proxy_pc, + }, + UnixRemote3.unix_local_root: { + UnixRemote3.not_connected: UnixRemote3.unix_local, + UnixRemote3.unix_remote: UnixRemote3.unix_local, + UnixRemote3.unix_remote_root: UnixRemote3.unix_local, + }, + UnixRemote3.proxy_pc: { + UnixRemote3.not_connected: UnixRemote3.unix_local, + UnixRemote3.unix_local_root: UnixRemote3.unix_local, + UnixRemote3.unix_remote_root: UnixRemote3.unix_remote, + }, + UnixRemote3.unix_local: { + UnixRemote3.unix_remote: UnixRemote3.proxy_pc, + UnixRemote3.unix_remote_root: UnixRemote3.proxy_pc, + }, + UnixRemote3.unix_remote_root: { + UnixRemote3.not_connected: UnixRemote3.unix_remote, + UnixRemote3.unix_local: UnixRemote3.unix_remote, + UnixRemote3.unix_local_root: UnixRemote3.unix_remote, + UnixRemote3.proxy_pc: UnixRemote3.unix_remote, + }, + } + 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(UnixRemote3, self)._get_packages_for_state(state, observer) + + if not available: + if state == UnixRemote3.unix_remote or state == UnixRemote3.unix_remote_root: + available = { + UnixRemote3.cmds: ["moler.cmd.unix"], + UnixRemote3.events: ["moler.events.shared", "moler.events.unix"], + } + if available: + return available[observer] + + return available diff --git a/moler/helpers.py b/moler/helpers.py index 5a47c9a93..a5d0e53c3 100644 --- a/moler/helpers.py +++ b/moler/helpers.py @@ -574,3 +574,89 @@ def regexp_without_anchors(regexp): if regexp_str == org_regexp_str: return regexp return re.compile(regexp_str) + + +def remove_state_from_sm(source_sm: dict, source_transitions: dict, state_to_remove: str) -> tuple: + """ + Remove a state from a state machine dict. + :param source_sm: a dict with a state machine description + :param source_transitions: a dict with a state machine transitions + :param state_to_remove: name of state to remove + :return: tuple with 2 dicts without state_to_remove, 0 - new state machine, 1 - new transitions + """ + new_sm = copy.deepcopy(source_sm) + new_transitions = copy.deepcopy(source_transitions) + + states_from_state_to_remove = [] + for from_state in source_sm.keys(): + for to_state in source_sm[from_state].keys(): + if to_state == state_to_remove: + states_from_state_to_remove.append(from_state) + + for to_state in source_sm[state_to_remove].keys(): + if to_state == state_to_remove: + continue + for new_from in states_from_state_to_remove: + if new_from != to_state: + break + if new_from not in new_sm: + 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 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: + new_transitions[new_from][to_state] = { + "action": [ + "_execute_command_to_change_state" + ], + } + + _delete_state(sm=new_sm, state_to_remove=state_to_remove) + _delete_state(sm=new_transitions, state_to_remove=state_to_remove) + + return (new_sm, new_transitions) + + +def _delete_state(sm: dict, state_to_remove: str) -> None: + """ + Delete state from a state machine dict (in place). + :param sm: dict with state machine + :param state_to_remove: name of state to delete + :return: None + """ + if state_to_remove in sm: + del sm[state_to_remove] + for from_state in sm: + if from_state in sm and state_to_remove in sm[from_state]: + del sm[from_state][state_to_remove] + + +def remove_state_hops_from_sm(source_hops: dict, state_to_remove: str) -> dict: + """ + Remove a state from a state machine dict. + :param source_sm: a dict with state machine description + :param state_to_remove: name of state to remove + :return: a new state machine hops dict without state_to_remove + """ + new_hops = copy.deepcopy(source_hops) + + for from_state in source_hops.keys(): + item = source_hops[from_state] + for dest_state in item.keys(): + 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]: + new_hops[from_state][dest_state] = source_hops[state_to_remove][dest_state] + else: + del new_hops[from_state][dest_state] + + for from_state in source_hops.keys(): + if from_state in new_hops and state_to_remove in new_hops[from_state]: + del new_hops[from_state][state_to_remove] + + if state_to_remove in new_hops: + del new_hops[state_to_remove] + + return new_hops diff --git a/moler/util/devices_SM.py b/moler/util/devices_SM.py index 5f2f4e05f..a957fce09 100644 --- a/moler/util/devices_SM.py +++ b/moler/util/devices_SM.py @@ -209,6 +209,7 @@ def _perform_device_tests( sleep_after_changed_state=sleep_after_changed_state, ) tested.add((source_state, target_state)) + if device.last_wrong_wait4_occurrence is not None: prompts = [{item["state"]: item["prompt_regex"]} for item in device.last_wrong_wait4_occurrence["list_matched"]] diff --git a/test/device/test_SM_unix_remote.py b/test/device/test_SM_unix_remote.py index 505737e6e..d8b8fdab2 100644 --- a/test/device/test_SM_unix_remote.py +++ b/test/device/test_SM_unix_remote.py @@ -14,22 +14,30 @@ from moler.exceptions import DeviceFailure -def test_unix_remote_device(device_connection, unix_remote_output): - unix_remote = get_device(name="UNIX_REMOTE", connection=device_connection, device_output=unix_remote_output, +unix_remotes=['UNIX_REMOTE', 'UNIX_REMOTE3'] +unix_remotes_proxy_pc=['UNIX_REMOTE_PROXY_PC', 'UNIX_REMOTE3_PROXY_PC'] +unix_remotes_real_io = ['UNIX_REMOTE_REAL_IO', 'UNIX_REMOTE3_REAL_IO'] + + +@pytest.mark.parametrize("device_name", unix_remotes) +def test_unix_remote_device(device_name, device_connection, unix_remote_output): + unix_remote = get_device(name=device_name, connection=device_connection, device_output=unix_remote_output, 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_REMOTE_PROXY_PC", connection=device_connection, +@pytest.mark.parametrize("device_name", unix_remotes_proxy_pc) +def test_unix_remote_proxy_pc_device(device_name, device_connection, unix_remote_proxy_pc_output): + unix_remote_proxy_pc = get_device(name=device_name, 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): +@pytest.mark.parametrize("device_name", unix_remotes_proxy_pc) +def test_unix_remote_proxy_pc_device_multiple_prompts(device_name, 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(): @@ -39,7 +47,7 @@ def test_unix_remote_proxy_pc_device_multiple_prompts(device_connection, unix_re 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_REMOTE_PROXY_PC", connection=device_connection, + unix_remote_proxy_pc = get_device(name=device_name, 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 @@ -50,8 +58,9 @@ def test_unix_remote_proxy_pc_device_multiple_prompts(device_connection, unix_re 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_REMOTE_PROXY_PC", connection=device_connection, +@pytest.mark.parametrize("device_name", unix_remotes_proxy_pc) +def test_unix_remote_proxy_pc_device_goto_state_bg(device_name, device_connection, unix_remote_proxy_pc_output): + unix_remote_proxy_pc = get_device(name=device_name, 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" @@ -76,8 +85,9 @@ def test_unix_remote_proxy_pc_device_goto_state_bg(device_connection, unix_remot 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_REMOTE_PROXY_PC", connection=device_connection, +@pytest.mark.parametrize("device_name", unix_remotes_proxy_pc) +def test_unix_remote_proxy_pc_device_goto_state_bg_and_goto(device_name, device_connection, unix_remote_proxy_pc_output): + unix_remote_proxy_pc = get_device(name=device_name, connection=device_connection, device_output=unix_remote_proxy_pc_output, test_file_path=__file__) unix_remote_proxy_pc._goto_state_in_production_mode = True @@ -91,8 +101,9 @@ def test_unix_remote_proxy_pc_device_goto_state_bg_and_goto(device_connection, u 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_REMOTE_PROXY_PC", connection=device_connection, +@pytest.mark.parametrize("device_name", unix_remotes_proxy_pc) +def test_unix_remote_proxy_pc_device_goto_state_bg_await(device_name, device_connection, unix_remote_proxy_pc_output): + unix_remote_proxy_pc = get_device(name=device_name, 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" @@ -105,8 +116,9 @@ def test_unix_remote_proxy_pc_device_goto_state_bg_await(device_connection, unix 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_REMOTE_PROXY_PC", connection=device_connection, +@pytest.mark.parametrize("device_name", unix_remotes_proxy_pc) +def test_unix_remote_proxy_pc_device_goto_state_bg_await_exception(device_name, device_connection, unix_remote_proxy_pc_output): + unix_remote_proxy_pc = get_device(name=device_name, 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" @@ -122,10 +134,11 @@ def test_unix_remote_proxy_pc_device_goto_state_bg_await_excption(device_connect assert unix_remote_proxy_pc.current_state == dst_state -def test_unix_remote_device_not_connected(): +@pytest.mark.parametrize("device_name", unix_remotes_real_io) +def test_unix_remote_device_not_connected(device_name): 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_REMOTE_REAL_IO", initial_state="UNIX_LOCAL") + unix_remote = DeviceFactory.get_device(name=device_name, 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() @@ -147,6 +160,18 @@ def test_unix_remote_device_not_connected(): execution += 1 +@pytest.mark.parametrize("devices", [unix_remotes, unix_remotes_proxy_pc, unix_remotes_real_io]) +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 unix_remote_output(): output = { diff --git a/test/device/test_SM_unix_remote3.py b/test/device/test_SM_unix_remote3.py new file mode 100644 index 000000000..ec0d84c61 --- /dev/null +++ b/test/device/test_SM_unix_remote3.py @@ -0,0 +1,195 @@ +__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 f41be4ac4..b9f5640c8 100644 --- a/test/resources/device_config.yml +++ b/test/resources/device_config.yml @@ -81,6 +81,62 @@ DEVICES: command_params: expected_prompt: "proxy_pc#" + UNIX_REMOTE3: + DEVICE_CLASS: moler.device.unixremote3.UnixRemote3 + 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: login + set_timeout: null + + UNIX_REMOTE3_REAL_IO: + DEVICE_CLASS: moler.device.unixremote3.UnixRemote3 + 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: login + set_timeout: null + + UNIX_REMOTE3_PROXY_PC: + DEVICE_CLASS: moler.device.unixremote3.UnixRemote3 + INITIAL_STATE: UNIX_LOCAL + LAZY_CMDS_EVENTS: True + CONNECTION_HOPS: + PROXY_PC: + UNIX_REMOTE: + execute_command: ssh + command_params: + host: remote_host + login: remote_login + password: password + expected_prompt: "remote#" + set_timeout: null + 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 + UNIX_REMOTE: + PROXY_PC: + command_params: + expected_prompt: "proxy_pc#" + JUNIPER_EX: DEVICE_CLASS: moler.device.juniper_ex.JuniperEX INITIAL_STATE: UNIX_LOCAL diff --git a/test/test_helpers.py b/test/test_helpers.py index 21f646948..deb3246d3 100644 --- a/test/test_helpers.py +++ b/test/test_helpers.py @@ -428,3 +428,269 @@ def test_date_parser_utc(): date_parsed = ConverterHelper.parse_date(date_str) date_expected = datetime(year=2024, month=5, day=22, hour=9, minute=11, second=48, tzinfo=tzoffset('UTC', 0)) assert date_parsed == date_expected + + +def test_remove_state_from_sm_dict(): + from moler.device.unixremote import UnixRemote + from moler.helpers import remove_state_from_sm + source_sm = { + UnixRemote.unix_local: { + UnixRemote.proxy_pc: { + "execute_command": "ssh", + "command_params": { + "target_newline": "\n" + }, + "required_command_params": [ + "host", + "login", + "password", + "expected_prompt" + ] + }, + }, + UnixRemote.proxy_pc: { # from + UnixRemote.unix_remote: { # to + "execute_command": "ssh", # using command + "command_params": { # with parameters + "target_newline": "\n" + }, + "required_command_params": [ + "host", + "login", + "password", + "expected_prompt" + ] + }, + UnixRemote.unix_local: { # to + "execute_command": "exit", # using command + "command_params": { # with parameters + "target_newline": "\n" + }, + "required_command_params": [ # with parameters + "expected_prompt" + ] + }, + }, + UnixRemote.unix_remote: { # from + UnixRemote.proxy_pc: { # to + "execute_command": "exit", # using command + "command_params": { # with parameters + "target_newline": "\n" + }, + "required_command_params": [ + "expected_prompt" + ] + }, + UnixRemote.unix_remote_root: { # to + "execute_command": "su", # using command + "command_params": { # with parameters + "password": "root_password", + "expected_prompt": r'remote_root_prompt', + "target_newline": "\n" + }, + "required_command_params": [ + ] + }, + }, + UnixRemote.unix_remote_root: { # from + UnixRemote.unix_remote: { # to + "execute_command": "exit", # using command + "command_params": { # with parameters + "target_newline": "\n", + "expected_prompt": r'remote_user_prompt' + }, + "required_command_params": [ + ] + } + } + } + + expected_sm = { + UnixRemote.unix_local: { + UnixRemote.unix_remote: { # to + "execute_command": "ssh", # using command + "command_params": { # with parameters + "target_newline": "\n" + }, + "required_command_params": [ + "host", + "login", + "password", + "expected_prompt" + ] + }, + }, + UnixRemote.unix_remote: { # from + UnixRemote.unix_local: { # to + "execute_command": "exit", # using command + "command_params": { # with parameters + "target_newline": "\n" + }, + "required_command_params": [ + "expected_prompt" + ] + }, + UnixRemote.unix_remote_root: { # to + "execute_command": "su", # using command + "command_params": { # with parameters + "password": "root_password", + "expected_prompt": r'remote_root_prompt', + "target_newline": "\n" + }, + "required_command_params": [ + ] + }, + }, + UnixRemote.unix_remote_root: { # from + UnixRemote.unix_remote: { # to + "execute_command": "exit", # using command + "command_params": { # with parameters + "target_newline": "\n", + "expected_prompt": r'remote_user_prompt' + }, + "required_command_params": [ + ] + } + } + } + + source_transitions = { + UnixRemote.unix_local: { + UnixRemote.proxy_pc: { + "action": [ + "_execute_command_to_change_state" + ], + } + }, + UnixRemote.proxy_pc: { + UnixRemote.unix_local: { + "action": [ + "_execute_command_to_change_state" + ], + }, + }, + UnixRemote.proxy_pc: { + UnixRemote.unix_remote: { + "action": [ + "_execute_command_to_change_state" + ], + } + }, + UnixRemote.unix_remote: { + UnixRemote.proxy_pc: { + "action": [ + "_execute_command_to_change_state" + ], + }, + UnixRemote.unix_remote_root: { + "action": [ + "_execute_command_to_change_state" + ], + } + }, + UnixRemote.unix_remote_root: { + UnixRemote.unix_remote: { + "action": [ + "_execute_command_to_change_state" + ], + } + } + } + + expected_transitions = { + UnixRemote.unix_remote: { + UnixRemote.unix_local: { + "action": [ + "_execute_command_to_change_state" + ], + }, + UnixRemote.unix_remote_root: { + "action": [ + "_execute_command_to_change_state" + ], + } + }, + UnixRemote.unix_local: { + UnixRemote.unix_remote: { + "action": [ + "_execute_command_to_change_state" + ], + } + }, + UnixRemote.unix_remote_root: { + UnixRemote.unix_remote: { + "action": [ + "_execute_command_to_change_state" + ], + } + } + } + + (current_sm, current_transitions) = remove_state_from_sm(source_sm=source_sm, source_transitions=source_transitions, state_to_remove=UnixRemote.proxy_pc) + assert expected_sm == current_sm + assert expected_transitions == current_transitions + + +def test_remove_state_hops_from_sm(): + from moler.device.unixremote import UnixRemote + from moler.helpers import remove_state_hops_from_sm + source_hops = { + UnixRemote.not_connected: { + UnixRemote.unix_remote: UnixRemote.unix_local, + UnixRemote.proxy_pc: UnixRemote.unix_local, + UnixRemote.unix_local_root: UnixRemote.unix_local, + UnixRemote.unix_remote_root: UnixRemote.unix_local + }, + UnixRemote.unix_remote: { + UnixRemote.not_connected: UnixRemote.proxy_pc, + UnixRemote.unix_local: UnixRemote.proxy_pc, + UnixRemote.unix_local_root: UnixRemote.proxy_pc + }, + UnixRemote.unix_local_root: { + UnixRemote.not_connected: UnixRemote.unix_local, + UnixRemote.unix_remote: UnixRemote.unix_local, + UnixRemote.unix_remote_root: UnixRemote.unix_local + }, + UnixRemote.proxy_pc: { + UnixRemote.not_connected: UnixRemote.unix_local, + UnixRemote.unix_local_root: UnixRemote.unix_local, + UnixRemote.unix_remote_root: UnixRemote.unix_remote + }, + UnixRemote.unix_local: { + UnixRemote.unix_remote: UnixRemote.proxy_pc, + UnixRemote.unix_remote_root: UnixRemote.proxy_pc + }, + UnixRemote.unix_remote_root: { + UnixRemote.not_connected: UnixRemote.unix_remote, + UnixRemote.unix_local: UnixRemote.unix_remote, + UnixRemote.unix_local_root: UnixRemote.unix_remote, + UnixRemote.proxy_pc: UnixRemote.unix_remote, + } + } + expected_hops = { + UnixRemote.not_connected: { + UnixRemote.unix_remote: UnixRemote.unix_local, + UnixRemote.unix_local_root: UnixRemote.unix_local, + UnixRemote.unix_remote_root: UnixRemote.unix_local, + }, + UnixRemote.unix_local: { + UnixRemote.unix_remote_root: UnixRemote.unix_remote + }, + UnixRemote.unix_local_root: { + UnixRemote.not_connected: UnixRemote.unix_local, + UnixRemote.unix_remote: UnixRemote.unix_local, + UnixRemote.unix_remote_root: UnixRemote.unix_local + }, + UnixRemote.unix_remote: { + UnixRemote.not_connected: UnixRemote.unix_local, + UnixRemote.unix_local_root: UnixRemote.unix_local + }, + UnixRemote.unix_remote_root: { + UnixRemote.not_connected: UnixRemote.unix_remote, + UnixRemote.unix_local: UnixRemote.unix_remote, + UnixRemote.unix_local_root: UnixRemote.unix_remote, + } + } + + current_hops = remove_state_hops_from_sm(source_hops, UnixRemote.proxy_pc) + assert expected_hops == current_hops