From 9b61b5454c1bc64fb8f42cd6fb95aafc90e73d13 Mon Sep 17 00:00:00 2001 From: Ahnaf Shahriar Date: Tue, 27 Oct 2026 16:11:16 -0700 Subject: [PATCH] Pyrite only: drop the Renode/C# backend; the sim binary is the simulator The C# engine is halted and its release product retired, yet Sim, run_firmware and the pytest plugin still defaulted to backend="renode" (pythonnet + Simantic.Core), which can no longer be fetched. The Rust batch path was a second runner beside the binary that rejected `parts:` and multi-machine scenarios. - Remove the renode backend, the pythonnet/.NET loader, the Simantic.Core install, the pythonnet dependency, --sim-backend and backend=. - run_firmware runs `sim run --elf ...`; run_scenario and test.yaml items run `sim run --scenario ... --record DIR` and read the bundle, so peers, several machines and media work because the binary does them. `args=` passes any other sim flag; ScenarioRun keeps stdout (traces) and stderr (peer output). - No wall-clock kill around sim: --timeout ends a run in virtual time; the plugin still judges a manifest's own `wall:` budget. - Sim stays on the in-process pyrite engine (single machine, stepping). - README/docs: the binary is the simulator; Python orchestrates. 0.5.0. Checked: `SIMANTIC_SIM= uv run pytest -q` 175 passed, 12 skipped. run_scenario on a customer two-peer STM32H753 scenario (2 Python peers + CAN peer, 10 s, args=["--trace-symbol","SysTick_Handler"]): 306 UART records, 9836 SysTick hits (same as the binary), 20 [gw] RX lines. Co-Authored-By: Claude Opus 5.5 --- README.md | 262 +++++++++++----------------- docs/session-api.md | 150 ++++------------ docs/writing-tests.md | 32 ++-- pyproject.toml | 6 +- src/simantic/__init__.py | 19 +-- src/simantic/_cli.py | 4 +- src/simantic/_elf.py | 5 +- src/simantic/_rust.py | 146 ++-------------- src/simantic/engine.py | 135 +-------------- src/simantic/install.py | 21 --- src/simantic/mcu.py | 166 ++++++++++-------- src/simantic/pytest_plugin.py | 163 ++++-------------- src/simantic/session.py | 310 ++-------------------------------- tests/test_fixtures.py | 63 +++---- tests/test_install.py | 20 +-- tests/test_pytest_surface.py | 12 +- tests/test_rust_backend.py | 151 ++--------------- tests/test_session.py | 30 +--- 18 files changed, 364 insertions(+), 1331 deletions(-) diff --git a/README.md b/README.md index 62fc854..fba6132 100644 --- a/README.md +++ b/README.md @@ -10,48 +10,34 @@ MIT licence

-Nothing to plug in, nothing to flash. Simantic boots your real ELF on a -simulated microcontroller and hands you the whole machine from Python. Watch it -print, press a button, read a variable straight out of RAM. +The `sim` binary is the simulator. This package installs it, runs it, and turns +its output into Python objects and pytest results. Everything the binary does +(several machines, Python scripted peers, media and bridges, `--record`, +`--trace-symbol`, `--real-time`) works from Python because Python just runs it. ```bash pip install simantic ``` ```python -from simantic import Sim +import simantic -with Sim(elf="fw.elf", mcu="STM32F401RE", uart="usart2") as sim: - sim.expect("ready") - sim.inject_gpio("gpioc", 13, True) # press the user button - sim.expect("button pressed") - assert sim.read_u32("press_count") == 1 +run = simantic.run_firmware("fw.elf", mcu="STM32F401RE", expect=["ready"]) +assert run.passed, run.failure_report() ``` -That is a whole test. No probe, no breakpoint, no waiting on hardware. - -Three things you get that a bench cannot give you: - -* **See inside.** Read any variable, register, or RTOS thread while the - firmware runs, without halting it. -* **Poke it.** Press buttons, send CAN frames, feed the radio, all from your - script. -* **Repeat exactly.** Time moves only when you ask, so a run comes out the same - every time, on your laptop and in CI. +That is a whole test. No probe, no breakpoint, no waiting on hardware. Time is +virtual, so a run comes out the same every time, on your laptop and in CI. -The simulator lives inside your Python process, so there is no server to start -and no port to talk to. - -> **Alpha, version 0.4.x.** We are still moving things around, so the API can -> change without a deprecation period. Pin an exact version -> (`simantic==0.4.2`) if you depend on it, and please hold off on production -> pipelines for now. Tell us what breaks. +> **Alpha, version 0.5.x.** 0.5.0 is breaking: the Renode/C# engine is gone and +> pyrite (the Rust `sim` binary) is the only engine. The API can still change +> without a deprecation period, so pin an exact version. ## Setup -`pip install` is the whole setup. The first `Sim(...)` downloads the engine it -needs into `~/.simantic/` and checks it against the published checksum. The -wheel on PyPI holds only Python code; the simulators are never inside it. +`pip install` is the whole setup. The first run downloads the `sim` binary into +`~/.simantic/` and checks it against the published checksum. The wheel on PyPI +holds only Python code. Sign in once if you want to name MCUs by part number: @@ -65,19 +51,99 @@ That opens a browser tab, much like `gh auth login`, and saves a token to Already have the `sim` binary? Put it on PATH or point `$SIMANTIC_SIM` at it. `simantic status` shows what resolved. -## Pick your engine +## Run a scenario -The same script runs on either engine. You choose per simulation: +`run_scenario` runs `sim run --scenario ... --record DIR` and reads the recorded +bundle back. A scenario can hold several machines, Python scripted peers +(`parts`) and the media that join them: ```python -Sim(elf="fw.elf", mcu="STM32F401RE", uart="usart2") # Renode engine, the default -Sim(elf="fw.elf", mcu="STM32F401RE", uart="usart2", backend="rust") # our Rust engine +scenario = { + "machines": { + "dut": {"mcu": "STM32H753IITX", "elf": "fc.elf", + "parts": [{"name": "gpspeer", "type": "uart-device", + "baud": 230400, "script": "gps.py"}]}, + "node": {"mcu": "STM32F407VG", "elf": "node.elf"}, + }, + "media": [ + {"type": "uart", "connect": ["dut.usart6", "dut.gpspeer"]}, + {"type": "can", "connect": ["dut.fdcan1", "node.can1"]}, + ], +} +run = simantic.run_scenario(scenario, timeout=12, args=["--trace-symbol", "can_send"]) + +assert "[dut] GPS fix: 3" in run.output # "[machine] "-prefixed UART lines +sends = [l for l in run.stdout.splitlines() if l.startswith("[node]") and "can_send" in l] ``` -The Rust engine is a small extension module, and is -considerably faster. Anything it cannot do yet will raise a `simantic.NotSupported` and names the gap -instead of quietly doing nothing. What each engine covers is listed in -[the engine table](https://github.com/simantic-dev/pippy/blob/main/docs/session-api.md#what-each-engine-supports). +The scenario dict is the same shape as `sim run --scenario`'s YAML. Records +carry virtual-time stamps, so a rate is a subtraction and an ordering is a +comparison. Changing what a peer does (a NACK for 50 ms, a stale frame, a node +that stops answering at t = 5 s) is an edit to its script. The `sim` guide +covers every output and the peer API: https://simantic.com/docs + +## Testing with pytest + +Installing the package turns each `test.yaml` manifest into its own pytest item, +so you get `-k` filtering, `--junitxml` and xdist. A manifest is a scenario plus +what must and must not appear: + +```yaml +machines: + dut: + mcu: STM32H753ZI + elf: fc.elf + parts: + - { name: gpspeer, type: uart-device, baud: 230400, script: gps.py } +media: + - { type: uart, connect: [dut.usart6, dut.gpspeer] } +timeout: 12 +expect: ["GPS fix: 3"] +expect_absent: ["RESULT: FAIL"] +``` + +```bash +pytest path/to/fixtures -k gps +``` + +Each item runs `sim run --scenario ... --record DIR` and checks the bundle. When +a test fails, its UART transcript is attached to the report. Single-machine +manifests (`mcu:` at the top level) are the one-machine case. Models resolve +through your account; no checkout of ours is needed. + +A run has a wall-clock budget: 30 s of host time by default, or whatever `wall:` +says. Going over it fails the test with the virtual time reached. Slow +simulation is a firmware busy-wait or a model gap, and the report says which. + +For a scenario with Python assertions in the middle, call `run_scenario` from an +ordinary test, as above; see [docs/writing-tests.md](docs/writing-tests.md). + +## Stepping one machine: `Sim` + +`Sim` runs the in-process pyrite engine on a single machine, so you can step +virtual time, inject, and read state between steps. It is a debugging and +stepping layer, not the way to run scenarios or suites. + +```python +from simantic import Sim + +with Sim(elf="fw.elf", mcu="STM32F401RE", uart="usart2") as sim: + sim.expect("ready") + sim.inject_gpio("gpioc", 13, True) # press the user button + sim.expect("button pressed") + assert sim.read_u32("press_count") == 1 +``` + +Time moves only when you ask. See [docs/session-api.md](docs/session-api.md). + +## When to use what + +* **The `sim` CLI directly:** one-off runs, `--real-time`, `--uart-tcp` bridges, + `--record` bundles, interactive poking. +* **`run_scenario` and pytest `test.yaml`:** anything with several machines, + scripted peers or media, and every suite that runs in CI. +* **`Sim`:** one machine, when you need to pause part-way, inject input, or read + memory and RTOS state between steps. ## ESP32 images @@ -106,128 +172,6 @@ release ESP-IDF installs. Offline, or with your own copy, pass `--rom`. `simantic esp-rom --chip esp32c3` downloads it ahead of time and prints where it went. -## Testing with pytest - -Take the `sim` fixture and write ordinary tests: - -```python -def test_timer_irq_fires(sim): - s = sim(elf="fw.elf", mcu="STM32F401RE", uart="usart2") - s.expect("fired=1", timeout=8) - assert s.read_u32("fired") == 1 -``` - -The engine starts once per worker rather than once per test, and every machine -is closed for you. When a test fails, its UART transcript is attached to the -report, because that is usually the evidence you want. - -### Test what went over the wire - -Firmware that prints "sensor OK" is reporting its own bookkeeping. The bus -tells you what happened. Put a scripted device on the other end of each bus -(a `parts` entry per device, each backed by a few dozen lines of Python), wire -the media in a scenario, and assert on `frames()`: - -```python -SCENARIO = { - "machines": {"dut": { - "mcu": "STM32H753IITX", "elf": "fc.elf", - "parts": [ - {"name": "imu0", "type": "spi-device", "bus": "spi1", "cs": "PC4", "script": "icm42688.py"}, - {"name": "baro", "type": "i2c-device", "bus": "i2c2", "address": 0x63, "script": "icp20100.py"}, - {"name": "gpspeer", "type": "uart-device", "baud": 230400, "script": "gps.py"}, - {"name": "canpeer", "type": "can-node", "script": "dronecan.py"}, - ], - }}, - "media": [ - {"type": "uart", "connect": ["dut.usart6", "dut.gpspeer"]}, - {"type": "can", "connect": ["dut.fdcan1", "dut.canpeer"]}, - ], -} - -def test_imu_configured_then_streams(sim): - s = sim(scenario=SCENARIO, machine="dut", uart="usart1") - s.run_for(3.0) # virtual seconds - imu = [f for f in s.frames() if f["label"] == "imu0"] # one dict per chip-select window - cfg = next(f for f in imu if f["data"][0] == 0x4F) # GYRO_CONFIG0 write - assert cfg["data"][1] & 0x0F == 0x06 # ODR the driver programmed - assert cfg["miso"][0] == 0x00 # what the slave answered - bursts = [f for f in imu if f["t"] > cfg["t"] and len(f["data"]) > 16] - assert len(bursts) > 100 # FIFO reads after configuration -``` - -Every record carries its virtual-time stamp, so a rate is a subtraction between -consecutive frames and an ordering is a comparison. The scenario dict is the -same shape as `sim --scenario`'s YAML, and because the test owns it, the -negative control is the same test with a peer removed from `media`. Changing -what a device does (a NACK for 50 ms, a stale frame, a node that stops -answering at t = 5 s) is an edit to the peer script, on the virtual clock. The -`sim` guide covers every output and the peer API: -https://simantic.com/docs - -`--sim-backend=renode|rust|both` chooses the engine. With `both`, each test runs -on each and the engine name appears in the test id. Anything an engine cannot do -is reported as a skip with the reason, so one suite can target both and stay -honest about what each covers. - -`test.yaml` manifests follow the same option. On `rust` a manifest, including a -multi machine one with `media:`, runs to its timeout in one call and its UART -output is checked afterwards; `expect_frames` checks are skipped there for now. -`simantic.run_firmware(..., backend="rust")` does the same for a single ELF. - -One tip worth real time: on the Renode engine, every hand off between Python and -the simulation costs a few hundred microseconds. Reading is free, pausing and -resuming is not. Prefer `expect()`, which crosses once, over a loop that polls -every millisecond. On the Rust engine, polling is essentially free. - -If you keep `test.yaml` fixture manifests, installing the package also turns -each one into its own pytest item, so you get `-k` filtering, `--junitxml`, and -xdist for free. A manifest is the fire-and-forget form of a test: name the -machine(s) with their `parts`, the `media` that wire UART and CAN peers to -the firmware's controllers, how long to run, and what must and must not -appear: - -```yaml -machines: - dut: - mcu: STM32H753ZI - elf: fc.elf - parts: - - { name: gpspeer, type: uart-device, baud: 230400, script: gps.py } -media: - - { type: uart, connect: [dut.usart6, dut.gpspeer] } -timeout: 12 -expect: ["GPS fix: 3"] -expect_absent: ["RESULT: FAIL"] -expect_frames: ["SPI Rx cs=0 len=2 mosi=4F 06"] # same line shape as `sim --frames` -expect_frames_absent: ["CAN Dropped"] -``` - -Single-machine manifests (`mcu:` at the top level) are the one-machine case. -Models resolve through your account; no checkout of ours is needed. A model can -require a minimum engine version; if yours is older, the failure names the -version and the fix (use a newer engine build). - -Three more keys cover fixtures that need files or peers on the host side: -`sparse_files` creates blank files of a given size before the run (a blank SD -card, say), `networkServices` adds scripted network peers, and `{TEST_DIR}` and -`{WORK_DIR}` in a board file expand to the manifest's directory and a per-test -scratch directory. A manifest whose `sim_args` asks for a `sim` -command-line flag with no equivalent here is skipped, with the flag named. - -A run also has a wall-clock budget: 30 s of host time by default, 100 s with -`wireless: true`, or whatever `wall:` says. Going over it fails the test with -the virtual time reached. Slow simulation is a firmware busy-wait or a model -gap, and the report tells you which to go find rather than waiting it out. - -For a single run with no assertions in the middle, there is `run_firmware(...)`: - -```python -run = simantic.run_firmware("build/zephyr.elf", mcu="STM32F401RE", - expect=["RESULT: PASS"]) -assert run.passed, run.failure_report() -``` - ## Telemetry Only when you are signed in, we report two things: diff --git a/docs/session-api.md b/docs/session-api.md index 1eba883..7c19658 100644 --- a/docs/session-api.md +++ b/docs/session-api.md @@ -1,16 +1,17 @@ -# Driving the engine: `simantic.Sim` +# Stepping one machine: `simantic.Sim` -`Sim` is the Python face of the simulator — the same capabilities as the -`sim` CLI, as a live object. No test framework is involved; pytest, a plain -script, a notebook or a process pool all use it the same way. It builds the -machine(s) inside your Python process and holds them at reset; every method either -advances virtual time by an exact, requested amount or observes state without -advancing it. Between calls nothing runs, so a script is deterministic and -Python think-time is free. +`Sim` is the in-process pyrite engine as a live object, for a single machine. It +builds the machine inside your Python process and holds it at reset; every method +either advances virtual time by an exact, requested amount or observes state +without advancing it. Between calls nothing runs, so a script is deterministic +and Python think-time is free. -It is the same class for a single board and for a multi-machine scenario with -radio/CAN/Ethernet media and scripted network peers — everything `sim ---scenario` accepts, `Sim(scenario=...)` accepts. +`Sim` is a debugging and stepping layer. It is not the way to run scenarios or +suites: several machines, scripted peers (`parts:` with scripts), media and +bridges belong to the `sim` binary, which Python reaches through +`simantic.run_scenario(...)` and `test.yaml` pytest items (both run +`sim run --scenario ... --record DIR` and read the bundle). There is no +`backend=` argument; pyrite is the only engine. ## Start @@ -18,43 +19,17 @@ radio/CAN/Ethernet media and scripted network peers — everything `sim from simantic import Sim # a model name: ~/.sim_cache, else fetched with your credentials and cached (like `sim --mcu`) -Sim(elf="fw.elf", mcu="STM32F401RE", uart="usart2", - parts=[{"name": "baro", "type": "i2c-device", "bus": "i2c1", "address": 0x76, "script": "baro.py"}]) - -# a scenario dict — the sim --scenario schema as Python -Sim(scenario={...}, machine="c6", uart="uart0", cwd=fixture_dir) +Sim(elf="fw.elf", mcu="STM32F401RE", uart="usart2") ``` -Scenario dicts use the `sim --scenario` schema verbatim: `machines` -(`mcu` + `parts` + `overlay` + `elf` + `symbolsElfPath`), `media` (BLE/CAN/Ethernet/UART -buses), `networkServices` (scripted peers), `quantum`. Relative paths resolve -against `cwd=` (default: the process cwd). - -`parts` is a list of devices on the board, one dict each: `name`, `type`, then -per type — `uart-device` (`script`, `baud`; wire it with a uart medium), -`i2c-device` (`bus`, `address`, `script`), `spi-device` (`bus`, `script`, `cs`: -an index, or a pin like `"PC4"` when several devices share the bus), -`can-node` (`script`; wire with a can medium), `ble-peer` / `wifi-peer` -(`ssid`, `passphrase`) / `eth-peer` (`mac`) (`script`; wire with the matching -medium), `sd-card` (`bus`, `size` such as `"16MiB"`, `image`: a formatted card image, since a blank card has nothing to mount), `peripheral` (`address`, -`size`, `script`, optional `irq` list), `model` (a stock part: `class` such as -`Sensors.BMP388`, `bus`, `address`, `properties`). -`overlay` is a board file (`.board`), a raw platform fragment for anything that -list cannot express. Needs engine 0.5.17+. - -A model can require a newer engine than the one installed (the STM32H7 parts -need 0.6.3). `Sim` then raises `SimError` naming the version it needs; use a newer engine build (the C# engine is no longer published). - -`symbols_elf=` (single-machine form) / `symbolsElfPath` (scenario machine -entries — same key as the CLI's scenario YAML, so a scenario dict is -copy-pasteable) attaches a companion ELF that carries debug symbols for a +`symbols_elf=` attaches a companion ELF that carries debug symbols for a stripped image, e.g. a PlatformIO/IDF `firmware.elf` alongside a stripped flash container passed as `elf=`. It only changes what `symbol()` and RTOS -introspection can resolve — the image that actually runs is still `elf=`. +introspection can resolve; the image that actually runs is still `elf=`. -`trace_symbols=[...]`, `trace_memory=[...]`, `trace_interrupts=True` and -`show_logs=True` turn on the non-halting instrumentation; read it back with -`symbol_trace()`, `interrupts()` and `logs()`. +`trace_symbols=[...]`, `trace_memory=[...]` and `trace_interrupts=True` turn +on the non-halting instrumentation; read it back with `symbol_trace()`, +`memory_trace()` and `interrupts()`. ## Drive @@ -84,8 +59,8 @@ one. `timeout` is wall-clock effort, not virtual time — assert on | `logs()` | `[{t, level, source, message}]` — unhandled registers, model warnings | | `interrupts()` | `[{t, machine, direction, exception, name}]` (with `trace_interrupts=True`) | | `symbol_trace()` | `[{t, machine, symbol, address, args}]` (with `trace_symbols=[...]`) | -| `memory_trace()` | `[{t, machine, watch, kind, address, value}]` (with `trace_memory=[...]`; `value` is `None` for a read). `backend="rust"` only | -| `itm()` | `[{t, machine, port, bytes, text}]`: ITM stimulus-port output (with `itm=True`, Cortex-M). `backend="rust"` only | +| `memory_trace()` | `[{t, machine, watch, kind, address, value}]` (with `trace_memory=[...]`; `value` is `None` for a read). | +| `itm()` | `[{t, machine, port, bytes, text}]`: ITM stimulus-port output (with `itm=True`, Cortex-M). | | `read_memory(addr_or_symbol, count)` / `read_u32(...)` | bytes / int from the system bus (RAM, flash, peripheral registers) | | `symbol(name)` | ELF symbol address | | `threads()` / `heap()` | RTOS thread snapshot / heap report, when recognised | @@ -93,39 +68,17 @@ one. `timeout` is wall-clock effort, not virtual time — assert on Every observer takes `machine=` in a scenario; the constructor's `machine=` and `uart=` are the defaults. -## What each engine supports +## Scope -| | `backend="renode"` | `backend="rust"` | -| --- | --- | --- | -| UART text: `expect()`, manifest `expect` / `expect_absent` | yes | yes | -| Bus frames (SPI, I2C, CAN, BLE, Ethernet): `frames()`, manifest `expect_frames` | yes | no: `frames()` is empty, a manifest with `expect_frames` is skipped | -| `interrupts()` | yes | yes, `Sim` and `run_scenario` | -| `symbol_trace()` | yes | yes, `Sim` and `run_scenario` | -| `memory_trace()` | no | yes, `Sim` and `run_scenario` | -| `itm()` | no | yes, `Sim` only | -| `read_memory()` / `read_u32()` / `symbol()` | yes | yes, `Sim` only | -| `threads()` / `heap()` | yes | yes, `Sim` only | -| `logs()` | yes | no: always empty | -| `inject_gpio()` | yes | yes | -| `inject_can()` | yes | yes | -| `inject_adc()` | no: raises `NotSupported` (use the overlay's `channelMillivolts`) | yes: STM32 F0/F4/G4, RP2040, ESP32-C3/P4, nRF52840 | -| `attach_i2c()` / `attach_spi()` | no: raises `NotSupported` | yes: I2C on STM32 (both IP versions), RP2040, NXP S32K3, nRF52840; SPI on STM32, RP2040, nRF52840 | -| `inject_radio()` | yes | no: raises `NotSupported` | -| `parts=` and `networkServices` | yes | no: raises `NotSupported` (a manifest is skipped) | -| Several machines joined by `media` | yes | `run_scenario` and `test.yaml` manifests; `Sim` drives one machine | - -The hardware list grows as engine models opt in; a peripheral that does not offer a -capability raises `SimError` naming it. - -"`Sim` only" means a Python test holding a `Sim` can ask for it; a `test.yaml` -manifest has keys for UART text and bus frames and nothing else, on either -engine. - -On the Rust engine a scenario with several machines runs in one call, to its -timeout, and what happened is read afterwards. It cannot be paused, so nothing -can be read or injected part-way; `Sim` needs exactly that, and its Rust -session holds a single machine. A manifest runs this way, and so does -`run_scenario`, which also returns the traces: +`Sim` drives one machine. A peripheral that does not offer a capability raises +`SimError` naming it. For anything else, use the binary: + +| need | use | +| --- | --- | +| several machines joined by `media` | `simantic.run_scenario(...)` or a `test.yaml` | +| Python scripted peers (`parts:`) | the same: they run in the `sim` process | +| bus frames, traces, UART records for every machine | `run.uart_records`, `run.interrupts`, `run.frames` from `run_scenario`; trace flags via `args=`, printed to `run.stdout` | +| `--real-time`, `--uart-tcp`, `--record` | the `sim` CLI | ```python run = simantic.run_scenario( @@ -133,48 +86,19 @@ run = simantic.run_scenario( "nodeb": {"mcu": "STM32F407VG", "elf": "node.elf"}}, "media": [{"type": "can", "connect": ["nodea.can1", "nodeb.can1"]}]}, timeout=10, # virtual seconds - trace_symbols=["can_send"], trace_memory=["rx_count:4"], trace_interrupts=True) + args=["--trace-symbol", "can_send", "--trace-memory", "rx_count:4"]) -assert "[nodeb] RESULT: PASS" in run.output # UART text, one "[machine] " prefixed line each -sends = [h for h in run.symbol_trace if h["machine"] == "nodea"] +assert "[nodeb] RESULT: PASS" in run.output ``` -`run.uart_records`, `run.interrupts`, `run.symbol_trace` and `run.memory_trace` -hold the same records the `Sim` methods of those names return, for every -machine, in virtual-time order. Tracing never halts a machine, and a traced -run prints what an untraced one does. A symbol is traced on every machine -whose ELF defines it. - -## Timing assertions and the quantum - -Renode delivers scheduled events on sync points, so a timing assertion is only -as fine as the scenario's `quantum`. The default (100 µs) is coarser than one -UART character at 115200 (86.8 µs). Set `quantum` in the scenario dict (seconds) -before asserting on intervals — an assertion at the default quantum measures -the time resolution, not the firmware. +A scenario runs to its timeout in one call and is read afterwards; it cannot be +paused part-way. Tracing never halts a machine. ## How it runs -`Sim` hosts the engine (`Simantic.Core`, .NET) inside the Python process via -pythonnet and holds a `Session` object — the same `SessionSpec`/`Session` API -the `sim` CLI is built on. There is no subprocess and no protocol: method -calls are method calls, records are objects. The engine is located in this -order: `$SIMANTIC_ENGINE_DIR`; a development `sim` publish directory -(`$SIMANTIC_SIM`); the managed install under `~/.simantic/engine//` -— and if none exists it is fetched from the public release on the spot. -The managed engine bundles its own .NET runtime, so a machine needs only -Python. - -Consequences worth knowing: - -- **One emulation per process.** The engine keeps process-global state, so - run N simulations as N processes (`ProcessPoolExecutor`), never N threads. -- **Scripted peers run in your interpreter.** A `ScriptedNetworkService` or - `ScriptedCellularPeer` script executes on the emulation thread of the same - Python process; `expect`/`run_for` release the GIL while the clock runs so - the peer can take it. Don't hold locks across those calls. -- **Exceptions are engine exceptions.** A bad platform or a missing ELF - surfaces as `SimError` with the engine's message. +`Sim` hosts pyrite in the Python process: no subprocess, no protocol. Method +calls are method calls. A bad platform or a missing ELF surfaces as `SimError` +with the engine's message. ## Why not MCP diff --git a/docs/writing-tests.md b/docs/writing-tests.md index 124dc22..21dc4dc 100644 --- a/docs/writing-tests.md +++ b/docs/writing-tests.md @@ -20,9 +20,9 @@ below are in [session-api.md](session-api.md). everything you will want (decoded output, bus frames), and let each test look at a different slice. 2. **Assert on the wire, not on the firmware's own claims.** Count frames per - peer (`Sim.frames()`, each record has `label` = the part name, `t`, `data`, + peer (the bundle's bus frames, each record has `label` = the part name, `t`, `data`, `miso`, ...) and decode the firmware's real output (here MAVLink read with - `Sim.read_uart_bytes()`, which is byte-accurate; `read_uart()` is not). + the recorded UART bytes, which is byte-accurate; `read_uart()` is not). 3. **Expected values live in the peer's constants.** The GPS peer reports a 3D fix at a fixed position, the IMU peer tilts the board by `PITCH_TRIM_DEG`. The test asserts the firmware reports what the peer put on the wire, and its @@ -38,23 +38,15 @@ below are in [session-api.md](session-api.md). def run(): scenario = yaml.safe_load((PEERS / "scenario.yaml").read_text()) scenario["machines"]["dut"]["elf"] = str(Path(elf).resolve()) - with Sim(scenario=scenario, machine="dut", uart=MAVLINK_UART, cwd=PEERS) as s: - s.run_for(RUN_SECONDS) # virtual seconds - msgs = list(mavlink_v2(s.read_uart_bytes(from_start=True))) - bus = Counter(f["label"] for f in s.frames(from_start=True)) - return msgs, bus + return simantic.run_scenario(scenario, timeout=RUN_SECONDS, base=PEERS) # virtual seconds ``` -- `Sim(scenario=...)` takes the same dict shape as `sim --scenario`'s YAML; - `machine=` picks the machine whose UART `uart=` names. `cwd=` is the base for - relative `elf:` and `script:` paths. -- Peers (`parts:`) need the default `renode` engine. The pytest plugin's - `--sim-backend=renode|rust|both` option chooses the engine for the `sim` - fixture, and `simantic.run_scenario` (a one-shot Rust-engine run) rejects - `parts:` with `NotSupported`. That is why the sample builds `Sim` itself. -- The `sim` fixture is function-scoped (a factory that closes each machine when - its test ends). For a module-wide run, as here, construct `Sim` in your own - module-scoped fixture, as above, and close it with the `with` block. +- `run_scenario` takes the same dict shape as `sim run --scenario`'s YAML and + runs the `sim` binary, peers included; `base=` is the base for relative + `elf:` and `script:` paths, and `args=` passes any other `sim run` flag. It + returns the recorded bundle as `run.output`, `run.uart_records`, + `run.interrupts`, `run.frames`, `run.stdout` (trace output) and `run.stderr` (peer output). +- `Sim` is single-machine and has no peers; use it for stepping, not for this. - The sample skips when `FIRMWARE` is unset: the firmware is yours to supply. ## The assertions @@ -75,8 +67,7 @@ def test_baro_reaches_scaled_pressure(run): Frames are timestamped in virtual time, so a rate is a subtraction between consecutive frames and an ordering is a comparison (see the README's "Test what -went over the wire"). Prefer `expect()` over polling loops on the Renode engine, -where each Python/engine hand-off costs a few hundred microseconds. +went over the wire"). ## Running it @@ -86,8 +77,7 @@ FIRMWARE=/path/to/firmware.elf pytest tests # from a simantic-dev/examples che ``` If a test fails, read the peer's own log first: peers report with `ctx.info` / -`ctx.warning`; pass `show_logs=True` to `Sim`, or read `s.logs()`, to see -the simulator's log. +`ctx.warning`; run the `sim` binary with `-v` to see the simulator's log. ## Things this sample does not do diff --git a/pyproject.toml b/pyproject.toml index 31a48eb..681c7ce 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,10 +4,8 @@ build-backend = "hatchling.build" [project] name = "simantic" -version = "0.4.2" +version = "0.5.0" description = "Python SDK and pytest plugin for the Simantic circuit and firmware simulators" -# PyYAML reads sim-fixtures test.yaml manifests; pythonnet hosts the -# simulation engine (Simantic.Core, .NET) in-process for `simantic.Sim`. readme = "README.md" requires-python = ">=3.11" license = "MIT" @@ -22,7 +20,7 @@ classifiers = [ "Programming Language :: Python :: 3", "Topic :: Scientific/Engineering :: Electronic Design Automation (EDA)", ] -dependencies = ["pyyaml>=6", "pythonnet>=3.0.5"] +dependencies = ["pyyaml>=6"] [project.urls] Homepage = "https://simantic.dev" diff --git a/src/simantic/__init__.py b/src/simantic/__init__.py index cba3e76..20ed487 100644 --- a/src/simantic/__init__.py +++ b/src/simantic/__init__.py @@ -1,6 +1,6 @@ """Python control of the Simantic simulators. -The firmware engine, hosted in your process. +The `sim` binary runs batches; the engine can also be hosted in your process. import simantic @@ -9,8 +9,8 @@ run = simantic.run_firmware("fw.elf", mcu="STM32F401RE", expect=["RESULT: PASS"]) # one-shot -The engine is fetched on first use. The `sim` CLI is optional; point -`$SIMANTIC_SIM` at one, or put it on PATH. +`run_firmware` and `run_scenario` need the `sim` binary: point `$SIMANTIC_SIM` +at one, or put it on PATH. `Sim` fetches its engine on first use. Installing this package also registers a pytest plugin that turns `test.yaml` fixture manifests into individually addressable pytest items. @@ -22,18 +22,17 @@ UnsupportedManifest, load_manifest, ) -from .mcu import ServerNotConfigured, SimError, SimRun, sim_binary +from .mcu import ScenarioRun, SimError, SimRun, run_scenario, sim_binary from .mcu import run as run_firmware -from .engine import EngineNotFound, engine_dir, rust_engine_dir -from .session import BACKENDS, ExpectTimeout, Match, Sim -from ._rust import NotSupported, ScenarioRun, run_scenario +from .engine import EngineNotFound, rust_engine_dir +from .session import ExpectTimeout, Match, Sim +from ._rust import NotSupported -__version__ = "0.4.2" +__version__ = "0.5.0" __all__ = [ # firmware "Manifest", - "ServerNotConfigured", "SimError", "SimRun", "UnsupportedManifest", @@ -48,8 +47,6 @@ "ExpectTimeout", "EngineNotFound", "NotSupported", - "BACKENDS", - "engine_dir", "rust_engine_dir", # shared "BinaryNotFound", diff --git a/src/simantic/_cli.py b/src/simantic/_cli.py index 9468b28..5b035dd 100644 --- a/src/simantic/_cli.py +++ b/src/simantic/_cli.py @@ -71,10 +71,8 @@ def _status(args) -> int: print(f" {name}: {locate(name, env)}") except BinaryNotFound: print(f" {name}: not found (run `simantic install {name}`)") - engine = install.installed_engine() - print(f" engine: {engine if engine else 'not found (no longer installable; local builds only)'}") rust = install.installed_rust_engine() - missing = 'not found (fetched on first use of backend="rust", or `simantic install engine-rust`)' + missing = 'not found (fetched on first use of `Sim`, or `simantic install engine-rust`)' print(f" engine-rust: {rust if rust else missing}") print(telemetry.describe()) return 0 diff --git a/src/simantic/_elf.py b/src/simantic/_elf.py index e583c27..38baee4 100644 --- a/src/simantic/_elf.py +++ b/src/simantic/_elf.py @@ -1,8 +1,7 @@ """Symbol addresses from a 32-bit little-endian ELF, with the standard library. -The Renode backend resolves symbols inside the engine. The Rust engine does -not carry a symbol table, so `Sim.symbol()` on that backend reads `.symtab` -here — the same answer, from the same file. +The in-process engine does not carry a symbol table, so `Sim.symbol()` reads +`.symtab` here. """ from __future__ import annotations diff --git a/src/simantic/_rust.py b/src/simantic/_rust.py index 436e4b3..f3448ab 100644 --- a/src/simantic/_rust.py +++ b/src/simantic/_rust.py @@ -1,17 +1,13 @@ -"""The Rust backend of `Sim`: `simantic_rust.Session` hosted in this process. +"""The engine behind `Sim`: `simantic_rust.Session` hosted in this process. -Same vocabulary as the Renode backend, one machine at a time. What the Rust -engine does not do yet raises `NotSupported` rather than silently doing -nothing — the gap list is the table in docs/session-api.md. +One machine at a time. What it does not do yet raises `NotSupported` rather +than silently doing nothing; `run_scenario` runs what it cannot. """ from __future__ import annotations -import math -import os import re import time -from dataclasses import dataclass, field from pathlib import Path from . import _elf @@ -30,30 +26,11 @@ def _exception_name(vector: int) -> str | None: return f"IRQ{vector - 16}" if vector >= 16 else _CORTEX_M_EXCEPTIONS.get(vector) -_NEWER_ENGINE = "on backend='rust' needs a newer Rust engine: run `simantic install engine-rust`" +_NEWER_ENGINE = "needs a newer Rust engine: run `simantic install engine-rust`" class NotSupported(SimError): - """The Rust backend has no implementation of this yet.""" - - -@dataclass -class ScenarioRun: - """What a one-shot scenario produced. Each list holds the records the - `Sim` method of the same name returns, for every machine, in time order.""" - - uart_records: list[dict] - warnings: list[str] - interrupts: list[dict] = field(default_factory=list) - symbol_trace: list[dict] = field(default_factory=list) - memory_trace: list[dict] = field(default_factory=list) - - @property - def output(self) -> str: - """The UART text as `sim --scenario` prints it, one `[machine] ` prefixed line each.""" - from .mcu import render_uart - - return render_uart(self.uart_records, multi=True) + """The in-process engine has no implementation of this yet.""" def _arm(elf: bytes) -> bool: @@ -61,119 +38,18 @@ def _arm(elf: bytes) -> bool: return int.from_bytes(elf[18:20], "little") == 40 -def run_scenario(scenario: dict, timeout: float, *, base: str | os.PathLike[str] | None = None, engine_dir=None, - trace_symbols: list[str] = (), trace_memory: list[str] = (), - trace_interrupts: bool = False) -> ScenarioRun: - """Run every machine of `scenario` with its media to `timeout` virtual - seconds on the Rust engine, in one call: the runner behind the Rust - `sim --scenario`. Relative paths resolve against `base` (default: the cwd). - - Nothing can be sent or read while it runs; that is what `Sim` is for, and - on this backend `Sim` drives one machine. What happened is read afterwards: - `trace_symbols`, `trace_memory` and `trace_interrupts` mean what they do on - `Sim`, none of them halts a machine, and each applies to every machine (a - symbol, to every machine whose ELF defines it). - """ - base = Path(base) if base else Path.cwd() - engine = load_rust(engine_dir) - if not hasattr(engine, "run_scenario"): - raise NotSupported(f"a one-shot scenario {_NEWER_ENGINE}") - if scenario.get("networkServices"): - raise NotSupported("backend='rust' does not take network services through simantic yet") - machines, declared = [], set() - for name, m in scenario["machines"].items(): - if m.get("parts"): - raise NotSupported("backend='rust' cannot build parts= yet; use backend='renode'") - if m.get("gdbPort"): - raise NotSupported("backend='rust' does not serve gdbPort through simantic; use `sim --scenario` with gdbPort, or backend='renode'") - board = (base / m["overlay"]).read_text() if m.get("overlay") else "" - declared |= {f"machine '{name}': {entry}" for entry in re.findall(r"^(\w+)\s*:", board, re.M)} - machines.append((name, m["mcu"], board, (base / m["elf"]).read_bytes())) - # Per machine: its symbols, and whether its registers are r0-r3 or a0-a3. - symbols = {name: _elf.symbols((base / (m.get("symbolsElfPath") or m["elf"])).read_bytes()) - for name, m in scenario["machines"].items()} if trace_symbols or trace_memory else {} - cortex_m = {name: _arm(elf) for name, _mcu, _board, elf in machines} - - def resolve(where: str) -> list[tuple[str, int]]: - if where[:2].lower() == "0x": - return [(name, int(where, 0)) for name in symbols] - found = [(name, table[where]) for name, table in symbols.items() if where in table] - if not found: - raise SimError(f"no symbol {where!r} in any machine's ELF") - return found - - traced = {(name, address): symbol for symbol in trace_symbols for name, address in resolve(symbol)} - watches = [] - for spec in trace_memory: - # A symbol or 0xADDR, optionally ":len" in bytes, as on `Sim`. - where, _, length = spec.partition(":") - watches += [(name, address, int(length, 0) if length else 4, spec) for name, address in resolve(where)] - # Only what was asked for is passed, so an engine that predates tracing - # still runs every scenario that does not use it. - options = {} - if traced: - options["trace_pc"] = list(traced) - if watches: - options["trace_memory"] = [w[:3] for w in watches] - if trace_interrupts: - options["trace_interrupts"] = True - media = [(med["type"], [tuple(end.split(".", 1)) for end in med.get("connect") or []]) - for med in scenario.get("media") or []] - # A net end is `machine.peripheral.pin`. - nets = [[(m, p, int(pin)) for m, p, pin in (end.rsplit(".", 2) for end in net)] for net in scenario.get("nets") or []] - # Script paths inside an overlay are relative to `base`, and the engine - # opens them relative to the working directory. - cwd = os.getcwd() - os.chdir(base) - try: - uarts, runs, warnings, skipped, *traces = engine.run_scenario( - machines, media, nets, [], max(1, math.ceil(timeout)), scenario.get("quantum"), **options) - except TypeError as exc: - raise NotSupported(f"tracing a one-shot scenario {_NEWER_ENGINE}") from exc - except Exception as exc: - raise SimError(str(exc)) from exc - finally: - os.chdir(cwd) - lost = [s for s in skipped if s.split(" (")[0] in declared] - if lost: - raise NotSupported(f"backend='rust' has no model for {', '.join(lost)}; use backend='renode'") - records = [{"t": t, "machine": uarts[i][0], "label": uarts[i][1], - "text": bytes(data).decode("latin-1"), "bytes": bytes(data)} for t, i, data in runs] - run = ScenarioRun(records, [*warnings, *(f"{s}: not modeled, reads as zero" for s in skipped)]) - if traces: - pc_hits, accesses, interrupts = traces[0] - # As `Sim` names a pointer argument: absolute symbols are not addresses. - names = {name: {v: k for k, v in _elf.symbols((base / (m.get("symbolsElfPath") or m["elf"])).read_bytes(), - absolute=False).items()} - for name, m in scenario["machines"].items()} if pc_hits else {} - run.symbol_trace = [ - {"t": t, "machine": m, "symbol": traced[m, pc], "address": pc, - "args": [{"register": reg, "value": v, "symbol": names[m].get(v)} - for reg, v in zip(("r0", "r1", "r2", "r3") if cortex_m[m] else ("a0", "a1", "a2", "a3"), args)]} - for t, m, _core, pc, args in pc_hits] - run.memory_trace = [ - {"t": t, "machine": m, "kind": "Write" if write else "Read", "address": address, "value": value, - "watch": next((spec for wm, lo, n, spec in watches if wm == m and lo < address + width and address < lo + n), None)} - for t, m, address, width, value, write in accesses] - run.interrupts = [ - {"t": t, "machine": m, "direction": "Enter" if entry else "Exit", "exception": vector, - "name": _exception_name(vector) if cortex_m[m] else None} - for t, m, _core, vector, entry in interrupts] - return run - - class RustBackend: def __init__(self, machines: list[dict], *, base: Path, media, services, quantum, trace_symbols, trace_memory, trace_interrupts, itm, engine_dir): if len(machines) != 1: - raise NotSupported("backend='rust' runs one machine; multi-machine scenarios need backend='renode'") + raise NotSupported("Sim runs one machine; use run_scenario for several") if media or services: - raise NotSupported("backend='rust' has no media or network services yet") + raise NotSupported("Sim has no media or network services; use run_scenario") m = machines[0] if m.get("parts"): - raise NotSupported("backend='rust' cannot build parts= yet; use backend='renode'") + raise NotSupported("Sim cannot build parts= yet; use run_scenario") if m.get("gdbPort"): - raise NotSupported("backend='rust' does not serve gdbPort through simantic; use `sim --scenario` with gdbPort, or backend='renode'") + raise NotSupported("Sim does not serve gdbPort; use `sim run --scenario` with gdbPort") self.machines = [m["name"]] self._elf = (base / m["elf"]).read_bytes() # Symbols come from the side ELF when the image that boots is stripped. @@ -198,7 +74,7 @@ def __init__(self, machines: list[dict], *, base: Path, media, services, quantum declared = set(re.findall(r"^(\w+)\s*:", board, re.M)) lost = [s for s in skipped if s.split(" ")[0] in declared] if lost: - raise NotSupported(f"backend='rust' has no model for {', '.join(lost)}; use backend='renode'") + raise NotSupported(f"no model for {', '.join(lost)}") self._symbols: dict[str, int] | None = None self._trace_interrupts, self._interrupts_seen = trace_interrupts, 0 self._itm = itm @@ -260,7 +136,7 @@ def _attach(self, method: str, what: str, *args) -> None: raise SimError(str(e)) from None def inject_radio(self, *_a, **_k) -> None: - raise NotSupported("backend='rust' has no radio injection yet") + raise NotSupported("no radio injection yet") # -- time ------------------------------------------------------------- diff --git a/src/simantic/engine.py b/src/simantic/engine.py index 8bd34e2..daae4f9 100644 --- a/src/simantic/engine.py +++ b/src/simantic/engine.py @@ -1,13 +1,8 @@ -"""Loading the simulation engine into this process. +"""Loading the in-process engine: the `simantic_rust` extension module. -`Simantic.Core` is a .NET library; `sim` is one face of it and this package is -another. `simantic install` (or `$SIMANTIC_SIM`) gives us the directory that -holds `sim`, `Simantic.Core.dll` and the runtime config, which is everything -needed to host the engine here via pythonnet. Nothing is spawned: the Python -objects *are* the session. - -One emulation per process — the engine keeps process-global state — so -parallel runs are parallel processes (`ProcessPoolExecutor`). +`simantic install engine-rust` (or `$SIMANTIC_RUST_ENGINE_DIR`) gives us the +directory that holds it. Nothing is spawned: the Python objects *are* the +session. """ from __future__ import annotations @@ -18,132 +13,12 @@ from pathlib import Path from . import install -from ._locate import BinaryNotFound -from .mcu import sim_binary -ENV_DIR = "SIMANTIC_ENGINE_DIR" RUST_ENV_DIR = "SIMANTIC_RUST_ENGINE_DIR" -#: Oldest engine this package can drive. The Session API it calls landed in -#: sim 0.5.4; an older engine fails with a missing-member error deep inside -#: pythonnet, so it is checked here where the message can say what to do. -MIN_ENGINE = (0, 5, 4) - - -class EngineTooOld(RuntimeError): - """The installed engine predates the API this package calls.""" - - -def _version_of(d: Path) -> tuple | None: - """The engine's version, from the directory name a managed install uses.""" - parts = [] - for piece in d.name.split("-")[0].split("."): - if not piece.isdigit(): - return None - parts.append(int(piece)) - return tuple(parts) if len(parts) >= 3 else None - - -def check_engine_version(d: Path) -> None: - """Refuse an engine older than MIN_ENGINE; unknown versions pass. - - A development publish directory has no version in its name — that path is - the developer's own problem, and blocking it would break local work. - """ - found = _version_of(d) - if found is not None and found < MIN_ENGINE: - want = ".".join(str(p) for p in MIN_ENGINE) - have = ".".join(str(p) for p in found) - raise EngineTooOld( - f"engine {have} at {d} is older than {want}, which this package needs. " - f"Use a newer engine build." - ) - class EngineNotFound(RuntimeError): - """The engine assemblies could not be located or loaded.""" - - -def _is_engine(d: Path) -> bool: - return (d / "Simantic.Core.dll").exists() and (d / "sim.runtimeconfig.json").exists() - - -def engine_dir(explicit: str | os.PathLike[str] | None = None, *, fetch: bool = True) -> Path: - """The directory holding Simantic.Core.dll and sim.runtimeconfig.json. - - Order: an explicit path, $SIMANTIC_ENGINE_DIR, a development `sim` whose - publish directory is beside it ($SIMANTIC_SIM), then the managed install - under ~/.simantic/engine. The C# engine is no longer published or - fetched; the Rust engine is the installable one. - """ - candidates = [] - if explicit is not None: - candidates.append(Path(explicit)) - if os.environ.get(ENV_DIR): - candidates.append(Path(os.environ[ENV_DIR])) - try: - candidates.append(sim_binary().resolve().parent) - except BinaryNotFound: - pass - for d in candidates: - if _is_engine(d): - return d - managed = install.installed_engine() - if managed is not None: - return managed - raise EngineNotFound( - f"Simantic.Core.dll not found. Set ${ENV_DIR}." - ) - - -def _resolve_plugin_assemblies_from(d: Path) -> None: - """Let the runtime find Renode plugin assemblies (e.g. Microsoft.Dynamic, - IronPython) that aren't in sim.deps.json. - - Renode loads some assemblies dynamically (a platform's PythonPeripheral, - for one) rather than through Simantic.Core's own dependency graph, so - they never make it into the CLI's deps.json. The `sim` executable finds - them anyway because a framework-dependent apphost falls back to probing - its own directory; pythonnet's manual coreclr host does not get that - fallback, so a plain AssemblyLoadContext.Resolving hook does it here. - """ - from System import AppDomain # type: ignore[import-not-found] - from System.IO import File as NetFile, Path as NetPath # type: ignore[import-not-found] - from System.Reflection import Assembly # type: ignore[import-not-found] - - def handler(_sender, args): - name = str(args.Name).split(",")[0] - candidate = NetPath.Combine(str(d), name + ".dll") - return Assembly.LoadFrom(candidate) if NetFile.Exists(candidate) else None - - AppDomain.CurrentDomain.AssemblyResolve += handler - - -@cache -def load(explicit: str | os.PathLike[str] | None = None): - """Host the .NET runtime and import Simantic.Core. Returns the Session namespace.""" - d = engine_dir(explicit) - check_engine_version(d) - try: - from pythonnet import load as load_runtime - except ImportError as exc: # pragma: no cover - dependency declared in pyproject - raise EngineNotFound("pythonnet is required to host the engine: pip install pythonnet") from exc - # A managed engine carries its own runtime in dotnet/; a development - # publish directory relies on the machine's ($DOTNET_ROOT / default). - bundled = d / "dotnet" - if (bundled / "host").is_dir(): - load_runtime("coreclr", runtime_config=str(d / "sim.runtimeconfig.json"), dotnet_root=str(bundled)) - else: - load_runtime("coreclr", runtime_config=str(d / "sim.runtimeconfig.json")) - import clr # noqa: F401 (provided by pythonnet after load) - - if str(d) not in sys.path: - sys.path.append(str(d)) - _resolve_plugin_assemblies_from(d) - clr.AddReference("Simantic.Core") - import Simantic.Core.Emulation.Session as session_ns # type: ignore[import-not-found] - - return session_ns + """The engine could not be located or loaded.""" def rust_engine_dir(explicit: str | os.PathLike[str] | None = None, *, fetch: bool = True) -> Path: diff --git a/src/simantic/install.py b/src/simantic/install.py index 6720753..c0b5bd9 100644 --- a/src/simantic/install.py +++ b/src/simantic/install.py @@ -374,27 +374,6 @@ def install(binary: str, *, force: bool = False, channel: str | None = None) -> return target -# -- the engine: Simantic.Core + a private .NET runtime, for in-process use ---- - -def engine_root() -> Path: - """Where engine releases live: ~/.simantic/engine//.""" - return simantic_home() / "engine" - - -def installed_engine() -> Path | None: - """The newest installed engine directory, or None.""" - root = engine_root() - if not root.exists(): - return None - candidates = [ - d for d in root.iterdir() - if (d / "Simantic.Core.dll").exists() and (d / "sim.runtimeconfig.json").exists() - ] - if not candidates: - return None - return max(candidates, key=lambda d: _version_key(d.name)) - - def _version_key(name: str) -> tuple: """Numeric release order. Anything that is not `N.N.N[-pre]` sorts last, so a stray directory under the engine root can never win the election.""" diff --git a/src/simantic/mcu.py b/src/simantic/mcu.py index 28f9da4..8df7834 100644 --- a/src/simantic/mcu.py +++ b/src/simantic/mcu.py @@ -3,51 +3,36 @@ `mcu=` names a model, which `sim` resolves for you and which requires authentication (`sim auth`); models are not distributed with this package. -`sim` emits no structured report: the only observable is UART text, written -to --output. The verdict therefore comes from substring matching, which is -the contract `test.yaml` manifests use (`expect` / `expect_absent`) and the -reason test firmware conventionally prints a `RESULT: PASS` marker. - -Some installations require a separate simulation server. This module does -not assume either way — it reads that from what the binary reports, so the -same code drives both. +A single-ELF `run()` reads the UART text `sim` prints. The verdict comes from +substring matching, which is the contract `test.yaml` manifests use (`expect` / +`expect_absent`) and the reason test firmware conventionally prints a +`RESULT: PASS` marker. `run_scenario()` runs a scenario file and reads the +structured files `sim --record` writes. """ from __future__ import annotations +import json import os import subprocess import tempfile from collections.abc import Sequence -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path +import yaml + from ._locate import locate from . import telemetry ENV_VAR = "SIMANTIC_SIM" BINARY = "sim" -#: The engines `run(backend=)` can use, the same names `Sim(backend=)` takes: -#: "renode" is the default, driven through the `sim` binary; "rust" is the -#: Rust engine, run in this process. They do not support the same targets and -#: peripherals. -BACKENDS = ("renode", "rust") - class SimError(RuntimeError): """The simulator could not run: a bad platform, a missing ELF, a failed start.""" -class ServerNotConfigured(SimError): - """This installation needs a simulation server and none was given. - - Separate from SimError because it is an unconfigured environment, not a - simulation result: a test runner should skip on it, the way it skips on - a missing binary, rather than report a firmware failure. - """ - - @dataclass(frozen=True) class SimRun: """One firmware run: the UART transcript plus the verdict against it.""" @@ -104,77 +89,32 @@ def run( elf: str | os.PathLike[str], *, mcu: str, - server: str | None = None, - timeout: int = 15, + timeout: float = 15, expect: Sequence[str] = (), expect_absent: Sequence[str] = (), - backend: str | None = None, - use_cached: bool = False, ascii_output: bool = True, only_messages: bool = True, binary: str | os.PathLike[str] | None = None, ) -> SimRun: - """Run one firmware ELF and check its UART against expectations. - - `mcu` names a backend-resolved model (needs auth). `server` defaults to - $SIM_SERVER_URL, which `sim` itself reads — it is passed explicitly only - when given here. - - `use_cached` reuses a previously fetched model from ~/.sim_cache, which - keeps a suite runnable without a round trip per test. + """Run one firmware ELF (`sim run --elf`) and check its UART against expectations. Defaults mirror what a fixture wants to read: ASCII rather than hex, and message text without the `[t] (LABEL)` prefix. """ - if backend is not None and backend not in BACKENDS: - raise ValueError(f"backend must be one of {BACKENDS}, got {backend!r}") telemetry.record("sdk.run_firmware") - if backend == "rust": - from ._rust import run_scenario - - records = run_scenario({"machines": {"machine": {"mcu": mcu, "elf": str(elf)}}}, timeout).uart_records - output = render_uart(records, multi=False) - return SimRun(output=output, exit_code=0, runner="rust engine", - missing=[t for t in expect if t not in output], - forbidden=[t for t in expect_absent if t in output]) with tempfile.TemporaryDirectory() as tmp: out_path = Path(tmp) / "uart.txt" - cmd = [ - str(sim_binary(binary)), - "--elf", str(elf), - "--timeout", str(timeout), - "--output", str(out_path), - "--mcu", mcu, - ] - if server is not None: - cmd += ["--server", server] - if use_cached: - cmd.append("--use-cached") + cmd = [str(sim_binary(binary)), mcu, "run", "--elf", str(elf), + "--timeout", str(timeout), "--output", str(out_path)] if ascii_output: cmd.append("--ascii") if only_messages: cmd.append("--only-messages") - - # Give the subprocess room past the simulated timeout before treating - # it as hung: --timeout bounds simulated time, not wall-clock. - try: - proc = subprocess.run( - cmd, capture_output=True, text=True, timeout=timeout * 4 + 30 - ) - except subprocess.TimeoutExpired as exc: - raise SimError(f"sim did not exit within its wall-clock budget: {exc}") from None - + proc = _run_sim(cmd) output = out_path.read_text(errors="replace") if out_path.exists() else "" if proc.returncode != 0 and not output: - stderr = proc.stderr.strip() - # Matched against what sim reports rather than pre-checked: whether a - # server is needed depends on the installation, not on this package. - if "sim-server" in stderr: - raise ServerNotConfigured( - f"{stderr}\nPass server= or set $SIM_SERVER_URL." - ) - raise SimError(f"sim exited {proc.returncode} with no output\n{stderr}") + raise SimError(f"sim exited {proc.returncode} with no output\n{proc.stderr.strip()}") return SimRun( output=output, @@ -182,3 +122,79 @@ def run( missing=[t for t in expect if t not in output], forbidden=[t for t in expect_absent if t in output], ) + + +def _run_sim(cmd: list[str], cwd: Path | None = None) -> subprocess.CompletedProcess: + # No wall-clock kill: `--timeout` ends the run in virtual time, and how long + # that takes on the host says nothing about the firmware (the pytest plugin + # judges a manifest's own `wall:` budget). + return subprocess.run(cmd, capture_output=True, text=True, cwd=cwd) + + +@dataclass +class ScenarioRun: + """What a one-shot scenario produced, for every machine, in time order.""" + + uart_records: list[dict] + warnings: list[str] + interrupts: list[dict] = field(default_factory=list) + #: `sim --frames` lines: each I2C/SPI/UART/CAN/BLE frame, `[machine] [t s] (label) ...`. + frames: str = "" + #: What `sim` printed on stderr, including scripted peers' own output. + stderr: str = "" + #: What `sim` printed on stdout, including `--trace-*` lines from `args=`. + stdout: str = "" + + @property + def output(self) -> str: + """The UART text as `sim run --scenario --ascii` prints it, one `[machine] ` prefixed line each.""" + return render_uart(self.uart_records, multi=True) + + +#: Scenario keys that name files, resolved against the scenario file by `sim`. +_MACHINE_PATHS = ("elf", "overlay", "symbolsElfPath") +_PART_PATHS = ("script", "image") + + +def _jsonl(path: Path) -> list[dict]: + return [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else [] + + +def run_scenario(scenario: dict, timeout: float, *, base: str | os.PathLike[str] | None = None, + args: Sequence[str] = (), binary: str | os.PathLike[str] | None = None) -> ScenarioRun: + """Run every machine of `scenario` (the `sim run --scenario` YAML, as a dict) + to `timeout` virtual seconds. Relative file paths resolve against `base` + (default: the cwd). Nothing can be sent or read while it runs; what + happened is read afterwards from the run's `--record` bundle. + + `args` go to `sim run` as given (`["--trace-symbol", "can_send"]`); what + they print lands in `stdout`. + """ + base = Path(base) if base else Path.cwd() + absolute = lambda v: str(base / v) # noqa: E731 (Path join keeps an absolute `v` as is) + machines = {} + for name, m in scenario["machines"].items(): + m = dict(m, **{k: absolute(m[k]) for k in _MACHINE_PATHS if m.get(k)}) + if m.get("parts"): + m["parts"] = [dict(p, **{k: absolute(p[k]) for k in _PART_PATHS if p.get(k)}) for p in m["parts"]] + machines[name] = m + telemetry.record("sdk.run_scenario") + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + (tmp / "scenario.yaml").write_text(yaml.safe_dump(dict(scenario, machines=machines))) + cmd = [str(sim_binary(binary)), "run", "--scenario", str(tmp / "scenario.yaml"), + "--timeout", str(timeout), "--record", str(tmp / "rec"), "--frames", str(tmp / "frames.txt"), *args] + # Script paths inside an overlay are relative to `base`, as `sim` opens them from its cwd. + proc = _run_sim(cmd, cwd=base) + rec = tmp / "rec" + if not (rec / "runs.jsonl").exists(): + raise SimError(f"sim exited {proc.returncode} without a record\n{proc.stderr.strip()}") + records = [{"t": r["t"], "machine": r["machine"], "label": r["label"], "text": r["text"], + "bytes": r["text"].encode("latin-1", "replace")} for r in _jsonl(rec / "uart.jsonl")] + interrupts = [{"t": r["t"], "machine": r["machine"], "direction": "Enter" if r["edge"] == "entry" else "Exit", + "exception": r["vector"], "name": r["name"]} for r in _jsonl(rec / "isr.jsonl")] + warnings = [r["message"] for r in _jsonl(rec / "log.jsonl")] + frames = (tmp / "frames.txt").read_text(errors="replace") if (tmp / "frames.txt").exists() else "" + if proc.returncode != 0: + raise SimError(f"sim exited {proc.returncode}\n{proc.stderr.strip()}") + return ScenarioRun(records, warnings, interrupts, frames, proc.stderr, proc.stdout) diff --git a/src/simantic/pytest_plugin.py b/src/simantic/pytest_plugin.py index c08e454..31e3a41 100644 --- a/src/simantic/pytest_plugin.py +++ b/src/simantic/pytest_plugin.py @@ -17,7 +17,7 @@ from ._locate import BinaryNotFound from .fixtures import UnsupportedManifest, load_manifest -from .mcu import ServerNotConfigured, SimError, render_uart, run as run_firmware +from .mcu import SimError, render_uart, run as run_firmware, run_scenario from . import telemetry @@ -79,39 +79,7 @@ def repr_failure(self, excinfo, style=None): class FixtureYamlFile(pytest.File): def collect(self): - choice = self.config.getoption(BACKEND_OPTION) - name = self.path.parent.name - for backend in ["renode", "rust"] if choice == "both" else [choice]: - item = FirmwareItem.from_parent(self, name=f"{name}[{backend}]" if choice == "both" else name) - item.backend = backend - yield item - - -def render_frames(frames: list[dict], multi: bool) -> str: - """Bus frames in the `sim --frames` line shape, so `expect_frames` strings - are the same ones a CLI run would be grepped for: - `[machine] [t s] (label) PROTO Dir summary`.""" - out = [] - for f in frames: - prefix = f"[{f['machine']}] " if multi else "" - out.append(f"{prefix}[{f['t']:.6f}s] ({f['label']}) {f['protocol'].upper()} {f['direction']} {f['summary']}") - return "\n".join(out) - - -def _run_within(sim, virtual_s: float, wall_s: float, step: float = 0.5) -> bool: - """Advance `virtual_s` of virtual time in `step` chunks, giving up once - `wall_s` of host time has passed. Returns False if the budget ran out. - A chunk boundary costs a few hundred microseconds, so half-second steps - are free at the budgets involved.""" - import time - - deadline = time.monotonic() + wall_s - end = sim.time + virtual_s - while sim.time < end - 1e-9: - if time.monotonic() >= deadline: - return False - sim.run_for(min(step, end - sim.time)) - return True + yield FirmwareItem.from_parent(self, name=self.path.parent.name) def _check(name: str, text: str, want: list[str], forbid: list[str]) -> list[str]: @@ -126,19 +94,17 @@ def _check(name: str, text: str, want: list[str], forbid: list[str]) -> list[str class FirmwareItem(_ReportingItem): - """One fixture manifest: boot the machine(s) with their peers, run for the + """One fixture manifest: run the machine(s) with their peers for the manifest's virtual seconds, then check the UART text and bus frames. - Every manifest runs as a scenario through `Sim`, so `mcu:` names resolve - through your account, overlay - fragments with scripted peers are applied, and `media:` wires peers to the - firmware's UARTs and CAN controllers. + Every manifest runs as a scenario through `sim run --scenario`, so `mcu:` + names resolve through your account, overlay fragments and scripted peers + are applied, and `media:` wires peers to the firmware's UARTs and CAN + controllers. """ - backend = "renode" - def runtest(self) -> None: - from .session import Sim + import time try: manifest = load_manifest(self.path) @@ -153,52 +119,31 @@ def runtest(self) -> None: with tempfile.TemporaryDirectory() as tmp: scenario = manifest.scenario(Path(tmp)) - first = next(iter(scenario["machines"])) + started = time.monotonic() try: - if self.backend == "rust": - uart, frames, over_budget, reached = self._run_on_rust(manifest, scenario) - else: - with Sim(scenario=scenario, machine=first, cwd=str(self.config.rootpath)) as sim: - over_budget = not _run_within(sim, manifest.timeout, manifest.wall_budget) - uart = render_uart(sim.uart_records(from_start=True), not manifest.single) - frames = render_frames(sim.frames(from_start=True), not manifest.single) - reached = sim.time - except (BinaryNotFound, ServerNotConfigured) as exc: + run = run_scenario(scenario, manifest.timeout, base=self.config.rootpath) + except BinaryNotFound as exc: pytest.skip(str(exc)) except SimError as exc: raise SimulationFailure(str(exc)) from None + # `sim` cannot be stopped part-way, so the wall budget is judged after the fact. + over_budget = time.monotonic() - started > manifest.wall_budget + uart = render_uart(run.uart_records, not manifest.single) problems = [] if over_budget: problems.append( - f"wall-clock budget of {manifest.wall_budget}s exceeded at virtual t={reached:.3f}s " - f"of {manifest.timeout}s (set `wall:` in test.yaml if this firmware is legitimately slow; " + f"wall-clock budget of {manifest.wall_budget}s exceeded " + f"(set `wall:` in test.yaml if this firmware is legitimately slow; " f"otherwise it is a model gap or a busy-wait worth finding)") problems += _check("UART output", uart, manifest.expect, manifest.expect_absent) - problems += _check("bus frames", frames, manifest.expect_frames, manifest.expect_frames_absent) + problems += _check("bus frames", run.frames, manifest.expect_frames, manifest.expect_frames_absent) if problems: report = "\n".join(problems) + "\n--- UART ---\n" + (uart or "") if manifest.expect_frames or manifest.expect_frames_absent: - report += "\n--- frames ---\n" + (frames or "") + report += "\n--- frames ---\n" + (run.frames or "") raise SimulationFailure(report) - def _run_on_rust(self, manifest, scenario): - """The whole scenario in one call on the Rust engine. It cannot be - stopped part-way, so the wall budget is judged after the fact.""" - import time - - from ._rust import NotSupported, run_scenario - - if manifest.expect_frames or manifest.expect_frames_absent: - pytest.skip("backend='rust' does not capture bus frames for expect_frames yet") - started = time.monotonic() - try: - records = run_scenario(scenario, manifest.timeout, base=self.config.rootpath).uart_records - except NotSupported as exc: - pytest.skip(str(exc)) - over_budget = time.monotonic() - started > manifest.wall_budget - return render_uart(records, not manifest.single), "", over_budget, float(manifest.timeout) - def reportinfo(self): return self.path, 0, f"firmware: {self.name}" @@ -226,77 +171,37 @@ def test_boot(firmware): # --- the in-process test surface ------------------------------------------- # # The fixtures above run a whole manifest through the `sim` binary: one -# subprocess, and with it ~3 s of engine start-up, per fixture. That is the -# right shape for a manifest, which is one coarse pass/fail. -# -# Hand-written tests are the other shape: many assertions against one running -# machine. For those, `sim` below drives the engine *in-process*, so start-up is -# paid once per worker instead of once per test. -# -# One discipline this surface exists to encode (measured; see -# docs/competitors/simantic-py-review-vs-pyrenode3.md §1b): on the Renode -# backend every hand-off between Python and the engine costs ~400-800 us, -# because resuming rendezvouses with Renode's time-source dispatcher threads. -# Reading is free -- it is the pause/resume that is not. So prefer `expect()`, -# which crosses once, over a poll loop that crosses per millisecond. On the -# Rust backend the same hand-off is ~1 us and the discipline does not apply. - -BACKEND_OPTION = "--sim-backend" - - -def pytest_addoption(parser): - group = parser.getgroup("simantic") - group.addoption( - BACKEND_OPTION, - default="renode", - choices=["renode", "rust", "both"], - help="engine the `sim` fixture drives; 'both' runs each test on each.", - ) - - -def pytest_generate_tests(metafunc): - """`both` becomes one test item per backend, so failures name the engine.""" - if "sim_backend" not in metafunc.fixturenames: - return - choice = metafunc.config.getoption(BACKEND_OPTION) - backends = ["renode", "rust"] if choice == "both" else [choice] - metafunc.parametrize("sim_backend", backends, scope="session") - - -@pytest.fixture(scope="session") -def sim_backend(request): - """The engine under test. Parametrized by --sim-backend=both.""" - return request.config.getoption(BACKEND_OPTION) +# subprocess per fixture, which is the right shape for a manifest's one coarse +# pass/fail. Hand-written tests are the other shape: many assertions against +# one running machine. For those, `sim` below drives the engine *in-process*, +# so start-up is paid once per worker instead of once per test. @pytest.fixture(scope="session") -def _sim_engine(sim_backend): +def _sim_engine(): """Load the engine once per worker, before any test is timed. Without this the first test in a process absorbs the whole start-up cost - and reads as mysteriously slow; with it, start-up is attributed to the - session where it belongs. Also turns a missing engine into one clear skip - rather than a failure per test. + and reads as mysteriously slow. Also turns a missing engine into one clear + skip rather than a failure per test. """ from . import engine try: - engine.load_rust() if sim_backend == "rust" else engine.load() + engine.load_rust() except engine.EngineNotFound as exc: - pytest.skip(f"no {sim_backend} engine: {exc}") - return sim_backend + pytest.skip(f"no engine: {exc}") @pytest.fixture -def sim(request, sim_backend, _sim_engine): +def sim(request, _sim_engine): """Factory for an in-process simulation, closed when the test ends. def test_timer_fires(sim): s = sim(elf="fw.elf", mcu="STM32F401RE", uart="usart2") s.expect("RESULT: PASS", timeout=8) - Anything the chosen backend cannot do skips rather than fails, so one suite - can run on both engines and report honestly what each covers. On failure the + Anything the engine cannot do skips rather than fails. On failure the UART transcript is attached to the report -- what the firmware printed is almost always the useful evidence, and it is gone once the session closes. """ @@ -306,7 +211,6 @@ def test_timer_fires(sim): made = [] def make(**kwargs): - kwargs.setdefault("backend", sim_backend) try: s = Sim(**kwargs) except NotSupported as exc: @@ -320,9 +224,7 @@ def make(**kwargs): for s in made: if failed: try: - request.node.add_report_section( - "call", f"UART ({s.backend})", s.read_uart(from_start=True) - ) + request.node.add_report_section("call", "UART", s.read_uart(from_start=True)) except Exception: pass s.close() @@ -330,11 +232,10 @@ def make(**kwargs): @pytest.hookimpl(hookwrapper=True) def pytest_runtest_call(item): - """A capability the chosen backend lacks is a skip, not a failure. + """A capability the engine lacks is a skip, not a failure. Scoped to tests that take the `sim` fixture, so this never reinterprets an - unrelated error. It is what lets one suite run on both engines and report - what each actually covers instead of a wall of red on the narrower one. + unrelated error. """ if "sim" not in getattr(item, "fixturenames", ()): yield diff --git a/src/simantic/session.py b/src/simantic/session.py index 90a359f..2ea12ca 100644 --- a/src/simantic/session.py +++ b/src/simantic/session.py @@ -15,47 +15,22 @@ sim.run_for(0.5) # advance exactly 500 virtual ms assert "Traceback" not in sim.read_uart() - scenario = { - "machines": {"c6": {"mcu": "ESP32-C6", "elf": "image.elf"}}, - "networkServices": [{"name": "broker", "host": "192.0.2.1", "port": 1883, - "type": "Antmicro.Renode.Peripherals.Network.ScriptedNetworkService", - "args": "mqtt_broker.py"}], - "quantum": 0.00001, - } - with Sim(scenario=scenario, machine="c6", uart="uart0") as sim: - assert sim.expect("CONNACK verified", timeout=60).virtual_seconds < 5 - -Platforms: `mcu=` names a model, resolved exactly like `sim --mcu` — -from `~/.sim_cache`, else fetched with your stored credentials and cached — -optionally with `parts=`, a list of peers on the board (one dict each: -`{"name": "baro", "type": "i2c-device", "bus": "i2c1", "address": 0x76, -"script": "baro.py"}`; types: uart-device, i2c-device, spi-device, can-node, -sd-card (`image:` a formatted card image), peripheral) and, for what that list cannot express, an `overlay=` -platform fragment. Scenario machines accept the same keys. - -`backend=` picks the engine, both hosted in-process: `"renode"` (the -default; `Simantic.Core`, see `engine.py`) or `"rust"` (`simantic_rust`, the -pure-Rust engine — one machine, faster, and missing some capabilities that -raise `NotSupported` rather than silently no-op; docs/session-api.md has the -table). A script written against one runs unchanged on the other wherever -both tick. This class adds vocabulary, not semantics. +Platforms: `mcu=` names a model, resolved by the engine like `sim ` +(your credentials; models are not distributed with this package). The engine +is `simantic_rust`, hosted in this process: one machine, and what it does not +do yet raises `NotSupported` rather than silently no-op. Peers on the board +(`parts=`), several machines and media are `run_scenario`'s. """ from __future__ import annotations -import json import os import re -import tempfile from pathlib import Path from typing import Any from . import telemetry -from .engine import _version_of, engine_dir as _engine_dir, load -from .fixtures import _resolve_paths -from .mcu import SimError - -BACKENDS = ("renode", "rust") +from ._rust import RustBackend class ExpectTimeout(AssertionError): @@ -85,95 +60,6 @@ def __repr__(self) -> str: return f"Match(t={self.virtual_seconds:.6f}, text={self.text!r})" -def _bytes(net_bytes) -> bytes: - return bytes(bytearray(net_bytes)) if net_bytes is not None else b"" - - -def _uart(r) -> dict: - # `Bytes` arrived in engine 0.5.17. Older engines carry only Text, a - # Latin-1 decoding of the wire bytes (one char per byte), so encoding it - # back is lossless there. - raw = getattr(r, "Bytes", None) - return {"t": r.T, "machine": r.Machine, "label": r.Label, "text": r.Text, - "bytes": _bytes(raw) if raw is not None else r.Text.encode("latin-1")} - - -def _frame(r) -> dict: - # `miso` (SPI only: the bytes the slave shifted back) arrived in engine - # 0.5.17; older engines have no such field, so read it defensively. - miso = getattr(r, "Miso", None) - return {"t": r.T, "machine": r.Machine, "label": r.Label, "protocol": r.Protocol, - "direction": r.Direction, "summary": r.Summary, "id": r.Id, - "data": _bytes(r.Data) if r.Data is not None else None, - "miso": _bytes(miso) if miso is not None else None} - - -def _log(r) -> dict: - return {"t": r.T, "level": r.Level, "source": r.Source, "message": r.Message} - - -def _interrupt(r) -> dict: - return {"t": r.T, "machine": r.Machine, "direction": r.Direction, - "exception": int(r.ExceptionIndex), "name": r.Name} - - -def _as_dict(net_obj) -> dict | None: - """An engine record as plain Python (camelCase keys), via the engine's own JSON.""" - if net_obj is None: - return None - import json - - import clr # type: ignore[import-not-found] - - clr.AddReference("System.Text.Json") - from System.Text.Json import JsonNamingPolicy, JsonSerializer, JsonSerializerOptions # type: ignore[import-not-found] - - opts = JsonSerializerOptions() - opts.PropertyNamingPolicy = JsonNamingPolicy.CamelCase - return json.loads(JsonSerializer.Serialize(net_obj, net_obj.GetType(), opts)) - - -def _symbol_trace(r) -> dict: - return {"t": r.T, "machine": r.Machine, "symbol": r.Symbol, "address": int(r.Address), - "args": [{"register": a.Register, "value": int(a.Value), "symbol": a.Symbol} for a in r.Args]} - - - -def _start_error(exc: Exception) -> str: - """What to tell the user when the engine refuses a platform. A model that - names a constructor argument the engine does not know was published for a - newer engine than the one installed, and that has one fix.""" - text = str(exc) - if "Could not find suitable constructor" in text: - return ("could not start the simulation: this model needs a newer engine " - "than the one installed.\n\n (the C# engine is no longer published; use a newer build)\n\n" - + text.splitlines()[0]) - return f"could not start the simulation: {text}" - - -def _needs_newer_engine(mcus, engine: Path) -> str | None: - """The model's own statement that it needs a newer engine, if it makes one. - - The engine caches each model it resolves with the model's `min_sim_version`. - It is read only after a failed start, so a run that works pays nothing; the - failure it explains can be any shape (a missing constructor argument, a - connection to an input the old engine does not have). - """ - have = _version_of(engine) - if have is None: - return None - for mcu in mcus: - try: - cached = Path(os.environ.get("HOME", "")) / ".sim_cache" / f"{mcu.lower()}.json" - want = json.loads(cached.read_text()).get("min_sim_version") - needed = tuple(int(p) for p in want.split(".")) - except (OSError, ValueError, AttributeError): - continue - if needed > have: - return (f"{mcu} needs engine {want} or newer; " - f"{'.'.join(map(str, have))} is installed.\n\n (the C# engine is no longer published; use a newer build)") - return None - class Sim: """One live simulation, driven from Python. Use as a context manager.""" @@ -192,14 +78,9 @@ def __init__( trace_memory: list[str] = (), trace_interrupts: bool = False, itm: bool = False, - show_logs: bool = False, cwd: str | os.PathLike[str] | None = None, engine_dir: str | os.PathLike[str] | None = None, - backend: str = "renode", ): - if backend not in BACKENDS: - raise ValueError(f"backend must be one of {BACKENDS}, got {backend!r}") - self.backend = backend self.machine = machine self.uart = uart self._cursors = {"uart": 0, "frames": 0, "logs": 0, "interrupts": 0, "symbol_trace": 0, @@ -208,7 +89,6 @@ def __init__( # consumed yet, so sequential expects never miss output that arrived # in a previous call's overshoot. self._pending: list[tuple[float, str]] = [] - self._work = Path(tempfile.mkdtemp(prefix="simantic-session-")) self._base = Path(cwd) if cwd else Path.cwd() # Argument errors are the caller's and must not depend on an engine @@ -231,25 +111,13 @@ def __init__( "overlay": overlay, "symbolsElfPath": symbols_elf}] scenario = scenario or {} - telemetry.record(f"sdk.session.{backend}") - if backend == "rust": - from ._rust import RustBackend - - self._b = RustBackend( - machines, base=self._base, media=scenario.get("media"), - services=scenario.get("networkServices"), quantum=scenario.get("quantum"), - trace_symbols=trace_symbols, trace_memory=trace_memory, trace_interrupts=trace_interrupts, - itm=itm, engine_dir=engine_dir, - ) - else: - if itm: - raise SimError("itm= needs backend='rust'") - self._b = _RenodeBackend( - machines, base=self._base, work=self._work, media=scenario.get("media"), - services=scenario.get("networkServices"), quantum=scenario.get("quantum"), - trace_symbols=trace_symbols, trace_memory=trace_memory, trace_interrupts=trace_interrupts, - show_logs=show_logs, engine_dir=engine_dir, - ) + telemetry.record("sdk.session") + self._b = RustBackend( + machines, base=self._base, media=scenario.get("media"), + services=scenario.get("networkServices"), quantum=scenario.get("quantum"), + trace_symbols=trace_symbols, trace_memory=trace_memory, trace_interrupts=trace_interrupts, + itm=itm, engine_dir=engine_dir, + ) self.machines: list[str] = list(self._b.machines) # -- stimulus ----------------------------------------------------------- @@ -386,18 +254,12 @@ def symbol_trace(self, from_start: bool = False) -> list[dict]: def memory_trace(self, from_start: bool = False) -> list[dict]: """Accesses to trace_memory= ranges (non-halting): {t, machine, watch, kind, - address, value}. `value` is None for a read. backend="rust" only.""" - return self._rust_records("memory_trace", from_start) + address, value}. `value` is None for a read.""" + return self._records("memory_trace", from_start) def itm(self, from_start: bool = False) -> list[dict]: - """ITM stimulus-port output (needs itm=True): {t, machine, port, bytes, text}. - backend="rust" only.""" - return self._rust_records("itm", from_start) - - def _rust_records(self, key: str, from_start: bool) -> list[dict]: - if self.backend != "rust": - raise SimError(f"{key}() needs backend='rust'") - return self._records(key, from_start) + """ITM stimulus-port output (needs itm=True): {t, machine, port, bytes, text}.""" + return self._records("itm", from_start) def read_memory(self, address: int | str, count: int = 4, machine: str | None = None) -> bytes: """Read bytes from the system bus; `address` is an int or a symbol name.""" @@ -469,141 +331,3 @@ def _consume(self, offset: int) -> None: else: self._pending[0] = (t, s[offset:]) offset = 0 - - -class _RenodeBackend: - """`Simantic.Core.Emulation.Session` via pythonnet — the default engine.""" - - _READERS = {"uart": ("ReadUart", _uart), "frames": ("ReadFrames", _frame), "logs": ("ReadLogs", _log), - "interrupts": ("ReadInterrupts", _interrupt), "symbol_trace": ("ReadSymbolTrace", _symbol_trace)} - - def __init__(self, machines: list[dict], *, base: Path, work: Path, media, services, quantum, - trace_symbols, trace_memory, trace_interrupts, show_logs, engine_dir): - self._base, self._work = base, work - ns = load(engine_dir) - spec = ns.SessionSpec() - spec.TraceInterrupts = trace_interrupts - spec.ShowBackendLogs = show_logs - for sym in trace_symbols: - spec.TraceSymbols.Add(sym) - for region in trace_memory: - spec.TraceMemory.Add(region) - for m in machines: - self._add_machine(spec, m["name"], m["mcu"], - self._fragment(ns, m.get("parts"), m.get("overlay")), m["elf"], - m.get("symbolsElfPath")) - for med in media or []: - sm = spec.AddMedium(med["type"], list(med.get("connect") or [])) - sm.Strict = bool(med.get("strict", False)) - if med.get("hostBridge"): - sm.HostBridge = med["hostBridge"] - for svc in services or []: - spec.AddService(svc["name"], svc["host"], int(svc.get("port", 0)), - svc.get("type", "Antmicro.Renode.Peripherals.Network.EchoService"), - self._service_args(svc.get("args", ""))) - if quantum is not None: - spec.QuantumSeconds = float(quantum) - try: - self._session = ns.Session.Start(spec) - except Exception as exc: # .NET exceptions surface as Python exceptions - # Chained, not swallowed: the engine's own exception stays - # reachable as __cause__ so a traceback shows what actually failed - # rather than only this wrapper's summary. - stale = _needs_newer_engine([m["mcu"] for m in machines if m.get("mcu")], - _engine_dir(engine_dir, fetch=False)) - raise SimError(stale or _start_error(exc)) from exc - self.machines = list(self._session.Machines) - - def _fragment(self, ns, parts, overlay) -> str | None: - """The platform fragment for one machine: `parts` (a list of peers, the - customer-facing form) rendered by the engine, then any raw `overlay` - file. Part script paths resolve against cwd= like every other path.""" - text = "" - if parts: - if not hasattr(ns, "Parts"): - raise SimError("parts= needs engine 0.5.17 or newer: use a newer engine build") - parts = [_resolve_paths(p, self._base) for p in parts] - text = ns.Parts.Fragment(json.dumps(parts)) - if overlay: - text += (self._base / overlay).read_text() - return text or None - - def _add_machine(self, spec, name: str, mcu, fragment, elf, symbols_elf) -> None: - """Model name → the engine's own resolver (~/.sim_cache, then the - backend with stored credentials — like `sim --mcu`).""" - elf_path = str(self._base / elf) - sm = spec.AddModel(name, mcu, elf_path, fragment) - if symbols_elf: - sm.SymbolsElfPath = str(self._base / symbols_elf) - - def _service_args(self, args: str) -> str: - # A script path is the common case; make it absolute against cwd=. - p = self._base / args - return str(p) if args and p.exists() else args - - def send(self, data, uart, machine): - self._session.Send(data, uart, machine) - - def inject_gpio(self, peripheral, pin, state, machine): - self._session.InjectGpio(peripheral, pin, state, machine) - - def inject_can(self, peripheral, can_id, data, extended, remote, fd, brs, machine): - self._session.InjectCan(peripheral, can_id, data, extended, remote, fd, brs, machine) - - def inject_radio(self, peripheral, frame, machine): - self._session.InjectRadio(peripheral, frame, machine) - - def inject_adc(self, peripheral, channel, volts): - from ._rust import NotSupported - raise NotSupported("backend='renode' has no ADC injection; put channelMillivolts in the board overlay") - - def attach_i2c(self, peripheral, address, device): - from ._rust import NotSupported - raise NotSupported("backend='renode' takes I2C devices from the platform, not Python objects") - - def attach_spi(self, peripheral, device): - from ._rust import NotSupported - raise NotSupported("backend='renode' takes SPI devices from the platform, not Python objects") - - def run_for(self, seconds: float) -> float: - return self._await(self._session.RunForAsync(seconds)) - - @property - def time(self) -> float: - return self._session.VirtualTime - - def expect(self, pattern, uart, machine, timeout): - r = self._await(self._session.ExpectAsync(pattern, uart, machine, timeout)) - return bool(r.Matched), r.Text, r.VirtualSeconds - - def records(self, kind, cursor, limit): - reader_name, convert = self._READERS[kind] - page = getattr(self._session, reader_name)(cursor, limit) - return [convert(r) for r in page.Records], page.Next, bool(page.Truncated) - - def read_memory(self, address, count, machine) -> bytes: - return _bytes(self._session.ReadMemory(address, count, machine)) - - def symbol(self, name, machine) -> int: - return int(self._session.ResolveSymbol(name, machine)) - - def threads(self, machine): - return _as_dict(self._session.Threads(machine)) - - def heap(self, machine): - return _as_dict(self._session.Heap(machine)) - - def close(self) -> None: - self._session.Dispose() - - @staticmethod - def _await(task): - """Wait for an engine task while releasing the GIL: scripted peers run - Python on the emulation thread and need it while the clock is running.""" - import time - - while not task.IsCompleted: - time.sleep(0.0005) - if task.IsFaulted: - raise SimError(str(task.Exception.GetBaseException().Message)) - return task.Result diff --git a/tests/test_fixtures.py b/tests/test_fixtures.py index 6c94bf6..76c069a 100644 --- a/tests/test_fixtures.py +++ b/tests/test_fixtures.py @@ -1,10 +1,11 @@ """test.yaml manifest handling and UART verdicts. No sim binary required.""" +import os + import pytest from pathlib import Path from simantic import ( - ServerNotConfigured, SimError, SimRun, UnsupportedManifest, @@ -90,37 +91,18 @@ def test_wall_budget_defaults_and_override(tmp_path): assert load_manifest(write(tmp_path, SINGLE + "wall: 7\n")).wall_budget == 7 -def test_run_within_stops_at_the_wall_budget(): - from simantic.pytest_plugin import _run_within - - class SlowSim: - time = 0.0 - def run_for(self, s): - import time - time.sleep(0.05) # 50 ms of host time per chunk - self.time += s - fast = SlowSim() - assert _run_within(fast, 1.0, wall_s=10, step=0.5) is True and abs(fast.time - 1.0) < 1e-9 - slow = SlowSim() - assert _run_within(slow, 100.0, wall_s=0.12, step=0.5) is False and slow.time < 100.0 - - def test_single_machine_manifest_is_a_one_machine_scenario(tmp_path): m = load_manifest(write(tmp_path, SINGLE)) assert m.single assert list(m.scenario(tmp_path)["machines"]) == ["dut"] -def test_render_uart_and_frames_match_the_cli_line_shapes(): - from simantic.pytest_plugin import render_frames, render_uart +def test_render_uart_matches_the_cli_line_shape(): + from simantic.pytest_plugin import render_uart uart = [{"machine": "dut", "label": "UART2", "text": "RESULT: \x01PASS\r\n"}, {"machine": "b", "label": "UART0", "text": "hi\n"}] assert render_uart(uart, multi=False).splitlines()[0] == "RESULT: .PASS" assert "[dut] RESULT: .PASS" in render_uart(uart, multi=True) - frames = [{"t": 1.0885, "machine": "dut", "label": "icp", "protocol": "I2c", "direction": "Tx", - "summary": "addr=0x63 read len=2 data=FF 00"}] - assert render_frames(frames, multi=False) == "[1.088500s] (icp) I2C Tx addr=0x63 read len=2 data=FF 00" - assert render_frames(frames, multi=True).startswith("[dut] [1.088500s]") def test_manifest_without_an_mcu_is_refused(tmp_path): @@ -162,31 +144,12 @@ def test_empty_transcript_is_labelled(): assert "(no UART output)" in run.failure_report() -def test_run_refuses_an_unknown_backend(): - with pytest.raises(ValueError, match="backend must be"): - run_firmware("a.elf", mcu="X", backend="qemu") - - -def test_missing_server_is_distinguished_from_a_firmware_failure(tmp_path): - """Detected from what sim reports, since not every installation needs one.""" - fake = tmp_path / "sim" - fake.write_text( - "#!/bin/sh\n" - "echo 'error: no sim-server configured (pass --server or set SIM_SERVER_URL)' >&2\n" - "exit 2\n" - ) - fake.chmod(0o755) - with pytest.raises(ServerNotConfigured, match="SIM_SERVER_URL"): - run_firmware("a.elf", mcu="X", binary=fake) - - -def test_other_failures_stay_plain_sim_errors(tmp_path): +def test_a_failed_start_is_a_sim_error(tmp_path): fake = tmp_path / "sim" fake.write_text("#!/bin/sh\necho 'error: cannot read elf' >&2\nexit 1\n") fake.chmod(0o755) - with pytest.raises(SimError, match="cannot read elf") as exc: + with pytest.raises(SimError, match="cannot read elf"): run_firmware("a.elf", mcu="X", binary=fake) - assert not isinstance(exc.value, ServerNotConfigured) def test_work_dir_sparse_files_and_services_reach_the_scenario(tmp_path): @@ -209,3 +172,17 @@ def test_sim_binary_flags_make_a_manifest_unsupported(tmp_path): load_manifest(tmp_path / "test.yaml") (tmp_path / "test.yaml").write_text("mcu: X\nelf: fw.elf\nsim_args: ['--ascii']\n") assert load_manifest(tmp_path / "test.yaml").mcu == "X" + + +PARTS_MANIFEST = Path(os.environ.get("SIMANTIC_PARTS_MANIFEST", "/nonexistent")) + + +@pytest.mark.skipif(not (os.environ.get("SIMANTIC_SIM") and PARTS_MANIFEST.exists()), + reason="needs $SIMANTIC_SIM and a test.yaml with `parts:` ($SIMANTIC_PARTS_MANIFEST)") +def test_run_scenario_with_a_scripted_peer_goes_through_the_binary(tmp_path): + from simantic import run_scenario + + m = load_manifest(PARTS_MANIFEST) + assert m.machines["dut"]["parts"] + run = run_scenario(m.scenario(tmp_path), m.timeout, base=m.path.parent) + assert all(text in run.output for text in m.expect) diff --git a/tests/test_install.py b/tests/test_install.py index 8f7e031..3669260 100644 --- a/tests/test_install.py +++ b/tests/test_install.py @@ -27,6 +27,7 @@ def home(tmp_path, monkeypatch): monkeypatch.setenv("HOME", str(tmp_path)) monkeypatch.delenv("SIMANTIC_HOME", raising=False) + monkeypatch.delenv("SIMANTIC_SIM", raising=False) # Fetching requires an account, so the common case here is authenticated. auth.save("smtc_" + "a" * 32, "dev@example.com") monkeypatch.setattr(install, "_release_url", lambda path: "https://releases.invalid/" + path) @@ -534,15 +535,6 @@ def test_version_order_ignores_junk_and_prereleases(): assert max(["zzz", "0.6.2"], key=install._version_key) == "0.6.2" -def test_installed_engine_picks_the_newest_version(home, monkeypatch): - for v in ("0.9.0", "0.10.0", "0.9.1", "zzz"): - d = install.engine_root() / v - d.mkdir(parents=True) - (d / "Simantic.Core.dll").write_bytes(b"") - (d / "sim.runtimeconfig.json").write_bytes(b"{}") - assert install.installed_engine().name == "0.10.0" - - # -- the Rust engine ---------------------------------------------------------- RUST_MANIFEST = { @@ -607,16 +599,6 @@ def test_a_directory_without_the_module_is_not(tmp_path): assert not install.is_rust_engine(tmp_path) -def test_engine_dir_does_not_fetch_the_retired_csharp_engine(home, monkeypatch): - from simantic import engine - - monkeypatch.delenv("SIMANTIC_SIM", raising=False) - monkeypatch.delenv("SIMANTIC_ENGINE_DIR", raising=False) - monkeypatch.setenv("PATH", "") - with pytest.raises(engine.EngineNotFound, match="Simantic.Core.dll not found"): - engine.engine_dir() - - def test_pyrite_manifest_resolves_sim_and_rust_engine(home, monkeypatch): h = "cd" * 32 manifest = {"version": "1.2.3", "artifacts": { diff --git a/tests/test_pytest_surface.py b/tests/test_pytest_surface.py index dab90cc..9cdb3e2 100644 --- a/tests/test_pytest_surface.py +++ b/tests/test_pytest_surface.py @@ -1,8 +1,6 @@ """The pytest surface, exercised against a real engine and real firmware. -These are the tests a user would write. They run on whichever engine ---sim-backend names (`both` runs each twice), so they double as the evidence -that one suite can target both. +These are the tests a user would write. """ import os import pytest @@ -46,15 +44,9 @@ def test_memory_and_time_advance(sim): assert s.time > first -def test_backend_is_the_one_requested(sim, sim_backend): - s = f401(sim, "mips-profile") - assert s.backend == sim_backend - - def test_unsupported_capability_skips_not_fails(sim): - """A multi-machine scenario is Renode-only; on rust this must skip.""" + """The in-process engine runs one machine; a two-machine scenario must skip.""" s = sim(scenario={"machines": { "a": {"elf": f"{FIX}/mips-profile/nucleo_f401re.elf", "mcu": MCU}, "b": {"elf": f"{FIX}/mips-profile/nucleo_f401re.elf", "mcu": MCU}, }}, uart=UART) - assert len(s.machines) == 2 diff --git a/tests/test_rust_backend.py b/tests/test_rust_backend.py index c20a74c..6c83b2d 100644 --- a/tests/test_rust_backend.py +++ b/tests/test_rust_backend.py @@ -141,8 +141,8 @@ def fake_engine(monkeypatch, tmp_path): def test_expect_records_and_symbols_on_rust(fake_engine): mcu, elf = fake_engine - with Sim(elf=elf, mcu=mcu, uart="usart2", backend="rust") as sim: - assert sim.backend == "rust" and sim.machines == ["machine"] + with Sim(elf=elf, mcu=mcu, uart="usart2") as sim: + assert sim.machines == ["machine"] assert FakeSession.instances[0].mcu == "FAKE-MCU" m = sim.expect("boot") assert m.virtual_seconds == pytest.approx(0.002, abs=1e-9) @@ -168,31 +168,16 @@ def test_expect_records_and_symbols_on_rust(fake_engine): def test_unsupported_calls_say_so(fake_engine): mcu, elf = fake_engine - with Sim(elf=elf, mcu=mcu, uart="usart2", backend="rust") as sim: + with Sim(elf=elf, mcu=mcu, uart="usart2") as sim: with pytest.raises(NotSupported): sim.inject_radio("radio", b"\x01") with pytest.raises(NotSupported, match="one machine"): Sim(scenario={"machines": {"a": {"elf": str(elf), "mcu": mcu}, - "b": {"elf": str(elf), "mcu": mcu}}}, backend="rust") - - -def test_backend_name_is_validated(): - with pytest.raises(ValueError, match="backend"): - Sim(elf="fw.elf", mcu="X", backend="qemu") + "b": {"elf": str(elf), "mcu": mcu}}}) # -- the contracts pyrenode3 lacks (docs/competitors/pyrenode3.md §4.7/§4.8) -- -def test_an_engine_older_than_the_api_is_refused(tmp_path): - from simantic.engine import EngineTooOld, check_engine_version - - check_engine_version(tmp_path / "0.5.4") # exactly the minimum - check_engine_version(tmp_path / "0.6.0") - check_engine_version(tmp_path / "dev-publish") # unversioned: developer's own - with pytest.raises(EngineTooOld, match="0.5.3"): - check_engine_version(tmp_path / "0.5.3") - - def test_engine_failures_keep_their_cause(fake_engine, monkeypatch): mcu, elf = fake_engine @@ -201,7 +186,7 @@ def boom(mcu, elf_bytes, board=""): monkeypatch.setattr(sys.modules["simantic_rust"], "Session", boom) with pytest.raises(Exception) as exc: - Sim(elf=elf, mcu=mcu, backend="rust") + Sim(elf=elf, mcu=mcu) assert "board parse error at line 5" in str(exc.value) assert isinstance(exc.value.__cause__, RuntimeError) @@ -209,15 +194,15 @@ def boom(mcu, elf_bytes, board=""): def test_parts_and_unbuilt_overlay_entries_are_not_supported(fake_engine, monkeypatch, tmp_path): mcu, elf = fake_engine with pytest.raises(NotSupported, match="parts="): - Sim(elf=elf, mcu=mcu, parts=[{"type": "spi-slave"}], backend="rust") + Sim(elf=elf, mcu=mcu, parts=[{"type": "spi-slave"}]) monkeypatch.setattr(FakeSession, "skipped", lambda self: ["dma1 (DMA.STM32DMA)", "imu (SPI.NoSuchImu)"], raising=False) overlay = tmp_path / "board.board" overlay.write_text("imu: SPI.NoSuchImu @ spi1\n") with pytest.raises(NotSupported, match="imu"): - Sim(elf=elf, mcu=mcu, overlay=overlay, backend="rust") + Sim(elf=elf, mcu=mcu, overlay=overlay) # The model's own gaps are logged, not fatal. - with Sim(elf=elf, mcu=mcu, backend="rust") as sim: + with Sim(elf=elf, mcu=mcu) as sim: assert any("dma1" in r["message"] for r in sim.logs()) @@ -247,7 +232,7 @@ def test_a_burst_crosses_the_boundary_as_runs_not_per_byte(fake_engine, monkeypa """ mcu, elf = fake_engine monkeypatch.setattr(sys.modules["simantic_rust"], "Session", CountingSession) - with Sim(elf=elf, mcu=mcu, uart="usart2", backend="rust") as sim: + with Sim(elf=elf, mcu=mcu, uart="usart2") as sim: sim.run_for(0.002) text = sim.read_uart(from_start=True) session = CountingSession.instances[-1] @@ -259,14 +244,14 @@ def test_records_are_paged_not_returned_whole(fake_engine): """Observation is pulled in bounded pages, so a long run cannot hand the caller one unbounded list built object by object.""" mcu, elf = fake_engine - with Sim(elf=elf, mcu=mcu, uart="usart2", backend="rust") as sim: + with Sim(elf=elf, mcu=mcu, uart="usart2") as sim: page, cursor, truncated = sim._b.records("uart", 0, 1) assert len(page) <= 1 and cursor <= 1 and isinstance(truncated, bool) def test_threads_heap_and_interrupts_on_rust(fake_engine): mcu, elf = fake_engine - with Sim(elf=elf, mcu=mcu, uart="usart2", backend="rust", trace_interrupts=True) as sim: + with Sim(elf=elf, mcu=mcu, uart="usart2", trace_interrupts=True) as sim: sim.run_for(0.002) assert [(r["direction"], r["name"]) for r in sim.interrupts()] == [("Enter", "SysTick"), ("Exit", "SysTick")] sim.run_for(0.003) @@ -276,14 +261,14 @@ def test_threads_heap_and_interrupts_on_rust(fake_engine): assert threads["threads"][0]["name"] == "main" and threads["threads"][0]["stackHighWaterMarkBytes"] == 200 assert sim.heap() == {"allocator": "sys_heap", "arenaSizeBytes": 4096, "usedBytes": 1096, "freeBytes": 3000, "minimumFreeBytes": None} - with Sim(elf=elf, mcu=mcu, uart="usart2", backend="rust") as sim: + with Sim(elf=elf, mcu=mcu, uart="usart2") as sim: sim.run_for(0.01) assert sim.interrupts() == [] def test_symbol_trace_memory_trace_and_itm_on_rust(fake_engine, tmp_path): mcu, elf = fake_engine - with Sim(elf=elf, mcu=mcu, uart="usart2", backend="rust", itm=True, + with Sim(elf=elf, mcu=mcu, uart="usart2", itm=True, trace_symbols=["main"], trace_memory=["counter", "0x20000010:8"]) as sim: main, counter = sim.symbol("main"), sim.symbol("counter") sim.run_for(0.001) @@ -296,7 +281,7 @@ def test_symbol_trace_memory_trace_and_itm_on_rust(fake_engine, tmp_path): assert [(r["watch"], r["kind"], r["value"]) for r in trace] == [ ("counter", "Write", 7), ("0x20000010:8", "Write", 7), ("counter", "Read", None), ("0x20000010:8", "Read", None)] assert [(r["port"], r["text"]) for r in sim.itm()] == [(0, "hi\n")] - with Sim(elf=elf, mcu=mcu, backend="rust", symbols_elf=elf) as sim: + with Sim(elf=elf, mcu=mcu, symbols_elf=elf) as sim: assert FakeSession.instances[-1].symbols_elf == elf.read_bytes() and sim.itm() == [] @@ -310,113 +295,13 @@ def __init__(self, mcu, elf, board=""): monkeypatch.setattr(sys.modules["simantic_rust"], "Session", Old) with pytest.raises(NotSupported, match="install engine-rust"): - Sim(elf=elf, mcu=mcu, backend="rust", itm=True) - with pytest.raises(NotSupported, match="install engine-rust"): - Sim(elf=elf, mcu=mcu, backend="rust", trace_symbols=["main"]) - - -# -- one-shot scenarios -------------------------------------------------------- - -def _fake_run_scenario(calls, skipped=()): - def run_scenario(machines, media, nets, services, timeout_seconds, quantum, **trace): - calls.append((machines, media, nets, timeout_seconds, quantum)) - result = ([("nodea", "usart2"), ("nodeb", "usart2")], - [(0.1, 0, b"RESULT: PASS\n"), (0.2, 1, b"RESULT: PASS\n")], ["w"], list(skipped)) - if not trace: - return result # an engine that predates tracing returns four - calls.append(trace) - (machine, main), (_, watched, _len) = trace["trace_pc"][0], trace["trace_memory"][0] - return (*result, ([(0.5, machine, 0, main, [1, 2, 3, 4])], - [(0.6, machine, watched, 4, 7, True)], - [(0.7, "nodeb", 0, 15, True)])) - return run_scenario - - -def test_a_scenario_runs_in_one_call_on_rust(fake_engine, monkeypatch, tmp_path): - from simantic._rust import run_scenario - from simantic.mcu import render_uart - - mcu, elf = fake_engine - calls = [] - monkeypatch.setattr(sys.modules["simantic_rust"], "run_scenario", _fake_run_scenario(calls), raising=False) - scenario = {"machines": {"nodea": {"mcu": mcu, "elf": str(elf)}, "nodeb": {"mcu": mcu, "elf": str(elf)}}, - "media": [{"type": "can", "connect": ["nodea.can1", "nodeb.can1"]}], - "nets": [["nodea.gpioa.3", "nodeb.gpiob.4"]], "quantum": 0.0001} - run = run_scenario(scenario, 2.5, base=tmp_path) - records, logs = run.uart_records, run.warnings - machines, media, nets, timeout, quantum = calls[0] - assert machines == [("nodea", mcu, "", elf.read_bytes()), ("nodeb", mcu, "", elf.read_bytes())] - assert media == [("can", [("nodea", "can1"), ("nodeb", "can1")])] - assert nets == [[("nodea", "gpioa", 3), ("nodeb", "gpiob", 4)]] - assert (timeout, quantum) == (3, 0.0001) - assert render_uart(records, multi=True) == "[nodea] RESULT: PASS\n[nodea] \n[nodeb] RESULT: PASS\n[nodeb] " - assert logs == ["w"] - - -def test_a_scenario_with_parts_or_an_old_engine_is_not_supported(fake_engine, monkeypatch, tmp_path): - from simantic._rust import run_scenario - - mcu, elf = fake_engine - one = {"machines": {"dut": {"mcu": mcu, "elf": str(elf)}}} + Sim(elf=elf, mcu=mcu, itm=True) with pytest.raises(NotSupported, match="install engine-rust"): - run_scenario(one, 1, base=tmp_path) - monkeypatch.setattr(sys.modules["simantic_rust"], "run_scenario", _fake_run_scenario([]), raising=False) - with pytest.raises(NotSupported, match="parts="): - run_scenario({"machines": {"dut": {"mcu": mcu, "elf": str(elf), "parts": [{"name": "x"}]}}}, 1, base=tmp_path) - overlay = tmp_path / "o.board" - overlay.write_text("imu: Sensors.Nope @ i2c1 0x68\n") - monkeypatch.setattr(sys.modules["simantic_rust"], "run_scenario", - _fake_run_scenario([], skipped=["machine 'dut': imu (Sensors.Nope)"])) - with pytest.raises(NotSupported, match="imu"): - run_scenario({"machines": {"dut": {"mcu": mcu, "elf": str(elf), "overlay": str(overlay)}}}, 1, base=tmp_path) - - -def test_run_firmware_on_rust_needs_no_sim_binary(fake_engine, monkeypatch, tmp_path): - from simantic import run_firmware - - mcu, elf = fake_engine - monkeypatch.setattr(sys.modules["simantic_rust"], "run_scenario", _fake_run_scenario([]), raising=False) - run = run_firmware(elf, mcu=mcu, backend="rust", expect=["RESULT: PASS"], expect_absent=["FAIL"]) - assert run.passed and run.runner == "rust engine" and "RESULT: PASS" in run.output + Sim(elf=elf, mcu=mcu, trace_symbols=["main"]) -def test_run_firmware_refuses_tlib(): - from simantic import run_firmware - - with pytest.raises(ValueError, match="renode"): - run_firmware("fw.elf", mcu="STM32F401RE", backend="tlib") - - -def test_a_one_shot_scenario_returns_its_traces(fake_engine, monkeypatch, tmp_path): - from simantic import _elf, run_scenario - +def test_gdb_port_is_refused_not_silently_dropped(fake_engine): mcu, elf = fake_engine - symbol, address = next(iter(_elf.symbols(elf.read_bytes()).items())) - calls = [] - monkeypatch.setattr(sys.modules["simantic_rust"], "run_scenario", _fake_run_scenario(calls), raising=False) - scenario = {"machines": {"nodea": {"mcu": mcu, "elf": str(elf)}, "nodeb": {"mcu": mcu, "elf": str(elf)}}} - run = run_scenario(scenario, 1, base=tmp_path, trace_symbols=[symbol], trace_memory=["0x20000000:8"], - trace_interrupts=True) - assert calls[1] == {"trace_pc": [("nodea", address), ("nodeb", address)], - "trace_memory": [("nodea", 0x20000000, 8), ("nodeb", 0x20000000, 8)], - "trace_interrupts": True} - hit, = run.symbol_trace - assert (hit["machine"], hit["symbol"], hit["args"][0]["register"], hit["args"][0]["value"]) == ("nodea", symbol, "r0", 1) - assert run.memory_trace == [{"t": 0.6, "machine": "nodea", "kind": "Write", "address": 0x20000000, "value": 7, - "watch": "0x20000000:8"}] - assert run.interrupts == [{"t": 0.7, "machine": "nodeb", "direction": "Enter", "exception": 15, "name": "SysTick"}] - assert run.output.startswith("[nodea] RESULT: PASS") - with pytest.raises(SimError, match="no symbol"): - run_scenario(scenario, 1, base=tmp_path, trace_symbols=["no_such_function"]) - - -def test_gdb_port_is_refused_not_silently_dropped(fake_engine, monkeypatch, tmp_path): - from simantic._rust import run_scenario - - mcu, elf = fake_engine - monkeypatch.setattr(sys.modules["simantic_rust"], "run_scenario", _fake_run_scenario([]), raising=False) machine = {"mcu": mcu, "elf": str(elf), "gdbPort": 3333} with pytest.raises(NotSupported, match="gdbPort"): - run_scenario({"machines": {"dut": machine}}, 1, base=tmp_path) - with pytest.raises(NotSupported, match="gdbPort"): - Sim(scenario={"machines": {"dut": machine}}, backend="rust") + Sim(scenario={"machines": {"dut": machine}}) diff --git a/tests/test_session.py b/tests/test_session.py index 76c2d9b..3fb54e5 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -7,10 +7,10 @@ import pytest from simantic import ExpectTimeout, Sim -from simantic.engine import EngineNotFound, engine_dir +from simantic.engine import EngineNotFound, rust_engine_dir try: - engine_dir() + rust_engine_dir(fetch=False) HAVE_ENGINE = True except EngineNotFound: HAVE_ENGINE = False @@ -58,7 +58,7 @@ def test_expect_timeout_is_assertion(): @needs_engine def test_symbols_elf_single_machine(): - """symbols_elf= reaches SessionMachine.SymbolsElfPath + """symbols_elf= reaches the engine without upsetting the run. No stripped-image fixture is wired up here, so the companion ELF is the image itself — a smoke test of the plumbing (loading the same ELF's symbols twice isn't a real symbol-resolution @@ -78,27 +78,3 @@ def test_symbols_elf_path_in_scenario(): with Sim(scenario=scenario, uart=UART) as sim: m = sim.expect(r"RESULT: (PASS|FAIL)", timeout=120) assert "PASS" in m - - -def test_start_error_names_the_fix_for_a_stale_engine(): - from simantic.session import _start_error - stale = _start_error(Exception( - "Error E25: Could not find suitable constructor for type 'STM32F4_RCC'.\nConstructor selection report: ...")) - assert "use a newer build" in stale - assert "STM32F4_RCC" in stale and "selection report" not in stale - assert _start_error(Exception("no such file")) == "could not start the simulation: no such file" - - -def test_needs_newer_engine_reads_the_models_minimum(tmp_path, monkeypatch): - from simantic.session import _needs_newer_engine - monkeypatch.setenv("HOME", str(tmp_path)) - (tmp_path / ".sim_cache").mkdir() - (tmp_path / ".sim_cache" / "stm32h753zi.json").write_text('{"min_sim_version": "0.6.3"}') - (tmp_path / ".sim_cache" / "stm32f401re.json").write_text('{"min_sim_version": null}') - old, new = tmp_path / "0.6.1", tmp_path / "0.6.3" - msg = _needs_newer_engine(["STM32F401RE", "STM32H753ZI"], old) - assert "STM32H753ZI needs engine 0.6.3 or newer; 0.6.1 is installed" in msg - assert "use a newer build" in msg - assert _needs_newer_engine(["STM32H753ZI"], new) is None - assert _needs_newer_engine(["NOT-CACHED"], old) is None - assert _needs_newer_engine(["STM32H753ZI"], tmp_path / "publish") is None # dev engine: no version