From 48be35dc7baad434041a0afee8a2e34cc79547b6 Mon Sep 17 00:00:00 2001 From: Noah Date: Wed, 9 Sep 2026 06:55:07 -0600 Subject: [PATCH] qemu: rs485 silence|restore takes a simulated device off the bus mid-run ui-drive.sh --rs485-devices fixed the roster for the whole boot, so no script could show the guest noticing a device go quiet. mbsim.py now serves a control socket (flare-edge --control); ui-drive.sh opens it next to the pty and hands its path to qmp.py drive (--rs485-control), whose new verb `rs485 silence|restore ADDR` sends one command and judges the reply. A run without a simulated bus records the step as fatal rather than a silent pass. Rig: unit 5 silenced reads online=false after 65 s in the status json, restored reads online=true after 64 s (flare-edge #184). Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01N3G6m9Aw5RyVY4ZowtKzEj --- qemu/tests/qmp.py | 42 ++++++++++++++++++++++++++++++++++-- qemu/tests/test_qmp_drive.py | 24 +++++++++++++++++++-- qemu/tests/ui-drive.sh | 5 ++++- 3 files changed, 66 insertions(+), 5 deletions(-) diff --git a/qemu/tests/qmp.py b/qemu/tests/qmp.py index b79e55f..385c4c9 100755 --- a/qemu/tests/qmp.py +++ b/qemu/tests/qmp.py @@ -37,6 +37,10 @@ a real panel, so these verbs mean the same thing on the rig and on hardware: eats the touch that ends either); every compiled flow script opens with it page | stats | hit X Y print the reply, judge nothing + rs485 silence|restore ADDR + take a simulated bus device off the bus (holds its + address, answers nothing) or put it back; needs + --rs485-control (ui-drive.sh --rs485-devices) ctl WORDS... raw passthrough for anything the channel grows assert_page MENU/TAB the active page is exactly this assert_hit X Y CLASS [TEXT...] @@ -486,7 +490,18 @@ def write_png(img, path): _PILImage.frombytes("RGB", (w, h), data).save(path) -def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=None, refs_path=None): +def rs485_send(control_path, line): + """One command to mbsim.py's control socket (flare-edge tools/modbus-sim, + --control): -> its one-line reply, `ok ...` or `error ...`.""" + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as c: + c.settimeout(5) + c.connect(control_path) + c.sendall((line + "\n").encode()) + return c.makefile("r").readline().strip() + + +def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=None, refs_path=None, + rs485_control=None): os.makedirs(outdir, exist_ok=True) with open(script_path) as fh: lines = fh.readlines() @@ -594,6 +609,26 @@ def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=None, ref need_ctl(lineno, cmd) reply = ctl.send(line if cmd != "ctl" else " ".join(args)) record(lineno, line, "ok", reply) + elif cmd == "rs485": + # rs485 silence|restore ADDR: take a simulated device off the bus + # (it holds its address and answers nothing, exactly an absent + # unit) or put it back, while the guest keeps polling. This is + # the only runtime lever on the roster ui-drive.sh --rs485-devices + # fixed at boot, and it exists so a flow can prove a screen + # noticing an adopted device go quiet (flare-edge #184). No + # simulated bus in this run is fatal: the step could not check + # anything, and a silent skip would read as a pass. + if rs485_control is None: + record(lineno, line, "fatal", "no simulated RS485 bus in this run " + "(ui-drive.sh --rs485-devices)") + elif len(args) != 2 or args[0] not in ("silence", "restore"): + record(lineno, line, "fatal", "expected: rs485 silence|restore ADDR") + else: + try: + reply = rs485_send(rs485_control, f"{args[0]} {args[1]}") + except OSError as e: + reply = f"error control socket: {e}" + record(lineno, line, "ok" if reply.startswith("ok") else "fail", reply) elif cmd == "assert_page": need_ctl(lineno, cmd) want = " ".join(args) @@ -815,6 +850,8 @@ def main(): ctl_path = sys.argv[sys.argv.index("--ctl") + 1] if "--ctl" in sys.argv else None console_path = sys.argv[sys.argv.index("--console") + 1] if "--console" in sys.argv else None refs_path = sys.argv[sys.argv.index("--refs") + 1] if "--refs" in sys.argv else None + rs485_control = (sys.argv[sys.argv.index("--rs485-control") + 1] + if "--rs485-control" in sys.argv else None) s = socket.socket(socket.AF_UNIX) s.connect(path) @@ -827,7 +864,8 @@ def main(): elif cmd == "tap": do_tap(s, f, int(sys.argv[3]), int(sys.argv[4])) elif cmd == "drive": - drive(s, f, sys.argv[3], sys.argv[4], size, ctl_path, console_path, refs_path) + drive(s, f, sys.argv[3], sys.argv[4], size, ctl_path, console_path, refs_path, + rs485_control) elif cmd == "quit": s.sendall(b'{"execute":"quit"}\n') diff --git a/qemu/tests/test_qmp_drive.py b/qemu/tests/test_qmp_drive.py index c5162f0..dfb3078 100755 --- a/qemu/tests/test_qmp_drive.py +++ b/qemu/tests/test_qmp_drive.py @@ -62,7 +62,7 @@ def fake_rpc(sock, sock_file, obj): return {} -def run_script(text, refs=None): +def run_script(text, refs=None, rs485_control=None): """-> (exit code or None, {cmd: row}, rows) for one drive() over TEXT.""" outdir = tempfile.mkdtemp(prefix="qmpdrive.") script = os.path.join(outdir, "s.txt") @@ -78,7 +78,7 @@ def run_script(text, refs=None): try: try: qmp.drive(None, None, script, outdir, SIZE, ctl_path="fake", - console_path=None, refs_path=refs_path) + console_path=None, refs_path=refs_path, rs485_control=rs485_control) except SystemExit as e: rc = e.code finally: @@ -190,6 +190,26 @@ class DriveVerbs(unittest.TestCase): self.assertIn("cannot capture", by["capture_region huge 0 0 999 999 exact"]["detail"]) self.assertIn("unknown tolerance", by["capture_region r2 0 0 8 8 fuzzy"]["detail"]) + def test_rs485_verb_talks_to_the_simulator_or_is_fatal(self): + sent = [] + + def fake_send(path, line): + sent.append((path, line)) + return "ok 5 silent" if line.startswith("silence") else "error no simulated device at 9" + + saved = qmp.rs485_send + qmp.rs485_send = fake_send + try: + rc, by, rows = run_script("rs485 silence 5\nrs485 restore 9\nrs485 bounce 5\n", + rs485_control="fake.ctl") + finally: + qmp.rs485_send = saved + self.assertEqual([r["status"] for r in rows], ["ok", "fail", "fatal"]) + self.assertEqual(sent, [("fake.ctl", "silence 5"), ("fake.ctl", "restore 9")]) + rc, by, rows = run_script("rs485 silence 5\n") + self.assertEqual(rows[0]["status"], "fatal", "no bus in the run must not look like a pass") + self.assertIn("no simulated RS485 bus", rows[0]["detail"]) + def test_unknown_verb_is_fatal_for_the_run(self): # A silently-ignored line is a test that proves nothing, so this one # is the documented exception to "the run continues": drive() exits diff --git a/qemu/tests/ui-drive.sh b/qemu/tests/ui-drive.sh index b0b39c7..067bc06 100755 --- a/qemu/tests/ui-drive.sh +++ b/qemu/tests/ui-drive.sh @@ -166,7 +166,9 @@ if [ -n "$RS485_DEVICES" ]; then dev_args=() IFS=',' read -r -a _devs <<< "$RS485_DEVICES" for d in "${_devs[@]}"; do dev_args+=(--device "$d"); done - python3 "$FLARE_EDGE/tools/modbus-sim/mbsim.py" --port "$WORK/rs.pty" "${dev_args[@]}" > "$WORK/mbsim.log" 2>&1 & + # --control is the runtime lever: qmp.py's `rs485 silence|restore ADDR` + # verb talks to it, so a flow can take a device off the bus mid-run. + python3 "$FLARE_EDGE/tools/modbus-sim/mbsim.py" --port "$WORK/rs.pty" --control "$WORK/rs.ctl" "${dev_args[@]}" > "$WORK/mbsim.log" 2>&1 & SIM_PIDS="$SIM_PIDS $!" echo "== rs485 simulator: $RS485_DEVICES on $WORK/rs.pty" fi @@ -235,6 +237,7 @@ DRIVE_ARGS=(--ctl "$WORK/ctl.sock" --console "$WORK/console.log") # default in two places. [ -n "$REFS_FILE" ] && DRIVE_ARGS+=(--refs "$REFS_FILE") python3 "$HERE/qmp.py" "$WORK/qmp.sock" drive "$SCRIPT" "$OUTDIR" \ + ${RS485_DEVICES:+--rs485-control "$WORK/rs.ctl"} \ "${DRIVE_ARGS[@]}" || drive_rc=$? # The UI must still be alive: see the header. The driver checks this after