Fix the reverse order of data processing in TextualEvent (Wait4prompts)

CHANGELOG.md

@@ -1,3 +1,6 @@
+## moler 3.11.1
+ * Fix the Wait4prompts (data processing in reverse order)
+
 ## moler 3.11.0
  * New Unix commands (bzip2, 7z)
  * uptime - new parser

README.md

@@ -1,4 +1,4 @@
-[![image](https://img.shields.io/badge/pypi-v3.11.0-blue.svg)](https://pypi.org/project/moler/)
+[![image](https://img.shields.io/badge/pypi-v3.11.1-blue.svg)](https://pypi.org/project/moler/)
 [![image](https://img.shields.io/badge/python-3.7%20%7C%203.8%20%7C%203.9%20%7C%203.10%20%7C%203.11%20%7C%203.12-blue.svg)](https://pypi.org/project/moler/)
 [![Build Status](https://github.com/nokia/moler/actions/workflows/ci.yaml/badge.svg?branch=main)](https://github.com/nokia/moler/actions)
 [![License](https://img.shields.io/badge/License-BSD%203--Clause-blue.svg)](./LICENSE)

docs/source/conf.py

@@ -24,7 +24,7 @@
 author = 'Nokia'
 
 # The short X.Y version
-version = '3.11.0'
+version = '3.11.1'
 # The full version, including alpha/beta/rc tags
 release = 'stable'
 

moler/device/textualdevice.py

@@ -1031,7 +1031,8 @@ def _prompts_observer_callback(self, event):
         occurrence = event.get_last_occurrence()
         state = occurrence["state"]
         line = occurrence["line"]
-        msg = f"Callback for state {state} for line >>{line}<<"
+        matched = occurrence["matched"]
+        msg = f"Callback for state {state} for line >>{line}<<, matched: '{matched}'."
         if self.current_state != state:
             self._log(level=logging.INFO, msg=msg)
             self._set_state(state)

moler/events/textualevent.py

@@ -55,12 +55,16 @@ def data_received(self, data, recv_time):
             self._last_recv_time_data_read_from_connection = recv_time
             try:
                 lines = data.splitlines(True)
-                if self._reverse_order:
-                    lines.reverse()
+                processed = []
                 for current_chunk in lines:
+                    line, is_full_line = self._update_from_cached_incomplete_line(current_chunk=current_chunk)
+                    processed.append((current_chunk, line, is_full_line))
+                if self._reverse_order:
+                    processed.reverse()
+                for data in processed:
+                    current_chunk, line, is_full_line = data
                     self._last_chunk_matched = False
                     if not self.done():
-                        line, is_full_line = self._update_from_cached_incomplete_line(current_chunk=current_chunk)
                         self._process_line_from_output(line=line, current_chunk=current_chunk,
                                                        is_full_line=is_full_line)
                         if self._paused:

moler/events/unix/wait4prompts.py

@@ -63,6 +63,7 @@ def _parse_prompts(self, line):
                 current_ret = {
                     "line": line,
                     "prompt_regex": prompt_regex.pattern,
+                    "matched": self._regex_helper.group(0),
                     "state": self.compiled_prompts_regex[prompt_regex],
                     "time": datetime.datetime.now(),
                 }
@@ -103,6 +104,7 @@ def _compile_prompts_patterns(self, prompts):
 EVENT_RESULT = [
     {
         "line": "host:~ #",
+        "matched": "host:~ #",
         "prompt_regex": "host:.*#",
         "state": "UNIX_LOCAL",
         "time": datetime.datetime(2019, 8, 22, 12, 42, 38, 278418),
@@ -127,6 +129,7 @@ def _compile_prompts_patterns(self, prompts):
 EVENT_RESULT_compiled = [
     {
         "line": "host:~ #",
+        "matched": "host:~ #",
         "prompt_regex": "host:.*#",
         "state": "UNIX_LOCAL",
         "time": datetime.datetime(2019, 8, 22, 12, 42, 38, 278418),

setup.py

@@ -12,7 +12,7 @@
 
 setup(
     name='moler',
-    version='3.11.0',
+    version='3.11.1',
     description='Moler is a library for working with terminals, mainly for automated tests',  # Required
     long_description=long_description,
     long_description_content_type='text/markdown',

test/cmd/unix/test_cmd_ls.py

@@ -3,11 +3,22 @@
 Testing of ls command.
 """
 __author__ = 'Marcin Usielski'
-__copyright__ = 'Copyright (C) 2018-2023, Nokia'
+__copyright__ = 'Copyright (C) 2018-2024, Nokia'
 __email__ = '[email protected]'
 
 import pytest
 from moler.cmd.unix.ls import Ls
+from moler.exceptions import CommandFailure
+
+
+def test_lines_split(buffer_connection):
+    output = r"ls *_SYSLOG*.log" +  chr(0x0D) + chr(0x0A) + "ls: cannot access '*_SYSLOG*.log': No such file or directory" + chr(0x0D) + chr(0x0A) + "moler_bash# "
+    buffer_connection.remote_inject_response([output])
+    ls_cmd = Ls(connection=buffer_connection.moler_connection, options="*_SYSLOG*.log")
+    ls_cmd.command_string = "ls *_SYSLOG*.log"
+    with pytest.raises(CommandFailure):
+        ls_cmd()
+
 
 def test_calling_ls_returns_result_parsed_from_command_output(buffer_connection, command_output_and_expected_result):
     command_output, expected_result = command_output_and_expected_result

test/events/unix/test_event_wait4prompts.py

@@ -11,6 +11,89 @@
 import re
 
 
+def test_split_lines(buffer_connection):
+    prompts = {re.compile(r'moler_bash#'): "UNIX_LOCAL", re.compile(r'(root@|[\\w-]+).*?:.*#\\s+'): 'UNIX_LOCAL_ROOT'}
+    outputs = [
+        r"ls *_SYSLOG*.log" + chr(0x0D) + chr(0x0A),
+        r"ls: ",
+        r"cannot access '*_SYSLOG*.log' ",
+        r": No such file or directory" + chr(0x0D) + chr(0x0A) + "moler_bash#"
+    ]
+    event = Wait4prompts(connection=buffer_connection.moler_connection, till_occurs_times=-1, prompts=prompts)
+    event.check_against_all_prompts = True
+    event._break_processing_when_found = False
+    event._reverse_order = True
+    was_callback_called = False
+    error = None
+
+    def _prompts_observer_callback(event):
+        occurrence = event.get_last_occurrence()
+        state = occurrence["state"]
+        line = occurrence["line"]
+        matched = occurrence["matched"]
+        nonlocal was_callback_called
+        was_callback_called = True
+        try:
+            assert state == "UNIX_LOCAL"
+            assert line == "moler_bash#"
+            assert matched == "moler_bash#"
+        except AssertionError as err:
+            nonlocal error
+            error = err
+
+    event.add_event_occurred_callback(callback=_prompts_observer_callback,
+            callback_params={
+                "event": event,
+            },)
+    event.start()
+    for output in outputs:
+        buffer_connection.moler_connection.data_received(output.encode("utf-8"), datetime.datetime.now())
+        time.sleep(0.01)
+    time.sleep(0.5)
+    event.cancel()
+    assert was_callback_called is True
+    if error:
+        raise error
+
+
+def test_split_lines_char_by_char(buffer_connection):
+    prompts = {re.compile(r'moler_bash#'): "UNIX_LOCAL", re.compile(r'(root@|[\\w-]+).*?:.*#\\s+'): 'UNIX_LOCAL_ROOT'}
+    output = r"ls *_SYSLOG*.log" + chr(0x0D) + chr(0x0A) + "ls: cannot access '*_SYSLOG*.log': No such file or directory" + chr(0x0D) + chr(0x0A) + "moler_bash# "
+    event = Wait4prompts(connection=buffer_connection.moler_connection, till_occurs_times=-1, prompts=prompts)
+    event.check_against_all_prompts = True
+    event._break_processing_when_found = False
+    was_callback_called = False
+    error = None
+
+    def _prompts_observer_callback(event):
+        occurrence = event.get_last_occurrence()
+        state = occurrence["state"]
+        line = occurrence["line"]
+        matched = occurrence["matched"]
+        nonlocal was_callback_called
+        was_callback_called =True
+        try:
+            assert state == "UNIX_LOCAL"
+            assert line == "moler_bash#"
+            assert matched == "moler_bash#"
+        except AssertionError as err:
+            nonlocal error
+            error = err
+
+    event.add_event_occurred_callback(callback=_prompts_observer_callback,
+            callback_params={
+                "event": event,
+            },)
+    event.start()
+    for char in output:
+        buffer_connection.moler_connection.data_received(char.encode("utf-8"), datetime.datetime.now())
+    time.sleep(0.5)
+    event.cancel()
+    assert was_callback_called is True
+    if error:
+        raise error
+
+
 def test_event_wait4prompts_good_2_prompts_from_1_line(buffer_connection):
     prompts = {re.compile(r'host:.*#'): "UNIX_LOCAL", re.compile(r'user@server.*#'): "USER"}
     output = "user@host:/home/#"
@@ -113,4 +196,3 @@ def callback(w4p_event):
     event.cancel()
     assert 2 == len(matched_states)
     assert matched_states == ['UNIX_LOCAL', 'USER']
-