|
|
@@ -0,0 +1,596 @@
|
|
|
+"""Profile defaults per product (#1326).
|
|
|
+
|
|
|
+A default holds the slicer preset, the per-model preset overrides and the K
|
|
|
+profiles for one Brand/Material/Subtype, for all colours or for one, and every
|
|
|
+way a spool is added copies it onto the new spool. Both inventory modes; the
|
|
|
+default itself is shared.
|
|
|
+"""
|
|
|
+
|
|
|
+from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
+
|
|
|
+import pytest
|
|
|
+from httpx import AsyncClient
|
|
|
+from sqlalchemy import select
|
|
|
+
|
|
|
+DEFAULTS = "/api/v1/inventory/profile-defaults"
|
|
|
+INTERNAL = "/api/v1/inventory/spools"
|
|
|
+SPOOLMAN = "/api/v1/spoolman/inventory"
|
|
|
+
|
|
|
+
|
|
|
+def _k(printer_id, *, extruder=0, diameter="0.4", cali_idx=5, k_value=0.02, name="PLA HF"):
|
|
|
+ return {
|
|
|
+ "printer_id": printer_id,
|
|
|
+ "extruder": extruder,
|
|
|
+ "nozzle_diameter": diameter,
|
|
|
+ "nozzle_type": None,
|
|
|
+ "k_value": k_value,
|
|
|
+ "name": name,
|
|
|
+ "cali_idx": cali_idx,
|
|
|
+ "setting_id": None,
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def _preset(model="H2D", diameter="0.4", code="PFUS01", name="My PLA @BBL H2D"):
|
|
|
+ return {"printer_model": model, "nozzle_diameter": diameter, "slicer_filament": code, "slicer_filament_name": name}
|
|
|
+
|
|
|
+
|
|
|
+def _default(color_name="", **overrides):
|
|
|
+ body = {
|
|
|
+ "brand": "Bambu Lab",
|
|
|
+ "material": "PLA",
|
|
|
+ "subtype": "Basic",
|
|
|
+ "color_name": color_name,
|
|
|
+ "slicer_filament": "PFUS00",
|
|
|
+ "slicer_filament_name": "My PLA",
|
|
|
+ "presets": [],
|
|
|
+ "k_profiles": [],
|
|
|
+ }
|
|
|
+ body.update(overrides)
|
|
|
+ return body
|
|
|
+
|
|
|
+
|
|
|
+def _spool_body(**overrides):
|
|
|
+ body = {"material": "PLA", "subtype": "Basic", "brand": "Bambu Lab", "color_name": "Jade White"}
|
|
|
+ body.update(overrides)
|
|
|
+ return body
|
|
|
+
|
|
|
+
|
|
|
+@pytest.fixture
|
|
|
+async def printers(printer_factory):
|
|
|
+ return [await printer_factory(model="H2D"), await printer_factory(model="X1C")]
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.integration
|
|
|
+class TestDefaultsCrud:
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_put_creates_then_replaces(self, async_client: AsyncClient, printers):
|
|
|
+ a, b = printers
|
|
|
+ response = await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id)], presets=[_preset()]))
|
|
|
+ assert response.status_code == 200, response.text
|
|
|
+ first = response.json()
|
|
|
+
|
|
|
+ response = await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(b.id)]))
|
|
|
+ assert response.status_code == 200, response.text
|
|
|
+ second = response.json()
|
|
|
+
|
|
|
+ assert second["id"] == first["id"]
|
|
|
+ assert [k["printer_id"] for k in second["k_profiles"]] == [b.id]
|
|
|
+ assert second["presets"] == []
|
|
|
+ listed = (await async_client.get(DEFAULTS)).json()
|
|
|
+ assert len(listed) == 1
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_product_matches_case_insensitively(self, async_client: AsyncClient):
|
|
|
+ await async_client.put(DEFAULTS, json=_default())
|
|
|
+ await async_client.put(DEFAULTS, json=_default(brand="bambu lab", material="pla", subtype="basic"))
|
|
|
+ assert len((await async_client.get(DEFAULTS)).json()) == 1
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_duplicate_hotend_is_refused_before_anything_changes(self, async_client: AsyncClient, printers):
|
|
|
+ a, _ = printers
|
|
|
+ await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id)]))
|
|
|
+ response = await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id), _k(a.id, cali_idx=7)]))
|
|
|
+ assert response.status_code == 422
|
|
|
+ kept = (await async_client.get(DEFAULTS)).json()[0]
|
|
|
+ assert [k["cali_idx"] for k in kept["k_profiles"]] == [5]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_unknown_printer_is_refused(self, async_client: AsyncClient):
|
|
|
+ response = await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(999)]))
|
|
|
+ assert response.status_code == 404
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_delete(self, async_client: AsyncClient):
|
|
|
+ row = (await async_client.put(DEFAULTS, json=_default())).json()
|
|
|
+ assert (await async_client.delete(f"{DEFAULTS}/{row['id']}")).status_code == 200
|
|
|
+ assert (await async_client.get(DEFAULTS)).json() == []
|
|
|
+ assert (await async_client.delete(f"{DEFAULTS}/{row['id']}")).status_code == 404
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_deleting_a_printer_takes_its_k_profiles_with_it(self, async_client: AsyncClient, printers):
|
|
|
+ a, b = printers
|
|
|
+ await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id), _k(b.id)]))
|
|
|
+ assert (await async_client.delete(f"/api/v1/printers/{a.id}")).status_code == 200
|
|
|
+ kept = (await async_client.get(DEFAULTS)).json()[0]
|
|
|
+ assert [k["printer_id"] for k in kept["k_profiles"]] == [b.id]
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.integration
|
|
|
+class TestResolve:
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_no_default(self, async_client: AsyncClient):
|
|
|
+ response = await async_client.get(f"{DEFAULTS}/resolve", params={"material": "PLA"})
|
|
|
+ assert response.status_code == 200
|
|
|
+ assert response.json() is None
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_all_colours_default_covers_every_colour(self, async_client: AsyncClient, printers):
|
|
|
+ a, _ = printers
|
|
|
+ await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id)]))
|
|
|
+ resolved = (await async_client.get(f"{DEFAULTS}/resolve", params=_spool_body(color_name="Never Seen"))).json()
|
|
|
+ assert resolved["slicer_filament"] == "PFUS00"
|
|
|
+ assert [k["cali_idx"] for k in resolved["k_profiles"]] == [5]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_colour_wins_per_hotend_and_keeps_the_rest(self, async_client: AsyncClient, printers):
|
|
|
+ a, b = printers
|
|
|
+ await async_client.put(
|
|
|
+ DEFAULTS,
|
|
|
+ json=_default(
|
|
|
+ k_profiles=[_k(a.id, extruder=0), _k(a.id, extruder=1), _k(b.id)],
|
|
|
+ presets=[_preset("H2D"), _preset("X1C", code="PFUS02")],
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ await async_client.put(
|
|
|
+ DEFAULTS,
|
|
|
+ json=_default(
|
|
|
+ "Jade White",
|
|
|
+ slicer_filament=None,
|
|
|
+ k_profiles=[_k(a.id, extruder=1, cali_idx=9, k_value=0.031)],
|
|
|
+ presets=[_preset("X1C", code="PFUS09")],
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ resolved = (await async_client.get(f"{DEFAULTS}/resolve", params=_spool_body())).json()
|
|
|
+ by_hotend = {(k["printer_id"], k["extruder"]): k["cali_idx"] for k in resolved["k_profiles"]}
|
|
|
+ assert by_hotend == {(a.id, 0): 5, (a.id, 1): 9, (b.id, 0): 5}
|
|
|
+ by_model = {p["printer_model"]: p["slicer_filament"] for p in resolved["presets"]}
|
|
|
+ assert by_model == {"H2D": "PFUS01", "X1C": "PFUS09"}
|
|
|
+ # The colour row sets no preset of its own, so the product's stays.
|
|
|
+ assert resolved["slicer_filament"] == "PFUS00"
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_other_products_are_not_matched(self, async_client: AsyncClient):
|
|
|
+ await async_client.put(DEFAULTS, json=_default())
|
|
|
+ for params in (_spool_body(subtype="Matte"), _spool_body(brand="Polymaker"), _spool_body(material="PETG")):
|
|
|
+ assert (await async_client.get(f"{DEFAULTS}/resolve", params=params)).json() is None
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.integration
|
|
|
+class TestInternalCreate:
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_new_spool_gets_the_default(self, async_client: AsyncClient, printers):
|
|
|
+ a, _ = printers
|
|
|
+ await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id)], presets=[_preset()]))
|
|
|
+
|
|
|
+ spool = (await async_client.post(INTERNAL, json=_spool_body())).json()
|
|
|
+ assert spool["slicer_filament"] == "PFUS00"
|
|
|
+ assert spool["slicer_filament_name"] == "My PLA"
|
|
|
+ assert [(k["printer_id"], k["cali_idx"]) for k in spool["k_profiles"]] == [(a.id, 5)]
|
|
|
+ presets = (await async_client.get(f"{INTERNAL}/{spool['id']}/filament-presets")).json()
|
|
|
+ assert [p["slicer_filament"] for p in presets] == ["PFUS01"]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_preset_in_the_request_is_kept(self, async_client: AsyncClient):
|
|
|
+ await async_client.put(DEFAULTS, json=_default())
|
|
|
+ spool = (
|
|
|
+ await async_client.post(
|
|
|
+ INTERNAL, json=_spool_body(slicer_filament="GFA00", slicer_filament_name="Bambu PLA Basic")
|
|
|
+ )
|
|
|
+ ).json()
|
|
|
+ assert spool["slicer_filament"] == "GFA00"
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_bulk_create_gives_every_copy_the_default(self, async_client: AsyncClient, printers):
|
|
|
+ a, _ = printers
|
|
|
+ await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id)]))
|
|
|
+ response = await async_client.post(f"{INTERNAL}/bulk", json={"spool": _spool_body(), "quantity": 3})
|
|
|
+ assert response.status_code == 200, response.text
|
|
|
+ spools = response.json()
|
|
|
+ assert len(spools) == 3
|
|
|
+ assert all([k["cali_idx"] for k in s["k_profiles"]] == [5] for s in spools)
|
|
|
+ assert all(s["slicer_filament"] == "PFUS00" for s in spools)
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_without_a_default_nothing_is_added(self, async_client: AsyncClient):
|
|
|
+ spool = (await async_client.post(INTERNAL, json=_spool_body())).json()
|
|
|
+ assert spool["slicer_filament"] is None
|
|
|
+ assert spool["k_profiles"] == []
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_rfid_auto_create_keeps_the_tag_preset_and_assigns_the_default_k(self, db_session, printers):
|
|
|
+ from backend.app.models.filament_profile_default import (
|
|
|
+ FilamentProfileDefault,
|
|
|
+ FilamentProfileDefaultKProfile,
|
|
|
+ )
|
|
|
+ from backend.app.services.spool_tag_matcher import create_spool_from_tray
|
|
|
+
|
|
|
+ a, _ = printers
|
|
|
+ row = FilamentProfileDefault(
|
|
|
+ match_key="bambu lab\x1fpla\x1fbasic\x1f",
|
|
|
+ brand="Bambu Lab",
|
|
|
+ material="PLA",
|
|
|
+ subtype="Basic",
|
|
|
+ slicer_filament="PFUS00",
|
|
|
+ )
|
|
|
+ db_session.add(row)
|
|
|
+ await db_session.flush()
|
|
|
+ db_session.add(FilamentProfileDefaultKProfile(profile_default_id=row.id, **_k(a.id)))
|
|
|
+ await db_session.commit()
|
|
|
+
|
|
|
+ spool = await create_spool_from_tray(
|
|
|
+ db_session,
|
|
|
+ {
|
|
|
+ "tray_type": "PLA",
|
|
|
+ "tray_sub_brands": "PLA Basic",
|
|
|
+ "tray_color": "FFFFFFFF",
|
|
|
+ "tray_info_idx": "GFA00",
|
|
|
+ "tray_weight": "1000",
|
|
|
+ "remain": 100,
|
|
|
+ "tag_uid": "A1B2C3D4E5F60708",
|
|
|
+ "tray_uuid": "0123456789ABCDEF0123456789ABCDEF",
|
|
|
+ },
|
|
|
+ )
|
|
|
+ assert spool.slicer_filament == "GFA00"
|
|
|
+ # In the in-memory list the RFID assign right after reads.
|
|
|
+ assert [k.cali_idx for k in spool.k_profiles] == [5]
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.integration
|
|
|
+class TestInternalApply:
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_fills_only_what_is_missing(self, async_client: AsyncClient, printers):
|
|
|
+ a, b = printers
|
|
|
+ # Spools first, so the create path has no default to apply.
|
|
|
+ bare = (await async_client.post(INTERNAL, json=_spool_body())).json()
|
|
|
+ own = (await async_client.post(INTERNAL, json=_spool_body(slicer_filament="GFA00"))).json()
|
|
|
+ await async_client.put(f"{INTERNAL}/{own['id']}/k-profiles", json=[_k(a.id, cali_idx=11, k_value=0.05)])
|
|
|
+ other = (await async_client.post(INTERNAL, json=_spool_body(subtype="Matte"))).json()
|
|
|
+ archived = (await async_client.post(INTERNAL, json=_spool_body())).json()
|
|
|
+ await async_client.post(f"{INTERNAL}/{archived['id']}/archive")
|
|
|
+
|
|
|
+ row = (await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id), _k(b.id)]))).json()
|
|
|
+ response = await async_client.post(f"{DEFAULTS}/{row['id']}/apply")
|
|
|
+ assert response.status_code == 200, response.text
|
|
|
+ assert response.json() == {"updated": 2}
|
|
|
+
|
|
|
+ def k_of(spool):
|
|
|
+ return {k["printer_id"]: k["cali_idx"] for k in spool["k_profiles"]}
|
|
|
+
|
|
|
+ bare = (await async_client.get(f"{INTERNAL}/{bare['id']}")).json()
|
|
|
+ assert bare["slicer_filament"] == "PFUS00"
|
|
|
+ assert k_of(bare) == {a.id: 5, b.id: 5}
|
|
|
+ own = (await async_client.get(f"{INTERNAL}/{own['id']}")).json()
|
|
|
+ assert own["slicer_filament"] == "GFA00"
|
|
|
+ assert k_of(own) == {a.id: 11, b.id: 5}
|
|
|
+ assert (await async_client.get(f"{INTERNAL}/{other['id']}")).json()["k_profiles"] == []
|
|
|
+ assert (await async_client.get(f"{INTERNAL}/{archived['id']}")).json()["k_profiles"] == []
|
|
|
+
|
|
|
+ # Nothing left to fill.
|
|
|
+ assert (await async_client.post(f"{DEFAULTS}/{row['id']}/apply")).json() == {"updated": 0}
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_all_colours_apply_gives_a_colour_its_own_values(self, async_client: AsyncClient, printers):
|
|
|
+ a, _ = printers
|
|
|
+ white = (await async_client.post(INTERNAL, json=_spool_body())).json()
|
|
|
+ black = (await async_client.post(INTERNAL, json=_spool_body(color_name="Black"))).json()
|
|
|
+ row = (await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id)]))).json()
|
|
|
+ await async_client.put(DEFAULTS, json=_default("Jade White", k_profiles=[_k(a.id, cali_idx=9)]))
|
|
|
+
|
|
|
+ assert (await async_client.post(f"{DEFAULTS}/{row['id']}/apply")).json() == {"updated": 2}
|
|
|
+ white = (await async_client.get(f"{INTERNAL}/{white['id']}")).json()
|
|
|
+ black = (await async_client.get(f"{INTERNAL}/{black['id']}")).json()
|
|
|
+ assert [k["cali_idx"] for k in white["k_profiles"]] == [9]
|
|
|
+ assert [k["cali_idx"] for k in black["k_profiles"]] == [5]
|
|
|
+
|
|
|
+
|
|
|
+# ── Spoolman mode ────────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+
|
|
|
+def _spoolman_spool(spool_id, *, extra=None, name="PLA Basic", archived=False):
|
|
|
+ return {
|
|
|
+ "id": spool_id,
|
|
|
+ "filament": {
|
|
|
+ "id": 7,
|
|
|
+ "name": name,
|
|
|
+ "material": "PLA",
|
|
|
+ "color_hex": "FFFFFF",
|
|
|
+ "weight": 1000,
|
|
|
+ "vendor": {"id": 3, "name": "Bambu Lab"},
|
|
|
+ },
|
|
|
+ "remaining_weight": 1000.0,
|
|
|
+ "used_weight": 0.0,
|
|
|
+ "location": None,
|
|
|
+ "comment": None,
|
|
|
+ "first_used": None,
|
|
|
+ "last_used": None,
|
|
|
+ "registered": "2024-01-01T00:00:00+00:00",
|
|
|
+ "archived": archived,
|
|
|
+ "price": None,
|
|
|
+ "extra": extra or {},
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+@pytest.fixture
|
|
|
+def spoolman_client():
|
|
|
+ client = MagicMock()
|
|
|
+ client.base_url = "http://localhost:7912"
|
|
|
+ client.health_check = AsyncMock(return_value=True)
|
|
|
+ ids = iter(range(100, 200))
|
|
|
+ client.create_spool = AsyncMock(side_effect=lambda **kwargs: _spoolman_spool(next(ids)))
|
|
|
+ client.find_or_create_filament = AsyncMock(return_value=7)
|
|
|
+ client.find_or_create_vendor = AsyncMock(return_value=3)
|
|
|
+ client.ensure_extra_field = AsyncMock(return_value=True)
|
|
|
+ client.merge_spool_extra = AsyncMock(side_effect=lambda spool_id, extra: _spoolman_spool(spool_id, extra=extra))
|
|
|
+ with patch("backend.app.api.routes.spoolman_inventory._get_client", AsyncMock(return_value=client)):
|
|
|
+ yield client
|
|
|
+
|
|
|
+
|
|
|
+async def _spoolman_rows(db_session, spool_id):
|
|
|
+ from backend.app.models.spool_filament_preset import SpoolmanFilamentPreset
|
|
|
+ from backend.app.models.spoolman_k_profile import SpoolmanKProfile
|
|
|
+
|
|
|
+ k = (
|
|
|
+ (await db_session.execute(select(SpoolmanKProfile).where(SpoolmanKProfile.spoolman_spool_id == spool_id)))
|
|
|
+ .scalars()
|
|
|
+ .all()
|
|
|
+ )
|
|
|
+ presets = (
|
|
|
+ (
|
|
|
+ await db_session.execute(
|
|
|
+ select(SpoolmanFilamentPreset).where(SpoolmanFilamentPreset.spoolman_spool_id == spool_id)
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .scalars()
|
|
|
+ .all()
|
|
|
+ )
|
|
|
+ return [r.cali_idx for r in k], [p.slicer_filament for p in presets]
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.integration
|
|
|
+class TestSpoolman:
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_create_gets_the_default(self, async_client: AsyncClient, db_session, printers, spoolman_client):
|
|
|
+ a, _ = printers
|
|
|
+ await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id)], presets=[_preset()]))
|
|
|
+
|
|
|
+ response = await async_client.post(f"{SPOOLMAN}/spools", json=_spool_body())
|
|
|
+ assert response.status_code == 200, response.text
|
|
|
+ assert response.json()["slicer_filament"] == "PFUS00"
|
|
|
+ extra = spoolman_client.merge_spool_extra.call_args.args[1]
|
|
|
+ assert extra["bambu_slicer_filament"] == '"PFUS00"'
|
|
|
+ assert extra["bambu_color_name"] == '"Jade White"'
|
|
|
+ assert await _spoolman_rows(db_session, 100) == ([5], ["PFUS01"])
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_bulk_create_writes_preset_colour_and_default_on_every_copy(
|
|
|
+ self, async_client: AsyncClient, db_session, printers, spoolman_client
|
|
|
+ ):
|
|
|
+ a, _ = printers
|
|
|
+ await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id)]))
|
|
|
+ response = await async_client.post(f"{SPOOLMAN}/spools/bulk", json={"spool": _spool_body(), "quantity": 2})
|
|
|
+ assert response.status_code == 200, response.text
|
|
|
+ # Bulk wrote neither the preset nor the colour name before.
|
|
|
+ written = {c.args[0]: c.args[1] for c in spoolman_client.merge_spool_extra.call_args_list}
|
|
|
+ assert set(written) == {100, 101}
|
|
|
+ assert all(e["bambu_color_name"] == '"Jade White"' for e in written.values())
|
|
|
+ assert all(e["bambu_slicer_filament"] == '"PFUS00"' for e in written.values())
|
|
|
+ for spool_id in (100, 101):
|
|
|
+ assert (await _spoolman_rows(db_session, spool_id))[0] == [5]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_apply_fills_spoolman_spools(self, async_client: AsyncClient, db_session, printers, spoolman_client):
|
|
|
+ a, _ = printers
|
|
|
+ spoolman_client.get_all_spools = AsyncMock(
|
|
|
+ return_value=[
|
|
|
+ _spoolman_spool(1, extra={"bambu_color_name": '"Jade White"'}),
|
|
|
+ _spoolman_spool(2, extra={"bambu_slicer_filament": '"GFA00"'}),
|
|
|
+ _spoolman_spool(3, name="PLA Matte"),
|
|
|
+ _spoolman_spool(4, archived=True),
|
|
|
+ ]
|
|
|
+ )
|
|
|
+ from backend.app.models.spoolman_k_profile import SpoolmanKProfile
|
|
|
+
|
|
|
+ # Spool 2 has its own K profile for the hotend the default covers.
|
|
|
+ db_session.add(SpoolmanKProfile(spoolman_spool_id=2, **_k(a.id, cali_idx=11)))
|
|
|
+ await db_session.commit()
|
|
|
+ row = (await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id)]))).json()
|
|
|
+
|
|
|
+ response = await async_client.post(f"{SPOOLMAN}/profile-defaults/{row['id']}/apply")
|
|
|
+ assert response.status_code == 200, response.text
|
|
|
+ # Spool 2 has both its preset and its K profile: nothing to fill.
|
|
|
+ assert response.json() == {"updated": 1}
|
|
|
+ assert (await _spoolman_rows(db_session, 1))[0] == [5]
|
|
|
+ assert (await _spoolman_rows(db_session, 2))[0] == [11]
|
|
|
+ assert (await _spoolman_rows(db_session, 3))[0] == []
|
|
|
+ assert (await _spoolman_rows(db_session, 4))[0] == []
|
|
|
+ # Only the spool without a preset of its own gets the default's.
|
|
|
+ assert [c.args[0] for c in spoolman_client.merge_spool_extra.call_args_list] == [1]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_ams_auto_add_gets_k_profiles(self, async_client: AsyncClient, db_session, printers):
|
|
|
+ from backend.app.services.filament_profile_default import fill_auto_added_spoolman_spool
|
|
|
+
|
|
|
+ a, _ = printers
|
|
|
+ await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id)], presets=[_preset()]))
|
|
|
+ await fill_auto_added_spoolman_spool(_spoolman_spool(55))
|
|
|
+ assert await _spoolman_rows(db_session, 55) == ([5], ["PFUS01"])
|
|
|
+
|
|
|
+
|
|
|
+@pytest.fixture
|
|
|
+def limited_to(request):
|
|
|
+ """Run the request as a caller limited to some printers (#1727)."""
|
|
|
+ from backend.app.core.auth import get_printer_scope_if_auth_enabled
|
|
|
+ from backend.app.core.printer_scope import PrinterScope
|
|
|
+ from backend.app.main import app
|
|
|
+
|
|
|
+ def _limit(*printer_ids):
|
|
|
+ app.dependency_overrides[get_printer_scope_if_auth_enabled] = lambda: PrinterScope(frozenset(printer_ids))
|
|
|
+
|
|
|
+ yield _limit
|
|
|
+ app.dependency_overrides.pop(get_printer_scope_if_auth_enabled, None)
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.integration
|
|
|
+class TestPrinterScope:
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_limited_caller_sees_only_its_printers(self, async_client: AsyncClient, printers, limited_to):
|
|
|
+ a, b = printers
|
|
|
+ await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id), _k(b.id)]))
|
|
|
+ limited_to(b.id)
|
|
|
+ listed = (await async_client.get(DEFAULTS)).json()[0]
|
|
|
+ assert [k["printer_id"] for k in listed["k_profiles"]] == [b.id]
|
|
|
+ resolved = (await async_client.get(f"{DEFAULTS}/resolve", params=_spool_body())).json()
|
|
|
+ assert [k["printer_id"] for k in resolved["k_profiles"]] == [b.id]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_limited_save_keeps_the_printers_it_cannot_see(
|
|
|
+ self, async_client: AsyncClient, printers, limited_to
|
|
|
+ ):
|
|
|
+ a, b = printers
|
|
|
+ await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id), _k(b.id)]))
|
|
|
+ limited_to(b.id)
|
|
|
+ response = await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(b.id, cali_idx=8)]))
|
|
|
+ assert response.status_code == 200, response.text
|
|
|
+ limited_to(a.id, b.id)
|
|
|
+ kept = (await async_client.get(DEFAULTS)).json()[0]
|
|
|
+ assert {k["printer_id"]: k["cali_idx"] for k in kept["k_profiles"]} == {a.id: 5, b.id: 8}
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_limited_caller_cannot_write_another_printer(self, async_client: AsyncClient, printers, limited_to):
|
|
|
+ # The row is ignored, not refused: a form opened on a spool carries
|
|
|
+ # every printer's profile, and saving it as the default must work.
|
|
|
+ a, b = printers
|
|
|
+ limited_to(b.id)
|
|
|
+ response = await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id), _k(b.id)]))
|
|
|
+ assert response.status_code == 200, response.text
|
|
|
+ limited_to(a.id, b.id)
|
|
|
+ kept = (await async_client.get(DEFAULTS)).json()[0]
|
|
|
+ assert [k["printer_id"] for k in kept["k_profiles"]] == [b.id]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_limited_caller_cannot_delete_another_printers_profiles(
|
|
|
+ self, async_client: AsyncClient, printers, limited_to
|
|
|
+ ):
|
|
|
+ a, b = printers
|
|
|
+ row = (await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id)]))).json()
|
|
|
+ limited_to(b.id)
|
|
|
+ assert (await async_client.delete(f"{DEFAULTS}/{row['id']}")).status_code == 403
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.integration
|
|
|
+class TestInternalPresetFill:
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_spool_keeps_its_own_preset_for_a_model(self, async_client: AsyncClient):
|
|
|
+ spool = (await async_client.post(INTERNAL, json=_spool_body())).json()
|
|
|
+ await async_client.put(f"{INTERNAL}/{spool['id']}/filament-presets", json=[_preset("H2D", code="OWN01")])
|
|
|
+ row = (
|
|
|
+ await async_client.put(DEFAULTS, json=_default(presets=[_preset("H2D"), _preset("X1C", code="PFUS02")]))
|
|
|
+ ).json()
|
|
|
+ assert (await async_client.post(f"{DEFAULTS}/{row['id']}/apply")).json() == {"updated": 1}
|
|
|
+ presets = (await async_client.get(f"{INTERNAL}/{spool['id']}/filament-presets")).json()
|
|
|
+ assert {p["printer_model"]: p["slicer_filament"] for p in presets} == {"H2D": "OWN01", "X1C": "PFUS02"}
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.integration
|
|
|
+class TestNullKeyFields:
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_null_brand_subtype_and_colour_read_as_empty(self, async_client: AsyncClient):
|
|
|
+ response = await async_client.put(DEFAULTS, json=_default(brand=None, subtype=None, color_name=None))
|
|
|
+ assert response.status_code == 200, response.text
|
|
|
+ assert {k: response.json()[k] for k in ("brand", "subtype", "color_name")} == {
|
|
|
+ "brand": "",
|
|
|
+ "subtype": "",
|
|
|
+ "color_name": "",
|
|
|
+ }
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_null_material_is_refused_not_a_500(self, async_client: AsyncClient):
|
|
|
+ assert (await async_client.put(DEFAULTS, json=_default(material=None))).status_code == 422
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.integration
|
|
|
+class TestReviewFixes:
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_non_ascii_colour_matches_its_own_row(self, async_client: AsyncClient, printers):
|
|
|
+ # SQLite's lower() folds ASCII only.
|
|
|
+ a, _ = printers
|
|
|
+ await async_client.put(DEFAULTS, json=_default("Красный", k_profiles=[_k(a.id, cali_idx=9)]))
|
|
|
+ response = await async_client.put(DEFAULTS, json=_default("Красный", k_profiles=[_k(a.id, cali_idx=8)]))
|
|
|
+ assert response.status_code == 200, response.text
|
|
|
+ assert len((await async_client.get(DEFAULTS)).json()) == 1
|
|
|
+ resolved = (await async_client.get(f"{DEFAULTS}/resolve", params=_spool_body(color_name="Красный"))).json()
|
|
|
+ assert [k["cali_idx"] for k in resolved["k_profiles"]] == [8]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_limited_k_profile_save_keeps_the_other_printers(
|
|
|
+ self, async_client: AsyncClient, printers, limited_to
|
|
|
+ ):
|
|
|
+ a, b = printers
|
|
|
+ await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id), _k(b.id)]))
|
|
|
+ spool = (await async_client.post(INTERNAL, json=_spool_body())).json()
|
|
|
+ limited_to(b.id)
|
|
|
+ # The limited form only ever saw printer b.
|
|
|
+ response = await async_client.put(f"{INTERNAL}/{spool['id']}/k-profiles", json=[_k(b.id, cali_idx=7)])
|
|
|
+ assert response.status_code == 200, response.text
|
|
|
+ limited_to(a.id, b.id)
|
|
|
+ spool = (await async_client.get(f"{INTERNAL}/{spool['id']}")).json()
|
|
|
+ assert {k["printer_id"]: k["cali_idx"] for k in spool["k_profiles"]} == {a.id: 5, b.id: 7}
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_limited_spoolman_k_profile_save_keeps_the_other_printers(
|
|
|
+ self, async_client: AsyncClient, db_session, printers, limited_to, spoolman_client
|
|
|
+ ):
|
|
|
+ from backend.app.models.spoolman_k_profile import SpoolmanKProfile
|
|
|
+
|
|
|
+ a, b = printers
|
|
|
+ spoolman_client.get_spool = AsyncMock(return_value=_spoolman_spool(9))
|
|
|
+ db_session.add(SpoolmanKProfile(spoolman_spool_id=9, **_k(a.id)))
|
|
|
+ db_session.add(SpoolmanKProfile(spoolman_spool_id=9, **_k(b.id)))
|
|
|
+ await db_session.commit()
|
|
|
+ limited_to(b.id)
|
|
|
+ response = await async_client.put(f"{SPOOLMAN}/spools/9/k-profiles", json=[_k(b.id, cali_idx=7)])
|
|
|
+ assert response.status_code == 200, response.text
|
|
|
+ rows = (
|
|
|
+ (await db_session.execute(select(SpoolmanKProfile).where(SpoolmanKProfile.spoolman_spool_id == 9)))
|
|
|
+ .scalars()
|
|
|
+ .all()
|
|
|
+ )
|
|
|
+ assert {r.printer_id: r.cali_idx for r in rows} == {a.id: 5, b.id: 7}
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ @pytest.mark.parametrize("include_slicer", [True, False])
|
|
|
+ async def test_an_auto_added_spoolman_spool_gets_the_preset_only_without_one_of_its_own(
|
|
|
+ self, async_client: AsyncClient, include_slicer
|
|
|
+ ):
|
|
|
+ from backend.app.services.filament_profile_default import fill_auto_added_spoolman_spool
|
|
|
+
|
|
|
+ await async_client.put(DEFAULTS, json=_default())
|
|
|
+ client = MagicMock()
|
|
|
+ client.ensure_extra_field = AsyncMock(return_value=True)
|
|
|
+ client.merge_spool_extra = AsyncMock(return_value={})
|
|
|
+ await fill_auto_added_spoolman_spool(_spoolman_spool(56), client, include_slicer=include_slicer)
|
|
|
+ if include_slicer:
|
|
|
+ client.merge_spool_extra.assert_awaited_once()
|
|
|
+ assert client.merge_spool_extra.call_args.args[1]["bambu_slicer_filament"] == '"PFUS00"'
|
|
|
+ else:
|
|
|
+ client.merge_spool_extra.assert_not_awaited()
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_apply_folds_non_ascii_case_the_same_way_as_the_lookup(self, async_client: AsyncClient, printers):
|
|
|
+ a, _ = printers
|
|
|
+ spool = (await async_client.post(INTERNAL, json=_spool_body(color_name="красный"))).json()
|
|
|
+ row = (await async_client.put(DEFAULTS, json=_default("Красный", k_profiles=[_k(a.id, cali_idx=9)]))).json()
|
|
|
+ await async_client.put(DEFAULTS, json=_default(k_profiles=[_k(a.id)]))
|
|
|
+ assert (await async_client.post(f"{DEFAULTS}/{row['id']}/apply")).json() == {"updated": 1}
|
|
|
+ spool = (await async_client.get(f"{INTERNAL}/{spool['id']}")).json()
|
|
|
+ assert [k["cali_idx"] for k in spool["k_profiles"]] == [9]
|