diff --git a/backend/app/services/archive.py b/backend/app/services/archive.py index f321d62a8..4e8f3785d 100644 --- a/backend/app/services/archive.py +++ b/backend/app/services/archive.py @@ -18,6 +18,7 @@ from backend.app.core.tasks import spawn_background_task from backend.app.models.archive import PrintArchive from backend.app.models.filament import Filament from backend.app.models.printer import Printer +from backend.app.utils.archive_paths import archive_dir as resolve_archive_dir from backend.app.utils.filename import clean_display_name from backend.app.utils.safe_path import PathTraversalError, safe_join_under @@ -1605,11 +1606,17 @@ class ArchiveService: if not archive: return False - # Get archive directory - file_path = ( - settings.base_dir / archive.file_path - ) # SEC-PATH-OK: archive.file_path is DB-stored, set by archive_print() under settings.archive_dir - archive_dir = file_path.parent + # Where this archive's files live. Deliberately the shared helper: an + # archive created without a 3MF has ``file_path == ""``, and deriving the + # directory here as ``(base_dir / "").parent`` resolved to the parent of + # base_dir — outside the data directory entirely. In Docker that is /app, + # so the write failed with EACCES and the timelapse was retried and + # discarded 25 times; where the parent happens to be writable it + # succeeded, dropped a stray video next to the install, and then failed + # anyway on the relative_to() below. Every H2-series and P2S print sent + # from the slicer takes that path, because the file goes to internal + # storage and no 3MF can be fetched. + archive_dir = resolve_archive_dir(archive) # Save timelapse - use thread pool to avoid blocking event loop # (timelapse files can be 100MB+, sync write blocks for seconds). @@ -1629,6 +1636,10 @@ class ArchiveService: archive_id, ) return False + # Created only once the name has been vetted, so a rejected filename + # leaves nothing behind. A no-3MF archive has never had a directory of + # its own, and the timelapse can be the first thing to want one. + await asyncio.to_thread(lambda: timelapse_file.parent.mkdir(parents=True, exist_ok=True)) await asyncio.to_thread(timelapse_file.write_bytes, timelapse_data) # Update archive record diff --git a/backend/app/services/usage_tracker.py b/backend/app/services/usage_tracker.py index c2b0c6543..dc6fe6514 100644 --- a/backend/app/services/usage_tracker.py +++ b/backend/app/services/usage_tracker.py @@ -10,6 +10,7 @@ AMS remain% delta is the fallback for trays not covered by 3MF data. import asyncio import json import logging +import re from dataclasses import dataclass, field from datetime import datetime, timezone @@ -738,7 +739,13 @@ async def on_print_complete( search_filename = data.get("filename") or data.get("subtask_name") or (session.print_name if session else "") if search_filename: - threemf_path = await _find_3mf_by_filename(printer_id, search_filename, db, app_settings.base_dir) + threemf_path = await _find_3mf_by_filename( + printer_id, + search_filename, + db, + app_settings.base_dir, + print_name=data.get("subtask_name") or (session.print_name if session else None), + ) if archive_id or threemf_path: threemf_results = await _track_from_3mf( @@ -830,6 +837,18 @@ async def on_print_complete( continue # Already tracked via 3MF if key not in session.tray_remain_start: + # No usable remain% when the print began, so there is no delta + # to charge. Said out loud for the same reason as the branches + # below: a slot the print used, holding a spool the operator + # assigned, otherwise vanished from the accounting without a + # word. Common on non-RFID spools, which report remain = -1 + # until a remaining amount is set by hand. + if not print_used_keys or key in print_used_keys: + logger.info( + "[UsageTracker] %s: no valid remain%% at print start, nothing to charge for printer %d", + tray_label, + printer_id, + ) continue # Skip trays the print never touched. Only enforce when we have @@ -994,6 +1013,60 @@ async def on_print_complete( return results +# A running print's ``filename`` is the path the printer is executing, and on a +# sliced job that is always ``…/Metadata/plate_.gcode``. Its stem names the +# *plate*, not the model, and every Bambu print in existence has one — so it +# identifies nothing and must never be used to match a 3MF. It reached the +# matcher for real on H2-series and P2S prints, where the file goes to internal +# eMMC, no 3MF can be fetched, and the archive keeps the gcode path as its +# filename: `plate_1` then matched an unrelated `lid_plate_1.gcode.3mf` and that +# print's filament figures were read off a different model entirely. +_GENERIC_PLATE_STEM = re.compile(r"^plate_?\d+$", re.IGNORECASE) + + +def _like_escape(value: str) -> str: + """Escape LIKE metacharacters so a stem matches literally. + + ``_`` is a single-character wildcard, and model names are full of them. + """ + return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + + +def _threemf_search_stem(*candidates: str | None) -> str | None: + """First candidate that names a model, or None if none of them do. + + Candidates are tried in order and the generic plate name is skipped rather + than accepted, so a print that only has one falls through to "no match" + instead of matching everything. + """ + for raw in candidates: + if not raw: + continue + stem = raw.split("/")[-1].strip() + for suffix in (".gcode.3mf", ".gcode", ".3mf"): + if stem.lower().endswith(suffix): + stem = stem[: -len(suffix)] + break + # Stripped only to judge the stem, never to change it: a real archive + # here is named "…Face Down .gcode.3mf", and a stem trimmed to + # "…Face Down" no longer matches the file it came from. + probe = stem.strip() + if probe and not _GENERIC_PLATE_STEM.match(probe): + return stem + return None + + +def _stem_matches(column, stem: str): + """Filter matching *stem* at a filename boundary rather than anywhere. + + ``ilike("%.%")`` also matched a *suffix* of a longer name, which is how + `plate_1` reached `lid_plate_1.gcode.3mf`. A name is either the whole + basename or the basename after a directory separator. + """ + escaped = _like_escape(stem) + return column.ilike(f"{escaped}.%", escape="\\") | column.ilike(f"%/{escaped}.%", escape="\\") + + async def _resolve_3mf_fallback(archive, db: AsyncSession, base_dir): """Try to find a 3MF file from library or a previous archive when the current archive has none. @@ -1005,13 +1078,9 @@ async def _resolve_3mf_fallback(archive, db: AsyncSession, base_dir): from backend.app.models.archive import PrintArchive from backend.app.models.library import LibraryFile - # Derive search name from archive filename (e.g. "benchy.3mf" or "benchy.gcode.3mf") - search_name = archive.filename or archive.print_name - if not search_name: - return None - # Normalize: strip path parts, get base name - search_name = search_name.split("/")[-1] - search_base = search_name.replace(".gcode.3mf", "").replace(".gcode", "").replace(".3mf", "") + # Derive search name from archive filename (e.g. "benchy.3mf" or "benchy.gcode.3mf"), + # falling back to the print name when the filename is only a plate path. + search_base = _threemf_search_stem(archive.filename, archive.print_name) if not search_base: return None @@ -1019,7 +1088,7 @@ async def _resolve_3mf_fallback(archive, db: AsyncSession, base_dir): try: lib_result = await db.execute( LibraryFile.active() - .where(LibraryFile.file_path.ilike(f"%/{search_base}.%") | LibraryFile.file_path.ilike(f"{search_base}.%")) + .where(_stem_matches(LibraryFile.file_path, search_base)) .where(LibraryFile.file_path.ilike("%.3mf")) .order_by(LibraryFile.created_at.desc()) .limit(3) @@ -1041,9 +1110,7 @@ async def _resolve_3mf_fallback(archive, db: AsyncSession, base_dir): .where(PrintArchive.printer_id == archive.printer_id) .where(PrintArchive.file_path != "") .where(PrintArchive.file_path.isnot(None)) - .where( - PrintArchive.filename.ilike(f"%{search_base}.%") | PrintArchive.filename.ilike(f"{search_base}.%"), - ) + .where(_stem_matches(PrintArchive.filename, search_base)) .order_by(PrintArchive.created_at.desc()) .limit(3) ) @@ -1067,19 +1134,22 @@ async def _find_3mf_by_filename( filename: str, db: AsyncSession, base_dir, + print_name: str | None = None, ): """Find a 3MF file by filename from library or previous archives. Used when auto-archive is disabled and there's no archive_id, but we still need the 3MF slicer data for filament usage tracking. + + ``print_name`` is the model name to fall back to when ``filename`` is the + printer's plate path, which names no model at all. """ from pathlib import Path from backend.app.models.archive import PrintArchive from backend.app.models.library import LibraryFile - search_name = filename.split("/")[-1] if "/" in filename else filename - search_base = search_name.replace(".gcode.3mf", "").replace(".gcode", "").replace(".3mf", "") + search_base = _threemf_search_stem(filename, print_name) if not search_base: return None @@ -1087,7 +1157,7 @@ async def _find_3mf_by_filename( try: lib_result = await db.execute( LibraryFile.active() - .where(LibraryFile.file_path.ilike(f"%/{search_base}.%") | LibraryFile.file_path.ilike(f"{search_base}.%")) + .where(_stem_matches(LibraryFile.file_path, search_base)) .where(LibraryFile.file_path.ilike("%.3mf")) .order_by(LibraryFile.created_at.desc()) .limit(3) @@ -1108,9 +1178,7 @@ async def _find_3mf_by_filename( .where(PrintArchive.printer_id == printer_id) .where(PrintArchive.file_path != "") .where(PrintArchive.file_path.isnot(None)) - .where( - PrintArchive.filename.ilike(f"%{search_base}.%") | PrintArchive.filename.ilike(f"{search_base}.%"), - ) + .where(_stem_matches(PrintArchive.filename, search_base)) .order_by(PrintArchive.created_at.desc()) .limit(3) ) diff --git a/backend/tests/integration/test_timelapse_scan_session.py b/backend/tests/integration/test_timelapse_scan_session.py index cc56a2965..85d93f77b 100644 --- a/backend/tests/integration/test_timelapse_scan_session.py +++ b/backend/tests/integration/test_timelapse_scan_session.py @@ -91,10 +91,9 @@ async def test_scan_timelapse_attaches_and_persists_via_fresh_session( # parent, then stores a base_dir-relative timelapse_path. Point base_dir at # tmp and stage the archive dir so the real write succeeds (mirrors # test_attach_timelapse_safe_path). - monkeypatch.setattr( - "backend.app.services.archive.settings", - MagicMock(base_dir=tmp_path), - ) + fake_settings = MagicMock(base_dir=tmp_path, archive_dir=tmp_path / "archive") + monkeypatch.setattr("backend.app.services.archive.settings", fake_settings) + monkeypatch.setattr("backend.app.utils.archive_paths.settings", fake_settings) archive_dir = tmp_path / "archives" / "test" archive_dir.mkdir(parents=True) diff --git a/backend/tests/unit/services/test_attach_timelapse_safe_path.py b/backend/tests/unit/services/test_attach_timelapse_safe_path.py index 3dfa9888c..dbbdc318a 100644 --- a/backend/tests/unit/services/test_attach_timelapse_safe_path.py +++ b/backend/tests/unit/services/test_attach_timelapse_safe_path.py @@ -25,10 +25,12 @@ async def test_attach_timelapse_rejects_dotdot_filename(tmp_path: Path, monkeypa archive_dir.mkdir(parents=True) # Repoint settings.base_dir so attach_timelapse's archive_dir = file_path.parent # resolves to our tmp directory. - monkeypatch.setattr( - "backend.app.services.archive.settings", - MagicMock(base_dir=tmp_path), - ) + # Both the service and the shared archive_dir helper read settings, through + # separate module-level bindings — patch both or the helper keeps the real + # data directory and the write escapes tmp_path. + fake_settings = MagicMock(base_dir=tmp_path, archive_dir=tmp_path / "archive") + monkeypatch.setattr("backend.app.services.archive.settings", fake_settings) + monkeypatch.setattr("backend.app.utils.archive_paths.settings", fake_settings) db = MagicMock() db.commit = AsyncMock() @@ -62,10 +64,12 @@ async def test_attach_timelapse_rejects_absolute_filename(tmp_path: Path, monkey """An absolute path in filename must not collapse the join.""" archive_dir = tmp_path / "archive" / "1" / "20260101_test" archive_dir.mkdir(parents=True) - monkeypatch.setattr( - "backend.app.services.archive.settings", - MagicMock(base_dir=tmp_path), - ) + # Both the service and the shared archive_dir helper read settings, through + # separate module-level bindings — patch both or the helper keeps the real + # data directory and the write escapes tmp_path. + fake_settings = MagicMock(base_dir=tmp_path, archive_dir=tmp_path / "archive") + monkeypatch.setattr("backend.app.services.archive.settings", fake_settings) + monkeypatch.setattr("backend.app.utils.archive_paths.settings", fake_settings) db = MagicMock() db.commit = AsyncMock() @@ -90,10 +94,12 @@ async def test_attach_timelapse_accepts_legit_filename(tmp_path: Path, monkeypat """The legitimate happy path must still work — the fix isn't over-strict.""" archive_dir = tmp_path / "archive" / "1" / "20260101_test" archive_dir.mkdir(parents=True) - monkeypatch.setattr( - "backend.app.services.archive.settings", - MagicMock(base_dir=tmp_path), - ) + # Both the service and the shared archive_dir helper read settings, through + # separate module-level bindings — patch both or the helper keeps the real + # data directory and the write escapes tmp_path. + fake_settings = MagicMock(base_dir=tmp_path, archive_dir=tmp_path / "archive") + monkeypatch.setattr("backend.app.services.archive.settings", fake_settings) + monkeypatch.setattr("backend.app.utils.archive_paths.settings", fake_settings) db = MagicMock() db.commit = AsyncMock() diff --git a/backend/tests/unit/test_3mf_fallback_matching.py b/backend/tests/unit/test_3mf_fallback_matching.py new file mode 100644 index 000000000..8f2730745 --- /dev/null +++ b/backend/tests/unit/test_3mf_fallback_matching.py @@ -0,0 +1,160 @@ +"""A no-3MF archive must not borrow a stranger's 3MF (#2843). + +H2-series and P2S firmware keeps a slicer-sent file on internal eMMC, which +Bambuddy cannot read, so the print becomes an archive with no file and its +``filename`` stays the path the printer is executing: +``/data/Metadata/plate_1.gcode``. + +Measured on the maintainer's H2D, 2026-08-17. The stem of that path is +``plate_1``, the old matcher searched ``filename ILIKE '%plate_1.%'``, and +``lid_plate_1.gcode.3mf`` matched — so a 1.6 g Cube was costed from a 207 g +four-colour ABS print: + + [UsageTracker] 3MF fallback: found previous archive 287 file for archive 345 + [UsageTracker] 3MF: slot_id=2 -> global_tray=4 -> AMS1-T0 (used_g=204.9 ...) + +Every Bambu print has a ``plate_N``, so this was not a near-miss between similar +names — it was a name that matches everything. +""" + +import pytest + +from backend.app.services.usage_tracker import ( + _like_escape, + _stem_matches, + _threemf_search_stem, +) + + +class TestSearchStem: + def test_a_real_filename_still_wins(self): + """Unchanged for every archive that has a 3MF of its own.""" + assert _threemf_search_stem("Cube.gcode.3mf", "Cube") == "Cube" + assert _threemf_search_stem("benchy.3mf", None) == "benchy" + + def test_the_plate_path_is_refused(self): + """The reported case: fall through to the model name instead.""" + assert _threemf_search_stem("/data/Metadata/plate_1.gcode", "Cube") == "Cube" + + @pytest.mark.parametrize("plate", ["plate_1", "plate_12", "plate1", "PLATE_3"]) + def test_every_plate_spelling_is_refused(self, plate): + assert _threemf_search_stem(f"/data/Metadata/{plate}.gcode", None) is None + + def test_no_usable_name_matches_nothing(self): + """Better to skip tracking than to charge a print for another model.""" + assert _threemf_search_stem("/data/Metadata/plate_1.gcode", None) is None + assert _threemf_search_stem(None, None) is None + assert _threemf_search_stem("", "") is None + + def test_a_model_named_after_a_plate_survives(self): + """`lid_plate_1` names a model — only a bare plate stem is generic.""" + assert _threemf_search_stem("lid_plate_1.gcode.3mf", None) == "lid_plate_1" + + def test_whitespace_before_the_extension_is_preserved(self): + """A real archive on the maintainer's install is named + "…Face Down .gcode.3mf". Trimming the stem to "…Face Down" would stop it + matching the very file it was derived from.""" + assert _threemf_search_stem("Steelers 6 Color Face Down .gcode.3mf", None) == "Steelers 6 Color Face Down " + + def test_surrounding_whitespace_is_still_ignored(self): + assert _threemf_search_stem(" Cube.3mf ", None) == "Cube" + + +class TestLikeEscaping: + def test_underscores_are_literal(self): + """``_`` is a single-character LIKE wildcard, and model names are full + of them — unescaped, `Cube_v1` also matches `CubeXv1`.""" + assert _like_escape("Cube_v1") == "Cube\\_v1" + + def test_percent_and_backslash(self): + assert _like_escape("100%_scale") == "100\\%\\_scale" + assert _like_escape("a\\b") == "a\\\\b" + + +class TestStemMatchesAtABoundary: + """The SQL the matcher builds, checked by rendering it.""" + + @staticmethod + def _patterns(stem): + from backend.app.models.archive import PrintArchive + + clause = _stem_matches(PrintArchive.filename, stem) + return str(clause.compile(compile_kwargs={"literal_binds": True})) + + def test_it_no_longer_matches_a_suffix_of_a_longer_name(self): + """The whole bug in one assertion: `%plate_1.%` is gone.""" + assert "%plate_1.%" not in self._patterns("plate_1") + + def test_it_anchors_the_basename(self): + sql = self._patterns("Cube") + assert "Cube.%" in sql + assert "%/Cube.%" in sql + + def test_it_escapes_the_stem(self): + assert "Cube\\_v1.%" in self._patterns("Cube_v1") + + +@pytest.mark.asyncio +async def test_the_h2d_collision_no_longer_resolves(db_session, tmp_path): + """End to end against the real rows: a Cube on eMMC must not resolve to + `lid_plate_1.gcode.3mf`.""" + from backend.app.models.archive import PrintArchive + from backend.app.services.usage_tracker import _resolve_3mf_fallback + + donor_file = tmp_path / "archive" / "1" / "lid_plate_1.gcode.3mf" + donor_file.parent.mkdir(parents=True) + donor_file.write_bytes(b"PK\x03\x04not-really-a-3mf") + + donor = PrintArchive( + printer_id=1, + print_name="lid_plate_1", + filename="lid_plate_1.gcode.3mf", + file_path="archive/1/lid_plate_1.gcode.3mf", + file_size=1, + status="completed", + ) + # The eMMC print: no file of its own, filename is the plate path. + orphan = PrintArchive( + printer_id=1, + print_name="Cube", + filename="/data/Metadata/plate_1.gcode", + file_path="", + file_size=0, + status="completed", + ) + db_session.add_all([donor, orphan]) + await db_session.commit() + + assert await _resolve_3mf_fallback(orphan, db_session, tmp_path) is None + + +@pytest.mark.asyncio +async def test_a_genuine_same_model_reprint_still_resolves(db_session, tmp_path): + """The fallback's actual purpose must survive the fix.""" + from backend.app.models.archive import PrintArchive + from backend.app.services.usage_tracker import _resolve_3mf_fallback + + donor_file = tmp_path / "archive" / "1" / "Cube.gcode.3mf" + donor_file.parent.mkdir(parents=True) + donor_file.write_bytes(b"PK\x03\x04not-really-a-3mf") + + donor = PrintArchive( + printer_id=1, + print_name="Cube", + filename="Cube.gcode.3mf", + file_path="archive/1/Cube.gcode.3mf", + file_size=1, + status="completed", + ) + orphan = PrintArchive( + printer_id=1, + print_name="Cube", + filename="/data/Metadata/plate_1.gcode", + file_path="", + file_size=0, + status="completed", + ) + db_session.add_all([donor, orphan]) + await db_session.commit() + + assert await _resolve_3mf_fallback(orphan, db_session, tmp_path) == donor_file diff --git a/backend/tests/unit/test_archive_filtering.py b/backend/tests/unit/test_archive_filtering.py index 7789f7cc9..0a6d99f72 100644 --- a/backend/tests/unit/test_archive_filtering.py +++ b/backend/tests/unit/test_archive_filtering.py @@ -687,9 +687,12 @@ class TestConvertTimelapseToMp4: patch("backend.app.services.camera.get_ffmpeg_path", return_value="/usr/bin/ffmpeg"), patch("backend.app.core.database.async_session", return_value=mock_session), patch("backend.app.services.archive.settings") as mock_settings, + patch("backend.app.utils.archive_paths.settings") as mock_paths_settings, patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_exec, ): mock_settings.base_dir = tmp_path + mock_paths_settings.base_dir = tmp_path + mock_paths_settings.archive_dir = tmp_path / "archive" mock_exec.return_value = mock_process # Create the expected output file (as FFmpeg would) mp4_path.write_bytes(b"fake mp4 output") @@ -772,9 +775,12 @@ class TestAttachTimelapseBackgroundConversion: with ( patch("backend.app.services.archive.settings") as mock_settings, + patch("backend.app.utils.archive_paths.settings") as mock_paths_settings, patch("asyncio.create_task") as mock_create_task, ): mock_settings.base_dir = tmp_path + mock_paths_settings.base_dir = tmp_path + mock_paths_settings.archive_dir = tmp_path / "archive" result = await service.attach_timelapse(1, b"fake mp4 data", "video.mp4") @@ -798,9 +804,12 @@ class TestAttachTimelapseBackgroundConversion: with ( patch("backend.app.services.archive.settings") as mock_settings, + patch("backend.app.utils.archive_paths.settings") as mock_paths_settings, patch("asyncio.create_task") as mock_create_task, ): mock_settings.base_dir = tmp_path + mock_paths_settings.base_dir = tmp_path + mock_paths_settings.archive_dir = tmp_path / "archive" result = await service.attach_timelapse(1, b"fake avi data", "video.avi") diff --git a/backend/tests/unit/test_no_3mf_archive_paths.py b/backend/tests/unit/test_no_3mf_archive_paths.py new file mode 100644 index 000000000..5506e1b8c --- /dev/null +++ b/backend/tests/unit/test_no_3mf_archive_paths.py @@ -0,0 +1,169 @@ +"""A no-3MF archive still owns a directory, and still explains itself (#2843). + +Both cases below are the same underlying situation: an H2-series or P2S print +sent from the slicer goes to internal eMMC, Bambuddy cannot fetch the 3MF, and +the archive is created with ``file_path == ""``. +""" + +import logging +from datetime import datetime, timezone +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + + +@pytest.fixture +def data_dirs(monkeypatch, tmp_path): + """Point base_dir/archive_dir at a scratch tree, as a real install has them.""" + from backend.app.core.config import settings + + base = tmp_path / "data" + (base / "archive").mkdir(parents=True) + monkeypatch.setattr(settings, "base_dir", base) + monkeypatch.setattr(settings, "archive_dir", base / "archive") + return base + + +class TestTimelapseDestination: + """``attach_timelapse`` must stay inside the data directory.""" + + @staticmethod + async def _archive(db_session, file_path: str): + from backend.app.models.archive import PrintArchive + + archive = PrintArchive( + printer_id=1, + print_name="Cube", + filename="/data/Metadata/plate_1.gcode" if not file_path else "Cube.gcode.3mf", + file_path=file_path, + file_size=0, + status="completed", + ) + db_session.add(archive) + await db_session.commit() + return archive + + @pytest.mark.asyncio + async def test_no_3mf_archive_writes_under_the_data_dir(self, db_session, data_dirs): + """Regression: this resolved to ``base_dir.parent`` — /app in Docker, so + the write failed EACCES; where the parent was writable it dropped a stray + video beside the install and then failed on relative_to() anyway.""" + from backend.app.services.archive import ArchiveService + + archive = await self._archive(db_session, "") + + ok = await ArchiveService(db_session).attach_timelapse(archive.id, b"video-bytes", "video_2026.mp4") + + assert ok is True + written = data_dirs / "archive" / str(archive.id) / "video_2026.mp4" + assert written.read_bytes() == b"video-bytes" + # Nothing may appear above the data directory. + assert not list(data_dirs.parent.glob("*.mp4")) + + @pytest.mark.asyncio + async def test_timelapse_path_is_stored_relative_to_base_dir(self, db_session, data_dirs): + """The old path could not be made relative to base_dir at all, which is + what raised ValueError and lost the video after a successful download.""" + from backend.app.services.archive import ArchiveService + + archive = await self._archive(db_session, "") + + await ArchiveService(db_session).attach_timelapse(archive.id, b"video-bytes", "video_2026.mp4") + + assert archive.timelapse_path == f"archive/{archive.id}/video_2026.mp4" + + @pytest.mark.asyncio + async def test_a_normal_archive_is_unaffected(self, db_session, data_dirs): + """An archive with a 3MF keeps writing beside it, exactly as before.""" + from backend.app.services.archive import ArchiveService + + archive = await self._archive(db_session, "archive/1/20260817_Cube/Cube.gcode.3mf") + (data_dirs / "archive" / "1" / "20260817_Cube").mkdir(parents=True) + + ok = await ArchiveService(db_session).attach_timelapse(archive.id, b"video-bytes", "video_2026.mp4") + + assert ok is True + assert archive.timelapse_path == "archive/1/20260817_Cube/video_2026.mp4" + + +class TestUnchargeableTrayIsAnnounced: + """A tray the print used but could not be charged must say so.""" + + @pytest.fixture(autouse=True) + def _clear_sessions(self): + from backend.app.services.usage_tracker import _active_sessions + + _active_sessions.clear() + yield + _active_sessions.clear() + + @pytest.mark.asyncio + async def test_used_tray_with_no_start_remain_is_logged(self, db_session, caplog): + """Non-RFID spools report remain = -1, so they never enter + ``tray_remain_start`` — and the loop skipped them with a bare + ``continue``. Nothing deducted, no reason given anywhere.""" + from backend.app.services.usage_tracker import PrintSession, _active_sessions, on_print_complete + + _active_sessions[1] = PrintSession( + printer_id=1, + print_name="Cube", + started_at=datetime.now(timezone.utc), + # AMS0-T0 is missing: it read -1 when the print began. + tray_remain_start={(1, 0): 50}, + tray_now_at_start=0, + ) + printer_manager = MagicMock() + printer_manager.get_status.return_value = SimpleNamespace( + raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 40}]}]}, + progress=100, + layer_num=50, + tray_now=0, + tray_change_log=[], + ) + + with caplog.at_level(logging.INFO, logger="backend.app.services.usage_tracker"): + await on_print_complete( + printer_id=1, + data={"status": "completed"}, + printer_manager=printer_manager, + db=db_session, + archive_id=None, + ams_mapping=[0], + ) + + assert "AMS0-T0: no valid remain% at print start" in caplog.text + + @pytest.mark.asyncio + async def test_a_tray_the_print_never_touched_stays_quiet(self, db_session, caplog): + """The loop walks every tray on the printer, so logging unconditionally + would narrate slots that had nothing to do with this print.""" + from backend.app.services.usage_tracker import PrintSession, _active_sessions, on_print_complete + + _active_sessions[1] = PrintSession( + printer_id=1, + print_name="Cube", + started_at=datetime.now(timezone.utc), + tray_remain_start={(1, 0): 50}, + tray_now_at_start=0, + ) + printer_manager = MagicMock() + printer_manager.get_status.return_value = SimpleNamespace( + raw_data={"ams": [{"id": 3, "tray": [{"id": 2, "remain": 40}]}]}, + progress=100, + layer_num=50, + tray_now=0, + tray_change_log=[], + ) + + with caplog.at_level(logging.INFO, logger="backend.app.services.usage_tracker"): + await on_print_complete( + printer_id=1, + data={"status": "completed"}, + printer_manager=printer_manager, + db=db_session, + archive_id=None, + ams_mapping=[0], + ) + + assert "AMS3-T2: no valid remain%" not in caplog.text