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/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/pyproject.toml b/pytest-embedded-wokwi/pyproject.toml index 683dcd80..bff6a3e7 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"} @@ -30,7 +31,7 @@ requires-python = ">=3.10" dependencies = [ "pytest-embedded~=1.17.0a2", - "toml~=0.10.2", + "wokwi-client>=0.2.0", ] [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..1c4f2b97 --- /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 pathlib import Path + +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: str | None = None, + app: t.Optional['IdfApp'] = None, + meta: Meta | None = None, + **kwargs, + ): + self.app = app + + # 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') + + # Initialize parent class + super().__init__(msg_queue=msg_queue, meta=meta, **kwargs) + + # Connect and start simulation + try: + 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() + 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) + 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, 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.serial_monitor(serial_callback) + + 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') + 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}') + + 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() + super().__del__() + + 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 afe1ce08..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: str | None = None, - wokwi_timeout: int | None = None, - wokwi_scenario: str | None = None, - wokwi_diagram: str | None = 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-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(""" diff --git a/pytest-embedded/pytest_embedded/dut_factory.py b/pytest-embedded/pytest_embedded/dut_factory.py index eb3fc019..8eee222c 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 @@ -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, @@ -319,14 +316,11 @@ 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, - 'wokwi_scenario': wokwi_scenario, 'wokwi_diagram': wokwi_diagram, 'msg_queue': msg_queue, 'app': None, @@ -523,7 +517,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 @@ -543,7 +537,7 @@ def dut_gn( app: App, serial: t.Union['Serial', 'LinuxSerial'] | None, qemu: t.Optional['Qemu'], - wokwi: t.Optional['WokwiCLI'], + wokwi: t.Optional['Wokwi'], ) -> Dut | list[Dut]: global DUT_GLOBAL_INDEX DUT_GLOBAL_INDEX += 1 @@ -687,9 +681,6 @@ def create( qemu_prog_path: str | None = None, qemu_cli_args: str | None = None, qemu_extra_args: str | None = None, - wokwi_cli_path: str | None = None, - wokwi_timeout: int | None = 0, - wokwi_scenario: str | None = None, wokwi_diagram: str | None = None, skip_regenerate_image: bool | None = None, encrypt: bool | None = None, @@ -736,9 +727,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. @@ -801,9 +789,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/log.py b/pytest-embedded/pytest_embedded/log.py index 65d6c549..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,18 +188,30 @@ 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 - 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 6fa5e3ee..baa6b8c8 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') @@ -290,20 +290,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)', @@ -988,27 +974,6 @@ def keyfile(request: FixtureRequest) -> str | None: ######### # Wokwi # ######### -@pytest.fixture -@multi_dut_argument -def wokwi_cli_path(request: FixtureRequest) -> str | None: - """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) -> str | None: - """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) -> str | None: - """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) -> str | None: @@ -1072,9 +1037,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, @@ -1167,7 +1129,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()) @@ -1181,7 +1143,7 @@ def dut( app: App, serial: t.Union['Serial', 'LinuxSerial'] | None, qemu: t.Optional['Qemu'], - wokwi: t.Optional['WokwiCLI'], + wokwi: t.Optional['Wokwi'], ) -> Dut | list[Dut]: """ A device under test (DUT) object that could gather output from various sources and redirect them to the pexpect 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