Skip to content

Commit

Permalink
AdbRemote3
Browse files Browse the repository at this point in the history
  • Loading branch information
@marcin-usielski
marcin-usielski committed Oct 30, 2024
1 parent c3b1026 commit d0be6a8
Show file tree
Hide file tree
Showing 8 changed files with 220 additions and 110 deletions.
5 changes: 1 addition & 4 deletions moler/device/adbremote.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -404,7 +404,6 @@ def _prepare_state_hops_without_proxy_pc(self):
AdbRemote.adb_shell_root: AdbRemote.unix_local,
},
AdbRemote.unix_remote: {
AdbRemote.unix_local_root: AdbRemote.unix_local,
AdbRemote.adb_shell_root: AdbRemote.adb_shell,
},
AdbRemote.unix_remote_root: {
Expand Down Expand Up @@ -459,8 +458,6 @@ def _prepare_state_hops_with_proxy_pc(self):
AdbRemote.adb_shell_root: AdbRemote.unix_local,
},
AdbRemote.unix_remote: {
AdbRemote.unix_local: AdbRemote.proxy_pc,
AdbRemote.unix_local_root: AdbRemote.proxy_pc,
AdbRemote.adb_shell_root: AdbRemote.adb_shell,
},
AdbRemote.unix_remote_root: {
Expand Down Expand Up @@ -490,7 +487,7 @@ def _prepare_state_hops_with_proxy_pc(self):
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_remote,
AdbRemote.not_connected: AdbRemote.unix_local,
AdbRemote.unix_local_root: AdbRemote.unix_local,
}
}
Expand Down
164 changes: 81 additions & 83 deletions moler/device/adbremote3.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@

import logging

from moler.device.textualdevice import TextualDevice
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
Expand Down Expand Up @@ -82,9 +81,8 @@ 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.
"""
print(f"\n\n****************\n{self.name} AdbRemote3._get_default_sm_configuration_with_proxy_pc\n\n")
config = {
TextualDevice.connection_hops: {
AdbRemote3.connection_hops: {
AdbRemote3.unix_remote: { # from
AdbRemote3.adb_shell: { # to
"execute_command": "adb_shell",
Expand Down Expand Up @@ -134,79 +132,46 @@ def _get_default_sm_configuration_with_proxy_pc(self):
}
return config

@mark_to_call_base_class_method_with_same_name
def _prepare_state_hops_with_proxy_pc(self):
def _overwrite_prompts(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.
Overwrite prompts for some states to easily configure the SM.
"""
state_hops = {
TextualDevice.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: {
TextualDevice.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: {
TextualDevice.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: {
TextualDevice.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: {
TextualDevice.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.unix_remote_root: AdbRemote3.unix_remote,
AdbRemote3.adb_shell: AdbRemote3.unix_remote,
AdbRemote3.adb_shell_root: AdbRemote3.unix_remote,
AdbRemote3.not_connected: AdbRemote3.unix_remote,
AdbRemote3.unix_local_root: AdbRemote3.unix_local,
}
}
return state_hops
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[TextualDevice.connection_hops]
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"]
Expand All @@ -233,7 +198,7 @@ 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]
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"]
Expand Down Expand Up @@ -261,7 +226,7 @@ def _serial_number(self):
:return: serial_number.
"""
hops_config = self._configurations[TextualDevice.connection_hops]
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
Expand All @@ -281,7 +246,7 @@ 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[TextualDevice.connection_hops]
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"]
Expand All @@ -301,7 +266,7 @@ 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_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"]
Expand All @@ -322,7 +287,7 @@ def _prepare_state_hops_with_proxy_pc(self):
:return: non direct transitions for each state without proxy_pc state.
"""
state_hops = {
TextualDevice.not_connected: {
AdbRemote3.not_connected: {
AdbRemote3.unix_local_root: AdbRemote3.unix_local,
AdbRemote3.proxy_pc: AdbRemote3.unix_local,
AdbRemote3.unix_remote: AdbRemote3.unix_local,
Expand All @@ -336,7 +301,7 @@ def _prepare_state_hops_with_proxy_pc(self):
AdbRemote3.adb_shell_root: AdbRemote3.proxy_pc,
},
AdbRemote3.unix_local_root: {
TextualDevice.not_connected: AdbRemote3.unix_local,
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,
Expand All @@ -349,34 +314,31 @@ def _prepare_state_hops_with_proxy_pc(self):
AdbRemote3.adb_shell_root: AdbRemote3.adb_shell,
},
AdbRemote3.unix_remote_root: {
TextualDevice.not_connected: AdbRemote3.unix_remote,
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: {
TextualDevice.not_connected: AdbRemote3.unix_remote,
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: {
TextualDevice.not_connected: AdbRemote3.adb_shell,
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.unix_remote_root: AdbRemote3.unix_remote,
AdbRemote3.adb_shell: AdbRemote3.unix_remote,
AdbRemote3.adb_shell_root: AdbRemote3.unix_remote,
AdbRemote3.not_connected: AdbRemote3.unix_remote,
AdbRemote3.unix_local_root: AdbRemote3.unix_local,
}
}
return state_hops
Expand All @@ -389,7 +351,7 @@ def _configure_state_machine(self, sm_params):
"""
super(AdbRemote3, self)._configure_state_machine(sm_params)

hops_config = self._configurations[TextualDevice.connection_hops]
hops_config = self._configurations[AdbRemote3.connection_hops]

# copy prompt for ADB_SHELL/exit from UNIX_LOCAL/ssh
if self._use_proxy_pc:
Expand Down Expand Up @@ -424,12 +386,48 @@ def _get_packages_for_state(self, state, observer):

if not available:
if (state == AdbRemote3.adb_shell) or (state == AdbRemote3.adb_shell_root):
available = {TextualDevice.cmds: ['moler.cmd.unix'],
TextualDevice.events: ['moler.events.shared']}
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 == TextualDevice.cmds:
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
2 changes: 0 additions & 2 deletions moler/device/proxy_pc.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,8 +41,6 @@ def __init__(self, sm_params, name=None, io_connection=None, io_type=None, varia
io_constructor_kwargs=io_constructor_kwargs,
sm_params=sm_params, initial_state=initial_state,
lazy_cmds_events=lazy_cmds_events)
from pprint import pformat
print(f"ProxyPc {self.name} sm config: {pformat(self._configurations)}")

def _get_default_sm_configuration(self):
"""
Expand Down
2 changes: 0 additions & 2 deletions moler/device/proxy_pc3.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -44,8 +44,6 @@ def __init__(self, sm_params, name=None, io_connection=None, io_type=None, varia
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.")
from pprint import pformat
print(f"ProxyPc3 {self.name} sm config: {pformat(self._configurations)}")

def _prepare_sm_data(self, sm_params):
self._prepare_dicts_for_sm(sm_params=sm_params)
Expand Down
1 change: 0 additions & 1 deletion moler/device/unixremote3.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -139,7 +139,6 @@ 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.
"""
print(f"\n\n****************\n{self.name} ProxyPC3._get_default_sm_configuration_with_proxy_pc\n\n")
config = {
UnixRemote3.connection_hops: {
UnixRemote3.proxy_pc: { # from
Expand Down
4 changes: 4 additions & 0 deletions moler/helpers.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
from types import FunctionType, MethodType

from six import string_types, integer_types
from moler.exceptions import MolerException


class ClassProperty(property):
Expand Down Expand Up @@ -666,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]
Expand Down
Loading

0 comments on commit d0be6a8

Please sign in to comment.