Common way to fail commands based on their output

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__ = '[email protected], [email protected]'
 
 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_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
@@ -337,6 +339,7 @@ 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)
 
     def is_end_of_cmd_output(self, line):
@@ -527,3 +530,46 @@ 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_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):
+        """
+        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)))
+
+    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_str
+        else:
+            current_indications = self.re_fail.pattern
+            new_indication = r'{}|{}'.format(current_indications, indication_str)
+        self.re_fail = re.compile(new_indication, flags)

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__ = '[email protected]'
 
 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)

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 == "":

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__ = '[email protected]'
 
 import re
@@ -19,13 +19,14 @@
                       'No such file or directory',
                       'running it may require superuser privileges',
                       'Cannot find device',
-                      'Input/output error']
+                      'Input/output error',
+                      ]
 r_cmd_failure_cause_alternatives = r'{}'.format("|".join(cmd_failure_causes))
 
 
 @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):
         """
@@ -37,31 +38,7 @@ 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
-
-    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
+        self.re_fail = re.compile(r_cmd_failure_cause_alternatives, re.IGNORECASE)
 
     def _decode_line(self, line):
         """

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__ = '[email protected]'
+__author__ = 'Tomasz Krol, Marcin Usielski'
+__copyright__ = 'Copyright (C) 2020-2024, Nokia'
+__email__ = '[email protected], [email protected]'
 
 
 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"""

moler/cmd/unix/touch.py

@@ -0,0 +1,63 @@
+# -*- coding: utf-8 -*-
+"""
+Command touch module
+"""
+
+
+__author__ = 'Marcin Usielski'
+__copyright__ = 'Copyright (C) 2018-2024, Nokia'
+__email__ = '[email protected]'
+
+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
+        self.add_failure_indication('touch: cannot touch')
+
+    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 = {}

test/cmd/unix/test_cmd_lxc_info.py

@@ -10,8 +10,8 @@
 """
 
 __author__ = 'Piotr Frydrych, Marcin Usielski'
-__copyright__ = 'Copyright (C) 2019-2020, Nokia'
-__email__ = '[email protected]'
+__copyright__ = 'Copyright (C) 2019-2024, Nokia'
+__email__ = '[email protected], [email protected]'
 
 
 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()

test/cmd/unix/test_cmd_touch.py

@@ -0,0 +1,41 @@
+# -*- coding: utf-8 -*-
+"""
+Touch command test module.
+"""
+
+__author__ = 'Marcin Usielski'
+__copyright__ = 'Copyright (C) 2024, Nokia'
+__email__ = '[email protected]'
+
+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()
+
+
+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")
+    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()