diff --git a/moler/events/textualevent.py b/moler/events/textualevent.py index 0bec6ee59..bc9856490 100644 --- a/moler/events/textualevent.py +++ b/moler/events/textualevent.py @@ -55,16 +55,12 @@ def data_received(self, data, recv_time): self._last_recv_time_data_read_from_connection = recv_time try: lines = data.splitlines(True) - 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 + 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: diff --git a/test/cmd/unix/test_cmd_ls.py b/test/cmd/unix/test_cmd_ls.py index 1ac468745..d1a24394d 100644 --- a/test/cmd/unix/test_cmd_ls.py +++ b/test/cmd/unix/test_cmd_ls.py @@ -3,22 +3,11 @@ Testing of ls command. """ __author__ = 'Marcin Usielski' -__copyright__ = 'Copyright (C) 2018-2024, Nokia' +__copyright__ = 'Copyright (C) 2018-2023, Nokia' __email__ = 'marcin.usielski@nokia.com' 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 diff --git a/test/events/unix/test_event_wait4prompts.py b/test/events/unix/test_event_wait4prompts.py index 150c8bc8f..78a9b05df 100644 --- a/test/events/unix/test_event_wait4prompts.py +++ b/test/events/unix/test_event_wait4prompts.py @@ -11,85 +11,6 @@ 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"] - nonlocal was_callback_called - was_callback_called = True - try: - assert state == "UNIX_LOCAL" - assert line == "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"] - nonlocal was_callback_called - was_callback_called =True - try: - assert state == "UNIX_LOCAL" - assert line == "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/#" @@ -192,3 +113,4 @@ def callback(w4p_event): event.cancel() assert 2 == len(matched_states) assert matched_states == ['UNIX_LOCAL', 'USER'] +