twister: pytest: Wrap iter_stdout

Added waiting for prompt instead of calling time.sleep,
to fix issue with longer setting up platforms with TF-M
enabled. Wrapped iter_stdout generator, to avoid issues
when generator is created and called multiple times
in test scenarios.
Fixes: #58747

Signed-off-by: Grzegorz Chwierut <[email protected]>

Author: gchwier

samples/subsys/testsuite/pytest/shell/pytest/test_shell.py

@@ -4,16 +4,15 @@
 
 import time
 import logging
-import pytest  # noqa # pylint: disable=unused-import
 
 from twister_harness.device.device_abstract import DeviceAbstract
 
 logger = logging.getLogger(__name__)
 
 
-def wait_for_message(iter_stdout, message, timeout=60):
+def wait_for_message(dut: DeviceAbstract, message, timeout=20):
     time_started = time.time()
-    for line in iter_stdout:
+    for line in dut.iter_stdout:
         if line:
             logger.debug("#: " + line)
         if message in line:
@@ -22,15 +21,25 @@ def wait_for_message(iter_stdout, message, timeout=60):
             return False
 
 
-def test_shell_print_help(dut: DeviceAbstract):
-    time.sleep(1)  # wait for application initialization on DUT
+def wait_for_prompt(dut: DeviceAbstract, prompt='uart:~$', timeout=20):
+    time_started = time.time()
+    while True:
+        dut.write(b'\n')
+        for line in dut.iter_stdout:
+            if prompt in line:
+                logger.debug('Got prompt')
+                return True
+        if time.time() > time_started + timeout:
+            return False
 
+
+def test_shell_print_help(dut: DeviceAbstract):
+    wait_for_prompt(dut)
     dut.write(b'help\n')
-    assert wait_for_message(dut.iter_stdout, "Available commands")
+    assert wait_for_message(dut, "Available commands")
 
 
 def test_shell_print_version(dut: DeviceAbstract):
-    time.sleep(1)  # wait for application initialization on DUT
-
+    wait_for_prompt(dut)
     dut.write(b'kernel version\n')
-    assert wait_for_message(dut.iter_stdout, "Zephyr version")
+    assert wait_for_message(dut, "Zephyr version")

scripts/pylib/pytest-twister-harness/src/twister_harness/device/device_abstract.py

@@ -25,6 +25,7 @@ def __init__(self, device_config: DeviceConfig, **kwargs) -> None:
         self.device_config: DeviceConfig = device_config
         self.handler_log_file: LogFile = NullLogFile.create()
         self.device_log_file: LogFile = NullLogFile.create()
+        self.iter_object: Generator[str, None, None] | None = None
 
     def __repr__(self) -> str:
         return f'{self.__class__.__name__}()'
@@ -55,11 +56,6 @@ def flash_and_run(self, timeout: float = 60.0) -> None:
         :param timeout: time out in seconds
         """
 
-    @property
-    @abc.abstractmethod
-    def iter_stdout(self) -> Generator[str, None, None]:
-        """Iterate stdout from a device."""
-
     @abc.abstractmethod
     def write(self, data: bytes) -> None:
         """Write data bytes to device"""
@@ -73,22 +69,17 @@ def initialize_log_files(self):
     def stop(self) -> None:
         """Stop device."""
 
-    # @abc.abstractmethod
-    # def read(self, size=1) -> None:
-    #     """Read size bytes from device"""
-
-    # def read_until(self, expected, size=None):
-    #     """Read until an expected bytes sequence is found"""
-    #     lenterm = len(expected)
-    #     line = bytearray()
-    #     while True:
-    #         c = self.read(1)
-    #         if c:
-    #             line += c
-    #             if line[-lenterm:] == expected:
-    #                 break
-    #             if size is not None and len(line) >= size:
-    #                 break
-    #         else:
-    #             break
-    #     return bytes(line)
+    @abc.abstractmethod
+    def iter_stdout_lines(self) -> Generator[str, None, None]:
+        """A generator that yields lines read from device"""
+
+    @property
+    def iter_stdout(self) -> Generator[str, None, None]:
+        """
+        Get generator object to iterate stdout from a device.
+        This wrapper method is added to avoid problems, when
+        user creates an instance of generator multiple times.
+        """
+        if not self.iter_object:
+            self.iter_object = self.iter_stdout_lines()
+        return self.iter_object

scripts/pylib/pytest-twister-harness/src/twister_harness/device/hardware_adapter.py

@@ -201,8 +201,7 @@ def flash_and_run(self, timeout: float = 60.0) -> None:
             if self.device_config.post_flash_script:
                 self.run_custom_script(self.device_config.post_flash_script, 30)
 
-    @property
-    def iter_stdout(self) -> Generator[str, None, None]:
+    def iter_stdout_lines(self) -> Generator[str, None, None]:
         """Return output from serial."""
         if not self.connection:
             return
@@ -212,6 +211,7 @@ def iter_stdout(self) -> Generator[str, None, None]:
             stream = self.connection.readline()
             self.handler_log_file.handle(data=stream)
             yield stream.decode(errors='ignore').strip()
+        self.iter_object = None
 
     def write(self, data: bytes) -> None:
         """Write data to serial"""

scripts/pylib/pytest-twister-harness/src/twister_harness/device/qemu_adapter.py

@@ -145,12 +145,11 @@ def _wait_for_fifo(self):
             logger.error(msg)
             raise TwisterHarnessException(msg)
 
-    @property
-    def iter_stdout(self) -> Generator[str, None, None]:
+    def iter_stdout_lines(self) -> Generator[str, None, None]:
         if not self.connection:
             return
-        # fifo file can be not create yet, so we need to wait for a while
-        self._wait_for_fifo()
+        if not self.connection.is_open:
+            self._wait_for_fifo()
 
         # create unblocking reading from fifo file
         q: Queue = Queue()
@@ -184,11 +183,15 @@ def read_lines():
             pass
         finally:
             t.join(1)
+            self.iter_object = None
 
     def write(self, data: bytes) -> None:
         """Write data to serial"""
-        if self.connection:
-            self.connection.write(data)
+        if not self.connection:
+            return
+        if not self.connection.is_open:
+            self._wait_for_fifo()
+        self.connection.write(data)
 
     def initialize_log_files(self):
         self.handler_log_file = HandlerLogFile.create(build_dir=self.device_config.build_dir)

scripts/pylib/pytest-twister-harness/src/twister_harness/device/simulator_adapter.py

@@ -166,8 +166,7 @@ def stop(self) -> None:
         if self._exc:
             raise self._exc
 
-    @property
-    def iter_stdout(self) -> Generator[str, None, None]:
+    def iter_stdout_lines(self) -> Generator[str, None, None]:
         """Return output from serial."""
         while True:
             stream = self.queue.get()
@@ -177,6 +176,7 @@ def iter_stdout(self) -> Generator[str, None, None]:
             self.handler_log_file.handle(data=stream + '\n')
             yield stream
             self.queue.task_done()
+        self.iter_object = None
 
     def write(self, data: bytes) -> None:
         """Write data to serial"""