Skip to content

Commit

Permalink
AtRemote
Browse files Browse the repository at this point in the history
  • Loading branch information
@marcin-usielski
marcin-usielski committed Oct 28, 2024
1 parent 373147d commit c1110a5
Show file tree
Hide file tree
Showing 3 changed files with 261 additions and 59 deletions.
248 changes: 192 additions & 56 deletions moler/device/atremote.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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__ = '[email protected]'
__author__ = 'Grzegorz Latuszek, Marcin Usielski'
__copyright__ = 'Copyright (C) 2020-2024, Nokia'
__email__ = '[email protected], [email protected]'

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
Expand All @@ -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"
Expand All @@ -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)
Expand All @@ -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
Expand All @@ -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
Expand All @@ -114,15 +112,72 @@ 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"
],
}
},
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"
],
Expand All @@ -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 = {
Expand All @@ -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"],
}
Expand All @@ -167,37 +250,90 @@ def _prepare_state_hops_without_proxy_pc(self):
"""
state_hops = {
TextualDevice.not_connected: {
UnixLocal.unix_local_root: UnixLocal.unix_local,
UnixRemote.unix_remote: UnixLocal.unix_local,
UnixRemote.unix_remote_root: UnixLocal.unix_local,
AtRemote.at_remote: UnixLocal.unix_local,
},
UnixLocal.unix_local: {
UnixRemote.unix_remote_root: UnixRemote.unix_remote,
AtRemote.at_remote: UnixRemote.unix_remote,
},
UnixLocal.unix_local_root: {
TextualDevice.not_connected: UnixLocal.unix_local,
UnixRemote.unix_remote: UnixLocal.unix_local,
UnixRemote.unix_remote_root: UnixLocal.unix_local,
AtRemote.at_remote: UnixLocal.unix_local,
},
UnixRemote.unix_remote: {
TextualDevice.not_connected: UnixLocal.unix_local,
UnixLocal.unix_local_root: UnixLocal.unix_local,
},
UnixRemote.unix_remote_root: {
TextualDevice.not_connected: UnixRemote.unix_remote,
UnixLocal.unix_local: UnixRemote.unix_remote,
UnixLocal.unix_local_root: UnixRemote.unix_remote,
AtRemote.at_remote: UnixRemote.unix_remote,
AtRemote.unix_local_root: AtRemote.unix_local,
AtRemote.unix_remote: AtRemote.unix_local,
AtRemote.unix_remote_root: AtRemote.unix_local,
AtRemote.at_remote: AtRemote.unix_local,
},
AtRemote.unix_local: {
AtRemote.unix_remote_root: AtRemote.unix_remote,
AtRemote.at_remote: AtRemote.unix_remote,
},
AtRemote.unix_local_root: {
TextualDevice.not_connected: AtRemote.unix_local,
AtRemote.unix_remote: AtRemote.unix_local,
AtRemote.unix_remote_root: AtRemote.unix_local,
AtRemote.at_remote: AtRemote.unix_local,
},
AtRemote.unix_remote: {
TextualDevice.not_connected: AtRemote.unix_local,
AtRemote.unix_local_root: AtRemote.unix_local,
},
AtRemote.unix_remote_root: {
TextualDevice.not_connected: AtRemote.unix_remote,
AtRemote.unix_local: AtRemote.unix_remote,
AtRemote.unix_local_root: AtRemote.unix_remote,
AtRemote.at_remote: AtRemote.unix_remote,
},
AtRemote.at_remote: {
TextualDevice.not_connected: AtRemote.unix_remote,
AtRemote.unix_local: AtRemote.unix_remote,
AtRemote.unix_local_root: AtRemote.unix_remote,
AtRemote.unix_remote_root: AtRemote.unix_remote,
},
}
return state_hops

@mark_to_call_base_class_method_with_same_name
def _prepare_state_hops_with_proxy_pc(self):
"""
Prepare non direct transitions for each state for State Machine without proxy_pc state.
:return: non direct transitions for each state without proxy_pc state.
"""
state_hops = {
TextualDevice.not_connected: {
AtRemote.unix_local_root: AtRemote.unix_local,
AtRemote.proxy_pc: AtRemote.unix_local,
AtRemote.unix_remote: AtRemote.unix_local,
AtRemote.unix_remote_root: AtRemote.unix_local,
AtRemote.at_remote: AtRemote.unix_local,
},
AtRemote.unix_local: {
AtRemote.unix_remote_root: AtRemote.proxy_pc,
AtRemote.at_remote: AtRemote.proxy_pc,
},
AtRemote.unix_local_root: {
TextualDevice.not_connected: AtRemote.unix_local,
AtRemote.proxy_pc: AtRemote.unix_local,
AtRemote.unix_remote: AtRemote.unix_local,
AtRemote.unix_remote_root: AtRemote.unix_local,
AtRemote.at_remote: AtRemote.unix_local,
},
AtRemote.unix_remote: {
AtRemote.unix_local: AtRemote.proxy_pc,
TextualDevice.not_connected: AtRemote.proxy_pc,
AtRemote.unix_local_root: AtRemote.proxy_pc,
},
AtRemote.unix_remote_root: {
TextualDevice.not_connected: AtRemote.unix_remote,
AtRemote.proxy_pc: AtRemote.unix_remote,
AtRemote.unix_local: AtRemote.unix_remote,
AtRemote.unix_local_root: AtRemote.unix_remote,
AtRemote.at_remote: AtRemote.unix_remote,
},
AtRemote.at_remote: {
TextualDevice.not_connected: UnixRemote.unix_remote,
UnixLocal.unix_local: UnixRemote.unix_remote,
UnixLocal.unix_local_root: UnixRemote.unix_remote,
UnixRemote.unix_remote_root: UnixRemote.unix_remote,
TextualDevice.not_connected: AtRemote.unix_remote,
AtRemote.proxy_pc: AtRemote.unix_remote,
AtRemote.unix_local: AtRemote.unix_remote,
AtRemote.unix_local_root: AtRemote.unix_remote,
AtRemote.unix_remote_root: AtRemote.unix_remote,
},
AtRemote.proxy_pc: {
AtRemote.unix_remote_root: AtRemote.unix_remote,
AtRemote.at_remote: AtRemote.unix_remote,
AtRemote.not_connected: AtRemote.unix_remote,
AtRemote.unix_local_root: AtRemote.unix_local,
}
}
return state_hops

Expand All @@ -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):
"""
Expand All @@ -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')
Expand Down
44 changes: 41 additions & 3 deletions test/device/test_SM_at_remote.py
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
__author__ = 'Grzegorz Latuszek'
__copyright__ = 'Copyright (C) 2020, Nokia'
__email__ = '[email protected]'
__author__ = 'Grzegorz Latuszek, Marcin Usielski'
__copyright__ = 'Copyright (C) 2020-2024, Nokia'
__email__ = '[email protected], [email protected]'

import pytest

Expand All @@ -14,6 +14,13 @@ def test_at_remote_device(device_connection, at_remote_output):
iterate_over_device_states(device=at_remote)


def test_at_remote_device_proxy_pc(device_connection, at_remote_output_proxy_pc):
at_remote = get_device(name="AT_REMOTE_PROXY_PC", connection=device_connection, device_output=at_remote_output_proxy_pc,
test_file_path=__file__)

iterate_over_device_states(device=at_remote)


@pytest.fixture
def at_remote_output():
plink_cmd_string = 'plink -serial COM5 |& awk \'BEGIN {print "COM5> port READY"} {print} END {print "^C"}\''
Expand All @@ -39,3 +46,34 @@ def at_remote_output():
}

return output


@pytest.fixture
def at_remote_output_proxy_pc():
plink_cmd_string = 'plink -serial COM5 |& awk \'BEGIN {print "COM5> port READY"} {print} END {print "^C"}\''
output = {
"UNIX_LOCAL": {
'TERM=xterm-mono ssh -l proxy_pc_login -o ServerAliveInterval=7 -o ServerAliveCountMax=2 proxy_pc_host': 'proxy_pc#',
'su': 'local_root_prompt'
},
"PROXY_PC": {
'TERM=xterm-mono ssh -l remote_login -o ServerAliveInterval=7 -o ServerAliveCountMax=2 remote_host': 'remote#',
'exit': 'moler_bash#',
},
"UNIX_LOCAL_ROOT": {
'exit': 'moler_bash#'
},
"UNIX_REMOTE": {
'exit': 'proxy_pc#',
'su': 'remote_root_prompt',
plink_cmd_string: 'COM5>'
},
"AT_REMOTE": {
'\x03': '^C\nremote#',
},
"UNIX_REMOTE_ROOT": {
'exit': 'remote#',
},
}

return output
Loading

0 comments on commit c1110a5

Please sign in to comment.