From 4cbc9ce2bf355d9fa38f0be9d10010d3728b1ffd Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Wed, 13 Aug 2025 11:40:33 +0200 Subject: [PATCH 01/13] refactor: replace WokwiCLI with Wokwi class and update related references --- docs/apis/pytest-embedded-wokwi.rst | 2 +- pytest-embedded-wokwi/pyproject.toml | 3 + .../pytest_embedded_wokwi/__init__.py | 4 +- .../pytest_embedded_wokwi/dut.py | 4 +- .../pytest_embedded_wokwi/wokwi.py | 205 ++++++++++++++++++ .../pytest_embedded_wokwi/wokwi_cli.py | 172 --------------- .../pytest_embedded/dut_factory.py | 10 +- pytest-embedded/pytest_embedded/log.py | 32 ++- pytest-embedded/pytest_embedded/plugin.py | 6 +- 9 files changed, 243 insertions(+), 195 deletions(-) create mode 100644 pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py delete mode 100644 pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi_cli.py diff --git a/docs/apis/pytest-embedded-wokwi.rst b/docs/apis/pytest-embedded-wokwi.rst index adf26760..5afa4cdf 100644 --- a/docs/apis/pytest-embedded-wokwi.rst +++ b/docs/apis/pytest-embedded-wokwi.rst @@ -17,7 +17,7 @@ :undoc-members: :show-inheritance: -.. automodule:: pytest_embedded_wokwi.wokwi_cli +.. automodule:: pytest_embedded_wokwi.wokwi :members: :undoc-members: :show-inheritance: diff --git a/pytest-embedded-wokwi/pyproject.toml b/pytest-embedded-wokwi/pyproject.toml index e37a70a9..a7ec3276 100644 --- a/pytest-embedded-wokwi/pyproject.toml +++ b/pytest-embedded-wokwi/pyproject.toml @@ -7,6 +7,7 @@ name = "pytest-embedded-wokwi" authors = [ {name = "Fu Hanxi", email = "fuhanxi@espressif.com"}, {name = "Uri Shaked", email = "uri@wokwi.com"}, + {name = "Jakub Andrysek", email = "jakub.andrysek@espressif.com"}, ] readme = "README.md" license = {file = "LICENSE"} @@ -34,6 +35,8 @@ requires-python = ">=3.7" dependencies = [ "pytest-embedded~=1.17.0a2", "toml~=0.10.2", + # Temporary workaround for Wokwi client - will be redirected to the official repo + "wokwi-client @ git+https://github.com/JakubAndrysek/wokwi-python-client.git@sync-wokwi-client", ] [project.optional-dependencies] diff --git a/pytest-embedded-wokwi/pytest_embedded_wokwi/__init__.py b/pytest-embedded-wokwi/pytest_embedded_wokwi/__init__.py index 41fae2e4..9324bcce 100644 --- a/pytest-embedded-wokwi/pytest_embedded_wokwi/__init__.py +++ b/pytest-embedded-wokwi/pytest_embedded_wokwi/__init__.py @@ -3,11 +3,11 @@ WOKWI_CLI_MINIMUM_VERSION = '0.10.1' from .dut import WokwiDut # noqa -from .wokwi_cli import WokwiCLI # noqa +from .wokwi import Wokwi # noqa __all__ = [ 'WOKWI_CLI_MINIMUM_VERSION', - 'WokwiCLI', + 'Wokwi', 'WokwiDut', ] diff --git a/pytest-embedded-wokwi/pytest_embedded_wokwi/dut.py b/pytest-embedded-wokwi/pytest_embedded_wokwi/dut.py index 132eec7b..60f9ba43 100644 --- a/pytest-embedded-wokwi/pytest_embedded_wokwi/dut.py +++ b/pytest-embedded-wokwi/pytest_embedded_wokwi/dut.py @@ -2,7 +2,7 @@ from pytest_embedded.dut import Dut -from .wokwi_cli import WokwiCLI +from .wokwi import Wokwi class WokwiDut(Dut): @@ -12,7 +12,7 @@ class WokwiDut(Dut): def __init__( self, - wokwi: WokwiCLI, + wokwi: Wokwi, **kwargs, ) -> None: self.wokwi = wokwi diff --git a/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py b/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py new file mode 100644 index 00000000..446b9c37 --- /dev/null +++ b/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py @@ -0,0 +1,205 @@ +import json +import logging +import os +import typing as t + +from packaging.version import Version +from pytest_embedded.log import DuplicateStdoutPopen, MessageQueue +from pytest_embedded.utils import Meta +from wokwi_client import GET_TOKEN_URL, WokwiClientSync + +from pytest_embedded_wokwi import WOKWI_CLI_MINIMUM_VERSION + +from .idf import IDFFirmwareResolver + +if t.TYPE_CHECKING: # pragma: no cover + from pytest_embedded_idf.app import IdfApp + + +target_to_board = { + 'esp32': 'board-esp32-devkit-c-v4', + 'esp32c3': 'board-esp32-c3-devkitm-1', + 'esp32c6': 'board-esp32-c6-devkitc-1', + 'esp32h2': 'board-esp32-h2-devkitm-1', + 'esp32p4': 'board-esp32-p4-function-ev', + 'esp32s2': 'board-esp32-s2-devkitm-1', + 'esp32s3': 'board-esp32-s3-devkitc-1', +} + + +class Wokwi(DuplicateStdoutPopen): + """Synchronous Wokwi integration that inherits from DuplicateStdoutPopen. + + This class provides a synchronous interface to the Wokwi simulator while maintaining + compatibility with pytest-embedded's logging and message queue infrastructure. + """ + + SOURCE = 'Wokwi' + REDIRECT_CLS = None # We'll handle output redirection manually + + def __init__( + self, + msg_queue: MessageQueue, + firmware_resolver: IDFFirmwareResolver, + wokwi_diagram: t.Optional[str] = None, + app: t.Optional['IdfApp'] = None, + _wokwi_cli_path: t.Optional[str] = None, # ignored for compatibility + _wokwi_timeout: t.Optional[int] = None, # ignored for compatibility + _wokwi_scenario: t.Optional[str] = None, # ignored for compatibility + meta: t.Optional[Meta] = None, + **kwargs, + ): + self.app = app + self.firmware_resolver = firmware_resolver + + # Get Wokwi API token + token = os.getenv('WOKWI_CLI_TOKEN') + if not token: + raise SystemExit(f'Set WOKWI_CLI_TOKEN in your environment. You can get it from {GET_TOKEN_URL}.') + + # Initialize synchronous Wokwi client + self.client = WokwiClientSync(token) + + # Check version compatibility + if Version(self.client.version) < Version(WOKWI_CLI_MINIMUM_VERSION): + logging.warning( + 'Wokwi client version %s < required %s (compatibility not guaranteed)', + self.client.version, + WOKWI_CLI_MINIMUM_VERSION, + ) + logging.info('Wokwi client library version: %s', self.client.version) + + # Prepare diagram file if not supplied + if wokwi_diagram is None: + self.create_diagram_json() + wokwi_diagram = os.path.join(self.app.app_path, 'diagram.json') + + # Filter out Wokwi-specific kwargs that shouldn't be passed to subprocess.Popen + wokwi_specific_kwargs = {'wokwi_timeout', 'wokwi_scenario', 'wokwi_diagram', 'firmware_resolver', 'app'} + filtered_kwargs = {k: v for k, v in kwargs.items() if k not in wokwi_specific_kwargs} + + # Initialize parent class + super().__init__(msg_queue=msg_queue, meta=meta, **filtered_kwargs) + + # Connect and start simulation + try: + firmware_path = self.firmware_resolver.resolve_firmware(app) + self._setup_simulation(wokwi_diagram, firmware_path, app.elf_file) + self._start_serial_monitoring() + except Exception as e: + self.close() + raise e + + def _setup_simulation(self, diagram: str, firmware_path: str, elf_path: str): + """Set up the Wokwi simulation.""" + hello = self.client.connect() + logging.info('Connected to Wokwi Simulator, server version: %s', hello.get('version', 'unknown')) + + # Upload files + self.client.upload_file('diagram.json', diagram) + self.client.upload_file('pytest.bin', firmware_path) + self.client.upload_file('pytest.elf', elf_path) + + # Start simulation + self.client.start_simulation(firmware='pytest.bin', elf='pytest.elf') + + 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: + decoded = data.decode('utf-8', errors='replace') + self._fw.write(decoded) + self._fw.flush() + except Exception as e: + logging.debug(f'Error writing to log file: {e}') + + # Put in message queue for expect() functionality + try: + if hasattr(self, '_q') and self._q: + self._q.put(data) + except Exception as e: + logging.debug(f'Error putting data in message queue: {e}') + + # Start monitoring in background + self.client.monitor_serial(serial_callback) + + def write(self, s: t.Union[str, bytes]) -> None: + """Write data to the Wokwi serial interface.""" + try: + data = s if isinstance(s, bytes) else s.encode('utf-8') + self.client.write_serial(data) + logging.debug(f'{self.SOURCE} ->: {s}') + except Exception as e: + logging.error(f'Failed to write to Wokwi serial: {e}') + + def close(self): + """Clean up resources.""" + try: + if hasattr(self, 'client') and self.client: + self.client.disconnect() + except Exception as e: + logging.debug(f'Error during Wokwi cleanup: {e}') + finally: + super().close() + + def __del__(self): + """Destructor to ensure cleanup when object is garbage collected.""" + self.close() + + def terminate(self): + """Terminate the Wokwi connection.""" + self.close() + super().terminate() + + def create_diagram_json(self): + """Create a diagram.json file for the simulation.""" + app = self.app + target_board = target_to_board[app.target] + + # Check for existing diagram.json file + diagram_json_path = os.path.join(app.app_path, 'diagram.json') + if os.path.exists(diagram_json_path): + with open(diagram_json_path) as f: + json_data = json.load(f) + if not any(part['type'] == target_board for part in json_data['parts']): + logging.warning( + f'diagram.json exists, no part with type "{target_board}" found. ' + + 'You may need to update the diagram.json file manually to match the target board.' + ) + return + + # Create default diagram + if app.target == 'esp32p4': + rx_pin = '38' + tx_pin = '37' + else: + rx_pin = 'RX' + tx_pin = 'TX' + + diagram = { + 'version': 1, + 'author': 'Uri Shaked', + 'editor': 'wokwi', + 'parts': [{'type': target_board, 'id': 'esp'}], + 'connections': [ + ['esp:' + tx_pin, '$serialMonitor:RX', ''], + ['esp:' + rx_pin, '$serialMonitor:TX', ''], + ], + } + + with open(diagram_json_path, 'w') as f: + json.dump(diagram, f, indent=2) + + def _hard_reset(self): + """Fake hard_reset to maintain API consistency.""" + raise NotImplementedError('Hard reset not supported in Wokwi simulation') diff --git a/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi_cli.py b/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi_cli.py deleted file mode 100644 index eba104f2..00000000 --- a/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi_cli.py +++ /dev/null @@ -1,172 +0,0 @@ -import json -import logging -import os -import re -import shutil -import subprocess -import typing as t -from pathlib import Path - -import toml -from packaging.version import Version -from pytest_embedded import __version__ -from pytest_embedded.log import DuplicateStdoutPopen - -from pytest_embedded_wokwi import WOKWI_CLI_MINIMUM_VERSION - -from .idf import IDFFirmwareResolver - -if t.TYPE_CHECKING: - from pytest_embedded_idf.app import IdfApp - - -target_to_board = { - 'esp32': 'board-esp32-devkit-c-v4', - 'esp32c3': 'board-esp32-c3-devkitm-1', - 'esp32c6': 'board-esp32-c6-devkitc-1', - 'esp32h2': 'board-esp32-h2-devkitm-1', - 'esp32p4': 'board-esp32-p4-function-ev', - 'esp32s2': 'board-esp32-s2-devkitm-1', - 'esp32s3': 'board-esp32-s3-devkitc-1', -} - - -class WokwiCLI(DuplicateStdoutPopen): - """ - WokwiCLI class - """ - - SOURCE = 'Wokwi' - WOKWI_CLI_PATH = 'wokwi-cli' - - def __init__( - self, - firmware_resolver: IDFFirmwareResolver, - wokwi_cli_path: t.Optional[str] = None, - wokwi_timeout: t.Optional[int] = None, - wokwi_scenario: t.Optional[str] = None, - wokwi_diagram: t.Optional[str] = None, - app: t.Optional['IdfApp'] = None, - **kwargs, - ): - """ - Args: - wokwi_cli_path: Wokwi CLI arguments - """ - self.app = app - self.firmware_resolver = firmware_resolver - - # first need to check if wokwi-cli exists in PATH - if shutil.which('wokwi-cli') is None: - raise RuntimeError('Please install wokwi-cli, by running: curl -L https://wokwi.com/ci/install.sh | sh') - - output = subprocess.check_output(['wokwi-cli', '--help']) - try: - wokwi_cli_version = re.match(r'Wokwi CLI v(\d+\.\d+\.\d+)', output.decode('utf-8')).group(1) - except AttributeError: - logging.warning('Failed to get wokwi-cli version, assume version requirements satisfied') - else: - if Version(wokwi_cli_version) < Version(WOKWI_CLI_MINIMUM_VERSION): - raise ValueError( - f'Wokwi CLI version {wokwi_cli_version} is not supported. ' - f'Minimum version required: {WOKWI_CLI_MINIMUM_VERSION}. ' - f'To update Wokwi CLI run: curl -L https://wokwi.com/ci/install.sh | sh' - ) - - self.create_wokwi_toml() - - if wokwi_diagram is None: - self.create_diagram_json() - - wokwi_cli = wokwi_cli_path or self.wokwi_cli_executable - cmd = [wokwi_cli, '--interactive', app.app_path] - if (wokwi_timeout is not None) and (wokwi_timeout > 0): - cmd.extend(['--timeout', str(wokwi_timeout)]) - if (wokwi_scenario is not None) and os.path.exists(wokwi_scenario): - cmd.extend(['--scenario', wokwi_scenario]) - if (wokwi_diagram is not None) and os.path.exists(wokwi_diagram): - cmd.extend(['--diagram-file', wokwi_diagram]) - - super().__init__( - cmd=cmd, - **kwargs, - ) - - @property - def wokwi_cli_executable(self): - return self.WOKWI_CLI_PATH - - def create_wokwi_toml(self): - app = self.app - flasher_args = self.firmware_resolver.resolve_firmware(app) - wokwi_toml_path = os.path.join(app.app_path, 'wokwi.toml') - firmware_path = Path(os.path.relpath(flasher_args, app.app_path)).as_posix() - elf_path = Path(os.path.relpath(app.elf_file, app.app_path)).as_posix() - - if os.path.exists(wokwi_toml_path): - with open(wokwi_toml_path) as f: - toml_data = toml.load(f) - - if 'wokwi' not in toml_data: - toml_data['wokwi'] = {'version': 1} - - wokwi_table = toml_data['wokwi'] - if wokwi_table.get('firmware') == firmware_path and wokwi_table.get('elf') == elf_path: - # No need to update - return - - wokwi_table.update({'firmware': firmware_path, 'elf': elf_path}) - else: - toml_data = { - 'wokwi': { - 'version': 1, - 'generatedBy': f'pytest-embedded-wokwi {__version__}', - 'firmware': firmware_path, - 'elf': elf_path, - } - } - - with open(wokwi_toml_path, 'w') as f: - toml.dump(toml_data, f) - - def create_diagram_json(self): - app = self.app - target_board = target_to_board[app.target] - - # Check for common diagram.json file - diagram_json_path = os.path.join(app.app_path, 'diagram.json') - if os.path.exists(diagram_json_path): - with open(diagram_json_path) as f: - json_data = json.load(f) - if not any(part['type'] == target_board for part in json_data['parts']): - logging.warning( - f'diagram.json exists, no part with type "{target_board}" found. ' - + 'You may need to update the diagram.json file manually to match the target board.' - ) - return - - if app.target == 'esp32p4': - rx_pin = '38' - tx_pin = '37' - else: - rx_pin = 'RX' - tx_pin = 'TX' - - diagram = { - 'version': 1, - 'author': 'Uri Shaked', - 'editor': 'wokwi', - 'parts': [{'type': target_board, 'id': 'esp'}], - 'connections': [ - ['esp:' + tx_pin, '$serialMonitor:RX', ''], - ['esp:' + rx_pin, '$serialMonitor:TX', ''], - ], - } - with open(diagram_json_path, 'w') as f: - f.write(json.dumps(diagram, indent=2)) - - def _hard_reset(self): - """ - This is a fake hard_reset. Keep this API to keep the consistency. - """ - raise NotImplementedError diff --git a/pytest-embedded/pytest_embedded/dut_factory.py b/pytest-embedded/pytest_embedded/dut_factory.py index ba2ed88a..dd529a03 100644 --- a/pytest-embedded/pytest_embedded/dut_factory.py +++ b/pytest-embedded/pytest_embedded/dut_factory.py @@ -17,7 +17,7 @@ from pytest_embedded_jtag import Gdb, OpenOcd from pytest_embedded_qemu import Qemu from pytest_embedded_serial import Serial - from pytest_embedded_wokwi import WokwiCLI + from pytest_embedded_wokwi import Wokwi from . import App, Dut from .log import MessageQueue, PexpectProcess @@ -309,9 +309,9 @@ def _fixture_classes_and_options_fn( } elif fixture == 'wokwi': if 'wokwi' in _services: - from pytest_embedded_wokwi import WokwiCLI + from pytest_embedded_wokwi import Wokwi - classes[fixture] = WokwiCLI + classes[fixture] = Wokwi kwargs[fixture].update({ 'wokwi_cli_path': wokwi_cli_path, 'wokwi_timeout': wokwi_timeout, @@ -495,7 +495,7 @@ def qemu_gn(_fixture_classes_and_options: ClassCliOptions, app) -> t.Optional['Q return cls(**_drop_none_kwargs(kwargs)) -def wokwi_gn(_fixture_classes_and_options: ClassCliOptions, app) -> t.Optional['WokwiCLI']: +def wokwi_gn(_fixture_classes_and_options: ClassCliOptions, app) -> t.Optional['Wokwi']: """A wokwi subprocess that could read/redirect/write""" if 'wokwi' not in _fixture_classes_and_options.classes: return None @@ -515,7 +515,7 @@ def dut_gn( app: App, serial: t.Optional[t.Union['Serial', 'LinuxSerial']], qemu: t.Optional['Qemu'], - wokwi: t.Optional['WokwiCLI'], + wokwi: t.Optional['Wokwi'], ) -> t.Union[Dut, t.List[Dut]]: global DUT_GLOBAL_INDEX DUT_GLOBAL_INDEX += 1 diff --git a/pytest-embedded/pytest_embedded/log.py b/pytest-embedded/pytest_embedded/log.py index 962e135f..800f9724 100644 --- a/pytest-embedded/pytest_embedded/log.py +++ b/pytest-embedded/pytest_embedded/log.py @@ -166,7 +166,7 @@ class DuplicateStdoutPopen(subprocess.Popen): SOURCE = 'POPEN' REDIRECT_CLS = _PopenRedirectProcess - def __init__(self, msg_queue: MessageQueue, cmd: Union[str, List[str]], meta: Optional[Meta] = None, **kwargs): + def __init__(self, msg_queue: MessageQueue, cmd: Union[str, List[str]] = [], meta: Optional[Meta] = None, **kwargs): self._q = msg_queue self._p = None @@ -188,16 +188,28 @@ def __init__(self, msg_queue: MessageQueue, cmd: Union[str, List[str]], meta: Op self._logfile_offset = 0 logging.debug(f'temp log file: {_log_file}') - kwargs.update({ - 'bufsize': 0, - 'stdin': subprocess.PIPE, - 'stdout': self._fw, - 'stderr': self._fw, - }) - self._cmd = cmd - logging.info('Executing %s', ' '.join(cmd) if isinstance(cmd, list) else cmd) - super().__init__(cmd, **kwargs) + + # Only start subprocess if command is not empty + if cmd and cmd != []: + kwargs.update({ + 'bufsize': 0, + 'stdin': subprocess.PIPE, + 'stdout': self._fw, + 'stderr': self._fw, + }) + + logging.info('Executing %s', ' '.join(cmd) if isinstance(cmd, list) else cmd) + super().__init__(cmd, **kwargs) + else: + # For empty commands, initialize minimal subprocess.Popen attributes + logging.debug('Empty command provided, not starting subprocess') + self.args = cmd + self.returncode = None + self.pid = None + self.stdin = None + self.stdout = None + self.stderr = None # some sub classes does not need to redirect to the message queue, they use blocking-IO instead and # return the response immediately in `write()` diff --git a/pytest-embedded/pytest_embedded/plugin.py b/pytest-embedded/pytest_embedded/plugin.py index b500efca..0808d4df 100644 --- a/pytest-embedded/pytest_embedded/plugin.py +++ b/pytest-embedded/pytest_embedded/plugin.py @@ -63,7 +63,7 @@ from pytest_embedded_jtag import Gdb, OpenOcd from pytest_embedded_qemu import Qemu from pytest_embedded_serial import Serial - from pytest_embedded_wokwi import WokwiCLI + from pytest_embedded_wokwi import Wokwi _T = t.TypeVar('_T') @@ -1174,7 +1174,7 @@ def qemu(_fixture_classes_and_options: ClassCliOptions, app) -> t.Optional['Qemu @pytest.fixture @multi_dut_generator_fixture -def wokwi(_fixture_classes_and_options: ClassCliOptions, app) -> t.Optional['WokwiCLI']: +def wokwi(_fixture_classes_and_options: ClassCliOptions, app) -> t.Optional['Wokwi']: """A wokwi subprocess that could read/redirect/write""" return wokwi_gn(**locals()) @@ -1188,7 +1188,7 @@ def dut( app: App, serial: t.Optional[t.Union['Serial', 'LinuxSerial']], qemu: t.Optional['Qemu'], - wokwi: t.Optional['WokwiCLI'], + wokwi: t.Optional['Wokwi'], ) -> t.Union[Dut, t.List[Dut]]: """ A device under test (DUT) object that could gather output from various sources and redirect them to the pexpect From 740989899dff3519b843fe74f2a17c0539fe0a7a Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Wed, 13 Aug 2025 13:59:02 +0200 Subject: [PATCH 02/13] refactor: remove deprecated Wokwi CLI options and related code --- pytest-embedded-wokwi/README.md | 44 ++++++++++++------- .../pytest_embedded_wokwi/wokwi.py | 10 +---- .../pytest_embedded/dut_factory.py | 15 ------- pytest-embedded/pytest_embedded/plugin.py | 38 ---------------- 4 files changed, 29 insertions(+), 78 deletions(-) diff --git a/pytest-embedded-wokwi/README.md b/pytest-embedded-wokwi/README.md index 8de5cd51..e5acb5f1 100644 --- a/pytest-embedded-wokwi/README.md +++ b/pytest-embedded-wokwi/README.md @@ -6,20 +6,6 @@ Wokwi supports most ESP32 targets, including: esp32, esp32s2, esp32s3, esp32c3, Running the tests with Wokwi requires an internet connection. Your firmware is uploaded to the Wokwi server for the duration of the simulation, but it is not saved on the server. On-premises Wokwi installations are available for enterprise customers. -#### Wokwi CLI installation - -The Wokwi plugin uses the [Wokwi CLI](https://github.com/wokwi/wokwi-cli) to interact with the wokwi simulation server. You can download the precompiled CLI binaries from the [releases page](https://github.com/wokwi/wokwi-cli/releases). Alternatively, on Linux or Mac OS, you can install the CLI using the following command: - -```bash -curl -L https://wokwi.com/ci/install.sh | sh -``` - -And on Windows: - -```powershell -iwr https://wokwi.com/ci/install.ps1 -useb | iex -``` - #### Wokwi API Tokens Before using this plugin, you need to create a free Wokwi account and [generate an API key](https://wokwi.com/dashboard/ci). You can then set the `WOKWI_CLI_TOKEN` environment variable to the API key. @@ -44,8 +30,32 @@ To run your tests with Wokwi, make sure to specify the `wokwi` service when runn pytest --embedded-services idf,wokwi ``` -To limit the amount of simulation time, use the `--wokwi-timeout` flag. For example, to set the simulation time limit to 60 seconds (60000 milliseconds): +#### Writing Tests -``` -pytest --embedded-services idf,wokwi --wokwi-timeout=60000 +When writing tests for your firmware, you can use the same pytest fixtures and assertions as you would for local testing. The main difference is that your tests will be executed in the Wokwi simulation environment and you have access to the Wokwi API for controlling the simulation through the `wokwi` fixture. + +All interactions with the Wokwi simulation is through the `wokwi.client` - [wokwi-python-client](https://github.com/wokwi/wokwi-python-client) + +For example, you can use `wokwi.client.set_control()` to control virtual components in the simulation, such as buttons, LEDs, and other peripherals. +Whole documentations can be found at [Wokwi Documentation](https://wokwi.github.io/wokwi-python-client/) + +Button test: +```py +import logging +from pytest_embedded_wokwi import Wokwi +from pytest_embedded import Dut + + +def test_gpio(dut: Dut, wokwi: Wokwi): + LOGGER = logging.getLogger(__name__) + + LOGGER.info("Waiting for Button test begin...") + dut.expect_exact("Butston test") + + for i in range(3): + LOGGER.info(f"Setting button pressed for {i + 1} seconds") + wokwi.client.set_control("btn1", "pressed", 1) + + dut.expect_exact(f"Button pressed {i + 1} times") + wokwi.client.set_control("btn1", "pressed", 0) ``` diff --git a/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py b/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py index 446b9c37..c48e21e0 100644 --- a/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py +++ b/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py @@ -43,9 +43,6 @@ def __init__( firmware_resolver: IDFFirmwareResolver, wokwi_diagram: t.Optional[str] = None, app: t.Optional['IdfApp'] = None, - _wokwi_cli_path: t.Optional[str] = None, # ignored for compatibility - _wokwi_timeout: t.Optional[int] = None, # ignored for compatibility - _wokwi_scenario: t.Optional[str] = None, # ignored for compatibility meta: t.Optional[Meta] = None, **kwargs, ): @@ -74,12 +71,8 @@ def __init__( self.create_diagram_json() wokwi_diagram = os.path.join(self.app.app_path, 'diagram.json') - # Filter out Wokwi-specific kwargs that shouldn't be passed to subprocess.Popen - wokwi_specific_kwargs = {'wokwi_timeout', 'wokwi_scenario', 'wokwi_diagram', 'firmware_resolver', 'app'} - filtered_kwargs = {k: v for k, v in kwargs.items() if k not in wokwi_specific_kwargs} - # Initialize parent class - super().__init__(msg_queue=msg_queue, meta=meta, **filtered_kwargs) + super().__init__(msg_queue=msg_queue, meta=meta, **kwargs) # Connect and start simulation try: @@ -155,6 +148,7 @@ def close(self): def __del__(self): """Destructor to ensure cleanup when object is garbage collected.""" self.close() + super().__del__() def terminate(self): """Terminate the Wokwi connection.""" diff --git a/pytest-embedded/pytest_embedded/dut_factory.py b/pytest-embedded/pytest_embedded/dut_factory.py index dd529a03..7fb5c18c 100644 --- a/pytest-embedded/pytest_embedded/dut_factory.py +++ b/pytest-embedded/pytest_embedded/dut_factory.py @@ -145,9 +145,6 @@ def _fixture_classes_and_options_fn( qemu_prog_path, qemu_cli_args, qemu_extra_args, - wokwi_cli_path, - wokwi_timeout, - wokwi_scenario, wokwi_diagram, skip_regenerate_image, encrypt, @@ -313,9 +310,6 @@ def _fixture_classes_and_options_fn( classes[fixture] = Wokwi kwargs[fixture].update({ - 'wokwi_cli_path': wokwi_cli_path, - 'wokwi_timeout': wokwi_timeout, - 'wokwi_scenario': wokwi_scenario, 'wokwi_diagram': wokwi_diagram, 'msg_queue': msg_queue, 'app': None, @@ -659,9 +653,6 @@ def create( qemu_prog_path: t.Optional[str] = None, qemu_cli_args: t.Optional[str] = None, qemu_extra_args: t.Optional[str] = None, - wokwi_cli_path: t.Optional[str] = None, - wokwi_timeout: t.Optional[int] = 0, - wokwi_scenario: t.Optional[str] = None, wokwi_diagram: t.Optional[str] = None, skip_regenerate_image: t.Optional[bool] = None, encrypt: t.Optional[bool] = None, @@ -708,9 +699,6 @@ def create( qemu_prog_path: QEMU program path. qemu_cli_args: QEMU CLI arguments. qemu_extra_args: Additional QEMU arguments. - wokwi_cli_path: Wokwi CLI path. - wokwi_timeout: Wokwi timeout. - wokwi_scenario: Wokwi scenario path. wokwi_diagram: Wokwi diagram path. skip_regenerate_image: Skip image regeneration flag. encrypt: Encryption flag. @@ -773,9 +761,6 @@ def create( 'qemu_prog_path': qemu_prog_path, 'qemu_cli_args': qemu_cli_args, 'qemu_extra_args': qemu_extra_args, - 'wokwi_cli_path': wokwi_cli_path, - 'wokwi_timeout': wokwi_timeout, - 'wokwi_scenario': wokwi_scenario, 'wokwi_diagram': wokwi_diagram, 'skip_regenerate_image': skip_regenerate_image, 'encrypt': encrypt, diff --git a/pytest-embedded/pytest_embedded/plugin.py b/pytest-embedded/pytest_embedded/plugin.py index 0808d4df..a0b7cf5a 100644 --- a/pytest-embedded/pytest_embedded/plugin.py +++ b/pytest-embedded/pytest_embedded/plugin.py @@ -297,20 +297,6 @@ def pytest_addoption(parser): ) wokwi_group = parser.getgroup('embedded-wokwi') - wokwi_group.addoption( - '--wokwi-cli-path', - help='Path to the wokwi-cli program (Default: "wokwi-cli")', - ) - wokwi_group.addoption( - '--wokwi-timeout', - default=86400000, - type=_gte_one_int, - help='Simulation timeout in milliseconds (Default: 86400000)', - ) - wokwi_group.addoption( - '--wokwi-scenario', - help='Path to the wokwi scenario file (Default: None)', - ) wokwi_group.addoption( '--wokwi-diagram', help='Path to the wokwi diagram file (Default: None)', @@ -995,27 +981,6 @@ def keyfile(request: FixtureRequest) -> t.Optional[str]: ######### # Wokwi # ######### -@pytest.fixture -@multi_dut_argument -def wokwi_cli_path(request: FixtureRequest) -> t.Optional[str]: - """Enable parametrization for the same cli option""" - return _request_param_or_config_option_or_default(request, 'wokwi_cli_path', None) - - -@pytest.fixture -@multi_dut_argument -def wokwi_timeout(request: FixtureRequest) -> t.Optional[str]: - """Enable parametrization for the same cli option""" - return _request_param_or_config_option_or_default(request, 'wokwi_timeout', None) - - -@pytest.fixture -@multi_dut_argument -def wokwi_scenario(request: FixtureRequest) -> t.Optional[str]: - """Enable parametrization for the same cli option""" - return _request_param_or_config_option_or_default(request, 'wokwi_scenario', None) - - @pytest.fixture @multi_dut_argument def wokwi_diagram(request: FixtureRequest) -> t.Optional[str]: @@ -1079,9 +1044,6 @@ def parametrize_fixtures( qemu_prog_path, qemu_cli_args, qemu_extra_args, - wokwi_cli_path, - wokwi_timeout, - wokwi_scenario, wokwi_diagram, skip_regenerate_image, encrypt, From e6f0697255e41edec5ec51e9b76ac35df5e21c5a Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Thu, 21 Aug 2025 15:34:02 +0200 Subject: [PATCH 03/13] fix(wokwi): update method calls for serial monitoring and writing --- pytest-embedded-wokwi/pyproject.toml | 2 +- pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/pytest-embedded-wokwi/pyproject.toml b/pytest-embedded-wokwi/pyproject.toml index a7ec3276..67caa820 100644 --- a/pytest-embedded-wokwi/pyproject.toml +++ b/pytest-embedded-wokwi/pyproject.toml @@ -36,7 +36,7 @@ dependencies = [ "pytest-embedded~=1.17.0a2", "toml~=0.10.2", # Temporary workaround for Wokwi client - will be redirected to the official repo - "wokwi-client @ git+https://github.com/JakubAndrysek/wokwi-python-client.git@sync-wokwi-client", + "wokwi-client @ git+https://github.com/JakubAndrysek/wokwi-python-client.git@add-sync-client", ] [project.optional-dependencies] diff --git a/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py b/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py index c48e21e0..89c9adea 100644 --- a/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py +++ b/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py @@ -124,13 +124,13 @@ def serial_callback(data: bytes): logging.debug(f'Error putting data in message queue: {e}') # Start monitoring in background - self.client.monitor_serial(serial_callback) + self.client.serial_monitor(serial_callback) def write(self, s: t.Union[str, bytes]) -> None: """Write data to the Wokwi serial interface.""" try: data = s if isinstance(s, bytes) else s.encode('utf-8') - self.client.write_serial(data) + self.client.serial_write(data) logging.debug(f'{self.SOURCE} ->: {s}') except Exception as e: logging.error(f'Failed to write to Wokwi serial: {e}') From 7453c82dff0a34a32299d1e5bdba5f3c6e31971d Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Mon, 25 Aug 2025 12:12:11 +0200 Subject: [PATCH 04/13] fix: update wokwi-client dependency to use stable release --- pytest-embedded-wokwi/pyproject.toml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pytest-embedded-wokwi/pyproject.toml b/pytest-embedded-wokwi/pyproject.toml index 67caa820..4e847a65 100644 --- a/pytest-embedded-wokwi/pyproject.toml +++ b/pytest-embedded-wokwi/pyproject.toml @@ -35,8 +35,8 @@ requires-python = ">=3.7" dependencies = [ "pytest-embedded~=1.17.0a2", "toml~=0.10.2", - # Temporary workaround for Wokwi client - will be redirected to the official repo - "wokwi-client @ git+https://github.com/JakubAndrysek/wokwi-python-client.git@add-sync-client", + # Temporary workaround - will be fixed for stable release + "wokwi-client", ] [project.optional-dependencies] From fa35a20b004aee124a9449249fd55eeb81d2e549 Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Thu, 28 Aug 2025 14:50:55 +0200 Subject: [PATCH 05/13] fix: update wokwi-client dependency to require a minimum version --- pytest-embedded-wokwi/pyproject.toml | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/pytest-embedded-wokwi/pyproject.toml b/pytest-embedded-wokwi/pyproject.toml index 4e847a65..b03f93cd 100644 --- a/pytest-embedded-wokwi/pyproject.toml +++ b/pytest-embedded-wokwi/pyproject.toml @@ -35,8 +35,7 @@ requires-python = ">=3.7" dependencies = [ "pytest-embedded~=1.17.0a2", "toml~=0.10.2", - # Temporary workaround - will be fixed for stable release - "wokwi-client", + "wokwi-client>=0.1.1", ] [project.optional-dependencies] From 27a9b4801e482e22abb5d9c746f431533cc5e6be Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Fri, 29 Aug 2025 10:16:15 +0200 Subject: [PATCH 06/13] feat(wokwi): log upload status before starting simulation --- pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py b/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py index 89c9adea..478b48d4 100644 --- a/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py +++ b/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py @@ -93,6 +93,8 @@ def _setup_simulation(self, diagram: str, firmware_path: str, elf_path: str): self.client.upload_file('pytest.bin', firmware_path) self.client.upload_file('pytest.elf', elf_path) + logging.info('Uploaded diagram and firmware to Wokwi. Starting simulation...') + # Start simulation self.client.start_simulation(firmware='pytest.bin', elf='pytest.elf') From 3f29de1de6dca531bfa9ec994cf0bbc057b79fbd Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Wed, 3 Sep 2025 22:42:13 +0200 Subject: [PATCH 07/13] fix: update wokwi-client dependency to require version 0.1.2 --- pytest-embedded-wokwi/pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pytest-embedded-wokwi/pyproject.toml b/pytest-embedded-wokwi/pyproject.toml index b03f93cd..5f7f615d 100644 --- a/pytest-embedded-wokwi/pyproject.toml +++ b/pytest-embedded-wokwi/pyproject.toml @@ -35,7 +35,7 @@ requires-python = ">=3.7" dependencies = [ "pytest-embedded~=1.17.0a2", "toml~=0.10.2", - "wokwi-client>=0.1.1", + "wokwi-client>=0.1.2", ] [project.optional-dependencies] From 0415a15542b5e8c35ba79e2f1db8d68083e1072b Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Thu, 4 Sep 2025 13:09:58 +0200 Subject: [PATCH 08/13] fix: remove unused wokwi-cli requirement from tests --- pytest-embedded-wokwi/tests/test_wokwi.py | 9 --------- 1 file changed, 9 deletions(-) diff --git a/pytest-embedded-wokwi/tests/test_wokwi.py b/pytest-embedded-wokwi/tests/test_wokwi.py index 60843724..39e7bcaf 100644 --- a/pytest-embedded-wokwi/tests/test_wokwi.py +++ b/pytest-embedded-wokwi/tests/test_wokwi.py @@ -1,21 +1,13 @@ import os -import shutil import pytest -wokwi_cli_required = pytest.mark.skipif( - shutil.which('wokwi-cli') is None, - reason='Please make sure that `wokwi-cli` is in your PATH env var. ' - + 'To install: https://docs.wokwi.com/wokwi-ci/getting-started#cli-installation', -) - wokwi_token_required = pytest.mark.skipif( os.getenv('WOKWI_CLI_TOKEN') is None, reason='Please make sure that `WOKWI_CLI_TOKEN` env var is set. Get a token here: https://wokwi.com/dashboard/ci', ) -@wokwi_cli_required @wokwi_token_required def test_pexpect_by_wokwi_esp32(testdir): testdir.makepyfile(""" @@ -40,7 +32,6 @@ def test_pexpect_by_wokwi(dut): result.assert_outcomes(passed=1) -@wokwi_cli_required @wokwi_token_required def test_pexpect_by_wokwi_esp32_arduino(testdir): testdir.makepyfile(""" From 938717cae74f72faf14d63756eecd6f8ee0eef6c Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Thu, 4 Sep 2025 18:07:38 +0200 Subject: [PATCH 09/13] fix: update wokwi-client dependency to require version 0.1.3 --- pytest-embedded-wokwi/pyproject.toml | 2 +- .../pytest_embedded_wokwi/wokwi.py | 14 +++++++++----- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/pytest-embedded-wokwi/pyproject.toml b/pytest-embedded-wokwi/pyproject.toml index 5f7f615d..79ad1e50 100644 --- a/pytest-embedded-wokwi/pyproject.toml +++ b/pytest-embedded-wokwi/pyproject.toml @@ -35,7 +35,7 @@ requires-python = ">=3.7" dependencies = [ "pytest-embedded~=1.17.0a2", "toml~=0.10.2", - "wokwi-client>=0.1.2", + "wokwi-client>=0.1.3", ] [project.optional-dependencies] diff --git a/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py b/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py index 478b48d4..7d3a305d 100644 --- a/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py +++ b/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py @@ -2,6 +2,7 @@ import logging import os import typing as t +from pathlib import Path from packaging.version import Version from pytest_embedded.log import DuplicateStdoutPopen, MessageQueue @@ -47,7 +48,6 @@ def __init__( **kwargs, ): self.app = app - self.firmware_resolver = firmware_resolver # Get Wokwi API token token = os.getenv('WOKWI_CLI_TOKEN') @@ -76,8 +76,11 @@ def __init__( # Connect and start simulation try: - firmware_path = self.firmware_resolver.resolve_firmware(app) - self._setup_simulation(wokwi_diagram, firmware_path, app.elf_file) + flasher_args = firmware_resolver.resolve_firmware(app) + firmware_path = Path(flasher_args).as_posix() + elf_path = Path(app.elf_file).as_posix() + + self._setup_simulation(wokwi_diagram, firmware_path, elf_path) self._start_serial_monitoring() except Exception as e: self.close() @@ -90,13 +93,14 @@ def _setup_simulation(self, diagram: str, firmware_path: str, elf_path: str): # Upload files self.client.upload_file('diagram.json', diagram) - self.client.upload_file('pytest.bin', firmware_path) + firmware = self.client.upload_file('pytest.bin', firmware_path) + self.client.upload_file('pytest.elf', elf_path) logging.info('Uploaded diagram and firmware to Wokwi. Starting simulation...') # Start simulation - self.client.start_simulation(firmware='pytest.bin', elf='pytest.elf') + self.client.start_simulation(firmware, elf='pytest.elf') def _start_serial_monitoring(self): """Start monitoring serial output and forward to stdout and message queue.""" From e5cb868a6f0541620992bb0244178d6e00b04021 Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Thu, 4 Sep 2025 18:18:28 +0200 Subject: [PATCH 10/13] feat(tests): add merged binary for hello_world_arduino example --- .../hello_world_arduino/build/hello_world_arduino.ino.merged.bin | 1 + 1 file changed, 1 insertion(+) create mode 120000 tests/fixtures/hello_world_arduino/build/hello_world_arduino.ino.merged.bin diff --git a/tests/fixtures/hello_world_arduino/build/hello_world_arduino.ino.merged.bin b/tests/fixtures/hello_world_arduino/build/hello_world_arduino.ino.merged.bin new file mode 120000 index 00000000..4c209aa1 --- /dev/null +++ b/tests/fixtures/hello_world_arduino/build/hello_world_arduino.ino.merged.bin @@ -0,0 +1 @@ +hello_world_arduino.ino.bin \ No newline at end of file From 9ff05431fb98d1ae5f0be2a96dbbca30cad96336 Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Wed, 10 Sep 2025 09:41:07 +0200 Subject: [PATCH 11/13] fix: update wokwi-client dependency to require version 0.2.0, remove toml --- pytest-embedded-wokwi/pyproject.toml | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/pytest-embedded-wokwi/pyproject.toml b/pytest-embedded-wokwi/pyproject.toml index 79ad1e50..1bf707e9 100644 --- a/pytest-embedded-wokwi/pyproject.toml +++ b/pytest-embedded-wokwi/pyproject.toml @@ -34,8 +34,7 @@ requires-python = ">=3.7" dependencies = [ "pytest-embedded~=1.17.0a2", - "toml~=0.10.2", - "wokwi-client>=0.1.3", + "wokwi-client>=0.2.0", ] [project.optional-dependencies] From 7eb79f9e069463e391ca134c69a0471912580e1c Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Wed, 10 Sep 2025 09:53:53 +0200 Subject: [PATCH 12/13] refactor: update type hints for wokwi diagram and write method --- pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py b/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py index 7d3a305d..1c4f2b97 100644 --- a/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py +++ b/pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py @@ -42,9 +42,9 @@ def __init__( self, msg_queue: MessageQueue, firmware_resolver: IDFFirmwareResolver, - wokwi_diagram: t.Optional[str] = None, + wokwi_diagram: str | None = None, app: t.Optional['IdfApp'] = None, - meta: t.Optional[Meta] = None, + meta: Meta | None = None, **kwargs, ): self.app = app @@ -132,7 +132,7 @@ def serial_callback(data: bytes): # Start monitoring in background self.client.serial_monitor(serial_callback) - def write(self, s: t.Union[str, bytes]) -> None: + def write(self, s: str | bytes) -> None: """Write data to the Wokwi serial interface.""" try: data = s if isinstance(s, bytes) else s.encode('utf-8') From 4b425c83445429b4dca8c8b860085d8de855b274 Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Wed, 10 Sep 2025 10:04:02 +0200 Subject: [PATCH 13/13] revert: revert back to the working logger with the wokwi --- pytest-embedded/pytest_embedded/log.py | 11 +---------- 1 file changed, 1 insertion(+), 10 deletions(-) diff --git a/pytest-embedded/pytest_embedded/log.py b/pytest-embedded/pytest_embedded/log.py index 2deacb99..20d2b4c6 100644 --- a/pytest-embedded/pytest_embedded/log.py +++ b/pytest-embedded/pytest_embedded/log.py @@ -166,7 +166,7 @@ class DuplicateStdoutPopen(subprocess.Popen): SOURCE = 'POPEN' REDIRECT_CLS = _PopenRedirectProcess - def __init__(self, msg_queue: MessageQueue, cmd: str | list[str], meta: Meta | None = None, **kwargs): + def __init__(self, msg_queue: MessageQueue, cmd: str | list[str] = [], meta: Meta | None = None, **kwargs): self._q = msg_queue self._p = None @@ -188,15 +188,6 @@ def __init__(self, msg_queue: MessageQueue, cmd: str | list[str], meta: Meta | N self._logfile_offset = 0 logging.debug(f'temp log file: {_log_file}') - kwargs.update( - { - 'bufsize': 0, - 'stdin': subprocess.PIPE, - 'stdout': self._fw, - 'stderr': self._fw, - } - ) - self._cmd = cmd # Only start subprocess if command is not empty