Wait4Prompts - reverse order of processing lines (#537)

* Wait4Prompts - reverse order of processing lines

* unit test

* test fix

moler/event.py

@@ -36,6 +36,7 @@ def __init__(self, connection: Optional[AbstractMolerConnection] = None, till_oc
         self.till_occurs_times = till_occurs_times
         self._log_every_occurrence = True
         self.event_name = Event.observer_name
+        self._last_chunk_matched = False
 
     def __str__(self):
         """
@@ -109,6 +110,7 @@ def event_occurred(self, event_data) -> None:
             raise ResultAlreadySet(self)
         self._prepare_result_from_occurred()
         self._occurred.append(event_data)
+        self._last_chunk_matched = True
         if self.till_occurs_times > 0:
             if len(self._occurred) >= self.till_occurs_times:
                 self.break_event()

moler/events/textualevent.py

@@ -24,6 +24,10 @@ def __init__(self, connection=None, till_occurs_times=-1, runner=None):
         self._ignore_unicode_errors = True  # If True then UnicodeDecodeError will be logged not raised in data_received
         self._last_recv_time_data_read_from_connection = None  # Time moment when data was really received from
         # connection (not when was passed to event).  Time is given as datetime.datetime instance
+        self._reverse_order: bool = False  # Order of processing lines in data_received
+        # method. False to process from the last line, True from the newest.
+        self._break_processing_when_found: bool = False  # Flag to break processing
+        # line when the first line is found.
 
     def event_occurred(self, event_data):
         self._consume_already_parsed_fragment()
@@ -51,14 +55,19 @@ 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()
                 for current_chunk in lines:
+                    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:
                             self._last_not_full_line = None
                             break
+                        if self._last_chunk_matched and self._break_processing_when_found:
+                            break
             except UnicodeDecodeError as ex:
                 if self._ignore_unicode_errors:
                     self._log(lvl=logging.WARNING,

moler/events/unix/wait4prompts.py

@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 __author__ = "Michal Ernst, Marcin Usielski"
-__copyright__ = "Copyright (C) 2019-2023, Nokia"
+__copyright__ = "Copyright (C) 2019-2024, Nokia"
 __email__ = "[email protected], [email protected]"
 
 import datetime
@@ -34,6 +34,10 @@ def __init__(self, connection, prompts, till_occurs_times=-1, runner=None):
         self.check_against_all_prompts = False
         self._ret_list_matched = []
 
+        # Change default order and behavior after matching the prompt
+        self._reverse_order = True
+        self._break_processing_when_found = True
+
     def on_new_line(self, line, is_full_line):
         try:
             self._parse_prompts(line)

test/events/unix/test_event_wait4prompts.py

@@ -1,9 +1,10 @@
 # -*- coding: utf-8 -*-
 
 __author__ = 'Marcin Usielski'
-__copyright__ = 'Copyright (C) 2021-2023, Nokia'
+__copyright__ = 'Copyright (C) 2021-2024, Nokia'
 __email__ = '[email protected]'
 
+import time
 
 from moler.events.unix.wait4prompts import Wait4prompts
 import datetime
@@ -63,3 +64,53 @@ def test_event_wait4prompts_wrong_1_prompt_from_1_line(buffer_connection):
     assert 'list_matched' not in ret
 
 
+def test_event_wait4prompts_reverse_order(buffer_connection):
+    matched_states = []
+
+    def callback(w4p_event):
+        occurrence = w4p_event.get_last_occurrence()
+        state = occurrence["state"]
+        matched_states.append(state)
+
+    prompts = {re.compile(r'host:.*#'): "UNIX_LOCAL", re.compile(r'user@server.*#'): "USER"}
+    output = "user@host:/home/#\nBLABLA\nuser@server:/home/#\n"
+    event = Wait4prompts(connection=buffer_connection.moler_connection,
+                         till_occurs_times=-1, prompts=prompts)
+    event.add_event_occurred_callback(callback=callback, callback_params={"w4p_event": event})
+    event._reverse_order = True
+    event.start()
+    buffer_connection.moler_connection.data_received(output.encode("utf-8"), datetime.datetime.now())
+    start_time = time.monotonic()
+    while 1 != len(matched_states) and time.monotonic() - start_time < 10:
+        time.sleep(0.1)
+    time.sleep(0.1)
+    event.cancel()
+    assert 1 == len(matched_states)
+    assert 'USER' in matched_states
+
+
+def test_event_wait4prompts_normal_order(buffer_connection):
+    matched_states = []
+
+    def callback(w4p_event):
+        occurrence = w4p_event.get_last_occurrence()
+        state = occurrence["state"]
+        matched_states.append(state)
+
+    prompts = {re.compile(r'host:.*#'): "UNIX_LOCAL", re.compile(r'user@server.*#'): "USER"}
+    output = "user@host:/home/#\nBLABLA\nuser@server:/home/#\n"
+    event = Wait4prompts(connection=buffer_connection.moler_connection,
+                         till_occurs_times=-1, prompts=prompts)
+    event._reverse_order = False
+    event._break_processing_when_found = False
+    event.add_event_occurred_callback(callback=callback, callback_params={"w4p_event": event})
+    event.start()
+    buffer_connection.moler_connection.data_received(output.encode("utf-8"), datetime.datetime.now())
+    start_time = time.monotonic()
+    while 2 != len(matched_states) and time.monotonic() - start_time < 10:
+        time.sleep(0.1)
+    time.sleep(0.1)
+    event.cancel()
+    assert 2 == len(matched_states)
+    assert matched_states == ['UNIX_LOCAL', 'USER']
+