From 3704797525be22b5da42c85a6a6468938cbffa86 Mon Sep 17 00:00:00 2001 From: Marcin Usielski <35992110+marcin-usielski@users.noreply.github.com> Date: Wed, 11 Oct 2023 10:36:49 +0200 Subject: [PATCH] SSh connections on Windows (#484) * SSh connections on Windows * dev * dev inv * dev * inv * dev * dev * 156 * info -> debug * version --- CHANGELOG.md | 6 +++ README.md | 2 +- docs/source/conf.py | 2 +- moler/device/proxy_pc2.py | 61 +++++++++++++++++++++++++++-- moler/device/unixremote2.py | 13 +++++- setup.py | 2 +- test/integration/test_devices_SM.py | 2 + 7 files changed, 79 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 25ec59c54..71401b1e3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,9 @@ +## moler 2.15.0 +* ConnectionObserver objects store full exception stack for Python3.5+ +* Support for failure regex for commands for devices based on Unix +* Support values without prefix in byte converter +* Improvement prompt detection for devices using Paramiko for the first ssh + ## moler 2.14.0 * Implementation of internal Quectel command for set NR earfcn lock * Fix typo in `nft` command diff --git a/README.md b/README.md index e95a94660..8cc1f3196 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -[![image](https://img.shields.io/badge/pypi-v2.14.0-blue.svg)](https://pypi.org/project/moler/) +[![image](https://img.shields.io/badge/pypi-v2.15.0-blue.svg)](https://pypi.org/project/moler/) [![image](https://img.shields.io/badge/python-2.7%20%7C%203.6%20%7C%203.7%20%7C%203.8%20%7C%203.9%20%7C%203.10%20%7C%203.11-blue.svg)](https://pypi.org/project/moler/) [![Build Status](https://travis-ci.org/nokia/moler.svg?branch=master)](https://travis-ci.org/nokia/moler) [![Coverage Status](https://coveralls.io/repos/github/nokia/moler/badge.svg?branch=master)](https://coveralls.io/github/nokia/moler?branch=master) diff --git a/docs/source/conf.py b/docs/source/conf.py index 7a2ef7777..e116912e2 100644 --- a/docs/source/conf.py +++ b/docs/source/conf.py @@ -24,7 +24,7 @@ author = 'Nokia' # The short X.Y version -version = '2.14.0' +version = '2.15.0' # The full version, including alpha/beta/rc tags release = 'stable' diff --git a/moler/device/proxy_pc2.py b/moler/device/proxy_pc2.py index dafcdf0ba..7a7623c7a 100644 --- a/moler/device/proxy_pc2.py +++ b/moler/device/proxy_pc2.py @@ -8,12 +8,16 @@ __author__ = 'Michal Ernst, Grzegorz Latuszek, Marcin Usielski' __copyright__ = 'Copyright (C) 2018-2023, Nokia' __email__ = 'michal.ernst@nokia.com, grzegorz.latuszek@nokia.com, marcin.usielski@nokia.com' + +import re +import time import six import abc import platform from moler.device.textualdevice import TextualDevice from moler.device.unixlocal import UnixLocal +from moler.exceptions import MolerException try: from moler.io.raw.terminal import ThreadedTerminal except ImportError: # ThreadedTerminal won't load on Windows @@ -62,6 +66,8 @@ def __init__(self, sm_params, name=None, io_connection=None, io_type=None, varia :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. """ + self._detecting_prompt_cmd = "echo DETECTING PROMPT" + self._prompt_detected = False self._use_local_unix_state = want_local_unix_state(io_type, io_connection) base_state = UNIX_LOCAL if self._use_local_unix_state else NOT_CONNECTED self._use_proxy_pc = self._should_use_proxy_pc(sm_params, PROXY_PC) @@ -72,7 +78,7 @@ 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) - self._prompt_detector_timeout = 0.5 + self._prompt_detector_timeout = 1.9 self._after_open_prompt_detector = None self._warn_about_temporary_life_of_class() @@ -238,7 +244,9 @@ def on_connection_made(self, connection): """ if self._use_local_unix_state: super(ProxyPc2, self).on_connection_made(connection) + self._prompt_detected = True # prompt defined in SM else: + self._prompt_detected = False # prompt not defined in SM self._set_state(PROXY_PC) self._detect_after_open_prompt(self._set_after_open_prompt) @@ -251,25 +259,33 @@ def on_connection_lost(self, connection): self._set_state(NOT_CONNECTED) def _detect_after_open_prompt(self, set_callback): - self._after_open_prompt_detector = Wait4(detect_patterns=[r'^(.+)echo DETECTING PROMPT'], + self._after_open_prompt_detector = Wait4(detect_patterns=[r'^(.+){}'.format(self._detecting_prompt_cmd)], connection=self.io_connection.moler_connection, till_occurs_times=1) detector = self._after_open_prompt_detector detector.add_event_occurred_callback(callback=set_callback, callback_params={"event": detector}) - self.io_connection.moler_connection.sendline("echo DETECTING PROMPT") detector.start(timeout=self._prompt_detector_timeout) + self.io_connection.moler_connection.sendline(self._detecting_prompt_cmd) + # detector.await_done(timeout=self._prompt_detector_timeout) def _set_after_open_prompt(self, event): + self.logger.debug("Old reverse_state_prompts_dict: {}".format( + self._reverse_state_prompts_dict)) occurrence = event.get_last_occurrence() - prompt = occurrence['groups'][0].rstrip() + prompt = re.escape(occurrence['groups'][0].rstrip()) state = self._get_current_state() + self.logger.debug("Found prompt '{}' for '{}'.".format(prompt, state)) with self._state_prompts_lock: self._state_prompts[state] = prompt + self.logger.debug("New prompts: {}".format(self._state_prompts)) self._prepare_reverse_state_prompts_dict() + self.logger.debug("After prepare_reverse_state_prompts_dict: {}".format(self._reverse_state_prompts_dict)) if self._prompts_event is not None: + self.logger.debug("prompts event is not none") self._prompts_event.change_prompts(prompts=self._reverse_state_prompts_dict) + self._prompt_detected = True def _prepare_state_prompts(self): """ @@ -411,3 +427,40 @@ def _get_packages_for_state(self, state, observer): return available[observer] return available + + def get_cmd(self, cmd_name, cmd_params=None, check_state=True, for_state=None): + """ + Returns instance of command connected with the device. + :param cmd_name: name of commands, name of class (without package), for example "cd". + :param cmd_params: dict with command parameters. + :param check_state: if True then before execute of command the state of device will be check if the same + as when command was created. If False the device state is not checked. + :param for_state: if None then command object for current state is returned, otherwise object for for_state is + returned. + :return: Instance of command + """ + if self._prompt_detected is False: + self._detect_prompt_get_cmd() + return super(ProxyPc2, self).get_cmd(cmd_name=cmd_name, cmd_params=cmd_params, + check_state=check_state, + for_state=for_state) + + def _detect_prompt_get_cmd(self): + self.logger.debug("get_cmd was called but prompt has not been detected yet.") + for try_nr in range(1, 10, 1): + if self._after_open_prompt_detector is None or self._after_open_prompt_detector.running() is False: + self._detect_after_open_prompt(self._set_after_open_prompt) + while time.time() - self._after_open_prompt_detector.life_status.start_time < self._prompt_detector_timeout: + time.sleep(0.1) + if self._prompt_detected is True: + break + self.io_connection.moler_connection.sendline(self._detecting_prompt_cmd) + self.logger.debug("get_cmd is canceling prompt detector.") + self._after_open_prompt_detector.cancel() + if self._prompt_detected is True: + break + + self._after_open_prompt_detector.cancel() + self._after_open_prompt_detector = None + if self._prompt_detected is False: + raise MolerException("Device {} cannot detect prompt!".format(self.public_name)) diff --git a/moler/device/unixremote2.py b/moler/device/unixremote2.py index 50e8db263..c15dd7648 100644 --- a/moler/device/unixremote2.py +++ b/moler/device/unixremote2.py @@ -381,15 +381,24 @@ def on_connection_made(self, connection): self._detect_after_open_prompt(self._set_after_open_prompt) def _set_after_open_prompt(self, event): + self.logger.debug("UnixRemote2 reverse_state_prompts_dict: {}".format( + self._reverse_state_prompts_dict)) occurrence = event.get_last_occurrence() - prompt = occurrence['groups'][0] + prompt = occurrence['groups'][0].rstrip() state = self._get_current_state() with self._state_prompts_lock: - self._state_prompts[state] = re.escape(prompt.rstrip()) + self._state_prompts[state] = re.escape(prompt) + self.logger.debug("Found prompt '{}' for '{}'.".format(prompt, state)) if state == UNIX_REMOTE: self._update_depending_on_ux_prompt() elif state == PROXY_PC: self._update_depending_on_proxy_prompt() + self.logger.debug("UnixRemote2. updated _reverse_state_prompts_dict: {}".format( + self._reverse_state_prompts_dict)) + if self._prompts_event is not None: + self.logger.debug("prompts event is not none") + self._prompts_event.change_prompts(prompts=self._reverse_state_prompts_dict) + self._prompt_detected = True @mark_to_call_base_class_method_with_same_name def _prepare_state_prompts_with_proxy_pc(self): diff --git a/setup.py b/setup.py index f63d69c8a..a13491101 100644 --- a/setup.py +++ b/setup.py @@ -13,7 +13,7 @@ setup( name='moler', # Required - version='2.14.0', # Required + version='2.15.0', # Required description='Moler is library to help in building automated tests', # Required long_description=long_description, # Optional long_description_content_type='text/markdown', # Optional (see note above) diff --git a/test/integration/test_devices_SM.py b/test/integration/test_devices_SM.py index aa3cf2f81..147ba2986 100644 --- a/test/integration/test_devices_SM.py +++ b/test/integration/test_devices_SM.py @@ -64,6 +64,7 @@ def test_unix_remote_with_sshshell_via_proxy_pc(loaded_unix_remote_config, proxy 'login': 'sshproxy', 'password': 'proxy_password'}, connection_hops=proxypc2uxroot_connection_hops) + dev._prompt_detected = True assert dev._use_proxy_pc is True assert dev.current_state == "PROXY_PC" dev.goto_state("UNIX_REMOTE") @@ -130,6 +131,7 @@ def test_adb_remote_with_sshshell_via_proxy_pc(loaded_adb_device_config, proxypc 'login': 'sshproxy', 'password': 'proxy_password'}, connection_hops=proxypc2adbshell_connection_hops) + dev._prompt_detected = True assert dev._use_proxy_pc is True assert dev.current_state == "PROXY_PC" dev.goto_state("UNIX_REMOTE")