test_spoolman_api.py 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707
  1. """Integration tests for Spoolman API endpoints."""
  2. from unittest.mock import AsyncMock, MagicMock, patch
  3. import pytest
  4. from httpx import AsyncClient
  5. class TestSpoolmanAPI:
  6. """Integration tests for /api/v1/spoolman/ endpoints."""
  7. @pytest.fixture
  8. async def spoolman_settings(self, db_session):
  9. """Create Spoolman settings in the database (enabled with URL)."""
  10. from backend.app.models.settings import Settings
  11. # Both settings are required for Spoolman to work
  12. enabled_setting = Settings(key="spoolman_enabled", value="true")
  13. url_setting = Settings(key="spoolman_url", value="http://localhost:7912")
  14. db_session.add(enabled_setting)
  15. db_session.add(url_setting)
  16. await db_session.commit()
  17. return {"enabled": enabled_setting, "url": url_setting}
  18. @pytest.fixture
  19. async def spoolman_url_only(self, db_session):
  20. """Create only the URL setting (not enabled)."""
  21. from backend.app.models.settings import Settings
  22. setting = Settings(key="spoolman_url", value="http://localhost:7912")
  23. db_session.add(setting)
  24. await db_session.commit()
  25. return setting
  26. @pytest.fixture
  27. def mock_spoolman_client(self):
  28. """Mock the Spoolman client functions."""
  29. mock_client = MagicMock()
  30. mock_client.is_connected = True
  31. mock_client.base_url = "http://localhost:7912"
  32. mock_client.health_check = AsyncMock(return_value=True)
  33. mock_client.ensure_tag_extra_field = AsyncMock(return_value=True)
  34. mock_client.get_spools = AsyncMock(return_value=[])
  35. mock_client.get_filaments = AsyncMock(return_value=[])
  36. mock_client.create_spool = AsyncMock(return_value={"id": 1})
  37. mock_client.update_spool = AsyncMock(return_value={"id": 1})
  38. mock_client.close = AsyncMock()
  39. with (
  40. patch(
  41. "backend.app.api.routes.spoolman.get_spoolman_client",
  42. AsyncMock(return_value=mock_client),
  43. ),
  44. patch(
  45. "backend.app.api.routes.spoolman.init_spoolman_client",
  46. AsyncMock(return_value=mock_client),
  47. ),
  48. patch(
  49. "backend.app.api.routes.spoolman.close_spoolman_client",
  50. AsyncMock(),
  51. ),
  52. ):
  53. yield mock_client
  54. @pytest.fixture
  55. def mock_spoolman_disconnected(self):
  56. """Mock the Spoolman client as disconnected (returns None)."""
  57. with (
  58. patch(
  59. "backend.app.api.routes.spoolman.get_spoolman_client",
  60. AsyncMock(return_value=None),
  61. ),
  62. patch(
  63. "backend.app.api.routes.spoolman.init_spoolman_client",
  64. AsyncMock(return_value=None),
  65. ),
  66. ):
  67. yield
  68. # =========================================================================
  69. # Status Endpoint Tests
  70. # =========================================================================
  71. @pytest.mark.asyncio
  72. @pytest.mark.integration
  73. async def test_get_status_not_configured(self, async_client: AsyncClient):
  74. """Verify status shows not enabled when no settings exist."""
  75. response = await async_client.get("/api/v1/spoolman/status")
  76. assert response.status_code == 200
  77. data = response.json()
  78. assert data["enabled"] is False
  79. assert data["connected"] is False
  80. @pytest.mark.asyncio
  81. @pytest.mark.integration
  82. async def test_get_status_url_only_not_enabled(self, async_client: AsyncClient, spoolman_url_only):
  83. """Verify status shows not enabled when only URL is set."""
  84. response = await async_client.get("/api/v1/spoolman/status")
  85. assert response.status_code == 200
  86. data = response.json()
  87. assert data["enabled"] is False
  88. assert data["url"] == "http://localhost:7912"
  89. @pytest.mark.asyncio
  90. @pytest.mark.integration
  91. async def test_get_status_enabled_and_connected(
  92. self, async_client: AsyncClient, spoolman_settings, mock_spoolman_client
  93. ):
  94. """Verify status shows enabled and connected when properly configured."""
  95. response = await async_client.get("/api/v1/spoolman/status")
  96. assert response.status_code == 200
  97. data = response.json()
  98. assert data["enabled"] is True
  99. assert data["connected"] is True
  100. assert data["url"] == "http://localhost:7912"
  101. # =========================================================================
  102. # Connect/Disconnect Tests
  103. # =========================================================================
  104. @pytest.mark.asyncio
  105. @pytest.mark.integration
  106. async def test_connect_not_enabled(self, async_client: AsyncClient):
  107. """Verify connect fails when not enabled."""
  108. response = await async_client.post("/api/v1/spoolman/connect")
  109. assert response.status_code == 400
  110. assert "not enabled" in response.json()["detail"].lower()
  111. @pytest.mark.asyncio
  112. @pytest.mark.integration
  113. async def test_connect_success(self, async_client: AsyncClient, spoolman_settings, mock_spoolman_client):
  114. """Verify successful connection to Spoolman."""
  115. response = await async_client.post("/api/v1/spoolman/connect")
  116. assert response.status_code == 200
  117. data = response.json()
  118. assert data["success"] is True
  119. assert "connected" in data["message"].lower()
  120. @pytest.mark.asyncio
  121. @pytest.mark.integration
  122. async def test_disconnect(self, async_client: AsyncClient, spoolman_settings, mock_spoolman_client):
  123. """Verify disconnect works."""
  124. response = await async_client.post("/api/v1/spoolman/disconnect")
  125. assert response.status_code == 200
  126. assert "disconnected" in response.json()["message"].lower()
  127. # =========================================================================
  128. # Spools Endpoint Tests
  129. # =========================================================================
  130. @pytest.mark.asyncio
  131. @pytest.mark.integration
  132. async def test_get_spools_not_enabled(self, async_client: AsyncClient):
  133. """Verify get spools fails when not enabled."""
  134. response = await async_client.get("/api/v1/spoolman/spools")
  135. assert response.status_code == 400
  136. assert "not enabled" in response.json()["detail"].lower()
  137. @pytest.mark.asyncio
  138. @pytest.mark.integration
  139. async def test_get_spools_success(self, async_client: AsyncClient, spoolman_settings, mock_spoolman_client):
  140. """Verify get spools returns data in expected format."""
  141. mock_spool = {
  142. "id": 1,
  143. "remaining_weight": 500,
  144. "used_weight": 500,
  145. "filament": {
  146. "id": 1,
  147. "name": "PLA Basic",
  148. "material": "PLA",
  149. "color_hex": "FF0000",
  150. },
  151. "first_used": "2024-01-01",
  152. "last_used": "2024-01-15",
  153. "location": "AMS1",
  154. "lot_nr": "LOT123",
  155. "comment": "Test spool",
  156. "extra": {"tag": '"ABC123"'},
  157. }
  158. mock_spoolman_client.get_spools = AsyncMock(return_value=[mock_spool])
  159. response = await async_client.get("/api/v1/spoolman/spools")
  160. assert response.status_code == 200
  161. data = response.json()
  162. assert "spools" in data
  163. assert isinstance(data["spools"], list)
  164. assert len(data["spools"]) == 1
  165. assert data["spools"][0]["id"] == 1
  166. # =========================================================================
  167. # Unlinked Spools Tests
  168. # =========================================================================
  169. @pytest.mark.asyncio
  170. @pytest.mark.integration
  171. async def test_get_unlinked_spools_not_enabled(self, async_client: AsyncClient):
  172. """Verify get unlinked spools fails when not enabled."""
  173. response = await async_client.get("/api/v1/spoolman/spools/unlinked")
  174. assert response.status_code == 400
  175. @pytest.mark.asyncio
  176. @pytest.mark.integration
  177. async def test_get_unlinked_spools_success(
  178. self, async_client: AsyncClient, spoolman_settings, mock_spoolman_client
  179. ):
  180. """Verify get unlinked spools returns spools without tags."""
  181. # Mock spool without extra.tag (unlinked)
  182. mock_spool = {
  183. "id": 1,
  184. "remaining_weight": 800,
  185. "used_weight": 200,
  186. "extra": {}, # No tag = unlinked
  187. "filament": {
  188. "id": 1,
  189. "name": "PLA Basic",
  190. "material": "PLA",
  191. "color_hex": "FF0000",
  192. },
  193. "location": "Shelf A",
  194. }
  195. mock_spoolman_client.get_spools = AsyncMock(return_value=[mock_spool])
  196. response = await async_client.get("/api/v1/spoolman/spools/unlinked")
  197. assert response.status_code == 200
  198. data = response.json()
  199. assert isinstance(data, list)
  200. assert len(data) == 1
  201. assert data[0]["id"] == 1
  202. assert data[0]["filament_name"] == "PLA Basic"
  203. @pytest.mark.asyncio
  204. @pytest.mark.integration
  205. async def test_get_unlinked_spools_excludes_linked(
  206. self, async_client: AsyncClient, spoolman_settings, mock_spoolman_client
  207. ):
  208. """Verify linked spools (with tag) are excluded."""
  209. # Mock spool with extra.tag (linked)
  210. mock_spool_linked = {
  211. "id": 1,
  212. "remaining_weight": 800,
  213. "used_weight": 200,
  214. "extra": {"tag": '"ABC123"'}, # Has tag = linked
  215. "filament": {"id": 1, "name": "PLA Red", "material": "PLA", "color_hex": "FF0000"},
  216. }
  217. # Mock spool without tag (unlinked)
  218. mock_spool_unlinked = {
  219. "id": 2,
  220. "remaining_weight": 900,
  221. "used_weight": 100,
  222. "extra": {}, # No tag = unlinked
  223. "filament": {"id": 2, "name": "PLA Blue", "material": "PLA", "color_hex": "0000FF"},
  224. }
  225. mock_spoolman_client.get_spools = AsyncMock(return_value=[mock_spool_linked, mock_spool_unlinked])
  226. response = await async_client.get("/api/v1/spoolman/spools/unlinked")
  227. assert response.status_code == 200
  228. data = response.json()
  229. assert len(data) == 1
  230. assert data[0]["id"] == 2 # Only unlinked spool
  231. # =========================================================================
  232. # Linked Spools Tests
  233. # =========================================================================
  234. @pytest.mark.asyncio
  235. @pytest.mark.integration
  236. async def test_get_linked_spools_not_enabled(self, async_client: AsyncClient):
  237. """Verify get linked spools fails when not enabled."""
  238. response = await async_client.get("/api/v1/spoolman/spools/linked")
  239. assert response.status_code == 400
  240. @pytest.mark.asyncio
  241. @pytest.mark.integration
  242. async def test_get_linked_spools_success(self, async_client: AsyncClient, spoolman_settings, mock_spoolman_client):
  243. """Verify get linked spools returns map of tag -> spool_id."""
  244. # Mock spool with extra.tag (linked)
  245. mock_spool = {
  246. "id": 42,
  247. "remaining_weight": 800,
  248. "extra": {"tag": '"A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4"'},
  249. "filament": {"id": 1, "name": "PLA Red", "material": "PLA"},
  250. }
  251. mock_spoolman_client.get_spools = AsyncMock(return_value=[mock_spool])
  252. response = await async_client.get("/api/v1/spoolman/spools/linked")
  253. assert response.status_code == 200
  254. data = response.json()
  255. assert "linked" in data
  256. assert isinstance(data["linked"], dict)
  257. # Tag should be uppercase and stripped of quotes
  258. assert "A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4" in data["linked"]
  259. assert data["linked"]["A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4"] == 42
  260. @pytest.mark.asyncio
  261. @pytest.mark.integration
  262. async def test_get_linked_spools_excludes_unlinked(
  263. self, async_client: AsyncClient, spoolman_settings, mock_spoolman_client
  264. ):
  265. """Verify unlinked spools (without tag) are excluded."""
  266. # Mock spool with tag (linked)
  267. mock_spool_linked = {
  268. "id": 1,
  269. "extra": {"tag": '"ABC12345678901234567890123456789A"'},
  270. "filament": {"id": 1, "name": "PLA Red", "material": "PLA"},
  271. }
  272. # Mock spool without tag (unlinked)
  273. mock_spool_unlinked = {
  274. "id": 2,
  275. "extra": {},
  276. "filament": {"id": 2, "name": "PLA Blue", "material": "PLA"},
  277. }
  278. mock_spoolman_client.get_spools = AsyncMock(return_value=[mock_spool_linked, mock_spool_unlinked])
  279. response = await async_client.get("/api/v1/spoolman/spools/linked")
  280. assert response.status_code == 200
  281. data = response.json()
  282. assert len(data["linked"]) == 1 # Only linked spool
  283. @pytest.mark.asyncio
  284. @pytest.mark.integration
  285. async def test_get_linked_spools_empty_tag_excluded(
  286. self, async_client: AsyncClient, spoolman_settings, mock_spoolman_client
  287. ):
  288. """Verify spools with empty tag (JSON-encoded empty string) are excluded."""
  289. # Mock spool with empty JSON-encoded tag
  290. mock_spool = {
  291. "id": 1,
  292. "extra": {"tag": '""'}, # JSON-encoded empty string
  293. "filament": {"id": 1, "name": "PLA Red", "material": "PLA"},
  294. }
  295. mock_spoolman_client.get_spools = AsyncMock(return_value=[mock_spool])
  296. response = await async_client.get("/api/v1/spoolman/spools/linked")
  297. assert response.status_code == 200
  298. data = response.json()
  299. assert len(data["linked"]) == 0 # Empty tag should be excluded
  300. # =========================================================================
  301. # Link Spool Tests
  302. # =========================================================================
  303. @pytest.mark.asyncio
  304. @pytest.mark.integration
  305. async def test_link_spool_not_enabled(self, async_client: AsyncClient):
  306. """Verify link spool fails when not enabled."""
  307. response = await async_client.post(
  308. "/api/v1/spoolman/spools/1/link",
  309. json={"tray_uuid": "A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4"},
  310. )
  311. assert response.status_code == 400
  312. @pytest.mark.asyncio
  313. @pytest.mark.integration
  314. async def test_link_spool_invalid_uuid_length(
  315. self, async_client: AsyncClient, spoolman_settings, mock_spoolman_client
  316. ):
  317. """Verify link spool fails with invalid UUID length."""
  318. response = await async_client.post(
  319. "/api/v1/spoolman/spools/1/link",
  320. json={"tray_uuid": "ABC123"}, # Too short
  321. )
  322. assert response.status_code == 400
  323. assert "32 hex characters" in response.json()["detail"]
  324. @pytest.mark.asyncio
  325. @pytest.mark.integration
  326. async def test_link_spool_invalid_uuid_format(
  327. self, async_client: AsyncClient, spoolman_settings, mock_spoolman_client
  328. ):
  329. """Verify link spool fails with non-hex UUID."""
  330. response = await async_client.post(
  331. "/api/v1/spoolman/spools/1/link",
  332. json={"tray_uuid": "ZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ"}, # Not hex
  333. )
  334. assert response.status_code == 400
  335. assert "hex" in response.json()["detail"].lower()
  336. @pytest.mark.asyncio
  337. @pytest.mark.integration
  338. async def test_link_spool_success(self, async_client: AsyncClient, spoolman_settings, mock_spoolman_client):
  339. """Verify successfully linking a spool to AMS tray."""
  340. mock_spoolman_client.update_spool = AsyncMock(
  341. return_value={"id": 1, "extra": {"tag": '"A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4"'}}
  342. )
  343. response = await async_client.post(
  344. "/api/v1/spoolman/spools/1/link",
  345. json={"tray_uuid": "A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4"},
  346. )
  347. assert response.status_code == 200
  348. data = response.json()
  349. assert data["success"] is True
  350. assert "linked" in data["message"].lower()
  351. # Verify update_spool was called
  352. mock_spoolman_client.update_spool.assert_called_once()
  353. # =========================================================================
  354. # Sync Tests
  355. # =========================================================================
  356. @pytest.mark.asyncio
  357. @pytest.mark.integration
  358. async def test_sync_printer_not_enabled(self, async_client: AsyncClient, printer_factory):
  359. """Verify sync fails when Spoolman not enabled."""
  360. printer = await printer_factory()
  361. response = await async_client.post(f"/api/v1/spoolman/sync/{printer.id}")
  362. assert response.status_code == 400
  363. assert "not enabled" in response.json()["detail"].lower()
  364. @pytest.mark.asyncio
  365. @pytest.mark.integration
  366. async def test_sync_printer_not_found(self, async_client: AsyncClient, spoolman_settings, mock_spoolman_client):
  367. """Verify sync fails for non-existent printer."""
  368. response = await async_client.post("/api/v1/spoolman/sync/9999")
  369. assert response.status_code == 404
  370. @pytest.mark.asyncio
  371. @pytest.mark.integration
  372. async def test_sync_returns_result_structure(
  373. self,
  374. async_client: AsyncClient,
  375. spoolman_settings,
  376. mock_spoolman_client,
  377. printer_factory,
  378. ):
  379. """Verify sync returns proper result structure."""
  380. printer = await printer_factory()
  381. # Mock printer manager to return AMS data
  382. with patch("backend.app.api.routes.spoolman.printer_manager") as pm_mock:
  383. mock_state = MagicMock()
  384. mock_state.raw_data = {"ams": [{"id": 0, "tray": []}]}
  385. pm_mock.get_status = MagicMock(return_value=mock_state)
  386. response = await async_client.post(f"/api/v1/spoolman/sync/{printer.id}")
  387. assert response.status_code == 200
  388. data = response.json()
  389. # Verify SyncResult structure
  390. assert "success" in data
  391. assert "synced_count" in data
  392. assert "skipped_count" in data
  393. assert "skipped" in data
  394. assert "errors" in data
  395. assert isinstance(data["skipped"], list)
  396. assert isinstance(data["errors"], list)
  397. @pytest.mark.asyncio
  398. @pytest.mark.integration
  399. async def test_sync_printer_not_connected(
  400. self,
  401. async_client: AsyncClient,
  402. spoolman_settings,
  403. mock_spoolman_client,
  404. printer_factory,
  405. ):
  406. """Verify sync fails when printer is not connected (no status)."""
  407. printer = await printer_factory()
  408. with patch("backend.app.api.routes.spoolman.printer_manager") as pm_mock:
  409. pm_mock.get_status = MagicMock(return_value=None)
  410. response = await async_client.post(f"/api/v1/spoolman/sync/{printer.id}")
  411. assert response.status_code == 404
  412. assert "not connected" in response.json()["detail"].lower()
  413. # =========================================================================
  414. # Filaments Endpoint Tests
  415. # =========================================================================
  416. @pytest.mark.asyncio
  417. @pytest.mark.integration
  418. async def test_get_filaments_not_enabled(self, async_client: AsyncClient):
  419. """Verify get filaments fails when not enabled."""
  420. response = await async_client.get("/api/v1/spoolman/filaments")
  421. assert response.status_code == 400
  422. assert "not enabled" in response.json()["detail"].lower()
  423. @pytest.mark.asyncio
  424. @pytest.mark.integration
  425. async def test_get_filaments_success(self, async_client: AsyncClient, spoolman_settings, mock_spoolman_client):
  426. """Verify get filaments returns data in expected format."""
  427. mock_filament = {
  428. "id": 1,
  429. "name": "PLA Basic",
  430. "material": "PLA",
  431. "color_hex": "FF0000",
  432. "vendor_id": 1,
  433. "weight": 1000,
  434. }
  435. mock_spoolman_client.get_filaments = AsyncMock(return_value=[mock_filament])
  436. response = await async_client.get("/api/v1/spoolman/filaments")
  437. assert response.status_code == 200
  438. data = response.json()
  439. assert "filaments" in data
  440. assert isinstance(data["filaments"], list)
  441. assert len(data["filaments"]) == 1
  442. assert data["filaments"][0]["name"] == "PLA Basic"
  443. # =========================================================================
  444. # Disable Weight Sync Tests
  445. # =========================================================================
  446. @pytest.fixture
  447. async def spoolman_settings_weight_sync_disabled(self, db_session):
  448. """Create Spoolman settings with weight sync disabled."""
  449. from backend.app.models.settings import Settings
  450. enabled_setting = Settings(key="spoolman_enabled", value="true")
  451. url_setting = Settings(key="spoolman_url", value="http://localhost:7912")
  452. disable_weight_setting = Settings(key="spoolman_disable_weight_sync", value="true")
  453. partial_usage_setting = Settings(key="spoolman_report_partial_usage", value="true")
  454. db_session.add(enabled_setting)
  455. db_session.add(url_setting)
  456. db_session.add(disable_weight_setting)
  457. db_session.add(partial_usage_setting)
  458. await db_session.commit()
  459. return {
  460. "enabled": enabled_setting,
  461. "url": url_setting,
  462. "disable_weight": disable_weight_setting,
  463. "partial_usage": partial_usage_setting,
  464. }
  465. @pytest.mark.asyncio
  466. @pytest.mark.integration
  467. async def test_settings_returns_disable_weight_sync(
  468. self, async_client: AsyncClient, spoolman_settings_weight_sync_disabled
  469. ):
  470. """Verify settings endpoint returns the disable_weight_sync setting."""
  471. response = await async_client.get("/api/v1/settings/spoolman")
  472. assert response.status_code == 200
  473. data = response.json()
  474. assert "spoolman_disable_weight_sync" in data
  475. assert data["spoolman_disable_weight_sync"] == "true"
  476. @pytest.mark.asyncio
  477. @pytest.mark.integration
  478. async def test_settings_update_disable_weight_sync(self, async_client: AsyncClient, spoolman_settings):
  479. """Verify settings endpoint can update the disable_weight_sync setting."""
  480. # First verify it's false by default
  481. response = await async_client.get("/api/v1/settings/spoolman")
  482. assert response.status_code == 200
  483. data = response.json()
  484. assert data.get("spoolman_disable_weight_sync", "false") == "false"
  485. # Update the setting
  486. response = await async_client.put(
  487. "/api/v1/settings/spoolman",
  488. json={"spoolman_disable_weight_sync": "true"},
  489. )
  490. assert response.status_code == 200
  491. data = response.json()
  492. assert data["spoolman_disable_weight_sync"] == "true"
  493. # Verify it persisted
  494. response = await async_client.get("/api/v1/settings/spoolman")
  495. assert response.status_code == 200
  496. data = response.json()
  497. assert data["spoolman_disable_weight_sync"] == "true"
  498. @pytest.mark.asyncio
  499. @pytest.mark.integration
  500. async def test_sync_with_weight_sync_disabled_updates_location_only(
  501. self,
  502. async_client: AsyncClient,
  503. spoolman_settings_weight_sync_disabled,
  504. mock_spoolman_client,
  505. printer_factory,
  506. ):
  507. """Verify sync only updates location when disable_weight_sync is enabled."""
  508. printer = await printer_factory()
  509. # Mock existing spool
  510. mock_existing_spool = {
  511. "id": 42,
  512. "remaining_weight": 800,
  513. "extra": {"tag": '"A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4"'},
  514. "filament": {"id": 1, "name": "PLA Red", "material": "PLA"},
  515. }
  516. mock_spoolman_client.find_spool_by_tag = AsyncMock(return_value=mock_existing_spool)
  517. mock_spoolman_client.parse_ams_tray = MagicMock()
  518. # Create mock AMSTray
  519. from backend.app.services.spoolman import AMSTray
  520. mock_tray = AMSTray(
  521. ams_id=0,
  522. tray_id=0,
  523. tray_type="PLA",
  524. tray_sub_brands="PLA Basic",
  525. tray_color="FF0000FF",
  526. remain=50,
  527. tag_uid="",
  528. tray_uuid="A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4",
  529. tray_info_idx="GFA00",
  530. tray_weight=1000,
  531. )
  532. mock_spoolman_client.parse_ams_tray.return_value = mock_tray
  533. mock_spoolman_client.is_bambu_lab_spool = MagicMock(return_value=True)
  534. mock_spoolman_client.convert_ams_slot_to_location = MagicMock(return_value="AMS A1")
  535. mock_spoolman_client.sync_ams_tray = AsyncMock(return_value={"id": 42})
  536. mock_spoolman_client.clear_location_for_removed_spools = AsyncMock(return_value=0)
  537. with patch("backend.app.api.routes.spoolman.printer_manager") as pm_mock:
  538. mock_state = MagicMock()
  539. mock_state.raw_data = {
  540. "ams": [
  541. {
  542. "id": 0,
  543. "tray": [
  544. {
  545. "id": 0,
  546. "tray_type": "PLA",
  547. "tray_sub_brands": "PLA Basic",
  548. "tray_color": "FF0000FF",
  549. "remain": 50,
  550. "tag_uid": "",
  551. "tray_uuid": "A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4",
  552. "tray_info_idx": "GFA00",
  553. "tray_weight": 1000,
  554. }
  555. ],
  556. }
  557. ]
  558. }
  559. pm_mock.get_status = MagicMock(return_value=mock_state)
  560. response = await async_client.post(f"/api/v1/spoolman/sync/{printer.id}")
  561. assert response.status_code == 200
  562. # Verify sync_ams_tray was called with disable_weight_sync=True
  563. mock_spoolman_client.sync_ams_tray.assert_called()
  564. call_kwargs = mock_spoolman_client.sync_ams_tray.call_args.kwargs
  565. assert call_kwargs.get("disable_weight_sync") is True
  566. # =========================================================================
  567. # Report Partial Usage Tests
  568. # =========================================================================
  569. @pytest.fixture
  570. async def spoolman_settings_partial_usage_disabled(self, db_session):
  571. """Create Spoolman settings with partial usage reporting disabled."""
  572. from backend.app.models.settings import Settings
  573. enabled_setting = Settings(key="spoolman_enabled", value="true")
  574. url_setting = Settings(key="spoolman_url", value="http://localhost:7912")
  575. partial_usage_setting = Settings(key="spoolman_report_partial_usage", value="false")
  576. db_session.add(enabled_setting)
  577. db_session.add(url_setting)
  578. db_session.add(partial_usage_setting)
  579. await db_session.commit()
  580. return {
  581. "enabled": enabled_setting,
  582. "url": url_setting,
  583. "partial_usage": partial_usage_setting,
  584. }
  585. @pytest.mark.asyncio
  586. @pytest.mark.integration
  587. async def test_settings_returns_report_partial_usage(
  588. self, async_client: AsyncClient, spoolman_settings_partial_usage_disabled
  589. ):
  590. """Verify settings endpoint returns the report_partial_usage setting."""
  591. response = await async_client.get("/api/v1/settings/spoolman")
  592. assert response.status_code == 200
  593. data = response.json()
  594. assert "spoolman_report_partial_usage" in data
  595. assert data["spoolman_report_partial_usage"] == "false"
  596. @pytest.mark.asyncio
  597. @pytest.mark.integration
  598. async def test_settings_update_report_partial_usage(self, async_client: AsyncClient, spoolman_settings):
  599. """Verify settings endpoint can update the report_partial_usage setting."""
  600. # First verify it's true by default
  601. response = await async_client.get("/api/v1/settings/spoolman")
  602. assert response.status_code == 200
  603. data = response.json()
  604. assert data.get("spoolman_report_partial_usage", "true") == "true"
  605. # Update the setting to false
  606. response = await async_client.put(
  607. "/api/v1/settings/spoolman",
  608. json={"spoolman_report_partial_usage": "false"},
  609. )
  610. assert response.status_code == 200
  611. data = response.json()
  612. assert data["spoolman_report_partial_usage"] == "false"
  613. # Verify it persisted
  614. response = await async_client.get("/api/v1/settings/spoolman")
  615. assert response.status_code == 200
  616. data = response.json()
  617. assert data["spoolman_report_partial_usage"] == "false"
  618. @pytest.mark.asyncio
  619. @pytest.mark.integration
  620. async def test_settings_report_partial_usage_defaults_to_true(self, async_client: AsyncClient, spoolman_settings):
  621. """Verify report_partial_usage defaults to true (unlike disable_weight_sync which defaults to false)."""
  622. response = await async_client.get("/api/v1/settings/spoolman")
  623. assert response.status_code == 200
  624. data = response.json()
  625. # Should default to "true"
  626. assert data["spoolman_report_partial_usage"] == "true"