mirror of
https://github.com/torlando-tech/pyxis.git
synced 2026-08-22 10:39:52 +00:00
test: on-device LXMF e2e harness with bz2-on-receive probe
tests/hardware/ drives a flashed T-Deck through LXMF round-trips against a Mac-side echo bot over the device's TCP-client -> rnsd link, via the firmware's -DPYXIS_TEST_HOOKS T: command surface. - run_e2e.sh: orchestrator -- autodetect serial port + Mac IP, verify rnsd on :4242, build+flash with the TCP target baked in, run the harness, assert. - tdeck_harness.py: resets the device, establishes path/identity with the bot, then runs a bz2-on-receive probe + DIRECT/OPPORTUNISTIC (and PROPAGATED when PYXIS_PROPAGATION_NODE_HEX is set) round-trips, watching for crash signatures. - lxmf_echo_bot.py: Mac-side LXMF echo bot; on a BZ2PROBE trigger it replies with a ~1.5KB highly-compressible payload so python LXMF sends a bz2-COMPRESSED Resource, exercising the receiver's decompress-on-receive path. Parametrized -- serial port, Mac IP, and propagation-node hash all come from the environment; nothing deployment-specific is committed. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UWZuYkHBRqNb6BZHV8sTG5
This commit is contained in:
torlando-agent[bot]
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
70d4aa6be9
commit
092fbedc44
@@ -0,0 +1,331 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Mac-side LXMF echo bot for diagnosing pyxis T-Deck delivery issues.
|
||||
|
||||
Joins the same RNS network as the T-Deck via AutoInterface (link-local
|
||||
IPv6 multicast). Announces an identity called "Mac Echo Bot" and echoes
|
||||
any received DIRECT message back to the sender. Every announce send,
|
||||
announce receive, message receive, and message send is logged with a
|
||||
timestamp.
|
||||
|
||||
Usage:
|
||||
pip install rns lxmf
|
||||
python3 /tmp/lxmf_echobot.py
|
||||
|
||||
Expected output when working:
|
||||
[12:00:00.123] Reticulum up. Identity hash=<hex>. Delivery dest hash=<hex>.
|
||||
[12:00:00.234] Sending announce
|
||||
[12:00:01.456] Received announce from <hex8>: Some Peer
|
||||
[12:00:05.789] Received DIRECT message from <hex8>: 'hello'
|
||||
[12:00:05.890] Echoing back: 'echo: hello'
|
||||
[12:00:06.012] Echo SENT (state=delivered)
|
||||
"""
|
||||
import os, sys, time, threading, datetime
|
||||
|
||||
# Pin to a local checkout of RNS + LXMF if one exists, so the bot
|
||||
# uses the same library pyxis interops with rather than whatever
|
||||
# pip points at. Override with RNS_REPO / LXMF_REPO env vars.
|
||||
for env_var, default_subdir in (("RNS_REPO", "repos/Reticulum"),
|
||||
("LXMF_REPO", "repos/LXMF")):
|
||||
p = os.environ.get(env_var) \
|
||||
or os.path.expanduser(os.path.join("~", default_subdir))
|
||||
if os.path.isdir(p):
|
||||
sys.path.insert(0, p)
|
||||
|
||||
import RNS
|
||||
import LXMF
|
||||
|
||||
|
||||
def ts():
|
||||
return datetime.datetime.now().strftime("%H:%M:%S.%f")[:-3]
|
||||
|
||||
|
||||
def log(msg):
|
||||
print(f"[{ts()}] {msg}", flush=True)
|
||||
|
||||
|
||||
CONFIG_DIR = "/tmp/echobot-rnsconfig"
|
||||
STORAGE_DIR = "/tmp/echobot-storage"
|
||||
DISPLAY_NAME = "Mac Echo Bot"
|
||||
|
||||
# Optional: when the harness launches the bot, it sets ECHOBOT_PEER_HEX
|
||||
# to pyxis's delivery destination hash. We use that to proactively
|
||||
# request a path / cached-announce so we have pyxis's identity in
|
||||
# `Identity.recall` cache before the harness drives messages — otherwise
|
||||
# we depend on rnsd's announce rebroadcast and miss it if rnsd's
|
||||
# per-announce rebroadcast limit was already reached.
|
||||
PEER_HEX = os.environ.get("ECHOBOT_PEER_HEX", "").strip()
|
||||
|
||||
# lxmd propagation node the bot routes PROPAGATED messages through.
|
||||
# Deployment-specific, so it comes from PYXIS_PROPAGATION_NODE_HEX (no
|
||||
# hardcoded default — keeps environment-specific hashes out of source).
|
||||
# When unset, the bot skips all propagation setup and the harness skips
|
||||
# its PROPAGATED rounds; DIRECT + OPPORTUNISTIC + the bz2 probe still run.
|
||||
PROPAGATION_NODE_HEX = os.environ.get("PYXIS_PROPAGATION_NODE_HEX", "").strip()
|
||||
PROPAGATION_STAMP_COST = 16 # lxmd default is 16; 13 is the floor.
|
||||
|
||||
# How often to pull queued messages from the PN. The bot's not running
|
||||
# any UI, so it relies on this loop to actually receive PROPAGATED
|
||||
# messages.
|
||||
PROP_SYNC_INTERVAL_SEC = 8
|
||||
|
||||
os.makedirs(CONFIG_DIR, exist_ok=True)
|
||||
os.makedirs(STORAGE_DIR, exist_ok=True)
|
||||
|
||||
# Minimal RNS config: TCPClient to the local rnsd at 127.0.0.1:4242.
|
||||
# Pyxis on the T-Deck connects as a TCP CLIENT to the same rnsd's
|
||||
# TCPServerInterface (built into rnsd's default config) so this bot
|
||||
# and pyxis end up on the same Reticulum network with the rnsd acting
|
||||
# as a hub.
|
||||
#
|
||||
# We use an isolated config dir / non-default control ports so this
|
||||
# bot does NOT accidentally join the user's primary rnsd shared
|
||||
# instance (which would mix the bot's identity into their personal
|
||||
# Reticulum state). The bot is its own RNS process.
|
||||
config_path = os.path.join(CONFIG_DIR, "config")
|
||||
with open(config_path, "w") as f:
|
||||
f.write("""\
|
||||
[reticulum]
|
||||
enable_transport = Yes
|
||||
share_instance = No
|
||||
shared_instance_port = 47428
|
||||
instance_control_port = 47429
|
||||
|
||||
[logging]
|
||||
loglevel = 4
|
||||
|
||||
[interfaces]
|
||||
[[TCP to local rnsd]]
|
||||
type = TCPClientInterface
|
||||
enabled = yes
|
||||
target_host = 127.0.0.1
|
||||
target_port = 4242
|
||||
""")
|
||||
|
||||
reticulum = RNS.Reticulum(configdir=CONFIG_DIR, loglevel=4)
|
||||
|
||||
# Persistent identity stored in the config dir so reruns reuse the
|
||||
# same delivery destination hash (peers don't have to re-learn the
|
||||
# path on every restart).
|
||||
ident_path = os.path.join(CONFIG_DIR, "identity")
|
||||
if os.path.exists(ident_path):
|
||||
identity = RNS.Identity.from_file(ident_path)
|
||||
log(f"Loaded identity from {ident_path}")
|
||||
else:
|
||||
identity = RNS.Identity()
|
||||
identity.to_file(ident_path)
|
||||
log(f"Created new identity, saved to {ident_path}")
|
||||
|
||||
router = LXMF.LXMRouter(identity=identity, storagepath=STORAGE_DIR)
|
||||
delivery_destination = router.register_delivery_identity(
|
||||
identity, display_name=DISPLAY_NAME
|
||||
)
|
||||
delivery_destination.set_default_app_data(
|
||||
lambda: router.get_announce_app_data(delivery_destination.hash)
|
||||
)
|
||||
|
||||
log(f"Reticulum up. Identity hash={identity.hash.hex()}.")
|
||||
log(f"Delivery dest hash={delivery_destination.hash.hex()}.")
|
||||
log(f"Display name: {DISPLAY_NAME}")
|
||||
|
||||
# Configure propagation node so the bot can SEND propagated and SYNC
|
||||
# its inbox. We have to wait for the PN to be path-known before
|
||||
# `set_outbound_propagation_node` is useful; the path arrives via
|
||||
# the rnsd<->lxmd shared instance. Set it right away anyway —
|
||||
# request_messages_from_propagation_node tolerates "no path yet"
|
||||
# and retries the path request internally.
|
||||
prop_node_hash = bytes.fromhex(PROPAGATION_NODE_HEX) if PROPAGATION_NODE_HEX else None
|
||||
if prop_node_hash is not None:
|
||||
router.set_outbound_propagation_node(prop_node_hash)
|
||||
router.outbound_propagation_node = prop_node_hash
|
||||
log(f"Configured outbound propagation node: {PROPAGATION_NODE_HEX}")
|
||||
else:
|
||||
log("No PYXIS_PROPAGATION_NODE_HEX set — propagation disabled (DIRECT/OPP/bz2 only)")
|
||||
|
||||
|
||||
def on_announce(destination_hash, announced_identity, app_data):
|
||||
name = LXMF.display_name_from_app_data(app_data) or "(no name)"
|
||||
log(f"Announce RX: dest={destination_hash.hex()} name={name!r}")
|
||||
|
||||
|
||||
announce_handler = type(
|
||||
"Handler", (), {
|
||||
"aspect_filter": "lxmf.delivery",
|
||||
# Without `receive_path_responses = True`, RNS only fires this
|
||||
# handler for LIVE announces (received via broadcast/forward).
|
||||
# Path responses (the cached announce sent back by a next-hop
|
||||
# in reply to RNS.Transport.request_path) are skipped. The
|
||||
# harness's eager-path-acquirer relies on path responses to
|
||||
# learn pyxis's identity quickly, so we opt in here.
|
||||
"receive_path_responses": True,
|
||||
"received_announce": staticmethod(on_announce),
|
||||
}
|
||||
)
|
||||
RNS.Transport.register_announce_handler(announce_handler)
|
||||
|
||||
|
||||
def on_delivery(message):
|
||||
src_hex = message.source_hash.hex() if message.source_hash else "?"
|
||||
method = {
|
||||
LXMF.LXMessage.OPPORTUNISTIC: "OPPORTUNISTIC",
|
||||
LXMF.LXMessage.DIRECT: "DIRECT",
|
||||
LXMF.LXMessage.PROPAGATED: "PROPAGATED",
|
||||
}.get(message.method, str(message.method))
|
||||
try:
|
||||
content = message.content.decode("utf-8")
|
||||
except Exception:
|
||||
content = repr(message.content)
|
||||
title = ""
|
||||
try:
|
||||
if message.title:
|
||||
title = message.title.decode("utf-8")
|
||||
except Exception:
|
||||
title = repr(message.title)
|
||||
log(
|
||||
f"Message RX: from={src_hex} method={method} "
|
||||
f"title={title!r} content={content!r}"
|
||||
)
|
||||
|
||||
# Echo back. Mirror the sender's method so PROPAGATED messages
|
||||
# echo via the PN (round-trip: sender uploads → PN persists → bot
|
||||
# syncs down → bot replies via PN → sender syncs down) and DIRECT
|
||||
# messages echo over a fresh link.
|
||||
try:
|
||||
# Recall the sender's identity so we can construct a destination.
|
||||
sender_identity = RNS.Identity.recall(message.source_hash)
|
||||
if sender_identity is None:
|
||||
log(f" Cannot echo: sender identity not yet known for {src_hex}")
|
||||
return
|
||||
dest = RNS.Destination(
|
||||
sender_identity, RNS.Destination.OUT, RNS.Destination.SINGLE,
|
||||
"lxmf", "delivery"
|
||||
)
|
||||
echo_method = (
|
||||
LXMF.LXMessage.PROPAGATED
|
||||
if message.method == LXMF.LXMessage.PROPAGATED
|
||||
else LXMF.LXMessage.DIRECT
|
||||
)
|
||||
# bz2-on-receive probe: when the harness sends "BZ2PROBE", reply with
|
||||
# a large, highly-compressible payload. It exceeds microLXMF's Resource
|
||||
# threshold AND compresses ~99%, so python LXMF sends it as a bz2-
|
||||
# compressed Resource (auto_compress, since microLXMF announces
|
||||
# compression-capable). That forces pyxis's Resource::assemble() to
|
||||
# bz2_decompress on receive — the path that, unpatched upstream, marks
|
||||
# the resource CORRUPT and tears down the link. If pyxis surfaces the
|
||||
# full payload via T:RX, decompress-on-receive works on hardware.
|
||||
if content.startswith("BZ2PROBE"):
|
||||
echo_body = "BZ2OK:" + ("ABCDABCDABCDABCD" * 96) # ~1542 bytes, repeating
|
||||
else:
|
||||
echo_body = f"echo: {content}"
|
||||
echo_kwargs = dict(
|
||||
destination=dest,
|
||||
source=delivery_destination,
|
||||
content=echo_body.encode("utf-8"),
|
||||
title=b"echo",
|
||||
desired_method=echo_method,
|
||||
)
|
||||
if echo_method == LXMF.LXMessage.PROPAGATED:
|
||||
# python LXMF requires `include_ticket=False` here so the
|
||||
# outbound propagation flow doesn't trip on missing ticket
|
||||
# state. The default is fine but spelled out for clarity.
|
||||
echo_kwargs["include_ticket"] = False
|
||||
echo = LXMF.LXMessage(**echo_kwargs)
|
||||
|
||||
def on_delivered(msg):
|
||||
log(f" Echo DELIVERED to {src_hex}")
|
||||
def on_failed(msg):
|
||||
log(f" Echo FAILED to {src_hex} state={msg.state}")
|
||||
def on_sent(msg):
|
||||
log(f" Echo SENT (PRF received) to {src_hex}")
|
||||
|
||||
echo.register_delivery_callback(on_delivered)
|
||||
echo.register_failed_callback(on_failed)
|
||||
# `sent` only fires on PROPAGATED in upstream LXMF; we still set it
|
||||
# for parity with the `delivery` callback nomenclature.
|
||||
try:
|
||||
echo.register_sent_callback(on_sent)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
router.handle_outbound(echo)
|
||||
log(f" Echo queued for delivery to {src_hex}")
|
||||
except Exception as e:
|
||||
log(f" Echo construction failed: {e}")
|
||||
|
||||
|
||||
router.register_delivery_callback(on_delivery)
|
||||
|
||||
|
||||
# Announce loop — every 30s so the T-Deck hears us multiple times.
|
||||
def announcer():
|
||||
while True:
|
||||
delivery_destination.announce()
|
||||
log("Announce TX")
|
||||
time.sleep(30)
|
||||
|
||||
|
||||
# Propagation sync loop — periodically pull queued messages from the PN.
|
||||
# Without this the bot would never receive PROPAGATED messages (no UI
|
||||
# tap to drive a manual sync; the LXMRouter only auto-syncs when its
|
||||
# `delivery_destination` has someone tapping `sync_inbound`).
|
||||
#
|
||||
# lxmd's default propagation announce_interval is 5 MINUTES (the value
|
||||
# in the config is multiplied by 60), so waiting passively for an
|
||||
# announce can stall the first sync several minutes. We proactively
|
||||
# `RNS.Transport.request_path()` until we get a path.
|
||||
def prop_syncer():
|
||||
while not RNS.Transport.has_path(prop_node_hash):
|
||||
log(" Requesting path to PN...")
|
||||
RNS.Transport.request_path(prop_node_hash)
|
||||
for _ in range(10):
|
||||
if RNS.Transport.has_path(prop_node_hash):
|
||||
break
|
||||
time.sleep(0.5)
|
||||
log(f"Path to PN known; starting periodic sync every {PROP_SYNC_INTERVAL_SEC}s")
|
||||
while True:
|
||||
try:
|
||||
router.request_messages_from_propagation_node(identity)
|
||||
except Exception as e:
|
||||
log(f" prop sync error: {e}")
|
||||
time.sleep(PROP_SYNC_INTERVAL_SEC)
|
||||
|
||||
|
||||
threading.Thread(target=announcer, daemon=True).start()
|
||||
if prop_node_hash is not None:
|
||||
threading.Thread(target=prop_syncer, daemon=True).start()
|
||||
|
||||
|
||||
# Eager peer-path acquisition. When the harness tells us pyxis's hash
|
||||
# via env, request a path immediately. RNS forwards the cached announce
|
||||
# from the next-hop, which fires our announce-handler and populates
|
||||
# Identity.recall_app_data. Without this, if the bot connects AFTER
|
||||
# rnsd hit its rebroadcast limit for the peer's announce, we never
|
||||
# learn pyxis's identity and can't echo PROPAGATED messages.
|
||||
if PEER_HEX:
|
||||
def peer_path_acquirer():
|
||||
try:
|
||||
peer_hash = bytes.fromhex(PEER_HEX)
|
||||
except Exception as e:
|
||||
log(f" Bad ECHOBOT_PEER_HEX={PEER_HEX!r}: {e}")
|
||||
return
|
||||
for _ in range(10):
|
||||
if RNS.Transport.has_path(peer_hash):
|
||||
log(f"Path to peer {PEER_HEX} known via cached announce")
|
||||
return
|
||||
log(f" Requesting path to peer {PEER_HEX}...")
|
||||
RNS.Transport.request_path(peer_hash)
|
||||
for _ in range(20):
|
||||
if RNS.Transport.has_path(peer_hash):
|
||||
log(f"Path to peer {PEER_HEX} known via cached announce")
|
||||
return
|
||||
time.sleep(0.5)
|
||||
log(f" WARN: never got path to peer {PEER_HEX}")
|
||||
|
||||
threading.Thread(target=peer_path_acquirer, daemon=True).start()
|
||||
log(f"Echo bot ready. Press Ctrl-C to stop. Storage: {STORAGE_DIR}")
|
||||
|
||||
try:
|
||||
while True:
|
||||
time.sleep(60)
|
||||
except KeyboardInterrupt:
|
||||
log("Shutting down.")
|
||||
Executable
+68
@@ -0,0 +1,68 @@
|
||||
#!/usr/bin/env bash
|
||||
#
|
||||
# Full end-to-end on-device test for pyxis: build + flash a T-Deck, bring up a
|
||||
# Mac-side LXMF echo bot over the local rnsd, and drive LXMF round-trips
|
||||
# (DIRECT / OPPORTUNISTIC / optional PROPAGATED) plus a bz2-on-receive probe
|
||||
# via the firmware's `T:` serial command surface (-DPYXIS_TEST_HOOKS).
|
||||
#
|
||||
# All environment-specific values are read from the environment (nothing
|
||||
# deployment-specific is committed):
|
||||
# PYXIS_TEST_TCP_HOST Mac IP the T-Deck dials for rnsd (default: en0 IPv4)
|
||||
# PYXIS_TEST_TCP_PORT rnsd TCPServerInterface port (default: 4242)
|
||||
# PYXIS_SERIAL_PORT T-Deck USB serial (default: first /dev/cu.usbmodem*)
|
||||
# PYXIS_ENV platformio env (default: tdeck)
|
||||
# PYXIS_PROPAGATION_NODE_HEX lxmd PN hash (optional; enables PROPAGATED rounds)
|
||||
#
|
||||
# Usage:
|
||||
# tests/hardware/run_e2e.sh # build, flash, one pass + bz2 probe
|
||||
# tests/hardware/run_e2e.sh --soak-hours 1 # ... then soak round-trips for 1h
|
||||
# PYXIS_SKIP_FLASH=1 tests/hardware/run_e2e.sh # reuse already-flashed fw
|
||||
#
|
||||
# Exits 0 only if every round (incl. the bz2 probe) passed and no crash fired.
|
||||
set -euo pipefail
|
||||
|
||||
HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
REPO="$(cd "$HERE/../.." && pwd)"
|
||||
SOAK_HOURS="0"
|
||||
[ "${1:-}" = "--soak-hours" ] && SOAK_HOURS="${2:-0}"
|
||||
|
||||
PYXIS_ENV="${PYXIS_ENV:-tdeck}"
|
||||
PYXIS_TEST_TCP_PORT="${PYXIS_TEST_TCP_PORT:-4242}"
|
||||
PYXIS_TEST_TCP_HOST="${PYXIS_TEST_TCP_HOST:-$(ipconfig getifaddr en0 2>/dev/null || true)}"
|
||||
PYXIS_SERIAL_PORT="${PYXIS_SERIAL_PORT:-$(ls /dev/cu.usbmodem* 2>/dev/null | head -1 || true)}"
|
||||
PIO="$(command -v pio || echo /opt/homebrew/bin/pio)"
|
||||
# The harness needs pyserial; PlatformIO's bundled python has it. Prefer an
|
||||
# explicit PYXIS_HARNESS_PY, else the penv python, else any python with serial.
|
||||
PIO_PY="${PYXIS_HARNESS_PY:-}"
|
||||
[ -z "$PIO_PY" ] && PIO_PY="$(ls "$HOME"/.platformio/penv/bin/python3 2>/dev/null | head -1 || true)"
|
||||
[ -z "$PIO_PY" ] && for c in /opt/homebrew/Cellar/platformio/*/libexec/bin/python3; do [ -x "$c" ] && PIO_PY="$c" && break; done
|
||||
|
||||
echo "== pyxis on-device e2e =="
|
||||
echo " repo: $REPO"
|
||||
echo " env: $PYXIS_ENV"
|
||||
echo " serial: ${PYXIS_SERIAL_PORT:-<none found>}"
|
||||
echo " rnsd target: ${PYXIS_TEST_TCP_HOST:-<unset>}:$PYXIS_TEST_TCP_PORT"
|
||||
echo " prop node: ${PYXIS_PROPAGATION_NODE_HEX:-<unset — PROPAGATED skipped>}"
|
||||
echo " soak hours: $SOAK_HOURS"
|
||||
echo " harness py: ${PIO_PY:-<not found>}"
|
||||
|
||||
[ -z "${PYXIS_SERIAL_PORT:-}" ] && { echo "ERROR: no T-Deck serial port found (set PYXIS_SERIAL_PORT)"; exit 2; }
|
||||
[ -z "${PYXIS_TEST_TCP_HOST:-}" ] && { echo "ERROR: could not determine Mac IP (set PYXIS_TEST_TCP_HOST)"; exit 2; }
|
||||
|
||||
# rnsd must be listening so both the T-Deck (LAN) and the bot (127.0.0.1) can
|
||||
# attach to the same Reticulum hub.
|
||||
if ! lsof -nP -iTCP:"$PYXIS_TEST_TCP_PORT" -sTCP:LISTEN >/dev/null 2>&1; then
|
||||
echo "ERROR: nothing listening on TCP:$PYXIS_TEST_TCP_PORT — start rnsd with a"
|
||||
echo " TCPServerInterface on that port first (e.g. \`rnsd\`)."
|
||||
exit 2
|
||||
fi
|
||||
|
||||
export PYXIS_TEST_TCP_HOST PYXIS_TEST_TCP_PORT PYXIS_SERIAL_PORT PYXIS_PROPAGATION_NODE_HEX
|
||||
|
||||
if [ "${PYXIS_SKIP_FLASH:-0}" != "1" ]; then
|
||||
echo "== build + flash ($PYXIS_ENV) with TCP target baked in =="
|
||||
"$PIO" run -e "$PYXIS_ENV" -t upload --upload-port "$PYXIS_SERIAL_PORT"
|
||||
fi
|
||||
|
||||
echo "== running harness (resets device, drives T: commands) =="
|
||||
exec "$PIO_PY" "$HERE/tdeck_harness.py" --soak-hours "$SOAK_HOURS"
|
||||
@@ -0,0 +1,645 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
T-Deck ↔ Mac echo-bot integration harness.
|
||||
|
||||
This script drives a pyxis-built T-Deck firmware (with -DPYXIS_TEST_HOOKS)
|
||||
and a Mac-side LXMF echo bot together, exercising LXMF DIRECT delivery
|
||||
in both directions over the local rnsd's TCPServerInterface
|
||||
(host:port configured at firmware-build-time via PYXIS_TEST_TCP_HOST /
|
||||
PYXIS_TEST_TCP_PORT in platformio.ini).
|
||||
|
||||
What it does:
|
||||
|
||||
1. Open the T-Deck serial port (/dev/cu.usbmodem1101) and reset the device.
|
||||
Stream every line into a captured log; parse `T:OK` / `T:ERR` /
|
||||
`T:RXMSG` / `T:PATH` lines for the harness, pass everything else
|
||||
through to the log.
|
||||
|
||||
2. Wait for pyxis boot: `BOOT END: ui_manager` line, then a TCP
|
||||
connect-success log to confirm the rnsd link is up.
|
||||
|
||||
3. Start the echo bot AFTER pyxis is fully up (subprocess) so its
|
||||
announce arrives while pyxis is listening. Bot uses the same rnsd
|
||||
the T-Deck talks to (shared instance via shared_instance_port).
|
||||
|
||||
4. Wait for pyxis to learn the bot's path (`T:HASPATH <bot_dest>`
|
||||
returns `T:OK 1`), with bot manually re-announcing if needed.
|
||||
|
||||
5. Test loop, repeating until either a failure or the soak deadline:
|
||||
- Send short message from pyxis to bot via `T:SEND`.
|
||||
- Wait for bot to log RX + ECHO SENT.
|
||||
- Wait for pyxis to log RX (poll `T:RX`).
|
||||
- Verify content matches `echo: <orig>`.
|
||||
- Send long (~1KB) message and repeat.
|
||||
- Sleep `--cadence` seconds between rounds.
|
||||
|
||||
6. Soak: run for at least `--soak-hours` hours (default 1.1). Crash =
|
||||
fail. Mid-test failures are logged but don't abort.
|
||||
|
||||
Usage:
|
||||
# Run with a python that has pyserial available. PlatformIO's
|
||||
# bundled python works:
|
||||
"$(pio system info | awk -F: '/Python Executable/{print $2}' | xargs)" \\
|
||||
tests/soak/tdeck_soak_harness.py [--soak-hours 1.1] [--cadence 30]
|
||||
|
||||
Outputs:
|
||||
/tmp/tdeck-harness.log — combined timestamped event log
|
||||
/tmp/tdeck-harness-tdeck.log — verbatim serial output from T-Deck
|
||||
/tmp/echobot.log — echo bot's own log (subprocess stdout)
|
||||
|
||||
Exits 0 on full success, 1 on failure.
|
||||
"""
|
||||
import os, sys, time, threading, subprocess, argparse, queue, signal
|
||||
import datetime
|
||||
|
||||
# pyserial: try the active python first, fall back to PlatformIO's
|
||||
# bundled site-packages if the user runs us through system python.
|
||||
try:
|
||||
import serial # noqa: F401
|
||||
except ImportError:
|
||||
pio_site = os.environ.get("PIO_SITE_PACKAGES")
|
||||
if pio_site and os.path.isdir(pio_site):
|
||||
sys.path.insert(0, pio_site)
|
||||
import serial # second try; let it raise if still missing
|
||||
|
||||
# Default serial port for a USB-attached T-Deck Plus on macOS. Override
|
||||
# with PYXIS_SERIAL_PORT (eg "/dev/ttyUSB0" on Linux).
|
||||
PORT = os.environ.get("PYXIS_SERIAL_PORT", "/dev/cu.usbmodem1101")
|
||||
BAUD = 115200
|
||||
HARNESS_LOG = "/tmp/tdeck-harness.log"
|
||||
TDECK_LOG = "/tmp/tdeck-harness-tdeck.log"
|
||||
ECHOBOT_LOG = "/tmp/echobot.log"
|
||||
ECHOBOT_PY = os.path.join(os.path.dirname(os.path.abspath(__file__)),
|
||||
"lxmf_echo_bot.py")
|
||||
|
||||
|
||||
def ts():
|
||||
return datetime.datetime.now().strftime("%H:%M:%S.%f")[:-3]
|
||||
|
||||
|
||||
_log_lock = threading.Lock()
|
||||
_log_fh = None
|
||||
|
||||
|
||||
def log(category, msg):
|
||||
line = f"[{ts()}] [{category}] {msg}\n"
|
||||
with _log_lock:
|
||||
sys.stdout.write(line)
|
||||
sys.stdout.flush()
|
||||
if _log_fh is not None:
|
||||
_log_fh.write(line)
|
||||
_log_fh.flush()
|
||||
|
||||
|
||||
class TDeck:
|
||||
"""Serial driver: reset, command, parse response."""
|
||||
|
||||
def __init__(self, port=PORT, baud=BAUD):
|
||||
self.ser = serial.Serial(port, baud, timeout=0.1)
|
||||
self._line_q = queue.Queue() # raw text lines
|
||||
self._tdeck_log = open(TDECK_LOG, "wb")
|
||||
self._stop = threading.Event()
|
||||
self._reader = threading.Thread(target=self._read_loop, daemon=True)
|
||||
self._reader.start()
|
||||
|
||||
def reset(self):
|
||||
"""Pulse DTR/RTS like esptool to reboot the ESP32 cleanly."""
|
||||
log("HARNESS", "Resetting T-Deck via DTR/RTS pulse")
|
||||
self.ser.dtr = False
|
||||
self.ser.rts = True
|
||||
time.sleep(0.1)
|
||||
self.ser.dtr = True
|
||||
self.ser.rts = False
|
||||
time.sleep(0.1)
|
||||
self.ser.dtr = False
|
||||
self.ser.rts = False
|
||||
|
||||
def _read_loop(self):
|
||||
buf = b""
|
||||
while not self._stop.is_set():
|
||||
try:
|
||||
d = self.ser.read(4096)
|
||||
except Exception as e:
|
||||
log("HARNESS", f"Serial read error: {e}")
|
||||
return
|
||||
if not d:
|
||||
continue
|
||||
self._tdeck_log.write(d)
|
||||
self._tdeck_log.flush()
|
||||
buf += d
|
||||
while b"\n" in buf:
|
||||
line, buf = buf.split(b"\n", 1)
|
||||
try:
|
||||
text = line.rstrip(b"\r").decode("utf-8", errors="replace")
|
||||
except Exception:
|
||||
text = repr(line)
|
||||
self._line_q.put(text)
|
||||
|
||||
def drain_lines(self):
|
||||
"""Pop all currently-buffered lines (non-blocking)."""
|
||||
out = []
|
||||
while True:
|
||||
try:
|
||||
out.append(self._line_q.get_nowait())
|
||||
except queue.Empty:
|
||||
break
|
||||
return out
|
||||
|
||||
def wait_for_line(self, predicate, timeout):
|
||||
"""Block until a line matches predicate(text) or timeout."""
|
||||
deadline = time.time() + timeout
|
||||
while time.time() < deadline:
|
||||
try:
|
||||
line = self._line_q.get(timeout=min(1.0, deadline - time.time()))
|
||||
except queue.Empty:
|
||||
continue
|
||||
if predicate(line):
|
||||
return line
|
||||
return None
|
||||
|
||||
def send_command(self, cmd, response_timeout=10.0):
|
||||
"""Send a `T:` command, wait for the next `T:OK` or `T:ERR` line.
|
||||
|
||||
Returns the response line (without the `T:OK` / `T:ERR` prefix
|
||||
portion left intact for inspection), or None on timeout.
|
||||
"""
|
||||
log("HARNESS-TX", cmd)
|
||||
# Drain any stale T: responses queued up from earlier
|
||||
deadline = time.time() + response_timeout
|
||||
# Send
|
||||
self.ser.write((cmd + "\n").encode("utf-8"))
|
||||
self.ser.flush()
|
||||
# Read until we see a T:OK or T:ERR
|
||||
while time.time() < deadline:
|
||||
try:
|
||||
line = self._line_q.get(timeout=min(1.0, deadline - time.time()))
|
||||
except queue.Empty:
|
||||
continue
|
||||
if line.startswith("T:OK") or line.startswith("T:ERR"):
|
||||
log("HARNESS-RX", line)
|
||||
return line
|
||||
log("HARNESS", f"send_command({cmd!r}) timed out")
|
||||
return None
|
||||
|
||||
def close(self):
|
||||
self._stop.set()
|
||||
self.ser.close()
|
||||
self._tdeck_log.close()
|
||||
|
||||
|
||||
def wait_for_pyxis_boot(t, timeout=90):
|
||||
log("HARNESS", "Waiting for pyxis boot to complete...")
|
||||
line = t.wait_for_line(
|
||||
lambda L: "BOOT" in L and "ui_manager" in L and "END" in L,
|
||||
timeout,
|
||||
)
|
||||
if not line:
|
||||
return False
|
||||
log("HARNESS", f"Boot complete: {line}")
|
||||
return True
|
||||
|
||||
|
||||
def wait_for_tcp_link(t, timeout=60):
|
||||
log("HARNESS", "Waiting for TCP interface to connect to rnsd...")
|
||||
line = t.wait_for_line(
|
||||
lambda L: "TCPClientInterface" in L and (
|
||||
"connected" in L.lower() or "Connected" in L or "started" in L.lower()
|
||||
),
|
||||
timeout,
|
||||
)
|
||||
if line:
|
||||
log("HARNESS", f"TCP link up: {line}")
|
||||
else:
|
||||
log("HARNESS", "(TCP-link line not seen — continuing anyway, "
|
||||
"pyxis may have already attached.)")
|
||||
return line is not None
|
||||
|
||||
|
||||
def start_echobot(peer_hex=""):
|
||||
log("HARNESS", f"Launching echo bot (peer_hex={peer_hex})...")
|
||||
fh = open(ECHOBOT_LOG, "w")
|
||||
env = dict(os.environ)
|
||||
if peer_hex:
|
||||
# The bot uses this to proactively request_path() so it picks up
|
||||
# pyxis's cached announce from rnsd, even if rnsd's announce
|
||||
# rebroadcast limit was already reached when the bot connected.
|
||||
env["ECHOBOT_PEER_HEX"] = peer_hex
|
||||
proc = subprocess.Popen(
|
||||
["/usr/bin/python3", ECHOBOT_PY],
|
||||
stdout=fh, stderr=subprocess.STDOUT,
|
||||
env=env,
|
||||
)
|
||||
return proc, fh
|
||||
|
||||
|
||||
def echobot_announce_dest(echobot_log_path, timeout=20):
|
||||
"""Read the echobot's own log to extract its delivery destination hash."""
|
||||
deadline = time.time() + timeout
|
||||
last = ""
|
||||
while time.time() < deadline:
|
||||
try:
|
||||
with open(echobot_log_path) as f:
|
||||
last = f.read()
|
||||
except FileNotFoundError:
|
||||
time.sleep(0.5)
|
||||
continue
|
||||
for line in last.splitlines():
|
||||
# `Delivery dest hash=<hex>.`
|
||||
if "Delivery dest hash=" in line:
|
||||
hex_part = line.split("Delivery dest hash=", 1)[1].rstrip(".").strip()
|
||||
# may have trailing punctuation
|
||||
hex_part = hex_part.split()[0].rstrip(".")
|
||||
return hex_part
|
||||
time.sleep(0.5)
|
||||
return None
|
||||
|
||||
|
||||
def wait_for_path(t, dest_hex, timeout=120):
|
||||
log("HARNESS", f"Polling pyxis for path to bot {dest_hex}...")
|
||||
deadline = time.time() + timeout
|
||||
last_paths_dump = 0
|
||||
while time.time() < deadline:
|
||||
resp = t.send_command(f"T:HASPATH {dest_hex}", response_timeout=5)
|
||||
if resp and resp.startswith("T:OK 1"):
|
||||
log("HARNESS", "Pyxis has path to bot.")
|
||||
return True
|
||||
# Every 20s, dump T:PATHS to see what pyxis DOES know about
|
||||
if time.time() - last_paths_dump > 20:
|
||||
last_paths_dump = time.time()
|
||||
t.ser.write(b"T:PATHS\n")
|
||||
t.ser.flush()
|
||||
ok = t.wait_for_line(lambda L: L.startswith("T:OK count="), timeout=5)
|
||||
if ok:
|
||||
log("HARNESS", f"T:PATHS {ok}")
|
||||
# Drain the T:PATH lines that follow
|
||||
try:
|
||||
count = int(ok.split("=", 1)[1])
|
||||
except Exception:
|
||||
count = 0
|
||||
for _ in range(count):
|
||||
p = t.wait_for_line(lambda L: L.startswith("T:PATH "), timeout=2)
|
||||
if p:
|
||||
log("HARNESS", f" {p}")
|
||||
time.sleep(2.0)
|
||||
return False
|
||||
|
||||
|
||||
def echobot_log_after(marker, predicate, timeout=30):
|
||||
"""Wait until the echobot log has a line matching predicate appearing
|
||||
after `marker` (a substring known to be present already, used to
|
||||
avoid matching old log entries from a previous round)."""
|
||||
deadline = time.time() + timeout
|
||||
last_size = 0
|
||||
while time.time() < deadline:
|
||||
try:
|
||||
with open(ECHOBOT_LOG) as f:
|
||||
txt = f.read()
|
||||
except FileNotFoundError:
|
||||
time.sleep(0.5)
|
||||
continue
|
||||
if marker:
|
||||
mi = txt.find(marker)
|
||||
if mi < 0:
|
||||
time.sleep(0.5)
|
||||
continue
|
||||
txt = txt[mi + len(marker):]
|
||||
for line in txt.splitlines():
|
||||
if predicate(line):
|
||||
return line
|
||||
time.sleep(0.5)
|
||||
return None
|
||||
|
||||
|
||||
def poll_tdeck_rx(t, expected_substring, timeout=60):
|
||||
"""Poll T:RX until a received message contains expected_substring."""
|
||||
deadline = time.time() + timeout
|
||||
while time.time() < deadline:
|
||||
# T:RX returns multiple lines; we have to read more than one
|
||||
log("HARNESS-TX", "T:RX")
|
||||
t.ser.write(b"T:RX\n")
|
||||
t.ser.flush()
|
||||
# Collect: T:OK count=N then N T:RXMSG lines
|
||||
ok_line = t.wait_for_line(
|
||||
lambda L: L.startswith("T:OK count=") or L.startswith("T:ERR"),
|
||||
timeout=5,
|
||||
)
|
||||
if not ok_line:
|
||||
time.sleep(2.0)
|
||||
continue
|
||||
try:
|
||||
count = int(ok_line.split("=", 1)[1])
|
||||
except Exception:
|
||||
count = 0
|
||||
rx_msgs = []
|
||||
for _ in range(count):
|
||||
line = t.wait_for_line(lambda L: L.startswith("T:RXMSG"), timeout=2)
|
||||
if line:
|
||||
rx_msgs.append(line)
|
||||
for line in rx_msgs:
|
||||
if expected_substring in line:
|
||||
log("HARNESS", f"Pyxis RX confirmed: {line}")
|
||||
return line
|
||||
time.sleep(2.0)
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
global _log_fh
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--soak-hours", type=float, default=0.0,
|
||||
help="0 = one pass of each method + the bz2 probe (default); "
|
||||
">0 loops the round-trip tests for a soak")
|
||||
parser.add_argument("--cadence", type=float, default=30,
|
||||
help="seconds between message rounds")
|
||||
parser.add_argument("--no-reset", action="store_true",
|
||||
help="don't pulse DTR (use if pyxis is already booted)")
|
||||
args = parser.parse_args()
|
||||
|
||||
_log_fh = open(HARNESS_LOG, "w")
|
||||
log("HARNESS", f"Harness starting; soak target = {args.soak_hours} hours")
|
||||
|
||||
# 1. Open T-Deck serial, reset, wait for boot
|
||||
t = TDeck()
|
||||
if not args.no_reset:
|
||||
t.reset()
|
||||
|
||||
if not wait_for_pyxis_boot(t, timeout=90):
|
||||
log("HARNESS", "FAILED: pyxis boot did not complete in 90s")
|
||||
return 1
|
||||
|
||||
wait_for_tcp_link(t, timeout=30)
|
||||
time.sleep(2.0)
|
||||
|
||||
pyxis_dest_resp = t.send_command("T:DEST", response_timeout=5)
|
||||
if not pyxis_dest_resp or not pyxis_dest_resp.startswith("T:OK"):
|
||||
log("HARNESS", f"FAILED: T:DEST returned {pyxis_dest_resp}")
|
||||
return 1
|
||||
pyxis_dest = pyxis_dest_resp.split(" ", 1)[1].strip()
|
||||
log("HARNESS", f"Pyxis delivery dest = {pyxis_dest}")
|
||||
|
||||
# 2. Start echo bot AFTER pyxis is up. Pass pyxis's destination
|
||||
# hash so the bot can proactively request_path() and pick up
|
||||
# the cached announce from rnsd.
|
||||
bot, bot_fh = start_echobot(peer_hex=pyxis_dest)
|
||||
bot_dest = echobot_announce_dest(ECHOBOT_LOG, timeout=20)
|
||||
if not bot_dest:
|
||||
log("HARNESS", "FAILED: could not extract bot destination hash")
|
||||
bot.terminate()
|
||||
return 1
|
||||
log("HARNESS", f"Echo bot delivery dest = {bot_dest}")
|
||||
|
||||
# 3. Wait for pyxis to learn the bot's path
|
||||
if not wait_for_path(t, bot_dest, timeout=120):
|
||||
log("HARNESS", "FAILED: pyxis never learned bot's path")
|
||||
bot.terminate()
|
||||
return 1
|
||||
|
||||
# 3b. Drive pyxis to announce so the bot learns pyxis's identity
|
||||
# for echo construction. Two ways the bot can end up with the
|
||||
# identity: (a) live announce delivered through rnsd while the bot
|
||||
# is connected — fires our announce_handler and we log "Announce
|
||||
# RX" — or (b) a path-response from rnsd's cache when the bot
|
||||
# calls RNS.Transport.request_path (which the bot does eagerly
|
||||
# for ECHOBOT_PEER_HEX). Path (b) populates Identity.recall but
|
||||
# doesn't always fire the announce_handler (path-response
|
||||
# delivery is gated on `receive_path_responses` and hits a thread
|
||||
# boundary). So treat "bot has pyxis path" OR "bot logged Announce
|
||||
# RX" as either-or success. Drive a few T:ANN cycles to give path
|
||||
# (a) a chance and not just rely on the eager request_path.
|
||||
log("HARNESS", "Driving pyxis to announce, "
|
||||
"waiting for bot to learn pyxis's identity...")
|
||||
bot_saw_pyxis = False
|
||||
for attempt in range(6):
|
||||
t.send_command("T:ANN", response_timeout=5)
|
||||
# Either: bot's announce_handler fired (live announce) OR
|
||||
# bot's eager request_path landed (path-response → cached
|
||||
# announce → Identity.remember). The latter logs "Path to
|
||||
# peer <hex> known via cached announce" via the
|
||||
# peer_path_acquirer thread.
|
||||
deadline = time.time() + 10
|
||||
while time.time() < deadline:
|
||||
try:
|
||||
with open(ECHOBOT_LOG) as f:
|
||||
txt = f.read()
|
||||
except FileNotFoundError:
|
||||
txt = ""
|
||||
if (("Announce RX:" in txt and pyxis_dest in txt)
|
||||
or f"Path to peer {pyxis_dest} known" in txt):
|
||||
bot_saw_pyxis = True
|
||||
break
|
||||
time.sleep(0.5)
|
||||
if bot_saw_pyxis:
|
||||
log("HARNESS", "Bot has pyxis identity (via announce or path-response)")
|
||||
break
|
||||
log("HARNESS", f" attempt {attempt+1}: bot not yet — re-announce")
|
||||
if not bot_saw_pyxis:
|
||||
log("HARNESS", "FAILED: bot never learned pyxis's identity — "
|
||||
"echoes will fail")
|
||||
bot.terminate()
|
||||
return 1
|
||||
|
||||
fails = 0
|
||||
successes = 0
|
||||
|
||||
# 4. bz2-on-receive probe — the graft's riskiest new code. Send a short
|
||||
# "BZ2PROBE" trigger (fits the USB-CDC single-line limit); the echo bot
|
||||
# replies with a ~1.5KB highly-compressible payload. python LXMF sends
|
||||
# that as a bz2-COMPRESSED Resource (auto_compress — microLXMF announces
|
||||
# compression-capable). pyxis must bz2_decompress it in
|
||||
# Resource::assemble() and surface the full body via T:RX. Unpatched
|
||||
# upstream marks a compressed inbound resource CORRUPT and tears down
|
||||
# the link, so a clean decompressed receive here proves the graft's
|
||||
# decompress-on-receive port works on real hardware.
|
||||
log("HARNESS", "=== bz2-on-receive probe ===")
|
||||
with open(ECHOBOT_LOG) as f:
|
||||
probe_marker = f.read()[-256:]
|
||||
probe_resp = t.send_command(f"T:SEND {bot_dest} BZ2PROBE", response_timeout=10)
|
||||
if probe_resp and probe_resp.startswith("T:OK"):
|
||||
bot_rx = echobot_log_after(
|
||||
probe_marker,
|
||||
lambda L: "Message RX:" in L and "BZ2PROBE" in L,
|
||||
timeout=60,
|
||||
)
|
||||
if not bot_rx:
|
||||
log("HARNESS", "FAIL [bz2-probe]: bot never received BZ2PROBE trigger")
|
||||
fails += 1
|
||||
else:
|
||||
rx = poll_tdeck_rx(t, "BZ2OK:", timeout=90)
|
||||
if rx and len(rx) > 800:
|
||||
log("HARNESS", f"PASS [bz2-probe]: pyxis received + decompressed "
|
||||
f"compressed Resource ({len(rx)}-char T:RXMSG line)")
|
||||
successes += 1
|
||||
else:
|
||||
log("HARNESS", f"FAIL [bz2-probe]: no decompressed payload surfaced "
|
||||
f"(got {rx!r}) — check for link teardown / CORRUPT")
|
||||
fails += 1
|
||||
t.send_command("T:RXCLR", response_timeout=5)
|
||||
else:
|
||||
log("HARNESS", f"FAIL [bz2-probe]: T:SEND BZ2PROBE returned {probe_resp!r}")
|
||||
fails += 1
|
||||
|
||||
# 5. Round-trip tests. DIRECT + OPPORTUNISTIC always run; PROPAGATED runs
|
||||
# only when PYXIS_PROPAGATION_NODE_HEX is set (deployment-specific hash
|
||||
# kept out of source). Runs one pass minimum, then loops for --soak-hours.
|
||||
deadline = time.time() + args.soak_hours * 3600
|
||||
round_n = 0
|
||||
payloads = [
|
||||
("direct-short", "T:SEND", "hi t-deck"),
|
||||
("direct-medium", "T:SEND", "this is a 100-char-ish payload to verify pyxis can frame a 1-packet LXMF message cleanly y"),
|
||||
("opportunistic-short", "T:SENDOPP", "opp t-deck"),
|
||||
("opportunistic-medium", "T:SENDOPP", "opp 100-char payload should fit a single LXMF packet without any link establishment xx"),
|
||||
]
|
||||
|
||||
propagation_node_hex = os.environ.get("PYXIS_PROPAGATION_NODE_HEX", "").strip()
|
||||
if propagation_node_hex:
|
||||
log("HARNESS", f"Configuring propagation node on pyxis: {propagation_node_hex}")
|
||||
set_resp = t.send_command(f"T:SETPROP {propagation_node_hex} 16", response_timeout=5)
|
||||
if not set_resp or not set_resp.startswith("T:OK"):
|
||||
log("HARNESS", f"WARN: T:SETPROP failed: {set_resp!r}")
|
||||
if not wait_for_path(t, propagation_node_hex, timeout=60):
|
||||
log("HARNESS", "WARN: pyxis never learned PN path; propagation rounds will fail")
|
||||
else:
|
||||
log("HARNESS", "Waiting for PN identity to land in recall cache...")
|
||||
identity_deadline = time.time() + 60
|
||||
while time.time() < identity_deadline:
|
||||
r = t.send_command(f"T:RECALL {propagation_node_hex}", response_timeout=5)
|
||||
if r and "size=" in r and int(r.split("size=", 1)[1].split()[0]) > 0:
|
||||
log("HARNESS", f"PN identity known: {r}")
|
||||
break
|
||||
time.sleep(2.0)
|
||||
payloads.append(("propagation-short", "T:SENDPROP", "prop t-deck"))
|
||||
else:
|
||||
log("HARNESS", "No PYXIS_PROPAGATION_NODE_HEX set — skipping PROPAGATED rounds")
|
||||
|
||||
first_pass = True
|
||||
while first_pass or time.time() < deadline:
|
||||
first_pass = False
|
||||
round_n += 1
|
||||
for label, send_cmd, content in payloads:
|
||||
log("HARNESS", f"=== Round {round_n} / {label} ===")
|
||||
# Mark current echobot log so we only look at lines after this point
|
||||
with open(ECHOBOT_LOG) as f:
|
||||
marker_text = f.read()
|
||||
marker = marker_text[-256:] if len(marker_text) > 256 else marker_text
|
||||
|
||||
# 4a. T-Deck → Bot
|
||||
send_resp = t.send_command(f"{send_cmd} {bot_dest} {content}", response_timeout=10)
|
||||
if not send_resp or not send_resp.startswith("T:OK"):
|
||||
log("HARNESS", f"FAIL [{label}]: T:SEND returned {send_resp}")
|
||||
fails += 1
|
||||
continue
|
||||
# Parse the message hash so we can poll state
|
||||
try:
|
||||
msg_hash = send_resp.split("hash=", 1)[1].split()[0]
|
||||
except Exception:
|
||||
msg_hash = None
|
||||
log("HARNESS", f"WARN [{label}]: could not parse msg_hash from {send_resp}")
|
||||
|
||||
# Wait for bot to log RX
|
||||
rx_line = echobot_log_after(
|
||||
marker,
|
||||
lambda L: "Message RX:" in L and "from=" + pyxis_dest in L,
|
||||
timeout=60,
|
||||
)
|
||||
if not rx_line:
|
||||
log("HARNESS", f"FAIL [{label}]: bot never received from pyxis")
|
||||
fails += 1
|
||||
continue
|
||||
log("HARNESS", f"OK [{label}]: bot RX: {rx_line.strip()}")
|
||||
|
||||
# Wait for echo back
|
||||
echo_line = echobot_log_after(
|
||||
marker,
|
||||
lambda L: "Echo queued for delivery" in L
|
||||
or "Echo SENT" in L or "Echo DELIVERED" in L,
|
||||
timeout=20,
|
||||
)
|
||||
if not echo_line:
|
||||
log("HARNESS", f"WARN [{label}]: bot did not log echo send "
|
||||
f"(may still arrive)")
|
||||
|
||||
# For PROPAGATED rounds the bot's echo doesn't go directly
|
||||
# to pyxis — it gets uploaded to the PN first, then pyxis
|
||||
# has to sync down. Wait for the bot's "Echo DELIVERED" (or
|
||||
# "Echo SENT") log line before kicking a sync, then retry
|
||||
# sync a few times because the first sync after bot upload
|
||||
# can race the PN's index update.
|
||||
if send_cmd == "T:SENDPROP":
|
||||
upload_line = echobot_log_after(
|
||||
marker,
|
||||
lambda L: ("Echo DELIVERED" in L
|
||||
or "Echo SENT" in L
|
||||
or "Echo queued" in L),
|
||||
timeout=20,
|
||||
)
|
||||
# Echo "DELIVERED" for PROPAGATED in python LXMF means
|
||||
# uploaded to PN. Even after that the PN index update
|
||||
# can lag a few seconds, so we give it a small grace
|
||||
# period.
|
||||
if upload_line and ("DELIVERED" in upload_line
|
||||
or "SENT" in upload_line):
|
||||
log("HARNESS", " Bot uploaded echo to PN; "
|
||||
"waiting 3s for PN to index it")
|
||||
time.sleep(3)
|
||||
else:
|
||||
log("HARNESS", " WARN: didn't see echo upload; "
|
||||
"syncing anyway and hoping for the best")
|
||||
|
||||
# Try up to 4 sync rounds. Each call to T:SYNCPROP
|
||||
# restarts the FSM if it's idle/complete/failed; we
|
||||
# poll T:SYNCSTATE until PR_COMPLETE (=6) or FAILED (=7).
|
||||
for sync_attempt in range(4):
|
||||
log("HARNESS",
|
||||
f" T:SYNCPROP attempt {sync_attempt + 1}/4")
|
||||
t.send_command("T:SYNCPROP", response_timeout=5)
|
||||
sync_deadline = time.time() + 30
|
||||
while time.time() < sync_deadline:
|
||||
sresp = t.send_command("T:SYNCSTATE",
|
||||
response_timeout=5)
|
||||
if sresp and ("state=6" in sresp
|
||||
or "state=7" in sresp):
|
||||
break
|
||||
time.sleep(2)
|
||||
# Quick peek for the echo before kicking another sync
|
||||
quick_match = poll_tdeck_rx(t, "echo:", timeout=5)
|
||||
if quick_match:
|
||||
break
|
||||
|
||||
# Wait for pyxis to RX the echo
|
||||
rx_timeout = 60 if send_cmd != "T:SENDPROP" else 30
|
||||
rx_match = poll_tdeck_rx(t, "echo:", timeout=rx_timeout)
|
||||
if not rx_match:
|
||||
log("HARNESS", f"FAIL [{label}]: pyxis never received echo")
|
||||
fails += 1
|
||||
continue
|
||||
log("HARNESS", f"OK [{label}]: pyxis RX echo: {rx_match}")
|
||||
|
||||
# Clear the pyxis RX ring so the next round starts clean
|
||||
t.send_command("T:RXCLR", response_timeout=5)
|
||||
|
||||
successes += 1
|
||||
log("HARNESS", f"=== Round {round_n} / {label}: PASS "
|
||||
f"(total: {successes} pass, {fails} fail) ===")
|
||||
|
||||
# Stability check: peek for any panic/abort/assert in the T-Deck log
|
||||
# (drain new lines but don't block on them).
|
||||
for line in t.drain_lines():
|
||||
if any(k in line for k in (
|
||||
"Guru Meditation", "PANIC", "abort", "assertion", "rst:0x"
|
||||
)):
|
||||
log("HARNESS", f"CRASH detected: {line}")
|
||||
fails += 1
|
||||
deadline = 0
|
||||
break
|
||||
|
||||
time.sleep(args.cadence)
|
||||
|
||||
log("HARNESS", f"Soak complete. successes={successes} fails={fails}")
|
||||
bot.terminate()
|
||||
bot_fh.close()
|
||||
t.close()
|
||||
return 0 if fails == 0 and successes > 0 else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user