feat(wokwi): Add support for specifying diagram path

pytest-embedded-wokwi/pytest_embedded_wokwi/__init__.py

@@ -1,9 +1,12 @@
 """Make pytest-embedded plugin work with the Wokwi CLI."""
 
-from .dut import WokwiDut
-from .wokwi_cli import WokwiCLI
+WOKWI_CLI_MINIMUM_VERSION = '0.10.1'
+
+from .dut import WokwiDut  # noqa
+from .wokwi_cli import WokwiCLI  # noqa
 
 __all__ = [
+    'WOKWI_CLI_MINIMUM_VERSION',
     'WokwiCLI',
     'WokwiDut',
 ]

pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi_cli.py

@@ -5,10 +5,14 @@
 import typing as t
 from pathlib import Path
 
+import pexpect
 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:
@@ -32,7 +36,6 @@ class WokwiCLI(DuplicateStdoutPopen):
     """
 
     SOURCE = 'Wokwi'
-
     WOKWI_CLI_PATH = 'wokwi-cli'
 
     def __init__(
@@ -41,6 +44,7 @@ def __init__(
         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,
     ):
@@ -51,15 +55,37 @@ def __init__(
         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')
+
+        child = pexpect.spawn('wokwi-cli --help')
+        try:
+            child.expect(r'Wokwi CLI v(\d+\.\d+\.\d+)', timeout=1)
+            wokwi_cli_version = child.match.group(1).decode('utf-8')
+        except pexpect.TIMEOUT:
+            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()
-        self.create_diagram_json()
+
+        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,
@@ -107,22 +133,6 @@ def create_diagram_json(self):
         app = self.app
         target_board = target_to_board[app.target]
 
-        # Check for specific target.diagram.json file first
-        diagram_json_path = os.path.join(app.app_path, (app.target + '.diagram.json'))
-        if os.path.exists(diagram_json_path):
-            # If there is also common diagram.json file, backup it first to diagram.json.old
-            if os.path.exists(os.path.join(app.app_path, 'diagram.json')):
-                logging.warning(
-                    'using %s instead. backup the original diagram.json to diagram.json.old', diagram_json_path
-                )
-                shutil.copyfile(
-                    os.path.join(app.app_path, 'diagram.json'),
-                    os.path.join(app.app_path, 'diagram.json.old'),
-                )
-            # Copy target.diagram.json to diagram.json
-            shutil.copyfile(diagram_json_path, os.path.join(app.app_path, 'diagram.json'))
-            return
-
         # Check for common diagram.json file
         diagram_json_path = os.path.join(app.app_path, 'diagram.json')
         if os.path.exists(diagram_json_path):

pytest-embedded-wokwi/tests/test_wokwi.py

@@ -54,6 +54,7 @@ def test_pexpect_by_wokwi(dut):
         '-s',
         '--embedded-services', 'arduino,wokwi',
         '--app-path', os.path.join(testdir.tmpdir, 'hello_world_arduino'),
+        '--wokwi-diagram', os.path.join(testdir.tmpdir, 'hello_world_arduino/esp32.diagram.json'),
     )
 
     result.assert_outcomes(passed=1)

pytest-embedded/pytest_embedded/dut_factory.py

@@ -149,6 +149,7 @@ def _fixture_classes_and_options_fn(
     wokwi_cli_path,
     wokwi_timeout,
     wokwi_scenario,
+    wokwi_diagram,
     skip_regenerate_image,
     encrypt,
     keyfile,
@@ -302,6 +303,7 @@ def _fixture_classes_and_options_fn(
                     'wokwi_cli_path': wokwi_cli_path,
                     'wokwi_timeout': wokwi_timeout,
                     'wokwi_scenario': wokwi_scenario,
+                    'wokwi_diagram': wokwi_diagram,
                     'msg_queue': msg_queue,
                     'app': None,
                     'meta': _meta,
@@ -608,6 +610,7 @@ def create(
         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,
         keyfile: t.Optional[str] = None,
@@ -655,6 +658,7 @@ def create(
             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.
             keyfile: Keyfile for encryption.
@@ -718,6 +722,7 @@ def create(
                 '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,
                 'keyfile': keyfile,

pytest-embedded/pytest_embedded/plugin.py

@@ -283,6 +283,10 @@ def pytest_addoption(parser):
         '--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)',
+    )
 
 
 ###########
@@ -973,6 +977,13 @@ def wokwi_scenario(request: FixtureRequest) -> t.Optional[str]:
     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]:
+    """Enable parametrization for the same cli option"""
+    return _request_param_or_config_option_or_default(request, 'wokwi_diagram', None)
+
+
 ####################
 # Private Fixtures #
 ####################
@@ -1031,6 +1042,7 @@ def parametrize_fixtures(
     wokwi_cli_path,
     wokwi_timeout,
     wokwi_scenario,
+    wokwi_diagram,
     skip_regenerate_image,
     encrypt,
     keyfile,