bambuddy/backend/tests/unit/services/test_usage_tracker.py

"""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,
)


async def _charge_in_memory(_db, spool, grams):
    """The fake session has no database for the tracker's UPDATE: add the
    grams to the mock spool, which is what these tests check. The statement
    itself is tested against a real database in test_usage_tracker_charge_race."""
    spool.weight_used = (spool.weight_used or 0) + grams
    spool.last_used = datetime.now(timezone.utc)
    return True


@pytest.fixture(autouse=True)
def _charge_spools_in_memory():
    with patch("backend.app.services.usage_tracker._charge_spool", _charge_in_memory):
        yield


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)),
                # Priced at the default rate, so it asks whether this is the
                # archive's first run before storing the cost.
                MagicMock(scalar=MagicMock(return_value=0)),
            ]
        )

        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"