crates/nornir-robotui/src/cmd_json.rs
//! # `cmd_json` — the wire-remote robot-drive channel
//!
//! The in-process [`RobotSession`](crate::RobotSession) drives an egui app that
//! lives in the SAME process (an `egui_kittest` harness in the drive job). It
//! cannot reach across a process / container boundary. `cmd_json` closes that
//! gap (the honest `cmd_json` gap of the full-system-autotest design §6.1): it is
//! a **transport-agnostic command/assertion protocol** so a driver drives a
//! **remote** egui process the SAME way it drives a co-located one.
//!
//! ## One vocabulary, two drivers (REUSE)
//!
//! There is exactly ONE robot command vocabulary in this crate —
//! [`Action`] / [`Assert`] / [`Target`] / [`FieldValue`] / [`Step`] from the
//! [`scenario`](crate::scenario) module. The in-process runner
//! ([`RobotSession::run_scenario`](crate::RobotSession::run_scenario)) drives it
//! locally; the wire channel here carries the *same enums* over a byte stream.
//! A [`CmdRequest`] is a thin envelope around them — the wire is not a parallel
//! command set, it is the same set with a transport under it.
//!
//! ```text
//! driver process host process (egui app / container)
//! ┌────────────────┐ JSON lines over ┌──────────────────────────────┐
//! │ CmdJsonClient │──── a byte stream ───▶│ CmdJsonServer<H: CmdHost> │
//! │ .click_button │ (Unix sock / TCP / │ → RobotSession::perform/check│
//! │ .read_state │◀── stdio / duplex) ─│ → egui_kittest harness │
//! └────────────────┘ CmdRequest/Response └──────────────────────────────┘
//! ```
//!
//! ## The three pieces
//! * [`CmdRequest`] / [`CmdResponse`] — the serde protocol (the command set:
//! locate-by-testid / act / assert / read-state / drive-step / ping).
//! * [`CmdJsonServer`] — the **host** side: an egui app mounts it and calls
//! [`serve`](CmdJsonServer::serve) (or [`serve_one`](CmdJsonServer::serve_one))
//! on a [`LineChannel`]; each request is applied to a [`CmdHost`] (the blanket
//! impl drives a [`RobotSession`]) and a [`CmdResponse`] is written back.
//! * [`CmdJsonClient`] — the **driver** side: mirrors the in-process drive API
//! ([`click_button`](CmdJsonClient::click_button),
//! [`type_into`](CmdJsonClient::type_into),
//! [`read_state`](CmdJsonClient::read_state), …) so a robot test / the fleet's
//! `RobotDriveMember` calls it exactly like the local session.
//!
//! ## Transport is swappable
//! The byte stream is abstracted behind [`LineChannel`]. [`IoChannel`] adapts any
//! `BufRead + Write` pair (a Unix socket via `try_clone()`, a TCP stream, or
//! stdio); [`duplex`] gives an in-memory pair for tests / in-process wiring. The
//! protocol never names a transport.
//!
//! ## Concurrency
//! The wire is plain **blocking I/O** — no CPU fan-out, no threads (ROOT LAW #0).
//! A real host runs [`serve`](CmdJsonServer::serve) inside the app's own I/O task
//! (the transport task the design blesses); this module spawns nothing.
use std::io::{self, BufRead, Write};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::scenario::{Action, Assert, Step, Target};
use crate::{status, Observable, RobotSession};
// ─────────────────────────────────────────────────────────────────────────────
// The protocol
// ─────────────────────────────────────────────────────────────────────────────
/// One command a driver sends to a host. The variants ARE the robot vocabulary:
/// each carries the same [`Action`] / [`Assert`] / [`Target`] / [`Step`] the
/// in-process runner uses — the wire adds no new verbs.
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(tag = "cmd", rename_all = "snake_case")]
pub enum CmdRequest {
/// Liveness / handshake — the host answers [`CmdResponse::Pong`].
Ping,
/// **locate-by-testid** — resolve a [`Target`] (label OR AccessKit node id,
/// the robot's "testid") to a concrete node id on the host this frame.
Locate {
/// The widget address to resolve.
target: Target,
},
/// Apply one [`Action`] (click / type / drag / set-field / key / scroll /
/// inject / step) to the host's egui app.
Act {
/// The gesture to perform.
action: Action,
},
/// Evaluate one [`Assert`] against the host's post-action observable state.
Assert {
/// The expectation to check.
assert: Assert,
},
/// Read the host app's full [`Observable::state_json`].
ReadState,
/// Read one RFC-6901 JSON pointer out of the host's `state_json`.
ReadStateAt {
/// The JSON pointer (e.g. `/clicks`).
pointer: String,
},
/// Drive one whole [`Step`] (act THEN assert) — the wire twin of the
/// in-process per-step matrix cell; the host answers [`CmdResponse::Stepped`].
DriveStep {
/// The `(action, assert)` step to drive.
step: Step,
},
}
/// The host's reply to a [`CmdRequest`]. `state_json` rides back inside
/// [`CmdResponse::State`] / [`CmdResponse::StateAt`] so a driver can assert on
/// exactly what the remote app rendered.
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(tag = "resp", rename_all = "snake_case")]
pub enum CmdResponse {
/// Answer to [`CmdRequest::Ping`].
Pong,
/// Answer to [`CmdRequest::Locate`] — the resolved node id, or `None` if the
/// widget is not present this frame.
Located {
/// The AccessKit node id the target resolved to, if any.
id: Option<u64>,
},
/// Answer to [`CmdRequest::Act`] — `ok` false + `error` when the gesture
/// could not be applied (missing widget, etc.).
Acted {
/// Whether the action landed.
ok: bool,
/// The reason it did not, when `ok` is false.
error: Option<String>,
},
/// Answer to [`CmdRequest::Assert`] — `pass` false + `reason` for a FAILED
/// assertion (the red-when-broken signal, surfaced not swallowed).
Asserted {
/// Whether the expectation held.
pass: bool,
/// Why it did not, when `pass` is false.
reason: Option<String>,
},
/// Answer to [`CmdRequest::ReadState`] — the full `state_json` blob.
State {
/// The host app's rendered state.
state: Value,
},
/// Answer to [`CmdRequest::ReadStateAt`] — the value at the pointer, or
/// `None` if the path is absent.
StateAt {
/// The value the pointer addressed.
value: Option<Value>,
},
/// Answer to [`CmdRequest::DriveStep`] — the step's matrix verdict.
Stepped {
/// Whether the step passed (action ok AND assertion held).
pass: bool,
/// The failure reason, or empty on pass.
message: String,
},
/// A protocol-level error (a malformed request line, an unknown command) —
/// distinct from an in-band `Acted { ok: false }` / `Asserted { pass: false }`
/// which are legitimate negative *results*, not transport faults.
Error {
/// What went wrong decoding / dispatching the request.
message: String,
},
}
// ─────────────────────────────────────────────────────────────────────────────
// The host seam
// ─────────────────────────────────────────────────────────────────────────────
/// What a [`CmdJsonServer`] applies each command to — the host-side egui context.
/// The blanket impl below drives a [`RobotSession`] (the same object the
/// in-process runner uses), so mounting the wire on an app is just wrapping its
/// kittest harness in a session. A different backend (a live eframe context) can
/// implement this trait to be driven over the same wire.
pub trait CmdHost {
/// Resolve a [`Target`] to a node id this frame (locate-by-testid).
fn locate(&self, target: &Target) -> Option<u64>;
/// Apply an [`Action`]; `Err(reason)` if it could not be performed.
fn act(&mut self, action: &Action) -> Result<(), String>;
/// Evaluate an [`Assert`]; `Err(reason)` if the expectation did not hold.
fn assert(&self, assert: &Assert) -> Result<(), String>;
/// The app's current `state_json`.
fn state(&self) -> Value;
/// Drive one whole step (act then assert) → `(passed, message)`.
fn drive_step(&mut self, step: &Step) -> (bool, String);
}
impl<A: Observable> CmdHost for RobotSession<'_, A> {
fn locate(&self, target: &Target) -> Option<u64> {
self.find_id(target)
}
fn act(&mut self, action: &Action) -> Result<(), String> {
// REUSE the in-process gesture dispatch — one act path, local or remote.
self.perform(action).map_err(|e| e.to_string())
}
fn assert(&self, assert: &Assert) -> Result<(), String> {
// REUSE the in-process assertion oracle — a wire assertion fails exactly
// where the local one does.
self.check(assert)
}
fn state(&self) -> Value {
// Inherent `RobotSession::state` (not this trait method) — reads the app.
RobotSession::state(self)
}
fn drive_step(&mut self, step: &Step) -> (bool, String) {
let (st, message) = self.run_step(step);
(st == status::PASS, message)
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The transport seam
// ─────────────────────────────────────────────────────────────────────────────
/// A bidirectional, line-framed byte stream — the swappable transport under the
/// protocol. The wire form is **JSON lines**: each [`CmdRequest`] /
/// [`CmdResponse`] is one `serde_json` object terminated by `\n`. Implement this
/// for any carrier (a socket, stdio, an in-memory pipe); the protocol above never
/// names one. [`IoChannel`] provides the impl for `BufRead + Write` carriers.
pub trait LineChannel {
/// Write one framed line (the impl appends the `\n`).
fn write_line(&mut self, line: &str) -> io::Result<()>;
/// Read the next framed line (newline stripped). `Ok(None)` = clean EOF.
fn read_line(&mut self) -> io::Result<Option<String>>;
}
/// A [`LineChannel`] over any split byte stream: a buffered reader half `R` and a
/// writer half `W`.
///
/// * **Unix socket / TCP:** `IoChannel::new(BufReader::new(sock.try_clone()?), sock)`.
/// * **stdio (a container child):** `IoChannel::new(stdin.lock(), stdout.lock())`.
/// * **in-memory (tests / in-process):** wrap the two ends of [`duplex`].
pub struct IoChannel<R: BufRead, W: Write> {
reader: R,
writer: W,
}
impl<R: BufRead, W: Write> IoChannel<R, W> {
/// Build a channel from a buffered read half and a write half.
pub fn new(reader: R, writer: W) -> Self {
Self { reader, writer }
}
}
impl<R: BufRead, W: Write> LineChannel for IoChannel<R, W> {
fn write_line(&mut self, line: &str) -> io::Result<()> {
self.writer.write_all(line.as_bytes())?;
self.writer.write_all(b"\n")?;
self.writer.flush()
}
fn read_line(&mut self) -> io::Result<Option<String>> {
let mut buf = String::new();
let n = self.reader.read_line(&mut buf)?;
if n == 0 {
return Ok(None); // EOF
}
// Strip the trailing newline (and a CR, tolerating CRLF carriers).
while buf.ends_with('\n') || buf.ends_with('\r') {
buf.pop();
}
Ok(Some(buf))
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The host: CmdJsonServer
// ─────────────────────────────────────────────────────────────────────────────
/// The **host** side: owns a [`CmdHost`] (drive an egui app's kittest harness via
/// the [`RobotSession`] blanket impl) and applies wire commands to it. Mount it on
/// an app and call [`serve`](Self::serve) on a [`LineChannel`]; it loops reading
/// requests and writing responses until the driver hangs up (EOF).
pub struct CmdJsonServer<H: CmdHost> {
host: H,
}
impl<H: CmdHost> CmdJsonServer<H> {
/// Mount the server over a host.
pub fn new(host: H) -> Self {
Self { host }
}
/// Borrow the mounted host (e.g. to seed fixtures before serving).
pub fn host(&self) -> &H {
&self.host
}
/// Borrow the mounted host mutably.
pub fn host_mut(&mut self) -> &mut H {
&mut self.host
}
/// Apply ONE decoded request to the host and produce its response. This is the
/// pure dispatch (no I/O) — the twin of the in-process `run_scenario` match.
pub fn handle(&mut self, req: &CmdRequest) -> CmdResponse {
match req {
CmdRequest::Ping => CmdResponse::Pong,
CmdRequest::Locate { target } => CmdResponse::Located {
id: self.host.locate(target),
},
CmdRequest::Act { action } => match self.host.act(action) {
Ok(()) => CmdResponse::Acted { ok: true, error: None },
Err(e) => CmdResponse::Acted { ok: false, error: Some(e) },
},
CmdRequest::Assert { assert } => match self.host.assert(assert) {
Ok(()) => CmdResponse::Asserted { pass: true, reason: None },
Err(e) => CmdResponse::Asserted { pass: false, reason: Some(e) },
},
CmdRequest::ReadState => CmdResponse::State {
state: self.host.state(),
},
CmdRequest::ReadStateAt { pointer } => CmdResponse::StateAt {
value: self.host.state().pointer(pointer).cloned(),
},
CmdRequest::DriveStep { step } => {
let (pass, message) = self.host.drive_step(step);
CmdResponse::Stepped { pass, message }
}
}
}
/// Read one request line, apply it, write one response line. Returns `Ok(true)`
/// if a request was served, `Ok(false)` on clean EOF (the driver hung up). A
/// malformed line becomes a [`CmdResponse::Error`] — a bad command never drops
/// the connection.
pub fn serve_one<C: LineChannel>(&mut self, chan: &mut C) -> io::Result<bool> {
let Some(line) = chan.read_line()? else {
return Ok(false);
};
let resp = match serde_json::from_str::<CmdRequest>(&line) {
Ok(req) => self.handle(&req),
Err(e) => CmdResponse::Error {
message: format!("malformed CmdRequest: {e}"),
},
};
let encoded = serde_json::to_string(&resp).map_err(io::Error::other)?;
chan.write_line(&encoded)?;
Ok(true)
}
/// Serve requests until the driver hangs up (EOF). A real host runs this inside
/// the app's own blocking I/O task — it spawns no thread of its own.
pub fn serve<C: LineChannel>(&mut self, chan: &mut C) -> io::Result<()> {
while self.serve_one(chan)? {}
Ok(())
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The driver: CmdJsonClient
// ─────────────────────────────────────────────────────────────────────────────
/// The **driver** side: sends [`CmdRequest`]s and reads [`CmdResponse`]s over a
/// [`LineChannel`], mirroring the in-process [`RobotSession`] drive API so a robot
/// test looks identical whether the app is co-located or remote.
///
/// The high-level verbs ([`click_button`](Self::click_button),
/// [`type_into`](Self::type_into), [`read_state`](Self::read_state), …) each do
/// one blocking [`roundtrip`](Self::roundtrip): they assume a host is serving the
/// other end of the channel (its own I/O task in production). For lockstep,
/// single-threaded wiring (and tests) use [`send`](Self::send) /
/// [`recv`](Self::recv) with a [`CmdJsonServer::serve_one`] pumped between them.
pub struct CmdJsonClient<C: LineChannel> {
chan: C,
}
impl<C: LineChannel> CmdJsonClient<C> {
/// Wrap a transport as a driver.
pub fn new(chan: C) -> Self {
Self { chan }
}
/// Borrow the underlying transport.
pub fn channel_mut(&mut self) -> &mut C {
&mut self.chan
}
/// Send a request without waiting for its response (the non-blocking half —
/// pair with [`recv`](Self::recv), pumping the host's
/// [`serve_one`](CmdJsonServer::serve_one) in between for single-threaded use).
pub fn send(&mut self, req: &CmdRequest) -> io::Result<()> {
let encoded = serde_json::to_string(req).map_err(io::Error::other)?;
self.chan.write_line(&encoded)
}
/// Receive the next response. Errors if the host hung up (EOF) before replying.
pub fn recv(&mut self) -> io::Result<CmdResponse> {
match self.chan.read_line()? {
Some(line) => serde_json::from_str(&line).map_err(io::Error::other),
None => Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
"host closed the channel before replying",
)),
}
}
/// Send a request and block for its response — the production drive path
/// (a host serving the other end answers it).
pub fn roundtrip(&mut self, req: &CmdRequest) -> io::Result<CmdResponse> {
self.send(req)?;
self.recv()
}
// ── the mirrored drive API (looks identical to RobotSession) ──────────────
/// Liveness check — `Ok(())` iff the host answered [`CmdResponse::Pong`].
pub fn ping(&mut self) -> io::Result<()> {
match self.roundtrip(&CmdRequest::Ping)? {
CmdResponse::Pong => Ok(()),
other => Err(unexpected("Pong", &other)),
}
}
/// Resolve a [`Target`] to a remote node id (locate-by-testid).
pub fn locate(&mut self, target: Target) -> io::Result<Option<u64>> {
match self.roundtrip(&CmdRequest::Locate { target })? {
CmdResponse::Located { id } => Ok(id),
other => Err(unexpected("Located", &other)),
}
}
/// Perform an arbitrary [`Action`] on the remote app. `Err` carries the host's
/// reason when the gesture could not be applied.
pub fn act(&mut self, action: Action) -> io::Result<()> {
match self.roundtrip(&CmdRequest::Act { action })? {
CmdResponse::Acted { ok: true, .. } => Ok(()),
CmdResponse::Acted { ok: false, error } => Err(acted_err(error)),
other => Err(unexpected("Acted", &other)),
}
}
/// Click the widget with this label (mirror of `RobotSession::click_label`).
pub fn click_label(&mut self, label: impl Into<String>) -> io::Result<()> {
self.act(Action::ClickLabel(label.into()))
}
/// Click a button by its text (mirror of `RobotSession::click_button`).
pub fn click_button(&mut self, text: impl Into<String>) -> io::Result<()> {
self.act(Action::ClickButton(text.into()))
}
/// Click a widget by AccessKit node id (mirror of `RobotSession::click_by_id`).
pub fn click_by_id(&mut self, id: u64) -> io::Result<()> {
self.act(Action::ClickById(id))
}
/// Type text into the field with this label (mirror of `type_into`).
pub fn type_into(
&mut self,
label: impl Into<String>,
text: impl Into<String>,
) -> io::Result<()> {
self.act(Action::TypeInto {
label: label.into(),
text: text.into(),
})
}
/// Evaluate an [`Assert`] on the remote app. Returns `Ok(None)` when it held,
/// `Ok(Some(reason))` when it FAILED — a failed assertion is a surfaced
/// negative result, not a transport error (the caller records the red cell).
pub fn assert(&mut self, assert: Assert) -> io::Result<Option<String>> {
match self.roundtrip(&CmdRequest::Assert { assert })? {
CmdResponse::Asserted { pass: true, .. } => Ok(None),
CmdResponse::Asserted { pass: false, reason } => {
Ok(Some(reason.unwrap_or_default()))
}
other => Err(unexpected("Asserted", &other)),
}
}
/// Read the remote app's full `state_json` (mirror of `RobotSession::state`).
pub fn read_state(&mut self) -> io::Result<Value> {
match self.roundtrip(&CmdRequest::ReadState)? {
CmdResponse::State { state } => Ok(state),
other => Err(unexpected("State", &other)),
}
}
/// Read one JSON pointer out of the remote `state_json` (mirror of `state_at`).
pub fn state_at(&mut self, pointer: impl Into<String>) -> io::Result<Option<Value>> {
match self.roundtrip(&CmdRequest::ReadStateAt {
pointer: pointer.into(),
})? {
CmdResponse::StateAt { value } => Ok(value),
other => Err(unexpected("StateAt", &other)),
}
}
/// Drive one whole [`Step`] (act then assert) remotely → `(passed, message)`,
/// the wire twin of the in-process per-step matrix cell.
pub fn drive_step(&mut self, step: Step) -> io::Result<(bool, String)> {
match self.roundtrip(&CmdRequest::DriveStep { step })? {
CmdResponse::Stepped { pass, message } => Ok((pass, message)),
other => Err(unexpected("Stepped", &other)),
}
}
}
/// A host reply that did not match the request — a protocol desync, surfaced as
/// an I/O error (includes a host-emitted [`CmdResponse::Error`]).
fn unexpected(want: &str, got: &CmdResponse) -> io::Error {
io::Error::new(
io::ErrorKind::InvalidData,
format!("expected a {want} response, got {got:?}"),
)
}
/// An `Acted { ok: false }` reason turned into an I/O error for the ergonomic
/// `act`/`click_*`/`type_into` verbs (which mirror the in-process `Result<()>`).
fn acted_err(reason: Option<String>) -> io::Error {
io::Error::new(
io::ErrorKind::Other,
reason.unwrap_or_else(|| "action failed on host".to_string()),
)
}
// ─────────────────────────────────────────────────────────────────────────────
// In-memory duplex (tests + in-process wiring; no socket needed)
// ─────────────────────────────────────────────────────────────────────────────
use std::collections::VecDeque;
use std::sync::{Arc, Condvar, Mutex};
/// One direction of an in-memory byte pipe: a shared buffer + a condvar so a
/// blocking read wakes when the writer appends (or closes). This is pure I/O
/// plumbing (a byte carrier), NOT a parallelism primitive — nothing here spawns.
#[derive(Default)]
struct PipeInner {
data: VecDeque<u8>,
closed: bool,
}
#[derive(Default)]
struct Pipe {
inner: Mutex<PipeInner>,
signal: Condvar,
}
/// One end of an in-memory duplex byte stream: reads from one direction, writes
/// to the other. `Read + Write`, so it wraps in a [`BufReader`](std::io::BufReader)
/// + an [`IoChannel`] exactly like a real socket. [`Clone`] is cheap (shared
/// `Arc`s) so the same end can back both the buffered read half and the write
/// half of an [`IoChannel`].
#[derive(Clone)]
pub struct DuplexEnd {
read_from: Arc<Pipe>,
write_to: Arc<Pipe>,
}
impl DuplexEnd {
/// Close this end's write direction — the peer's next read past the buffered
/// bytes sees EOF (`serve` returns, a blocked `read` unblocks).
pub fn close(&self) {
let mut g = self.write_to.inner.lock().unwrap();
g.closed = true;
self.write_to.signal.notify_all();
}
}
impl io::Read for DuplexEnd {
fn read(&mut self, out: &mut [u8]) -> io::Result<usize> {
if out.is_empty() {
return Ok(0);
}
let mut g = self.read_from.inner.lock().unwrap();
loop {
if !g.data.is_empty() {
let n = out.len().min(g.data.len());
for slot in out.iter_mut().take(n) {
*slot = g.data.pop_front().unwrap();
}
return Ok(n);
}
if g.closed {
return Ok(0); // EOF
}
// No bytes yet and not closed → wait for the writer. In the lockstep,
// single-threaded use the bytes are always already present, so this
// wait is never actually reached.
g = self.read_from.signal.wait(g).unwrap();
}
}
}
impl io::Write for DuplexEnd {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
let mut g = self.write_to.inner.lock().unwrap();
if g.closed {
return Err(io::Error::new(io::ErrorKind::BrokenPipe, "duplex end closed"));
}
g.data.extend(buf.iter().copied());
self.write_to.signal.notify_all();
Ok(buf.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
/// An in-memory duplex byte stream — two cross-wired [`DuplexEnd`]s. What one end
/// writes, the other reads. Use it to wire a [`CmdJsonClient`] to a
/// [`CmdJsonServer`] with no real socket (tests, or an in-process remote-shaped
/// harness). Returns `(a, b)`; give one to each side.
pub fn duplex() -> (DuplexEnd, DuplexEnd) {
let a2b = Arc::new(Pipe::default()); // a writes → b reads
let b2a = Arc::new(Pipe::default()); // b writes → a reads
(
DuplexEnd {
read_from: b2a.clone(),
write_to: a2b.clone(),
},
DuplexEnd {
read_from: a2b,
write_to: b2a,
},
)
}
/// Build an [`IoChannel`] over one [`DuplexEnd`] — the common test/in-process
/// wrap (`BufReader` for the read half, a clone for the write half).
pub fn channel_over(end: DuplexEnd) -> IoChannel<io::BufReader<DuplexEnd>, DuplexEnd> {
IoChannel::new(io::BufReader::new(end.clone()), end)
}
// ─────────────────────────────────────────────────────────────────────────────
// Tests — the LOOPBACK proof: a driver drives a host egui app over the channel
// ─────────────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
use crate::demo::{bump_draw, BumpApp};
use crate::scenario::Step as UiStep;
use egui_kittest::Harness;
use serde_json::json;
/// A `CmdJsonServer` over a `RobotSession` driving the demo Bump app. This is
/// the "host process" — in production the container's egui app builds the same
/// session over ITS harness and calls `serve()` on a socket.
fn bump_host() -> CmdJsonServer<RobotSession<'static, BumpApp>> {
let harness = Harness::new_ui_state(
|ui, app: &mut BumpApp| {
let mut t = crate::Targets::default();
bump_draw(ui, app, &mut t);
},
BumpApp::default(),
);
CmdJsonServer::new(RobotSession::new(harness, "cmd-json-host"))
}
/// The whole point: a DRIVER drives a HOST egui app OVER THE CHANNEL. Single
/// threaded, lockstep — `send` a request, pump the host's `serve_one`, `recv`
/// the reply — proving the full serialize → byte stream → dispatch → apply →
/// serialize-back → deserialize round trip. RED-WHEN-BROKEN: a click sent over
/// the wire must move the remote counter and the wired-back `state_json` must
/// reflect it; break the (de)serialize or the dispatch and this goes red.
#[test]
fn driver_drives_host_over_the_channel() {
let (driver_end, host_end) = duplex();
let mut client = CmdJsonClient::new(channel_over(driver_end));
let mut host = bump_host();
let mut host_chan = channel_over(host_end);
// A tiny lockstep pump: send → host serves one → recv. Stands in for the
// host's own I/O task without any thread.
macro_rules! call {
($req:expr) => {{
client.send(&$req).expect("send");
assert!(host.serve_one(&mut host_chan).expect("serve"), "host served");
client.recv().expect("recv")
}};
}
// Handshake travels the wire.
assert!(matches!(call!(CmdRequest::Ping), CmdResponse::Pong));
// Initial remote state: clicks == 0.
match call!(CmdRequest::ReadStateAt { pointer: "/clicks".into() }) {
CmdResponse::StateAt { value } => assert_eq!(value, Some(json!(0))),
other => panic!("want StateAt, got {other:?}"),
}
// locate-by-testid: the Bump button resolves to a real node id remotely.
let bump_id = match call!(CmdRequest::Locate { target: Target::label("Bump") }) {
CmdResponse::Located { id } => id.expect("Bump located remotely"),
other => panic!("want Located, got {other:?}"),
};
// Drive a CLICK over the wire, twice — the remote app applies each.
for _ in 0..2 {
assert!(matches!(
call!(CmdRequest::Act { action: Action::ClickButton("Bump".into()) }),
CmdResponse::Acted { ok: true, error: None },
));
}
// The wired-back state reflects the two remote clicks (round-trip proof).
match call!(CmdRequest::ReadState) {
CmdResponse::State { state } => {
assert_eq!(state.pointer("/clicks"), Some(&json!(2)),
"two clicks sent over the channel moved the REMOTE counter");
}
other => panic!("want State, got {other:?}"),
}
// Click-by-id over the wire drives the SAME widget (id from Locate).
assert!(matches!(
call!(CmdRequest::Act { action: Action::ClickById(bump_id) }),
CmdResponse::Acted { ok: true, error: None },
));
// Type into a remote field, then read it back through state_json.
assert!(matches!(
call!(CmdRequest::Act {
action: Action::TypeInto { label: "Name".into(), text: "Heimdall".into() }
}),
CmdResponse::Acted { ok: true, .. }
));
match call!(CmdRequest::ReadStateAt { pointer: "/name".into() }) {
CmdResponse::StateAt { value } => assert_eq!(value, Some(json!("Heimdall"))),
other => panic!("want StateAt, got {other:?}"),
}
// A PASSING assertion over the wire.
assert!(matches!(
call!(CmdRequest::Assert {
assert: Assert::State { pointer: "/clicks".into(), expected: json!(3) }
}),
CmdResponse::Asserted { pass: true, reason: None },
));
// RED-WHEN-BROKEN #2: a FAILING assertion must surface as a failed RESULT
// (Asserted { pass: false }), NOT a silent pass or a dropped connection.
match call!(CmdRequest::Assert {
assert: Assert::State { pointer: "/clicks".into(), expected: json!(999) },
}) {
CmdResponse::Asserted { pass: false, reason: Some(r) } => {
assert!(r.contains("999"), "the failure reason travels the wire: {r}");
}
other => panic!("a failed assertion must surface as Asserted{{pass:false}}, got {other:?}"),
}
}
/// The high-level CLIENT mirror API (`click_button` / `read_state` / `assert`)
/// drives the remote host and looks identical to the in-process session. We
/// pump the host between the client's `send` and `recv` via a `Coupled`
/// transport so `roundtrip` completes single-threaded (in production the host's
/// own I/O task answers).
#[test]
fn client_mirror_api_drives_the_remote_host() {
// A transport that, on each read, first lets the host serve the pending
// request — so a blocking `roundtrip` resolves on one thread. The bytes
// still cross a real duplex + full (de)serialize; only the scheduling is
// collapsed.
struct Coupled {
driver: IoChannel<io::BufReader<DuplexEnd>, DuplexEnd>,
host: CmdJsonServer<RobotSession<'static, BumpApp>>,
host_chan: IoChannel<io::BufReader<DuplexEnd>, DuplexEnd>,
}
impl LineChannel for Coupled {
fn write_line(&mut self, line: &str) -> io::Result<()> {
self.driver.write_line(line)
}
fn read_line(&mut self) -> io::Result<Option<String>> {
// Serve the request the driver just wrote, then read the reply.
self.host.serve_one(&mut self.host_chan)?;
self.driver.read_line()
}
}
let (driver_end, host_end) = duplex();
let coupled = Coupled {
driver: channel_over(driver_end),
host: bump_host(),
host_chan: channel_over(host_end),
};
let mut client = CmdJsonClient::new(coupled);
// Reads identically to a local RobotSession — but the app is "remote".
client.ping().expect("host alive");
assert_eq!(client.state_at("/clicks").unwrap(), Some(json!(0)));
client.click_button("Bump").expect("remote click");
client.click_button("Bump").expect("remote click");
assert_eq!(
client.state_at("/clicks").unwrap(),
Some(json!(2)),
"the high-level mirror API drove the remote counter to 2",
);
client.type_into("Name", "Odin").expect("remote type");
assert_eq!(client.state_at("/name").unwrap(), Some(json!("Odin")));
// A whole step (act + assert) driven remotely → a green matrix verdict.
let (pass, msg) = client
.drive_step(
UiStep::new("third bump → clicks == 3")
.action(Action::ClickButton("Bump".into()))
.assert(Assert::State { pointer: "/clicks".into(), expected: json!(3) }),
)
.expect("drive step");
assert!(pass, "remote step passed: {msg}");
// A FAILING assertion via the mirror API surfaces the reason (Ok(Some)),
// not an error and not a silent pass.
let reason = client
.assert(Assert::State { pointer: "/clicks".into(), expected: json!(999) })
.expect("assert round-trips")
.expect("the assertion FAILED, so a reason is returned");
assert!(reason.contains("999"), "mirror API surfaces the failure: {reason}");
}
/// A malformed request line becomes a `CmdResponse::Error` — a bad command is
/// an in-band protocol error, it does NOT drop the connection or panic.
#[test]
fn malformed_request_is_an_error_not_a_drop() {
let (driver_end, host_end) = duplex();
let mut raw = channel_over(driver_end);
let mut host = bump_host();
let mut host_chan = channel_over(host_end);
raw.write_line("{not valid json").expect("write garbage");
assert!(host.serve_one(&mut host_chan).expect("served"), "host stayed up");
match raw.read_line().expect("reply").as_deref() {
Some(line) => {
let resp: CmdResponse = serde_json::from_str(line).expect("decodes");
assert!(matches!(resp, CmdResponse::Error { .. }), "got {resp:?}");
}
None => panic!("host dropped the connection on a bad line"),
}
// And the host is still serving: a valid Ping after the bad line works.
raw.write_line(&serde_json::to_string(&CmdRequest::Ping).unwrap()).unwrap();
assert!(host.serve_one(&mut host_chan).expect("served"));
let line = raw.read_line().expect("reply").expect("a line");
assert!(matches!(
serde_json::from_str::<CmdResponse>(&line).unwrap(),
CmdResponse::Pong
));
}
/// The protocol enums round-trip through JSON unchanged (the wire form is
/// stable serde) — including an [`Action`] carrying `egui` types.
#[test]
fn protocol_json_roundtrips() {
let reqs = vec![
CmdRequest::Ping,
CmdRequest::Locate { target: Target::id(7) },
CmdRequest::Act { action: Action::ClickButton("Bump".into()) },
CmdRequest::Act { action: Action::Key(egui::Key::Enter) },
CmdRequest::Assert {
assert: Assert::State { pointer: "/x".into(), expected: json!(1) },
},
CmdRequest::ReadState,
CmdRequest::ReadStateAt { pointer: "/y".into() },
];
for req in reqs {
let line = serde_json::to_string(&req).unwrap();
let back: CmdRequest = serde_json::from_str(&line).unwrap();
// The wire enums intentionally carry non-PartialEq egui/Value payloads,
// so compare the canonical JSON forms: decode → re-encode must match.
assert_eq!(
serde_json::to_value(&req).unwrap(),
serde_json::to_value(&back).unwrap(),
"request survives the wire form: {line}",
);
}
}
// Silence "unused" for the demo Observable import in some cfgs.
#[allow(dead_code)]
fn _assert_observable(a: &BumpApp) -> Value {
a.state_json()
}
}