From 51089c169b6f6b1e12b7c0a67823ae43fbc2b1a3 Mon Sep 17 00:00:00 2001 From: Marcin Usielski Date: Fri, 12 Jan 2024 14:00:28 +0100 Subject: [PATCH 1/9] Command touch --- moler/cmd/unix/genericunix.py | 6 ++-- moler/cmd/unix/touch.py | 51 +++++++++++++++++++++++++++++++++ test/cmd/unix/test_cmd_touch.py | 21 ++++++++++++++ 3 files changed, 76 insertions(+), 2 deletions(-) create mode 100644 moler/cmd/unix/touch.py create mode 100644 test/cmd/unix/test_cmd_touch.py diff --git a/moler/cmd/unix/genericunix.py b/moler/cmd/unix/genericunix.py index 67c96f3b3..42def36f4 100644 --- a/moler/cmd/unix/genericunix.py +++ b/moler/cmd/unix/genericunix.py @@ -4,7 +4,7 @@ """ __author__ = 'Marcin Usielski' -__copyright__ = 'Copyright (C) 2018-2023, Nokia' +__copyright__ = 'Copyright (C) 2018-2024, Nokia' __email__ = 'marcin.usielski@nokia.com' import re @@ -19,7 +19,9 @@ 'No such file or directory', 'running it may require superuser privileges', 'Cannot find device', - 'Input/output error'] + 'Input/output error', + 'Permission denied', + ] r_cmd_failure_cause_alternatives = r'{}'.format("|".join(cmd_failure_causes)) diff --git a/moler/cmd/unix/touch.py b/moler/cmd/unix/touch.py new file mode 100644 index 000000000..63f29bb4f --- /dev/null +++ b/moler/cmd/unix/touch.py @@ -0,0 +1,51 @@ +from moler.cmd.unix.genericunix import GenericUnixCommand + +class Touch(GenericUnixCommand): + + def __init__(self, connection, path, prompt=None, newline_chars=None, runner=None, options=None): + """ + Unix command touch. + + :param connection: Moler connection to device, terminal when command is executed. + :param path: path to file to be created + :param prompt: prompt (on system where command runs). + :param newline_chars: Characters to split lines - list. + :param runner: Runner to run command. + :param options: Options of unix touch command + """ + super(Touch, self).__init__(connection=connection, prompt=prompt, newline_chars=newline_chars, runner=runner) + self.ret_required = False + self.options = options + self.path = path + + def build_command_string(self): + """ + Builds command string from parameters passed to object. + + :return: String representation of command to send over connection to device. + """ + cmd = "touch" + if self.options: + cmd = "{} {}".format(cmd, self.options) + cmd = "{} {}".format(cmd, self.path) + return cmd + + +COMMAND_OUTPUT = """touch file1.txt +moler_bash#""" + + +COMMAND_KWARGS = {'path': 'file1.txt'} + + +COMMAND_RESULT = {} + + +COMMAND_OUTPUT_options = """touch -a file1.txt +moler_bash#""" + + +COMMAND_KWARGS_options = {'path': 'file1.txt', 'options': '-a'} + + +COMMAND_RESULT_options = {} diff --git a/test/cmd/unix/test_cmd_touch.py b/test/cmd/unix/test_cmd_touch.py new file mode 100644 index 000000000..d05891ff0 --- /dev/null +++ b/test/cmd/unix/test_cmd_touch.py @@ -0,0 +1,21 @@ +# -*- coding: utf-8 -*- +""" +Touch command test module. +""" + +__author__ = 'Marcin Usielski' +__copyright__ = 'Copyright (C) 2024, Nokia' +__email__ = 'marcin.usielski@nokia.com' + +import pytest +from moler.cmd.unix.touch import Touch +from moler.exceptions import CommandFailure + + +def test_touch_cannot_create(buffer_connection): + touch_cmd = Touch(connection=buffer_connection.moler_connection, path="file.asc") + assert "touch file.asc" == touch_cmd.command_string + command_output = "touch file.asc\ntouch: cannot touch 'file.asc': Permission denied\nmoler_bash#" + buffer_connection.remote_inject_response([command_output]) + with pytest.raises(CommandFailure): + touch_cmd() From 2feceb41d3954230a525cb491a8a8a3a55696466 Mon Sep 17 00:00:00 2001 From: Marcin Usielski Date: Fri, 12 Jan 2024 15:42:52 +0100 Subject: [PATCH 2/9] add_failure_indication [AI] --- moler/cmd/unix/genericunix.py | 15 ++++++++++++++- moler/cmd/unix/touch.py | 12 ++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/moler/cmd/unix/genericunix.py b/moler/cmd/unix/genericunix.py index 42def36f4..9a38b3de4 100644 --- a/moler/cmd/unix/genericunix.py +++ b/moler/cmd/unix/genericunix.py @@ -20,7 +20,6 @@ 'running it may require superuser privileges', 'Cannot find device', 'Input/output error', - 'Permission denied', ] r_cmd_failure_cause_alternatives = r'{}'.format("|".join(cmd_failure_causes)) @@ -75,3 +74,17 @@ def _decode_line(self, line): if self.remove_all_known_special_chars_from_terminal_output: line = remove_all_known_special_chars(line) return line + + def add_failure_indication(self, indication): + """ + Add failure indication to command. + + :param indication: String or regexp with ndication of failure. + :return: None + """ + if self._re_fail is None: + new_indication = indication + else: + current_indications = self._re_fail.pattern + new_indication = r'{}|{}'.format(current_indications, indication) + self._re_fail = re.compile(new_indication, re.IGNORECASE) diff --git a/moler/cmd/unix/touch.py b/moler/cmd/unix/touch.py index 63f29bb4f..d52c8b6f9 100644 --- a/moler/cmd/unix/touch.py +++ b/moler/cmd/unix/touch.py @@ -1,5 +1,16 @@ +# -*- coding: utf-8 -*- +""" +Command touch module +""" + + +__author__ = 'Marcin Usielski' +__copyright__ = 'Copyright (C) 2018-2024, Nokia' +__email__ = 'marcin.usielski@nokia.com' + from moler.cmd.unix.genericunix import GenericUnixCommand + class Touch(GenericUnixCommand): def __init__(self, connection, path, prompt=None, newline_chars=None, runner=None, options=None): @@ -17,6 +28,7 @@ def __init__(self, connection, path, prompt=None, newline_chars=None, runner=Non self.ret_required = False self.options = options self.path = path + self.add_failure_indication('touch: cannot touch') def build_command_string(self): """ From 9c855a6766f7def977613f9111e61fbd8abed6dd Mon Sep 17 00:00:00 2001 From: Marcin Usielski Date: Fri, 12 Jan 2024 15:53:31 +0100 Subject: [PATCH 3/9] test for coverage [AI] --- test/cmd/unix/test_cmd_touch.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/test/cmd/unix/test_cmd_touch.py b/test/cmd/unix/test_cmd_touch.py index d05891ff0..27db78f8b 100644 --- a/test/cmd/unix/test_cmd_touch.py +++ b/test/cmd/unix/test_cmd_touch.py @@ -19,3 +19,14 @@ def test_touch_cannot_create(buffer_connection): buffer_connection.remote_inject_response([command_output]) with pytest.raises(CommandFailure): touch_cmd() + + +def test_touch_read_only(buffer_connection): + touch_cmd = Touch(connection=buffer_connection.moler_connection, path="file.asc") + touch_cmd._re_fail = None + touch_cmd.add_failure_indication("Read-only file system") + assert "touch file.asc" == touch_cmd.command_string + command_output = "touch file.asc\ntouch: cannot touch 'file.asc': Read-only file system\nmoler_bash#" + buffer_connection.remote_inject_response([command_output]) + with pytest.raises(CommandFailure): + touch_cmd() From e2f3f87caebe2ed7a586d8a3e20eb5ac5feb06a9 Mon Sep 17 00:00:00 2001 From: Marcin Usielski Date: Fri, 12 Jan 2024 16:16:35 +0100 Subject: [PATCH 4/9] new test [AI] --- test/cmd/unix/test_cmd_touch.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/test/cmd/unix/test_cmd_touch.py b/test/cmd/unix/test_cmd_touch.py index 27db78f8b..fa8dc8ff4 100644 --- a/test/cmd/unix/test_cmd_touch.py +++ b/test/cmd/unix/test_cmd_touch.py @@ -22,6 +22,15 @@ def test_touch_cannot_create(buffer_connection): def test_touch_read_only(buffer_connection): + touch_cmd = Touch(connection=buffer_connection.moler_connection, path="file.asc") + assert "touch file.asc" == touch_cmd.command_string + command_output = "touch file.asc\ntouch: cannot touch 'file.asc': Read-only file system\nmoler_bash#" + buffer_connection.remote_inject_response([command_output]) + with pytest.raises(CommandFailure): + touch_cmd() + + +def test_touch_read_only_remove(buffer_connection): touch_cmd = Touch(connection=buffer_connection.moler_connection, path="file.asc") touch_cmd._re_fail = None touch_cmd.add_failure_indication("Read-only file system") From 3bf6851cfa8959d5aeb5ed9cbfc7ff2d4e26a2e1 Mon Sep 17 00:00:00 2001 From: Marcin Usielski Date: Tue, 16 Jan 2024 09:29:18 +0100 Subject: [PATCH 5/9] CommandTextualGeneric [AI] --- moler/cmd/commandtextualgeneric.py | 28 +++++++++++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/moler/cmd/commandtextualgeneric.py b/moler/cmd/commandtextualgeneric.py index 6a0ebb4c5..48bc0b4b9 100644 --- a/moler/cmd/commandtextualgeneric.py +++ b/moler/cmd/commandtextualgeneric.py @@ -4,7 +4,7 @@ """ __author__ = 'Marcin Usielski, Michal Ernst' -__copyright__ = 'Copyright (C) 2018-2023, Nokia' +__copyright__ = 'Copyright (C) 2018-2024, Nokia' __email__ = 'marcin.usielski@nokia.com, michal.ernst@nokia.com' import abc @@ -13,6 +13,7 @@ import six +from moler.exceptions import CommandFailure from moler.cmd import RegexHelper from moler.command import Command from moler.helpers import regexp_without_anchors @@ -81,6 +82,7 @@ def __init__(self, connection, prompt=None, newline_chars=None, runner=None): # False to consider also chunks. self.enter_on_prompt_without_anchors = False # Set True to try to match prompt in line without ^ and $. self.debug_data_received = False # Set True to log as hex all data received by command in data_received + self.re_failure = None # Regex to failure the command if it occurrs in the command output if not self._newline_chars: self._newline_chars = CommandTextualGeneric._default_newline_chars @@ -338,6 +340,7 @@ def on_new_line(self, line, is_full_line): " nor empty.") else: self._break_exec_on_regex(line=line, is_full_line=is_full_line) + self._failure_indiction(line=line, is_full_line=is_full_line) def is_end_of_cmd_output(self, line): """ @@ -527,3 +530,26 @@ def __str__(self): expected_prompt = self._re_prompt.pattern # having expected prompt visible simplifies troubleshooting return "{}, prompt_regex:r'{}')".format(base_str[:-1], expected_prompt) + + def _is_failure_indication(self, line, is_full_line): + """ + Checks if the given line is a failure indication. + + :param line: The line to check. + :param is_full_line: Indicates if the line is a full line or a partial line. + :return: True if the line is a failure indication, False otherwise. + """ + if self.re_failure is not None and is_full_line and self.is_failure_indication(line, self.re_failure): + return True + return False + + def _failure_indiction(self, line, is_full_line): + """ + Set CommandException if failure string in the line. + + :param line: The line to check. + :param is_full_line: Indicates if the line is a full line or a partial line. + :return: None + """ + if self._is_failure_indication(line=line, is_full_line=is_full_line): + self.set_exception(CommandFailure(self, "command failed in line '{}'".format(line))) From 1ac2ae01111ee063cbd1ae9a943ad340cf0dff24 Mon Sep 17 00:00:00 2001 From: Marcin Usielski Date: Tue, 16 Jan 2024 13:32:17 +0100 Subject: [PATCH 6/9] fix the commands --- moler/cmd/commandtextualgeneric.py | 29 ++++++++--- moler/cmd/pdu_aten/pdu/generic_pdu.py | 27 +---------- moler/cmd/unix/cat.py | 4 +- moler/cmd/unix/genericunix.py | 70 ++++++++++++++------------- moler/cmd/unix/tail_latest_file.py | 15 +++--- test/cmd/unix/test_cmd_lxc_info.py | 15 +++--- test/cmd/unix/test_cmd_touch.py | 2 +- 7 files changed, 81 insertions(+), 81 deletions(-) diff --git a/moler/cmd/commandtextualgeneric.py b/moler/cmd/commandtextualgeneric.py index 48bc0b4b9..9681ab11f 100644 --- a/moler/cmd/commandtextualgeneric.py +++ b/moler/cmd/commandtextualgeneric.py @@ -82,7 +82,7 @@ def __init__(self, connection, prompt=None, newline_chars=None, runner=None): # False to consider also chunks. self.enter_on_prompt_without_anchors = False # Set True to try to match prompt in line without ^ and $. self.debug_data_received = False # Set True to log as hex all data received by command in data_received - self.re_failure = None # Regex to failure the command if it occurrs in the command output + self.re_fail = None # Regex to failure the command if it occurrs in the command output if not self._newline_chars: self._newline_chars = CommandTextualGeneric._default_newline_chars @@ -251,6 +251,7 @@ def data_received(self, data, recv_time): self.logger.debug("{} is done".format(self)) break except UnicodeDecodeError as ex: + print("UnicodeDecodeError") if self._ignore_unicode_errors: self._log(lvl=logging.WARNING, msg="Processing data from '{}' with unicode problem: '{}'.".format(self, ex)) @@ -260,6 +261,7 @@ def data_received(self, data, recv_time): msg="Processing data from '{}' raised: '{}'.".format(self, ex)) raise ex except Exception as ex: # pragma: no cover # log it just to catch that rare hanging thread issue + print("Exception {}".format(ex)) self._log(lvl=logging.WARNING, msg="Processing data from '{}' raised: '{}'.".format(self, ex)) raise ex @@ -339,8 +341,8 @@ def on_new_line(self, line, is_full_line): msg="Found candidate for final prompt but current ret is None or empty, required not None" " nor empty.") else: + self.failure_indiction(line=line, is_full_line=is_full_line) self._break_exec_on_regex(line=line, is_full_line=is_full_line) - self._failure_indiction(line=line, is_full_line=is_full_line) def is_end_of_cmd_output(self, line): """ @@ -531,7 +533,7 @@ def __str__(self): # having expected prompt visible simplifies troubleshooting return "{}, prompt_regex:r'{}')".format(base_str[:-1], expected_prompt) - def _is_failure_indication(self, line, is_full_line): + def is_failure_indication(self, line, is_full_line): """ Checks if the given line is a failure indication. @@ -539,11 +541,12 @@ def _is_failure_indication(self, line, is_full_line): :param is_full_line: Indicates if the line is a full line or a partial line. :return: True if the line is a failure indication, False otherwise. """ - if self.re_failure is not None and is_full_line and self.is_failure_indication(line, self.re_failure): + if self.re_fail is not None and is_full_line and self._regex_helper.search_compiled( + compiled=self.re_fail, string=line): return True return False - def _failure_indiction(self, line, is_full_line): + def failure_indiction(self, line, is_full_line): """ Set CommandException if failure string in the line. @@ -551,5 +554,19 @@ def _failure_indiction(self, line, is_full_line): :param is_full_line: Indicates if the line is a full line or a partial line. :return: None """ - if self._is_failure_indication(line=line, is_full_line=is_full_line): + if self.is_failure_indication(line=line, is_full_line=is_full_line): self.set_exception(CommandFailure(self, "command failed in line '{}'".format(line))) + + def add_failure_indication(self, indication): + """ + Add failure indication to command. + + :param indication: String or regexp with ndication of failure. + :return: None + """ + if self.re_fail is None: + new_indication = indication + else: + current_indications = self.re_fail.pattern + new_indication = r'{}|{}'.format(current_indications, indication) + self.re_fail = re.compile(new_indication, re.IGNORECASE) diff --git a/moler/cmd/pdu_aten/pdu/generic_pdu.py b/moler/cmd/pdu_aten/pdu/generic_pdu.py index dbe088b16..d00178f34 100644 --- a/moler/cmd/pdu_aten/pdu/generic_pdu.py +++ b/moler/cmd/pdu_aten/pdu/generic_pdu.py @@ -10,7 +10,7 @@ from moler.exceptions import CommandFailure __author__ = 'Marcin Usielski' -__copyright__ = 'Copyright (C) 2020, Nokia' +__copyright__ = 'Copyright (C) 2020-2024, Nokia' __email__ = 'marcin.usielski@nokia.com' cmd_failure_causes = ['Not Support', @@ -32,27 +32,4 @@ def __init__(self, connection, prompt=None, newline_chars=None, runner=None): """ super(GenericPdu, self).__init__(connection=connection, prompt=prompt, newline_chars=newline_chars, runner=runner) - self._re_fail = re.compile(r_cmd_failure_cause_alternatives, re.IGNORECASE) - - def on_new_line(self, line, is_full_line): - """ - Method to parse command output. Will be called after line with command echo. - Write your own implementation but don't forget to call on_new_line from base class - - :param line: Line to parse, new lines are trimmed - :param is_full_line: False for chunk of line; True on full line (NOTE: new line character removed) - :return: None - """ - if is_full_line and self.is_failure_indication(line): - self.set_exception(CommandFailure(self, "command failed in line '{}'".format(line))) - return super(GenericPdu, self).on_new_line(line, is_full_line) - - def is_failure_indication(self, line): - """ - Method to detect if passed line contains part indicating failure of command - - :param line: Line from command output on device - :return: Match object if find regex in line, None otherwise. - """ - ret = self._regex_helper.search_compiled(self._re_fail, line) if self._re_fail else None - return ret + self.re_fail = re.compile(r_cmd_failure_cause_alternatives, re.IGNORECASE) diff --git a/moler/cmd/unix/cat.py b/moler/cmd/unix/cat.py index 9dfc79185..7dab873cc 100644 --- a/moler/cmd/unix/cat.py +++ b/moler/cmd/unix/cat.py @@ -81,14 +81,14 @@ def set_exception(self, exception): return return super(Cat, self).set_exception(exception) - def is_failure_indication(self, line): + def is_failure_indication(self, line, is_full_line): """ Method to detect if passed line contains part indicating failure of command. :param line: Line from command output on device :return: Match object if find regex in line, None otherwise. """ - return self._regex_helper.search_compiled(Cat._re_parse_error, line) + return self._regex_helper.search_compiled(Cat._re_parse_error, line) is not None def _parse_line(self, line): if not line == "": diff --git a/moler/cmd/unix/genericunix.py b/moler/cmd/unix/genericunix.py index 9a38b3de4..70d1afdae 100644 --- a/moler/cmd/unix/genericunix.py +++ b/moler/cmd/unix/genericunix.py @@ -26,7 +26,7 @@ @six.add_metaclass(abc.ABCMeta) class GenericUnixCommand(CommandTextualGeneric): - _re_fail = re.compile(r_cmd_failure_cause_alternatives, re.IGNORECASE) + # _re_fail = re.compile(r_cmd_failure_cause_alternatives, re.IGNORECASE) def __init__(self, connection, prompt=None, newline_chars=None, runner=None): """ @@ -38,31 +38,33 @@ def __init__(self, connection, prompt=None, newline_chars=None, runner=None): super(GenericUnixCommand, self).__init__(connection=connection, prompt=prompt, newline_chars=newline_chars, runner=runner) self.remove_all_known_special_chars_from_terminal_output = True + self.re_fail = re.compile(r_cmd_failure_cause_alternatives, re.IGNORECASE) + # self._re_fail = None - def on_new_line(self, line, is_full_line): - """ - Method to parse command output. Will be called after line with command echo. - Write your own implementation but don't forget to call on_new_line from base class + # def on_new_line(self, line, is_full_line): + # """ + # Method to parse command output. Will be called after line with command echo. + # Write your own implementation but don't forget to call on_new_line from base class - :param line: Line to parse, new lines are trimmed - :param is_full_line: False for chunk of line; True on full line (NOTE: new line character removed) - :return: None - """ - if is_full_line and self.is_failure_indication(line) is not None: - self.set_exception(CommandFailure(self, "command failed in line '{}'".format(line))) - return super(GenericUnixCommand, self).on_new_line(line, is_full_line) + # :param line: Line to parse, new lines are trimmed + # :param is_full_line: False for chunk of line; True on full line (NOTE: new line character removed) + # :return: None + # """ + # if is_full_line and self.is_failure_indication(line) is not None: + # self.set_exception(CommandFailure(self, "command failed in line '{}'".format(line))) + # return super(GenericUnixCommand, self).on_new_line(line, is_full_line) - def is_failure_indication(self, line): - """ - Method to detect if passed line contains part indicating failure of command + # def is_failure_indication(self, line): + # """ + # Method to detect if passed line contains part indicating failure of command - :param line: Line from command output on device - :return: Match object if find regex in line, None otherwise. - """ - if self._re_fail: - return self._regex_helper.search_compiled(compiled=self._re_fail, - string=line) - return None + # :param line: Line from command output on device + # :return: Match object if find regex in line, None otherwise. + # """ + # if self._re_fail: + # return self._regex_helper.search_compiled(compiled=self._re_fail, + # string=line) + # return None def _decode_line(self, line): """ @@ -75,16 +77,16 @@ def _decode_line(self, line): line = remove_all_known_special_chars(line) return line - def add_failure_indication(self, indication): - """ - Add failure indication to command. + # def add_failure_indication(self, indication): + # """ + # Add failure indication to command. - :param indication: String or regexp with ndication of failure. - :return: None - """ - if self._re_fail is None: - new_indication = indication - else: - current_indications = self._re_fail.pattern - new_indication = r'{}|{}'.format(current_indications, indication) - self._re_fail = re.compile(new_indication, re.IGNORECASE) + # :param indication: String or regexp with ndication of failure. + # :return: None + # """ + # if self._re_fail is None: + # new_indication = indication + # else: + # current_indications = self._re_fail.pattern + # new_indication = r'{}|{}'.format(current_indications, indication) + # self._re_fail = re.compile(new_indication, re.IGNORECASE) diff --git a/moler/cmd/unix/tail_latest_file.py b/moler/cmd/unix/tail_latest_file.py index f9de4c214..e43dfcfd3 100644 --- a/moler/cmd/unix/tail_latest_file.py +++ b/moler/cmd/unix/tail_latest_file.py @@ -7,9 +7,9 @@ import time from moler.cmd.unix.genericunix import GenericUnixCommand -__author__ = 'Tomasz Krol' -__copyright__ = 'Copyright (C) 2020, Nokia' -__email__ = 'tomasz.krol@nokia.com' +__author__ = 'Tomasz Krol, Marcin Usielski' +__copyright__ = 'Copyright (C) 2020-2024, Nokia' +__email__ = 'tomasz.krol@nokia.com, marcin.usielski@nokia.com' class TailLatestFile(GenericUnixCommand): @@ -89,19 +89,20 @@ def on_new_line(self, line, is_full_line): self._first_line_time = time.time() super(TailLatestFile, self).on_new_line(line=line, is_full_line=is_full_line) - def is_failure_indication(self, line): + def is_failure_indication(self, line, is_full_line): """ Check if line has info about failure indication. :param line: Line from device - :return: None if line does not match regex with failure, Match object if line matches the failure regex. + :param is_full_line: True if line had new line chars, False otherwise + :return: False if line does not match regex with failure, True if line matches the failure regex. """ if self._check_failure_indication: if time.time() - self._first_line_time < self.time_for_failure: - return self._regex_helper.search_compiled(self._re_fail, line) + return self._regex_helper.search_compiled(self.re_fail, line) is not None else: self._check_failure_indication = False # do not check time for future output. It's too late already. - return None + return False COMMAND_OUTPUT = r""" diff --git a/test/cmd/unix/test_cmd_lxc_info.py b/test/cmd/unix/test_cmd_lxc_info.py index eec7d29f9..140f50cca 100644 --- a/test/cmd/unix/test_cmd_lxc_info.py +++ b/test/cmd/unix/test_cmd_lxc_info.py @@ -10,8 +10,8 @@ """ __author__ = 'Piotr Frydrych, Marcin Usielski' -__copyright__ = 'Copyright (C) 2019-2020, Nokia' -__email__ = 'piotr.frydrych@nokia.com' +__copyright__ = 'Copyright (C) 2019-2024, Nokia' +__email__ = 'piotr.frydrych@nokia.com, marcin.usielski@nokia.com' from moler.cmd.unix.lxc_info import LxcInfo @@ -24,10 +24,12 @@ def test_lxc_info_raise_command_error(buffer_connection, command_output_and_expe buffer_connection.remote_inject_response(data) cmd = LxcInfo(name="0xe049", connection=buffer_connection.moler_connection, options="-z") from time import time - print("test_lxc_info_raise_command_error S {}".format(time())) + start_time = time() with pytest.raises(CommandFailure): cmd() - print("test_lxc_info_raise_command_error E {}".format(time())) + end_time = time() + assert (end_time - start_time) < cmd.timeout + def test_lxc_info_raise_container_name_error(buffer_connection, container_name_error_and_expected_result): @@ -36,10 +38,11 @@ def test_lxc_info_raise_container_name_error(buffer_connection, container_name_e cmd = LxcInfo(name="0xe0499", connection=buffer_connection.moler_connection) cmd.terminating_timeout = 0 from time import time - print("test_lxc_info_raise_container_name_error S {}".format(time())) + start_time = time() with pytest.raises(CommandFailure): cmd() - print("test_lxc_info_raise_container_name_error E {}".format(time())) + end_time = time() + assert (end_time - start_time) < cmd.timeout @pytest.fixture() diff --git a/test/cmd/unix/test_cmd_touch.py b/test/cmd/unix/test_cmd_touch.py index fa8dc8ff4..f04fddf9d 100644 --- a/test/cmd/unix/test_cmd_touch.py +++ b/test/cmd/unix/test_cmd_touch.py @@ -32,7 +32,7 @@ def test_touch_read_only(buffer_connection): def test_touch_read_only_remove(buffer_connection): touch_cmd = Touch(connection=buffer_connection.moler_connection, path="file.asc") - touch_cmd._re_fail = None + touch_cmd.re_fail = None touch_cmd.add_failure_indication("Read-only file system") assert "touch file.asc" == touch_cmd.command_string command_output = "touch file.asc\ntouch: cannot touch 'file.asc': Read-only file system\nmoler_bash#" From 3dc13be25a0d610615f735e15a396b2f11efa070 Mon Sep 17 00:00:00 2001 From: Marcin Usielski Date: Tue, 16 Jan 2024 13:41:48 +0100 Subject: [PATCH 7/9] style --- moler/cmd/commandtextualgeneric.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/moler/cmd/commandtextualgeneric.py b/moler/cmd/commandtextualgeneric.py index 9681ab11f..bf7ca8d45 100644 --- a/moler/cmd/commandtextualgeneric.py +++ b/moler/cmd/commandtextualgeneric.py @@ -541,8 +541,8 @@ def is_failure_indication(self, line, is_full_line): :param is_full_line: Indicates if the line is a full line or a partial line. :return: True if the line is a failure indication, False otherwise. """ - if self.re_fail is not None and is_full_line and self._regex_helper.search_compiled( - compiled=self.re_fail, string=line): + if self.re_fail is not None and is_full_line and\ + self._regex_helper.search_compiled(compiled=self.re_fail, string=line): return True return False From ab36feaa9807bdea20176b696dd91a8130a9100a Mon Sep 17 00:00:00 2001 From: Marcin Usielski Date: Thu, 18 Jan 2024 09:55:56 +0100 Subject: [PATCH 8/9] -Print --- moler/cmd/commandtextualgeneric.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/moler/cmd/commandtextualgeneric.py b/moler/cmd/commandtextualgeneric.py index bf7ca8d45..cc061d4f3 100644 --- a/moler/cmd/commandtextualgeneric.py +++ b/moler/cmd/commandtextualgeneric.py @@ -251,7 +251,6 @@ def data_received(self, data, recv_time): self.logger.debug("{} is done".format(self)) break except UnicodeDecodeError as ex: - print("UnicodeDecodeError") if self._ignore_unicode_errors: self._log(lvl=logging.WARNING, msg="Processing data from '{}' with unicode problem: '{}'.".format(self, ex)) @@ -261,7 +260,6 @@ def data_received(self, data, recv_time): msg="Processing data from '{}' raised: '{}'.".format(self, ex)) raise ex except Exception as ex: # pragma: no cover # log it just to catch that rare hanging thread issue - print("Exception {}".format(ex)) self._log(lvl=logging.WARNING, msg="Processing data from '{}' raised: '{}'.".format(self, ex)) raise ex @@ -557,16 +555,21 @@ def failure_indiction(self, line, is_full_line): if self.is_failure_indication(line=line, is_full_line=is_full_line): self.set_exception(CommandFailure(self, "command failed in line '{}'".format(line))) - def add_failure_indication(self, indication): + def add_failure_indication(self, indication, flags=re.IGNORECASE): """ Add failure indication to command. :param indication: String or regexp with ndication of failure. + :param flags: Flags for compiled regexp. :return: None """ + try: + indication_str = indication.pattern + except AttributeError: + indication_str = indication if self.re_fail is None: - new_indication = indication + new_indication = indication_str else: current_indications = self.re_fail.pattern - new_indication = r'{}|{}'.format(current_indications, indication) - self.re_fail = re.compile(new_indication, re.IGNORECASE) + new_indication = r'{}|{}'.format(current_indications, indication_str) + self.re_fail = re.compile(new_indication, flags) From b680067574e9f210b33b37a282d77248133baece Mon Sep 17 00:00:00 2001 From: Marcin Usielski Date: Thu, 18 Jan 2024 10:16:03 +0100 Subject: [PATCH 9/9] deleted commented code --- moler/cmd/unix/genericunix.py | 40 ----------------------------------- 1 file changed, 40 deletions(-) diff --git a/moler/cmd/unix/genericunix.py b/moler/cmd/unix/genericunix.py index 70d1afdae..e5c48cab8 100644 --- a/moler/cmd/unix/genericunix.py +++ b/moler/cmd/unix/genericunix.py @@ -39,32 +39,6 @@ def __init__(self, connection, prompt=None, newline_chars=None, runner=None): runner=runner) self.remove_all_known_special_chars_from_terminal_output = True self.re_fail = re.compile(r_cmd_failure_cause_alternatives, re.IGNORECASE) - # self._re_fail = None - - # def on_new_line(self, line, is_full_line): - # """ - # Method to parse command output. Will be called after line with command echo. - # Write your own implementation but don't forget to call on_new_line from base class - - # :param line: Line to parse, new lines are trimmed - # :param is_full_line: False for chunk of line; True on full line (NOTE: new line character removed) - # :return: None - # """ - # if is_full_line and self.is_failure_indication(line) is not None: - # self.set_exception(CommandFailure(self, "command failed in line '{}'".format(line))) - # return super(GenericUnixCommand, self).on_new_line(line, is_full_line) - - # def is_failure_indication(self, line): - # """ - # Method to detect if passed line contains part indicating failure of command - - # :param line: Line from command output on device - # :return: Match object if find regex in line, None otherwise. - # """ - # if self._re_fail: - # return self._regex_helper.search_compiled(compiled=self._re_fail, - # string=line) - # return None def _decode_line(self, line): """ @@ -76,17 +50,3 @@ def _decode_line(self, line): if self.remove_all_known_special_chars_from_terminal_output: line = remove_all_known_special_chars(line) return line - - # def add_failure_indication(self, indication): - # """ - # Add failure indication to command. - - # :param indication: String or regexp with ndication of failure. - # :return: None - # """ - # if self._re_fail is None: - # new_indication = indication - # else: - # current_indications = self._re_fail.pattern - # new_indication = r'{}|{}'.format(current_indications, indication) - # self._re_fail = re.compile(new_indication, re.IGNORECASE)