mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-04 05:01:37 +02:00
fix(archives): MQTT-derived filament type/color on fallback archives (#1533)
When the source .3mf can't be downloaded at print start (P1S/A1/P2S
firmwares lock the file mid-print), main.py creates a fallback
PrintArchive with file_path="" and every filament field NULL — even
though the MQTT payload already has the AMS state and the slicer's
slot-per-print-filament mapping (data["ams"]["ams"] and
data["ams_mapping"]).
New _extract_filament_data_from_mqtt(data, ams_mapping) builds a
{global_tray_id: (type, color)} map from the AMS units, then narrows
to slots referenced by ams_mapping (slicer order preserved, -1 VT-tray
sentinels skipped) or falls back to every loaded slot when no mapping
is present. Returns comma-separated filament_type and filament_color
matching the 3MF-extraction shape, so the inventory page, Quick Stats
rollup, and len(filament_type.split(",")) per-print count behave
identically for fallback rows.
The constructor at the fallback site now passes the resulting values
into the PrintArchive row.
This does NOT recover per-filament gram usage — that needs the .3mf's
slice_info.config or a deeper layer-delta integration via usage_tracker.
The reporter (maker-space lead evaluating Bambuddy partly for AMS
expansion planning) asked specifically for "the number of filaments
used", which is what this gives them.
15 unit tests cover empty/malformed payloads, the no-mapping path,
mapping filtering and reordering, VT-tray sentinels, dual-AMS global
ids, column-limit truncation, and defensive garbage handling.
This commit is contained in:
maziggy
@@ -29,6 +29,7 @@ All notable changes to Bambuddy will be documented in this file.
|
||||
- **Trivy DS-0026 (`Dockerfile.test` missing HEALTHCHECK): silenced via `HEALTHCHECK NONE`** — The test image runs `pytest` and exits; there is no long-running service to probe, so any HEALTHCHECK we added would be cargo-cult noise. `HEALTHCHECK NONE` is the documented Docker directive to explicitly opt out of any inherited healthcheck and is the way Trivy expects projects to signal "this image is not a service." Closes code-scanning alert #813.
|
||||
|
||||
### Fixed
|
||||
- **Fallback archives now carry MQTT-derived filament type + colour when the 3MF can't be downloaded (#1533, reported by @JmanB52D)** — Reporter (lead of a maker-space 3D Fab area) was evaluating Bambuddy partly to count filaments per print for AMS expansion planning; print log was showing "—" in the filament column for every job. Trace: a P2S in VP proxy mode where the slicer's .3mf upload lands on the real printer's SD card, then the printer locks the file mid-print and refuses every FTP read (the existing fallback-archive code path in `main.py:2596`, originally added for P1S/A1 printers, anticipates this: *"FTP has file size limitations"* — same effective behaviour on P2S). The user log shows ~12 FTP candidate paths attempted on every print start, every one returning 550, then directory listings on `/cache /model /data /data/Metadata` also returning 550, then the fallback archive being created with `file_path=""` and **every filament column NULL** — even though the MQTT print-start payload already had the AMS state and the slicer's slot-per-print-filament mapping sitting in `data["ams"]["ams"]` / `data["ams_mapping"]`. **Fix**: new `_extract_filament_data_from_mqtt(data, ams_mapping)` helper in `backend/app/main.py` (placed next to the existing `_get_start_ams_mapping`) walks `data["ams"]["ams"][*].tray[*]` to build a global-tray-id → (tray_type, tray_color) map, then narrows to slots referenced by `ams_mapping` if present (slicer order preserved; -1 entries for VT-tray skipped), or falls back to every loaded slot otherwise. Output is a comma-separated `filament_type` + `filament_color` in the same shape the 3MF extractor produces — so the inventory page, Quick Stats filament rollup, and `len(filament_type.split(','))` per-print count all light up identically for fallback rows. Truncated to the model's column limits (50 / 200). Defensive against malformed MQTT shapes (non-dict entries, non-int ids, missing fields) since this runs in the print-start hot path and a raise would break print logging entirely. The fallback `PrintArchive(...)` constructor now passes `filament_type=` / `filament_color=` from the helper. **What this is NOT**: not per-filament gram usage (that needs the 3MF's `slice_info.config` or a deep AMS layer-delta integration via `usage_tracker`) — only types and colours. The user explicitly asked for "the number of filaments used to know if or when we need to expand AMS units", which is exactly what this gives them (`SELECT COUNT(DISTINCT split(filament_type, ',')) ...` or the existing inventory count surfaces). A separate, larger piece of work to capture the .3mf in VP proxy mode at upload time (by sniffing FTP STOR in `tcp_proxy.py`) is the real long-term fix for any user who wants full 3MF-derived archive metadata in proxy mode; it's not bundled here. **Tests**: 15 in `test_fallback_archive_mqtt_filament.py` (`backend/tests/unit/`) covering: empty / malformed / no-loaded-slot payloads return `{}`; the no-mapping path lists every loaded slot in ascending global-id order with colours uppercased; an `ams_mapping` filters to and reorders by the slicer's order; VT-tray sentinels (`-1`) are filtered; dual-AMS layouts resolve `unit*4 + tray` correctly across units; a mapping pointing at unknown slots falls through to the known subset, but an entirely-unknown mapping returns `{}` rather than misreporting from the all-slots fallback; both column-limit truncations enforced; missing-colour-but-present-type emits `filament_type` only; defensive against non-dict/non-int garbage in the AMS list without raising. Existing 22 print-start unit tests untouched and green. Backend ruff clean.
|
||||
- **SpoolBuddy: Tare status banner no longer sits at "Waiting for device..." forever (#1536, reported by @flom89)** — On the SpoolBuddy kiosk's Settings → Scale (Waage) tab, pressing TARE wrote the "Tare command sent. Waiting for device..." banner but had no mechanism to resolve it. The daemon writes back through `POST /spoolbuddy/devices/{id}/calibration/set-tare` (which stamps `tare_offset` + `last_calibrated_at` on the device row), the device list query already polls every 10 s, but `handleTare` in `frontend/src/pages/spoolbuddy/SpoolBuddySettingsPage.tsx` was set-and-forget — the banner persisted indefinitely. The "Calibration complete!" banner on the full calibration flow had the same shape and stayed forever too. **Fix**: a completion watcher that snapshots `device.last_calibrated_at` when TARE is pressed, sets an `awaitingTareSince` state, invalidates the device-list query every 1 s while that state is active (so detection responds within ~1 s instead of waiting on the 10 s background poll), and when `last_calibrated_at` advances past the snapshot flips the banner to "Tare complete!" with a 3 s auto-dismiss timer. A 15 s timeout on the watcher fails open to "Tare timed out — is the SpoolBuddy daemon running?" so a dead daemon doesn't leave the user staring at the spinner. The Calibration-complete success banner and the calibration-failed error banner now share the same auto-dismiss helper (3 s success, 5 s error). All timers are owned by a `useRef` that cleans up on unmount; pressing TARE while a previous dismiss is queued cancels the old timer. **i18n**: two new keys (`spoolbuddy.settings.tareComplete`, `spoolbuddy.settings.tareTimedOut`) translated into all 9 locales (de/es/fr/it/ja/pt-BR/zh-CN/zh-TW + en) per [[feedback_translate_dont_fallback]] — no English fallbacks. Parity script passes at 4997 keys × 9 locales. Frontend build clean.
|
||||
- **ntfy notifications: honest User-Agent + actionable error when the server is behind a Cloudflare challenge (#1534, reported by @apizz)** — Reporter pointed an ntfy server behind a Cloudflare Tunnel at Bambuddy and got `HTTP 403: <!DOCTYPE html>...Just a moment...` on every Test click. They reproduced the same response with a plain `curl -H "Authorization: Bearer <token>" -d "test" https://ntfy.example/<topic>` — confirming the 403 originates from Cloudflare's JS challenge intercept (Bot Fight Mode / "Under Attack" mode), not from Bambuddy or ntfy. Cloudflare returns its interstitial HTML to any non-browser client at the edge, so the request never reaches the user's ntfy backend at all. Bambuddy can't solve a JS challenge from a backend — the only real fix is on the user's Cloudflare side (a security-skip rule for the hostname/path, disabling Bot Fight Mode for that hostname, or fronting the server with Cloudflare Access using a service token). Two improvements shipped to make this footgun self-diagnosable for the next user who hits it. **(1) Honest User-Agent on the notification HTTP client.** `backend/app/services/notification_service.py` was the one outbound httpx client in the codebase that didn't set the project-standard `Bambuddy/1.0 (+https://github.com/maziggy/bambuddy)` UA — it leaked `python-httpx/<version>` instead. Brings it in line with `bambu_cloud` / `makerworld` / `firmware_check` / `inventory` (all unified during the May 2026 compliance pass) and makes Bambuddy a more obvious citizen to upstream WAFs and proxy operators. Won't defeat Cloudflare's JS challenge (the user's curl test proves CF blocks regardless of UA) but it's a consistency / hygiene fix with no regression risk. **(2) Cloudflare-challenge detection on the ntfy error path.** New `_looks_like_cloudflare_challenge(response)` helper checks the response shape (`Server: cloudflare` or `cf-mitigated` header, or `<!DOCTYPE html>...Just a moment...` body). When a 403/non-success response matches, the error returned to the UI now reads: *"HTTP 403 — ntfy server is behind a Cloudflare challenge. Bambuddy was served the JS challenge page instead of reaching ntfy. Cloudflare cannot be solved from a backend; add a Cloudflare security-skip rule for this hostname, disable Bot Fight Mode, or front the server with Cloudflare Access using a service token. (#1534)"* — actionable, points at the real fix, removes the raw HTML dump. A regular 403 (e.g. ntfy auth failure with a plain `forbidden: invalid auth token` body) still surfaces the original body so genuine auth errors stay debuggable; the interceptor only fires on the Cloudflare shape. **Tests**: 3 new in `TestNtfyOutbound` in `test_notification_service.py` — (a) the lazy-constructed httpx client carries the honest UA header on first use; (b) a 403 with `Server: cloudflare` + `Just a moment...` body produces the actionable error and does not echo `<!DOCTYPE` to the user; (c) a 403 with a plain text auth-failure body keeps the original `HTTP 403: forbidden: invalid auth token` so we don't hide real errors. 110/110 in the notification suites green under `pytest -n 30`. Backend ruff clean.
|
||||
- **Source-3MF upload on "fallback" archives no longer crashes with HTTP 500 (and stops orphaning files outside the data volume) (#1531, reported by @d3nn3s08)** — When MQTT reports a print start but Bambuddy never saw the source 3MF (cloud-initiated prints, Bambu Handy, prints already on the printer's SD card when Bambuddy connected), `main.py:2596` creates a "fallback" `PrintArchive` row with `file_path=""`. The two `Archives → Source 3MF Upload` routes computed the destination directory as `(settings.base_dir / archive.file_path).parent / "source"` — which on a fallback row collapsed to `Path('/app/data') / '' = Path('/app/data')`, whose `.parent` is `Path('/app')`, sending the upload to `/app/source/<filename>.3mf`. The file was physically written there (a path outside the user's mounted data volume — orphaned on container restart) and only the *final* `source_path.relative_to(settings.base_dir)` raised, so every retry left another orphan. Affected reporter is on a QNAP Docker host with the standard `/app/data` mount; both maintainer and triage initially diagnosed it as a Docker volume misconfiguration, but the traceback shows the bug is purely on Bambuddy's side — the user's setup was correct. **Fix**: new private helper `_resolve_source_3mf_path(archive, source_filename)` in `backend/app/api/routes/archives.py` centralises the destination computation. Normal archives still nest the source under `<archive_file_dir>/source/<filename>`. Fallback archives (empty `file_path`) now land under `<base_dir>/archive/no_source/<archive_id>/<filename>` instead — a deterministic, addressable location that stays inside the data volume, and the existing read sites (`download_source_3mf`, `download_source_3mf_by_filename`, the slicer-token routes, `delete_source_3mf`) all continue to work because they read back via `settings.base_dir / archive.source_3mf_path`. The helper also defensively asserts the resolved directory is inside `base_dir.resolve()` regardless of where it came from, so a row corrupted by an old import or a manual SQL edit fails with a clear 500 message ("Archive N resolves to a path outside the data directory; cannot attach source.") instead of silently writing outside the volume. Both upload sites (`upload_source_3mf` and `upload_source_3mf_by_name`, the slicer-post-processing endpoint) now route through the helper, so neither can independently drift back into the bug. **Tests**: 2 new in `TestUploadSourceThreeMF` in `backend/tests/integration/test_archives_api.py` — (a) `test_fallback_archive_source_upload_lands_under_base_dir` creates an archive with `file_path=""`, uploads a minimal valid 3MF, asserts 200 status, that the returned `source_3mf_path` is relative (not `/app/source/...`), that the file physically exists under the patched `base_dir`, and that the path is the deterministic fallback location keyed off `archive.id`; (b) `test_normal_archive_source_upload_unchanged` is the same flow against an archive with a populated `file_path`, asserting the existing `archives/test/source/<filename>.3mf` layout is preserved (regression guard against the helper accidentally changing the normal path). 57/57 in `test_archives_api.py` green under `pytest -n 30`. Backend ruff clean. **Note**: existing orphan files at `/app/source/<filename>.3mf` from prior failed retries inside an affected user's container can be safely deleted; they were never indexed in the DB, never reachable from the UI, and would have vanished on the next container restart anyway.
|
||||
|
||||
@@ -628,6 +628,74 @@ def _get_start_ams_mapping(data: dict, archive_id: int | None) -> list[int] | No
|
||||
return stored_ams_mapping
|
||||
|
||||
|
||||
def _extract_filament_data_from_mqtt(data: dict, ams_mapping: list[int] | None = None) -> dict[str, str]:
|
||||
"""Best-effort filament metadata from the MQTT print-start snapshot.
|
||||
|
||||
Used when the 3MF can't be downloaded (P1S/A1/P2S firmwares lock the
|
||||
file during print, see #1533) so the fallback PrintArchive still has
|
||||
enough filament info to support the inventory views and AMS-expansion
|
||||
planning the operator opens it for. Returns a dict with optional
|
||||
``filament_type`` and ``filament_color`` keys in the same
|
||||
comma-separated format the 3MF extractor produces, so the rest of the
|
||||
codebase treats the fallback archive identically to a normal one.
|
||||
|
||||
``ams_mapping`` is the slicer's slot-per-print-filament list captured
|
||||
from the MQTT print payload (global tray IDs, possibly -1 for VT-tray
|
||||
entries). When supplied, only the slots actually consumed by this
|
||||
print contribute. Without it the function falls back to every loaded
|
||||
AMS slot — less accurate but still useful.
|
||||
"""
|
||||
result: dict[str, str] = {}
|
||||
ams_root = (data or {}).get("ams") or {}
|
||||
ams_units = ams_root.get("ams") if isinstance(ams_root, dict) else None
|
||||
if not isinstance(ams_units, list) or not ams_units:
|
||||
return result
|
||||
|
||||
# Map global tray id (unit * 4 + tray) → (type, color).
|
||||
loaded: dict[int, tuple[str, str]] = {}
|
||||
for unit in ams_units:
|
||||
if not isinstance(unit, dict):
|
||||
continue
|
||||
try:
|
||||
unit_id = int(unit.get("id", 0))
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
for tray in unit.get("tray") or []:
|
||||
if not isinstance(tray, dict):
|
||||
continue
|
||||
try:
|
||||
tray_id = int(tray.get("id", 0))
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
ttype = (tray.get("tray_type") or "").strip()
|
||||
tcolor = (tray.get("tray_color") or "").strip().upper()
|
||||
if not ttype:
|
||||
continue # Empty / unloaded slot.
|
||||
loaded[unit_id * 4 + tray_id] = (ttype, tcolor)
|
||||
|
||||
if not loaded:
|
||||
return result
|
||||
|
||||
if ams_mapping:
|
||||
used_ids = [int(x) for x in ams_mapping if isinstance(x, (int, float)) and int(x) >= 0]
|
||||
filaments = [loaded[g] for g in used_ids if g in loaded]
|
||||
if not filaments:
|
||||
return result # Mapping points entirely at slots we have no data for.
|
||||
else:
|
||||
filaments = [loaded[g] for g in sorted(loaded.keys())]
|
||||
|
||||
types_joined = ",".join(f[0] for f in filaments)
|
||||
colors_joined = ",".join(f[1] for f in filaments if f[1])
|
||||
|
||||
# Column limits per backend/app/models/archive.py: filament_type=50,
|
||||
# filament_color=200.
|
||||
if types_joined:
|
||||
result["filament_type"] = types_joined[:50]
|
||||
if colors_joined:
|
||||
result["filament_color"] = colors_joined[:200]
|
||||
return result
|
||||
|
||||
|
||||
def _maybe_start_layer_timelapse(printer, printer_id: int, archive_id: int) -> bool:
|
||||
"""Start a layer-timelapse session for *archive_id* when the printer has
|
||||
an external camera configured. Returns True if a session was started.
|
||||
@@ -2592,6 +2660,14 @@ async def on_print_start(printer_id: int, data: dict):
|
||||
if mc_remaining and isinstance(mc_remaining, (int, float)) and mc_remaining > 0:
|
||||
fallback_print_time = int(mc_remaining * 60)
|
||||
|
||||
# Best-effort filament metadata from MQTT — see
|
||||
# _extract_filament_data_from_mqtt. Without this the fallback
|
||||
# archive's filament fields stayed NULL even though the AMS
|
||||
# state at print start was sitting right there in `data`.
|
||||
# The slicer's ams_mapping (when present) narrows the result
|
||||
# to slots actually used by the print (#1533).
|
||||
mqtt_filament_meta = _extract_filament_data_from_mqtt(data, _get_start_ams_mapping(data, None))
|
||||
|
||||
# Create minimal archive entry
|
||||
fallback_archive = PrintArchive(
|
||||
printer_id=printer_id,
|
||||
@@ -2603,6 +2679,8 @@ async def on_print_start(printer_id: int, data: dict):
|
||||
status="printing",
|
||||
started_at=datetime.now(timezone.utc),
|
||||
subtask_id=subtask_id,
|
||||
filament_type=mqtt_filament_meta.get("filament_type"),
|
||||
filament_color=mqtt_filament_meta.get("filament_color"),
|
||||
extra_data={"no_3mf_available": True, "original_subtask": subtask_name, "_print_data": data},
|
||||
)
|
||||
|
||||
|
||||
@@ -0,0 +1,206 @@
|
||||
"""Tests for _extract_filament_data_from_mqtt (#1533).
|
||||
|
||||
The fallback PrintArchive path in main.py fires when the source 3MF can't
|
||||
be downloaded from the printer at print start — common on P1S / A1 / P2S
|
||||
firmwares that lock the file during printing. Before this fix the
|
||||
fallback archive had every filament field NULL even though the MQTT
|
||||
print-start payload already carried the AMS state and the slicer's
|
||||
slot-per-print-filament mapping. The helper extracts a comma-separated
|
||||
``filament_type`` / ``filament_color`` from that payload so the inventory
|
||||
views can at least show what's loaded, and operators planning AMS
|
||||
expansion can count filaments per print.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from backend.app.main import _extract_filament_data_from_mqtt
|
||||
|
||||
|
||||
def _ams_unit(unit_id: int, trays: list[dict]) -> dict:
|
||||
return {"id": unit_id, "tray": trays}
|
||||
|
||||
|
||||
def _tray(tray_id: int, ttype: str | None, color: str | None) -> dict:
|
||||
out: dict = {"id": tray_id}
|
||||
if ttype is not None:
|
||||
out["tray_type"] = ttype
|
||||
if color is not None:
|
||||
out["tray_color"] = color
|
||||
return out
|
||||
|
||||
|
||||
class TestExtractFilamentDataFromMqtt:
|
||||
def test_empty_payload_returns_empty_dict(self):
|
||||
assert _extract_filament_data_from_mqtt({}) == {}
|
||||
assert _extract_filament_data_from_mqtt({"ams": None}) == {}
|
||||
assert _extract_filament_data_from_mqtt({"ams": {}}) == {}
|
||||
assert _extract_filament_data_from_mqtt({"ams": {"ams": []}}) == {}
|
||||
|
||||
def test_no_loaded_slots_returns_empty(self):
|
||||
"""All slots empty (no tray_type) → nothing to report."""
|
||||
data = {
|
||||
"ams": {
|
||||
"ams": [
|
||||
_ams_unit(0, [_tray(i, "", "") for i in range(4)]),
|
||||
],
|
||||
}
|
||||
}
|
||||
assert _extract_filament_data_from_mqtt(data) == {}
|
||||
|
||||
def test_no_mapping_lists_all_loaded_slots_sorted(self):
|
||||
data = {
|
||||
"ams": {
|
||||
"ams": [
|
||||
_ams_unit(
|
||||
0,
|
||||
[
|
||||
_tray(0, "PLA", "FF0000"),
|
||||
_tray(1, "PETG", "00FF00"),
|
||||
_tray(2, "", ""), # Empty slot — skipped.
|
||||
_tray(3, "ABS", "0000ff"),
|
||||
],
|
||||
),
|
||||
],
|
||||
}
|
||||
}
|
||||
result = _extract_filament_data_from_mqtt(data)
|
||||
# Order is by ascending global tray id, colors uppercased.
|
||||
assert result == {"filament_type": "PLA,PETG,ABS", "filament_color": "FF0000,00FF00,0000FF"}
|
||||
|
||||
def test_ams_mapping_narrows_to_used_slots(self):
|
||||
"""The slicer's slot-per-print-filament mapping wins — only used
|
||||
slots contribute, in the slicer's order (which is the order the
|
||||
print materially consumes them)."""
|
||||
data = {
|
||||
"ams": {
|
||||
"ams": [
|
||||
_ams_unit(
|
||||
0,
|
||||
[
|
||||
_tray(0, "PLA", "FF0000"),
|
||||
_tray(1, "PETG", "00FF00"),
|
||||
_tray(2, "ABS", "0000FF"),
|
||||
_tray(3, "TPU", "FFFF00"),
|
||||
],
|
||||
),
|
||||
],
|
||||
}
|
||||
}
|
||||
# Print uses slots 3 then 0 then 1 (slot 2 untouched, no entry).
|
||||
result = _extract_filament_data_from_mqtt(data, ams_mapping=[3, 0, 1])
|
||||
assert result == {"filament_type": "TPU,PLA,PETG", "filament_color": "FFFF00,FF0000,00FF00"}
|
||||
|
||||
def test_ams_mapping_with_vt_tray_sentinels_filtered_out(self):
|
||||
"""ams_mapping entries equal to -1 represent the VT tray (external
|
||||
spool feed). We have no AMS tray data for them — they must be
|
||||
skipped, not treated as global tray id 0."""
|
||||
data = {
|
||||
"ams": {
|
||||
"ams": [
|
||||
_ams_unit(
|
||||
0,
|
||||
[
|
||||
_tray(0, "PLA", "FF0000"),
|
||||
_tray(1, "PETG", "00FF00"),
|
||||
],
|
||||
),
|
||||
],
|
||||
}
|
||||
}
|
||||
result = _extract_filament_data_from_mqtt(data, ams_mapping=[-1, 0, 1])
|
||||
assert result == {"filament_type": "PLA,PETG", "filament_color": "FF0000,00FF00"}
|
||||
|
||||
def test_dual_ams_global_ids_use_unit4_offset(self):
|
||||
"""A dual-AMS rig has unit 0 → trays 0-3, unit 1 → trays 4-7.
|
||||
``ams_mapping=4`` must resolve to unit 1, tray 0 — not unit 0."""
|
||||
data = {
|
||||
"ams": {
|
||||
"ams": [
|
||||
_ams_unit(0, [_tray(0, "PLA", "FF0000")]),
|
||||
_ams_unit(1, [_tray(0, "PETG-CF", "112233")]),
|
||||
],
|
||||
}
|
||||
}
|
||||
result = _extract_filament_data_from_mqtt(data, ams_mapping=[4, 0])
|
||||
assert result == {"filament_type": "PETG-CF,PLA", "filament_color": "112233,FF0000"}
|
||||
|
||||
def test_mapping_pointing_at_unknown_slot_falls_through_to_known_only(self):
|
||||
data = {
|
||||
"ams": {
|
||||
"ams": [
|
||||
_ams_unit(0, [_tray(0, "PLA", "FF0000")]),
|
||||
],
|
||||
}
|
||||
}
|
||||
# Slot 7 isn't in our AMS — entry skipped, only slot 0 remains.
|
||||
result = _extract_filament_data_from_mqtt(data, ams_mapping=[7, 0])
|
||||
assert result == {"filament_type": "PLA", "filament_color": "FF0000"}
|
||||
|
||||
def test_mapping_entirely_unknown_returns_empty(self):
|
||||
"""If every mapped slot is unknown the helper returns {} rather
|
||||
than silently misreporting from the all-slots fallback — the
|
||||
slicer was explicit about which slots to use."""
|
||||
data = {
|
||||
"ams": {
|
||||
"ams": [
|
||||
_ams_unit(0, [_tray(0, "PLA", "FF0000")]),
|
||||
],
|
||||
}
|
||||
}
|
||||
assert _extract_filament_data_from_mqtt(data, ams_mapping=[5, 6]) == {}
|
||||
|
||||
def test_color_truncation_at_column_limit(self):
|
||||
"""filament_color column is VARCHAR(200); long multi-color prints
|
||||
must not exceed it."""
|
||||
# 16 trays of 6-char colors + 15 commas = 96+15 = 111 chars. Safe.
|
||||
# Construct an oversized synthetic case with many distinct colors.
|
||||
trays = [_tray(i, "PLA", f"{i:06X}") for i in range(4)]
|
||||
data = {"ams": {"ams": [_ams_unit(u, trays) for u in range(8)]}}
|
||||
result = _extract_filament_data_from_mqtt(data)
|
||||
assert "filament_color" in result
|
||||
assert len(result["filament_color"]) <= 200
|
||||
|
||||
def test_type_truncation_at_column_limit(self):
|
||||
"""filament_type column is VARCHAR(50). Many filaments must truncate."""
|
||||
# 16 PETG-CF entries: 7 chars × 16 + 15 commas = 127 chars.
|
||||
trays = [_tray(i, "PETG-CF", "AABBCC") for i in range(4)]
|
||||
data = {"ams": {"ams": [_ams_unit(u, trays) for u in range(4)]}}
|
||||
result = _extract_filament_data_from_mqtt(data)
|
||||
assert "filament_type" in result
|
||||
assert len(result["filament_type"]) <= 50
|
||||
|
||||
def test_color_missing_only_emits_type(self):
|
||||
"""A tray with type but blank color still contributes to filament_type."""
|
||||
data = {
|
||||
"ams": {
|
||||
"ams": [
|
||||
_ams_unit(0, [_tray(0, "PLA", "")]),
|
||||
],
|
||||
}
|
||||
}
|
||||
result = _extract_filament_data_from_mqtt(data)
|
||||
assert result == {"filament_type": "PLA"}
|
||||
# filament_color absent — not empty string.
|
||||
assert "filament_color" not in result
|
||||
|
||||
def test_malformed_unit_skipped_without_crash(self):
|
||||
"""Defensive: unexpected MQTT shapes (non-dict in ams list, missing
|
||||
id, string tray.id) must not raise. The fallback-archive write
|
||||
runs in a hot path during print start — anything that throws here
|
||||
would bubble up and break the print log entirely."""
|
||||
data = {
|
||||
"ams": {
|
||||
"ams": [
|
||||
"garbage",
|
||||
{"id": "not-an-int", "tray": []},
|
||||
_ams_unit(0, [_tray(0, "PLA", "FF0000"), {"id": "x", "tray_type": "PETG"}]),
|
||||
],
|
||||
}
|
||||
}
|
||||
result = _extract_filament_data_from_mqtt(data)
|
||||
# Only the well-formed entry contributes; no exception.
|
||||
assert result.get("filament_type") == "PLA"
|
||||
|
||||
@pytest.mark.parametrize("data", [None, {}, {"ams": "weird-string"}])
|
||||
def test_garbage_top_level_is_empty(self, data):
|
||||
assert _extract_filament_data_from_mqtt(data or {}) == {}
|
||||
Reference in New Issue
Block a user