From 7da6ccd7cceab4c26bd77159a669a346b2dae8d9 Mon Sep 17 00:00:00 2001 From: Lucas Saavedra Vaz <32426024+lucasssvaz@users.noreply.github.com> Date: Wed, 20 May 2026 15:09:42 -0300 Subject: [PATCH] feat: Add option to capture payloads --- docs/concepts/key-concepts.rst | 38 + docs/usages/payload_capture.rst | 121 +++ .../pytest_embedded_wokwi/wokwi.py | 11 +- pytest-embedded/pytest_embedded/dut.py | 137 +++ .../pytest_embedded/dut_factory.py | 80 +- pytest-embedded/pytest_embedded/plugin.py | 43 +- pytest-embedded/tests/test_base.py | 886 ++++++++++++++++++ 7 files changed, 1299 insertions(+), 17 deletions(-) create mode 100644 docs/usages/payload_capture.rst diff --git a/docs/concepts/key-concepts.rst b/docs/concepts/key-concepts.rst index a451eb5f..25ce6033 100644 --- a/docs/concepts/key-concepts.rst +++ b/docs/concepts/key-concepts.rst @@ -149,3 +149,41 @@ The ``--qemu-cli-args`` option applies to the first DUT (with the ``qemu`` servi ``pytest-embedded`` prints all DUT output with a timestamp. To remove the timestamp, run pytest with the ``--with-timestamp n`` option. By default, ``pytest`` swallows ``stdout``. To see the live output, run pytest with the ``-s`` option. + +************* + Echo Muting +************* + +The listener process that echoes serial output to ``stdout`` supports two complementary muting mechanisms. When muted, output is still written to the log file — only the terminal echo is suppressed. + +Automatic Muting via ``mute_patterns`` +====================================== + +Override the ``mute_patterns`` fixture in your ``conftest.py`` to provide ``(start, end)`` string pairs. When the listener sees *start* in the serial stream it suppresses echo until *end* is seen: + +.. code:: python + + @pytest.fixture + def mute_patterns(): + return [("<<>>", "<<>>")] + +This is the recommended approach because it operates inside the listener process itself, so there is no race condition between the DUT producing output and the test code reacting to it. + +Manual Muting via ``Dut`` +========================= + +For ad-hoc suppression you can call :func:`~pytest_embedded.dut.Dut.mute_echo` / :func:`~pytest_embedded.dut.Dut.unmute_echo`, or use the :func:`~pytest_embedded.dut.Dut.muted_echo` context manager: + +.. code:: python + + def test_verbose_section(dut): + with dut.muted_echo(): + dut.expect("noisy output") + +***************** + Payload Capture +***************** + +:func:`~pytest_embedded.dut.Dut.capture_payload` and :func:`~pytest_embedded.dut.Dut.capture_payload_to_file` let you extract delimited data blocks from the serial output — for example Base64-encoded binary dumps, coverage data, or diagnostic payloads. + +Both methods pair naturally with ``mute_patterns``: register the same start/end markers so the raw dump never clutters the terminal, then call ``capture_payload`` to retrieve the data programmatically. diff --git a/docs/usages/payload_capture.rst b/docs/usages/payload_capture.rst new file mode 100644 index 00000000..b3601a43 --- /dev/null +++ b/docs/usages/payload_capture.rst @@ -0,0 +1,121 @@ +############################### + Payload Capture & Echo Muting +############################### + +Embedded devices sometimes emit large blocks of data over the serial port — Base64-encoded binary blobs, coverage dumps, diagnostic payloads, etc. ``pytest-embedded`` provides tools to **capture** these blocks programmatically and **mute** the terminal echo so the console stays readable. + +*********************** + Automatic Echo Muting +*********************** + +The listener process that echoes serial output to ``stdout`` supports pattern-based muting. Override the ``mute_patterns`` fixture in your project's ``conftest.py``: + +.. code:: python + + # conftest.py + import pytest + + @pytest.fixture + def mute_patterns(): + return [ + ("<<>>", "<<>>"), + ] + +Each ``(start, end)`` pair defines a muting region. When the listener detects *start* in the serial stream, it suppresses ``stdout`` echo until *end* is seen. Log-file writing is **never** affected — all data is always recorded to the pexpect log file. + +Multiple patterns can be registered simultaneously: + +.. code:: python + + @pytest.fixture + def mute_patterns(): + return [ + ("<<>>", "<<>>"), + ("<<>>", "<<>>"), + ] + +.. note:: + + ``mute_patterns`` operates inside the listener process itself, so there is no race condition between the device producing output and the test code reacting to it. + +******************** + Manual Echo Muting +******************** + +For sections that are not delimited by fixed markers, use the :func:`~pytest_embedded.dut.Dut.muted_echo` context manager: + +.. code:: python + + def test_noisy_section(dut): + with dut.muted_echo(): + dut.expect("some verbose output") + # stdout is silent, but the log file still records everything + +You can also call :func:`~pytest_embedded.dut.Dut.mute_echo` and :func:`~pytest_embedded.dut.Dut.unmute_echo` directly for fine-grained control. + +******************** + Capturing Payloads +******************** + +:func:`~pytest_embedded.dut.Dut.capture_payload` waits for a *start* marker, collects all bytes until the *end* marker, and returns them as raw ``bytes``: + +.. code:: python + + def test_extract_data(dut): + data = dut.capture_payload( + start="<<>>", + end="<<>>", + start_timeout=10, + timeout=60, + ) + assert data is not None + # process `data` ... + +If you prefer to persist the captured data to a file (for example, to accumulate data from multiple reboots), use :func:`~pytest_embedded.dut.Dut.capture_payload_to_file`: + +.. code:: python + + def test_multi_boot_capture(dut): + for boot in range(3): + dut.write("reboot\n") + ok = dut.capture_payload_to_file( + start="<<>>", + end="<<>>", + filepath="/tmp/dump.txt", + append=True, # accumulate across boots + include_markers=True, # wrap each block with markers + ) + assert ok + +************************* + Combining Both Features +************************* + +Registering start/end markers in ``mute_patterns`` **and** using them in ``capture_payload`` is the recommended pattern — the listener mutes the echo automatically while the test code captures the data: + +.. code:: python + + # conftest.py + import pytest + + DUMP_START = "<<>>" + DUMP_END = "<<>>" + + @pytest.fixture + def mute_patterns(): + return [(DUMP_START, DUMP_END)] + +.. code:: python + + # test_coverage.py + from conftest import DUMP_START, DUMP_END + + def test_gcov(dut): + ok = dut.capture_payload_to_file( + start=DUMP_START, + end=DUMP_END, + filepath="gcov_raw.txt", + ) + assert ok + +The console stays clean, the log file has everything, and ``gcov_raw.txt`` contains just the payload. diff --git a/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py b/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py index 438fd4f0..c2c2b500 100644 --- a/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py +++ b/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py @@ -225,13 +225,6 @@ def _start_serial_monitoring(self): """Start monitoring serial output and forward to stdout and message queue.""" def serial_callback(data: bytes): - # Write to stdout for live monitoring - try: - decoded = data.decode('utf-8', errors='replace') - print(decoded, end='', flush=True) - except Exception as e: - logging.debug(f'Error writing to stdout: {e}') - # Write to log file if available try: if hasattr(self, '_fw') and self._fw and not self._fw.closed: @@ -241,7 +234,9 @@ def serial_callback(data: bytes): except Exception as e: logging.debug(f'Error writing to log file: {e}') - # Put in message queue for expect() functionality + # Put in message queue for expect() functionality. + # The listener process handles timestamped stdout echo and + # respects mute_patterns / mute_event, so we don't print here. try: if hasattr(self, '_q') and self._q: self._q.put(data) diff --git a/pytest-embedded/pytest_embedded/dut.py b/pytest-embedded/pytest_embedded/dut.py index 6104b200..ad736bee 100644 --- a/pytest-embedded/pytest_embedded/dut.py +++ b/pytest-embedded/pytest_embedded/dut.py @@ -1,5 +1,7 @@ +import contextlib import functools import logging +import os import os.path import re from collections.abc import Callable @@ -33,10 +35,12 @@ def __init__( pexpect_logfile: str, test_case_name: str, meta: Meta | None = None, + _mute_event=None, **kwargs, ) -> None: self._q = msg_queue self._meta = meta + self._mute_event = _mute_event self.pexpect_proc = pexpect_proc self.app = app @@ -59,6 +63,139 @@ def close(self) -> None: self.testsuite.dump(junit_report) logging.info(f'Created unity output junit report: {junit_report}') + def mute_echo(self) -> None: + """Suppress stdout echo from the listener process. Log file writing is unaffected.""" + if self._mute_event is not None: + self._mute_event.set() + + def unmute_echo(self) -> None: + """Resume stdout echo from the listener process.""" + if self._mute_event is not None: + self._mute_event.clear() + + @contextlib.contextmanager + def muted_echo(self): + """Context manager that suppresses stdout echo while still logging to file.""" + self.mute_echo() + try: + yield + finally: + self.unmute_echo() + + def capture_payload( + self, + start: str, + end: str, + start_timeout: float = 10, + timeout: float = 60, + mute: bool = False, + ) -> bytes | None: + """Capture a delimited payload from the serial output. + + Waits for the *start* marker, then reads until the *end* marker and + returns the raw bytes **between** the two markers. + + Stdout echo suppression is best handled automatically by + registering the markers via the ``mute_patterns`` fixture (see + :func:`pytest_embedded.plugin.mute_patterns`). When *mute* is + ``True`` the event-based manual mute is also engaged as a + fallback. + + Args: + start: exact string that marks the beginning of the payload. + end: exact string that marks the end of the payload. + start_timeout: seconds to wait for *start* before giving up. + timeout: seconds to wait for *end* after *start* has been seen. + mute: additionally engage the manual mute event. (default: False) + + Returns: + The captured bytes between the markers, or ``None`` if the start + marker was not seen or the end marker timed out. + """ + ctx = self.muted_echo() if mute else contextlib.nullcontext() + with ctx: + try: + self.pexpect_proc.expect(start, timeout=start_timeout) + except (pexpect.EOF, pexpect.TIMEOUT): + logging.debug('capture_payload: start marker %r not found within %ss', start, start_timeout) + return None + except Exception: + logging.debug('capture_payload: unexpected error waiting for start marker', exc_info=True) + return None + + try: + self.pexpect_proc.expect(end, timeout=timeout) + return self.pexpect_proc.before + except pexpect.TIMEOUT: + logging.warning('capture_payload: end marker %r not found within %ss', end, timeout) + return None + except Exception: + logging.debug('capture_payload: unexpected error waiting for end marker', exc_info=True) + return None + + def capture_payload_to_file( + self, + start: str, + end: str, + filepath: str, + start_timeout: float = 10, + timeout: float = 60, + mute: bool = True, + append: bool = True, + include_markers: bool = True, + ) -> bool: + """Capture a delimited payload and write it to a file. + + This is a convenience wrapper around :meth:`capture_payload` that + persists the result to *filepath*. + + Args: + start: exact string that marks the beginning of the payload. + end: exact string that marks the end of the payload. + filepath: destination file path (parent directories are created + automatically). + start_timeout: seconds to wait for *start* before giving up. + timeout: seconds to wait for *end* after *start* has been seen. + mute: suppress stdout echo while capturing. (default: True) + append: open the file in append mode so that successive calls + accumulate data (e.g. one call per reboot). (default: True) + include_markers: wrap the written data with ``start`` / ``end`` + lines so downstream parsers can re-identify each block. + (default: True) + + Returns: + ``True`` if the payload was captured and written, ``False`` + otherwise. + """ + data = self.capture_payload( + start=start, + end=end, + start_timeout=start_timeout, + timeout=timeout, + mute=mute, + ) + if data is None: + return False + + if isinstance(data, bytes): + data = data.decode('utf-8', errors='replace') + + try: + os.makedirs(os.path.dirname(filepath) or '.', exist_ok=True) + mode = 'a' if append else 'w' + with open(filepath, mode) as fh: + if include_markers: + fh.write(start + '\n') + fh.write(data) + if include_markers: + fh.write(end + '\n') + logging.info('capture_payload_to_file: payload written to %s', filepath) + except Exception: + logging.debug('capture_payload_to_file: failed to write %s', filepath, exc_info=True) + return False + + return True + def write(self, s: AnyStr) -> None: """ Write to the `MessageQueue` instance diff --git a/pytest-embedded/pytest_embedded/dut_factory.py b/pytest-embedded/pytest_embedded/dut_factory.py index 1b11e301..e86708a2 100644 --- a/pytest-embedded/pytest_embedded/dut_factory.py +++ b/pytest-embedded/pytest_embedded/dut_factory.py @@ -52,10 +52,20 @@ def set_stdout_lock(lock) -> None: def _listen( - q: MessageQueue, filepath: str, with_timestamp: bool = True, count: int = 1, total: int = 1, _stdout_lock=None + q: MessageQueue, + filepath: str, + with_timestamp: bool = True, + count: int = 1, + total: int = 1, + _stdout_lock=None, + mute_event=None, + mute_patterns=(), ) -> None: shall_add_prefix = True _pending = '' + _active_mute_pair = None + _max_pat_len = max((len(s) for pair in mute_patterns for s in pair), default=0) + _match_buf = '' while True: msg = q.get() if not msg: @@ -69,6 +79,8 @@ def _listen( if not _s: continue + raw = _s + prefix = '' if total > 1: prefix = f'[dut-{count}] ' @@ -79,22 +91,62 @@ def _listen( if shall_add_prefix: _s = prefix + _s + _was_in_mute_at_chunk_start = _active_mute_pair is not None + _entered_mute_this_chunk = False + if mute_patterns: + _match_buf += raw + changed = True + while changed: + changed = False + if _active_mute_pair is None: + for start_pat, end_pat in mute_patterns: + if start_pat in _match_buf: + _active_mute_pair = (start_pat, end_pat) + _entered_mute_this_chunk = True + _match_buf = _match_buf[_match_buf.index(start_pat) + len(start_pat) :] + changed = True + break + else: + _, end_pat = _active_mute_pair + if end_pat in _match_buf: + _match_buf = _match_buf[_match_buf.index(end_pat) + len(end_pat) :] + _active_mute_pair = None + changed = True + if _max_pat_len > 1: + _match_buf = _match_buf[-(_max_pat_len - 1) :] if len(_match_buf) >= _max_pat_len else _match_buf + + _muted = ( + _was_in_mute_at_chunk_start + or _entered_mute_this_chunk + or _active_mute_pair is not None + or (mute_event is not None and mute_event.is_set()) + ) + _s = _s.replace('\r\n', '\n') # remove extra \r. since multi-dut \r would mess up the log if _s.endswith('\n'): # complete line shall_add_prefix = True _s = _s[:-1].replace('\n', '\n' + prefix) + '\n' - with _stdout_lock if _stdout_lock else contextlib.nullcontext(): - _stdout.write(_pending + _s) - _stdout.flush() + if not _muted: + with _stdout_lock if _stdout_lock else contextlib.nullcontext(): + _stdout.write(_pending + _s) + _stdout.flush() _pending = '' else: shall_add_prefix = False _s = _s.replace('\n', '\n' + prefix) - _pending += _s + if not _muted: + _pending += _s def _listener_gn( - msg_queue, _pexpect_logfile, with_timestamp, dut_index, dut_total, _stdout_lock=None + msg_queue, + _pexpect_logfile, + with_timestamp, + dut_index, + dut_total, + _stdout_lock=None, + _mute_event=None, + mute_patterns=(), ) -> multiprocessing.Process: os.makedirs(os.path.dirname(_pexpect_logfile), exist_ok=True) kwargs = { @@ -102,6 +154,8 @@ def _listener_gn( 'count': dut_index, 'total': dut_total, '_stdout_lock': _stdout_lock, + 'mute_event': _mute_event, + 'mute_patterns': mute_patterns, } return _ctx.Process( @@ -167,6 +221,7 @@ def _fixture_classes_and_options_fn( pexpect_proc, msg_queue, _meta, + _mute_event=None, **kwargs, ) -> ClassCliOptions: classes: dict[str, type] = {} @@ -350,6 +405,7 @@ def _fixture_classes_and_options_fn( 'pexpect_logfile': _pexpect_logfile, 'test_case_name': test_case_name, 'meta': _meta, + '_mute_event': _mute_event, } if 'idf' in _services and 'esp' not in _services: # esp,idf will use IdfDut, which based on IdfUnityDutMixin already @@ -770,8 +826,17 @@ def create( ) logging.debug('You can get your custom DUT log file at the following path: %s.', _pexpect_logfile) + _mute_event = _ctx.Event() + layout.append(_mute_event) + _listener = _listener_gn( - msg_queue, _pexpect_logfile, True, DUT_GLOBAL_INDEX, DUT_GLOBAL_INDEX + 1, _stdout_lock=_STDOUT_LOCK + msg_queue, + _pexpect_logfile, + True, + DUT_GLOBAL_INDEX, + DUT_GLOBAL_INDEX + 1, + _stdout_lock=_STDOUT_LOCK, + _mute_event=_mute_event, ) layout.append(_listener) @@ -824,6 +889,7 @@ def create( '_pexpect_logfile': _pexpect_logfile, 'pexpect_proc': pexpect_proc, 'msg_queue': msg_queue, + '_mute_event': _mute_event, } _fixture_classes_and_options = _fixture_classes_and_options_fn(**_kwargs) diff --git a/pytest-embedded/pytest_embedded/plugin.py b/pytest-embedded/pytest_embedded/plugin.py index 38d54503..3c6663f9 100644 --- a/pytest-embedded/pytest_embedded/plugin.py +++ b/pytest-embedded/pytest_embedded/plugin.py @@ -763,10 +763,44 @@ def with_timestamp(request: FixtureRequest) -> bool: return _request_param_or_config_option_or_default(request, 'with_timestamp', None) +@pytest.fixture +@multi_dut_generator_fixture +def _mute_event(): + """Per-DUT event that, when set, suppresses stdout echo in the listener while still logging to file.""" + return _ctx.Event() + + +@pytest.fixture +def mute_patterns(): + """Return a sequence of ``(start, end)`` string pairs. + + When the listener process sees *start* in the serial output it + suppresses stdout echo until *end* is seen. Log-file writing is + unaffected. Override this fixture in ``conftest.py`` to register + project-specific patterns:: + + @pytest.fixture + def mute_patterns(): + return [('<<>>', '<<>>')] + """ + return () + + +@pytest.fixture +def _mute_patterns_safe(mute_patterns): + """Wrap mute_patterns in a dict to prevent multi-DUT splitting. + + ``multi_dut_generator_fixture`` splits all ``list``/``tuple`` kwargs + by DUT index. ``mute_patterns`` is a sequence of pattern pairs shared + across all DUTs, not a per-DUT value, so it must not be split. + """ + return {'patterns': tuple(mute_patterns)} + + @pytest.fixture @multi_dut_generator_fixture def _listener( - msg_queue, _pexpect_logfile, with_timestamp, dut_index, dut_total, _stdout_lock + msg_queue, _pexpect_logfile, with_timestamp, dut_index, dut_total, _stdout_lock, _mute_event, _mute_patterns_safe ) -> multiprocessing.Process: """ The listener would create a `_listen` process. The `_listen` process would get the string from the message queue, @@ -776,8 +810,12 @@ def _listener( 2. write the string to `_pexpect_logfile` A shared lock (_stdout_lock) is used to prevent interleaved output when multiple DUTs print simultaneously. + When ``_mute_event`` is set, stdout echo is suppressed while log file writing continues. + ``mute_patterns`` provides automatic muting based on start/end marker pairs. """ - return _listener_gn(**locals()) + kwargs = dict(locals()) + kwargs['mute_patterns'] = kwargs.pop('_mute_patterns_safe')['patterns'] + return _listener_gn(**kwargs) @pytest.fixture @@ -1160,6 +1198,7 @@ def _fixture_classes_and_options( _pexpect_logfile, pexpect_proc, msg_queue, + _mute_event, ) -> ClassCliOptions: """ classes: the class that the fixture should instantiate diff --git a/pytest-embedded/tests/test_base.py b/pytest-embedded/tests/test_base.py index 89b1cab2..5a8ded4f 100644 --- a/pytest-embedded/tests/test_base.py +++ b/pytest-embedded/tests/test_base.py @@ -1010,3 +1010,889 @@ def test_concurrent_dut_writes(dut): result = testdir.runpytest() result.assert_outcomes(passed=1) + + +# --------------------------------------------------------------------------- +# Tests for echo muting (_listen mute_event / mute_patterns) +# --------------------------------------------------------------------------- + + +def _run_listen_with_stdout_capture(q, logfile, stdout_filepath, listen_kwargs): + """Run _listen in a child process, redirecting module stdout to a file.""" + import pytest_embedded.dut_factory as dut_factory + from pytest_embedded.dut_factory import _listen + + with open(stdout_filepath, 'w', encoding='utf-8') as captured: + dut_factory._stdout = captured + _listen(q, logfile, **listen_kwargs) + + +def test_listen_backwards_compat_no_mute_args(tmp_path): + """_listen still works when called without mute_event or mute_patterns (pre-existing API).""" + import time + + from pytest_embedded.dut_factory import _ctx, _listen + from pytest_embedded.log import MessageQueue + + logfile = str(tmp_path / 'test.log') + q = MessageQueue() + messages = [f'compat_{i}\n'.encode() for i in range(10)] + + p = _ctx.Process(target=_listen, args=(q, logfile), kwargs={'with_timestamp': False}) + p.start() + try: + for msg in messages: + q.put(msg) + + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + try: + content = open(logfile, 'rb').read() + if all(msg in content for msg in messages): + break + except OSError: + pass + time.sleep(0.05) + finally: + p.terminate() + p.join(timeout=5) + assert p.exitcode is not None + + content = open(logfile, 'rb').read() + for msg in messages: + assert msg in content, f'{msg!r} missing from logfile' + + +def test_listen_mute_event_suppresses_stdout(tmp_path): + """When mute_event is set, _listen still writes to the logfile but skips stdout.""" + import time + + from pytest_embedded.dut_factory import _ctx, _listen + from pytest_embedded.log import MessageQueue + + logfile = str(tmp_path / 'test.log') + q = MessageQueue() + event = _ctx.Event() + event.set() + + messages = [f'muted_line_{i}\n'.encode() for i in range(10)] + + p = _ctx.Process( + target=_listen, + args=(q, logfile), + kwargs={'with_timestamp': False, 'mute_event': event}, + ) + p.start() + try: + for msg in messages: + q.put(msg) + + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + try: + content = open(logfile, 'rb').read() + if all(msg in content for msg in messages): + break + except OSError: + pass + time.sleep(0.05) + finally: + p.terminate() + p.join(timeout=5) + + content = open(logfile, 'rb').read() + for msg in messages: + assert msg in content, f'{msg!r} should still be in logfile even when muted' + + +def test_listen_mute_event_cleared_allows_stdout(tmp_path): + """When mute_event is cleared (default), _listen writes to both logfile and stdout.""" + import time + + from pytest_embedded.dut_factory import _ctx, _listen + from pytest_embedded.log import MessageQueue + + logfile = str(tmp_path / 'test.log') + q = MessageQueue() + event = _ctx.Event() # not set = not muted + + messages = [f'unmuted_{i}\n'.encode() for i in range(5)] + + p = _ctx.Process( + target=_listen, + args=(q, logfile), + kwargs={'with_timestamp': False, 'mute_event': event}, + ) + p.start() + try: + for msg in messages: + q.put(msg) + + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + try: + content = open(logfile, 'rb').read() + if all(msg in content for msg in messages): + break + except OSError: + pass + time.sleep(0.05) + finally: + p.terminate() + p.join(timeout=5) + + content = open(logfile, 'rb').read() + for msg in messages: + assert msg in content + + +def test_listen_mute_patterns_auto_mutes(tmp_path): + """mute_patterns causes _listen to auto-mute between start/end markers; logfile is unaffected.""" + import time + + from pytest_embedded.dut_factory import _ctx, _listen + from pytest_embedded.log import MessageQueue + + logfile = str(tmp_path / 'test.log') + q = MessageQueue() + patterns = [('<<>>', '<<>>')] + + all_messages = [ + b'before\n', + b'<<>>\n', + b'muted_payload_1\n', + b'muted_payload_2\n', + b'<<>>\n', + b'after\n', + ] + + p = _ctx.Process( + target=_listen, + args=(q, logfile), + kwargs={'with_timestamp': False, 'mute_patterns': patterns}, + ) + p.start() + try: + for msg in all_messages: + q.put(msg) + + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + try: + content = open(logfile, 'rb').read() + if b'after' in content: + break + except OSError: + pass + time.sleep(0.05) + finally: + p.terminate() + p.join(timeout=5) + + content = open(logfile, 'rb').read() + for msg in all_messages: + assert msg in content, f'{msg!r} should be in logfile regardless of muting' + + +def test_listen_mute_patterns_multiple_pairs(tmp_path): + """Multiple mute_patterns pairs can be registered simultaneously.""" + import time + + from pytest_embedded.dut_factory import _ctx, _listen + from pytest_embedded.log import MessageQueue + + logfile = str(tmp_path / 'test.log') + q = MessageQueue() + patterns = [ + ('<<>>', '<<>>'), + ('<<>>', '<<>>'), + ] + + all_messages = [ + b'normal_1\n', + b'<<>>\n', + b'gcov_data\n', + b'<<>>\n', + b'normal_2\n', + b'<<>>\n', + b'diag_data\n', + b'<<>>\n', + b'normal_3\n', + ] + + p = _ctx.Process( + target=_listen, + args=(q, logfile), + kwargs={'with_timestamp': False, 'mute_patterns': patterns}, + ) + p.start() + try: + for msg in all_messages: + q.put(msg) + + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + try: + content = open(logfile, 'rb').read() + if b'normal_3' in content: + break + except OSError: + pass + time.sleep(0.05) + finally: + p.terminate() + p.join(timeout=5) + + content = open(logfile, 'rb').read() + for msg in all_messages: + assert msg in content + + +def test_listen_empty_mute_patterns_no_effect(tmp_path): + """Empty mute_patterns tuple has no effect — same as not passing it.""" + import time + + from pytest_embedded.dut_factory import _ctx, _listen + from pytest_embedded.log import MessageQueue + + logfile = str(tmp_path / 'test.log') + q = MessageQueue() + messages = [f'pass_through_{i}\n'.encode() for i in range(10)] + + p = _ctx.Process( + target=_listen, + args=(q, logfile), + kwargs={'with_timestamp': False, 'mute_patterns': ()}, + ) + p.start() + try: + for msg in messages: + q.put(msg) + + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + try: + content = open(logfile, 'rb').read() + if all(msg in content for msg in messages): + break + except OSError: + pass + time.sleep(0.05) + finally: + p.terminate() + p.join(timeout=5) + + content = open(logfile, 'rb').read() + for msg in messages: + assert msg in content + + +def test_listen_mute_patterns_wrong_end_does_not_unmute(tmp_path): + """Matching startA then endB must NOT unmute — only endA should.""" + import time + + from pytest_embedded.dut_factory import _ctx, _listen + from pytest_embedded.log import MessageQueue + + logfile = str(tmp_path / 'test.log') + q = MessageQueue() + patterns = [ + ('<<>>', '<<>>'), + ('<<>>', '<<>>'), + ] + + all_messages = [ + b'visible_before\n', + b'<<>>\n', + b'should_be_muted\n', + b'<<>>\n', + b'still_muted_after_wrong_end\n', + b'<<>>\n', + b'visible_after\n', + ] + + p = _ctx.Process( + target=_listen, + args=(q, logfile), + kwargs={'with_timestamp': False, 'mute_patterns': patterns}, + ) + p.start() + try: + for msg in all_messages: + q.put(msg) + + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + try: + content = open(logfile, 'rb').read() + if b'visible_after' in content: + break + except OSError: + pass + time.sleep(0.05) + finally: + p.terminate() + p.join(timeout=5) + + content = open(logfile, 'rb').read() + for msg in all_messages: + assert msg in content, f'{msg!r} must always be written to the logfile' + + +def test_listen_mute_patterns_split_across_chunks(tmp_path): + """A mute pattern split across two serial chunks must still trigger muting.""" + import time + + from pytest_embedded.dut_factory import _ctx, _listen + from pytest_embedded.log import MessageQueue + + logfile = str(tmp_path / 'test.log') + q = MessageQueue() + patterns = [('<<>>', '<<>>')] + + all_messages = [ + b'before\n', + b'<<>>\n', + b'muted_payload\n', + b'<<>>\n', + b'after\n', + ] + + p = _ctx.Process( + target=_listen, + args=(q, logfile), + kwargs={'with_timestamp': False, 'mute_patterns': patterns}, + ) + p.start() + try: + for msg in all_messages: + q.put(msg) + + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + try: + content = open(logfile, 'rb').read() + if b'after' in content: + break + except OSError: + pass + time.sleep(0.05) + finally: + p.terminate() + p.join(timeout=5) + + content = open(logfile, 'rb').read() + for msg in all_messages: + assert msg in content, f'{msg!r} must be in logfile' + + +def test_listen_mute_patterns_both_in_one_chunk(tmp_path): + """When one chunk contains both start and end markers, payload must stay off stdout.""" + import time + + from pytest_embedded.dut_factory import _ctx + from pytest_embedded.log import MessageQueue + + logfile = str(tmp_path / 'test.log') + stdout_file = str(tmp_path / 'stdout.log') + q = MessageQueue() + patterns = [('<<>>', '<<>>')] + + all_messages = [ + b'before\n', + b'<<>>payload<<>>\n', + b'after\n', + ] + + p = _ctx.Process( + target=_run_listen_with_stdout_capture, + args=(q, logfile, stdout_file), + kwargs={'listen_kwargs': {'with_timestamp': False, 'mute_patterns': patterns}}, + ) + p.start() + try: + for msg in all_messages: + q.put(msg) + + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + try: + content = open(logfile, 'rb').read() + if b'after' in content: + break + except OSError: + pass + time.sleep(0.05) + finally: + p.terminate() + p.join(timeout=5) + + content = open(logfile, 'rb').read() + for msg in all_messages: + assert msg in content, f'{msg!r} must be in logfile' + + stdout_content = open(stdout_file, encoding='utf-8').read() + assert 'before' in stdout_content + assert 'after' in stdout_content + assert 'payload' not in stdout_content + assert '<<>>' not in stdout_content + assert '<<>>' not in stdout_content + + +# --------------------------------------------------------------------------- +# Tests for Dut mute_echo / unmute_echo / muted_echo +# --------------------------------------------------------------------------- + + +def _make_dut_with_pipe(tmp_path, mute_event=None): + """Helper: create a Dut backed by an OS pipe for unit testing.""" + import os + + from pytest_embedded.dut import Dut + from pytest_embedded.log import MessageQueue, PexpectProcess + + rd, wr = os.pipe() + fr = os.fdopen(rd, 'rb', 0) + fw = os.fdopen(wr, 'wb', 0) + proc = PexpectProcess(fr) + q = MessageQueue() + dut = Dut( + pexpect_proc=proc, + msg_queue=q, + app=None, + pexpect_logfile=str(tmp_path / 'dut.log'), + test_case_name='test', + _mute_event=mute_event, + ) + return dut, fw + + +def test_dut_mute_echo_sets_event(tmp_path): + """Dut.mute_echo() sets the event; unmute_echo() clears it.""" + from pytest_embedded.dut_factory import _ctx + + event = _ctx.Event() + assert not event.is_set() + + dut, fw = _make_dut_with_pipe(tmp_path, mute_event=event) + + dut.mute_echo() + assert event.is_set() + + dut.unmute_echo() + assert not event.is_set() + fw.close() + + +def test_dut_muted_echo_context_manager(tmp_path): + """muted_echo() context manager sets event on entry, clears on exit.""" + from pytest_embedded.dut_factory import _ctx + + event = _ctx.Event() + dut, fw = _make_dut_with_pipe(tmp_path, mute_event=event) + + assert not event.is_set() + with dut.muted_echo(): + assert event.is_set() + assert not event.is_set() + fw.close() + + +def test_dut_mute_echo_without_event_is_noop(tmp_path): + """mute_echo / unmute_echo are no-ops when _mute_event is None (backwards compat).""" + dut, fw = _make_dut_with_pipe(tmp_path, mute_event=None) + + dut.mute_echo() + dut.unmute_echo() + with dut.muted_echo(): + pass + fw.close() + + +# --------------------------------------------------------------------------- +# Tests for Dut.capture_payload / capture_payload_to_file +# --------------------------------------------------------------------------- + + +def test_capture_payload_returns_bytes_between_markers(tmp_path): + """capture_payload returns the raw bytes between start and end markers.""" + import threading + + from pytest_embedded.dut import Dut + from pytest_embedded.log import MessageQueue, PexpectProcess + + q = MessageQueue() + logfile = str(tmp_path / 'dut.log') + + # PexpectProcess uses the file reader end for pexpect + import os + + _pexpect_fr_rd, _pexpect_fr_wr = os.pipe() + fr = os.fdopen(_pexpect_fr_rd, 'rb', 0) + fw = os.fdopen(_pexpect_fr_wr, 'wb', 0) + proc = PexpectProcess(fr) + + dut = Dut( + pexpect_proc=proc, + msg_queue=q, + app=None, + pexpect_logfile=logfile, + test_case_name='test_cap', + _mute_event=None, + ) + + payload = b'<<>>\nHELLO_PAYLOAD\n<<>>\n' + + def _write(): + import time + + time.sleep(0.2) + fw.write(payload) + fw.flush() + + t = threading.Thread(target=_write, daemon=True) + t.start() + + result = dut.capture_payload('<<>>', '<<>>', start_timeout=5, timeout=5) + t.join(timeout=5) + fw.close() + + assert result is not None + assert b'HELLO_PAYLOAD' in result + + +def test_capture_payload_returns_none_on_timeout(tmp_path): + """capture_payload returns None when start marker is not seen.""" + import os + + from pytest_embedded.dut import Dut + from pytest_embedded.log import MessageQueue, PexpectProcess + + _rd, _wr = os.pipe() + fr = os.fdopen(_rd, 'rb', 0) + fw = os.fdopen(_wr, 'wb', 0) + proc = PexpectProcess(fr) + + dut = Dut( + pexpect_proc=proc, + msg_queue=MessageQueue(), + app=None, + pexpect_logfile=str(tmp_path / 'dut.log'), + test_case_name='test_timeout', + ) + + result = dut.capture_payload('<<>>', '<<>>', start_timeout=0.5, timeout=0.5) + fw.close() + + assert result is None + + +def test_capture_payload_to_file_writes_data(tmp_path): + """capture_payload_to_file creates a file with the captured payload.""" + import os + import threading + + from pytest_embedded.dut import Dut + from pytest_embedded.log import MessageQueue, PexpectProcess + + _rd, _wr = os.pipe() + fr = os.fdopen(_rd, 'rb', 0) + fw = os.fdopen(_wr, 'wb', 0) + proc = PexpectProcess(fr) + + out_file = str(tmp_path / 'captured.txt') + + dut = Dut( + pexpect_proc=proc, + msg_queue=MessageQueue(), + app=None, + pexpect_logfile=str(tmp_path / 'dut.log'), + test_case_name='test_cap_file', + _mute_event=None, + ) + + payload = b'<<>>\nDATA_LINE_1\nDATA_LINE_2\n<<>>\n' + + def _write(): + import time + + time.sleep(0.2) + fw.write(payload) + fw.flush() + + t = threading.Thread(target=_write, daemon=True) + t.start() + + ok = dut.capture_payload_to_file( + '<<>>', + '<<>>', + filepath=out_file, + start_timeout=5, + timeout=5, + mute=False, + append=False, + include_markers=True, + ) + t.join(timeout=5) + fw.close() + + assert ok is True + content = open(out_file).read() + assert '<<>>' in content + assert 'DATA_LINE_1' in content + assert 'DATA_LINE_2' in content + assert '<<>>' in content + + +def test_capture_payload_to_file_append_mode(tmp_path): + """capture_payload_to_file with append=True accumulates data across calls.""" + import os + import threading + + from pytest_embedded.dut import Dut + from pytest_embedded.log import MessageQueue, PexpectProcess + + _rd, _wr = os.pipe() + fr = os.fdopen(_rd, 'rb', 0) + fw = os.fdopen(_wr, 'wb', 0) + proc = PexpectProcess(fr) + + out_file = str(tmp_path / 'appended.txt') + + dut = Dut( + pexpect_proc=proc, + msg_queue=MessageQueue(), + app=None, + pexpect_logfile=str(tmp_path / 'dut.log'), + test_case_name='test_append', + _mute_event=None, + ) + + def _write_two_payloads(): + import time + + time.sleep(0.2) + fw.write(b'<<>>\nBOOT1_DATA\n<<>>\n') + fw.flush() + time.sleep(0.2) + fw.write(b'<<>>\nBOOT2_DATA\n<<>>\n') + fw.flush() + + t = threading.Thread(target=_write_two_payloads, daemon=True) + t.start() + + ok1 = dut.capture_payload_to_file('<<>>', '<<>>', filepath=out_file, start_timeout=5, timeout=5, mute=False) + ok2 = dut.capture_payload_to_file('<<>>', '<<>>', filepath=out_file, start_timeout=5, timeout=5, mute=False) + t.join(timeout=5) + fw.close() + + assert ok1 is True + assert ok2 is True + content = open(out_file).read() + assert 'BOOT1_DATA' in content + assert 'BOOT2_DATA' in content + + +def test_capture_payload_to_file_returns_false_on_timeout(tmp_path): + """capture_payload_to_file returns False when start marker is never seen.""" + import os + + from pytest_embedded.dut import Dut + from pytest_embedded.log import MessageQueue, PexpectProcess + + _rd, _wr = os.pipe() + fr = os.fdopen(_rd, 'rb', 0) + fw = os.fdopen(_wr, 'wb', 0) + proc = PexpectProcess(fr) + + out_file = str(tmp_path / 'nodata.txt') + + dut = Dut( + pexpect_proc=proc, + msg_queue=MessageQueue(), + app=None, + pexpect_logfile=str(tmp_path / 'dut.log'), + test_case_name='test_no_cap', + ) + + ok = dut.capture_payload_to_file( + '<<>>', '<<>>', filepath=out_file, start_timeout=0.5, timeout=0.5, mute=False + ) + fw.close() + + assert ok is False + assert not os.path.exists(out_file) + + +# --------------------------------------------------------------------------- +# Tests for mute_patterns / _mute_event fixtures (via testdir) +# --------------------------------------------------------------------------- + + +def test_mute_patterns_fixture_default(testdir): + """The default mute_patterns fixture returns an empty tuple.""" + testdir.makepyfile(""" + def test_default_patterns(mute_patterns): + assert mute_patterns == () + """) + + result = testdir.runpytest() + result.assert_outcomes(passed=1) + + +def test_mute_event_fixture_exists(testdir): + """The _mute_event fixture is created and is usable.""" + testdir.makepyfile(""" + def test_event_exists(_mute_event): + assert _mute_event is not None + assert not _mute_event.is_set() + _mute_event.set() + assert _mute_event.is_set() + _mute_event.clear() + """) + + result = testdir.runpytest() + result.assert_outcomes(passed=1) + + +def test_mute_patterns_override_in_conftest(testdir): + """Projects can override mute_patterns in conftest.py to register custom patterns.""" + testdir.makeconftest(""" + import pytest + + @pytest.fixture + def mute_patterns(): + return [("<<>>", "<<>>")] + """) + testdir.makepyfile(""" + def test_custom_patterns(mute_patterns): + assert len(mute_patterns) == 1 + assert mute_patterns[0] == ("<<>>", "<<>>") + """) + + result = testdir.runpytest() + result.assert_outcomes(passed=1) + + +# --------------------------------------------------------------------------- +# Integration: mute_patterns + dut echo suppression (via testdir) +# --------------------------------------------------------------------------- + + +def test_mute_patterns_integration_with_dut(testdir): + """mute_patterns auto-mutes output between markers but data still reaches pexpect.""" + testdir.makeconftest(""" + import pytest + + @pytest.fixture + def mute_patterns(): + return [("<<>>", "<<>>")] + """) + testdir.makepyfile(""" + def test_muted_payload(dut, redirect): + with redirect(): + print("before_marker") + print("<<>>") + print("secret_payload_data") + print("<<>>") + print("after_marker") + + dut.expect_exact("before_marker") + dut.expect_exact("<<>>") + dut.expect_exact("secret_payload_data") + dut.expect_exact("<<>>") + dut.expect_exact("after_marker") + """) + + result = testdir.runpytest('-s') + result.assert_outcomes(passed=1) + + +def test_dut_capture_payload_integration(testdir): + """capture_payload works inside a real pytest-embedded session.""" + testdir.makepyfile(""" + import threading + + def test_capture(dut, redirect): + def _emit(): + import time + time.sleep(0.3) + with redirect(): + print("<<>>") + print("captured_line_1") + print("captured_line_2") + print("<<>>") + + t = threading.Thread(target=_emit, daemon=True) + t.start() + + data = dut.capture_payload("<<>>", "<<>>", start_timeout=5, timeout=5) + t.join(timeout=5) + + assert data is not None + text = data.decode('utf-8', errors='replace') + assert "captured_line_1" in text + assert "captured_line_2" in text + """) + + result = testdir.runpytest('-s') + result.assert_outcomes(passed=1) + + +def test_backwards_compat_no_mute_patterns_no_event(testdir): + """Tests that don't use mute_patterns or _mute_event work identically to before.""" + testdir.makepyfile(""" + def test_basic_expect(dut, redirect): + with redirect(): + print("hello world") + + dut.expect_exact("hello world") + """) + + result = testdir.runpytest('-s') + result.assert_outcomes(passed=1) + + +def test_multi_dut_with_mute_patterns(testdir): + """mute_patterns works correctly in multi-DUT mode without IndexError.""" + testdir.makeconftest(""" + import pytest + + @pytest.fixture + def mute_patterns(): + return [("<<>>", "<<>>")] + """) + testdir.makepyfile(r""" + import pytest + + @pytest.mark.parametrize('count', [2], indirect=True) + def test_multi_dut_mute(dut): + dut[0].write(b'hello_from_dut0') + dut[1].write(b'hello_from_dut1') + + dut[0].expect_exact('hello_from_dut0') + dut[1].expect_exact('hello_from_dut1') + """) + + result = testdir.runpytest('-s') + result.assert_outcomes(passed=1) + + +def test_multi_dut_default_mute_patterns(testdir): + """Default empty mute_patterns does not break multi-DUT mode.""" + testdir.makepyfile(r""" + import pytest + + @pytest.mark.parametrize('count', [2], indirect=True) + def test_multi_dut_default(dut): + dut[0].write(b'msg_a') + dut[1].write(b'msg_b') + + dut[0].expect_exact('msg_a') + dut[1].expect_exact('msg_b') + """) + + result = testdir.runpytest('-s') + result.assert_outcomes(passed=1)