mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 19:21:33 +02:00
maziggy
0623cc46df
Two faults behind the same kind of print: one that arrives without a
retrievable 3MF, which on an H2S is any job started from the printer's
own internal library.
Such an archive has no file_path, and Path("").parent is Path("."), so
every site that derived the archive's folder from it landed on the data
directory itself. The finish-photo capture spotted that and wrote to
<archive_dir>/<id>/photos instead. Nothing else did. The photo was
written in one place and looked for in another: reads 404'd, deletes
dropped the name and left the file, and the notification attachment
never found the image. Hand-uploaded photos worked only because upload
and read agreed with each other rather than with the capture. Give the
question one owner in utils/archive_paths and have all four sites ask
it. Lookups check the old shared location too, so photos already
uploaded there stay reachable; uploads now go where captures go.
Separately, the remain%-delta fallback that stands in for a missing 3MF
can charge nothing for several reasons, and did so without a word. The
AMS reading is coarse and, on the reporter's printer, noisy: it rises
mid-print, swings five points over a job, sits at 100% through a
36-minute print on a fresh spool, and goes negative on a nearly empty
one -- which the start-of-print gate rejects, dropping the only slot
that was printing. Two of their prints went uncounted for two different
reasons and both read as "no spools updated", which is also what a print
with nothing to charge prints. Name the slot and the two readings in
each case, on the Spoolman path and on the internal-inventory path,
which has carried the same gates since #1119.
The Spoolman path also had no notion of which slots the print used, so a
spool swapped into an idle slot mid-print reads as consumption and is
billed to whoever that slot is assigned to -- the fault #1269 fixed for
the internal tracker, still open here, and likeliest on exactly the
prints this fallback serves, where nothing else narrows the field. Use
the same three pieces of evidence it does: the print's mapping, its
mid-print tray changes, and the tray it started on. The last needs
storing, because the internal tracker's row is deleted before this runs
and a screen-started print has no mapping to fall back on -- hence a new
nullable column, and no backfill, since a row from before it existed has
nothing to say. Where no evidence exists at all, every slot is still
considered.
Both paths also treated tray_now == 255 as naming a slot. It does not:
it is the field's initial value, the fallback for an unparseable
reading, and what it reports with nothing loaded. Mapped as a tray id it
becomes (255, 1), so as the only evidence it excluded every real slot
and charged nothing at all -- this issue's own bug, arriving by a new
route. On the internal path that is live today; on the Spoolman path it
would have shipped with the guard above. The external holder reports 254
when it is genuinely in use.
The arithmetic is untouched: at one percent per step this cannot resolve
a small print, and pretending otherwise would be worse than saying so.
1020 lines
40 KiB
Python
1020 lines
40 KiB
Python
"""Unit tests for the filament usage tracker.
|
|
|
|
Tests 3MF-primary tracking (Path 1) and AMS remain% delta fallback
|
|
(Path 2) for spools not covered by 3MF data.
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from backend.app.services.usage_tracker import (
|
|
PrintSession,
|
|
_active_sessions,
|
|
_archive_colors_from_spools,
|
|
_spool_color_to_hex,
|
|
_track_from_3mf,
|
|
on_print_complete,
|
|
on_print_start,
|
|
)
|
|
|
|
|
|
def _make_spool(*, id=1, label_weight=1000, weight_used=0, tag_uid=None, tray_uuid=None, rgba=None):
|
|
"""Create a mock Spool object."""
|
|
spool = MagicMock()
|
|
spool.id = id
|
|
spool.label_weight = label_weight
|
|
spool.weight_used = weight_used
|
|
spool.tag_uid = tag_uid
|
|
spool.tray_uuid = tray_uuid
|
|
spool.last_used = None
|
|
spool.cost_per_kg = None
|
|
spool.material = "PLA"
|
|
spool.rgba = rgba
|
|
return spool
|
|
|
|
|
|
def _make_assignment(*, spool_id=1, printer_id=1, ams_id=0, tray_id=0, created_at=None):
|
|
"""Create a mock SpoolAssignment object."""
|
|
assignment = MagicMock()
|
|
assignment.spool_id = spool_id
|
|
assignment.printer_id = printer_id
|
|
assignment.ams_id = ams_id
|
|
assignment.tray_id = tray_id
|
|
assignment.created_at = created_at or datetime.now(timezone.utc)
|
|
return assignment
|
|
|
|
|
|
def _make_printer_state(ams_data, progress=0, layer_num=0, tray_now=255):
|
|
"""Create a mock printer state with AMS data."""
|
|
state = MagicMock()
|
|
state.raw_data = {"ams": ams_data}
|
|
state.progress = progress
|
|
state.layer_num = layer_num
|
|
state.tray_now = tray_now
|
|
return state
|
|
|
|
|
|
def _make_printer_manager(state=None):
|
|
"""Create a mock printer manager."""
|
|
pm = MagicMock()
|
|
pm.get_status.return_value = state
|
|
return pm
|
|
|
|
|
|
class TestOnPrintStart:
|
|
"""Tests for on_print_start — capturing AMS remain%."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_sessions(self):
|
|
_active_sessions.clear()
|
|
yield
|
|
_active_sessions.clear()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_creates_session_with_valid_remain(self):
|
|
"""Session created with remain% data for trays reporting 0-100."""
|
|
ams_data = [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]
|
|
pm = _make_printer_manager(_make_printer_state(ams_data))
|
|
|
|
await on_print_start(1, {"subtask_name": "test_print"}, pm)
|
|
|
|
assert 1 in _active_sessions
|
|
session = _active_sessions[1]
|
|
assert session.print_name == "test_print"
|
|
assert session.tray_remain_start == {(0, 0): 80}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_creates_session_even_without_valid_remain(self):
|
|
"""Session still created when remain=-1 (for 3MF fallback path)."""
|
|
ams_data = [{"id": 0, "tray": [{"id": 0, "remain": -1}]}]
|
|
pm = _make_printer_manager(_make_printer_state(ams_data))
|
|
|
|
await on_print_start(1, {"subtask_name": "test_print"}, pm)
|
|
|
|
assert 1 in _active_sessions
|
|
session = _active_sessions[1]
|
|
assert session.tray_remain_start == {} # Empty, no valid remain
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skips_without_ams_data(self):
|
|
"""No session created when no AMS data available."""
|
|
state = MagicMock()
|
|
state.raw_data = {"ams": []}
|
|
pm = _make_printer_manager(state)
|
|
|
|
await on_print_start(1, {"subtask_name": "test"}, pm)
|
|
|
|
assert 1 not in _active_sessions
|
|
|
|
|
|
class TestOnPrintCompleteAMSDelta:
|
|
"""Tests for Path 1: AMS remain% delta tracking."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_sessions(self):
|
|
_active_sessions.clear()
|
|
yield
|
|
_active_sessions.clear()
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _mock_get_setting(self):
|
|
with patch(
|
|
"backend.app.api.routes.settings.get_setting",
|
|
new_callable=AsyncMock,
|
|
return_value=None,
|
|
):
|
|
yield
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_computes_delta_and_updates_spool(self):
|
|
"""Spool weight_used updated by remain% delta * label_weight."""
|
|
# Set up session with start remain = 80%
|
|
_active_sessions[1] = PrintSession(
|
|
printer_id=1,
|
|
print_name="test",
|
|
started_at=datetime.now(timezone.utc),
|
|
tray_remain_start={(0, 0): 80},
|
|
)
|
|
|
|
# Current remain = 70% → 10% consumed → 100g on 1000g spool
|
|
ams_data = [{"id": 0, "tray": [{"id": 0, "remain": 70}]}]
|
|
pm = _make_printer_manager(_make_printer_state(ams_data))
|
|
|
|
spool = _make_spool(label_weight=1000, weight_used=50)
|
|
assignment = _make_assignment()
|
|
|
|
db = AsyncMock()
|
|
# First 2 executes → _find_3mf_by_filename (library + archive search, uses scalars().all()),
|
|
# then assignment, then spool for the AMS fallback path
|
|
db.execute = AsyncMock(
|
|
side_effect=[
|
|
MagicMock(), # _find_3mf_by_filename: library search
|
|
MagicMock(), # _find_3mf_by_filename: archive search
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=assignment)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
|
|
]
|
|
)
|
|
|
|
results = await on_print_complete(1, {"status": "completed"}, pm, db)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["weight_used"] == 100.0
|
|
assert results[0]["percent_used"] == 10
|
|
# weight_used should be old (50) + delta (100)
|
|
assert spool.weight_used == 150.0
|
|
db.commit.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skips_negative_delta(self):
|
|
"""No tracking when remain increased (spool refilled)."""
|
|
_active_sessions[1] = PrintSession(
|
|
printer_id=1,
|
|
print_name="test",
|
|
started_at=datetime.now(timezone.utc),
|
|
tray_remain_start={(0, 0): 50},
|
|
)
|
|
|
|
# Remain went UP: 50 → 80 (refilled)
|
|
ams_data = [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]
|
|
pm = _make_printer_manager(_make_printer_state(ams_data))
|
|
db = AsyncMock()
|
|
|
|
results = await on_print_complete(1, {"status": "completed"}, pm, db)
|
|
|
|
assert results == []
|
|
db.commit.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_an_unloaded_tray_at_start_does_not_exclude_every_slot(self):
|
|
"""tray_now reads 255 at rest -- its initial value, and what an
|
|
unparseable reading falls back to. Mapped as a tray id that is (255, 1),
|
|
so taking it as evidence of which slots the print used would exclude
|
|
every real one and charge nothing (#1820)."""
|
|
_active_sessions[1] = PrintSession(
|
|
printer_id=1,
|
|
print_name="test",
|
|
started_at=datetime.now(timezone.utc),
|
|
tray_remain_start={(0, 0): 80},
|
|
tray_now_at_start=255,
|
|
)
|
|
ams_data = [{"id": 0, "tray": [{"id": 0, "remain": 70}]}]
|
|
pm = _make_printer_manager(_make_printer_state(ams_data))
|
|
|
|
spool = _make_spool(label_weight=1000, weight_used=0)
|
|
db = AsyncMock()
|
|
db.execute = AsyncMock(
|
|
side_effect=[
|
|
MagicMock(), # _find_3mf_by_filename: library search
|
|
MagicMock(), # _find_3mf_by_filename: archive search
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=_make_assignment())),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
|
|
]
|
|
)
|
|
|
|
results = await on_print_complete(1, {"status": "completed"}, pm, db)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["weight_used"] == 100.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_delta_that_charges_nothing_says_so(self, caplog):
|
|
"""Charging nothing has to be distinguishable from having nothing to
|
|
charge (#1820). A fresh spool reads 100% for its first tens of grams
|
|
and the AMS estimate drifts upward on its own, so this fires on real
|
|
prints, not only on refills -- and used to fire in complete silence."""
|
|
import logging
|
|
|
|
_active_sessions[1] = PrintSession(
|
|
printer_id=1,
|
|
print_name="test",
|
|
started_at=datetime.now(timezone.utc),
|
|
tray_remain_start={(0, 0): 100},
|
|
)
|
|
ams_data = [{"id": 0, "tray": [{"id": 0, "remain": 100}]}]
|
|
pm = _make_printer_manager(_make_printer_state(ams_data))
|
|
db = AsyncMock()
|
|
|
|
with caplog.at_level(logging.INFO, logger="backend.app.services.usage_tracker"):
|
|
results = await on_print_complete(1, {"status": "completed"}, pm, db)
|
|
|
|
assert results == []
|
|
assert "did not fall" in caplog.text
|
|
assert "100% -> 100%" in caplog.text
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_session_falls_through_to_3mf(self):
|
|
"""When no session exists, AMS delta path skipped (3MF may still run)."""
|
|
pm = _make_printer_manager()
|
|
db = AsyncMock()
|
|
|
|
results = await on_print_complete(1, {"status": "completed"}, pm, db)
|
|
|
|
assert results == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skips_fallback_for_trays_outside_print_mapping(self):
|
|
"""#1269: swapping a spool in an UNUSED slot mid-print must NOT charge the old spool.
|
|
|
|
Reproduces maugsburger's report: single-color print on AMS0-T3
|
|
(ams_mapping=[3]). User swaps spools in T1 and T2 during the print —
|
|
those slots report remain=0 at completion (new spool with no tag).
|
|
The fallback must skip T1 and T2 because they were never in the
|
|
print's tray mapping or runtime tray_change_log.
|
|
"""
|
|
_active_sessions[1] = PrintSession(
|
|
printer_id=1,
|
|
print_name="splitter",
|
|
started_at=datetime.now(timezone.utc),
|
|
tray_remain_start={(0, 1): 100, (0, 2): 17, (0, 3): 100},
|
|
tray_now_at_start=3,
|
|
ams_mapping=[3],
|
|
)
|
|
|
|
# User swapped T1 and T2 mid-print → both report remain=0 now.
|
|
# T3 was actually used but it's also at 0 now. Without the fix the
|
|
# fallback would charge the originally-assigned spools at T1 and T2.
|
|
ams_data = [
|
|
{
|
|
"id": 0,
|
|
"tray": [
|
|
{"id": 1, "remain": 0},
|
|
{"id": 2, "remain": 0},
|
|
{"id": 3, "remain": 0},
|
|
],
|
|
}
|
|
]
|
|
state = _make_printer_state(ams_data, tray_now=3)
|
|
state.tray_change_log = [(3, 0)] # only T3 was loaded during the print
|
|
pm = _make_printer_manager(state)
|
|
|
|
# Only T3 should reach the spool lookup; T1 and T2 must be filtered
|
|
# out before any DB query is issued for them.
|
|
t3_spool = _make_spool(id=8, label_weight=1000, weight_used=0)
|
|
t3_assignment = _make_assignment(spool_id=8, ams_id=0, tray_id=3)
|
|
db = AsyncMock()
|
|
db.execute = AsyncMock(
|
|
side_effect=[
|
|
MagicMock(), # _find_3mf_by_filename: library search
|
|
MagicMock(), # _find_3mf_by_filename: archive search
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=t3_assignment)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=t3_spool)),
|
|
]
|
|
)
|
|
|
|
results = await on_print_complete(1, {"status": "completed"}, pm, db)
|
|
|
|
# Only T3 should be charged. T1 (spool 27 in the report) and T2
|
|
# (spool 24) must NOT appear in the results.
|
|
assert len(results) == 1
|
|
assert results[0]["ams_id"] == 0
|
|
assert results[0]["tray_id"] == 3
|
|
|
|
|
|
class TestTrackFrom3MF:
|
|
"""Tests for Path 2: 3MF per-filament fallback tracking."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_updates_non_bl_spool_from_3mf(self):
|
|
"""Non-BL spool gets weight_used from 3MF used_g for completed print."""
|
|
spool = _make_spool(id=5, label_weight=1000, weight_used=100)
|
|
assignment = _make_assignment(spool_id=5)
|
|
archive = MagicMock()
|
|
archive.file_path = "archives/test.3mf"
|
|
|
|
db = AsyncMock()
|
|
# archive, queue_item(None), assignment, spool
|
|
db.execute = AsyncMock(
|
|
side_effect=[
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=archive)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=None)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=assignment)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
|
|
]
|
|
)
|
|
|
|
pm = _make_printer_manager(_make_printer_state([], tray_now=0))
|
|
filament_usage = [{"slot_id": 1, "used_g": 25.5, "type": "PLA", "color": "#FF0000"}]
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=filament_usage),
|
|
):
|
|
mock_path = MagicMock()
|
|
mock_path.exists.return_value = True
|
|
mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
|
results = await _track_from_3mf(
|
|
printer_id=1,
|
|
archive_id=10,
|
|
status="completed",
|
|
print_name="test_print",
|
|
handled_trays=set(),
|
|
printer_manager=pm,
|
|
db=db,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["spool_id"] == 5
|
|
assert results[0]["weight_used"] == 25.5
|
|
# weight_used = old (100) + 3MF (25.5)
|
|
assert spool.weight_used == 125.5
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scales_by_progress_for_failed_print(self):
|
|
"""Failed print scales 3MF estimate by progress percentage."""
|
|
spool = _make_spool(id=1, label_weight=1000, weight_used=0)
|
|
assignment = _make_assignment()
|
|
archive = MagicMock()
|
|
archive.file_path = "archives/test.3mf"
|
|
|
|
db = AsyncMock()
|
|
# archive, queue_item(None), assignment, spool
|
|
db.execute = AsyncMock(
|
|
side_effect=[
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=archive)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=None)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=assignment)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
|
|
]
|
|
)
|
|
|
|
# Print failed at 50% progress → 50g consumed from 100g estimate
|
|
pm = _make_printer_manager(_make_printer_state([], progress=50, tray_now=0))
|
|
filament_usage = [{"slot_id": 1, "used_g": 100.0, "type": "PLA", "color": ""}]
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=filament_usage),
|
|
):
|
|
mock_path = MagicMock()
|
|
mock_path.exists.return_value = True
|
|
mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
|
results = await _track_from_3mf(
|
|
printer_id=1,
|
|
archive_id=10,
|
|
status="failed",
|
|
print_name="test",
|
|
handled_trays=set(),
|
|
printer_manager=pm,
|
|
db=db,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["weight_used"] == 50.0
|
|
assert spool.weight_used == 50.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tracks_bl_spools_via_3mf(self):
|
|
"""BL spools (with tag_uid) ARE now tracked via 3MF (unified tracking)."""
|
|
spool = _make_spool(tag_uid="ABCD1234", tray_uuid="A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4")
|
|
assignment = _make_assignment()
|
|
archive = MagicMock()
|
|
archive.file_path = "archives/test.3mf"
|
|
|
|
db = AsyncMock()
|
|
# archive, queue_item(None), assignment, spool
|
|
db.execute = AsyncMock(
|
|
side_effect=[
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=archive)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=None)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=assignment)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
|
|
]
|
|
)
|
|
|
|
pm = _make_printer_manager(_make_printer_state([], tray_now=0))
|
|
filament_usage = [{"slot_id": 1, "used_g": 50.0, "type": "PLA", "color": ""}]
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=filament_usage),
|
|
):
|
|
mock_path = MagicMock()
|
|
mock_path.exists.return_value = True
|
|
mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
|
results = await _track_from_3mf(
|
|
printer_id=1,
|
|
archive_id=10,
|
|
status="completed",
|
|
print_name="test",
|
|
handled_trays=set(),
|
|
printer_manager=pm,
|
|
db=db,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["spool_id"] == 1
|
|
assert results[0]["weight_used"] == 50.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skips_already_handled_trays(self):
|
|
"""Trays handled by AMS remain% delta are not double-tracked via 3MF."""
|
|
archive = MagicMock()
|
|
archive.file_path = "archives/test.3mf"
|
|
|
|
db = AsyncMock()
|
|
# archive, queue_item(None)
|
|
db.execute = AsyncMock(
|
|
side_effect=[
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=archive)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=None)),
|
|
]
|
|
)
|
|
|
|
pm = _make_printer_manager(_make_printer_state([], tray_now=0))
|
|
filament_usage = [{"slot_id": 1, "used_g": 50.0, "type": "PLA", "color": ""}]
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=filament_usage),
|
|
):
|
|
mock_path = MagicMock()
|
|
mock_path.exists.return_value = True
|
|
mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
|
results = await _track_from_3mf(
|
|
printer_id=1,
|
|
archive_id=10,
|
|
status="completed",
|
|
print_name="test",
|
|
handled_trays={(0, 0)}, # slot_id=1 → ams_id=0, tray_id=0
|
|
printer_manager=pm,
|
|
db=db,
|
|
)
|
|
|
|
assert results == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_slot_to_tray_mapping(self):
|
|
"""3MF slot_id maps correctly to (ams_id, tray_id) via tray_now."""
|
|
# tray_now=4 → ams_id=1, tray_id=0 (single filament uses tray_now)
|
|
spool = _make_spool(id=9)
|
|
assignment = _make_assignment(spool_id=9, ams_id=1, tray_id=0)
|
|
archive = MagicMock()
|
|
archive.file_path = "archives/test.3mf"
|
|
|
|
db = AsyncMock()
|
|
# archive, queue_item(None), assignment, spool
|
|
db.execute = AsyncMock(
|
|
side_effect=[
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=archive)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=None)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=assignment)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
|
|
]
|
|
)
|
|
|
|
pm = _make_printer_manager(_make_printer_state([], tray_now=4))
|
|
filament_usage = [{"slot_id": 5, "used_g": 30.0, "type": "PETG", "color": ""}]
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=filament_usage),
|
|
):
|
|
mock_path = MagicMock()
|
|
mock_path.exists.return_value = True
|
|
mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
|
results = await _track_from_3mf(
|
|
printer_id=1,
|
|
archive_id=10,
|
|
status="completed",
|
|
print_name="test",
|
|
handled_trays=set(),
|
|
printer_manager=pm,
|
|
db=db,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["ams_id"] == 1
|
|
assert results[0]["tray_id"] == 0
|
|
|
|
|
|
class TestSpoolAssignmentSnapshot:
|
|
"""Tests for spool assignment snapshotting at print start (#459).
|
|
|
|
When a spool runs empty mid-print, on_ams_change deletes the SpoolAssignment.
|
|
The snapshot captured at print start ensures usage is still attributed correctly.
|
|
"""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_sessions(self):
|
|
_active_sessions.clear()
|
|
yield
|
|
_active_sessions.clear()
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _mock_get_setting(self):
|
|
with patch(
|
|
"backend.app.api.routes.settings.get_setting",
|
|
new_callable=AsyncMock,
|
|
return_value=None,
|
|
):
|
|
yield
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_print_start_snapshots_assignments_with_db(self):
|
|
"""on_print_start captures spool assignments when db is provided."""
|
|
ams_data = [{"id": 0, "tray": [{"id": 0, "remain": 80}, {"id": 1, "remain": 60}]}]
|
|
pm = _make_printer_manager(_make_printer_state(ams_data, tray_now=0))
|
|
|
|
assignment_0 = _make_assignment(spool_id=10, printer_id=1, ams_id=0, tray_id=0)
|
|
assignment_1 = _make_assignment(spool_id=20, printer_id=1, ams_id=0, tray_id=1)
|
|
|
|
db = AsyncMock()
|
|
scalars_mock = MagicMock()
|
|
scalars_mock.all.return_value = [assignment_0, assignment_1]
|
|
result_mock = MagicMock()
|
|
result_mock.scalars.return_value = scalars_mock
|
|
db.execute = AsyncMock(return_value=result_mock)
|
|
|
|
await on_print_start(1, {"subtask_name": "Benchy"}, pm, db=db)
|
|
|
|
session = _active_sessions[1]
|
|
assert session.spool_assignments == {(0, 0): 10, (0, 1): 20}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_print_start_empty_snapshot_without_db(self):
|
|
"""on_print_start creates empty snapshot when no db provided."""
|
|
ams_data = [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]
|
|
pm = _make_printer_manager(_make_printer_state(ams_data, tray_now=0))
|
|
|
|
await on_print_start(1, {"subtask_name": "Benchy"}, pm)
|
|
|
|
session = _active_sessions[1]
|
|
assert session.spool_assignments == {}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_3mf_uses_snapshot_instead_of_live_query(self):
|
|
"""_track_from_3mf uses snapshot spool_id without querying SpoolAssignment."""
|
|
spool = _make_spool(id=42, label_weight=1000)
|
|
archive = MagicMock()
|
|
archive.file_path = "archives/test.3mf"
|
|
|
|
# db: archive, queue_item(None), spool — NO assignment query needed
|
|
db = AsyncMock()
|
|
db.execute = AsyncMock(
|
|
side_effect=[
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=archive)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=None)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
|
|
]
|
|
)
|
|
|
|
pm = _make_printer_manager(_make_printer_state([], tray_now=0))
|
|
filament_usage = [{"slot_id": 1, "used_g": 15.0, "type": "PLA", "color": "#FF0000"}]
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=filament_usage),
|
|
):
|
|
mock_path = MagicMock()
|
|
mock_path.exists.return_value = True
|
|
mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
|
results = await _track_from_3mf(
|
|
printer_id=1,
|
|
archive_id=10,
|
|
status="completed",
|
|
print_name="Test",
|
|
handled_trays=set(),
|
|
printer_manager=pm,
|
|
db=db,
|
|
spool_assignments={(0, 0): 42},
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["spool_id"] == 42
|
|
assert results[0]["weight_used"] == 15.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_3mf_falls_back_to_live_query_without_snapshot(self):
|
|
"""_track_from_3mf queries SpoolAssignment when no snapshot exists."""
|
|
spool = _make_spool(id=5, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=5)
|
|
archive = MagicMock()
|
|
archive.file_path = "archives/test.3mf"
|
|
|
|
# db: archive, queue_item(None), assignment, spool
|
|
db = AsyncMock()
|
|
db.execute = AsyncMock(
|
|
side_effect=[
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=archive)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=None)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=assignment)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
|
|
]
|
|
)
|
|
|
|
pm = _make_printer_manager(_make_printer_state([], tray_now=0))
|
|
filament_usage = [{"slot_id": 1, "used_g": 10.0, "type": "PLA", "color": "#FF0000"}]
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=filament_usage),
|
|
):
|
|
mock_path = MagicMock()
|
|
mock_path.exists.return_value = True
|
|
mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
|
results = await _track_from_3mf(
|
|
printer_id=1,
|
|
archive_id=10,
|
|
status="completed",
|
|
print_name="Test",
|
|
handled_trays=set(),
|
|
printer_manager=pm,
|
|
db=db,
|
|
spool_assignments=None,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["spool_id"] == 5
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ams_delta_uses_snapshot_over_live_query(self):
|
|
"""AMS remain% fallback uses snapshot spool_id instead of live query."""
|
|
spool = _make_spool(id=77, label_weight=1000)
|
|
|
|
_active_sessions[1] = PrintSession(
|
|
printer_id=1,
|
|
print_name="Benchy",
|
|
started_at=datetime.now(timezone.utc),
|
|
tray_remain_start={(0, 0): 80},
|
|
spool_assignments={(0, 0): 77},
|
|
)
|
|
|
|
# Current remain = 70% → 10% delta → 100g
|
|
ams_data = [{"id": 0, "tray": [{"id": 0, "remain": 70}]}]
|
|
pm = _make_printer_manager(_make_printer_state(ams_data))
|
|
|
|
# First 2 executes → _find_3mf_by_filename (library + archive search),
|
|
# then live assignment check (returns None), then spool lookup by snapshot spool_id
|
|
db = AsyncMock()
|
|
db.execute = AsyncMock(
|
|
side_effect=[
|
|
MagicMock(), # _find_3mf_by_filename: library search
|
|
MagicMock(), # _find_3mf_by_filename: archive search
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=None)), # live assignment
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
|
|
]
|
|
)
|
|
|
|
results = await on_print_complete(
|
|
printer_id=1,
|
|
data={"status": "completed"},
|
|
printer_manager=pm,
|
|
db=db,
|
|
archive_id=None,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["spool_id"] == 77
|
|
assert results[0]["weight_used"] == 100.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ams_delta_falls_back_to_live_query_without_snapshot(self):
|
|
"""AMS remain% fallback queries SpoolAssignment when snapshot is empty."""
|
|
spool = _make_spool(id=33, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=33)
|
|
|
|
_active_sessions[1] = PrintSession(
|
|
printer_id=1,
|
|
print_name="Benchy",
|
|
started_at=datetime.now(timezone.utc),
|
|
tray_remain_start={(0, 0): 80},
|
|
spool_assignments={}, # Empty snapshot (pre-upgrade session)
|
|
)
|
|
|
|
ams_data = [{"id": 0, "tray": [{"id": 0, "remain": 70}]}]
|
|
pm = _make_printer_manager(_make_printer_state(ams_data))
|
|
|
|
# First 2 executes → _find_3mf_by_filename (library + archive search),
|
|
# then assignment and spool for the AMS fallback path
|
|
db = AsyncMock()
|
|
db.execute = AsyncMock(
|
|
side_effect=[
|
|
MagicMock(), # _find_3mf_by_filename: library search
|
|
MagicMock(), # _find_3mf_by_filename: archive search
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=assignment)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
|
|
]
|
|
)
|
|
|
|
results = await on_print_complete(
|
|
printer_id=1,
|
|
data={"status": "completed"},
|
|
printer_manager=pm,
|
|
db=db,
|
|
archive_id=None,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["spool_id"] == 33
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snapshot_survives_mid_print_unlink(self):
|
|
"""Core bug scenario: snapshot provides spool_id after mid-print unlink.
|
|
|
|
Simulates the #459 scenario: spool runs empty mid-print, on_ams_change
|
|
deletes the SpoolAssignment, but the snapshot from print start still
|
|
has the spool_id so usage is correctly attributed at print completion.
|
|
"""
|
|
spool = _make_spool(id=8, label_weight=1000, weight_used=50)
|
|
archive = MagicMock()
|
|
archive.file_path = "archives/big_print.3mf"
|
|
# Explicit numeric so the #1344 top-up branch doesn't trip a
|
|
# MagicMock-vs-float comparison.
|
|
archive.filament_used_grams = 14.2
|
|
|
|
# Session was created at print start WITH snapshot
|
|
_active_sessions[1] = PrintSession(
|
|
printer_id=1,
|
|
print_name="Big Print",
|
|
started_at=datetime.now(timezone.utc),
|
|
tray_remain_start={(0, 0): 90},
|
|
spool_assignments={(0, 0): 8}, # Snapshot from print start
|
|
)
|
|
|
|
pm = _make_printer_manager(
|
|
_make_printer_state(
|
|
[{"id": 0, "tray": [{"id": 0, "remain": 75}]}],
|
|
tray_now=0,
|
|
)
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 14.2, "type": "PLA", "color": "#FF0000"}]
|
|
|
|
# db: archive, queue_item(None), live assignment(None), spool,
|
|
# then cost aggregation queries
|
|
# NOTE: No assignment in db — it was deleted by on_ams_change mid-print!
|
|
db = AsyncMock()
|
|
db.execute = AsyncMock(
|
|
side_effect=[
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=archive)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=None)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=None)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
|
|
# Cost-update block re-selects the archive to mutate cost.
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=archive)),
|
|
]
|
|
)
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=filament_usage),
|
|
):
|
|
mock_path = MagicMock()
|
|
mock_path.exists.return_value = True
|
|
mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
|
results = await on_print_complete(
|
|
printer_id=1,
|
|
data={"status": "completed"},
|
|
printer_manager=pm,
|
|
db=db,
|
|
archive_id=100,
|
|
)
|
|
|
|
# Usage should be tracked despite assignment being deleted mid-print
|
|
assert len(results) >= 1
|
|
assert results[0]["spool_id"] == 8
|
|
assert results[0]["weight_used"] == 14.2
|
|
# Spool weight should be updated: 50 + 14.2 = 64.2
|
|
assert spool.weight_used == 64.2
|
|
|
|
|
|
class TestSpoolColorToHex:
|
|
"""`_spool_color_to_hex` normalises Spool.rgba (RRGGBBAA, no #) to #RRGGBB."""
|
|
|
|
def test_strips_alpha_and_adds_hash(self):
|
|
assert _spool_color_to_hex("000000FF") == "#000000"
|
|
assert _spool_color_to_hex("EC984CFF") == "#EC984C"
|
|
|
|
def test_uppercases(self):
|
|
assert _spool_color_to_hex("ec984cff") == "#EC984C"
|
|
|
|
def test_accepts_six_char_value(self):
|
|
"""A value with no alpha is still valid."""
|
|
assert _spool_color_to_hex("161616") == "#161616"
|
|
|
|
def test_tolerates_leading_hash(self):
|
|
assert _spool_color_to_hex("#000000FF") == "#000000"
|
|
|
|
def test_none_and_too_short_return_none(self):
|
|
"""Missing / malformed colour falls back to the 3MF value."""
|
|
assert _spool_color_to_hex(None) is None
|
|
assert _spool_color_to_hex("") is None
|
|
assert _spool_color_to_hex("FFF") is None
|
|
|
|
|
|
class TestArchiveColorsFromSpools:
|
|
"""`_archive_colors_from_spools` rebuilds an archive's filament_color from
|
|
the inventory spools that fed the print (#1494). All-or-nothing: a partial
|
|
match returns None so the 3MF colour is left intact."""
|
|
|
|
def test_single_slot_matched(self):
|
|
"""The #1494 case: one used slot, matched to a #000000 spool."""
|
|
usage = [{"slot_id": 1, "used_g": 15.9, "color": "#161616"}]
|
|
results = [{"slot_id": 1, "color": "#000000"}]
|
|
assert _archive_colors_from_spools(usage, results) == ["#000000"]
|
|
|
|
def test_multi_slot_all_matched_keeps_slot_order(self):
|
|
usage = [
|
|
{"slot_id": 1, "used_g": 10.0, "color": "#111111"},
|
|
{"slot_id": 2, "used_g": 20.0, "color": "#222222"},
|
|
]
|
|
# results deliberately out of slot order — output must be slot-ordered
|
|
results = [
|
|
{"slot_id": 2, "color": "#00FF00"},
|
|
{"slot_id": 1, "color": "#FF0000"},
|
|
]
|
|
assert _archive_colors_from_spools(usage, results) == ["#FF0000", "#00FF00"]
|
|
|
|
def test_duplicate_colors_deduplicated(self):
|
|
"""Two slots of the same spool colour collapse to one entry, as the
|
|
3MF-derived path also de-duplicates."""
|
|
usage = [
|
|
{"slot_id": 1, "used_g": 10.0, "color": "#111111"},
|
|
{"slot_id": 2, "used_g": 20.0, "color": "#222222"},
|
|
]
|
|
results = [
|
|
{"slot_id": 1, "color": "#000000"},
|
|
{"slot_id": 2, "color": "#000000"},
|
|
]
|
|
assert _archive_colors_from_spools(usage, results) == ["#000000"]
|
|
|
|
def test_partial_match_returns_none(self):
|
|
"""Slot 2 was used but never matched to a spool — leave the 3MF colour
|
|
untouched rather than dropping slot 2 from the archive."""
|
|
usage = [
|
|
{"slot_id": 1, "used_g": 10.0, "color": "#111111"},
|
|
{"slot_id": 2, "used_g": 20.0, "color": "#222222"},
|
|
]
|
|
results = [{"slot_id": 1, "color": "#000000"}]
|
|
assert _archive_colors_from_spools(usage, results) is None
|
|
|
|
def test_matched_spool_without_color_returns_none(self):
|
|
"""A spool with no rgba (color None) does not count as matched."""
|
|
usage = [{"slot_id": 1, "used_g": 15.0, "color": "#161616"}]
|
|
results = [{"slot_id": 1, "color": None}]
|
|
assert _archive_colors_from_spools(usage, results) is None
|
|
|
|
def test_unused_slot_not_required(self):
|
|
"""A slot with zero usage need not be matched."""
|
|
usage = [
|
|
{"slot_id": 1, "used_g": 15.0, "color": "#161616"},
|
|
{"slot_id": 2, "used_g": 0.0, "color": "#888888"},
|
|
]
|
|
results = [{"slot_id": 1, "color": "#000000"}]
|
|
assert _archive_colors_from_spools(usage, results) == ["#000000"]
|
|
|
|
def test_no_used_slots_returns_none(self):
|
|
assert _archive_colors_from_spools([], []) is None
|
|
|
|
def test_ams_fallback_results_excluded(self):
|
|
"""AMS remain%-delta fallback results carry slot_id=None and must not
|
|
satisfy the match for a real 3MF slot."""
|
|
usage = [{"slot_id": 1, "used_g": 15.0, "color": "#161616"}]
|
|
results = [{"slot_id": None, "color": "#000000"}]
|
|
assert _archive_colors_from_spools(usage, results) is None
|
|
|
|
|
|
class TestArchiveFilamentColorRewrite:
|
|
"""`_track_from_3mf` overwrites the archive's filament_color with the
|
|
matched inventory spool colour at print completion (#1494)."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archive_color_adopts_spool_color(self):
|
|
"""A print from a #000000 inventory spool whose 3MF says #161616 ends
|
|
up with the archive showing the spool's #000000."""
|
|
spool = _make_spool(id=5, label_weight=1000, weight_used=100, rgba="000000FF")
|
|
assignment = _make_assignment(spool_id=5)
|
|
archive = MagicMock()
|
|
archive.file_path = "archives/test.3mf"
|
|
archive.filament_color = "#161616" # what archive.py set from the 3MF
|
|
|
|
db = AsyncMock()
|
|
db.execute = AsyncMock(
|
|
side_effect=[
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=archive)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=None)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=assignment)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
|
|
]
|
|
)
|
|
|
|
pm = _make_printer_manager(_make_printer_state([], tray_now=0))
|
|
filament_usage = [{"slot_id": 1, "used_g": 25.5, "type": "PETG", "color": "#161616"}]
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=filament_usage),
|
|
):
|
|
mock_path = MagicMock()
|
|
mock_path.exists.return_value = True
|
|
mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
|
results = await _track_from_3mf(
|
|
printer_id=1,
|
|
archive_id=10,
|
|
status="completed",
|
|
print_name="test_print",
|
|
handled_trays=set(),
|
|
printer_manager=pm,
|
|
db=db,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["color"] == "#000000"
|
|
assert results[0]["slot_id"] == 1
|
|
# The archive colour was rewritten from the slicer's #161616 to the
|
|
# inventory spool's #000000.
|
|
assert archive.filament_color == "#000000"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archive_color_untouched_when_spool_has_no_color(self):
|
|
"""A spool with no rgba leaves the 3MF colour in place."""
|
|
spool = _make_spool(id=5, label_weight=1000, weight_used=100, rgba=None)
|
|
assignment = _make_assignment(spool_id=5)
|
|
archive = MagicMock()
|
|
archive.file_path = "archives/test.3mf"
|
|
archive.filament_color = "#161616"
|
|
|
|
db = AsyncMock()
|
|
db.execute = AsyncMock(
|
|
side_effect=[
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=archive)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=None)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=assignment)),
|
|
MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
|
|
]
|
|
)
|
|
|
|
pm = _make_printer_manager(_make_printer_state([], tray_now=0))
|
|
filament_usage = [{"slot_id": 1, "used_g": 25.5, "type": "PETG", "color": "#161616"}]
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=filament_usage),
|
|
):
|
|
mock_path = MagicMock()
|
|
mock_path.exists.return_value = True
|
|
mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
|
await _track_from_3mf(
|
|
printer_id=1,
|
|
archive_id=10,
|
|
status="completed",
|
|
print_name="test_print",
|
|
handled_trays=set(),
|
|
printer_manager=pm,
|
|
db=db,
|
|
)
|
|
|
|
assert archive.filament_color == "#161616"
|