test_slicer_presets.py 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638
  1. """Tests for the unified slicer-presets endpoint helpers.
  2. The endpoint stitches together three preset sources (cloud / local /
  3. standard) with name-based dedup. These tests pin the dedup logic, the
  4. cloud-status mapping, and the per-user / sidecar caches at the
  5. helper level — full HTTP integration is covered by the routes test.
  6. """
  7. from __future__ import annotations
  8. import time
  9. from unittest.mock import AsyncMock, MagicMock, patch
  10. import pytest
  11. from backend.app.api.routes import slicer_presets as sp
  12. from backend.app.schemas.slicer_presets import UnifiedPreset
  13. def _slot(items: list[tuple[str, str, str]]) -> dict[str, list[UnifiedPreset]]:
  14. """Helper: build a single-slot dict from (id, name, source) tuples placed
  15. on the printer slot. Process / filament default to empty so each test
  16. only exercises the slot it cares about."""
  17. return {
  18. "printer": [UnifiedPreset(id=i, name=n, source=s) for i, n, s in items],
  19. "process": [],
  20. "filament": [],
  21. }
  22. class TestDedupeByName:
  23. """Cloud > local > standard, by ``name``, order preserved within tier."""
  24. def test_cloud_wins_over_local_and_standard(self):
  25. cloud = _slot([("cid1", "Bambu PLA Basic", "cloud")])
  26. local = _slot([("lid1", "Bambu PLA Basic", "local")])
  27. standard = _slot([("Bambu PLA Basic", "Bambu PLA Basic", "standard")])
  28. c, l_, s = sp._dedupe_by_name(cloud, local, standard)
  29. assert [p.source for p in c["printer"]] == ["cloud"]
  30. assert l_["printer"] == []
  31. assert s["printer"] == []
  32. def test_local_filtered_only_when_present_in_cloud(self):
  33. cloud = _slot([("cid1", "Custom PLA", "cloud")])
  34. local = _slot(
  35. [
  36. ("lid1", "Custom PLA", "local"), # filtered (in cloud)
  37. ("lid2", "My Workhorse PLA", "local"), # kept
  38. ]
  39. )
  40. standard = _slot([])
  41. _c, l_, _s = sp._dedupe_by_name(cloud, local, standard)
  42. assert [p.name for p in l_["printer"]] == ["My Workhorse PLA"]
  43. def test_standard_filtered_against_both_higher_tiers(self):
  44. cloud = _slot([("c1", "A", "cloud")])
  45. local = _slot([("l1", "B", "local")])
  46. standard = _slot(
  47. [
  48. ("A", "A", "standard"), # filtered (in cloud)
  49. ("B", "B", "standard"), # filtered (in local)
  50. ("C", "C", "standard"), # kept
  51. ]
  52. )
  53. _c, _l, s = sp._dedupe_by_name(cloud, local, standard)
  54. assert [p.name for p in s["printer"]] == ["C"]
  55. def test_preserves_order_within_tier(self):
  56. """A tier's input order must be preserved in its output — nothing in
  57. the dedupe pass should sort, reverse, or otherwise reorder entries."""
  58. cloud = _slot(
  59. [
  60. ("c1", "Z-First", "cloud"),
  61. ("c2", "A-Second", "cloud"),
  62. ("c3", "M-Third", "cloud"),
  63. ]
  64. )
  65. c, _l, _s = sp._dedupe_by_name(cloud, _slot([]), _slot([]))
  66. assert [p.name for p in c["printer"]] == ["Z-First", "A-Second", "M-Third"]
  67. def test_dedupe_is_per_slot(self):
  68. """A name colliding across DIFFERENT slots must NOT cross-filter —
  69. a "Custom" filament shouldn't hide a "Custom" printer."""
  70. cloud = {
  71. "printer": [],
  72. "process": [],
  73. "filament": [UnifiedPreset(id="cf1", name="Custom", source="cloud")],
  74. }
  75. local = {
  76. "printer": [UnifiedPreset(id="lp1", name="Custom", source="local")],
  77. "process": [],
  78. "filament": [],
  79. }
  80. _c, l_, _s = sp._dedupe_by_name(cloud, local, _slot([]))
  81. # The filament-tier collision must NOT remove the printer-tier "Custom".
  82. assert [p.name for p in l_["printer"]] == ["Custom"]
  83. def _user_with_cloud_auth(user_id: int = 1) -> MagicMock:
  84. """Construct a mock User that passes the CLOUD_AUTH permission check.
  85. `MagicMock` defaults `.has_permission(...)` to a truthy MagicMock object,
  86. which would coincidentally pass the gate — but explicit is better than
  87. accidental. Setting `.return_value = True` documents the intent."""
  88. user = MagicMock(id=user_id)
  89. user.has_permission = MagicMock(return_value=True)
  90. return user
  91. class TestFetchCloudPresets:
  92. """`_fetch_cloud_presets` translates token state and cloud errors into
  93. the four ``cloud_status`` values the SliceModal banner consumes."""
  94. @pytest.mark.asyncio
  95. async def test_no_token_returns_not_authenticated(self):
  96. sp._cloud_cache.clear()
  97. with patch.object(sp, "get_stored_token", AsyncMock(return_value=(None, None, None))):
  98. slots, status = await sp._fetch_cloud_presets(MagicMock(), _user_with_cloud_auth())
  99. assert status == "not_authenticated"
  100. assert slots == {"printer": [], "process": [], "filament": []}
  101. @pytest.mark.asyncio
  102. async def test_user_without_cloud_auth_returns_not_authenticated(self):
  103. """Defence-in-depth: a user lacking CLOUD_AUTH must NOT see cloud
  104. presets even if their User row carries a stale cloud_token from a
  105. previous permission state. Token lookup is skipped entirely."""
  106. sp._cloud_cache.clear()
  107. user = MagicMock(id=1)
  108. user.has_permission = MagicMock(return_value=False)
  109. with patch.object(sp, "get_stored_token", AsyncMock(return_value=("leftover-token", None, None))) as get_tok:
  110. slots, status = await sp._fetch_cloud_presets(MagicMock(), user)
  111. assert status == "not_authenticated"
  112. assert slots["printer"] == []
  113. # Token was never read — the perm check short-circuits ahead of it.
  114. get_tok.assert_not_called()
  115. @pytest.mark.asyncio
  116. async def test_auth_error_returns_expired(self):
  117. sp._cloud_cache.clear()
  118. cloud_mock = MagicMock()
  119. cloud_mock.set_token = MagicMock()
  120. cloud_mock.get_slicer_settings = AsyncMock(side_effect=sp.BambuCloudAuthError("expired"))
  121. cloud_mock.close = AsyncMock()
  122. with (
  123. patch.object(sp, "get_stored_token", AsyncMock(return_value=("tok", "e@x", None))),
  124. patch.object(sp, "BambuCloudService", return_value=cloud_mock),
  125. ):
  126. slots, status = await sp._fetch_cloud_presets(MagicMock(), _user_with_cloud_auth())
  127. assert status == "expired"
  128. assert slots["printer"] == []
  129. cloud_mock.close.assert_awaited_once()
  130. @pytest.mark.asyncio
  131. async def test_cloud_error_returns_unreachable(self):
  132. sp._cloud_cache.clear()
  133. cloud_mock = MagicMock()
  134. cloud_mock.set_token = MagicMock()
  135. cloud_mock.get_slicer_settings = AsyncMock(side_effect=sp.BambuCloudError("net down"))
  136. cloud_mock.close = AsyncMock()
  137. with (
  138. patch.object(sp, "get_stored_token", AsyncMock(return_value=("tok", None, None))),
  139. patch.object(sp, "BambuCloudService", return_value=cloud_mock),
  140. ):
  141. _slots, status = await sp._fetch_cloud_presets(MagicMock(), _user_with_cloud_auth())
  142. assert status == "unreachable"
  143. @pytest.mark.asyncio
  144. async def test_happy_path_shapes_private_then_public(self):
  145. """Cloud presets split into private (user-custom) + public (Bambu's
  146. stock cloud presets). Private should sort before public so a user's
  147. own customisations sit at the top of the dropdown."""
  148. sp._cloud_cache.clear()
  149. cloud_mock = MagicMock()
  150. cloud_mock.set_token = MagicMock()
  151. cloud_mock.get_slicer_settings = AsyncMock(
  152. return_value={
  153. "printer": {
  154. "private": [{"setting_id": "PFUprivate1", "name": "My X1C"}],
  155. "public": [{"setting_id": "PFUpublic1", "name": "Bambu X1C Stock"}],
  156. },
  157. "print": {"private": [], "public": []},
  158. "filament": {"private": [], "public": []},
  159. }
  160. )
  161. cloud_mock.close = AsyncMock()
  162. with (
  163. patch.object(sp, "get_stored_token", AsyncMock(return_value=("tok", None, None))),
  164. patch.object(sp, "BambuCloudService", return_value=cloud_mock),
  165. ):
  166. slots, status = await sp._fetch_cloud_presets(MagicMock(), _user_with_cloud_auth())
  167. assert status == "ok"
  168. names = [p.name for p in slots["printer"]]
  169. assert names == ["My X1C", "Bambu X1C Stock"]
  170. @pytest.mark.asyncio
  171. async def test_cache_hit_skips_cloud_call(self):
  172. """A second call within TTL must reuse the cached slots and NOT
  173. hit Bambu Cloud again."""
  174. sp._cloud_cache.clear()
  175. cloud_mock = MagicMock()
  176. cloud_mock.set_token = MagicMock()
  177. cloud_mock.get_slicer_settings = AsyncMock(
  178. return_value={
  179. "printer": {"private": [{"setting_id": "id1", "name": "X1C"}], "public": []},
  180. "print": {"private": [], "public": []},
  181. "filament": {"private": [], "public": []},
  182. }
  183. )
  184. cloud_mock.close = AsyncMock()
  185. user = _user_with_cloud_auth(user_id=42)
  186. with (
  187. patch.object(sp, "get_stored_token", AsyncMock(return_value=("tok", None, None))),
  188. patch.object(sp, "BambuCloudService", return_value=cloud_mock),
  189. ):
  190. await sp._fetch_cloud_presets(MagicMock(), user)
  191. await sp._fetch_cloud_presets(MagicMock(), user)
  192. cloud_mock.get_slicer_settings.assert_awaited_once()
  193. @pytest.mark.asyncio
  194. async def test_cache_is_per_user(self):
  195. """User A's cached cloud presets must not surface for user B."""
  196. sp._cloud_cache.clear()
  197. def make_mock(name: str):
  198. m = MagicMock()
  199. m.set_token = MagicMock()
  200. m.get_slicer_settings = AsyncMock(
  201. return_value={
  202. "printer": {"private": [{"setting_id": f"id-{name}", "name": name}], "public": []},
  203. "print": {"private": [], "public": []},
  204. "filament": {"private": [], "public": []},
  205. }
  206. )
  207. m.close = AsyncMock()
  208. return m
  209. sequence = [make_mock("AliceX1C"), make_mock("BobX1C")]
  210. with (
  211. patch.object(sp, "get_stored_token", AsyncMock(return_value=("tok", None, None))),
  212. patch.object(sp, "BambuCloudService", side_effect=sequence),
  213. ):
  214. alice_slots, _ = await sp._fetch_cloud_presets(MagicMock(), _user_with_cloud_auth(1))
  215. bob_slots, _ = await sp._fetch_cloud_presets(MagicMock(), _user_with_cloud_auth(2))
  216. assert alice_slots["printer"][0].name == "AliceX1C"
  217. assert bob_slots["printer"][0].name == "BobX1C"
  218. @pytest.mark.asyncio
  219. async def test_cache_invalidates_on_token_change(self):
  220. """A token change (logout + login, admin reset, region switch) must
  221. bypass the cache for that user — pinning a real-world auth bug
  222. where user re-login + cache-stuck-on-old-cloud-account would
  223. silently serve a different account's preset list for ~5 minutes."""
  224. sp._cloud_cache.clear()
  225. def make_mock(name: str):
  226. m = MagicMock()
  227. m.set_token = MagicMock()
  228. m.get_slicer_settings = AsyncMock(
  229. return_value={
  230. "printer": {"private": [{"setting_id": f"id-{name}", "name": name}], "public": []},
  231. "print": {"private": [], "public": []},
  232. "filament": {"private": [], "public": []},
  233. }
  234. )
  235. m.close = AsyncMock()
  236. return m
  237. # Same user_id, different token between calls — the second call must
  238. # NOT serve the first call's cached slots.
  239. services = [make_mock("OldAccountX1C"), make_mock("NewAccountX1C")]
  240. token_sequence = [("tok-old", None, None), ("tok-new", None, None)]
  241. user = _user_with_cloud_auth(user_id=7)
  242. with (
  243. patch.object(sp, "get_stored_token", AsyncMock(side_effect=token_sequence)),
  244. patch.object(sp, "BambuCloudService", side_effect=services),
  245. ):
  246. first, _ = await sp._fetch_cloud_presets(MagicMock(), user)
  247. second, _ = await sp._fetch_cloud_presets(MagicMock(), user)
  248. assert first["printer"][0].name == "OldAccountX1C"
  249. assert second["printer"][0].name == "NewAccountX1C"
  250. class TestFetchBundledPresets:
  251. """Standard tier reaches out to the slicer-api sidecar; tolerate the
  252. sidecar being absent / unreachable so the modal still works."""
  253. @pytest.mark.asyncio
  254. async def test_no_sidecar_url_returns_empty(self):
  255. sp._bundled_cache = None
  256. with patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value=None)):
  257. slots = await sp._fetch_bundled_presets(MagicMock())
  258. assert slots == {"printer": [], "process": [], "filament": []}
  259. # No URL means no useful cache result either — second call should
  260. # try again (so users who configure a URL mid-session see results).
  261. assert sp._bundled_cache is None
  262. @pytest.mark.asyncio
  263. async def test_sidecar_error_returns_empty(self):
  264. sp._bundled_cache = None
  265. svc_mock = MagicMock()
  266. svc_mock.list_bundled_profiles = AsyncMock(side_effect=sp.SlicerApiError("boom"))
  267. svc_mock.__aenter__ = AsyncMock(return_value=svc_mock)
  268. svc_mock.__aexit__ = AsyncMock(return_value=False)
  269. with (
  270. patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value="http://nope")),
  271. patch.object(sp, "SlicerApiService", return_value=svc_mock),
  272. ):
  273. slots = await sp._fetch_bundled_presets(MagicMock())
  274. assert slots == {"printer": [], "process": [], "filament": []}
  275. @pytest.mark.asyncio
  276. async def test_happy_path_shapes_response(self):
  277. sp._bundled_cache = None
  278. svc_mock = MagicMock()
  279. svc_mock.list_bundled_profiles = AsyncMock(
  280. return_value={
  281. "printer": [{"name": "Bambu X1C 0.4", "base_id": None}],
  282. "process": [{"name": "0.20mm Standard", "base_id": "fdm_process_common"}],
  283. "filament": [{"name": "Bambu PLA Basic", "base_id": "fdm_filament_pla"}],
  284. }
  285. )
  286. svc_mock.__aenter__ = AsyncMock(return_value=svc_mock)
  287. svc_mock.__aexit__ = AsyncMock(return_value=False)
  288. with (
  289. patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value="http://ok")),
  290. patch.object(sp, "SlicerApiService", return_value=svc_mock),
  291. ):
  292. slots = await sp._fetch_bundled_presets(MagicMock())
  293. assert slots["printer"][0].name == "Bambu X1C 0.4"
  294. assert slots["printer"][0].source == "standard"
  295. # Bundled presets are addressed by name (the slicer's inheritance
  296. # walker resolves them by name), so id == name.
  297. assert slots["printer"][0].id == "Bambu X1C 0.4"
  298. @pytest.mark.asyncio
  299. async def test_cache_hit_skips_sidecar(self):
  300. """A second call within TTL must serve from the cached entry and not
  301. re-hit the sidecar HTTP."""
  302. sp._bundled_cache = (
  303. time.monotonic(),
  304. {
  305. "printer": [UnifiedPreset(id="Cached", name="Cached", source="standard")],
  306. "process": [],
  307. "filament": [],
  308. },
  309. )
  310. # If `SlicerApiService` is constructed at all we've missed the cache.
  311. with patch.object(sp, "SlicerApiService", side_effect=AssertionError("cache miss!")):
  312. slots = await sp._fetch_bundled_presets(MagicMock())
  313. assert slots["printer"][0].name == "Cached"
  314. class TestResolveSlicerApiUrl:
  315. """`_resolve_slicer_api_url` must respect the user's `preferred_slicer`
  316. setting just like the slice route does. The bundled-listing fetch
  317. used to be hardcoded to OrcaSlicer's URL, which left the Standard
  318. tier permanently empty for BambuStudio installs."""
  319. @pytest.mark.asyncio
  320. async def test_bambu_studio_preference_uses_bambu_url(self):
  321. """When the user prefers Bambu Studio, the listing fetch must hit
  322. the bambu-studio-api sidecar (port 3001 by default), not orca's
  323. port 3003."""
  324. async def fake_get_setting(_db, key):
  325. return {
  326. "preferred_slicer": "bambu_studio",
  327. "bambu_studio_api_url": "http://bambu-studio-api:3000",
  328. }.get(key)
  329. with patch(
  330. "backend.app.api.routes.settings.get_setting",
  331. new=fake_get_setting,
  332. ):
  333. url = await sp._resolve_slicer_api_url(MagicMock())
  334. assert url == "http://bambu-studio-api:3000"
  335. @pytest.mark.asyncio
  336. async def test_orcaslicer_preference_uses_orca_url(self):
  337. async def fake_get_setting(_db, key):
  338. return {
  339. "preferred_slicer": "orcaslicer",
  340. "orcaslicer_api_url": "http://orca-slicer-api:3000",
  341. }.get(key)
  342. with patch(
  343. "backend.app.api.routes.settings.get_setting",
  344. new=fake_get_setting,
  345. ):
  346. url = await sp._resolve_slicer_api_url(MagicMock())
  347. assert url == "http://orca-slicer-api:3000"
  348. @pytest.mark.asyncio
  349. async def test_default_preference_is_bambu_studio(self):
  350. """Empty preferred_slicer → bambu_studio (matches the slice route's
  351. default at library.py:_run_slicer_with_fallback)."""
  352. async def fake_get_setting(_db, key):
  353. return {
  354. # preferred_slicer not set
  355. "bambu_studio_api_url": "http://bambu-default:3000",
  356. }.get(key)
  357. with patch(
  358. "backend.app.api.routes.settings.get_setting",
  359. new=fake_get_setting,
  360. ):
  361. url = await sp._resolve_slicer_api_url(MagicMock())
  362. assert url == "http://bambu-default:3000"
  363. @pytest.mark.asyncio
  364. async def test_unknown_preference_returns_none(self):
  365. """An unrecognised preferred_slicer value (e.g. set out-of-band by
  366. a stale migration) returns None so the modal degrades to "no
  367. Standard tier" rather than crashing — the slice route raises 400
  368. in this case but the listing is informational, so be lenient."""
  369. async def fake_get_setting(_db, key):
  370. return {"preferred_slicer": "prusaslicer"}.get(key)
  371. with patch(
  372. "backend.app.api.routes.settings.get_setting",
  373. new=fake_get_setting,
  374. ):
  375. url = await sp._resolve_slicer_api_url(MagicMock())
  376. assert url is None
  377. class TestBundleRoutes:
  378. """Route-level coverage for the bundle proxy endpoints. Each route
  379. resolves the sidecar URL via _resolve_slicer_api_url, then proxies the
  380. operation through SlicerApiService. We mock both pieces so we can pin
  381. the HTTP-status mapping (sidecar input error → 400, BundleNotFoundError
  382. → 404, unreachable → 503) without spinning up a sidecar.
  383. """
  384. SAMPLE_SUMMARY = sp.BundleSummary(
  385. id="abc123def456abcd",
  386. printer_preset_name="# Bambu Lab H2D 0.4 nozzle",
  387. printer=["# Bambu Lab H2D 0.4 nozzle"],
  388. process=["# 0.20mm Standard @BBL H2D"],
  389. filament=["# Bambu PLA Basic @BBL H2D"],
  390. version="02.06.00.50",
  391. )
  392. def _patched_service(self, **methods) -> MagicMock:
  393. """Build a SlicerApiService mock that supports `async with` and
  394. exposes the bundle methods via AsyncMock per the override dict."""
  395. svc = MagicMock()
  396. svc.__aenter__ = AsyncMock(return_value=svc)
  397. svc.__aexit__ = AsyncMock(return_value=False)
  398. for name, mock in methods.items():
  399. setattr(svc, name, mock)
  400. return svc
  401. @pytest.mark.asyncio
  402. async def test_import_bundle_happy_path(self):
  403. from io import BytesIO
  404. from fastapi import UploadFile
  405. svc = self._patched_service(
  406. import_bundle=AsyncMock(return_value=self.SAMPLE_SUMMARY),
  407. )
  408. with (
  409. patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value="http://ok")),
  410. patch.object(sp, "SlicerApiService", return_value=svc),
  411. ):
  412. file = UploadFile(filename="H2D.bbscfg", file=BytesIO(b"PK\x03\x04"))
  413. result = await sp.import_slicer_bundle(file=file, db=MagicMock(), _=None)
  414. assert result["id"] == "abc123def456abcd"
  415. assert result["printer"] == ["# Bambu Lab H2D 0.4 nozzle"]
  416. svc.import_bundle.assert_awaited_once()
  417. kwargs = svc.import_bundle.await_args.kwargs
  418. assert kwargs["filename"] == "H2D.bbscfg"
  419. @pytest.mark.asyncio
  420. async def test_import_bundle_no_sidecar_returns_503(self):
  421. from io import BytesIO
  422. from fastapi import HTTPException, UploadFile
  423. with (
  424. patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value=None)),
  425. pytest.raises(HTTPException) as exc,
  426. ):
  427. await sp.import_slicer_bundle(
  428. file=UploadFile(filename="x.bbscfg", file=BytesIO(b"x")),
  429. db=MagicMock(),
  430. _=None,
  431. )
  432. assert exc.value.status_code == 503
  433. @pytest.mark.asyncio
  434. async def test_import_bundle_empty_file_returns_400(self):
  435. from io import BytesIO
  436. from fastapi import HTTPException, UploadFile
  437. with (
  438. patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value="http://ok")),
  439. pytest.raises(HTTPException) as exc,
  440. ):
  441. await sp.import_slicer_bundle(
  442. file=UploadFile(filename="x.bbscfg", file=BytesIO(b"")),
  443. db=MagicMock(),
  444. _=None,
  445. )
  446. assert exc.value.status_code == 400
  447. @pytest.mark.asyncio
  448. async def test_import_bundle_sidecar_400_passes_through(self):
  449. from io import BytesIO
  450. from fastapi import HTTPException, UploadFile
  451. svc = self._patched_service(
  452. import_bundle=AsyncMock(side_effect=sp.SlicerInputError("bad zip")),
  453. )
  454. with (
  455. patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value="http://ok")),
  456. patch.object(sp, "SlicerApiService", return_value=svc),
  457. pytest.raises(HTTPException) as exc,
  458. ):
  459. await sp.import_slicer_bundle(
  460. file=UploadFile(filename="x.bbscfg", file=BytesIO(b"x")),
  461. db=MagicMock(),
  462. _=None,
  463. )
  464. assert exc.value.status_code == 400
  465. @pytest.mark.asyncio
  466. async def test_import_bundle_sidecar_unreachable_returns_503(self):
  467. from io import BytesIO
  468. from fastapi import HTTPException, UploadFile
  469. svc = self._patched_service(
  470. import_bundle=AsyncMock(side_effect=sp.SlicerApiUnavailableError("offline")),
  471. )
  472. with (
  473. patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value="http://ok")),
  474. patch.object(sp, "SlicerApiService", return_value=svc),
  475. pytest.raises(HTTPException) as exc,
  476. ):
  477. await sp.import_slicer_bundle(
  478. file=UploadFile(filename="x.bbscfg", file=BytesIO(b"x")),
  479. db=MagicMock(),
  480. _=None,
  481. )
  482. assert exc.value.status_code == 503
  483. @pytest.mark.asyncio
  484. async def test_list_bundles_happy_path(self):
  485. svc = self._patched_service(
  486. list_bundles=AsyncMock(return_value=[self.SAMPLE_SUMMARY]),
  487. )
  488. with (
  489. patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value="http://ok")),
  490. patch.object(sp, "SlicerApiService", return_value=svc),
  491. ):
  492. result = await sp.list_slicer_bundles(db=MagicMock(), _=None)
  493. assert len(result) == 1
  494. assert result[0]["id"] == "abc123def456abcd"
  495. @pytest.mark.asyncio
  496. async def test_list_bundles_no_sidecar_returns_empty(self):
  497. # Differs from import: list returns [] instead of 503 so the
  498. # SliceModal still renders cleanly when no sidecar is configured
  499. # (matches bundled-tier behaviour above).
  500. with patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value=None)):
  501. result = await sp.list_slicer_bundles(db=MagicMock(), _=None)
  502. assert result == []
  503. @pytest.mark.asyncio
  504. async def test_list_bundles_sidecar_unreachable_returns_503(self):
  505. from fastapi import HTTPException
  506. svc = self._patched_service(
  507. list_bundles=AsyncMock(side_effect=sp.SlicerApiUnavailableError("offline")),
  508. )
  509. with (
  510. patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value="http://ok")),
  511. patch.object(sp, "SlicerApiService", return_value=svc),
  512. pytest.raises(HTTPException) as exc,
  513. ):
  514. await sp.list_slicer_bundles(db=MagicMock(), _=None)
  515. assert exc.value.status_code == 503
  516. @pytest.mark.asyncio
  517. async def test_get_bundle_404(self):
  518. from fastapi import HTTPException
  519. svc = self._patched_service(
  520. get_bundle=AsyncMock(side_effect=sp.BundleNotFoundError("not found")),
  521. )
  522. with (
  523. patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value="http://ok")),
  524. patch.object(sp, "SlicerApiService", return_value=svc),
  525. pytest.raises(HTTPException) as exc,
  526. ):
  527. await sp.get_slicer_bundle("missing", db=MagicMock(), _=None)
  528. assert exc.value.status_code == 404
  529. @pytest.mark.asyncio
  530. async def test_delete_bundle_204(self):
  531. # delete returns None on success; FastAPI sends 204 because the route
  532. # declares status_code=204.
  533. svc = self._patched_service(delete_bundle=AsyncMock(return_value=None))
  534. with (
  535. patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value="http://ok")),
  536. patch.object(sp, "SlicerApiService", return_value=svc),
  537. ):
  538. result = await sp.delete_slicer_bundle("abc", db=MagicMock(), _=None)
  539. assert result is None
  540. svc.delete_bundle.assert_awaited_once_with("abc")
  541. @pytest.mark.asyncio
  542. async def test_delete_bundle_404(self):
  543. from fastapi import HTTPException
  544. svc = self._patched_service(
  545. delete_bundle=AsyncMock(side_effect=sp.BundleNotFoundError("not found")),
  546. )
  547. with (
  548. patch.object(sp, "_resolve_slicer_api_url", AsyncMock(return_value="http://ok")),
  549. patch.object(sp, "SlicerApiService", return_value=svc),
  550. pytest.raises(HTTPException) as exc,
  551. ):
  552. await sp.delete_slicer_bundle("missing", db=MagicMock(), _=None)
  553. assert exc.value.status_code == 404