bambuddy/backend/tests/unit/test_slicer_presets.py

"""Tests for the unified slicer-presets endpoint helpers.

The endpoint stitches together four preset sources (local / orca_cloud /
cloud / standard). It does NOT dedup across tiers — every tier surfaces
its full list so the user can pick any source. Bambu Cloud filament
metadata is enriched from same-named entries in the other tiers so it
can still score in the SliceModal's auto-pick. These tests pin the
enrich behaviour, the cloud-status mapping, and the per-user / sidecar
caches at the helper level — full HTTP integration is covered by the
routes test.
"""

from __future__ import annotations

import time
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from backend.app.api.routes import slicer_presets as sp
from backend.app.schemas.slicer_presets import UnifiedPreset


def _slot(items: list[tuple[str, str, str]]) -> dict[str, list[UnifiedPreset]]:
    """Helper: build a single-slot dict from (id, name, source) tuples placed
    on the printer slot. Process / filament default to empty so each test
    only exercises the slot it cares about."""
    return {
        "printer": [UnifiedPreset(id=i, name=n, source=s) for i, n, s in items],
        "process": [],
        "filament": [],
    }


class TestEnrichCloudMetadata:
    """No cross-tier dedup — every tier's full list comes back; Bambu Cloud
    filament metadata is enriched from same-named entries in other tiers."""

    def test_same_name_in_all_tiers_appears_in_every_tier(self):
        """Critical regression guard for #1712: a user who has imported a
        local profile AND signed in to Orca AND has Bambu Cloud with the
        same name should see it under EACH source, not just the highest-
        priority tier. The order is used for auto-pick + group rendering;
        it is NOT used to hide profiles."""
        orca = _slot([("oid1", "Bambu PLA Basic", "orca_cloud")])
        cloud = _slot([("cid1", "Bambu PLA Basic", "cloud")])
        local = _slot([("lid1", "Bambu PLA Basic", "local")])
        standard = _slot([("Bambu PLA Basic", "Bambu PLA Basic", "standard")])

        oc, c, l_, s = sp._enrich_cloud_metadata(orca, cloud, local, standard)

        assert [p.source for p in l_["printer"]] == ["local"]
        assert [p.source for p in oc["printer"]] == ["orca_cloud"]
        assert [p.source for p in c["printer"]] == ["cloud"]
        assert [p.source for p in s["printer"]] == ["standard"]

    def test_preserves_order_within_tier(self):
        """A tier's input order must be preserved — nothing in the enrich
        pass should sort, reverse, or otherwise reorder entries."""
        cloud = _slot(
            [
                ("c1", "Z-First", "cloud"),
                ("c2", "A-Second", "cloud"),
                ("c3", "M-Third", "cloud"),
            ]
        )
        _oc, c, _l, _s = sp._enrich_cloud_metadata(_slot([]), cloud, _slot([]), _slot([]))
        assert [p.name for p in c["printer"]] == ["Z-First", "A-Second", "M-Third"]

    def test_bambu_cloud_filament_metadata_backfilled_from_local(self):
        """Bambu Cloud's list response omits filament_type/colour for
        rate-limit reasons. A same-named local entry's metadata fills in
        so the cloud entry can still score in pickFilamentForSlot."""
        local = {
            "printer": [],
            "process": [],
            "filament": [
                UnifiedPreset(
                    id="lp1",
                    name="Bambu PLA Basic",
                    source="local",
                    filament_type="PLA",
                    filament_colour="#FF0000",
                )
            ],
        }
        cloud = {
            "printer": [],
            "process": [],
            "filament": [UnifiedPreset(id="cp1", name="Bambu PLA Basic", source="cloud")],
        }
        _oc, c, _l, _s = sp._enrich_cloud_metadata(_slot([]), cloud, local, _slot([]))
        # Cloud entry now carries the local entry's metadata.
        assert c["filament"][0].filament_type == "PLA"
        assert c["filament"][0].filament_colour == "#FF0000"
        # Local entry is untouched.
        assert local["filament"][0].filament_type == "PLA"

    def test_bambu_cloud_metadata_falls_back_through_orca_and_standard(self):
        """When local doesn't carry the name, orca_cloud / standard fill in."""
        orca = {
            "printer": [],
            "process": [],
            "filament": [
                UnifiedPreset(
                    id="o1",
                    name="Bambu PLA Basic",
                    source="orca_cloud",
                    filament_type="PLA",
                    filament_colour="#00FF00",
                )
            ],
        }
        cloud = {
            "printer": [],
            "process": [],
            "filament": [UnifiedPreset(id="cp1", name="Bambu PLA Basic", source="cloud")],
        }
        _oc, c, _l, _s = sp._enrich_cloud_metadata(orca, cloud, _slot([]), _slot([]))
        assert c["filament"][0].filament_type == "PLA"
        assert c["filament"][0].filament_colour == "#00FF00"

    def test_bambu_cloud_keeps_its_own_metadata_when_present(self):
        """If Bambu Cloud already has filament_type / filament_colour the
        enrich pass must not overwrite them with a different same-named
        entry's values."""
        local = {
            "printer": [],
            "process": [],
            "filament": [
                UnifiedPreset(
                    id="lp1",
                    name="Bambu PLA Basic",
                    source="local",
                    filament_type="PETG",
                    filament_colour="#000000",
                )
            ],
        }
        cloud = {
            "printer": [],
            "process": [],
            "filament": [
                UnifiedPreset(
                    id="cp1",
                    name="Bambu PLA Basic",
                    source="cloud",
                    filament_type="PLA",
                    filament_colour="#FFFFFF",
                )
            ],
        }
        _oc, c, _l, _s = sp._enrich_cloud_metadata(_slot([]), cloud, local, _slot([]))
        assert c["filament"][0].filament_type == "PLA"
        assert c["filament"][0].filament_colour == "#FFFFFF"


class TestEnrichCompatiblePrinters:
    """#2628: the same name bridge carries ``compatible_printers`` onto the
    tiers that don't ship one. Bambu Cloud never does — so a profile whose
    name carries no printer model reads as "compatibility unknown", which the
    SliceModal treats as usable and auto-picks for the wrong printer."""

    COMPAT = ["Bambu Lab X1 Carbon 0.2 nozzle", "Bambu Lab P1S 0.2 nozzle"]

    def _tier(self, source: str, slot: str, compat: list[str] | None) -> dict[str, list[UnifiedPreset]]:
        empty: dict[str, list[UnifiedPreset]] = {"printer": [], "process": [], "filament": []}
        empty[slot] = [
            UnifiedPreset(id=f"{source}1", name="Overture PLA Matte @0.2", source=source, compatible_printers=compat)
        ]
        return empty

    def test_bambu_cloud_borrows_the_list_from_a_same_named_local_import(self):
        local = self._tier("local", "filament", self.COMPAT)
        cloud = self._tier("cloud", "filament", None)

        _oc, c, _l, _s = sp._enrich_cloud_metadata(_slot([]), cloud, local, _slot([]))

        assert c["filament"][0].compatible_printers == self.COMPAT

    def test_bambu_cloud_borrows_from_orca_cloud_too(self):
        orca = self._tier("orca_cloud", "filament", self.COMPAT)
        cloud = self._tier("cloud", "filament", None)

        _oc, c, _l, _s = sp._enrich_cloud_metadata(orca, cloud, _slot([]), _slot([]))

        assert c["filament"][0].compatible_printers == self.COMPAT

    def test_orca_cloud_borrows_when_its_own_content_had_no_list(self):
        orca = self._tier("orca_cloud", "filament", None)
        local = self._tier("local", "filament", self.COMPAT)

        oc, _c, _l, _s = sp._enrich_cloud_metadata(orca, _slot([]), local, _slot([]))

        assert oc["filament"][0].compatible_printers == self.COMPAT

    def test_process_slot_is_bridged_as_well(self):
        local = self._tier("local", "process", self.COMPAT)
        cloud = self._tier("cloud", "process", None)

        _oc, c, _l, _s = sp._enrich_cloud_metadata(_slot([]), cloud, local, _slot([]))

        assert c["process"][0].compatible_printers == self.COMPAT

    def test_never_overwrites_a_list_the_entry_already_has(self):
        own = ["Bambu Lab P2S 0.4 nozzle"]
        local = self._tier("local", "filament", self.COMPAT)
        cloud = self._tier("cloud", "filament", own)

        _oc, c, _l, _s = sp._enrich_cloud_metadata(_slot([]), cloud, local, _slot([]))

        assert c["filament"][0].compatible_printers == own

    def test_no_donor_leaves_the_entry_unclassified(self):
        """Absent evidence the entry must stay None — the SliceModal then
        falls back to the name matcher instead of hiding the profile."""
        cloud = self._tier("cloud", "filament", None)

        _oc, c, _l, _s = sp._enrich_cloud_metadata(_slot([]), cloud, _slot([]), _slot([]))

        assert c["filament"][0].compatible_printers is None

    def test_borrowed_list_is_copied_not_shared(self):
        """A later mutation of one tier's list must not reach through to the
        other — these objects are cached per user between requests."""
        local = self._tier("local", "filament", list(self.COMPAT))
        cloud = self._tier("cloud", "filament", None)

        _oc, c, l_, _s = sp._enrich_cloud_metadata(_slot([]), cloud, local, _slot([]))
        l_["filament"][0].compatible_printers.append("Bambu Lab H2D 0.4 nozzle")

        assert c["filament"][0].compatible_printers == self.COMPAT


def _user_with_cloud_auth(user_id: int = 1) -> MagicMock:
    """Construct a mock User that passes the CLOUD_AUTH permission check.

    `MagicMock` defaults `.has_permission(...)` to a truthy MagicMock object,
    which would coincidentally pass the gate — but explicit is better than
    accidental. Setting `.return_value = True` documents the intent."""
    user = MagicMock(id=user_id)
    user.has_permission = MagicMock(return_value=True)
    return user


class TestFetchOrcaCloudPresets:
    """``_fetch_orca_cloud_presets`` mirrors the Bambu Cloud fetcher's status
    vocabulary (``ok`` / ``not_authenticated`` / ``expired`` / ``unreachable``)
    and the same permission-shortcut + caching behaviour. Tests pin the
    contract so a future bug in either fetcher doesn't silently desync them."""

    def _orca_creds(self, token: str | None = "tok") -> MagicMock:
        creds = MagicMock()
        creds.token = token
        return creds

    @pytest.mark.asyncio
    async def test_no_token_returns_not_authenticated(self):
        sp._orca_cloud_cache.clear()
        with patch.object(sp, "_load_orca_credentials", AsyncMock(return_value=self._orca_creds(None))):
            user = MagicMock(id=1)
            user.has_permission = MagicMock(return_value=True)
            slots, status = await sp._fetch_orca_cloud_presets(MagicMock(), user)
        assert status == "not_authenticated"
        assert slots == {"printer": [], "process": [], "filament": []}

    @pytest.mark.asyncio
    async def test_user_without_orca_cloud_auth_returns_not_authenticated(self):
        """Defence-in-depth — a user lacking ORCA_CLOUD_AUTH must not see Orca
        presets even if their User row carries a stale token. Credentials
        lookup must short-circuit ahead of the token read."""
        sp._orca_cloud_cache.clear()
        user = MagicMock(id=1)
        user.has_permission = MagicMock(return_value=False)
        with patch.object(sp, "_load_orca_credentials", AsyncMock(return_value=self._orca_creds("tok"))) as load:
            slots, status = await sp._fetch_orca_cloud_presets(MagicMock(), user)
        assert status == "not_authenticated"
        assert slots["printer"] == []
        load.assert_not_called()

    @pytest.mark.asyncio
    async def test_auth_error_returns_expired(self):
        sp._orca_cloud_cache.clear()
        svc_mock = MagicMock()
        svc_mock.list_profiles = AsyncMock(side_effect=sp.OrcaCloudAuthError("expired"))
        svc_mock.close = AsyncMock()
        user = MagicMock(id=1)
        user.has_permission = MagicMock(return_value=True)
        with (
            patch.object(sp, "_load_orca_credentials", AsyncMock(return_value=self._orca_creds("tok"))),
            patch.object(sp, "_build_orca_service", AsyncMock(return_value=svc_mock)),
        ):
            _slots, status = await sp._fetch_orca_cloud_presets(MagicMock(), user)
        assert status == "expired"
        svc_mock.close.assert_awaited_once()

    @pytest.mark.asyncio
    async def test_orca_error_returns_unreachable(self):
        sp._orca_cloud_cache.clear()
        svc_mock = MagicMock()
        svc_mock.list_profiles = AsyncMock(side_effect=sp.OrcaCloudError("net down"))
        svc_mock.close = AsyncMock()
        user = MagicMock(id=1)
        user.has_permission = MagicMock(return_value=True)
        with (
            patch.object(sp, "_load_orca_credentials", AsyncMock(return_value=self._orca_creds("tok"))),
            patch.object(sp, "_build_orca_service", AsyncMock(return_value=svc_mock)),
        ):
            _slots, status = await sp._fetch_orca_cloud_presets(MagicMock(), user)
        assert status == "unreachable"

    @pytest.mark.asyncio
    async def test_happy_path_shapes_grouped_by_type(self):
        """Orca content.type values map onto Bambu Cloud's preset type vocab
        (``printer`` / ``print`` → ``process`` / ``filament``). Verify the
        full mapping by feeding one of each shape."""
        sp._orca_cloud_cache.clear()
        svc_mock = MagicMock()
        svc_mock.list_profiles = AsyncMock(
            return_value=[
                {"id": "m1", "name": "Orca X1C", "content": {"type": "printer"}},
                {"id": "p1", "name": "Orca 0.20mm", "content": {"type": "print"}},
                {
                    "id": "f1",
                    "name": "Orca PLA",
                    "content": {
                        "type": "filament",
                        "filament_type": ["PLA"],
                        "default_filament_colour": ["#000000"],
                    },
                },
            ]
        )
        svc_mock.close = AsyncMock()
        user = MagicMock(id=1)
        user.has_permission = MagicMock(return_value=True)
        with (
            patch.object(sp, "_load_orca_credentials", AsyncMock(return_value=self._orca_creds("tok"))),
            patch.object(sp, "_build_orca_service", AsyncMock(return_value=svc_mock)),
        ):
            slots, status = await sp._fetch_orca_cloud_presets(MagicMock(), user)
        assert status == "ok"
        assert [p.name for p in slots["printer"]] == ["Orca X1C"]
        assert [p.name for p in slots["process"]] == ["Orca 0.20mm"]
        filament = slots["filament"]
        assert [p.name for p in filament] == ["Orca PLA"]
        # Inline metadata extracted from the content blob (Orca's sync_pull
        # returns full content, so unlike Bambu Cloud we don't need a second
        # per-preset fetch to enrich filament_type / filament_colour).
        assert filament[0].filament_type == "PLA"
        assert filament[0].filament_colour == "#000000"

    @pytest.mark.asyncio
    async def test_extracts_compatible_printers_from_content(self):
        """#2628: Orca's sync_pull already carries the profile's own
        compatible-printer list. Surfacing it lets the SliceModal reject a
        profile built for another printer instead of falling back to reading
        the model out of the NAME — which fails outright for names that carry
        no model ("Overture PLA Matte @0.2")."""
        sp._orca_cloud_cache.clear()
        compat = ["Bambu Lab X1 Carbon 0.2 nozzle", "Bambu Lab P1S 0.2 nozzle"]
        svc_mock = MagicMock()
        svc_mock.list_profiles = AsyncMock(
            return_value=[
                {
                    "id": "f1",
                    "name": "Overture PLA Matte @0.2",
                    "content": {
                        "type": "filament",
                        "filament_type": ["PLA"],
                        "compatible_printers": compat,
                    },
                },
                {
                    "id": "p1",
                    "name": "Orca 0.20mm",
                    "content": {"type": "print", "compatible_printers": "Bambu Lab P2S 0.4 nozzle"},
                },
                {"id": "m1", "name": "Orca X1C", "content": {"type": "printer"}},
            ]
        )
        svc_mock.close = AsyncMock()
        user = MagicMock(id=1)
        user.has_permission = MagicMock(return_value=True)
        with (
            patch.object(sp, "_load_orca_credentials", AsyncMock(return_value=self._orca_creds("tok"))),
            patch.object(sp, "_build_orca_service", AsyncMock(return_value=svc_mock)),
        ):
            slots, status = await sp._fetch_orca_cloud_presets(MagicMock(), user)

        assert status == "ok"
        assert slots["filament"][0].compatible_printers == compat
        # A single-printer profile may store a bare string — normalised to a list.
        assert slots["process"][0].compatible_printers == ["Bambu Lab P2S 0.4 nozzle"]
        # Printer presets have nothing to be compatible with.
        assert slots["printer"][0].compatible_printers is None

    @pytest.mark.asyncio
    async def test_printer_preset_carries_the_preset_it_was_saved_from(self):
        """#3250: a printer preset saved under a name of its own is in no
        process's compatible_printers; the SliceModal needs its parent."""
        sp._orca_cloud_cache.clear()
        svc_mock = MagicMock()
        svc_mock.list_profiles = AsyncMock(
            return_value=[
                {
                    "id": "m1",
                    "name": "Bambu Lab H2D 0.4 nozzle - Skew",
                    "content": {"type": "printer", "inherits": " Bambu Lab H2D 0.4 nozzle "},
                },
                {"id": "m2", "name": "Orca X1C", "content": {"type": "printer", "inherits": ""}},
                {"id": "m3", "name": "Odd", "content": {"type": "printer", "inherits": ["x"]}},
            ]
        )
        svc_mock.close = AsyncMock()
        user = MagicMock(id=1)
        user.has_permission = MagicMock(return_value=True)
        with (
            patch.object(sp, "_load_orca_credentials", AsyncMock(return_value=self._orca_creds("tok"))),
            patch.object(sp, "_build_orca_service", AsyncMock(return_value=svc_mock)),
        ):
            slots, _ = await sp._fetch_orca_cloud_presets(MagicMock(), user)

        assert [p.inherits for p in slots["printer"]] == ["Bambu Lab H2D 0.4 nozzle", None, None]

    @pytest.mark.asyncio
    async def test_missing_or_malformed_compatible_printers_stays_none(self):
        """No data must read as "unknown", never as "compatible with nothing" —
        the SliceModal falls back to the name matcher for those."""
        sp._orca_cloud_cache.clear()
        svc_mock = MagicMock()
        svc_mock.list_profiles = AsyncMock(
            return_value=[
                {"id": "f1", "name": "No list", "content": {"type": "filament"}},
                {"id": "f2", "name": "Empty list", "content": {"type": "filament", "compatible_printers": []}},
                {"id": "f3", "name": "Blanks", "content": {"type": "filament", "compatible_printers": ["", "  "]}},
                {"id": "f4", "name": "Wrong type", "content": {"type": "filament", "compatible_printers": {"a": 1}}},
            ]
        )
        svc_mock.close = AsyncMock()
        user = MagicMock(id=1)
        user.has_permission = MagicMock(return_value=True)
        with (
            patch.object(sp, "_load_orca_credentials", AsyncMock(return_value=self._orca_creds("tok"))),
            patch.object(sp, "_build_orca_service", AsyncMock(return_value=svc_mock)),
        ):
            slots, _status = await sp._fetch_orca_cloud_presets(MagicMock(), user)

        assert [p.compatible_printers for p in slots["filament"]] == [None, None, None, None]

    @pytest.mark.asyncio
    async def test_cache_hit_skips_orca_call(self):
        """A second call within TTL must reuse the cached slots and NOT
        hit the Orca service again — same TTL as Bambu Cloud (5 min)."""
        sp._orca_cloud_cache.clear()
        svc_mock = MagicMock()
        svc_mock.list_profiles = AsyncMock(return_value=[])
        svc_mock.close = AsyncMock()
        user = MagicMock(id=1)
        user.has_permission = MagicMock(return_value=True)
        with (
            patch.object(sp, "_load_orca_credentials", AsyncMock(return_value=self._orca_creds("tok"))),
            patch.object(sp, "_build_orca_service", AsyncMock(return_value=svc_mock)) as build,
        ):
            await sp._fetch_orca_cloud_presets(MagicMock(), user)
            await sp._fetch_orca_cloud_presets(MagicMock(), user)
        # Build is the cache miss signal — second call reused the cache.
        build.assert_awaited_once()


class TestFetchCloudPresets:
    """`_fetch_cloud_presets` translates token state and cloud errors into
    the four ``cloud_status`` values the SliceModal banner consumes."""

    @pytest.mark.asyncio
    async def test_no_token_returns_not_authenticated(self):
        sp._cloud_cache.clear()
        with patch.object(sp, "get_stored_token", AsyncMock(return_value=(None, None, None))):
            slots, status = await sp._fetch_cloud_presets(MagicMock(), _user_with_cloud_auth())
        assert status == "not_authenticated"
        assert slots == {"printer": [], "process": [], "filament": []}

    @pytest.mark.asyncio
    async def test_user_without_cloud_auth_returns_not_authenticated(self):
        """Defence-in-depth: a user lacking CLOUD_AUTH must NOT see cloud
        presets even if their User row carries a stale cloud_token from a
        previous permission state. Token lookup is skipped entirely."""
        sp._cloud_cache.clear()
        user = MagicMock(id=1)
        user.has_permission = MagicMock(return_value=False)
        with patch.object(sp, "get_stored_token", AsyncMock(return_value=("leftover-token", None, None))) as get_tok:
            slots, status = await sp._fetch_cloud_presets(MagicMock(), user)
        assert status == "not_authenticated"
        assert slots["printer"] == []
        # Token was never read — the perm check short-circuits ahead of it.
        get_tok.assert_not_called()

    @pytest.mark.asyncio
    async def test_auth_error_returns_expired(self):
        sp._cloud_cache.clear()
        cloud_mock = MagicMock()
        cloud_mock.set_token = MagicMock()
        cloud_mock.get_slicer_settings = AsyncMock(side_effect=sp.BambuCloudAuthError("expired"))
        cloud_mock.close = AsyncMock()
        with (
            patch.object(sp, "get_stored_token", AsyncMock(return_value=("tok", "e@x", None))),
            patch.object(sp, "BambuCloudService", return_value=cloud_mock),
        ):
            slots, status = await sp._fetch_cloud_presets(MagicMock(), _user_with_cloud_auth())
        assert status == "expired"
        assert slots["printer"] == []
        cloud_mock.close.assert_awaited_once()

    @pytest.mark.asyncio
    async def test_cloud_error_returns_unreachable(self):
        sp._cloud_cache.clear()
        cloud_mock = MagicMock()
        cloud_mock.set_token = MagicMock()
        cloud_mock.get_slicer_settings = AsyncMock(side_effect=sp.BambuCloudError("net down"))
        cloud_mock.close = AsyncMock()
        with (
            patch.object(sp, "get_stored_token", AsyncMock(return_value=("tok", None, None))),
            patch.object(sp, "BambuCloudService", return_value=cloud_mock),
        ):
            _slots, status = await sp._fetch_cloud_presets(MagicMock(), _user_with_cloud_auth())
        assert status == "unreachable"

    @pytest.mark.asyncio
    async def test_happy_path_shapes_private_then_public(self):
        """Cloud presets split into private (user-custom) + public (Bambu's
        stock cloud presets). Private should sort before public so a user's
        own customisations sit at the top of the dropdown."""
        sp._cloud_cache.clear()
        cloud_mock = MagicMock()
        cloud_mock.set_token = MagicMock()
        cloud_mock.get_slicer_settings = AsyncMock(
            return_value={
                "printer": {
                    "private": [{"setting_id": "PFUprivate1", "name": "My X1C"}],
                    "public": [{"setting_id": "PFUpublic1", "name": "Bambu X1C Stock"}],
                },
                "print": {"private": [], "public": []},
                "filament": {"private": [], "public": []},
            }
        )
        cloud_mock.close = AsyncMock()
        with (
            patch.object(sp, "get_stored_token", AsyncMock(return_value=("tok", None, None))),
            patch.object(sp, "BambuCloudService", return_value=cloud_mock),
        ):
            slots, status = await sp._fetch_cloud_presets(MagicMock(), _user_with_cloud_auth())
        assert status == "ok"
        names = [p.name for p in slots["printer"]]
        assert names == ["My X1C", "Bambu X1C Stock"]

    @pytest.mark.asyncio
    async def test_cache_hit_skips_cloud_call(self):
        """A second call within TTL must reuse the cached slots and NOT
        hit Bambu Cloud again."""
        sp._cloud_cache.clear()
        cloud_mock = MagicMock()
        cloud_mock.set_token = MagicMock()
        cloud_mock.get_slicer_settings = AsyncMock(
            return_value={
                "printer": {"private": [{"setting_id": "id1", "name": "X1C"}], "public": []},
                "print": {"private": [], "public": []},
                "filament": {"private": [], "public": []},
            }
        )
        cloud_mock.close = AsyncMock()
        user = _user_with_cloud_auth(user_id=42)
        with (
            patch.object(sp, "get_stored_token", AsyncMock(return_value=("tok", None, None))),
            patch.object(sp, "BambuCloudService", return_value=cloud_mock),
        ):
            await sp._fetch_cloud_presets(MagicMock(), user)
            await sp._fetch_cloud_presets(MagicMock(), user)
        cloud_mock.get_slicer_settings.assert_awaited_once()

    @pytest.mark.asyncio
    async def test_cache_is_per_user(self):
        """User A's cached cloud presets must not surface for user B."""
        sp._cloud_cache.clear()

        def make_mock(name: str):
            m = MagicMock()
            m.set_token = MagicMock()
            m.get_slicer_settings = AsyncMock(
                return_value={
                    "printer": {"private": [{"setting_id": f"id-{name}", "name": name}], "public": []},
                    "print": {"private": [], "public": []},
                    "filament": {"private": [], "public": []},
                }
            )
            m.close = AsyncMock()
            return m

        sequence = [make_mock("AliceX1C"), make_mock("BobX1C")]
        with (
            patch.object(sp, "get_stored_token", AsyncMock(return_value=("tok", None, None))),
            patch.object(sp, "BambuCloudService", side_effect=sequence),
        ):
            alice_slots, _ = await sp._fetch_cloud_presets(MagicMock(), _user_with_cloud_auth(1))
            bob_slots, _ = await sp._fetch_cloud_presets(MagicMock(), _user_with_cloud_auth(2))

        assert alice_slots["printer"][0].name == "AliceX1C"
        assert bob_slots["printer"][0].name == "BobX1C"

    @pytest.mark.asyncio
    async def test_cache_invalidates_on_token_change(self):
        """A token change (logout + login, admin reset, region switch) must
        bypass the cache for that user — pinning a real-world auth bug
        where user re-login + cache-stuck-on-old-cloud-account would
        silently serve a different account's preset list for ~5 minutes."""
        sp._cloud_cache.clear()

        def make_mock(name: str):
            m = MagicMock()
            m.set_token = MagicMock()
            m.get_slicer_settings = AsyncMock(
                return_value={
                    "printer": {"private": [{"setting_id": f"id-{name}", "name": name}], "public": []},
                    "print": {"private": [], "public": []},
                    "filament": {"private": [], "public": []},
                }
            )
            m.close = AsyncMock()
            return m

        # Same user_id, different token between calls — the second call must
        # NOT serve the first call's cached slots.
        services = [make_mock("OldAccountX1C"), make_mock("NewAccountX1C")]
        token_sequence = [("tok-old", None, None), ("tok-new", None, None)]
        user = _user_with_cloud_auth(user_id=7)

        with (
            patch.object(sp, "get_stored_token", AsyncMock(side_effect=token_sequence)),
            patch.object(sp, "BambuCloudService", side_effect=services),
        ):
            first, _ = await sp._fetch_cloud_presets(MagicMock(), user)
            second, _ = await sp._fetch_cloud_presets(MagicMock(), user)

        assert first["printer"][0].name == "OldAccountX1C"
        assert second["printer"][0].name == "NewAccountX1C"

    @pytest.mark.asyncio
    async def test_refresh_bypasses_cloud_cache(self):
        """``refresh=True`` must skip an otherwise-warm cache entry and hit
        Bambu Cloud again — wiring for the SliceModal's Refresh button so a
        user who deletes a cloud preset in Bambu Studio / Handy doesn't have
        to wait for the 5-minute TTL to expire (#1581)."""
        sp._cloud_cache.clear()
        cloud_mock = MagicMock()
        cloud_mock.set_token = MagicMock()
        cloud_mock.get_slicer_settings = AsyncMock(
            return_value={
                "printer": {"private": [{"setting_id": "id1", "name": "X1C"}], "public": []},
                "print": {"private": [], "public": []},
                "filament": {"private": [], "public": []},
            }
        )
        cloud_mock.close = AsyncMock()
        user = _user_with_cloud_auth(user_id=99)
        with (
            patch.object(sp, "get_stored_token", AsyncMock(return_value=("tok", None, None))),
            patch.object(sp, "BambuCloudService", return_value=cloud_mock),
        ):
            await sp._fetch_cloud_presets(MagicMock(), user)
            # Without refresh, the second call hits cache (covered by
            # test_cache_hit_skips_cloud_call). With refresh=True it MUST
            # re-fetch.
            await sp._fetch_cloud_presets(MagicMock(), user, refresh=True)
        assert cloud_mock.get_slicer_settings.await_count == 2

    @pytest.mark.asyncio
    async def test_refresh_writes_back_to_cache(self):
        """A refresh call must still update the cache so a subsequent normal
        call doesn't re-hit the cloud immediately afterwards."""
        sp._cloud_cache.clear()
        cloud_mock = MagicMock()
        cloud_mock.set_token = MagicMock()
        cloud_mock.get_slicer_settings = AsyncMock(
            return_value={
                "printer": {"private": [{"setting_id": "id1", "name": "X1C"}], "public": []},
                "print": {"private": [], "public": []},
                "filament": {"private": [], "public": []},
            }
        )
        cloud_mock.close = AsyncMock()
        user = _user_with_cloud_auth(user_id=101)
        with (
            patch.object(sp, "get_stored_token", AsyncMock(return_value=("tok", None, None))),
            patch.object(sp, "BambuCloudService", return_value=cloud_mock),
        ):
            await sp._fetch_cloud_presets(MagicMock(), user, refresh=True)
            await sp._fetch_cloud_presets(MagicMock(), user)
        # Two calls — first refresh, second a normal cache hit.
        assert cloud_mock.get_slicer_settings.await_count == 1


class TestFetchBundledPresets:
    """Standard tier reaches out to the slicer-api sidecar; tolerate the
    sidecar being absent / unreachable so the modal still works."""

    @pytest.mark.asyncio
    async def test_no_sidecar_url_returns_empty(self):
        sp._bundled_cache = None
        with patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value=None)):
            slots = await sp._fetch_bundled_presets(MagicMock())
        assert slots == {"printer": [], "process": [], "filament": []}
        # No URL means no useful cache result either — second call should
        # try again (so users who configure a URL mid-session see results).
        assert sp._bundled_cache is None

    @pytest.mark.asyncio
    async def test_sidecar_error_returns_empty(self):
        sp._bundled_cache = None
        svc_mock = MagicMock()
        svc_mock.list_bundled_profiles = AsyncMock(side_effect=sp.SlicerApiError("boom"))
        svc_mock.__aenter__ = AsyncMock(return_value=svc_mock)
        svc_mock.__aexit__ = AsyncMock(return_value=False)
        with (
            patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value="http://nope")),
            patch.object(sp, "SlicerApiService", return_value=svc_mock),
        ):
            slots = await sp._fetch_bundled_presets(MagicMock())
        assert slots == {"printer": [], "process": [], "filament": []}

    @pytest.mark.asyncio
    async def test_happy_path_shapes_response(self):
        sp._bundled_cache = None
        svc_mock = MagicMock()
        svc_mock.list_bundled_profiles = AsyncMock(
            return_value={
                "printer": [{"name": "Bambu X1C 0.4", "base_id": None}],
                "process": [{"name": "0.20mm Standard", "base_id": "fdm_process_common"}],
                "filament": [{"name": "Bambu PLA Basic", "base_id": "fdm_filament_pla"}],
            }
        )
        svc_mock.__aenter__ = AsyncMock(return_value=svc_mock)
        svc_mock.__aexit__ = AsyncMock(return_value=False)
        with (
            patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value="http://ok")),
            patch.object(sp, "SlicerApiService", return_value=svc_mock),
        ):
            slots = await sp._fetch_bundled_presets(MagicMock())
        assert slots["printer"][0].name == "Bambu X1C 0.4"
        assert slots["printer"][0].source == "standard"
        # Bundled presets are addressed by name (the slicer's inheritance
        # walker resolves them by name), so id == name.
        assert slots["printer"][0].id == "Bambu X1C 0.4"

    @pytest.mark.asyncio
    async def test_cache_hit_skips_sidecar(self):
        """A second call within TTL must serve from the cached entry and not
        re-hit the sidecar HTTP."""
        sp._bundled_cache = (
            time.monotonic(),
            {
                "printer": [UnifiedPreset(id="Cached", name="Cached", source="standard")],
                "process": [],
                "filament": [],
            },
        )
        # If `SlicerApiService` is constructed at all we've missed the cache.
        with patch.object(sp, "SlicerApiService", side_effect=AssertionError("cache miss!")):
            slots = await sp._fetch_bundled_presets(MagicMock())
        assert slots["printer"][0].name == "Cached"

    @pytest.mark.asyncio
    async def test_refresh_bypasses_bundled_cache(self):
        """``refresh=True`` must re-hit the sidecar even when the in-process
        cache is warm — paired with the cloud-cache refresh, this is what
        powers the SliceModal's Refresh button (#1581)."""
        sp._bundled_cache = (
            time.monotonic(),
            {
                "printer": [UnifiedPreset(id="Stale", name="Stale", source="standard")],
                "process": [],
                "filament": [],
            },
        )
        svc_mock = MagicMock()
        svc_mock.list_bundled_profiles = AsyncMock(
            return_value={
                "printer": [{"name": "Fresh", "base_id": None}],
                "process": [],
                "filament": [],
            }
        )
        svc_mock.__aenter__ = AsyncMock(return_value=svc_mock)
        svc_mock.__aexit__ = AsyncMock(return_value=False)
        with (
            patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value="http://ok")),
            patch.object(sp, "SlicerApiService", return_value=svc_mock),
        ):
            slots = await sp._fetch_bundled_presets(MagicMock(), refresh=True)
        svc_mock.list_bundled_profiles.assert_awaited_once()
        assert [p.name for p in slots["printer"]] == ["Fresh"]
        # The fresh result must also be written back to the cache so a
        # subsequent normal (non-refresh) call doesn't re-hit the sidecar.
        assert sp._bundled_cache is not None
        assert [p.name for p in sp._bundled_cache[1]["printer"]] == ["Fresh"]


class TestResolveSlicerApiUrl:
    """`_resolve_slicer_api_url` must respect the user's `preferred_slicer`
    setting just like the slice route does. The bundled-listing fetch
    used to be hardcoded to OrcaSlicer's URL, which left the Standard
    tier permanently empty for BambuStudio installs."""

    @pytest.mark.asyncio
    async def test_bambu_studio_preference_uses_bambu_url(self):
        """When the user prefers Bambu Studio, the listing fetch must hit
        the bambu-studio-api sidecar (port 3001 by default), not orca's
        port 3003."""

        async def fake_get_setting(_db, key):
            return {
                "preferred_slicer": "bambu_studio",
                "bambu_studio_api_url": "http://bambu-studio-api:3000",
            }.get(key)

        with patch(
            "backend.app.api.routes.settings.get_setting",
            new=fake_get_setting,
        ):
            url = await sp._resolve_slicer_api_url(MagicMock())
        assert url == "http://bambu-studio-api:3000"

    @pytest.mark.asyncio
    async def test_orcaslicer_preference_uses_orca_url(self):
        async def fake_get_setting(_db, key):
            return {
                "preferred_slicer": "orcaslicer",
                "orcaslicer_api_url": "http://orca-slicer-api:3000",
            }.get(key)

        with patch(
            "backend.app.api.routes.settings.get_setting",
            new=fake_get_setting,
        ):
            url = await sp._resolve_slicer_api_url(MagicMock())
        assert url == "http://orca-slicer-api:3000"

    @pytest.mark.asyncio
    async def test_default_preference_is_bambu_studio(self):
        """Empty preferred_slicer → bambu_studio (matches the slice route's
        default at library.py:_run_slicer_with_fallback)."""

        async def fake_get_setting(_db, key):
            return {
                # preferred_slicer not set
                "bambu_studio_api_url": "http://bambu-default:3000",
            }.get(key)

        with patch(
            "backend.app.api.routes.settings.get_setting",
            new=fake_get_setting,
        ):
            url = await sp._resolve_slicer_api_url(MagicMock())
        assert url == "http://bambu-default:3000"

    @pytest.mark.asyncio
    async def test_unknown_preference_returns_none(self):
        """An unrecognised preferred_slicer value (e.g. set out-of-band by
        a stale migration) returns None so the modal degrades to "no
        Standard tier" rather than crashing — the slice route raises 400
        in this case but the listing is informational, so be lenient."""

        async def fake_get_setting(_db, key):
            return {"preferred_slicer": "prusaslicer"}.get(key)

        with patch(
            "backend.app.api.routes.settings.get_setting",
            new=fake_get_setting,
        ):
            url = await sp._resolve_slicer_api_url(MagicMock())
        assert url is None


class TestParseCompatiblePrinters:
    """``compatible_printers`` exposed for local process / filament presets so
    the SliceModal can filter the dropdowns by the selected printer (#1325)."""

    def test_parses_json_array(self):
        raw = '["Bambu Lab X1 Carbon 0.4 nozzle", "Bambu Lab X1 0.4 nozzle"]'
        assert sp._parse_compatible_printers(raw) == [
            "Bambu Lab X1 Carbon 0.4 nozzle",
            "Bambu Lab X1 0.4 nozzle",
        ]

    def test_none_and_empty_return_none(self):
        assert sp._parse_compatible_printers(None) is None
        assert sp._parse_compatible_printers("") is None
        assert sp._parse_compatible_printers("[]") is None

    def test_malformed_json_returns_none(self):
        assert sp._parse_compatible_printers("not json") is None
        # A JSON value that isn't an array is treated as absent, not an error.
        assert sp._parse_compatible_printers('"a string"') is None

    def test_drops_non_string_and_blank_entries(self):
        assert sp._parse_compatible_printers('["X1C", 5, "", "  ", "A1"]') == [
            "X1C",
            "A1",
        ]


class TestListPrinterModels:
    """``GET /slicer/printer-models`` exposes ``PRINTER_MODEL_MAP`` so the
    frontend doesn't duplicate the Bambu model registry (#1325 follow-up)."""

    def test_returns_canonical_printer_model_map(self):
        from backend.app.utils.printer_models import PRINTER_MODEL_MAP

        result = sp.list_printer_models()
        # Same shape - mapping from "Bambu Lab <model>" to short code.
        assert result == PRINTER_MODEL_MAP
        # Spot-check a few entries: the SliceModal name-fallback (#1325)
        # specifically depends on these resolving.
        assert result["Bambu Lab X1 Carbon"] == "X1C"
        assert result["Bambu Lab P2S"] == "P2S"
        assert result["Bambu Lab A1 mini"] == "A1 Mini"
        assert result["Bambu Lab H2D Pro"] == "H2D Pro"

    def test_returns_a_copy_not_the_module_dict(self):
        # A response handler must never hand out the live module-level dict —
        # accidental mutation by middleware / serialisers would silently
        # corrupt the registry for every subsequent request.
        from backend.app.utils.printer_models import PRINTER_MODEL_MAP

        result = sp.list_printer_models()
        assert result is not PRINTER_MODEL_MAP


class TestNoUserWithAuthOn:
    """The sign-in stored without a user is the auth-off install's, and may be
    left over after auth was turned on. A caller with no user while auth is
    on (an API key without Allow Cloud Access) must not list its presets."""

    @pytest.mark.asyncio
    async def test_bambu_cloud(self):
        sp._cloud_cache.clear()
        with (
            patch.object(sp, "is_auth_enabled", AsyncMock(return_value=True)),
            patch.object(sp, "get_stored_token", AsyncMock(return_value=("global-tok", None, None))) as get_tok,
        ):
            slots, status = await sp._fetch_cloud_presets(MagicMock(), None)
        assert status == "not_authenticated"
        assert slots == {"printer": [], "process": [], "filament": []}
        get_tok.assert_not_called()

    @pytest.mark.asyncio
    async def test_orca_cloud(self):
        sp._orca_cloud_cache.clear()
        with (
            patch.object(sp, "is_auth_enabled", AsyncMock(return_value=True)),
            patch.object(sp, "_load_orca_credentials", AsyncMock()) as load,
        ):
            slots, status = await sp._fetch_orca_cloud_presets(MagicMock(), None)
        assert status == "not_authenticated"
        assert slots == {"printer": [], "process": [], "filament": []}
        load.assert_not_called()


class TestLocalPrinterInherits:
    """#3250: local printer presets report the preset they were saved from."""

    @pytest.mark.asyncio
    async def test_local_printer_preset_carries_inherits(self, db_session):
        from backend.app.models.local_preset import LocalPreset

        db_session.add_all(
            [
                LocalPreset(
                    name="Bambu Lab H2D 0.4 nozzle - Skew",
                    preset_type="printer",
                    source="orcaslicer",
                    setting="{}",
                    inherits="Bambu Lab H2D 0.4 nozzle",
                ),
                LocalPreset(name="Plain", preset_type="printer", source="orcaslicer", setting="{}", inherits="  "),
                LocalPreset(
                    name="0.20mm Mine",
                    preset_type="process",
                    source="orcaslicer",
                    setting="{}",
                    inherits="0.20mm Standard @BBL H2D",
                ),
            ]
        )
        await db_session.commit()

        slots = await sp._fetch_local_presets(db_session)

        by_name = {p.name: p for p in slots["printer"]}
        assert by_name["Bambu Lab H2D 0.4 nozzle - Skew"].inherits == "Bambu Lab H2D 0.4 nozzle"
        assert by_name["Plain"].inherits is None
        # Only printer presets need it; a process is matched by its own list.
        assert slots["process"][0].inherits is None