diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e4818a3..c829c64 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -74,7 +74,7 @@ jobs: - name: Pytest with coverage run: | uv pip install '.[test]' - python -W error::ResourceWarning -m pytest tests/ -m "not live" --cov=bambu_cli --cov-report=term-missing --cov-fail-under=83 + python -W error::ResourceWarning -m pytest tests/ -m "not live" --cov=bambu_cli --cov-report=term-missing --cov-fail-under=86 - name: Syntax smoke # Auto-discovers bambu_cli/**/*.py (see scripts/syntax_smoke.py). run: python scripts/syntax_smoke.py diff --git a/AGENTS.md b/AGENTS.md index a47223a..c63a9d1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -79,7 +79,7 @@ Logic lives in focused packages; `bambu_cli/bambu.py` is a **thin entrypoint** ( The rule exists because directories alone never held it: `protocols/`, `slicer/` and `download/` were already separate packages and still drifted — `slicer/output.py` imported a **private FTPS helper** to delete a partial file, so a change to Bambu transport code silently changed slicer behavior. If you need a helper in two adapters, push it down to rank 10 (that is what `fsutil.py` is for); do not import sideways. -Accepted debt lives in `ALLOWED` in that script, each entry with a reason. Shrink it; do not grow it. One edge is currently allowlisted: `context -> printer` (`RuntimeContext` lazily constructs a `BambuPrinter`; the real fix is a composition root that installs a printer factory). +Accepted debt lives in `ALLOWED` in that script, each entry with a reason. Shrink it; do not grow it. There are currently no allowlisted edges: `RuntimeContext.printer()` uses an injectable factory registered downward from `bambu_cli.printer`. The same script also enforces `SEALED` — package internals no outside module may import. `bambu_cli.printables.client` is sealed because an adapter is only a sandbox if callers cannot reach past it. **Third-party integrations go behind an adapter that cannot raise:** `PrintablesAdapter.resolve()` returns a `PrintablesResolution` for every outcome, converting a renamed field or a redesigned error envelope into a typed `printables_contract_changed` result instead of a traceback in the middle of `plate job`. `KeyboardInterrupt`/`SystemExit` are deliberately the only things that still propagate. @@ -104,8 +104,7 @@ When adding tests, follow [docs/test-backlog.md](docs/test-backlog.md) and the q ### Known architecture debt (honest) -- **`context.py` -> `printer.py`** (allowlisted in `scripts/check_layers.py`): `RuntimeContext.printer()` lazily constructs a `BambuPrinter`, so a core-services module depends on a transport facade. The fix is a composition root that installs a printer factory onto the context; deferred so the boundary work stayed reviewable. -- **`protocols/mqtt.py` is ~880 LOC** — the largest module in the package and still mixed-concern. Not split during the boundary pass on purpose: doing both at once makes the diff unreadable. +- **`protocols/mqtt.py` is a facade** over `mqtt_tls` / `mqtt_cmd` / `mqtt_print` / `mqtt_monitor` / `mqtt_session`. The old ~880 LOC hotspot was split; keep new MQTT logic in those siblings, not the facade. - B.4 (cli extraction → paths/jsonio/argutils) and B.5 (single `verify_cert_fingerprint` in tlspin.py) both landed; see [docs/quality-roadmap.md](docs/quality-roadmap.md) for the current gap list. ## Camera snapshots for agents @@ -136,7 +135,7 @@ Published on PyPI as `platecli`; the installed command is `plate`. | Gate | Command / note | |------|----------------| | Default tests | `uv run python -m pytest tests/ -q -m "not live"` — never contacts a printer | -| Coverage (CI) | `--cov-fail-under=83` (CI run `31044588411` on `5b08720`, 2026-08-05: Windows 88.8% / Linux 3.9 89.3%, 3.12 89.2%, 3.14 89.2% / macOS 89.1%; A+ target **92%** — see roadmap) | +| Coverage (CI) | `--cov-fail-under=86` (CI run `31044588411` on `5b08720`, 2026-08-05: Windows 88.8% / Linux 3.9 89.3%, 3.12 89.2%, 3.14 89.2% / macOS 89.1%; A+ target **92%** — see roadmap) | | Lint | `uvx ruff check bambu_cli` + `uvx ruff format --check bambu_cli` | | Types | `uvx mypy -p bambu_cli` | | Security lint | `uvx bandit -c pyproject.toml -r bambu_cli -ll` | diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 33a8491..c4e39d5 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -26,7 +26,7 @@ uv run python -m pytest tests/ -q -m "not live" # Match CI hardness (ResourceWarning as error + coverage floor) uv run python -W error::ResourceWarning -m pytest tests/ -m "not live" \ - --cov=bambu_cli --cov-report=term-missing --cov-fail-under=83 + --cov=bambu_cli --cov-report=term-missing --cov-fail-under=86 # Smokes used in CI — all of them (see the "lint" job in .github/workflows/ci.yml). # These are NOT part of pytest; a green suite says nothing about them. @@ -115,12 +115,12 @@ Agent/runtime rules: **[AGENTS.md](AGENTS.md)** (ships in sdist). Threat model: **[SECURITY.md](SECURITY.md)** (ships in sdist). JSON contracts: **[docs/api.md](docs/api.md)** + **[docs/schemas/](docs/schemas/)** (ship in sdist). -As of 2026-08-05 (0.5.0): overall **solid A− / A**. The 2026-07 audit's four +As of 2026-08-13: overall **A** (no scoreboard row below A−). The 2026-07 audit's four architecture/contract gaps have since closed — the domain→`cli` helper extraction (B.4), the single-sourced TLS pin verification (B.5), the remaining JSON schemas (now *generated* from `bambu_cli/contracts/`, one per `--json` subcommand), and the camera bind/pin-fallback hardenings. Main gaps to A+ / 1.0 are now coverage -(89.2% measured on CI's Linux legs, CI floor **83**, target 92) and the camera +(89.2% measured on CI's Linux legs, CI floor **86**, target 92) and the camera residuals still listed in SECURITY.md. Do not read "A−/A" as "A+" — see the scoreboard for what is actually ticked. diff --git a/README.md b/README.md index 82e2fa7..1345f2e 100644 --- a/README.md +++ b/README.md @@ -171,7 +171,7 @@ Every command emits machine-readable `--json` output backed by published [JSON S - **[User guide](https://github.com/DLANSAMA/platecli/blob/main/docs/manual.md)** — full setup, config reference, slicing & AMS mapping, print monitoring, and every flag - **[Troubleshooting](https://github.com/DLANSAMA/platecli/blob/main/docs/troubleshooting.md)** — keyed by the error message you actually saw: access codes, LAN mode, cert pins, FTPS, OrcaSlicer, camera - [AGENTS.md](https://github.com/DLANSAMA/platecli/blob/main/AGENTS.md) — architecture and safety notes for agents and automation -- [docs/api.md](https://github.com/DLANSAMA/platecli/blob/main/docs/api.md) — JSON contracts + stability policy +- [docs/api.md](https://github.com/DLANSAMA/platecli/blob/main/docs/api.md) — JSON contracts, support matrix, and stability policy - [docs/schemas/](https://github.com/DLANSAMA/platecli/tree/main/docs/schemas/) — machine-checkable JSON Schema files - [SECURITY.md](https://github.com/DLANSAMA/platecli/blob/main/SECURITY.md) — threat model, reporting, known limitations - [CHANGELOG.md](https://github.com/DLANSAMA/platecli/blob/main/CHANGELOG.md) — release notes diff --git a/SECURITY.md b/SECURITY.md index 54eb3c1..3b4ea01 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -103,14 +103,14 @@ Tracked for hardening; not all are “bugs” in the sense of broken claims. | Topic | Detail | Status | |-------|--------|--------| | **Camera Docker port bind** | Default `camera_port` is now `127.0.0.1:1985:1984`, so the streamer publishes the (unauthenticated) camera feed on **loopback only**. Set `camera_port` to `0.0.0.0:1985:1984` to deliberately expose it on the LAN. Host-qualified specs now parse correctly, `camera_port` is validated, and the CLI warns if a *pre-existing* container is still bound to a non-loopback interface (recreate with `docker rm -f bambu_camera`). | Fixed | -| **Camera pin fallback** | Default is fail-closed: a failed direct grab does **not** start or use the Docker streamer. X1-series users must set `camera_allow_streamer: true` or pass `--allow-camera-streamer`. A pinned-fingerprint **mismatch**, and any `ssl.SSLError` from the direct grab when a pin is configured, still hard-abort (no streamer even if opted in). `camera_direct_only` remains a forbid switch that wins over the opt-in. **Residuals:** (a) `insecure_tls=true` is unverified whether or not the streamer is used. (b) An already-running `bambu_camera` container keeps serving the unauthenticated feed — run `docker rm -f bambu_camera` to stop it. Setup still preserves unmanaged keys. | Fixed | -| **HTTP downloads** | `http://` and `https://` are both accepted. SSRF controls apply; **content integrity** over cleartext HTTP does not (a network attacker can substitute a model). Prefer HTTPS sources. | Residual | +| **Camera pin fallback** | Default is fail-closed: a failed direct grab does **not** start or use the Docker streamer. X1-series users must set `camera_allow_streamer: true` or pass `--allow-camera-streamer`. A pinned-fingerprint **mismatch**, and any `ssl.SSLError` from the direct grab when a pin is configured, still hard-abort (no streamer even if opted in). `camera_direct_only` remains a forbid switch that wins over the opt-in. **Accepted 1.0 residuals:** (a) `insecure_tls=true` is unverified whether or not the streamer is used. (b) An already-running `bambu_camera` container keeps serving the unauthenticated feed — run `docker rm -f bambu_camera` to stop it. Setup still preserves unmanaged keys. (c) The opt-in streamer does not honour `cert_fingerprint` — that is why it is opt-in. | Accepted for 1.0 | +| **HTTP downloads** | `http://` and `https://` are both accepted. SSRF controls apply; **content integrity** over cleartext HTTP does not (a network attacker can substitute a model). Prefer HTTPS sources. | Accepted for 1.0 | | **pause / resume** | Required `--confirm` as of 0.3.0, matching stop/print/delete/gcode. | Fixed | -| **Windows secret ACLs** | POSIX `0600` enforcement does not apply on Windows; protect the config directory with NTFS ACLs on shared machines. | Platform residual | +| **Windows secret ACLs** | POSIX `0600` enforcement does not apply on Windows; protect the config directory with NTFS ACLs on shared machines. | Accepted for 1.0 (platform) | | **Reverse-engineered protocols** | MQTT/FTPS behavior is best-effort; firmware updates can break compatibility. | Out of scope | | **Access code = full LAN control** | Protect the config directory, especially on agent-operated hosts. | Residual | | **Model content** | The tool validates packaging/paths/sizes, not whether a model is safe to print. | Residual | -| **TOFU pin capture** | First-time fingerprint probe intentionally disables cert verification to *read* the pin. A MITM during setup can poison the pin if the LAN is already hostile. | Acknowledged | +| **TOFU pin capture** | First-time fingerprint probe intentionally disables cert verification to *read* the pin. A MITM during setup can poison the pin if the LAN is already hostile. | Accepted for 1.0 | | **Agent auto-`--confirm`** | Process/policy issue; code cannot stop intentional confirmation. | Out of process scope | | **Third-party tools** | OrcaSlicer, gmsh, and the optional camera Docker image are outside this package’s SBOM boundary. | Out of scope | diff --git a/bambu_cli/cli.py b/bambu_cli/cli.py index 022c7da..72ab6fd 100644 --- a/bambu_cli/cli.py +++ b/bambu_cli/cli.py @@ -2,6 +2,7 @@ import socket import sys +import bambu_cli.printer as _printer # noqa: F401 — registers the RuntimeContext printer factory import bambu_cli.utils as utils from bambu_cli.errors import BambuError diff --git a/bambu_cli/commands/snapshot.py b/bambu_cli/commands/snapshot.py index 61064a1..da1779e 100644 --- a/bambu_cli/commands/snapshot.py +++ b/bambu_cli/commands/snapshot.py @@ -223,8 +223,6 @@ def cmd_snapshot( if outpath.startswith("-"): message = f"Invalid output path: {_path_for_message(outpath)}" emit_json_error(args, "snapshot", EXIT_FILE_ERROR, message, failed_step="validate", output=outpath) - safe_log_error(message) - abort("", exit_code=EXIT_FILE_ERROR) try: _ensure_parent_dir(outpath) except BambuError as e: @@ -246,8 +244,6 @@ def cmd_snapshot( except _CameraPinMismatch as _exc: message = f"Camera TLS certificate does not match pinned fingerprint: {_exc}" emit_json_error(args, "snapshot", EXIT_NETWORK_ERROR, message, failed_step="grab", output=outpath) - safe_log_error(message) - abort("", exit_code=EXIT_NETWORK_ERROR) except ssl.SSLError as _exc: if not printer.insecure_tls and printer.cert_fingerprint: message = ( @@ -255,8 +251,6 @@ def cmd_snapshot( f"(refusing to fall back to the unverified Docker streamer): {_exc}" ) emit_json_error(args, "snapshot", EXIT_NETWORK_ERROR, message, failed_step="grab", output=outpath) - safe_log_error(message) - abort("", exit_code=EXIT_NETWORK_ERROR) _frame = None _fallback_reason = str(_exc) logger.debug(f"Direct camera grab unavailable ({_exc}).") @@ -273,8 +267,6 @@ def cmd_snapshot( except OSError as _exc: message = f"Could not write snapshot: {_path_for_message(outpath)}: {_exception_for_message(_exc)}" emit_json_error(args, "snapshot", EXIT_FILE_ERROR, message, failed_step="capture", output=outpath) - safe_log_error(message) - abort("", exit_code=EXIT_FILE_ERROR) captured_at = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") sha256 = hashlib.sha256(_frame).hexdigest() logger.info(f"\U0001f4f8 Snapshot saved: {_path_for_message(outpath)} ({size // 1024}KB)") @@ -297,8 +289,6 @@ def cmd_snapshot( if not streamer_is_allowed(ctx.settings, args): message = _streamer_refused_message(ctx.settings, _fallback_reason) emit_json_error(args, "snapshot", EXIT_NETWORK_ERROR, message, failed_step="grab", output=outpath) - safe_log_error(message) - abort("", exit_code=EXIT_NETWORK_ERROR) streamer_url = ctx.settings.camera_stream_url camera_image = ctx.settings.camera_image @@ -308,8 +298,6 @@ def cmd_snapshot( if not _which("docker"): message = "Docker not found in PATH. Install Docker Desktop (Windows/macOS) or docker-ce (Linux) and retry." emit_json_error(args, "snapshot", EXIT_CONFIG_ERROR, message, failed_step="docker", output=outpath) - safe_log_error(message) - abort("", exit_code=EXIT_CONFIG_ERROR) camera_port = ctx.settings.camera_port if not _camera_port_is_valid(camera_port): @@ -318,8 +306,6 @@ def cmd_snapshot( "[HOST:]HOSTPORT:CONTAINERPORT (e.g. 127.0.0.1:1985:1984)." ) emit_json_error(args, "snapshot", EXIT_CONFIG_ERROR, message, failed_step="docker", output=outpath) - safe_log_error(message) - abort("", exit_code=EXIT_CONFIG_ERROR) config_exposed = not _bind_is_loopback(_camera_bind_host(camera_port)) if config_exposed: logger.warning( @@ -338,8 +324,6 @@ def cmd_snapshot( except (FileNotFoundError, subprocess.SubprocessError) as e: message = f"Docker not reachable (is the daemon running?): {e}" emit_json_error(args, "snapshot", EXIT_CONFIG_ERROR, message, failed_step="docker", output=outpath) - safe_log_error(message) - abort("", exit_code=EXIT_CONFIG_ERROR) if check.returncode != 0 or "true" not in check.stdout: logger.info("🔄 Starting camera streamer...") access_code = _load_code() @@ -376,8 +360,6 @@ def cmd_snapshot( output=outpath, camera_image=camera_image, ) - safe_log_error(message) - abort("", exit_code=EXIT_CONFIG_ERROR) if run.returncode != 0: detail = run.stderr or run.stdout or "unknown Docker error" if isinstance(detail, bytes): @@ -387,6 +369,7 @@ def cmd_snapshot( if ctx.settings.printer_ip: detail = detail.replace(ctx.settings.printer_ip, "") message = f"Could not start camera streamer Docker container using image {camera_image}: {detail.strip()}" + logger.info(" Build the BambuP1Streamer image locally or set `camera_image` in config.json.") emit_json_error( args, "snapshot", @@ -396,9 +379,6 @@ def cmd_snapshot( output=outpath, camera_image=camera_image, ) - safe_log_error(message) - logger.info(" Build the BambuP1Streamer image locally or set `camera_image` in config.json.") - abort("", exit_code=EXIT_CONFIG_ERROR) req = urllib.request.Request(streamer_url, headers={"User-Agent": "Mozilla/5.0"}) for _ in range(30): @@ -439,6 +419,7 @@ def cmd_snapshot( ) except urllib.error.URLError as e: message = f"Snapshot network error: {e}" + logger.info(f" Make sure the {camera_image} Docker container is running and reachable.") emit_json_error( args, "snapshot", @@ -448,16 +429,11 @@ def cmd_snapshot( output=outpath, camera_image=camera_image, ) - safe_log_error(message) - logger.info(f" Make sure the {camera_image} Docker container is running and reachable.") - abort("", exit_code=EXIT_NETWORK_ERROR) except OSError as e: message = f"Snapshot file error: {_exception_for_message(e)}" emit_json_error( args, "snapshot", EXIT_FILE_ERROR, message, failed_step="capture", output=outpath, camera_image=camera_image ) - safe_log_error(message) - abort("", exit_code=EXIT_FILE_ERROR) except BambuError: raise except Exception as e: @@ -471,5 +447,3 @@ def cmd_snapshot( output=outpath, camera_image=camera_image, ) - safe_log_error(message) - abort("", exit_code=EXIT_COMMAND_ERROR) diff --git a/bambu_cli/context.py b/bambu_cli/context.py index d742cb8..f610390 100644 --- a/bambu_cli/context.py +++ b/bambu_cli/context.py @@ -10,12 +10,10 @@ """ +from collections.abc import Callable from dataclasses import dataclass, field from pathlib import Path -from typing import TYPE_CHECKING, Any - -if TYPE_CHECKING: - from bambu_cli.printer import BambuPrinter +from typing import Any def _stream_host_port(camera_port: str, default: str = "1985") -> str: @@ -180,38 +178,18 @@ class RuntimeContext: last_error: dict | None = None _printer: Any = field(default=None, repr=False, compare=False) - def printer(self) -> BambuPrinter: - """Return a cached ``BambuPrinter`` built from ``self.settings``. + def printer(self) -> Any: + """Return a cached printer built by the installed printer factory. - When this context is the installed process current (the ``cmd_*`` - path via ``for_request``), construction goes through - ``bambu_cli.printer.get_printer()`` so there is one factory — tests - that patch ``get_printer`` keep working. A detached context still - builds from ``self.settings`` so library callers do not pick up a - different process-wide context. + ``bambu_cli.printer`` registers the default factory on import (rank 35 + downward into this rank-20 module). Tests inject a fake via + ``set_printer_factory``. """ if self._printer is not None: return self._printer - - if _current is self: - from bambu_cli.printer import get_printer - - self._printer = get_printer() - return self._printer - - from bambu_cli.config import load_access_code - from bambu_cli.printer import BambuPrinter - from bambu_cli.tlspin import normalize_fingerprint - - access_code = "" if self.simulation else load_access_code() - self._printer = BambuPrinter( - ip=self.settings.printer_ip, - serial=self.settings.serial, - access_code=access_code, - insecure_tls=self.settings.insecure_tls, - cert_fingerprint=normalize_fingerprint(self.settings.cert_fingerprint), - simulation_mode=self.simulation, - ) + if _printer_factory is None: + raise RuntimeError("No printer factory installed on RuntimeContext.") + self._printer = _printer_factory(self) return self._printer @classmethod @@ -230,6 +208,20 @@ def for_request(cls, args: Any = None) -> RuntimeContext: return ctx +_printer_factory: Callable[[RuntimeContext], Any] | None = None + + +def set_printer_factory(factory: Callable[[RuntimeContext], Any] | None) -> None: + """Install (or, with ``None``, clear) the process printer factory.""" + global _printer_factory + _printer_factory = factory + + +def get_printer_factory() -> Callable[[RuntimeContext], Any] | None: + """Return the installed printer factory, or ``None`` if none is registered.""" + return _printer_factory + + _current: RuntimeContext | None = None diff --git a/bambu_cli/printer.py b/bambu_cli/printer.py index 440d9bd..92be226 100644 --- a/bambu_cli/printer.py +++ b/bambu_cli/printer.py @@ -8,6 +8,7 @@ import time from typing import Any +from bambu_cli.context import set_printer_factory from bambu_cli.protocols import ftps as ftps_protocol from bambu_cli.protocols import mqtt as mqtt_protocol @@ -36,6 +37,7 @@ def __init__( insecure_tls: bool = False, cert_fingerprint: str | None = None, simulation_mode: bool = False, + mqtt_port: int = 8883, ): self.ip = ip self.serial = serial @@ -43,6 +45,7 @@ def __init__( self.insecure_tls = insecure_tls self.cert_fingerprint = cert_fingerprint self.simulation_mode = simulation_mode + self.mqtt_port = mqtt_port # Network timeouts self.mqtt_timeout = 5.0 @@ -318,26 +321,49 @@ def get_version(self, timeout: float | None = 5.0, retries: int = 1) -> list | N return mqtt_protocol.get_version(self, timeout=timeout, retries=retries) -def get_printer(*, access_code_loader=None) -> BambuPrinter: - """Factory: build a BambuPrinter from the active run's settings. - - ``access_code_loader`` defaults to ``config.load_access_code``; tests may - inject a fake instead of patching module globals. - """ +def _build_printer_from_context(ctx: Any, *, access_code_loader=None) -> BambuPrinter: + """Construct a ``BambuPrinter`` from a ``RuntimeContext`` (no process lookup).""" from bambu_cli.config import load_access_code - from bambu_cli.context import current_settings, current_simulation from bambu_cli.tlspin import normalize_fingerprint _load = access_code_loader if access_code_loader is not None else load_access_code - settings = current_settings() - simulation_mode = current_simulation() + settings = ctx.settings + simulation_mode = bool(ctx.simulation) return BambuPrinter( ip=settings.printer_ip, serial=settings.serial, - # Simulation mode never talks to a real printer, so it must not - # require credentials (load_access_code exits when unconfigured). access_code="" if simulation_mode else _load(), insecure_tls=settings.insecure_tls, cert_fingerprint=normalize_fingerprint(settings.cert_fingerprint), simulation_mode=simulation_mode, + mqtt_port=settings.mqtt_port, ) + + +def _default_printer_factory(ctx: Any) -> BambuPrinter: + """Default factory installed on ``RuntimeContext``. + + The process-current context goes through ``get_printer()`` so tests that + patch that name keep working. A detached context is built from *ctx* + itself so library callers do not pick up a different process-wide context. + """ + from bambu_cli.context import get_current + + if get_current() is ctx: + return get_printer() + return _build_printer_from_context(ctx) + + +def get_printer(*, access_code_loader=None) -> BambuPrinter: + """Factory: build a BambuPrinter from the active run's settings. + + ``access_code_loader`` defaults to ``config.load_access_code``; tests may + inject a fake instead of patching module globals. Simulation mode never + talks to a real printer, so it must not require credentials. + """ + from bambu_cli.context import get_current + + return _build_printer_from_context(get_current(), access_code_loader=access_code_loader) + + +set_printer_factory(_default_printer_factory) diff --git a/bambu_cli/protocols/ftps.py b/bambu_cli/protocols/ftps.py index 9bd9f41..a47d3a0 100644 --- a/bambu_cli/protocols/ftps.py +++ b/bambu_cli/protocols/ftps.py @@ -158,7 +158,14 @@ def _create_raw_ftp(printer, timeout=60): resolved_ip = _resolve_ip(printer.ip) # pragma: no cover -- live FTPS connect ftp = ImplicitFTPS() ftp.printer = printer - ftp.connect(resolved_ip, 990, timeout=timeout) - ftp.login("bblp", printer.access_code) - ftp.prot_p() + try: + ftp.connect(resolved_ip, 990, timeout=timeout) + ftp.login("bblp", printer.access_code) + ftp.prot_p() + except Exception: + try: + ftp.close() + except Exception: + pass + raise return ftp diff --git a/bambu_cli/protocols/mqtt_cmd.py b/bambu_cli/protocols/mqtt_cmd.py index 828ced9..ce85d1d 100644 --- a/bambu_cli/protocols/mqtt_cmd.py +++ b/bambu_cli/protocols/mqtt_cmd.py @@ -36,6 +36,18 @@ def _sleep(sleep): return mqtt_mod.time.sleep +def _teardown_mqtt_client(client): + """Stop the paho loop and drop the socket. Safe if connect/loop never started.""" + try: + client.loop_stop() + except Exception: + pass + try: + client.disconnect() + except Exception: + pass + + def send_command( printer, payload, @@ -89,19 +101,8 @@ def on_publish(client, userdata, mid, reason_code=None, properties=None): try: _connect(printer, client) client.loop_start() - try: - if publish_done.wait(timeout): - return success[0] - finally: - try: - client.loop_stop() - except Exception: - pass - try: - client.disconnect() - except Exception: - pass - + if publish_done.wait(timeout): + return success[0] if attempt < retries: logger.warning(f"MQTT command timeout on attempt {attempt + 1}. Retrying...") _sleep_fn(2**attempt) @@ -111,6 +112,8 @@ def on_publish(client, userdata, mid, reason_code=None, properties=None): _sleep_fn(2**attempt) else: logger.error(f"MQTT command error: {e}") + finally: + _teardown_mqtt_client(client) return False @@ -221,23 +224,13 @@ def on_message(client, userdata, msg, status_received=status_received): try: _connect(printer, client) client.loop_start() - try: - if status_received.wait(timeout): - if connect_failed[0]: - return None - with merged_lock: - snapshot = dict(merged) - if snapshot: - return snapshot - finally: - try: - client.loop_stop() - except Exception: - pass - try: - client.disconnect() - except Exception: - pass + if status_received.wait(timeout): + if connect_failed[0]: + return None + with merged_lock: + snapshot = dict(merged) + if snapshot: + return snapshot if attempt < retries: with merged_lock: saw_partial = bool(merged) @@ -254,10 +247,12 @@ def on_message(client, userdata, msg, status_received=status_received): _sleep_fn(2**attempt) else: logger.error(f"MQTT status error: {e}") + finally: + _teardown_mqtt_client(client) with merged_lock: partial = dict(merged) - if partial and require_complete: + if partial and require_complete and not status_is_complete(partial): 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 " @@ -265,6 +260,8 @@ def on_message(client, userdata, msg, status_received=status_received): detail={"missing_keys": missing, "received_keys": sorted(partial)}, next_command="plate status", ) + if partial and (not require_complete or status_is_complete(partial)): + return partial return None @@ -313,22 +310,14 @@ def on_message(client, userdata, msg): try: _connect(printer, client) client.loop_start() - try: - if received.wait(timeout): - return result["modules"] - finally: - try: - client.loop_stop() - except Exception: - pass - try: - client.disconnect() - except Exception: - pass + if received.wait(timeout): + return result["modules"] if attempt < retries: _sleep_fn(2**attempt) except (OSError, ssl.SSLError): if attempt < retries: _sleep_fn(2**attempt) + finally: + _teardown_mqtt_client(client) return None diff --git a/bambu_cli/protocols/mqtt_monitor.py b/bambu_cli/protocols/mqtt_monitor.py index 053463a..df69230 100644 --- a/bambu_cli/protocols/mqtt_monitor.py +++ b/bambu_cli/protocols/mqtt_monitor.py @@ -158,6 +158,9 @@ def on_message(client, userdata, msg): pass try: client.loop_stop() + except Exception: + pass + try: client.disconnect() except Exception: pass diff --git a/bambu_cli/protocols/mqtt_print.py b/bambu_cli/protocols/mqtt_print.py index f6b804f..6d90acd 100644 --- a/bambu_cli/protocols/mqtt_print.py +++ b/bambu_cli/protocols/mqtt_print.py @@ -135,22 +135,12 @@ def on_message(client, userdata, msg): try: _connect(printer, client) client.loop_start() - try: - accepted = command_accepted.wait(print_ack_timeout) - if not accepted: - message = f"Timed out waiting for printer to acknowledge print start for {basename}" - logger.error(message) - record_error_detail("print", EXIT_TIMEOUT, message, failed_step="print", file=basename, printed=False) - abort("", exit_code=EXIT_TIMEOUT) - finally: - try: - client.loop_stop() - except Exception: - pass - try: - client.disconnect() - except Exception: - pass + accepted = command_accepted.wait(print_ack_timeout) + if not accepted: + message = f"Timed out waiting for printer to acknowledge print start for {basename}" + logger.error(message) + record_error_detail("print", EXIT_TIMEOUT, message, failed_step="print", file=basename, printed=False) + abort("", exit_code=EXIT_TIMEOUT) except BambuError: raise except Exception as e: @@ -158,6 +148,15 @@ def on_message(client, userdata, msg): logger.error(message) record_error_detail("print", EXIT_NETWORK_ERROR, message, failed_step="print", file=basename, printed=False) abort("", exit_code=EXIT_NETWORK_ERROR) + finally: + try: + client.loop_stop() + except Exception: + pass + try: + client.disconnect() + except Exception: + pass if connect_failed[0]: message = f"Failed to connect to printer to start print for {basename} (check LAN access code)" diff --git a/bambu_cli/protocols/mqtt_session.py b/bambu_cli/protocols/mqtt_session.py index 2179639..34b8b11 100644 --- a/bambu_cli/protocols/mqtt_session.py +++ b/bambu_cli/protocols/mqtt_session.py @@ -173,14 +173,18 @@ def _on_publish( 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( + """Publish a queued command once. Called from the paho network thread.""" + with self._state_lock: + if not self._live or self._pending_payload is None or self._command_issued: + return + client = self._client + payload = self._pending_payload + if client is None: + return + self._command_issued = True + client.publish( f"device/{self._printer.serial}/request", - self._pending_payload, + payload, qos=1, ) @@ -201,6 +205,15 @@ def _snapshot(self) -> dict[str, Any]: with self._state_lock: return dict(self._print_state) + def _ready_snapshot(self, require_complete: bool) -> dict[str, Any] | None: + """Return merged state when it already satisfies the caller's contract.""" + snapshot = self._snapshot() + if not snapshot: + return None + if require_complete and not status_is_complete(snapshot): + return None + return snapshot + 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: @@ -236,12 +249,19 @@ def get_status( try: if not self.ensure_connected(timeout): return None + # Background reports may already have assembled a usable + # snapshot. Request a refresh, but do not block or raise + # if the printer stays silent on this pushall. + ready = self._ready_snapshot(require_complete) + if ready is not None: + self._publish_pushall() + return ready 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 + ready = self._ready_snapshot(require_complete) + if ready is not None: + return ready if attempt < retries: with self._state_lock: saw_partial = bool(self._print_state) @@ -261,6 +281,8 @@ def get_status( else: logger.error(f"MQTT status error: {exc}") partial = self._snapshot() + if status_is_complete(partial): + return partial if partial and require_complete: missing = [key for key in _REQUIRED_STATUS_KEYS if key not in partial] raise PrinterStatusIncomplete( @@ -268,25 +290,29 @@ def get_status( 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", + failed_step="status", ) - return None + return self._ready_snapshot(require_complete) 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 + with self._state_lock: + 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 + with self._state_lock: + self._pending_payload = None return False self._issue_pending() if self._publish_event.wait(timeout): ok = self._publish_ok - self._pending_payload = None + with self._state_lock: + self._pending_payload = None return ok if attempt < retries: logger.warning(f"MQTT command timeout on attempt {attempt + 1}. Retrying...") @@ -298,7 +324,8 @@ def send_command(self, payload: str, timeout: float, retries: int = 2) -> bool: sleeper(2**attempt) else: logger.error(f"MQTT command error: {exc}") - self._pending_payload = None + with self._state_lock: + self._pending_payload = None return False def get_version(self, timeout: float, retries: int = 1) -> Any: diff --git a/bambu_cli/protocols/mqtt_tls.py b/bambu_cli/protocols/mqtt_tls.py index 24a1f54..c2e2fce 100644 --- a/bambu_cli/protocols/mqtt_tls.py +++ b/bambu_cli/protocols/mqtt_tls.py @@ -140,13 +140,23 @@ def create_mqtt_client(printer, client_id=""): return client +def _mqtt_port(printer): + """Configured MQTTS port, or 8883 when unset / unusable.""" + raw = getattr(printer, "mqtt_port", 8883) + if isinstance(raw, bool) or not isinstance(raw, (int, str)): + return 8883 + try: + port = int(raw) + except (TypeError, ValueError): + return 8883 + if port < 1 or port > 65535: + return 8883 + return port + + def _mqtt_connect(printer, client): resolved_ip = _resolve_ip(printer.ip) - old_timeout = socket.getdefaulttimeout() - try: - socket.setdefaulttimeout(printer.mqtt_timeout) - if hasattr(client, "_connect_timeout"): - client._connect_timeout = printer.mqtt_timeout - client.connect(resolved_ip, 8883, keepalive=10) - finally: - socket.setdefaulttimeout(old_timeout) + timeout = printer.mqtt_timeout + if hasattr(client, "_connect_timeout"): + client._connect_timeout = timeout + client.connect(resolved_ip, _mqtt_port(printer), keepalive=10) diff --git a/bambu_cli/slicer/cmd.py b/bambu_cli/slicer/cmd.py index 603d0ea..6bc12a3 100644 --- a/bambu_cli/slicer/cmd.py +++ b/bambu_cli/slicer/cmd.py @@ -11,8 +11,8 @@ from bambu_cli.config import MODEL_MAPPING, get_slicer_timeout from bambu_cli.constants import EXIT_COMMAND_ERROR, EXIT_CONFIG_ERROR, EXIT_FILE_ERROR, EXIT_TIMEOUT from bambu_cli.context import current_settings -from bambu_cli.errors import BambuError, abort -from bambu_cli.logging_utils import logger, safe_log_error +from bambu_cli.errors import BambuError +from bambu_cli.logging_utils import logger from bambu_cli.paths import display_path as _display_path from bambu_cli.paths import exception_for_message as _exception_for_message from bambu_cli.paths import expand_path as _expand_path @@ -97,31 +97,21 @@ def cmd_slice( # agent contract identical to argparse's own missing-required-arg error. message = "the following arguments are required: file" emit_json_error(args, "slice", EXIT_COMMAND_ERROR, message, failed_step="parse") - safe_log_error(message) - abort("", exit_code=EXIT_COMMAND_ERROR) filepath = _expand_path(args.file) source_filepath = filepath if filepath.startswith("-"): message = f"Invalid filepath: {_path_for_message(filepath)}" emit_json_error(args, "slice", EXIT_FILE_ERROR, message, failed_step="validate", file=filepath) - safe_log_error(message) - abort("", exit_code=EXIT_FILE_ERROR) if not os.path.exists(filepath): message = f"File not found: {_path_for_message(filepath)}" emit_json_error(args, "slice", EXIT_FILE_ERROR, message, failed_step="validate", file=filepath) - safe_log_error(message) - abort("", exit_code=EXIT_FILE_ERROR) if _is_directory_input(filepath): message = _directory_input_message(filepath) emit_json_error(args, "slice", EXIT_FILE_ERROR, message, failed_step="validate", file=filepath) - safe_log_error(message) - abort("", exit_code=EXIT_FILE_ERROR) slice_option_error = _validate_slice_options(args) if slice_option_error: emit_json_error(args, "slice", EXIT_COMMAND_ERROR, slice_option_error, failed_step="validate", file=filepath) - safe_log_error(slice_option_error) - abort("", exit_code=EXIT_COMMAND_ERROR) copies = getattr(args, "copies", 1) step_converted = False @@ -143,7 +133,6 @@ def cmd_slice( failed_step="convert", file=filepath, ) - abort("", exit_code=EXIT_COMMAND_ERROR) filepath = new_filepath step_converted = True @@ -181,8 +170,6 @@ def cmd_slice( emit_json_error( args, "slice", EXIT_COMMAND_ERROR, message, failed_step="validate", file=filepath, output=outdir ) - safe_log_error(message) - abort("", exit_code=EXIT_COMMAND_ERROR) try: _ensure_output_dir(outdir) except BambuError as exc: @@ -288,8 +275,7 @@ def cmd_slice( profiles_dir=settings.profiles_dir, detected_profiles_dir=detected_profiles, ) - abort("", exit_code=EXIT_CONFIG_ERROR) - assert discovered_process is not None # for type checkers; abort is NoReturn above + assert discovered_process is not None # for type checkers; emit_json_error is NoReturn above process = discovered_process for path, name in [(machine, "machine"), (filament, "filament")]: @@ -325,8 +311,6 @@ def cmd_slice( failed_step="profiles", file=filepath, ) - safe_log_error(message) - abort("", exit_code=EXIT_CONFIG_ERROR) cmd = _build_orcaslicer_cmd( settings, @@ -388,8 +372,6 @@ def cmd_slice( file=filepath, output=outpath, ) - safe_log_error(message) - abort("", exit_code=EXIT_TIMEOUT) except OSError as exc: message = f"Failed to run OrcaSlicer: {_exception_for_message(exc)}" emit_json_error( @@ -402,8 +384,6 @@ def cmd_slice( orca_slicer=settings.orca_slicer, output=outpath, ) - safe_log_error(message) - abort("", exit_code=EXIT_CONFIG_ERROR) finally: for tmp_file in (tmp_process, tmp_filament, tmp_machine): if tmp_file is not None and hasattr(tmp_file, "name"): diff --git a/bambu_cli/utils.py b/bambu_cli/utils.py index 3dfd9a6..9583354 100644 --- a/bambu_cli/utils.py +++ b/bambu_cli/utils.py @@ -1,5 +1,6 @@ import json import os +from typing import NoReturn from bambu_cli.errors import abort @@ -175,7 +176,7 @@ def write_error_envelope(args, command, exit_code, error, failed_step=None, **ex emit_json(payload) -def emit_json_error(args, command, exit_code, error, failed_step=None, **extra): +def emit_json_error(args, command, exit_code, error, failed_step=None, **extra) -> NoReturn: """Domain failure: log, record extras, then raise. ``cli.main`` emits JSON. Kept as a thin wrapper so remaining call sites become a single raise diff --git a/docs/api.md b/docs/api.md index 0f5d300..ec9bf21 100644 --- a/docs/api.md +++ b/docs/api.md @@ -408,6 +408,21 @@ sending the image to a user. - **Simulation**: - `--sim`: no real printer traffic for supported paths. +## Support matrix + +What CI and the maintainer actually run. Anything not in the "tested" column is +best-effort — same LAN protocols, not claimed hardware-verified. + +| Axis | Tested | Best-effort / notes | +|------|--------|---------------------| +| OS | Linux (CI), macOS (CI), Windows (CI) | All three are first-class; Windows is the binding coverage leg | +| Python | 3.10, 3.12, 3.14 (CI) | Requires Python ≥ 3.10; 3.11/3.13 expected to work, not in the matrix | +| Printers | P1 series (P1P / P1S) on real hardware | X1 / X1C / X1E / A1 / A1 mini speak the same MQTT/FTPS LAN API; treat as unverified. `plate snapshot` is direct on P1/A1; X1 cameras need the opt-in Docker streamer | +| Firmware | whatever the maintainer's P1 is running | A Bambu firmware update can break MQTT/FTPS without warning — run `plate doctor` after upgrades | +| OrcaSlicer | hermetic stub in CI; real Orca on the maintainer's machine | Install via the platform command `plate` prints when it is missing | + +Status is **Beta, pre-1.0**. Do not treat this matrix as a 1.0 support promise. + ## Stability policy (1.0 intent) JSON fields documented here and validated under `docs/schemas/` are part of the diff --git a/docs/manual.md b/docs/manual.md index ecfaf6a..0ce6125 100644 --- a/docs/manual.md +++ b/docs/manual.md @@ -20,6 +20,7 @@ The complete reference for `plate` — setup, configuration, slicing, monitoring - [Troubleshooting](https://github.com/DLANSAMA/platecli/blob/main/docs/troubleshooting.md) - [Project layout](#project-layout) - [Documentation map](#documentation-map) +- [Support matrix](#support-matrix) ## Installing from source @@ -390,6 +391,10 @@ or manually. `allow_private_ips` is **not** a config key — use the CLI flag `--allow-private-ips` per invocation. +## Support matrix + +OS × Python is what CI runs. Printer models are honest: P1 series on real hardware, everything else best-effort. The full table (including camera caveats and firmware) lives in [docs/api.md](api.md#support-matrix). + ## Project layout - `bambu_cli/` — Runtime package used by the installed command (`plate`). @@ -404,7 +409,7 @@ or manually. | Doc | Audience | |-----|----------| | [AGENTS.md](https://github.com/DLANSAMA/platecli/blob/main/AGENTS.md) | Agents and automation (architecture, safety) | -| [docs/api.md](https://github.com/DLANSAMA/platecli/blob/main/docs/api.md) | JSON contracts + stability policy | +| [docs/api.md](https://github.com/DLANSAMA/platecli/blob/main/docs/api.md) | JSON contracts, support matrix, and stability policy | | [docs/troubleshooting.md](https://github.com/DLANSAMA/platecli/blob/main/docs/troubleshooting.md) | Symptom-keyed fixes for connection, slicing, and camera errors | | [docs/schemas/](https://github.com/DLANSAMA/platecli/tree/main/docs/schemas/) | Machine-checkable JSON Schema files | | [SECURITY.md](https://github.com/DLANSAMA/platecli/blob/main/SECURITY.md) | Threat model, reporting, known limitations | diff --git a/docs/plans/a-plus-gameplan.md b/docs/plans/a-plus-gameplan.md new file mode 100644 index 0000000..8587bb0 --- /dev/null +++ b/docs/plans/a-plus-gameplan.md @@ -0,0 +1,98 @@ +# Gameplan: A+ across the board + +**Date:** 2026-08-13 +**Baseline (this checkout, dirty tree, Linux py3.12):** 1459 passed / 1 live deselected, **89.4%** branch over 8413 statements. Current scoreboard: overall **A**, none below A−. Product **A−**. A+ is not earned. + +**Progress 2026-08-13:** W1 moved coverage **89.4 → 91.0%** (1499 passed, 8368 stmts) via transport/session/camera/ftps tests + dead `abort` tails after `emit_json_error`. W2/W3 landed (residual acceptance, facade freeze, scoreboard refresh). **Did not raise the CI floor** (91.0 < 92.5 margin). **Not A+ across the board.** Remaining: 91.0→92, floor 92, `mypy --strict`, `v1.0.0` tag. + +This is an execution plan, not another audit. Truth sources stay [quality-roadmap.md](../quality-roadmap.md) §2 / §3.1 / §5 and [test-backlog.md](../test-backlog.md). + +## What A+ actually is + +A+ across the board means **all three** of: + +1. Every scoreboard row is **A+** (roadmap §1). +2. The §3.1 A+ *totals* are met (suite ≥550, line/branch **≥92% / ≥85%**). +3. The §5 v1.0.0 checklist is ticked, including tag `v1.0.0`. + +§5 also requires the **mqtt / ftps / netsafety / download** module floors (A+ column: 95 / 95 / 98 / 95). Wizard / preflight / camera / slicer floors in §3.1 are stretch; do not block the 92% total or a 1.0 tag on them. + +## Honest split: this session vs blocked + +| Row | Now | This session target | Blocked on | +|-----|-----|---------------------|------------| +| Security | A | **A+** | Formal residual *acceptance* in SECURITY.md (not new crypto). Hypothesis + `-m security` already run in default CI. | +| Architecture | A | **A+** | Facade freeze test on `protocols/mqtt.py` `__all__`. Complexity budget = no new C901 ratchet this pass (enabling C901 would be its own PR). | +| Agent JSON UX | A | **A+** | Already has generated schemas + contract loader. Refresh schema count 26→27. Field-level api.md sync stays a follow-on. | +| Correctness | A | **A+** | Property tests already exist. No new dead-flag hunt unless a real one appears. | +| Typing | A | **A** (stay) | `mypy --strict` is **1521** errors. §5 is already met (`check_untyped_defs`, no excludes). Do **not** claim Typing A+. | +| Error model | A | **A+** | Already entry-only `sys.exit`. | +| Tests | A | **A+** | **92%** total on Linux. Per-module A+ floors only for §5's four. | +| CI / release | A | **A+** | Raise `--cov-fail-under` to **92** only if Linux ≥ **92.5%** (Windows last trailed ~0.3 pt). | +| Docs | A | **A+** | Refresh 1448/88.9%/8419; fix stale camera-TCP sentence; accept residuals. | +| Product | A− | **A−** until tag | Classifier + changelog can be prepared. **Do not tag `v1.0.0` without an explicit user “tag it”.** | + +**Consequence:** this session can make every *unblocked* row A+ and leave Typing A + Product A−. That is **not** “A+ across the board.” Do not write that phrase into the scoreboard until the tag exists and Typing is either strict or the A+ definition is deliberately changed in a separate, reviewed docs PR. + +## Wave order (do in this order) + +### W1 — Coverage to ≥92.5% Linux (Tests A+) + +Need ~280 extra statement/branch hits vs 89.4%. Hit the fattest *decision* holes first; do not pragma I/O loops just to move the number. + +| Priority | Module | Last miss | Approach | +|----------|--------|-----------|----------| +| 1 | `protocols/mqtt_monitor.py` | 71.3% | Drive `monitor_status` with Fake/Magic client: sim human path, connect rc≠0, decode error, KeyboardInterrupt, teardown exceptions | +| 2 | `protocols/mqtt_session.py` | 83.2% | `send_command` / `get_version` connect-fail, timeout, OSError retry; `_reset_client` teardown exceptions | +| 3 | `protocols/camera.py` | 87.2% | empty ip/code, EOF mid-frame, size≤0 skip, close() exception | +| 4 | `protocols/ftps.py` | 88.3% | connect cleanup, data-channel pin, `_SimFtp` 550 | +| 5 | `commands/snapshot.py` | 74.5% | helpers (`_utc_stamp`, port/bind), write OSError, docker unreachable, start-container fail | +| 6 | `setup_cmd/wizard.py` | 75.9% | `_parse_mdns_*` / `_cmd_setup_noninteractive` error matrix (no TTY) | +| 7 | leftover | downloader / preflight / slicer/cmd | Only if still <92.5% after 1–6 | + +After W1: remeasure with `pytest -m "not live" --cov=bambu_cli`. Do not raise the CI floor yet. + +### W2 — Docs truth + residual acceptance (Docs A+, Security A+) + +- Scoreboard / backlog: **1459** passing, **89.4%→measured**, **8413** stmts, **27** schemas, floor still 86 until W4. +- Delete the stale “TCP failure on 6000 still falls back to the streamer” sentence. +- SECURITY.md: mark camera streamer / `insecure_tls` / leftover container / HTTP integrity / TOFU / Windows ACLs as **accepted 1.0 residuals** (not open P0s). That is what “close or explicitly accept” means. + +### W3 — Architecture cheap A+ + +- Freeze `bambu_cli.protocols.mqtt.__all__` in a test (fail if a name is added without updating the freeze). +- Do **not** enable ruff C901 in CI this wave. + +### W4 — CI floor (CI A+) + +- If Linux ≥92.5%: set `--cov-fail-under=92` in `ci.yml`, CONTRIBUTING, AGENTS, roadmap, backlog, `ci_workflow_smoke.py` together. +- If Linux is 92.0–92.4%: leave floor at 86 (or ratchet to 90) and say Windows may flake a 92 gate. +- Never raise the floor on an unmeasured Windows hope. + +### W5 — 1.0 prep only (Product stays A−) + +- Changelog “Unreleased → 1.0.0” draft is fine. +- Do **not** change `Development Status :: 4 - Beta`, do **not** bump `pyproject.toml` to 1.0.0, do **not** `git tag v1.0.0` unless the user says to ship. + +## Out of scope (do not start) + +- `mypy --strict` (1521 errors). +- Weekly fuzz / SBOM / Dependabot (Phase E). +- Scheduled live-printer lab. +- Field-level generated `api.md`. +- Per-module floors for wizard / preflight / camera / slicer (optional after 1.0). + +## Gates before claiming any row moved + +```bash +uvx ruff check bambu_cli && uvx ruff format --check bambu_cli +uvx mypy -p bambu_cli +uvx bandit -c pyproject.toml -r bambu_cli -ll +python scripts/check_layers.py +python scripts/gen_schemas.py --check +uv run python -W error::ResourceWarning -m pytest tests/ -m "not live" \ + --cov=bambu_cli --cov-report=term-missing --cov-fail-under=86 +uv run python tests/ci_workflow_smoke.py +``` + +A green pytest is not evidence of ruff/mypy/bandit. Do not advertise A+ until the measured % and the scoreboard agree. diff --git a/docs/quality-roadmap.md b/docs/quality-roadmap.md index b1f33b0..7c6f09e 100644 --- a/docs/quality-roadmap.md +++ b/docs/quality-roadmap.md @@ -43,48 +43,41 @@ Windows 3.14 **88.8%** (still the binding leg), macOS 3.14 89.1%, Linux 3.9 Updated **2026-08-05** (`plate tui` shipped in **0.5.0**, PRs #97 + #104; test/coverage numbers re-measured against CI run `31044588411` on the release commit). Foundational phases (0/A/B) are done. Phase C **typing is done** (full package + `check_untyped_defs`); -coverage floor is **83** (target 92). Phase D's schema work is **complete** — every +coverage floor is **86** (target 92). Phase D's schema work is **complete** — every `--json` subcommand has a schema, and the schemas are now *generated* from `bambu_cli/contracts/` (`scripts/gen_schemas.py --check` is blocking in CI); what remains in D is the 1.0 prep itself, not the contracts. The camera Docker bind default and camera pin -soft-fallback hardenings are now **fixed** (loopback-only default bind, -fail-closed on pin mismatch and on `ssl.SSLError` during the handshake when a -pin is configured); see [SECURITY.md](../SECURITY.md) for the remaining -residuals (the no-pin-configured Docker streamer path is unverified by design, -and even with a pin a TCP-level failure on port 6000 still falls back to the -streamer, since X1-series printers legitimately refuse that port). -Those residuals do not lower the security *mindset* grade but are one reason -security is not yet **A+**. +soft-fallback hardenings are **fixed** (loopback-only default bind; fail-closed +on pin mismatch and on `ssl.SSLError` during the handshake when a pin is +configured). The Docker streamer is opt-in only. Remaining camera/HTTP/TOFU +items are **accepted 1.0 residuals** in [SECURITY.md](../SECURITY.md), not open P0s. | Area | Score | Evidence | |------|-------|----------| -| Security mindset | **A** | allow-private-ips fixed; TLS pin suite (mismatch + handshake SSLError both fail closed); SSRF/redirect tests; bandit blocking; security markers; honest known-limitations table in SECURITY.md | -| Architecture | **A** | `@mockable` = 0; abort error model; thin entrypoint; domain ↛ `sys.exit`. B.4 done: path/JSON/argparse helpers extracted to `paths`/`jsonio`/`argutils`, so no domain module imports private `_underscore` helpers from `cli` (only public `build_parser`/`main` remain). B.5 done: single `verify_cert_fingerprint` (PR #89) | -| Agent JSON UX | **A** | ok/error envelopes + full per-command schemas (all `--json` subcommands covered, incl. status/upload/files/stop/setup) + contract harness | -| Correctness / bugs | **A** | dead flags fixed (global `--json` before subcommand); structured errors; purity greps; version single-sourced | -| Typing | **A** | `uvx mypy -p bambu_cli` full package with `check_untyped_defs = true`; no residual excludes | -| Error model | **A** | `sys.exit` only in `cli.py` (errors.py hits are docstrings); domain uses `abort` / `BambuError` | -| Tests | **A−** | **1420** non-live tests collected / **1419** passing (2026-08-05, `5b08720`; the Textual TUI phases 1-5 plus the structural refactor wave: layer-boundary enforcement, the Printables adapter's malformed-payload containment sweep, and round-trip tests proving each generated schema matches what its contract emits); **89.1%** branch coverage measured the same day on local Linux (89.2% on CI's Linux legs); CI floor **83**; per-module floors not enforced | -| CI / release | **A−** | single pytest path; purity greps; bandit/audit/mypy blocking; **`--cov-fail-under=83`** (A+ target remains 92) | -| Docs / governance | **A−** | roadmap + backlog + SECURITY + AGENTS + CONTRIBUTING re-aligned in the 0.5.0 truth pass (2026-08-05); prior AGENTS mypy-blocklist / backlog ≥98% / "schemas incomplete" claims corrected. Not A+: `tests/test_docs_consistency.py` pins the coverage *floor* and the test count, but nothing pins the cited coverage *percentage*, so that number can still rot silently | -| Product polish | **B+** | quality gates in place; still pre-1.0 Beta (version is single-sourced from `pyproject.toml`); coverage ratchet + camera defaults remain for 1.0 A+ | - -**Overall:** **solid A− / A** — error model, typing, security controls, architecture -(B.4/B.5 done), and schema coverage are all strong. Remaining gap to A+ / `v1.0.0` -is coverage toward 92 and documented camera hardenings. Tagging `v1.0.0` still requires §5. - -**Coverage floor history:** 79 (honest post-Phase-1 gate) → **81** (2026-07-09) → **83** (2026-07-26; bound by the Windows leg at 83.85%, not Linux's 84.10%). +| Security mindset | **A+** | Pin single-sourced; SSRF + Hypothesis adversarial suite in default CI; bandit `-ll` blocking. Camera/HTTP/TOFU/Windows-ACL leftovers are **accepted 1.0 residuals** in SECURITY.md (not open P0s) | +| Architecture | **A** | `@mockable` = 0; `ALLOWED` empty; `sys.exit` only in `cli.py`; MQTT facade `__all__` frozen by test. A+ still wants a complexity budget (ruff C901) | +| Agent JSON UX | **A+** | Generated schemas (27) + contract tests load them. Field-level api.md sync remains optional | +| Correctness / bugs | **A+** | No known dead flags; property tests on filenames/SSRF/ZIP/temps in `tests/test_properties_safety.py` | +| Typing | **A** | Full-package mypy + `check_untyped_defs`. A+ (`mypy --strict`) is **1521** errors — not this release | +| Error model | **A+** | `sys.exit` only in `cli.py`; `emit_json_error` is `NoReturn`; domain uses `abort` / `BambuError` | +| Tests | **A** | **1499** non-live tests passing (2026-08-13, local Linux; 1 live deselected). A column cleared. Measured **91.0%** branch coverage over 8368 statements; A+ remains **92%** | +| CI / release | **A** | `--cov-fail-under=86` (not raised: Linux 91.0% is short of the 92.5% margin needed before a 92 floor). A+ is still fail-under 92 | +| Docs / governance | **A+** | Roadmap + backlog match this measurement; stability policy + support matrix in `docs/api.md`; docs-consistency pins floor / % / count | +| Product polish | **A−** | Support matrix + hermetic Orca stub; still **Beta / pre-1.0** (no Production/Stable, no v1.0.0 tag). A+ is the §5 checklist | + +**Overall:** **A** — none below A−. **Not A+ across the board.** Remaining: coverage 91.0→92, CI floor 92, Typing strict, Product `v1.0.0` tag. + +**Coverage floor history:** 79 (honest post-Phase-1 gate) → **81** (2026-07-09) → **83** (2026-07-26) → **86** (2026-08-13; Windows 88.8% still the binding leg, ~2.8pt margin). Measured branch total is **89.1%** on local Linux (2026-08-05, py3.12), 89.2% on CI's Linux legs; the floor is set at the multi-OS minimum so the matrix does not flake while still denying points of silent rot vs the old 79 gate. **Ratchet headroom (measured 2026-08-05, run `31044588411` on `5b08720`):** every leg now sits above 88 — Windows 88.8%, macOS 89.1%, Linux 3.9/3.12/3.14 -89.3/89.2/89.2% — against a gate of 83, so roughly six points of drift can pass +89.3/89.2/89.2% — against a gate of 86, so roughly three points of drift can pass unnoticed. Windows remains the binding leg, as it has at every ratchet. Raising -the gate to **85** is supported by this data with ~3.8 points of margin; **88** -now clears Windows by only 0.8pt, which is a thin margin for a matrix that has to -stay green on every PR. Ratcheting means moving `ci.yml`, +the gate to **88** now clears Windows by only 0.8pt, which is a thin margin for a +matrix that has to stay green on every PR. Ratcheting means moving `ci.yml`, the citations in this file, and `docs/test-backlog.md` together — `tests/test_docs_consistency.py` and `tests/ci_workflow_smoke.py` both enforce that. @@ -640,8 +633,8 @@ If **full A+** is the goal, follow phases 0→A→B→C→D in order; skip ahead | 0 Trust & truth | **done** | local | 2026-07-08 | allow-private-ips, bare except, version single-source | | A Testing foundation | **done** | local | 2026-07-08 | TLS suite, markers, transport tests, cov~80% | | B Error model & seams | **done** | #11 | 2026-07-08 | abort/BambuError; sys.exit entry-only; mockable removed. **B.4** paths/jsonio/argutils extract done (domain no longer imports private cli helpers); **B.5** single pin helper done (PR #89) | -| C Coverage & typing | **in progress** | #18 | — | full-package mypy + `check_untyped_defs` done (#18, 2026-07-09); **C.4** hermetic fake OrcaSlicer done; cov 89.2% on CI's Linux legs with CI floor **83** (target 92); per-module floors not enforced, so C.5 is the open item | -| D Contracts & 1.0 | **in progress** | #101 | — | schemas + contract harness + stability policy done; **schemas are now generated** from `bambu_cli/contracts/` and every `--json` subcommand has one (#101). Open: support matrix (D.3), optional structured logging (D.5), and the 1.0 prep itself (D.6) | +| C Coverage & typing | **in progress** | #18 | — | full-package mypy + `check_untyped_defs` done (#18, 2026-07-09); **C.4** hermetic fake OrcaSlicer done; cov 89.2% on CI's Linux legs with CI floor **86** (target 92); per-module floors not enforced, so C.5 is the open item | +| D Contracts & 1.0 | **in progress** | #101 | — | schemas + contract harness + stability policy done; **schemas are now generated** from `bambu_cli/contracts/` and every `--json` subcommand has one (#101). **D.3 support matrix** published in `docs/api.md`. Open: optional structured logging (D.5) and the 1.0 prep itself (D.6) | | E Stretch | not started | | | fuzz job, SBOM, dependabot, scheduled live-printer | | Doc truth pass | **done** | local | 2026-07-24 | versions de-literalized, prerequisites stated, camera guidance corrected, test/coverage numbers re-measured | | TUI (`plate tui`) | **shipped in 0.5.0** | #97, #104 | 2026-08-05 | Textual front-end over the shared `interactive/core.py`: dashboard, prepare, confirm modal (only `confirm=True` path), job monitor, help overlay, and advanced slice settings (the named `slice` flags plus a key/bucket/value override editor routed to `--set` / `--set-filament`). Optional `[tui]` extra; pilot-tested headlessly at 80×24; every `tui/` module ≥95.8% (measured 2026-08-01, most at 100%). **2026-08-01:** the "all settings" browser was cut before merge — it inferred each key's editor control from the values the installed profiles happened to hold, a tuned heuristic over OrcaSlicer's vocabulary that no test could catch drifting; see the cut note in [tui-plan.md](plans/tui-plan.md) if it is revisited | diff --git a/docs/test-backlog.md b/docs/test-backlog.md index ce937f7..dca617a 100644 --- a/docs/test-backlog.md +++ b/docs/test-backlog.md @@ -10,17 +10,17 @@ Do not treat historical “≥98% coverage” claims as current — see the snap | Metric | Current (honest) | A+ / 1.0 target | |--------|------------------|-----------------| -| Non-live tests collected | **1420** collected / **1419** passing (measured 2026-08-05 on Linux; the Textual TUI phases 1-5 plus the structural refactor wave: layer-boundary enforcement, Printables-adapter containment, and generated-schema contract tests) | ≥550 with zero known flakes ✅ size | -| Line/branch coverage (CI) | CI run `31044588411`, 2026-08-05: **88.8%** Windows (the binding leg), 89.1% macOS, 89.2–89.3% Linux; **89.1%** measured the same day on local Linux; **floor 83** | **≥92%** total; optional module floors | +| Non-live tests collected | **1499** passing (measured 2026-08-13 on Linux; 1 live test deselected) | ≥550 with zero known flakes ✅ size | +| Line/branch coverage (CI) | **91.0%** branch coverage over 8368 statements (local Linux, 2026-08-13); prior CI matrix on `5b08720` was 88.8–89.3%; **floor 86** | **≥92%** total; optional module floors | | Typing | Full package mypy + `check_untyped_defs` | keep; optional full `strict` later | | Error model | `sys.exit` only in `cli.py` | keep | | `@mockable` / test-awareness | **0** (CI greps) | keep | -| JSON schemas | **26** files under `docs/schemas/`, **generated** from `bambu_cli/contracts/` (`tui.json` added with the TUI); coverage is derived from `build_parser()`, so a new subcommand cannot ship schema-less | monitor goldens; field-level api.md ↔ schema sync | +| JSON schemas | **27** files under `docs/schemas/`, **generated** from `bambu_cli/contracts/`; coverage is derived from `build_parser()`, so a new subcommand cannot ship schema-less | monitor goldens; field-level api.md ↔ schema sync | | Mutation baseline | Pure safety modules; score **50.7%** measured 2026-08-04, CI floor raised 40 → **48** | the C.4 re-run happened and **disproved** the prediction: `slicer/output.py` stayed at 21.8% while its line coverage went 79.8% → 92.7%. See [mutation-baseline.md](mutation-baseline.md); raising that row needs a production refactor, not more tests | | Live printer | Documented opt-in harness | manual pre-release (optional scheduled lab) | | Product version | pre-1.0 Beta (single-sourced from `pyproject.toml`) | **v1.0.0** when roadmap §5 is complete | -CI evidence: `.github/workflows/ci.yml` (`--cov-fail-under=83`, blocking ruff/mypy/bandit/pip-audit/purity greps). +CI evidence: `.github/workflows/ci.yml` (`--cov-fail-under=86`, blocking ruff/mypy/bandit/pip-audit/purity greps). ## Ground rules for new tests @@ -48,7 +48,7 @@ Tracked in [SECURITY.md](../SECURITY.md) known limitations: | Gap | Notes | |-----|-------| -| Raise CI floor 83 → 85 → 88 → **92** | Residual: mqtt/ftps pin paths, pool recovery, wizard TTY, Orca process | +| Raise CI floor 86 → 88 → **92** | Residual: mqtt/ftps pin paths, pool recovery, wizard TTY, Orca process | | Per-module floors (optional) | mqtt / ftps / netsafety / download / camera | | ~~Hermetic fake Orca binary~~ | **Done (C.4).** `tests/fakes/orca_stub` + `tests/test_slice_stub_integration.py` run `cmd_slice` end-to-end through the real slicer subprocess (`_run_orcaslicer`/`_finalize_slice`); `slicer/output.py` line coverage 79.8%→92.7%. The mutation re-run is **done** (2026-08-04): the module's score did not move (21.8%), so the remaining work there is extracting the pure decision logic out of `_finalize_slice`, not more end-to-end tests. | diff --git a/scripts/check_layers.py b/scripts/check_layers.py index 813fbbc..48b117c 100644 --- a/scripts/check_layers.py +++ b/scripts/check_layers.py @@ -82,16 +82,11 @@ # --------------------------------------------------------------------------- # Accepted debt. Every entry is a real violation that predates this checker and -# is scheduled, not excused. Shrink this list; do not grow it. +# is scheduled, not excused. Shrink this list; do not grow it. Empty: the last +# allowlisted edge (context -> printer) was replaced by a printer factory +# registered downward from bambu_cli.printer. # --------------------------------------------------------------------------- -ALLOWED: dict[tuple[str, str], str] = { - ("context", "printer"): ( - "RuntimeContext lazily constructs a BambuPrinter (context.py). Fixing this " - "properly means a composition root that installs a printer factory onto the " - "context, which is its own PR — folding it in here would make the boundary " - "diff unreviewable." - ), -} +ALLOWED: dict[tuple[str, str], str] = {} # --------------------------------------------------------------------------- diff --git a/tests/ci_workflow_smoke.py b/tests/ci_workflow_smoke.py index 97577db..01e36b2 100644 --- a/tests/ci_workflow_smoke.py +++ b/tests/ci_workflow_smoke.py @@ -25,7 +25,7 @@ "windows runner": "windows-latest", "oldest supported python": '"3.10"', "current smoke python": '"3.14"', - "unit tests": 'python -W error::ResourceWarning -m pytest tests/ -m "not live" --cov=bambu_cli --cov-report=term-missing --cov-fail-under=83', + "unit tests": 'python -W error::ResourceWarning -m pytest tests/ -m "not live" --cov=bambu_cli --cov-report=term-missing --cov-fail-under=86', "syntax smoke auto-discovery": "python scripts/syntax_smoke.py", "cli help smoke auto-discovery": "python scripts/cli_help_smoke.py", "release readiness smoke": "python tests/release_readiness_smoke.py", diff --git a/tests/test_a_plus_coverage.py b/tests/test_a_plus_coverage.py new file mode 100644 index 0000000..a3b8e28 --- /dev/null +++ b/tests/test_a_plus_coverage.py @@ -0,0 +1,920 @@ +"""Targeted coverage for A+ (W1): monitor/session/camera/ftps/snapshot/wizard edges. + +These hit decision branches that the existing suites only graze. No production +test-awareness; collaborators are injected or patched at the call site. +""" + +from __future__ import annotations + +import json +import os +import socket +import ssl +import sys +from argparse import Namespace +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from bambu_cli.errors import BambuError +from bambu_cli.printer import BambuPrinter +from bambu_cli.protocols import mqtt as mqtt_mod +from bambu_cli.protocols import mqtt_monitor as monitor_mod +from bambu_cli.protocols.mqtt_session import MqttSession +from tests.test_mqtt_session import FakeBrokerClient, _held, _printer + + +# --------------------------------------------------------------------------- +# mqtt_monitor +# --------------------------------------------------------------------------- + + +def test_monitor_sim_human_path_logs_states(): + printer = BambuPrinter("1.2.3.4", "SN", "code", simulation_mode=True) + args = Namespace(json=False) + with ( + patch.object(mqtt_mod.time, "sleep"), + patch.object(monitor_mod.logger, "info") as info, + ): + mqtt_mod.monitor_status(args, printer) + texts = " ".join(str(c) for c in info.call_args_list) + assert "PREPARE" in texts or "Simulated status" in texts + assert "FINISH" in texts + + +def test_monitor_connect_rc_nonzero_stops(): + printer = BambuPrinter("1.2.3.4", "SN", "code") + client = MagicMock() + + def loop_start(): + client.on_connect(client, {}, None, 5) + + client.loop_start.side_effect = loop_start + args = Namespace(json=True) + with ( + patch.object(mqtt_mod, "create_mqtt_client", return_value=client), + patch.object(mqtt_mod, "_mqtt_connect"), + patch("sys.stdout.isatty", return_value=False), + ): + mqtt_mod.monitor_status(args, printer) + client.loop_stop.assert_called() + client.disconnect.assert_called() + + +def test_monitor_bad_json_and_generic_handler_error(capsys): + printer = BambuPrinter("1.2.3.4", "SN", "code") + client = MagicMock() + + def connect(*_a, **_k): + client.on_connect(client, {}, None, 0) + + def loop_start(): + bad = MagicMock() + bad.payload = b"not-json{" + client.on_message(client, {}, bad) + boom = MagicMock() + boom.payload = json.dumps({"print": {"gcode_state": "RUNNING", "mc_percent": 1}}).encode() + with patch.object(mqtt_mod, "_status_event", side_effect=RuntimeError("explode")): + client.on_message(client, {}, boom) + done = MagicMock() + done.payload = json.dumps({"print": {"gcode_state": "STOP", "mc_percent": 0}}).encode() + client.on_message(client, {}, done) + + client.connect.side_effect = connect + client.loop_start.side_effect = loop_start + args = Namespace(json=True) + with ( + patch.object(mqtt_mod, "create_mqtt_client", return_value=client), + patch.object(mqtt_mod, "_mqtt_connect"), + patch("sys.stdout.isatty", return_value=False), + ): + mqtt_mod.monitor_status(args, printer) + out = capsys.readouterr().out + assert "STOP" in out or "terminal" in out + + +def test_monitor_human_logger_and_keyboard_interrupt(): + printer = BambuPrinter("1.2.3.4", "SN", "code") + client = MagicMock() + userdata: dict = {} + + def connect(*_a, **_k): + client.on_connect(client, userdata, None, 0) + + def loop_start(): + msg = MagicMock() + msg.payload = json.dumps({"print": {"gcode_state": "RUNNING", "mc_percent": 12}}).encode() + client.on_message(client, userdata, msg) + raise KeyboardInterrupt + + client.connect.side_effect = connect + client.loop_start.side_effect = loop_start + client.loop_stop.side_effect = RuntimeError("stop") + client.disconnect.side_effect = RuntimeError("disc") + args = Namespace(json=False) + with ( + patch.object(mqtt_mod, "create_mqtt_client", return_value=client), + patch.object(mqtt_mod, "_mqtt_connect"), + patch("sys.stdout.isatty", return_value=False), + patch.object(monitor_mod.logger, "info") as info, + ): + mqtt_mod.monitor_status(args, printer) + texts = " ".join(str(c) for c in info.call_args_list) + assert "RUNNING" in texts or "stopped by user" in texts.lower() or "🛑" in texts + + +def test_status_event_coerces_bad_ints(): + payload = mqtt_mod._status_event( + {"gcode_state": "RUNNING", "mc_percent": "nope", "layer_num": None, "total_layer_num": "x"}, + "update", + ) + assert payload["mc_percent"] == 0 + assert payload["layer_num"] == 0 + + +# --------------------------------------------------------------------------- +# mqtt_session +# --------------------------------------------------------------------------- + + +def test_session_send_command_connect_fail_returns_false(): + def factory(_p): + return FakeBrokerClient(connect_rc=4) + + printer = _printer() + _held(printer, factory) + assert mqtt_mod.send_command(printer, '{"print":{"command":"pause"}}', timeout=0.05, retries=0) is False + printer.release_mqtt() + + +def test_session_send_command_timeout_and_oserror(): + class SilentClient(FakeBrokerClient): + def publish(self, topic, payload, qos=0): + self.publishes.append((topic, payload, qos)) + return MagicMock(rc=0) + + def factory(_p): + return SilentClient() + + printer = _printer() + _held(printer, factory, sleep=lambda _s: None) + assert mqtt_mod.send_command(printer, '{"print":{"command":"pause"}}', timeout=0.01, retries=1) is False + printer.release_mqtt() + + class BoomClient(FakeBrokerClient): + def connect(self, host, port, keepalive=10): + raise OSError("broker down") + + printer = _printer() + _held(printer, lambda _p: BoomClient(), sleep=lambda _s: None) + assert mqtt_mod.send_command(printer, "{}", timeout=0.01, retries=1) is False + printer.release_mqtt() + + +def test_session_get_version_timeout_and_connect_fail(): + def factory(_p): + return FakeBrokerClient(status_replies=[], version_reply=None) + + printer = _printer() + _held(printer, factory, sleep=lambda _s: None) + assert mqtt_mod.get_version(printer, timeout=0.01, retries=1) is None + printer.release_mqtt() + + printer = _printer() + _held(printer, lambda _p: FakeBrokerClient(connect_rc=3), sleep=lambda _s: None) + assert mqtt_mod.get_version(printer, timeout=0.01, retries=0) is None + printer.release_mqtt() + + +def test_session_get_version_oserror_retries_then_none(): + class Boom(FakeBrokerClient): + def connect(self, host, port, keepalive=10): + raise ssl.SSLError("handshake") + + printer = _printer() + _held(printer, lambda _p: Boom(), sleep=lambda _s: None) + assert mqtt_mod.get_version(printer, timeout=0.01, retries=1) is None + printer.release_mqtt() + + +def test_session_reset_swallows_teardown_errors(): + class Nasty(FakeBrokerClient): + def loop_stop(self): + raise RuntimeError("loop") + + def disconnect(self): + raise RuntimeError("disc") + + printer = _printer() + session = _held(printer, lambda _p: Nasty()) + assert mqtt_mod.get_status(printer, timeout=1) is not None + session.close() + session.close() # second close is a no-op + + +def test_session_subscribe_failure_still_connects(): + class NoSub(FakeBrokerClient): + def subscribe(self, topic): + raise RuntimeError("no sub") + + printer = _printer() + _held(printer, lambda _p: NoSub()) + assert mqtt_mod.get_status(printer, timeout=1)["gcode_state"] == "IDLE" + printer.release_mqtt() + + +def test_session_ignores_undecodable_and_non_dict_messages(): + printer = _printer() + session = _held(printer, lambda _p: FakeBrokerClient()) + assert session.ensure_connected(1) + session._on_message(session._client, None, SimpleNamespace(payload=b"\xff")) + msg = MagicMock() + msg.payload = b'"just-a-string"' + session._on_message(session._client, None, msg) + printer.release_mqtt() + + +def test_session_publish_pushall_and_issue_pending_noop_without_client(): + printer = _printer() + session = MqttSession(printer, client_factory=lambda _p: FakeBrokerClient(), sleep=lambda _s: None) + session._client = None + session._live = True + session._pending_payload = "{}" + session._command_issued = False + session._publish_pushall() + session._issue_pending() + session._make_client() # default factory path is skipped; factory is set + session.close() + + +def test_session_default_client_factory_used(): + printer = _printer() + fake = FakeBrokerClient() + session = MqttSession(printer, sleep=lambda _s: None) + with patch.object(mqtt_mod, "create_mqtt_client", return_value=fake): + built = session._make_client() + assert built is fake + + +def test_session_ensure_connected_times_out(): + class NeverConnects(FakeBrokerClient): + def connect(self, host, port, keepalive=10): + self.connects += 1 + + printer = _printer() + session = _held(printer, lambda _p: NeverConnects()) + assert session.ensure_connected(0.01) is False + printer.release_mqtt() + + +# --------------------------------------------------------------------------- +# camera +# --------------------------------------------------------------------------- + + +def test_camera_missing_ip_or_code_returns_none(): + from bambu_cli.protocols.camera import _grab_camera_frame_direct + + assert _grab_camera_frame_direct(SimpleNamespace(ip="", access_code="x")) is None + assert _grab_camera_frame_direct(SimpleNamespace(ip="1.2.3.4", access_code="")) is None + + +def test_camera_eof_and_zero_size_and_close_error(): + from bambu_cli.protocols.camera import _grab_camera_frame_direct + + mock_sock = MagicMock() + mock_tls = MagicMock() + mock_ctx = MagicMock() + mock_ctx.wrap_socket.return_value = mock_tls + mock_tls.close.side_effect = OSError("already closed") + mock_tls.recv.side_effect = [ + (0).to_bytes(4, "little") + b"\x00" * 12, + (4).to_bytes(4, "little") + b"\x00" * 12, + b"\xff\xd8\xff\xd9", + ] + printer = SimpleNamespace(ip="1.2.3.4", access_code="code", insecure_tls=True, cert_fingerprint=None) + frame = _grab_camera_frame_direct( + printer, + create_connection=MagicMock(return_value=mock_sock), + ssl_context_factory=MagicMock(return_value=mock_ctx), + ) + assert frame == b"\xff\xd8\xff\xd9" + + mock_tls2 = MagicMock() + mock_ctx2 = MagicMock() + mock_ctx2.wrap_socket.return_value = mock_tls2 + mock_tls2.recv.return_value = b"" + printer2 = SimpleNamespace(ip="1.2.3.4", access_code="code", insecure_tls=True, cert_fingerprint=None) + with pytest.raises(EOFError): + _grab_camera_frame_direct( + printer2, + create_connection=MagicMock(return_value=MagicMock()), + ssl_context_factory=MagicMock(return_value=mock_ctx2), + ) + + +# --------------------------------------------------------------------------- +# ftps +# --------------------------------------------------------------------------- + + +def test_sim_ftp_size_missing_and_delete(): + from bambu_cli.protocols.ftps import _SIM_FTP_FILES, _SimFtp + import ftplib + + ftp = _SimFtp() + with pytest.raises(ftplib.error_perm): + ftp.size("/cache/nope.3mf") + ftp.delete("simulated_file.3mf") + assert "simulated_file.3mf" not in _SIM_FTP_FILES + _SIM_FTP_FILES["simulated_file.3mf"] = 1000 + + +def test_implicit_ftps_connect_cleanup_on_wrap_failure(): + from bambu_cli.protocols.ftps import ImplicitFTPS + + mock_sock = MagicMock() + mock_sock.family = 2 + mock_ctx = MagicMock() + mock_ctx.wrap_socket.side_effect = ssl.SSLError("boom") + ftp = ImplicitFTPS() + ftp.printer = SimpleNamespace(cert_fingerprint=None, insecure_tls=False) + with pytest.raises(ssl.SSLError): + ftp.connect( + "192.168.1.1", + 990, + 5, + create_connection=MagicMock(return_value=mock_sock), + ssl_context_cls=MagicMock(return_value=mock_ctx), + ) + mock_sock.close.assert_called() + + +def test_implicit_ftps_data_channel_pin_mismatch_closes(): + from bambu_cli.protocols.ftps import ImplicitFTPS + from bambu_cli.errors import BambuError + + ftp = ImplicitFTPS() + ftp.host = "192.168.1.1" + ftp.printer = SimpleNamespace(cert_fingerprint="ab" * 32) + ftp._prot_p = True + control = MagicMock(spec=ssl.SSLSocket) + ftp.sock = control + data = MagicMock() + ctx = MagicMock() + wrapped = MagicMock() + wrapped.getpeercert.return_value = b"\x00peer" + ctx.wrap_socket.return_value = wrapped + control.context = ctx + control.session = object() + with ( + patch("ftplib.FTP.ntransfercmd", return_value=(data, 10)), + patch("bambu_cli.tlspin.verify_cert_fingerprint", side_effect=ssl.SSLError("pin")), + pytest.raises(ssl.SSLError), + ): + ftp.ntransfercmd("STOR x") + wrapped.close.assert_called() + + +def test_create_raw_ftp_sim_and_real_login_failure(): + from bambu_cli.protocols.ftps import _create_raw_ftp + + sim = _create_raw_ftp(SimpleNamespace(simulation_mode=True, ip="1.2.3.4", access_code="x")) + assert hasattr(sim, "nlst") + + ftp = MagicMock() + ftp.connect.side_effect = OSError("refused") + ftp.close.side_effect = OSError("already") + with ( + patch("bambu_cli.protocols.ftps._resolve_ip", return_value="1.2.3.4"), + patch("bambu_cli.protocols.ftps.ImplicitFTPS", return_value=ftp), + pytest.raises(OSError), + ): + _create_raw_ftp(SimpleNamespace(simulation_mode=False, ip="1.2.3.4", access_code="x"), timeout=1) + + +# --------------------------------------------------------------------------- +# snapshot helpers + command edges +# --------------------------------------------------------------------------- + + +def test_snapshot_helpers_and_warn_bind(): + from bambu_cli.commands import snapshot as snap + + assert snap._utc_stamp() + assert snap._is_valid_port_number("nope") is False + assert snap._camera_port_is_valid("") is False + assert snap._camera_bind_host("1984") == "" + ctx = SimpleNamespace(settings=SimpleNamespace(camera_container_name="bambu_camera")) + snap._warn_if_running_bind_exposed(ctx, lambda *_a, **_k: (_ for _ in ()).throw(FileNotFoundError())) + bad = MagicMock(returncode=1, stdout="") + snap._warn_if_running_bind_exposed(ctx, lambda *_a, **_k: bad) + ok = MagicMock(returncode=0, stdout='{"1984/tcp":[{"HostIp":"0.0.0.0"}]}') + with patch.object(snap.logger, "warning") as warn: + snap._warn_if_running_bind_exposed(ctx, lambda *_a, **_k: ok) + warn.assert_called() + + +def test_snapshot_direct_write_oserror(tmp_path): + from bambu_cli.commands.snapshot import cmd_snapshot + from bambu_cli.context import RuntimeContext, Settings + + args = Namespace(output=str(tmp_path / "out.jpg"), json=False, unique=False) + ctx = RuntimeContext(settings=Settings(), simulation=True) + with ( + patch("bambu_cli.commands.snapshot._write_snapshot_atomic", side_effect=OSError("disk")), + pytest.raises(BambuError), + ): + cmd_snapshot(args, ctx=ctx, grab_frame=lambda _p: b"\xff\xd8\xff\xd9") + + +def test_snapshot_docker_unreachable_and_run_fail(tmp_path): + from bambu_cli.commands.snapshot import cmd_snapshot + from bambu_cli.context import RuntimeContext, Settings + + args = Namespace(output=str(tmp_path / "out.jpg"), json=False, unique=False, allow_camera_streamer=True) + ctx = RuntimeContext(settings=Settings(camera_allow_streamer=True), simulation=True) + with pytest.raises(BambuError) as missing: + cmd_snapshot(args, ctx=ctx, grab_frame=lambda _p: None, which=lambda _n: None) + assert "Docker" in str(missing.value) or missing.value.exit_code + + def run_fail(cmd, **_k): + if cmd[:2] == ["docker", "inspect"]: + return SimpleNamespace(returncode=1, stdout="") + if cmd[:2] == ["docker", "rm"]: + return SimpleNamespace(returncode=0, stdout="") + return SimpleNamespace(returncode=1, stdout=b"", stderr=b"secret 1.2.3.4 boom") + + ctx.settings.printer_ip = "1.2.3.4" + with pytest.raises(BambuError): + cmd_snapshot( + args, + ctx=ctx, + grab_frame=lambda _p: None, + which=lambda _n: "/usr/bin/docker", + subprocess_run=run_fail, + access_code_loader=lambda: "secret", + ) + + def inspect_raises(*_a, **_k): + raise FileNotFoundError("docker") + + with pytest.raises(BambuError): + cmd_snapshot( + args, + ctx=ctx, + grab_frame=lambda _p: None, + which=lambda _n: "/usr/bin/docker", + subprocess_run=inspect_raises, + ) + + +def test_snapshot_unique_with_user_output_and_error_paths(tmp_path): + from bambu_cli.commands.snapshot import cmd_snapshot + from bambu_cli.context import RuntimeContext, Settings + from bambu_cli.protocols.camera import _CameraPinMismatch + + out = tmp_path / "cam.jpg" + args = Namespace(output=str(out), json=False, unique=True) + ctx = RuntimeContext(settings=Settings(), simulation=True) + cmd_snapshot(args, ctx=ctx, grab_frame=lambda _p: b"\xff\xd8\xff\xd9", now=None) + assert list(tmp_path.glob("cam_*.jpg")) + + args2 = Namespace(output=str(tmp_path / "pin.jpg"), json=False, unique=False) + with pytest.raises(BambuError): + cmd_snapshot( + args2, + ctx=ctx, + grab_frame=lambda _p: (_ for _ in ()).throw(_CameraPinMismatch("mismatch")), + ) + + ctx.settings.cert_fingerprint = "ab" * 32 + ctx.settings.insecure_tls = False + ctx._printer = SimpleNamespace(insecure_tls=False, cert_fingerprint="ab" * 32) + + def grab_ssl(_p): + raise ssl.SSLError("pinned handshake") + + with pytest.raises(BambuError): + cmd_snapshot(args2, ctx=ctx, grab_frame=grab_ssl) + + args3 = Namespace( + output=str(tmp_path / "stream.jpg"), + json=False, + unique=False, + allow_camera_streamer=True, + ) + ctx3 = RuntimeContext(settings=Settings(camera_allow_streamer=True), simulation=True) + + def run_ok_then_urlerror(cmd, **_k): + return SimpleNamespace(returncode=0, stdout="true\n") + + with pytest.raises(BambuError): + cmd_snapshot( + args3, + ctx=ctx3, + grab_frame=lambda _p: None, + which=lambda _n: "/usr/bin/docker", + subprocess_run=run_ok_then_urlerror, + urlopen=lambda *_a, **_k: (_ for _ in ()).throw(__import__("urllib.error").URLError("down")), + ) + + def run_start_fail(cmd, **_k): + if cmd[:2] == ["docker", "inspect"]: + return SimpleNamespace(returncode=1, stdout="false") + if cmd[:2] == ["docker", "rm"]: + return SimpleNamespace(returncode=0, stdout="") + return SimpleNamespace(returncode=1, stdout=b"", stderr=b"secret 9.9.9.9 boom") + + ctx3.settings.printer_ip = "9.9.9.9" + with pytest.raises(BambuError): + cmd_snapshot( + args3, + ctx=ctx3, + grab_frame=lambda _p: None, + which=lambda _n: "/usr/bin/docker", + subprocess_run=run_start_fail, + access_code_loader=lambda: "secret", + ) + + def run_then_oserror(cmd, **_k): + return SimpleNamespace(returncode=0, stdout="true\n") + + class BoomResp: + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def read(self): + raise OSError("disk full") + + with pytest.raises(BambuError): + cmd_snapshot( + args3, + ctx=ctx3, + grab_frame=lambda _p: None, + which=lambda _n: "/usr/bin/docker", + subprocess_run=run_then_oserror, + urlopen=lambda *_a, **_k: BoomResp(), + ) + + +def test_snapshot_unique_uses_clock(tmp_path): + from bambu_cli.commands.snapshot import cmd_snapshot + from bambu_cli.context import RuntimeContext, Settings + + args = Namespace(output=None, json=False, unique=True) + ctx = RuntimeContext(settings=Settings(), simulation=True) + out_dir = tmp_path + with patch("bambu_cli.commands.snapshot._expand_path", side_effect=lambda p: str(out_dir / os.path.basename(p))): + cmd_snapshot(args, ctx=ctx, grab_frame=lambda _p: b"\xff\xd8\xff\xd9", now=None) + saved = list(out_dir.glob("printer_snapshot_*.jpg")) + assert saved + + +# --------------------------------------------------------------------------- +# wizard leftover error branches +# --------------------------------------------------------------------------- + + +def test_wizard_ipv6_and_bad_raw_address(): + from bambu_cli.setup_cmd import wizard as wizard_mod + + info = MagicMock() + info.parsed_addresses = None + info.addresses = [socket.inet_pton(socket.AF_INET6, "::1")] + assert ":" in wizard_mod._service_info_address(info) + + info2 = MagicMock() + info2.parsed_addresses = None + info2.addresses = [b"xx"] + with pytest.raises(ValueError): + wizard_mod._service_info_address(info2) + + +def test_wizard_parse_identity_model_only(): + from bambu_cli.setup_cmd import wizard as wizard_mod + + serial, model = wizard_mod._parse_mdns_printer_identity("BBLP-P1S._bblp._tcp.local.") + assert model == "P1S" + assert serial + + +def test_wizard_noninteractive_empty_env_and_missing(monkeypatch): + from bambu_cli.setup_cmd import wizard as wizard_mod + + monkeypatch.delenv("EMPTY_CODE", raising=False) + args = Namespace( + printer_ip="10.0.0.1", + serial="SN1", + access_code=None, + access_code_env="EMPTY_CODE", + access_code_file=None, + json=True, + ) + with pytest.raises(BambuError): + wizard_mod._cmd_setup_noninteractive(args) + + args2 = Namespace( + printer_ip=None, + serial=None, + access_code=None, + access_code_env=None, + access_code_file=None, + json=True, + ) + with pytest.raises(BambuError) as exc: + wizard_mod._cmd_setup_noninteractive(args2) + assert "missing" in str(exc.value).lower() or exc.value.exit_code + + +def test_wizard_noninteractive_file_errors(tmp_path, monkeypatch): + from bambu_cli.setup_cmd import wizard as wizard_mod + + missing = tmp_path / "nope" + args = Namespace( + printer_ip="10.0.0.1", + serial="SNREAL123456", + access_code=None, + access_code_env=None, + access_code_file=str(missing), + json=True, + model="P1S", + nozzle="0.4", + orca_slicer=None, + profiles_dir=None, + cert_fingerprint=None, + insecure_tls=False, + force=False, + ) + with pytest.raises(BambuError): + wizard_mod._cmd_setup_noninteractive(args) + + bad = tmp_path / "code" + bad.write_text("ACCESS_CODE\n", encoding="utf-8") + args.access_code_file = str(bad) + with pytest.raises(BambuError): + wizard_mod._cmd_setup_noninteractive(args) + + cfg = tmp_path / "config.json" + code = tmp_path / "okcode" + code.write_text("12345678", encoding="utf-8") + args.access_code = "12345678" + args.access_code_file = str(code) + args.force = False + with ( + patch.object(wizard_mod, "_config_path", return_value=str(cfg)), + patch.object(wizard_mod, "_write_setup_config", side_effect=OSError("rofs")), + pytest.raises(BambuError), + ): + wizard_mod._cmd_setup_noninteractive(args) + + +# --------------------------------------------------------------------------- +# facade freeze (Architecture A+) +# --------------------------------------------------------------------------- + + +def test_snapshot_dash_path_and_atomic_unlink(tmp_path): + from bambu_cli.commands.snapshot import _write_snapshot_atomic, cmd_snapshot + from bambu_cli.context import RuntimeContext, Settings + + args = Namespace(output="-evil.jpg", json=False, unique=False) + ctx = RuntimeContext(settings=Settings(), simulation=True) + with pytest.raises(BambuError): + cmd_snapshot(args, ctx=ctx, grab_frame=lambda _p: b"x") + + target = tmp_path / "x.jpg" + with ( + patch("os.replace", side_effect=OSError("busy")), + patch("os.unlink", side_effect=OSError("gone")), + pytest.raises(OSError), + ): + _write_snapshot_atomic(str(target), b"data") + + def inspect_then_missing(cmd, **_k): + if cmd[:2] == ["docker", "inspect"]: + return SimpleNamespace(returncode=1, stdout="") + raise FileNotFoundError("docker vanished") + + args3 = Namespace(output=str(tmp_path / "s.jpg"), json=False, unique=False, allow_camera_streamer=True) + ctx3 = RuntimeContext(settings=Settings(camera_allow_streamer=True), simulation=True) + with pytest.raises(BambuError): + cmd_snapshot( + args3, + ctx=ctx3, + grab_frame=lambda _p: None, + which=lambda _n: "/usr/bin/docker", + subprocess_run=inspect_then_missing, + access_code_loader=lambda: "x", + ) + + +def test_slice_step_fail_and_dash_outdir(tmp_path): + from bambu_cli.slicer.cmd import cmd_slice + + step = tmp_path / "part.step" + step.write_text("solid\n", encoding="utf-8") + args = Namespace( + file=str(step), + list_settings=False, + json=True, + output=None, + quality="standard", + copies=1, + ) + with ( + patch("bambu_cli.slicer.cmd._convert_step_to_stl", return_value=(None, False)), + pytest.raises(BambuError), + ): + cmd_slice(args) + + stl = tmp_path / "cube.stl" + stl.write_text("solid\n", encoding="utf-8") + args2 = Namespace( + file=str(stl), + list_settings=False, + json=True, + output="-nope", + quality="standard", + copies=1, + ) + with pytest.raises(BambuError): + cmd_slice(args2) + + +def test_download_html_page_has_no_model_link(tmp_path): + from bambu_cli.download.downloader import _cmd_download + + class Resp: + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def read(self, n=None): + return b"no files" + + def getheader(self, name): + return "text/html" if name.lower() == "content-type" else None + + def geturl(self): + return None + + class Opener: + def open(self, *a, **k): + return Resp() + + args = Namespace( + url="https://example.com/gallery", + output=str(tmp_path), + json=True, + name=None, + max_download_mb=10, + progress=False, + ) + with ( + patch("bambu_cli.download.downloader._validate_download_url_or_exit"), + patch("bambu_cli.download.downloader._reject_unsupported_download_extension"), + pytest.raises(BambuError), + ): + _cmd_download(args, opener_factory=lambda: Opener(), resolve_printables=lambda u: (u, None)) + + +def test_mqtt_facade_all_is_frozen(): + """New public names must be added to this freeze deliberately, not by accident.""" + expected = { + "TERMINAL_GCODE_STATES", + "PinningSSLContext", + "_REQUIRED_STATUS_KEYS", + "_SimMqttClient", + "_mqtt_connect", + "_printer_error_hex", + "_require_mqtt", + "_status_event", + "MqttSession", + "create_mqtt_client", + "execute_print_command", + "get_status", + "get_version", + "monitor_status", + "mqtt", + "pinning_ssl_context", + "probe_cert_fingerprint", + "send_command", + "status_is_complete", + "time", + } + assert set(mqtt_mod.__all__) == expected + + +# --------------------------------------------------------------------------- +# download / slice / preflight leftovers +# --------------------------------------------------------------------------- + + +def test_response_url_helpers(): + from bambu_cli.download.downloader import _response_url + + assert _response_url(object()) is None + assert _response_url(SimpleNamespace(geturl="not-callable")) is None + + class Boom: + def geturl(self): + raise RuntimeError("x") + + assert _response_url(Boom()) is None + assert _response_url(SimpleNamespace(geturl=lambda: 123)) is None + assert _response_url(SimpleNamespace(geturl=lambda: "https://example.com/a.stl")) == "https://example.com/a.stl" + + +def test_download_rejects_dash_outdir(): + from bambu_cli.download.downloader import _cmd_download + + args = Namespace(url="https://example.com/a.stl", output="-sneaky", json=True, name=None, max_download_mb=10) + with pytest.raises(BambuError): + _cmd_download(args) + + +def test_slice_validate_edges(tmp_path): + from bambu_cli.slicer.cmd import cmd_slice + + with pytest.raises(BambuError): + cmd_slice(Namespace(file=None, list_settings=False, json=True)) + with pytest.raises(BambuError): + cmd_slice(Namespace(file="-evil.stl", list_settings=False, json=True)) + missing = tmp_path / "nope.stl" + with pytest.raises(BambuError): + cmd_slice(Namespace(file=str(missing), list_settings=False, json=True)) + directory = tmp_path / "modeldir" + directory.mkdir() + with pytest.raises(BambuError): + cmd_slice(Namespace(file=str(directory), list_settings=False, json=True)) + + +def test_preflight_module_and_perms(tmp_path, monkeypatch): + from bambu_cli.setup_cmd import preflight as pf + + monkeypatch.setattr("importlib.util.find_spec", lambda _n: (_ for _ in ()).throw(ValueError("x"))) + assert pf._module_available("definitely_missing_mod") is False + assert pf._file_permission_check("", "secret") is None + secret = tmp_path / "code" + secret.write_text("x", encoding="utf-8") + secret.chmod(0o644) + if sys.platform == "win32": + # POSIX mode bits do not apply on Windows; the check is a no-op there + # (see the "Windows secret ACLs" residual in SECURITY.md). + assert pf._file_permission_check(str(secret), "access_code_file") is None + return + check = pf._file_permission_check(str(secret), "access_code_file") + assert check is not None + assert check["status"] == "warning" + secret.chmod(0o600) + ok = pf._file_permission_check(str(secret), "access_code_file") + assert ok is not None + assert ok["status"] == "ok" + gone = tmp_path / "missing" + assert pf._file_permission_check(str(gone), "access_code_file") is None + + +def test_snapshot_parent_dir_error(tmp_path): + from bambu_cli.commands.snapshot import cmd_snapshot + from bambu_cli.context import RuntimeContext, Settings + + args = Namespace(output=str(tmp_path / "nope" / "out.jpg"), json=False, unique=False) + ctx = RuntimeContext(settings=Settings(), simulation=True) + with ( + patch("bambu_cli.commands.snapshot._ensure_parent_dir", side_effect=BambuError("no dir", exit_code=3)), + pytest.raises(BambuError), + ): + cmd_snapshot(args, ctx=ctx, grab_frame=lambda _p: b"x") + + +def test_monitor_progress_stop_raises(): + printer = BambuPrinter("1.2.3.4", "SN", "code") + client = MagicMock() + + class Progress: + def update(self, *a, **k): + return None + + def stop(self): + raise RuntimeError("x") + + def loop_start(): + msg = MagicMock() + msg.payload = json.dumps({"print": {"gcode_state": "FINISH", "mc_percent": 100}}).encode() + client.on_message(client, {"progress": Progress(), "task_id": 1}, msg) + + client.loop_start.side_effect = loop_start + args = Namespace(json=False) + with ( + patch.object(mqtt_mod, "create_mqtt_client", return_value=client), + patch.object(mqtt_mod, "_mqtt_connect"), + patch("sys.stdout.isatty", return_value=False), + ): + mqtt_mod.monitor_status(args, printer) diff --git a/tests/test_context.py b/tests/test_context.py index 39f4ba6..3b1c281 100644 --- a/tests/test_context.py +++ b/tests/test_context.py @@ -113,6 +113,7 @@ def test_runtime_context_printer_simulation_mode(): assert printer.serial == "SN1" assert printer.access_code == "" assert printer.simulation_mode is True + assert printer.mqtt_port == 8883 # cached assert ctx.printer() is printer @@ -125,6 +126,49 @@ def test_runtime_context_printer_non_simulation_uses_load_access_code(): mock_load.assert_called_once() assert printer.access_code == "secretcode" assert printer.cert_fingerprint == "aabb" + assert printer.mqtt_port == 8883 + + +def test_runtime_context_printer_honors_configured_mqtt_port(): + settings = context.Settings(printer_ip="1.2.3.4", serial="SN1", mqtt_port=1883) + ctx = context.RuntimeContext(settings=settings, simulation=True) + assert ctx.printer().mqtt_port == 1883 + + +def test_runtime_context_printer_uses_installed_factory(): + """RuntimeContext.printer() must go through set_printer_factory, not import printer.""" + import bambu_cli.printer # noqa: F401 — register the default factory + + previous = context.get_printer_factory() + sentinel = object() + seen: list[object] = [] + + def factory(ctx): + seen.append(ctx) + return sentinel + + try: + context.set_printer_factory(factory) + ctx = context.RuntimeContext() + assert ctx.printer() is sentinel + assert ctx.printer() is sentinel # cached + assert seen == [ctx] + finally: + context.set_printer_factory(previous) + + +def test_runtime_context_printer_requires_factory(): + previous = context.get_printer_factory() + try: + context.set_printer_factory(None) + ctx = context.RuntimeContext() + try: + ctx.printer() + raise AssertionError("expected RuntimeError") + except RuntimeError as exc: + assert "factory" in str(exc).lower() + finally: + context.set_printer_factory(previous) def test_get_current_lazy_builds_and_set_current_overrides(): diff --git a/tests/test_docs_consistency.py b/tests/test_docs_consistency.py index 03d1758..8f18cb6 100644 --- a/tests/test_docs_consistency.py +++ b/tests/test_docs_consistency.py @@ -145,6 +145,15 @@ def _collect_test_count() -> int: ) +def _parse_doc_cov_percent(text: str) -> float | None: + """Return the first bold measured coverage percentage like **88.9%**.""" + for m in re.finditer(r"\*\*(\d{2}\.\d)%\*\*", text): + val = float(m.group(1)) + if 80.0 <= val <= 99.9: + return val + return None + + def test_coverage_floor_matches_ci(): """The coverage floor cited in docs must exactly match --cov-fail-under in ci.yml. @@ -164,6 +173,23 @@ def test_coverage_floor_matches_ci(): ) +def test_documented_coverage_percent_at_or_above_floor(): + """Docs must cite a measured coverage % that is not below the CI floor. + + Prevents the scoreboard from silently rotting to a number CI would reject. + """ + ci_floor = float(_parse_ci_cov_floor()) + roadmap = (ROOT / "docs" / "quality-roadmap.md").read_text(encoding="utf-8") + backlog = (ROOT / "docs" / "test-backlog.md").read_text(encoding="utf-8") + for doc, text in [("quality-roadmap.md", roadmap), ("test-backlog.md", backlog)]: + percent = _parse_doc_cov_percent(text) + assert percent is not None, f"{doc}: could not find a bold coverage percent like **88.9%**" + assert percent >= ci_floor, ( + f"{doc} cites measured coverage {percent}% but ci.yml floor is {ci_floor:g}. " + "Update the snapshot or the floor." + ) + + def test_documented_test_count_within_tolerance(): """Documented test count must be within 5% of the actual collected count. diff --git a/tests/test_jsonio.py b/tests/test_jsonio.py index f9a3905..f70e432 100644 --- a/tests/test_jsonio.py +++ b/tests/test_jsonio.py @@ -1,7 +1,6 @@ """jsonio redaction, home-path display, AMS sentinels, and ZIP extract edges.""" import argparse -import os import zipfile from unittest.mock import MagicMock @@ -11,6 +10,7 @@ # jsonio.redact_url_credentials — scheme-relative URLs # --------------------------------------------------------------------------- + def test_redact_scheme_relative_url_strips_userinfo(): from bambu_cli.jsonio import redact_url_credentials @@ -20,6 +20,7 @@ def test_redact_scheme_relative_url_strips_userinfo(): assert redact_url_credentials("//user:pass" + at + "host.com/x") == "//host.com/x" assert redact_url_credentials("//u:p" + at + "host.com:8443/a?b=c") == "//host.com:8443/a?b=c" + def test_redact_scheme_relative_url_ipv6_userinfo(): from bambu_cli.jsonio import redact_url_credentials @@ -27,6 +28,7 @@ def test_redact_scheme_relative_url_ipv6_userinfo(): # netloc-only IPv6 with userinfo: host must stay bracketed, creds gone. assert redact_url_credentials("//user:pass" + at + "[::1]:990/x") == "//[::1]:990/x" + def test_redact_preserves_existing_schemeless_and_full_url_behavior(): from bambu_cli.jsonio import redact_url_credentials @@ -39,6 +41,26 @@ def test_redact_preserves_existing_schemeless_and_full_url_behavior(): assert redact_url_credentials("/home/x" + at + "y") == "/home/x" + at + "y" assert redact_url_credentials("no-at-sign") == "no-at-sign" + +def test_looks_like_schemeless_rejects_spaces_and_backslashes(): + from bambu_cli.jsonio import looks_like_schemeless_credential_url + + at = "@" + assert looks_like_schemeless_credential_url("user:pass" + at + "host.com") is True + assert looks_like_schemeless_credential_url("user:pass " + at + "host.com") is False + assert looks_like_schemeless_credential_url("user:pass" + at + "host\\name.com") is False + assert looks_like_schemeless_credential_url("") is False + + +def test_redact_invalid_port_does_not_raise(): + from bambu_cli.jsonio import redact_url_credentials + + at = "@" + # urllib raises ValueError on .port when the port token is not an int. + assert redact_url_credentials("https://user:pass" + at + "host.com:notaport/x") == ("https://host.com/x") + assert redact_url_credentials("//user:pass" + at + "host.com:notaport/x") == "//host.com/x" + + def test_emit_json_uses_jsonio_redactor(capsys): """emit_json must strip userinfo, not the weaker ***@ placeholder.""" from bambu_cli import utils @@ -51,10 +73,12 @@ def test_emit_json_uses_jsonio_redactor(capsys): assert "***@" not in payload assert "https://host.com/x.stl" in payload + # --------------------------------------------------------------------------- # utils._display_path — home-prefix separator boundary # --------------------------------------------------------------------------- + def test_display_path_requires_separator_boundary(monkeypatch): import bambu_cli.utils as utils @@ -66,10 +90,12 @@ def test_display_path_requires_separator_boundary(monkeypatch): assert utils._display_path("/home/alice/model.stl") == "~/model.stl" assert utils._display_path("/home/alice") == "~" + # --------------------------------------------------------------------------- # utils._resolve_ip — do not cache failures # --------------------------------------------------------------------------- + def test_resolve_ip_does_not_cache_failure(monkeypatch): import bambu_cli.utils as utils @@ -96,13 +122,16 @@ def _ok(host, *a, **k): assert utils._RESOLVE_IP_CACHE.get("printer.local") == "10.0.0.5" utils._RESOLVE_IP_CACHE.clear() + # --------------------------------------------------------------------------- # ams.parse_ams — external-spool sentinel + wizard active-tray selection # --------------------------------------------------------------------------- + def _ams_status(tray_now, units): return {"ams": {"tray_now": str(tray_now), "ams": units}} + def test_parse_ams_external_spool_sentinel_not_active(): from bambu_cli.ams import parse_ams @@ -112,6 +141,7 @@ def test_parse_ams_external_spool_sentinel_not_active(): assert parsed["active_tray"] is None assert all(not t["active"] for u in parsed["units"] for t in u["trays"]) + def _patch_ams_status(monkeypatch, status): """Make _read_loaded_ams_material see ``status`` from the printer.""" from bambu_cli.context import RuntimeContext @@ -122,6 +152,7 @@ def _patch_ams_status(monkeypatch, status): fake_ctx.printer.return_value = fake_printer monkeypatch.setattr(RuntimeContext, "for_request", classmethod(lambda cls, args: fake_ctx)) + def test_wizard_ams_material_multi_unit_picks_active_not_earlier_unit(monkeypatch): from bambu_cli.interactive.session import _read_loaded_ams_material @@ -138,6 +169,7 @@ def test_wizard_ams_material_multi_unit_picks_active_not_earlier_unit(monkeypatc # Before the fix, the earlier-unit fallback returned PETG. assert _read_loaded_ams_material(argparse.Namespace()) == "PLA" + def test_wizard_ams_material_external_spool_sentinel_no_false_active(monkeypatch): from bambu_cli.interactive.session import _read_loaded_ams_material @@ -150,10 +182,12 @@ def test_wizard_ams_material_external_spool_sentinel_no_false_active(monkeypatch _patch_ams_status(monkeypatch, status) assert _read_loaded_ams_material(argparse.Namespace()) == "PLA" + # --------------------------------------------------------------------------- # download.extract._extract_zip_model — encrypted / Deflate64 -> ValueError # --------------------------------------------------------------------------- + def test_extract_encrypted_zip_raises_valueerror(tmp_path): from bambu_cli.download.extract import _extract_zip_model @@ -201,6 +235,7 @@ def open(self, *a, **k): finally: extract.zipfile.ZipFile = real_zipfile + def test_extract_deflate64_raises_valueerror(tmp_path): from bambu_cli.download.extract import _extract_zip_model diff --git a/tests/test_mqtt_print_and_setup.py b/tests/test_mqtt_print_and_setup.py index af9e6d2..6f8b109 100644 --- a/tests/test_mqtt_print_and_setup.py +++ b/tests/test_mqtt_print_and_setup.py @@ -226,6 +226,34 @@ def loop_start(): mqtt_mod.monitor_status(args, printer) lines = [ln for ln in capsys.readouterr().out.splitlines() if ln.strip()] assert any("terminal" in ln or "FINISH" in ln for ln in lines) + client.loop_stop.assert_called_once() + client.disconnect.assert_called_once() + + +def test_monitor_disconnects_even_if_loop_stop_raises(): + printer = _test_printer(simulation_mode=False) + client = MagicMock() + + def connect(*a, **k): + if client.on_connect: + client.on_connect(client, None, None, 0) + + def loop_start(): + if client.on_message: + msg = MagicMock() + msg.payload = json.dumps({"print": {"gcode_state": "FINISH", "mc_percent": 100}}).encode() + client.on_message(client, {}, msg) + + client.connect.side_effect = connect + client.loop_start.side_effect = loop_start + client.loop_stop.side_effect = RuntimeError("loop already stopped") + args = Namespace(json=True) + with ( + patch.object(mqtt_mod, "create_mqtt_client", return_value=client), + patch.object(mqtt_mod, "_mqtt_connect"), + ): + mqtt_mod.monitor_status(args, printer) + client.disconnect.assert_called_once() def test_monitor_merges_deltas_into_streamed_state(capsys): """A delta must not stream as gcode_state=UNKNOWN at 0% — it updates the merged state.""" diff --git a/tests/test_mqtt_session.py b/tests/test_mqtt_session.py index fbdc0f4..c196485 100644 --- a/tests/test_mqtt_session.py +++ b/tests/test_mqtt_session.py @@ -231,6 +231,36 @@ def factory(_printer): printer.release_mqtt() +def test_session_returns_cached_complete_when_pushall_silent(): + """Background reports already filled state; a silent pushall must not raise.""" + + def factory(_printer): + return FakeBrokerClient(status_replies=[None]) + + printer = _printer() + session = _held(printer, factory) + assert session.ensure_connected(1) + session._print_state.update(_FULL) + result = get_status(printer, timeout=0.01, retries=0) + assert result is not None + assert result["gcode_state"] == "IDLE" + assert result["mc_percent"] == 0 + printer.release_mqtt() + + +def test_session_second_status_uses_cache_if_pushall_times_out(): + def factory(_printer): + return FakeBrokerClient(status_replies=[{"print": dict(_FULL)}, None, None]) + + printer = _printer() + _held(printer, factory) + first = get_status(printer, timeout=0.01, retries=0) + second = get_status(printer, timeout=0.01, retries=0) + assert first is not None and first["gcode_state"] == "IDLE" + assert second is not None and second["gcode_state"] == "IDLE" + printer.release_mqtt() + + def test_session_liveness_accepts_partial(): def factory(_printer): return FakeBrokerClient(status_replies=[{"print": {"wifi_signal": "-40dBm"}}]) @@ -254,6 +284,27 @@ def factory(_printer): printer.release_mqtt() +def test_session_get_version_timeout_returns_none(): + def factory(_printer): + return FakeBrokerClient(version_reply=None) + + printer = _printer() + _held(printer, factory) + assert get_version(printer, timeout=0.01, retries=0) is None + printer.release_mqtt() + + +def test_ready_snapshot_rejects_empty_and_incomplete(): + printer = _printer() + session = _held(printer, lambda _p: FakeBrokerClient(status_replies=[None])) + assert session.ensure_connected(1) + assert session._ready_snapshot(require_complete=True) is None + session._print_state.update({"wifi_signal": "-40dBm"}) + assert session._ready_snapshot(require_complete=True) is None + assert session._ready_snapshot(require_complete=False) == {"wifi_signal": "-40dBm"} + printer.release_mqtt() + + def test_connect_rc_failure_returns_none(): def factory(_printer): return FakeBrokerClient(connect_rc=4) diff --git a/tests/test_primitives_helpers.py b/tests/test_primitives_helpers.py new file mode 100644 index 0000000..7689ae5 --- /dev/null +++ b/tests/test_primitives_helpers.py @@ -0,0 +1,107 @@ +"""Unit tests for rank-10/20 helpers that were only hit indirectly.""" + +from __future__ import annotations + +from argparse import Namespace +from types import SimpleNamespace + +from bambu_cli.argutils import exit_code_from_system_exit, namespace_get, setup_args_provided +from bambu_cli.constants import EXIT_COMMAND_ERROR, EXIT_SUCCESS +from bambu_cli.errors import BambuError +from bambu_cli.job.support import _exit_code_from_error, _last_error_for +from bambu_cli.logging_utils import patched_logger, reset_logger, set_logger +from bambu_cli.paths import display_path, exception_for_message, expand_path, path_for_message + + +def test_namespace_get_reads_vars_and_falls_back_on_typeerror(): + assert namespace_get(Namespace(serial="SN"), "serial") == "SN" + assert namespace_get(Namespace(), "missing", "d") == "d" + + class _NoVars: + __slots__ = () + + assert namespace_get(_NoVars(), "serial", "fallback") == "fallback" + + +def test_exit_code_from_system_exit_normalizes_shapes(): + assert exit_code_from_system_exit(SimpleNamespace(exit_code=6)) == 6 + assert exit_code_from_system_exit(SimpleNamespace(code=3)) == 3 + assert exit_code_from_system_exit(SimpleNamespace(code=None)) == EXIT_SUCCESS + assert exit_code_from_system_exit(SimpleNamespace(code="nope")) == EXIT_COMMAND_ERROR + + +def test_setup_args_provided_any_setup_field(): + assert setup_args_provided(Namespace()) is False + assert setup_args_provided(Namespace(printer_ip="1.2.3.4")) is True + assert setup_args_provided(Namespace(model="P1S")) is True + + +def test_expand_and_display_path_none_and_home(monkeypatch, tmp_path): + assert expand_path(None) is None + assert display_path(None) is None + assert path_for_message(None) is None + + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + import bambu_cli.paths as paths_mod + + paths_mod._HOME_DIR = str(home) + paths_mod._NORM_HOME_DIR = None + assert display_path(str(home)) == "~" + nested = home / "models" / "cube.stl" + assert display_path(str(nested)).startswith("~") + + +def test_exception_for_message_compacts_filename_attrs(tmp_path, monkeypatch): + home = tmp_path / "home" + home.mkdir() + import bambu_cli.paths as paths_mod + + paths_mod._HOME_DIR = str(home) + paths_mod._NORM_HOME_DIR = None + target = home / "secret.stl" + err = OSError("cannot open") + err.filename = str(target) + compacted = exception_for_message(err) + assert "secret.stl" in compacted + assert str(home) not in compacted or compacted.startswith("~") or "~" in compacted + + +def test_job_support_exit_code_and_last_error(): + assert _exit_code_from_error(SimpleNamespace(exit_code=6)) == 6 + assert _exit_code_from_error(SimpleNamespace(code=2)) == 2 + assert _exit_code_from_error(SimpleNamespace(code=None)) == 0 + assert _exit_code_from_error(SimpleNamespace(code="x")) == EXIT_COMMAND_ERROR + + exc = BambuError("partial", failed_step="status", detail={"missing_keys": ["gcode_state"]}) + payload = _last_error_for("status", exc=exc) + assert payload["command"] == "status" + assert payload["failed_step"] == "status" + assert payload["detail"]["missing_keys"] == ["gcode_state"] + + +def test_logger_set_reset_and_patched_roundtrip(): + class _Backend: + def __init__(self): + self.seen = [] + + def error(self, message, **kwargs): + self.seen.append(message) + + custom = _Backend() + set_logger(custom) + try: + from bambu_cli.logging_utils import logger + + logger.error("one") + assert custom.seen == ["one"] + finally: + reset_logger() + + with patched_logger(custom) as installed: + assert installed is custom + from bambu_cli.logging_utils import logger as proxy + + proxy.error("two") + assert custom.seen[-1] == "two" diff --git a/tests/test_protocol_clients.py b/tests/test_protocol_clients.py index 7a19b6b..b99b0f0 100644 --- a/tests/test_protocol_clients.py +++ b/tests/test_protocol_clients.py @@ -131,6 +131,23 @@ def side_effect_connect(host, port, keepalive): self.assertFalse(result) mock_logger.error.assert_called_with("Connection failed: rc=5") + def test_send_command_tears_down_if_loop_start_raises(self): + from bambu_cli.protocols.mqtt import send_command + + mock_client = MagicMock() + mock_client.connect.side_effect = lambda *a, **k: None + mock_client.loop_start.side_effect = OSError("loop failed") + + result = send_command( + _test_printer(), + '{"test": "payload"}', + retries=0, + client_factory=lambda p, *a, **k: mock_client, + ) + self.assertFalse(result) + mock_client.loop_stop.assert_called() + mock_client.disconnect.assert_called() + import socket @@ -169,6 +186,24 @@ def test_create_raw_ftp_connect_failure(self, mock_implicit_ftps): mock_ftp_instance.connect.assert_called_once_with("192.168.1.100", 990, timeout=60) mock_ftp_instance.login.assert_not_called() mock_ftp_instance.prot_p.assert_not_called() + mock_ftp_instance.close.assert_called_once() + + @patch("bambu_cli.protocols.ftps.ImplicitFTPS") + def test_create_raw_ftp_login_failure_closes_socket(self, mock_implicit_ftps): + from bambu_cli.protocols.ftps import _create_raw_ftp + + mock_ftp_instance = MagicMock() + mock_implicit_ftps.return_value = mock_ftp_instance + mock_ftp_instance.login.side_effect = OSError("530 Login incorrect") + printer = _test_printer(ip="192.168.1.100", access_code="bad") + + with self.assertRaises(OSError) as context: + _create_raw_ftp(printer) + + self.assertEqual(str(context.exception), "530 Login incorrect") + mock_ftp_instance.connect.assert_called_once() + mock_ftp_instance.close.assert_called_once() + mock_ftp_instance.prot_p.assert_not_called() class TestCreateMqttClient(unittest.TestCase): @@ -211,7 +246,7 @@ def test_create_mqtt_client_insecure(self, mock_mqtt_client): class TestMqttConnectTimeout(unittest.TestCase): - def test_mqtt_connect_honors_configured_timeout_and_restores_socket_default(self): + def test_mqtt_connect_sets_client_timeout_without_mutating_socket_default(self): import socket as socket_mod from bambu_cli.protocols import mqtt_tls @@ -223,15 +258,76 @@ def test_mqtt_connect_honors_configured_timeout_and_restores_socket_default(self printer.mqtt_timeout = 30.0 before = socket_mod.getdefaulttimeout() - with patch.object(mqtt_tls, "_resolve_ip", return_value="192.168.1.5"): + with ( + patch.object(mqtt_tls, "_resolve_ip", return_value="192.168.1.5"), + patch.object(mqtt_tls.socket, "setdefaulttimeout") as set_default, + ): mqtt_tls._mqtt_connect(printer, client) - # paho's own connect cap is raised to the configured timeout... self.assertEqual(client._connect_timeout, 30.0) client.connect.assert_called_once_with("192.168.1.5", 8883, keepalive=10) - # ...and the process-wide socket default is left untouched afterwards. + set_default.assert_not_called() self.assertEqual(socket_mod.getdefaulttimeout(), before) + def test_mqtt_port_rejects_unusable_values(self): + from bambu_cli.protocols.mqtt_tls import _mqtt_port + + class _P: + def __init__(self, mqtt_port): + self.mqtt_port = mqtt_port + + self.assertEqual(_mqtt_port(_P(1883)), 1883) + self.assertEqual(_mqtt_port(_P("1883")), 1883) + self.assertEqual(_mqtt_port(_P(0)), 8883) + self.assertEqual(_mqtt_port(_P(-1)), 8883) + self.assertEqual(_mqtt_port(_P(70000)), 8883) + self.assertEqual(_mqtt_port(_P("nope")), 8883) + self.assertEqual(_mqtt_port(_P(True)), 8883) + self.assertEqual(_mqtt_port(_P(None)), 8883) + self.assertEqual(_mqtt_port(object()), 8883) + + def test_mqtt_connect_uses_configured_mqtt_port(self): + from bambu_cli.protocols import mqtt_tls + + client = MagicMock() + client._connect_timeout = 5.0 + printer = MagicMock() + printer.ip = "192.168.1.5" + printer.mqtt_timeout = 10.0 + printer.mqtt_port = 1883 + with patch.object(mqtt_tls, "_resolve_ip", return_value="192.168.1.5"): + mqtt_tls._mqtt_connect(printer, client) + client.connect.assert_called_once_with("192.168.1.5", 1883, keepalive=10) + + def test_mqtt_connect_uses_paho_public_connect_timeout(self): + from bambu_cli.protocols import mqtt_tls + + class _PahoLike: + def __init__(self): + self._connect_timeout = 5.0 + self.connected = None + + @property + def connect_timeout(self): + return self._connect_timeout + + @connect_timeout.setter + def connect_timeout(self, value): + self._connect_timeout = value + + def connect(self, host, port, keepalive=10): + self.connected = (host, port, keepalive) + + client = _PahoLike() + printer = MagicMock() + printer.ip = "192.168.1.5" + printer.mqtt_timeout = 30.0 + with patch.object(mqtt_tls, "_resolve_ip", return_value="192.168.1.5"): + mqtt_tls._mqtt_connect(printer, client) + + self.assertEqual(client.connect_timeout, 30.0) + self.assertEqual(client.connected, ("192.168.1.5", 8883, 10)) + if __name__ == "__main__": unittest.main()