diff --git a/AGENTS.md b/AGENTS.md index 4b55e30..a47223a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -30,6 +30,7 @@ Core printer interaction is `BambuPrinter` in `bambu_cli/printer.py`. Agents and - Secret-bearing files are tightened to `0600` automatically on POSIX: config.json on load, and the `access_code_file` when `load_access_code()` reads it. Windows relies on NTFS ACLs (see [SECURITY.md](SECURITY.md)). - Network operations support `timeout` and `retries` through `printer.send_command()` and `printer.status()`. - `printer.status()` returns a **complete** state snapshot, never a partial MQTT delta: the printer streams incremental updates on its report topic and only answers `pushall` with the whole state, so report messages are merged and the wait continues until `gcode_state`, `mc_percent`, `bed_temper`, and `nozzle_temper` are all present. If only deltas arrive it raises `PrinterStatusIncomplete` (exit `6`) instead of returning a partial — so `plate --json status` never emits a `printer` object missing `gcode_state`. Pass `require_complete=False` for liveness probes that only need to know MQTT works (this is what `doctor` and `print --dry-run` do). +- A long-lived process can call `printer.hold_mqtt()` so `status` / `send_command` / `get_version` reuse one TLS session. The TUI does this: dashboard and monitor refreshes must not open a new connection every few seconds. One-shot CLI commands still connect and tear down. Always pair with `release_mqtt()` so the paho loop thread does not leak (CI runs with `-W error::ResourceWarning`). ### Module layout diff --git a/CHANGELOG.md b/CHANGELOG.md index 1b0bab2..7bc323d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version ### Changed +- The TUI dashboard and monitor keep one MQTT TLS session for the process + instead of opening a new connection on every refresh. One-shot CLI commands + still connect and tear down. + - `--json` emitters construct `bambu_cli.contracts` objects. `status --json` no longer flattens firmware fields onto the envelope; they stay under `printer`. Errors go through `ErrorEnvelope`; `job` success/failure through diff --git a/bambu_cli/printer.py b/bambu_cli/printer.py index 516a61d..1f36327 100644 --- a/bambu_cli/printer.py +++ b/bambu_cli/printer.py @@ -4,6 +4,7 @@ import os import secrets import ssl +import threading import time from typing import Any, Optional @@ -47,6 +48,38 @@ def __init__( self.mqtt_timeout = 5.0 self.ftps_timeout = 15.0 + # Long-lived MQTT session. None for one-shot CLI (connect + teardown). + # The TUI calls hold_mqtt() so dashboard/monitor refreshes reuse one + # TLS connection instead of opening a new one every 10s. + self._mqtt_session: Any = None + self._mqtt_hold_lock = threading.Lock() + + def hold_mqtt(self, *, client_factory=None) -> None: + """Keep one MQTT client for subsequent status/send_command/get_version. + + No-op in simulation mode (nothing opens a socket). Safe to call twice. + Pair with ``release_mqtt()`` so the loop thread and socket do not leak. + """ + if self.simulation_mode: + return + with self._mqtt_hold_lock: + if self._mqtt_session is None: + from bambu_cli.protocols.mqtt_session import MqttSession + + self._mqtt_session = MqttSession(self, client_factory=client_factory) + + def release_mqtt(self) -> None: + """Disconnect a held MQTT session. Safe to call twice.""" + with self._mqtt_hold_lock: + session = self._mqtt_session + self._mqtt_session = None + if session is not None: + session.close() + + @property + def mqtt_held(self) -> bool: + return self._mqtt_session is not None + def send_command(self, payload: str, timeout: Optional[float] = None, retries: int = 2) -> bool: """Send a JSON command payload via MQTT.""" return mqtt_protocol.send_command(self, payload, timeout=timeout, retries=retries) diff --git a/bambu_cli/protocols/mqtt.py b/bambu_cli/protocols/mqtt.py index 63a27b0..0b07303 100644 --- a/bambu_cli/protocols/mqtt.py +++ b/bambu_cli/protocols/mqtt.py @@ -1,8 +1,9 @@ """MQTT transport facade. -Implementations live in ``mqtt_tls``, ``mqtt_cmd``, ``mqtt_print``, and -``mqtt_monitor``. This module re-exports the public names so existing imports -and test patches on ``bambu_cli.protocols.mqtt`` keep working. +Implementations live in ``mqtt_tls``, ``mqtt_cmd``, ``mqtt_print``, +``mqtt_monitor``, and ``mqtt_session``. This module re-exports the public +names so existing imports and test patches on ``bambu_cli.protocols.mqtt`` +keep working. """ from __future__ import annotations @@ -19,6 +20,7 @@ ) from bambu_cli.protocols.mqtt_monitor import _status_event, monitor_status from bambu_cli.protocols.mqtt_print import _printer_error_hex, execute_print_command +from bambu_cli.protocols.mqtt_session import MqttSession from bambu_cli.protocols.mqtt_tls import ( PinningSSLContext, _mqtt_connect, @@ -39,6 +41,7 @@ "_printer_error_hex", "_require_mqtt", "_status_event", + "MqttSession", "create_mqtt_client", "execute_print_command", "get_status", diff --git a/bambu_cli/protocols/mqtt_cmd.py b/bambu_cli/protocols/mqtt_cmd.py index f9a4683..828ced9 100644 --- a/bambu_cli/protocols/mqtt_cmd.py +++ b/bambu_cli/protocols/mqtt_cmd.py @@ -59,6 +59,10 @@ def send_command( logger.info(f"🤖 [SIM] Sending command: {payload}") return True + session = getattr(printer, "_mqtt_session", None) + if session is not None: + return session.send_command(payload, timeout, retries) + for attempt in range(retries + 1): client = _factory(printer) client.user_data_set({}) @@ -164,6 +168,10 @@ def get_status(printer, timeout=None, retries=2, *, require_complete=True): }, } + session = getattr(printer, "_mqtt_session", None) + if session is not None: + return session.get_status(timeout, retries, require_complete=require_complete) + merged: dict = {} merged_lock = threading.Lock() _factory = _client_factory(None) @@ -265,6 +273,10 @@ def get_version(printer, timeout=5, retries=1): if printer.simulation_mode: return [{"name": "ota", "sw_ver": "01.00.00.00", "hw_ver": "P1P-SIM"}] + session = getattr(printer, "_mqtt_session", None) + if session is not None: + return session.get_version(timeout, retries) + _factory = _client_factory(None) _sleep_fn = _sleep(None) diff --git a/bambu_cli/protocols/mqtt_session.py b/bambu_cli/protocols/mqtt_session.py new file mode 100644 index 0000000..6ab2716 --- /dev/null +++ b/bambu_cli/protocols/mqtt_session.py @@ -0,0 +1,323 @@ +"""A long-lived MQTT client for one ``BambuPrinter``. + +One-shot CLI commands still connect and tear down in ``mqtt_cmd``. The TUI +(and any other long-lived process) calls ``BambuPrinter.hold_mqtt()`` so +``status`` / ``send_command`` / ``get_version`` reuse a single TLS session +instead of opening a new one on every dashboard refresh. + +Threading: operations are serialized on ``_op_lock``. paho callbacks run on +the network thread and only touch Events / the merged print dict — they never +take ``_op_lock``. Reconnect cannot re-issue a state-changing command: each +``send_command`` has a once-flag that survives ``on_connect`` firing again. +""" + +from __future__ import annotations + +import json +import ssl +import threading +from typing import Any, Callable + +from bambu_cli.errors import PrinterStatusIncomplete +from bambu_cli.logging_utils import logger +from bambu_cli.protocols.mqtt_cmd import _REQUIRED_STATUS_KEYS, status_is_complete +from bambu_cli.utils import get_sequence_id + +ClientFactory = Callable[[Any], Any] + + +def _sleep_fn(sleep: Callable[[float], None] | None): + if sleep is not None: + return sleep + from bambu_cli.protocols import mqtt as mqtt_mod + + return mqtt_mod.time.sleep + + +class MqttSession: + """One paho client, reused until ``close()``.""" + + def __init__( + self, + printer: Any, + *, + client_factory: ClientFactory | None = None, + sleep: Callable[[float], None] | None = None, + ) -> None: + self._printer = printer + self._client_factory = client_factory + self._sleep = sleep + self._op_lock = threading.RLock() + self._state_lock = threading.Lock() + self._client: Any = None + self._live = False + self._print_state: dict[str, Any] = {} + self._awaiting_status = False + self._awaiting_require_complete = True + self._status_event = threading.Event() + self._pending_payload: str | None = None + self._command_issued = False + self._publish_ok = False + self._publish_event = threading.Event() + self._version_modules: Any = None + self._version_event = threading.Event() + self._connect_failed = False + self._connected_event = threading.Event() + + def _make_client(self) -> Any: + if self._client_factory is not None: + return self._client_factory(self._printer) + from bambu_cli.protocols import mqtt as mqtt_mod + + return mqtt_mod.create_mqtt_client(self._printer) + + def _connect(self, client: Any) -> None: + from bambu_cli.protocols import mqtt as mqtt_mod + + mqtt_mod._mqtt_connect(self._printer, client) + + def close(self) -> None: + """Stop the loop and disconnect. Safe to call twice.""" + with self._op_lock: + self._reset_client() + + def _reset_client(self) -> None: + client = self._client + self._client = None + self._live = False + self._connect_failed = False + with self._state_lock: + self._print_state = {} + if client is None: + return + try: + client.loop_stop() + except Exception: + pass + try: + client.disconnect() + except Exception: + pass + + def _bind_callbacks(self, client: Any) -> None: + client.on_connect = self._on_connect + client.on_disconnect = self._on_disconnect + client.on_message = self._on_message + client.on_publish = self._on_publish + + def _on_connect( + self, + client: Any, + userdata: Any, + flags: Any, + rc: int, + properties: Any = None, + ) -> None: + if rc == 0: + self._live = True + self._connect_failed = False + try: + client.subscribe(f"device/{self._printer.serial}/report") + except Exception as exc: + logger.debug(f"MQTT session subscribe failed: {exc}") + self._issue_pending() + else: + self._live = False + self._connect_failed = True + logger.error(f"Connection failed: rc={rc}") + self._connected_event.set() + + def _on_disconnect( + self, + client: Any, + userdata: Any, + rc: Any = None, + properties: Any = None, + *args: Any, + **kwargs: Any, + ) -> None: + self._live = False + + def _on_message(self, client: Any, userdata: Any, msg: Any) -> None: + try: + data = json.loads(msg.payload.decode("utf-8")) + except (json.JSONDecodeError, UnicodeDecodeError, AttributeError) as exc: + logger.debug(f"MQTT decode error: {exc}") + return + if not isinstance(data, dict): + return + info = data.get("info") + if isinstance(info, dict) and info.get("command") == "get_version" and "module" in info: + self._version_modules = info["module"] + self._version_event.set() + print_data = data.get("print") + if not isinstance(print_data, dict): + return + with self._state_lock: + self._print_state.update(print_data) + complete = status_is_complete(self._print_state) + if self._awaiting_status and (complete or not self._awaiting_require_complete): + self._awaiting_status = False + self._status_event.set() + + def _on_publish( + self, + client: Any, + userdata: Any, + mid: Any, + reason_code: Any = None, + properties: Any = None, + ) -> None: + self._publish_ok = True + self._publish_event.set() + + def _issue_pending(self) -> None: + if not self._live or self._pending_payload is None or self._command_issued: + return + if self._client is None: + return + self._command_issued = True + self._client.publish( + f"device/{self._printer.serial}/request", + self._pending_payload, + qos=1, + ) + + def _publish_pushall(self) -> None: + if self._client is None: + return + push = json.dumps({"pushing": {"sequence_id": get_sequence_id(), "command": "pushall"}}) + self._client.publish(f"device/{self._printer.serial}/request", push) + + def _arm_status_wait(self, require_complete: bool) -> threading.Event: + with self._state_lock: + self._awaiting_status = True + self._awaiting_require_complete = require_complete + self._status_event = threading.Event() + return self._status_event + + def _snapshot(self) -> dict[str, Any]: + with self._state_lock: + return dict(self._print_state) + + def ensure_connected(self, timeout: float) -> bool: + """Connect (or reconnect after a drop). Returns False on broker rc != 0.""" + if self._client is not None and self._live: + return True + self._reset_client() + client = self._make_client() + if hasattr(client, "user_data_set"): + client.user_data_set({}) + self._connected_event = threading.Event() + self._bind_callbacks(client) + self._client = client + self._connect(client) + try: + client.loop_start() + except Exception: + pass + # connect() may already have fired on_connect (tests); real paho fires + # after loop_start. Wait so we never publish before the subscribe. + if not self._connected_event.wait(timeout): + return False + return self._live + + def get_status( + self, + timeout: float, + retries: int = 2, + *, + require_complete: bool = True, + ) -> dict[str, Any] | None: + sleeper = _sleep_fn(self._sleep) + with self._op_lock: + for attempt in range(retries + 1): + try: + if not self.ensure_connected(timeout): + return None + event = self._arm_status_wait(require_complete) + self._publish_pushall() + if event.wait(timeout): + snapshot = self._snapshot() + if snapshot and (not require_complete or status_is_complete(snapshot)): + return snapshot + if attempt < retries: + with self._state_lock: + saw_partial = bool(self._print_state) + if saw_partial: + logger.warning( + f"Printer sent only partial status on attempt {attempt + 1}. " + "Re-requesting full state..." + ) + else: + logger.warning(f"MQTT status timeout on attempt {attempt + 1}. Retrying...") + sleeper(2**attempt) + except (OSError, ssl.SSLError) as exc: + self._reset_client() + if attempt < retries: + logger.warning(f"MQTT status attempt {attempt + 1} failed: {exc}. Retrying...") + sleeper(2**attempt) + else: + logger.error(f"MQTT status error: {exc}") + partial = self._snapshot() + if partial and require_complete: + missing = [key for key in _REQUIRED_STATUS_KEYS if key not in partial] + raise PrinterStatusIncomplete( + "Printer returned only partial status updates, never a full snapshot " + f"(missing {', '.join(missing)}). It may be busy mid-print; retry the command.", + detail={"missing_keys": missing, "received_keys": sorted(partial)}, + next_command="plate status", + ) + return None + + def send_command(self, payload: str, timeout: float, retries: int = 2) -> bool: + sleeper = _sleep_fn(self._sleep) + with self._op_lock: + for attempt in range(retries + 1): + try: + self._pending_payload = payload + self._command_issued = False + self._publish_ok = False + self._publish_event = threading.Event() + if not self.ensure_connected(timeout): + self._pending_payload = None + return False + self._issue_pending() + if self._publish_event.wait(timeout): + ok = self._publish_ok + self._pending_payload = None + return ok + if attempt < retries: + logger.warning(f"MQTT command timeout on attempt {attempt + 1}. Retrying...") + sleeper(2**attempt) + except (OSError, ssl.SSLError) as exc: + self._reset_client() + if attempt < retries: + logger.warning(f"MQTT command attempt {attempt + 1} failed: {exc}. Retrying...") + sleeper(2**attempt) + else: + logger.error(f"MQTT command error: {exc}") + self._pending_payload = None + return False + + def get_version(self, timeout: float, retries: int = 1) -> Any: + sleeper = _sleep_fn(self._sleep) + request = json.dumps({"info": {"sequence_id": get_sequence_id(), "command": "get_version"}}) + with self._op_lock: + for attempt in range(retries + 1): + try: + self._version_modules = None + self._version_event = threading.Event() + if not self.ensure_connected(timeout): + return None + if self._client is not None: + self._client.publish(f"device/{self._printer.serial}/request", request) + if self._version_event.wait(timeout): + return self._version_modules + if attempt < retries: + sleeper(2**attempt) + except (OSError, ssl.SSLError): + self._reset_client() + if attempt < retries: + sleeper(2**attempt) + return None diff --git a/bambu_cli/tui/app.py b/bambu_cli/tui/app.py index a228445..d162501 100644 --- a/bambu_cli/tui/app.py +++ b/bambu_cli/tui/app.py @@ -88,13 +88,6 @@ def set_job_in_flight(self, value: bool) -> None: def job_in_flight(self) -> bool: return self._job_in_flight - def action_quit(self) -> None: - """Quit, unless a job worker is mid-flight (then say so and stay).""" - if self._job_in_flight: - self.notify("Upload in progress — wait for it to finish.", severity="warning") - return - self.exit() - def action_help(self) -> None: """Open the key reference (never stacks a second copy). @@ -127,6 +120,24 @@ def action_refresh(self) -> None: if callable(refresh): refresh() + def on_unmount(self) -> None: + self.release_mqtt() + + def release_mqtt(self) -> None: + """Tear down the dashboard MQTT session. Idempotent.""" + provider = getattr(self._deps, "status_provider", None) + closer = getattr(provider, "close", None) + if callable(closer): + closer() + + def action_quit(self) -> None: + """Quit, unless a job worker is mid-flight (then say so and stay).""" + if self._job_in_flight: + self.notify("Upload in progress — wait for it to finish.", severity="warning") + return + self.release_mqtt() + self.exit() + def run_app(args: argparse.Namespace, deps: TuiDeps | None = None) -> None: """Construct and run the Textual app (blocks until the user quits). @@ -135,4 +146,8 @@ def run_app(args: argparse.Namespace, deps: TuiDeps | None = None) -> None: actually launches the UI, and so tests can drive ``PlateApp`` directly via ``run_test()`` without going through this blocking call. """ - PlateApp(args, deps).run() + app = PlateApp(args, deps) + try: + app.run() + finally: + app.release_mqtt() diff --git a/bambu_cli/tui/deps.py b/bambu_cli/tui/deps.py index cc034a9..c5c466b 100644 --- a/bambu_cli/tui/deps.py +++ b/bambu_cli/tui/deps.py @@ -48,9 +48,9 @@ class TuiDeps: poll_interval: float = 3.0 def get_status_provider(self) -> Any: - if self.status_provider is not None: - return self.status_provider - return StatusService() + if self.status_provider is None: + self.status_provider = StatusService() + return self.status_provider def get_steps(self) -> Any: if self.steps is not None: @@ -63,9 +63,9 @@ def get_pipeline(self) -> Any: return PipelineService(steps=self.get_steps()) def get_monitor_service(self) -> Any: - if self.monitor_service is not None: - return self.monitor_service - return MonitorService(self.get_status_provider()) + if self.monitor_service is None: + self.monitor_service = MonitorService(self.get_status_provider()) + return self.monitor_service def get_poll_interval(self) -> float: return self.poll_interval diff --git a/bambu_cli/tui/screens/dashboard.py b/bambu_cli/tui/screens/dashboard.py index e6b6908..182eb95 100644 --- a/bambu_cli/tui/screens/dashboard.py +++ b/bambu_cli/tui/screens/dashboard.py @@ -3,10 +3,12 @@ The blocking ``StatusService.fetch`` (``printer.status()`` waits on a ``threading.Event``) runs in a Textual *thread* worker so the UI never freezes; its result is applied to the widgets on the main thread via -``App.call_from_thread`` (widget updates must not run off-thread). A refresh -fires on ``r`` and on a 10 s interval that is only armed while this screen is -active (disarmed on suspend, re-armed on resume) and cancelled on unmount — a -leaked timer would trip the ``-W error::ResourceWarning`` CI mode. +``App.call_from_thread`` (widget updates must not run off-thread). The service +holds one MQTT session for the process, so a refresh does not open a new TLS +connection. A refresh fires on ``r`` and on a 10 s interval that is only armed +while this screen is active (disarmed on suspend, re-armed on resume) and +cancelled on unmount — a leaked timer would trip the ``-W error::ResourceWarning`` +CI mode. A failed fetch is an ordinary ``StatusSnapshot(ok=False)`` value: the panels render an inline "unreachable" state and the app keeps running. diff --git a/bambu_cli/tui/services.py b/bambu_cli/tui/services.py index d080025..348306e 100644 --- a/bambu_cli/tui/services.py +++ b/bambu_cli/tui/services.py @@ -38,19 +38,30 @@ def gcode_state(self) -> str: class StatusService: """Fetch a normalized status snapshot from the configured/simulated printer. - Wraps ``RuntimeContext.for_request(args).printer().status()`` and - ``parse_ams``. NEVER raises: any failure (MQTT error, timeout, missing - config) is captured into ``StatusSnapshot(ok=False, error=...)`` so the - dashboard renders an inline "printer unreachable" state instead of crashing. + Holds one ``RuntimeContext`` / printer for the TUI lifetime and calls + ``hold_mqtt()`` so dashboard and monitor refreshes reuse a single TLS + session. ``close()`` releases it. + + NEVER raises: any failure (MQTT error, timeout, missing config) is + captured into ``StatusSnapshot(ok=False, error=...)`` so the dashboard + renders an inline "printer unreachable" state instead of crashing. """ + def __init__(self, ctx: Any = None) -> None: + self._ctx = ctx + self._printer: Any = None + def fetch(self, args: argparse.Namespace) -> StatusSnapshot: try: from bambu_cli.ams import parse_ams from bambu_cli.context import RuntimeContext - ctx = RuntimeContext.for_request(args) - data = ctx.printer().status() + if self._printer is None: + if self._ctx is None: + self._ctx = RuntimeContext.for_request(args) + self._printer = self._ctx.printer() + self._printer.hold_mqtt() + data = self._printer.status() if not isinstance(data, dict) or not data: return StatusSnapshot(ok=False, error="Printer returned no status.") ams = parse_ams(data) @@ -58,6 +69,13 @@ def fetch(self, args: argparse.Namespace) -> StatusSnapshot: except Exception as exc: # noqa: BLE001 -- see class docstring: never crash the UI return StatusSnapshot(ok=False, error=_short_error(exc)) + def close(self) -> None: + """Drop the held MQTT session. Safe to call twice or before any fetch.""" + printer = self._printer + self._printer = None + if printer is not None: + printer.release_mqtt() + def _short_error(exc: Exception) -> str: """Render an exception as a single short line for the unreachable state.""" diff --git a/tests/test_mqtt_session.py b/tests/test_mqtt_session.py new file mode 100644 index 0000000..fbdc0f4 --- /dev/null +++ b/tests/test_mqtt_session.py @@ -0,0 +1,329 @@ +"""Reusable MQTT session: one TLS client across status/send_command.""" + +from __future__ import annotations + +import json +from unittest.mock import MagicMock + +from bambu_cli.errors import PrinterStatusIncomplete +from bambu_cli.printer import BambuPrinter +from bambu_cli.protocols.mqtt import get_status, get_version, send_command +from bambu_cli.protocols.mqtt_session import MqttSession + +_FULL = { + "gcode_state": "IDLE", + "mc_percent": 0, + "bed_temper": 25.0, + "nozzle_temper": 25.0, +} + + +class FakeBrokerClient: + """In-memory paho stand-in: connect/publish drive the session callbacks.""" + + def __init__(self, *, status_replies=None, version_reply=None, connect_rc=0, double_connect=False): + self.on_connect = None + self.on_disconnect = None + self.on_message = None + self.on_publish = None + self.connects = 0 + self.disconnects = 0 + self.publishes: list[tuple] = [] + self.subscribes: list[str] = [] + self.loop_started = False + self._connected = False + self._status_replies = list(status_replies or []) + self._status_repeat = None if status_replies else {"print": dict(_FULL)} + self._version_reply = version_reply + self._connect_rc = connect_rc + self._double_connect = double_connect + + def user_data_set(self, data): + pass + + def connect(self, host, port, keepalive=10): + self.connects += 1 + self._connected = self._connect_rc == 0 + if self.on_connect: + self.on_connect(self, None, None, self._connect_rc) + if self._double_connect and self._connect_rc == 0: + self.on_connect(self, None, None, 0) + + def subscribe(self, topic): + self.subscribes.append(topic) + + def publish(self, topic, payload, qos=0): + self.publishes.append((topic, payload, qos)) + try: + data = json.loads(payload) + except (TypeError, ValueError): + data = {} + if isinstance(data, dict) and "pushing" in data: + reply = self._status_replies.pop(0) if self._status_replies else self._status_repeat + if reply is not None: + self._deliver(reply) + elif isinstance(data, dict) and isinstance(data.get("info"), dict) and self._version_reply is not None: + self._deliver(self._version_reply) + if self.on_publish: + self.on_publish(self, None, 1) + return MagicMock(rc=0) + + def _deliver(self, payload): + if self.on_message is None: + return + msg = MagicMock() + raw = payload if isinstance(payload, (bytes, bytearray)) else json.dumps(payload).encode() + msg.payload = raw + self.on_message(self, None, msg) + + def loop_start(self): + self.loop_started = True + + def loop_stop(self): + self.loop_started = False + + def disconnect(self): + self.disconnects += 1 + self._connected = False + if self.on_disconnect: + self.on_disconnect(self, None, 0) + + def force_drop(self): + self._connected = False + if self.on_disconnect: + self.on_disconnect(self, None, 0) + + def is_connected(self): + return self._connected + + +def _printer(**kwargs): + return BambuPrinter( + ip="192.168.1.9", + serial="01S", + access_code="code", + **kwargs, + ) + + +def _held(printer, factory, sleep=lambda _s: None): + session = MqttSession(printer, client_factory=factory, sleep=sleep) + printer._mqtt_session = session + return session + + +def test_hold_mqtt_is_noop_in_simulation(): + printer = _printer(simulation_mode=True) + printer.hold_mqtt() + assert printer.mqtt_held is False + assert printer.status()["gcode_state"] == "IDLE" + printer.release_mqtt() + + +def test_two_status_calls_reuse_one_client(): + clients = [] + + def factory(_printer): + client = FakeBrokerClient() + clients.append(client) + return client + + printer = _printer() + _held(printer, factory) + first = get_status(printer, timeout=1) + second = get_status(printer, timeout=1) + assert first["gcode_state"] == "IDLE" + assert second["gcode_state"] == "IDLE" + assert len(clients) == 1 + assert clients[0].connects == 1 + assert clients[0].disconnects == 0 + assert len([p for p in clients[0].publishes if "pushall" in str(p[1])]) == 2 + printer.release_mqtt() + assert clients[0].disconnects == 1 + assert printer.mqtt_held is False + + +def test_drop_reconnects_on_next_status(): + clients = [] + + def factory(_printer): + client = FakeBrokerClient() + clients.append(client) + return client + + printer = _printer() + _held(printer, factory) + assert get_status(printer, timeout=1)["mc_percent"] == 0 + clients[0].force_drop() + assert get_status(printer, timeout=1)["gcode_state"] == "IDLE" + assert len(clients) == 2 + assert clients[0].disconnects >= 1 + printer.release_mqtt() + + +def test_send_command_on_connect_publishes_once(): + clients = [] + + def factory(_printer): + client = FakeBrokerClient(double_connect=True) + clients.append(client) + return client + + printer = _printer() + _held(printer, factory) + assert send_command(printer, '{"print":{"command":"pause"}}', timeout=1) is True + command_publishes = [p for p in clients[0].publishes if p[2] == 1] + assert len(command_publishes) == 1 + printer.release_mqtt() + + +def test_reconnect_after_command_does_not_republish(): + clients = [] + + def factory(_printer): + client = FakeBrokerClient() + clients.append(client) + return client + + printer = _printer() + _held(printer, factory) + assert send_command(printer, '{"print":{"command":"pause"}}', timeout=1) is True + clients[0].force_drop() + assert get_status(printer, timeout=1) is not None + pause_publishes = [ + payload + for _topic, payload, qos in clients[0].publishes + clients[1].publishes + if qos == 1 and "pause" in str(payload) + ] + assert len(pause_publishes) == 1 + printer.release_mqtt() + + +def test_oneshot_status_still_disconnects(): + clients = [] + + def factory(_printer): + client = FakeBrokerClient() + clients.append(client) + return client + + printer = _printer() + from unittest.mock import patch + + with patch("bambu_cli.protocols.mqtt.create_mqtt_client", side_effect=factory): + result = get_status(printer, timeout=1) + assert result["gcode_state"] == "IDLE" + assert len(clients) == 1 + assert clients[0].disconnects == 1 + + +def test_session_incomplete_raises(): + def factory(_printer): + return FakeBrokerClient(status_replies=[{"print": {"wifi_signal": "-40dBm"}}] * 4) + + printer = _printer() + _held(printer, factory) + try: + get_status(printer, timeout=0.01, retries=0) + raise AssertionError("expected PrinterStatusIncomplete") + except PrinterStatusIncomplete as exc: + assert "missing" in str(exc).lower() + printer.release_mqtt() + + +def test_session_liveness_accepts_partial(): + def factory(_printer): + return FakeBrokerClient(status_replies=[{"print": {"wifi_signal": "-40dBm"}}]) + + printer = _printer() + _held(printer, factory) + result = get_status(printer, timeout=1, retries=0, require_complete=False) + assert result == {"wifi_signal": "-40dBm"} + printer.release_mqtt() + + +def test_session_get_version(): + def factory(_printer): + return FakeBrokerClient( + version_reply={"info": {"command": "get_version", "module": [{"name": "ota", "sw_ver": "1.0"}]}} + ) + + printer = _printer() + _held(printer, factory) + assert get_version(printer, timeout=1) == [{"name": "ota", "sw_ver": "1.0"}] + printer.release_mqtt() + + +def test_connect_rc_failure_returns_none(): + def factory(_printer): + return FakeBrokerClient(connect_rc=4) + + printer = _printer() + _held(printer, factory) + assert get_status(printer, timeout=1, retries=0) is None + assert send_command(printer, "{}", timeout=1, retries=0) is False + printer.release_mqtt() + + +def test_oserror_on_connect_retries_then_succeeds(): + calls = {"n": 0} + + def factory(_printer): + calls["n"] += 1 + if calls["n"] == 1: + + class Boom: + def user_data_set(self, data): + pass + + def connect(self, *args, **kwargs): + raise OSError("down") + + def loop_start(self): + pass + + def loop_stop(self): + pass + + def disconnect(self): + pass + + return Boom() + return FakeBrokerClient() + + printer = _printer() + _held(printer, factory) + assert get_status(printer, timeout=1, retries=1)["gcode_state"] == "IDLE" + assert calls["n"] == 2 + printer.release_mqtt() + + +def test_status_timeout_then_retry_succeeds(): + def factory(_printer): + return FakeBrokerClient(status_replies=[None, {"print": dict(_FULL)}]) + + printer = _printer() + _held(printer, factory) + result = get_status(printer, timeout=0.01, retries=1) + assert result is not None + assert result["gcode_state"] == "IDLE" + printer.release_mqtt() + + +def test_hold_mqtt_and_release_are_idempotent(): + printer = _printer() + created = [] + + def factory(_printer): + created.append(FakeBrokerClient()) + return created[-1] + + printer.hold_mqtt(client_factory=factory) + printer.hold_mqtt(client_factory=factory) + assert printer.mqtt_held is True + get_status(printer, timeout=1) + assert len(created) == 1 + printer.release_mqtt() + printer.release_mqtt() + assert printer.mqtt_held is False + assert created[0].disconnects == 1 diff --git a/tests/test_tui_dashboard.py b/tests/test_tui_dashboard.py index 16e326b..a47eaa9 100644 --- a/tests/test_tui_dashboard.py +++ b/tests/test_tui_dashboard.py @@ -255,3 +255,34 @@ async def test_ams_panel_survives_a_markup_shaped_filament_type(): await pilot.pause() text = _all_text(app) assert "a[/b]c" in text + + +async def test_dashboard_timer_disarms_on_suspend_and_unmount(): + """A leaked interval would trip ResourceWarning-as-error in CI.""" + provider = FakeStatusProvider([_IDLE_SNAPSHOT]) + app = PlateApp(_args(), TuiDeps(status_provider=provider)) + async with app.run_test() as pilot: + await pilot.pause() + dash = app.screen + assert dash._timer is not None + dash.on_screen_suspend() + assert dash._timer is None + dash.on_screen_resume() + assert dash._timer is not None + assert dash._timer is None + + +async def test_quit_releases_the_status_provider(): + closed = [] + + class ClosingProvider(FakeStatusProvider): + def close(self): + closed.append(True) + + provider = ClosingProvider([_IDLE_SNAPSHOT]) + app = PlateApp(_args(), TuiDeps(status_provider=provider)) + async with app.run_test() as pilot: + await pilot.pause() + await pilot.press("q") + await pilot.pause() + assert closed diff --git a/tests/test_tui_polish.py b/tests/test_tui_polish.py index dcf110f..85bdb10 100644 --- a/tests/test_tui_polish.py +++ b/tests/test_tui_polish.py @@ -466,10 +466,15 @@ def test_tuideps_defaults_build_the_real_collaborators(): from bambu_cli.tui.services import MonitorService, PipelineService, StatusService deps = TuiDeps() - assert isinstance(deps.get_status_provider(), StatusService) + provider = deps.get_status_provider() + assert isinstance(provider, StatusService) + assert deps.get_status_provider() is provider assert isinstance(deps.get_steps(), CoreGoSteps) assert isinstance(deps.get_pipeline(), PipelineService) - assert isinstance(deps.get_monitor_service(), MonitorService) + monitor = deps.get_monitor_service() + assert isinstance(monitor, MonitorService) + assert deps.get_monitor_service() is monitor + assert monitor._provider() is provider assert deps.get_ams_detector() is read_loaded_ams_material assert deps.get_poll_interval() == 3.0 @@ -692,14 +697,32 @@ def test_status_service_captures_every_failure_shape(monkeypatch): from bambu_cli.tui.services import StatusService, _short_error class EmptyPrinter: + def hold_mqtt(self, **kwargs): + pass + + def release_mqtt(self): + pass + def status(self): return {} class BoomPrinter: + def hold_mqtt(self, **kwargs): + pass + + def release_mqtt(self): + pass + def status(self): raise RuntimeError("mqtt exploded") class SilentPrinter: + def hold_mqtt(self, **kwargs): + pass + + def release_mqtt(self): + pass + def status(self): raise OSError() @@ -719,6 +742,45 @@ def status(self): assert _short_error(ValueError("plain")) == "plain" +def test_status_service_holds_one_printer_and_releases_mqtt(): + from bambu_cli.tui.services import StatusService + + class RecordingPrinter: + def __init__(self): + self.holds = 0 + self.releases = 0 + self.status_calls = 0 + + def hold_mqtt(self, **kwargs): + self.holds += 1 + + def release_mqtt(self): + self.releases += 1 + + def status(self): + self.status_calls += 1 + return { + "gcode_state": "IDLE", + "mc_percent": 0, + "bed_temper": 25, + "nozzle_temper": 25, + } + + printer = RecordingPrinter() + ctx = MagicMock() + ctx.printer.return_value = printer + service = StatusService(ctx=ctx) + first = service.fetch(_args()) + second = service.fetch(_args()) + assert first.ok and second.ok + assert printer.holds == 1 + assert printer.status_calls == 2 + assert ctx.printer.call_count == 1 + service.close() + service.close() + assert printer.releases == 1 + + def test_status_lines_show_targets_and_file(): from bambu_cli.tui.services import status_lines