Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
125 changes: 0 additions & 125 deletions src/simantic/_replx.py

This file was deleted.

25 changes: 14 additions & 11 deletions src/simantic/_rust.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
from dataclasses import dataclass, field
from pathlib import Path

from . import _elf, _replx
from . import _elf
from .engine import load_rust
from .mcu import SimError

Expand Down Expand Up @@ -56,6 +56,11 @@ def output(self) -> str:
return render_uart(self.uart_records, multi=True)


def _arm(elf: bytes) -> bool:
"""Whether the ELF's `e_machine` is ARM (r0-r3 arguments) rather than RISC-V (a0-a3)."""
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:
Expand All @@ -81,14 +86,13 @@ def run_scenario(scenario: dict, timeout: float, *, base: str | os.PathLike[str]
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'")
overlay = base / m["overlay"] if m.get("overlay") else None
if overlay:
declared |= {f"machine '{name}': {entry}" for entry in re.findall(r"^(\w+)\s*:", overlay.read_text(), re.M)}
machines.append((name, _replx.platform_text(mcu=m["mcu"], overlay=overlay), (base / m["elf"]).read_bytes()))
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: "CPU.CortexM" in text for name, text, _elf_bytes in machines}
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":
Expand Down Expand Up @@ -174,8 +178,7 @@ def __init__(self, machines: list[dict], *, base: Path, media, services, quantum
self._elf = (base / m["elf"]).read_bytes()
# Symbols come from the side ELF when the image that boots is stripped.
self._symbols_elf = (base / m["symbolsElfPath"]).read_bytes() if m.get("symbolsElfPath") else self._elf
overlay = base / m["overlay"] if m.get("overlay") else None
text = _replx.platform_text(mcu=m["mcu"], overlay=overlay)
board = (base / m["overlay"]).read_text() if m.get("overlay") else ""
engine = load_rust(engine_dir)
# Only what was asked for is passed, so an engine that predates an
# option still runs every script that does not use it.
Expand All @@ -185,21 +188,21 @@ def __init__(self, machines: list[dict], *, base: Path, media, services, quantum
if itm:
options["itm"] = True
try:
self._s = engine.Session(text, self._elf, **options)
self._s = engine.Session(m["mcu"], self._elf, board=board, **options)
except TypeError as exc:
raise NotSupported(f"{'/'.join(options)} {_NEWER_ENGINE}") from exc
except Exception as exc:
raise SimError(str(exc)) from exc
# An older engine does not report what it left out.
skipped = getattr(self._s, "skipped", list)()
declared = set(re.findall(r"^(\w+)\s*:", overlay.read_text(), re.M)) if overlay else set()
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'")
self._symbols: dict[str, int] | None = None
self._trace_interrupts, self._interrupts_seen = trace_interrupts, 0
self._itm = itm
self._cortex_m = "CPU.CortexM" in text
self._cortex_m = _arm(self._elf)
self._names: dict[int, str] | None = None
self._records: dict[str, list[dict]] = {k: [] for k in (
"uart", "frames", "logs", "interrupts", "symbol_trace", "memory_trace", "itm")}
Expand Down
8 changes: 0 additions & 8 deletions tests/test_fixtures.py
Original file line number Diff line number Diff line change
Expand Up @@ -209,11 +209,3 @@ 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"


def test_model_name_cannot_escape_the_cache_dir(tmp_path, monkeypatch):
from simantic import _replx
from simantic.session import SimError
monkeypatch.setenv("HOME", str(tmp_path))
with pytest.raises(SimError, match="invalid model name"):
_replx.model_replx("../../evil")
18 changes: 0 additions & 18 deletions tests/test_model_auth_error.py

This file was deleted.

46 changes: 14 additions & 32 deletions tests/test_rust_backend.py
Original file line number Diff line number Diff line change
@@ -1,32 +1,16 @@
"""The Rust backend through Sim, against a fake engine module. No Rust build
required: what is tested is the Python half — platform rendering, symbol
lookup, the expect loop, record paging, and that unsupported calls say so."""
required: what is tested is the Python half — symbol lookup, the expect
loop, record paging, and that unsupported calls say so."""
import struct
import sys
import types

import pytest

from simantic import NotSupported, Sim, SimError, _elf, _replx
from simantic import NotSupported, Sim, SimError, _elf
from simantic.engine import load_rust


# -- platform rendering -------------------------------------------------------

def test_render_takes_defaults_and_folds_arithmetic():
text = "cpu: CPU.CortexM\n PerformanceInMips: {{RCC.AHBFreq_Value:84000000 / 1000000 * 1.25}}\n" \
"nvic: X\n systickFrequency: {{RCC.AHBFreq_Value:84000000}}\n cpuType: \"{{cpu:cortex-m4f}}\""
out = _replx.render(text)
assert "PerformanceInMips: 105\n" in out
assert "systickFrequency: 84000000\n" in out
assert 'cpuType: "cortex-m4f"' in out


def test_render_refuses_code_in_templates():
assert _replx.render("x: {{a:abc}}") == "x: abc"
assert _replx.render("x: {{a:1+2}}") == "x: 3"


# -- ELF symbols --------------------------------------------------------------

def _tiny_elf(symbols: dict[str, tuple[int, int]]) -> bytes:
Expand Down Expand Up @@ -63,8 +47,8 @@ class FakeSession:
"""Prints 'boot\\n' then 'RESULT: PASS\\n' on usart2, 1 ms apart, from t=2 ms."""
instances = []

def __init__(self, repl_text, elf, symbols_elf=None, itm=False):
self.repl_text, self.elf, self.t = repl_text, elf, 0.0
def __init__(self, mcu, elf, board="", symbols_elf=None, itm=False):
self.mcu, self.board, self.elf, self.t = mcu, board, elf, 0.0
self.symbols_elf, self.itm_on, self.pcs, self.watches = symbols_elf, itm, [], []
self.sent, self.gpio, self.can, self.attached, self.adc = [], [], [], [], []
self._script = [(0.002, b"boot\n"), (0.003, b"RESULT: PASS\n")]
Expand Down Expand Up @@ -149,7 +133,6 @@ def fake_engine(monkeypatch, tmp_path):
load_rust.cache_clear()
FakeSession.instances.clear()
mcu = "FAKE-MCU"
monkeypatch.setattr(_replx, "model_replx", lambda _mcu: "cpu: CPU.CortexM\n freq: {{F:84000000}}\n")
elf = tmp_path / "fw.elf"
elf.write_bytes(_tiny_elf({"main": (0x08000495, 0x12), "counter": (0x20000004, 0x11)}))
yield mcu, elf
Expand All @@ -160,7 +143,7 @@ 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"]
assert "freq: 84000000" in FakeSession.instances[0].repl_text
assert FakeSession.instances[0].mcu == "FAKE-MCU"
m = sim.expect("boot")
assert m.virtual_seconds == pytest.approx(0.002, abs=1e-9)
m = sim.expect(r"RESULT: (PASS|FAIL)")
Expand Down Expand Up @@ -213,13 +196,13 @@ def test_an_engine_older_than_the_api_is_refused(tmp_path):
def test_engine_failures_keep_their_cause(fake_engine, monkeypatch):
mcu, elf = fake_engine

def boom(repl_text, elf_bytes):
raise RuntimeError("repl parse error at line 5")
def boom(mcu, elf_bytes, board=""):
raise RuntimeError("board parse error at line 5")

monkeypatch.setattr(sys.modules["simantic_rust"], "Session", boom)
with pytest.raises(Exception) as exc:
Sim(elf=elf, mcu=mcu, backend="rust")
assert "repl parse error at line 5" in str(exc.value)
assert "board parse error at line 5" in str(exc.value)
assert isinstance(exc.value.__cause__, RuntimeError)


Expand All @@ -243,8 +226,8 @@ def test_parts_and_unbuilt_overlay_entries_are_not_supported(fake_engine, monkey
class CountingSession(FakeSession):
"""Reports how many objects crossed the boundary, against bytes delivered."""

def __init__(self, repl_text, elf):
super().__init__(repl_text, elf)
def __init__(self, mcu, elf, board=""):
super().__init__(mcu, elf, board)
self.handed_over = 0
self._script = [(0.001, b"x" * 10_000)]

Expand Down Expand Up @@ -321,8 +304,8 @@ def test_an_engine_without_the_debug_calls_says_to_upgrade(fake_engine, monkeypa
mcu, elf = fake_engine

class Old(FakeSession):
def __init__(self, repl_text, elf):
super().__init__(repl_text, elf)
def __init__(self, mcu, elf, board=""):
super().__init__(mcu, elf, board)
trace_pc = property()

monkeypatch.setattr(sys.modules["simantic_rust"], "Session", Old)
Expand Down Expand Up @@ -362,8 +345,7 @@ def test_a_scenario_runs_in_one_call_on_rust(fake_engine, monkeypatch, tmp_path)
run = run_scenario(scenario, 2.5, base=tmp_path)
records, logs = run.uart_records, run.warnings
machines, media, nets, timeout, quantum = calls[0]
assert [(n, e) for n, _text, e in machines] == [("nodea", elf.read_bytes()), ("nodeb", elf.read_bytes())]
assert "freq: 84000000" in machines[0][1]
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)
Expand Down
Loading