test_archives_api.py 38 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919
  1. """Integration tests for Archives API endpoints.
  2. Tests the full request/response cycle for /api/v1/archives/ endpoints.
  3. """
  4. import pytest
  5. from httpx import AsyncClient
  6. class TestArchivesAPI:
  7. """Integration tests for /api/v1/archives/ endpoints."""
  8. # ========================================================================
  9. # List endpoints
  10. # ========================================================================
  11. @pytest.mark.asyncio
  12. @pytest.mark.integration
  13. async def test_list_archives_empty(self, async_client: AsyncClient):
  14. """Verify empty list is returned when no archives exist."""
  15. response = await async_client.get("/api/v1/archives/")
  16. assert response.status_code == 200
  17. data = response.json()
  18. assert isinstance(data, list)
  19. assert len(data) == 0
  20. @pytest.mark.asyncio
  21. @pytest.mark.integration
  22. async def test_list_archives_with_data(
  23. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  24. ):
  25. """Verify list returns existing archives."""
  26. printer = await printer_factory()
  27. await archive_factory(printer.id, print_name="Test Archive")
  28. response = await async_client.get("/api/v1/archives/")
  29. assert response.status_code == 200
  30. data = response.json()
  31. assert isinstance(data, list)
  32. assert len(data) >= 1
  33. assert any(a["print_name"] == "Test Archive" for a in data)
  34. @pytest.mark.asyncio
  35. @pytest.mark.integration
  36. async def test_list_archives_pagination(
  37. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  38. ):
  39. """Verify pagination works correctly."""
  40. printer = await printer_factory()
  41. # Create 5 archives
  42. for i in range(5):
  43. await archive_factory(printer.id, print_name=f"Archive {i}")
  44. # Get first page with limit 2
  45. response = await async_client.get("/api/v1/archives/?limit=2&offset=0")
  46. assert response.status_code == 200
  47. data = response.json()
  48. assert isinstance(data, list)
  49. assert len(data) == 2
  50. @pytest.mark.asyncio
  51. @pytest.mark.integration
  52. async def test_list_archives_filter_by_printer(
  53. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  54. ):
  55. """Verify filtering by printer_id works."""
  56. printer1 = await printer_factory(name="Printer 1", serial_number="00M09A000000001")
  57. printer2 = await printer_factory(name="Printer 2", serial_number="00M09A000000002")
  58. await archive_factory(printer1.id, print_name="Printer 1 Archive")
  59. await archive_factory(printer2.id, print_name="Printer 2 Archive")
  60. response = await async_client.get(f"/api/v1/archives/?printer_id={printer1.id}")
  61. assert response.status_code == 200
  62. data = response.json()
  63. assert all(a["printer_id"] == printer1.id for a in data)
  64. # ========================================================================
  65. # Get single endpoint
  66. # ========================================================================
  67. @pytest.mark.asyncio
  68. @pytest.mark.integration
  69. async def test_get_archive(self, async_client: AsyncClient, archive_factory, printer_factory, db_session):
  70. """Verify single archive can be retrieved."""
  71. printer = await printer_factory()
  72. archive = await archive_factory(printer.id, print_name="Get Test Archive")
  73. response = await async_client.get(f"/api/v1/archives/{archive.id}")
  74. assert response.status_code == 200
  75. result = response.json()
  76. assert result["id"] == archive.id
  77. assert result["print_name"] == "Get Test Archive"
  78. @pytest.mark.asyncio
  79. @pytest.mark.integration
  80. async def test_get_archive_not_found(self, async_client: AsyncClient):
  81. """Verify 404 for non-existent archive."""
  82. response = await async_client.get("/api/v1/archives/9999")
  83. assert response.status_code == 404
  84. # ========================================================================
  85. # Update endpoints
  86. # ========================================================================
  87. @pytest.mark.asyncio
  88. @pytest.mark.integration
  89. async def test_update_archive_name(self, async_client: AsyncClient, archive_factory, printer_factory, db_session):
  90. """Verify archive name can be updated."""
  91. printer = await printer_factory()
  92. archive = await archive_factory(printer.id, print_name="Original Name")
  93. response = await async_client.patch(f"/api/v1/archives/{archive.id}", json={"print_name": "Updated Name"})
  94. assert response.status_code == 200
  95. assert response.json()["print_name"] == "Updated Name"
  96. @pytest.mark.asyncio
  97. @pytest.mark.integration
  98. async def test_update_archive_notes(self, async_client: AsyncClient, archive_factory, printer_factory, db_session):
  99. """Verify archive notes can be updated."""
  100. printer = await printer_factory()
  101. archive = await archive_factory(printer.id)
  102. response = await async_client.patch(f"/api/v1/archives/{archive.id}", json={"notes": "Great print!"})
  103. assert response.status_code == 200
  104. assert response.json()["notes"] == "Great print!"
  105. @pytest.mark.asyncio
  106. @pytest.mark.integration
  107. async def test_update_archive_favorite(
  108. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  109. ):
  110. """Verify archive favorite status can be updated."""
  111. printer = await printer_factory()
  112. archive = await archive_factory(printer.id)
  113. response = await async_client.patch(f"/api/v1/archives/{archive.id}", json={"is_favorite": True})
  114. assert response.status_code == 200
  115. assert response.json()["is_favorite"] is True
  116. @pytest.mark.asyncio
  117. @pytest.mark.integration
  118. async def test_update_archive_external_url(
  119. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  120. ):
  121. """Verify archive external_url can be updated."""
  122. printer = await printer_factory()
  123. archive = await archive_factory(printer.id)
  124. response = await async_client.patch(
  125. f"/api/v1/archives/{archive.id}", json={"external_url": "https://printables.com/model/12345"}
  126. )
  127. assert response.status_code == 200
  128. assert response.json()["external_url"] == "https://printables.com/model/12345"
  129. # Verify it can be cleared
  130. response = await async_client.patch(f"/api/v1/archives/{archive.id}", json={"external_url": None})
  131. assert response.status_code == 200
  132. assert response.json()["external_url"] is None
  133. # ========================================================================
  134. # Delete endpoints
  135. # ========================================================================
  136. @pytest.mark.asyncio
  137. @pytest.mark.integration
  138. async def test_delete_archive(self, async_client: AsyncClient, archive_factory, printer_factory, db_session):
  139. """Verify archive can be deleted."""
  140. printer = await printer_factory()
  141. archive = await archive_factory(printer.id)
  142. archive_id = archive.id
  143. response = await async_client.delete(f"/api/v1/archives/{archive_id}")
  144. assert response.status_code == 200
  145. # Verify deleted
  146. response = await async_client.get(f"/api/v1/archives/{archive_id}")
  147. assert response.status_code == 404
  148. @pytest.mark.asyncio
  149. @pytest.mark.integration
  150. async def test_delete_nonexistent_archive(self, async_client: AsyncClient):
  151. """Verify deleting non-existent archive returns 404."""
  152. response = await async_client.delete("/api/v1/archives/9999")
  153. assert response.status_code == 404
  154. @pytest.mark.asyncio
  155. @pytest.mark.integration
  156. async def test_soft_delete_preserves_stats_contribution(
  157. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  158. ):
  159. """#1343: deleting an archive without ``purge_stats`` keeps its
  160. contribution in Quick Stats. The row vanishes from listings but the
  161. filament / time / cost totals stay intact.
  162. """
  163. printer = await printer_factory()
  164. await archive_factory(
  165. printer.id,
  166. status="completed",
  167. print_time_seconds=3600,
  168. filament_used_grams=50.0,
  169. cost=1.50,
  170. )
  171. archive_to_delete = await archive_factory(
  172. printer.id,
  173. status="completed",
  174. print_time_seconds=7200,
  175. filament_used_grams=100.0,
  176. cost=3.00,
  177. )
  178. # Pre-delete: stats include both archives.
  179. pre = (await async_client.get("/api/v1/archives/stats")).json()
  180. assert pre["total_prints"] == 2
  181. assert pre["total_filament_grams"] == 150.0
  182. assert pre["total_cost"] == 4.50
  183. # Soft delete (default — no purge_stats param).
  184. resp = await async_client.delete(f"/api/v1/archives/{archive_to_delete.id}")
  185. assert resp.status_code == 200
  186. body = resp.json()
  187. assert body["purged_from_stats"] is False
  188. # Listing hides the deleted archive…
  189. listing = (await async_client.get("/api/v1/archives/")).json()
  190. assert all(a["id"] != archive_to_delete.id for a in listing)
  191. # …but stats still reflect both prints (the whole point of #1343).
  192. post = (await async_client.get("/api/v1/archives/stats")).json()
  193. assert post["total_prints"] == 2
  194. assert post["total_filament_grams"] == 150.0
  195. assert post["total_cost"] == 4.50
  196. @pytest.mark.asyncio
  197. @pytest.mark.integration
  198. async def test_soft_delete_clears_thumbnail_path_on_linked_log_entries(
  199. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  200. ):
  201. """#1348 follow-up: soft-deleting an archive removes its files from disk;
  202. the cached thumbnail_path on linked PrintLogEntry rows must be NULLed
  203. in the same transaction so the print-log view doesn't 404-storm on the
  204. now-deleted thumbnail file."""
  205. from sqlalchemy import select
  206. from backend.app.models.print_log import PrintLogEntry
  207. printer = await printer_factory()
  208. archive = await archive_factory(
  209. printer.id,
  210. status="completed",
  211. thumbnail_path="archives/test/test_print/thumbnail.png",
  212. )
  213. # The factory's auto-PrintLogEntry doesn't copy thumbnail_path; set it
  214. # manually to mirror what the production write_log_entry path stores.
  215. run_query = await db_session.execute(select(PrintLogEntry).where(PrintLogEntry.archive_id == archive.id))
  216. run = run_query.scalar_one()
  217. run.thumbnail_path = "archives/test/test_print/thumbnail.png"
  218. await db_session.commit()
  219. assert run.thumbnail_path is not None
  220. resp = await async_client.delete(f"/api/v1/archives/{archive.id}")
  221. assert resp.status_code == 200
  222. assert resp.json()["purged_from_stats"] is False
  223. await db_session.refresh(run)
  224. assert run.thumbnail_path is None, "soft-delete must NULL thumbnail_path on linked log entry"
  225. # The log entry itself survives the soft delete (its filament/cost
  226. # contribution still needs to flow into stats per #1343).
  227. assert run.id is not None
  228. assert run.archive_id == archive.id
  229. @pytest.mark.asyncio
  230. @pytest.mark.integration
  231. async def test_hard_delete_clears_thumbnail_path_before_fk_cascade(
  232. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  233. ):
  234. """#1348 follow-up: the auto-purge sweeper (and any caller of
  235. ArchiveService.delete_archive) hard-deletes the archive row but leaves
  236. PrintLogEntry rows alive via ON DELETE SET NULL. The eager
  237. thumbnail_path clear must run inside delete_archive so even orphaned
  238. log entries don't surface stale paths."""
  239. from sqlalchemy import select
  240. from backend.app.models.print_log import PrintLogEntry
  241. from backend.app.services.archive import ArchiveService
  242. printer = await printer_factory()
  243. archive = await archive_factory(
  244. printer.id,
  245. status="completed",
  246. thumbnail_path="archives/test/test_print/thumbnail.png",
  247. )
  248. run_query = await db_session.execute(select(PrintLogEntry).where(PrintLogEntry.archive_id == archive.id))
  249. run = run_query.scalar_one()
  250. run.thumbnail_path = "archives/test/test_print/thumbnail.png"
  251. await db_session.commit()
  252. run_id = run.id
  253. service = ArchiveService(db_session)
  254. assert await service.delete_archive(archive.id) is True
  255. # Log entry survives the hard-delete (the FK is ON DELETE SET NULL
  256. # in production; SQLite test config doesn't enable foreign_keys=ON
  257. # by default so archive_id may still be set, but the row itself
  258. # remains for audit). The thumbnail_path was cleared eagerly by
  259. # _null_print_log_thumbnail_paths before db.delete(archive).
  260. refetch = await db_session.execute(select(PrintLogEntry).where(PrintLogEntry.id == run_id))
  261. survivor = refetch.scalar_one()
  262. assert survivor.thumbnail_path is None, (
  263. "delete_archive must NULL thumbnail_path before removing the archive row"
  264. )
  265. @pytest.mark.asyncio
  266. @pytest.mark.integration
  267. async def test_print_log_thumbnail_route_lazy_nulls_missing_file(
  268. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  269. ):
  270. """#1348 follow-up: GET /print-log/{id}/thumbnail self-heals when the
  271. thumbnail_path on a log entry points at a missing file (failed print
  272. whose thumbnail was never written, or a stale path that escaped the
  273. delete-time cleanup)."""
  274. from sqlalchemy import select
  275. from backend.app.models.print_log import PrintLogEntry
  276. printer = await printer_factory()
  277. archive = await archive_factory(printer.id, status="failed")
  278. run_query = await db_session.execute(select(PrintLogEntry).where(PrintLogEntry.archive_id == archive.id))
  279. run = run_query.scalar_one()
  280. # Path points at a file that never existed (failed-print case where
  281. # archive.thumbnail_path was set but the extractor never produced one).
  282. run.thumbnail_path = "archives/missing/never_written/thumbnail.png"
  283. await db_session.commit()
  284. # Auth is disabled in the integration test config, so the stream-token
  285. # guard is bypassed — the route runs the lazy-NULL branch directly.
  286. resp = await async_client.get(f"/api/v1/print-log/{run.id}/thumbnail")
  287. assert resp.status_code == 404
  288. await db_session.refresh(run)
  289. assert run.thumbnail_path is None, "missing file must self-heal to NULL"
  290. @pytest.mark.asyncio
  291. @pytest.mark.integration
  292. async def test_purge_stats_drops_archive_from_quick_stats(
  293. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  294. ):
  295. """#1343: deleting with ``?purge_stats=true`` hard-deletes the row,
  296. dropping its contribution from Quick Stats (the original behaviour,
  297. now opt-in)."""
  298. printer = await printer_factory()
  299. keep = await archive_factory(printer.id, status="completed", filament_used_grams=50.0)
  300. purge = await archive_factory(printer.id, status="completed", filament_used_grams=100.0)
  301. resp = await async_client.delete(f"/api/v1/archives/{purge.id}?purge_stats=true")
  302. assert resp.status_code == 200
  303. assert resp.json()["purged_from_stats"] is True
  304. stats = (await async_client.get("/api/v1/archives/stats")).json()
  305. assert stats["total_prints"] == 1
  306. assert stats["total_filament_grams"] == 50.0
  307. # The kept archive is still listed.
  308. listing = (await async_client.get("/api/v1/archives/")).json()
  309. assert [a["id"] for a in listing] == [keep.id]
  310. @pytest.mark.asyncio
  311. @pytest.mark.integration
  312. async def test_soft_deleted_archive_404_on_detail(
  313. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  314. ):
  315. """A soft-deleted archive must 404 on GET — a stale bookmark or
  316. direct URL should not expose a row the user has already removed."""
  317. printer = await printer_factory()
  318. archive = await archive_factory(printer.id)
  319. await async_client.delete(f"/api/v1/archives/{archive.id}")
  320. resp = await async_client.get(f"/api/v1/archives/{archive.id}")
  321. assert resp.status_code == 404
  322. @pytest.mark.asyncio
  323. @pytest.mark.integration
  324. async def test_soft_deleted_archive_hidden_from_search(
  325. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  326. ):
  327. """Search must skip soft-deleted archives. Uses the LIKE fallback by
  328. querying a single-character pattern that the SQLite FTS5 rejects, so
  329. the test covers the fallback path that the production FTS path also
  330. respects."""
  331. printer = await printer_factory()
  332. archive = await archive_factory(printer.id, print_name="UniqueSoftDeleteCandidate")
  333. await async_client.delete(f"/api/v1/archives/{archive.id}")
  334. resp = await async_client.get("/api/v1/archives/search?q=UniqueSoftDeleteCandidate")
  335. assert resp.status_code == 200
  336. assert resp.json() == []
  337. # ========================================================================
  338. # Statistics endpoints
  339. # ========================================================================
  340. @pytest.mark.asyncio
  341. @pytest.mark.integration
  342. async def test_get_archive_stats(self, async_client: AsyncClient, archive_factory, printer_factory, db_session):
  343. """Verify archive statistics can be retrieved."""
  344. printer = await printer_factory()
  345. await archive_factory(
  346. printer.id,
  347. status="completed",
  348. print_time_seconds=3600,
  349. filament_used_grams=50.0,
  350. )
  351. await archive_factory(
  352. printer.id,
  353. status="completed",
  354. print_time_seconds=7200,
  355. filament_used_grams=100.0,
  356. )
  357. response = await async_client.get("/api/v1/archives/stats")
  358. assert response.status_code == 200
  359. result = response.json()
  360. # Check for actual stats fields
  361. assert "total_prints" in result
  362. assert "successful_prints" in result
  363. class TestArchivesSlimAPI:
  364. """Integration tests for /api/v1/archives/slim endpoint."""
  365. @pytest.mark.asyncio
  366. @pytest.mark.integration
  367. async def test_slim_empty(self, async_client: AsyncClient):
  368. """Verify empty list when no archives exist."""
  369. response = await async_client.get("/api/v1/archives/slim")
  370. assert response.status_code == 200
  371. assert response.json() == []
  372. @pytest.mark.asyncio
  373. @pytest.mark.integration
  374. async def test_slim_returns_only_expected_fields(
  375. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  376. ):
  377. """Verify response contains only slim fields, not full archive data."""
  378. printer = await printer_factory()
  379. await archive_factory(
  380. printer.id,
  381. print_name="Slim Test",
  382. status="completed",
  383. filament_type="PLA",
  384. filament_color="#FF0000",
  385. filament_used_grams=50.0,
  386. print_time_seconds=3600,
  387. cost=1.50,
  388. quantity=2,
  389. )
  390. response = await async_client.get("/api/v1/archives/slim")
  391. assert response.status_code == 200
  392. data = response.json()
  393. assert len(data) == 1
  394. item = data[0]
  395. # Expected fields present
  396. assert item["printer_id"] == printer.id
  397. assert item["print_name"] == "Slim Test"
  398. assert item["status"] == "completed"
  399. assert item["filament_type"] == "PLA"
  400. assert item["filament_color"] == "#FF0000"
  401. assert item["filament_used_grams"] == 50.0
  402. assert item["print_time_seconds"] == 3600
  403. assert item["cost"] == 1.50
  404. assert item["quantity"] == 2
  405. assert "created_at" in item
  406. # Full archive fields must NOT be present
  407. assert "id" not in item
  408. assert "filename" not in item
  409. assert "file_path" not in item
  410. assert "file_size" not in item
  411. assert "extra_data" not in item
  412. assert "notes" not in item
  413. assert "tags" not in item
  414. assert "photos" not in item
  415. assert "thumbnail_path" not in item
  416. assert "content_hash" not in item
  417. assert "duplicates" not in item
  418. assert "duplicate_count" not in item
  419. @pytest.mark.asyncio
  420. @pytest.mark.integration
  421. async def test_slim_computes_actual_time(
  422. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  423. ):
  424. """Verify actual_time_seconds is computed from started_at/completed_at."""
  425. from datetime import datetime, timezone
  426. printer = await printer_factory()
  427. started = datetime(2024, 1, 1, 10, 0, 0, tzinfo=timezone.utc)
  428. completed = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) # 2 hours = 7200s
  429. await archive_factory(
  430. printer.id,
  431. status="completed",
  432. started_at=started,
  433. completed_at=completed,
  434. )
  435. response = await async_client.get("/api/v1/archives/slim")
  436. assert response.status_code == 200
  437. item = response.json()[0]
  438. assert item["actual_time_seconds"] == 7200
  439. @pytest.mark.asyncio
  440. @pytest.mark.integration
  441. async def test_slim_actual_time_null_for_failed(
  442. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  443. ):
  444. """Verify actual_time_seconds is null for non-completed prints."""
  445. from datetime import datetime, timezone
  446. printer = await printer_factory()
  447. await archive_factory(
  448. printer.id,
  449. status="failed",
  450. started_at=datetime(2024, 1, 1, 10, 0, 0, tzinfo=timezone.utc),
  451. completed_at=datetime(2024, 1, 1, 11, 0, 0, tzinfo=timezone.utc),
  452. )
  453. response = await async_client.get("/api/v1/archives/slim")
  454. assert response.status_code == 200
  455. item = response.json()[0]
  456. assert item["actual_time_seconds"] is None
  457. @pytest.mark.asyncio
  458. @pytest.mark.integration
  459. async def test_slim_date_filtering(self, async_client: AsyncClient, archive_factory, printer_factory, db_session):
  460. """Verify date_from and date_to filters work."""
  461. from datetime import datetime, timezone
  462. printer = await printer_factory()
  463. await archive_factory(
  464. printer.id,
  465. print_name="Old Print",
  466. created_at=datetime(2024, 1, 1, tzinfo=timezone.utc),
  467. )
  468. await archive_factory(
  469. printer.id,
  470. print_name="New Print",
  471. created_at=datetime(2024, 6, 15, tzinfo=timezone.utc),
  472. )
  473. # Filter to only June 2024
  474. response = await async_client.get("/api/v1/archives/slim?date_from=2024-06-01&date_to=2024-06-30")
  475. assert response.status_code == 200
  476. data = response.json()
  477. assert len(data) == 1
  478. assert data[0]["print_name"] == "New Print"
  479. @pytest.mark.asyncio
  480. @pytest.mark.integration
  481. async def test_slim_pagination(self, async_client: AsyncClient, archive_factory, printer_factory, db_session):
  482. """Verify limit and offset work."""
  483. printer = await printer_factory()
  484. for i in range(5):
  485. await archive_factory(printer.id, print_name=f"Print {i}")
  486. response = await async_client.get("/api/v1/archives/slim?limit=2&offset=0")
  487. assert response.status_code == 200
  488. assert len(response.json()) == 2
  489. class TestArchiveDataIntegrity:
  490. """Tests for archive data integrity."""
  491. @pytest.mark.asyncio
  492. @pytest.mark.integration
  493. async def test_archive_linked_to_printer(
  494. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  495. ):
  496. """Verify archive is properly linked to printer."""
  497. printer = await printer_factory(name="My Printer")
  498. archive = await archive_factory(printer.id)
  499. response = await async_client.get(f"/api/v1/archives/{archive.id}")
  500. assert response.status_code == 200
  501. result = response.json()
  502. assert result["printer_id"] == printer.id
  503. @pytest.mark.asyncio
  504. @pytest.mark.integration
  505. async def test_archive_stores_print_data(
  506. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  507. ):
  508. """Verify archive stores all print data correctly."""
  509. printer = await printer_factory()
  510. archive = await archive_factory(
  511. printer.id,
  512. print_name="Test Print",
  513. filename="test.3mf",
  514. status="completed",
  515. filament_type="PLA",
  516. filament_used_grams=75.5,
  517. print_time_seconds=5400,
  518. )
  519. response = await async_client.get(f"/api/v1/archives/{archive.id}")
  520. assert response.status_code == 200
  521. result = response.json()
  522. assert result["print_name"] == "Test Print"
  523. assert result["filename"] == "test.3mf"
  524. assert result["status"] == "completed"
  525. assert result["filament_type"] == "PLA"
  526. assert result["filament_used_grams"] == 75.5
  527. assert result["print_time_seconds"] == 5400
  528. @pytest.mark.asyncio
  529. @pytest.mark.integration
  530. async def test_archive_update_persists(
  531. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  532. ):
  533. """CRITICAL: Verify archive updates persist."""
  534. printer = await printer_factory()
  535. archive = await archive_factory(printer.id, notes="Original notes")
  536. # Update
  537. await async_client.patch(f"/api/v1/archives/{archive.id}", json={"notes": "Updated notes", "is_favorite": True})
  538. # Verify persistence
  539. response = await async_client.get(f"/api/v1/archives/{archive.id}")
  540. result = response.json()
  541. assert result["notes"] == "Updated notes"
  542. assert result["is_favorite"] is True
  543. class TestArchiveF3DEndpoints:
  544. """Tests for F3D (Fusion 360 design file) attachment endpoints."""
  545. @pytest.mark.asyncio
  546. @pytest.mark.integration
  547. async def test_archive_response_includes_f3d_path(
  548. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  549. ):
  550. """Verify f3d_path is included in archive response."""
  551. printer = await printer_factory()
  552. archive = await archive_factory(printer.id, f3d_path="archives/test/design.f3d")
  553. response = await async_client.get(f"/api/v1/archives/{archive.id}")
  554. assert response.status_code == 200
  555. result = response.json()
  556. assert "f3d_path" in result
  557. assert result["f3d_path"] == "archives/test/design.f3d"
  558. @pytest.mark.asyncio
  559. @pytest.mark.integration
  560. async def test_archive_response_f3d_path_null_when_not_set(
  561. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  562. ):
  563. """Verify f3d_path is null when no F3D file attached."""
  564. printer = await printer_factory()
  565. archive = await archive_factory(printer.id)
  566. response = await async_client.get(f"/api/v1/archives/{archive.id}")
  567. assert response.status_code == 200
  568. result = response.json()
  569. assert "f3d_path" in result
  570. assert result["f3d_path"] is None
  571. @pytest.mark.asyncio
  572. @pytest.mark.integration
  573. async def test_upload_f3d_to_nonexistent_archive(self, async_client: AsyncClient):
  574. """Verify 404 when uploading F3D to non-existent archive."""
  575. # Create a minimal file-like upload
  576. files = {"file": ("design.f3d", b"fake f3d content", "application/octet-stream")}
  577. response = await async_client.post("/api/v1/archives/9999/f3d", files=files)
  578. assert response.status_code == 404
  579. @pytest.mark.asyncio
  580. @pytest.mark.integration
  581. async def test_download_f3d_not_found_when_no_file(
  582. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  583. ):
  584. """Verify 404 when downloading F3D from archive without F3D file."""
  585. printer = await printer_factory()
  586. archive = await archive_factory(printer.id)
  587. response = await async_client.get(f"/api/v1/archives/{archive.id}/f3d")
  588. assert response.status_code == 404
  589. @pytest.mark.asyncio
  590. @pytest.mark.integration
  591. async def test_download_f3d_nonexistent_archive(self, async_client: AsyncClient):
  592. """Verify 404 when downloading F3D from non-existent archive."""
  593. response = await async_client.get("/api/v1/archives/9999/f3d")
  594. assert response.status_code == 404
  595. @pytest.mark.asyncio
  596. @pytest.mark.integration
  597. async def test_delete_f3d_nonexistent_archive(self, async_client: AsyncClient):
  598. """Verify 404 when deleting F3D from non-existent archive."""
  599. response = await async_client.delete("/api/v1/archives/9999/f3d")
  600. assert response.status_code == 404
  601. @pytest.mark.asyncio
  602. @pytest.mark.integration
  603. async def test_delete_f3d_when_no_file(
  604. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  605. ):
  606. """Verify 404 when deleting F3D from archive without F3D file."""
  607. printer = await printer_factory()
  608. archive = await archive_factory(printer.id)
  609. response = await async_client.delete(f"/api/v1/archives/{archive.id}/f3d")
  610. assert response.status_code == 404
  611. @pytest.mark.asyncio
  612. @pytest.mark.integration
  613. async def test_list_archives_includes_f3d_path(
  614. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  615. ):
  616. """Verify f3d_path is included in archive list responses."""
  617. printer = await printer_factory()
  618. await archive_factory(printer.id, print_name="With F3D", f3d_path="archives/test/design.f3d")
  619. await archive_factory(printer.id, print_name="Without F3D")
  620. response = await async_client.get("/api/v1/archives/")
  621. assert response.status_code == 200
  622. data = response.json()
  623. assert len(data) >= 2
  624. with_f3d = next((a for a in data if a["print_name"] == "With F3D"), None)
  625. without_f3d = next((a for a in data if a["print_name"] == "Without F3D"), None)
  626. assert with_f3d is not None
  627. assert with_f3d["f3d_path"] == "archives/test/design.f3d"
  628. assert without_f3d is not None
  629. assert without_f3d["f3d_path"] is None
  630. # ========================================================================
  631. # Multi-Plate 3MF endpoints (Issue #93)
  632. # ========================================================================
  633. @pytest.mark.asyncio
  634. @pytest.mark.integration
  635. async def test_get_archive_plates_not_found(self, async_client: AsyncClient):
  636. """Verify 404 when fetching plates for non-existent archive."""
  637. response = await async_client.get("/api/v1/archives/999999/plates")
  638. assert response.status_code == 404
  639. @pytest.mark.asyncio
  640. @pytest.mark.integration
  641. async def test_get_plate_thumbnail_not_found(self, async_client: AsyncClient):
  642. """Verify 404 when fetching plate thumbnail for non-existent archive."""
  643. response = await async_client.get("/api/v1/archives/999999/plate-thumbnail/1")
  644. assert response.status_code == 404
  645. @pytest.mark.asyncio
  646. @pytest.mark.integration
  647. async def test_filament_requirements_not_found(self, async_client: AsyncClient):
  648. """Verify filament-requirements returns 404 for non-existent archive."""
  649. response = await async_client.get("/api/v1/archives/999999/filament-requirements")
  650. assert response.status_code == 404
  651. @pytest.mark.asyncio
  652. @pytest.mark.integration
  653. async def test_filament_requirements_with_plate_id_not_found(self, async_client: AsyncClient):
  654. """Verify filament-requirements with plate_id returns 404 for non-existent archive."""
  655. response = await async_client.get("/api/v1/archives/999999/filament-requirements?plate_id=1")
  656. assert response.status_code == 404
  657. # ========================================================================
  658. # Tag Management endpoints (Issue #183)
  659. # ========================================================================
  660. @pytest.mark.asyncio
  661. @pytest.mark.integration
  662. async def test_get_tags_empty(self, async_client: AsyncClient):
  663. """Verify empty list when no tags exist."""
  664. response = await async_client.get("/api/v1/archives/tags")
  665. assert response.status_code == 200
  666. data = response.json()
  667. assert isinstance(data, list)
  668. assert len(data) == 0
  669. @pytest.mark.asyncio
  670. @pytest.mark.integration
  671. async def test_get_tags_with_data(self, async_client: AsyncClient, archive_factory, printer_factory, db_session):
  672. """Verify tags are returned with counts."""
  673. printer = await printer_factory()
  674. await archive_factory(printer.id, print_name="Archive 1", tags="functional, test")
  675. await archive_factory(printer.id, print_name="Archive 2", tags="functional, calibration")
  676. await archive_factory(printer.id, print_name="Archive 3", tags="test")
  677. response = await async_client.get("/api/v1/archives/tags")
  678. assert response.status_code == 200
  679. data = response.json()
  680. assert isinstance(data, list)
  681. # Convert to dict for easier lookup
  682. tags_dict = {t["name"]: t["count"] for t in data}
  683. assert tags_dict.get("functional") == 2
  684. assert tags_dict.get("test") == 2
  685. assert tags_dict.get("calibration") == 1
  686. @pytest.mark.asyncio
  687. @pytest.mark.integration
  688. async def test_get_tags_sorted_by_count(
  689. self, async_client: AsyncClient, archive_factory, printer_factory, db_session
  690. ):
  691. """Verify tags are sorted by count descending, then by name."""
  692. printer = await printer_factory()
  693. await archive_factory(printer.id, tags="alpha")
  694. await archive_factory(printer.id, tags="beta, alpha")
  695. await archive_factory(printer.id, tags="gamma, beta, alpha")
  696. response = await async_client.get("/api/v1/archives/tags")
  697. assert response.status_code == 200
  698. data = response.json()
  699. # alpha=3, beta=2, gamma=1
  700. assert data[0]["name"] == "alpha"
  701. assert data[0]["count"] == 3
  702. assert data[1]["name"] == "beta"
  703. assert data[1]["count"] == 2
  704. assert data[2]["name"] == "gamma"
  705. assert data[2]["count"] == 1
  706. @pytest.mark.asyncio
  707. @pytest.mark.integration
  708. async def test_rename_tag(self, async_client: AsyncClient, archive_factory, printer_factory, db_session):
  709. """Verify renaming a tag updates all archives."""
  710. printer = await printer_factory()
  711. a1 = await archive_factory(printer.id, print_name="Archive 1", tags="old-tag, other")
  712. a2 = await archive_factory(printer.id, print_name="Archive 2", tags="old-tag")
  713. await archive_factory(printer.id, print_name="Archive 3", tags="different")
  714. response = await async_client.put("/api/v1/archives/tags/old-tag", json={"new_name": "new-tag"})
  715. assert response.status_code == 200
  716. data = response.json()
  717. assert data["affected"] == 2
  718. # Verify the archives were updated
  719. response = await async_client.get(f"/api/v1/archives/{a1.id}")
  720. assert "new-tag" in response.json()["tags"]
  721. assert "old-tag" not in response.json()["tags"]
  722. response = await async_client.get(f"/api/v1/archives/{a2.id}")
  723. assert response.json()["tags"] == "new-tag"
  724. @pytest.mark.asyncio
  725. @pytest.mark.integration
  726. async def test_rename_tag_no_change(self, async_client: AsyncClient):
  727. """Verify renaming to same name returns 0 affected."""
  728. response = await async_client.put("/api/v1/archives/tags/some-tag", json={"new_name": "some-tag"})
  729. assert response.status_code == 200
  730. assert response.json()["affected"] == 0
  731. @pytest.mark.asyncio
  732. @pytest.mark.integration
  733. async def test_rename_tag_empty_name_error(self, async_client: AsyncClient):
  734. """Verify renaming to empty name returns error."""
  735. response = await async_client.put("/api/v1/archives/tags/some-tag", json={"new_name": ""})
  736. assert response.status_code == 400
  737. @pytest.mark.asyncio
  738. @pytest.mark.integration
  739. async def test_delete_tag(self, async_client: AsyncClient, archive_factory, printer_factory, db_session):
  740. """Verify deleting a tag removes it from all archives."""
  741. printer = await printer_factory()
  742. a1 = await archive_factory(printer.id, print_name="Archive 1", tags="delete-me, keep")
  743. a2 = await archive_factory(printer.id, print_name="Archive 2", tags="delete-me")
  744. await archive_factory(printer.id, print_name="Archive 3", tags="different")
  745. response = await async_client.delete("/api/v1/archives/tags/delete-me")
  746. assert response.status_code == 200
  747. data = response.json()
  748. assert data["affected"] == 2
  749. # Verify the archives were updated
  750. response = await async_client.get(f"/api/v1/archives/{a1.id}")
  751. assert response.json()["tags"] == "keep"
  752. response = await async_client.get(f"/api/v1/archives/{a2.id}")
  753. # Should be None or empty when last tag is removed
  754. assert response.json()["tags"] is None or response.json()["tags"] == ""
  755. @pytest.mark.asyncio
  756. @pytest.mark.integration
  757. async def test_delete_tag_not_found(self, async_client: AsyncClient):
  758. """Verify deleting non-existent tag returns 0 affected."""
  759. response = await async_client.delete("/api/v1/archives/tags/nonexistent-tag")
  760. assert response.status_code == 200
  761. assert response.json()["affected"] == 0