"""Integration tests for Projects API endpoints.""" import pytest from httpx import AsyncClient class TestProjectsAPI: """Integration tests for /api/v1/projects endpoints.""" @pytest.fixture async def project_factory(self, db_session): """Factory to create test projects.""" _counter = [0] async def _create_project(**kwargs): from backend.app.models.project import Project _counter[0] += 1 counter = _counter[0] defaults = { "name": f"Test Project {counter}", "description": "Test project description", "color": "#FF0000", } defaults.update(kwargs) project = Project(**defaults) db_session.add(project) await db_session.commit() await db_session.refresh(project) return project return _create_project @pytest.mark.asyncio @pytest.mark.integration async def test_list_projects_empty(self, async_client: AsyncClient): """Verify empty list when no projects exist.""" response = await async_client.get("/api/v1/projects/") assert response.status_code == 200 assert isinstance(response.json(), list) @pytest.mark.asyncio @pytest.mark.integration async def test_list_projects_with_data(self, async_client: AsyncClient, project_factory, db_session): """Verify list returns existing projects.""" await project_factory(name="My Project") response = await async_client.get("/api/v1/projects/") assert response.status_code == 200 data = response.json() assert any(p["name"] == "My Project" for p in data) @pytest.mark.asyncio @pytest.mark.integration async def test_create_project(self, async_client: AsyncClient): """Verify project can be created.""" data = { "name": "New Project", "description": "A new project", "color": "#00FF00", } response = await async_client.post("/api/v1/projects/", json=data) assert response.status_code == 200 result = response.json() assert result["name"] == "New Project" assert result["color"] == "#00FF00" @pytest.mark.asyncio @pytest.mark.integration async def test_get_project(self, async_client: AsyncClient, project_factory, db_session): """Verify single project can be retrieved.""" project = await project_factory(name="Get Test Project") response = await async_client.get(f"/api/v1/projects/{project.id}") assert response.status_code == 200 assert response.json()["name"] == "Get Test Project" @pytest.mark.asyncio @pytest.mark.integration async def test_get_project_not_found(self, async_client: AsyncClient): """Verify 404 for non-existent project.""" response = await async_client.get("/api/v1/projects/9999") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_update_project(self, async_client: AsyncClient, project_factory, db_session): """Verify project can be updated.""" project = await project_factory(name="Original") response = await async_client.patch( f"/api/v1/projects/{project.id}", json={"name": "Updated", "description": "Updated description"} ) assert response.status_code == 200 result = response.json() assert result["name"] == "Updated" assert result["description"] == "Updated description" @pytest.mark.asyncio @pytest.mark.integration async def test_delete_project(self, async_client: AsyncClient, project_factory, db_session): """Verify project can be deleted.""" project = await project_factory() response = await async_client.delete(f"/api/v1/projects/{project.id}") assert response.status_code == 200 data = response.json() assert data["message"] == "Project deleted" @pytest.mark.asyncio @pytest.mark.integration async def test_delete_project_not_found(self, async_client: AsyncClient): """Verify 404 for deleting non-existent project.""" response = await async_client.delete("/api/v1/projects/9999") assert response.status_code == 404 class TestProjectUrlAndCoverImage: """Tests for #1155 — url field + cover image upload/get/delete.""" @pytest.fixture async def project_factory(self, db_session): async def _create(**kwargs): from backend.app.models.project import Project defaults = {"name": "URL/Cover Project", "color": "#00ff00"} defaults.update(kwargs) project = Project(**defaults) db_session.add(project) await db_session.commit() await db_session.refresh(project) return project return _create @pytest.mark.asyncio @pytest.mark.integration async def test_create_project_accepts_https_url(self, async_client: AsyncClient): response = await async_client.post( "/api/v1/projects/", json={"name": "With URL", "url": "https://makerworld.com/models/12345"}, ) assert response.status_code == 200 body = response.json() assert body["url"] == "https://makerworld.com/models/12345" @pytest.mark.asyncio @pytest.mark.integration async def test_create_project_rejects_javascript_url(self, async_client: AsyncClient): # `` rendering would execute javascript: URLs — schema must reject. response = await async_client.post( "/api/v1/projects/", json={"name": "Hostile", "url": "javascript:alert(1)"}, ) assert response.status_code == 422 @pytest.mark.asyncio @pytest.mark.integration async def test_create_project_rejects_data_url(self, async_client: AsyncClient): response = await async_client.post( "/api/v1/projects/", json={"name": "Hostile", "url": "data:text/html,"}, ) assert response.status_code == 422 @pytest.mark.asyncio @pytest.mark.integration async def test_patch_project_clears_url_when_explicitly_null(self, async_client: AsyncClient, project_factory): project = await project_factory(url="https://example.com") response = await async_client.patch(f"/api/v1/projects/{project.id}", json={"url": None}) assert response.status_code == 200 assert response.json()["url"] is None @pytest.mark.asyncio @pytest.mark.integration async def test_upload_cover_image_then_serve_then_delete(self, async_client: AsyncClient, project_factory): project = await project_factory() # 1x1 PNG (smallest valid PNG bytes) png_bytes = bytes.fromhex( "89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c4" "890000000d49444154789c63f80f00000100010000000000000049454e44ae42" "6082" ) upload = await async_client.post( f"/api/v1/projects/{project.id}/cover-image", files={"file": ("cover.png", png_bytes, "image/png")}, ) assert upload.status_code == 200, upload.text body = upload.json() assert body["status"] == "success" assert body["filename"].endswith(".png") cover_filename = body["filename"] # GET should serve the bytes back served = await async_client.get(f"/api/v1/projects/{project.id}/cover-image") assert served.status_code == 200 assert served.headers["content-type"] == "image/png" assert served.content == png_bytes # Project response should reflect the cover_image_filename field view = await async_client.get(f"/api/v1/projects/{project.id}") assert view.json()["cover_image_filename"] == cover_filename # DELETE should clear the field deleted = await async_client.delete(f"/api/v1/projects/{project.id}/cover-image") assert deleted.status_code == 200 view2 = await async_client.get(f"/api/v1/projects/{project.id}") assert view2.json()["cover_image_filename"] is None # And subsequent GET should 404 served2 = await async_client.get(f"/api/v1/projects/{project.id}/cover-image") assert served2.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_upload_cover_image_rejects_non_image(self, async_client: AsyncClient, project_factory): project = await project_factory() response = await async_client.post( f"/api/v1/projects/{project.id}/cover-image", files={"file": ("evil.exe", b"MZ\x00\x00", "application/octet-stream")}, ) assert response.status_code == 400 @pytest.mark.integration def test_cover_image_get_uses_stream_token_gate(self): """Regression guard: GET /projects/{id}/cover-image MUST be gated by ``RequireCameraStreamTokenIfAuthEnabled`` (accepts ``?token=…`` query string) rather than by the bearer-token gate, because browsers can't attach an ``Authorization`` header to ```` requests. Swapping back to the bearer gate would silently 401 every cover image when auth is enabled.""" from fastapi.routing import APIRoute from backend.app.api.routes.projects import router # Find the GET cover-image route. The router exposes path/methods/ # dependencies via APIRoute objects. cover_get = None for route in router.routes: if isinstance(route, APIRoute) and route.path.endswith("/cover-image") and "GET" in route.methods: cover_get = route break assert cover_get is not None, "GET cover-image route missing" # The route's dependant tree includes a Depends(require_camera_stream_token_if_auth_enabled()) # — its `call` is the inner check function returned by that factory. # Walk the dependant tree and assert one of the dependencies came from # the stream-token factory, NOT from require_permission_if_auth_enabled. from backend.app.core.auth import ( require_camera_stream_token_if_auth_enabled, ) # The factory returns a fresh closure each call; the most reliable # signature is the qualified name of the function in the closure chain. expected_qualname = require_camera_stream_token_if_auth_enabled().__qualname__ gate_qualnames = [dep.call.__qualname__ for dep in cover_get.dependant.dependencies if dep.call] assert expected_qualname in gate_qualnames, ( f"GET cover-image route is not gated by RequireCameraStreamTokenIfAuthEnabled. Found: {gate_qualnames}" ) class TestProjectPartsTracking: """Tests for project parts tracking feature.""" @pytest.fixture async def project_factory(self, db_session): """Factory to create test projects.""" async def _create_project(**kwargs): from backend.app.models.project import Project defaults = { "name": "Parts Test Project", "description": "Test project", "color": "#FF0000", } defaults.update(kwargs) project = Project(**defaults) db_session.add(project) await db_session.commit() await db_session.refresh(project) return project return _create_project @pytest.fixture async def archive_factory(self, db_session): """Factory to create a test archive plus a matching PrintLogEntry. Project stats aggregate from ``print_log_entries`` (#1593), so a test that only writes archives wouldn't exercise the production path — production always writes one log entry per run. The factory mirrors that: every archive whose status is anything other than ``"archived"`` (file shelved without printing) gets a log entry whose status matches the archive. """ async def _create_archive(**kwargs): from backend.app.models.archive import PrintArchive from backend.app.models.print_log import PrintLogEntry defaults = { "filename": "test.3mf", "file_path": "test/test.3mf", "file_size": 1000, "print_name": "Test Print", "status": "completed", "quantity": 1, } defaults.update(kwargs) archive = PrintArchive(**defaults) db_session.add(archive) await db_session.commit() await db_session.refresh(archive) if archive.status != "archived": db_session.add( PrintLogEntry( archive_id=archive.id, print_name=archive.print_name, status=archive.status, ) ) await db_session.commit() return archive return _create_archive @pytest.mark.asyncio @pytest.mark.integration async def test_create_project_with_target_parts_count(self, async_client: AsyncClient): """Verify project can be created with target_parts_count.""" data = { "name": "Parts Project", "target_count": 10, # 10 plates "target_parts_count": 50, # 50 parts total } response = await async_client.post("/api/v1/projects/", json=data) assert response.status_code == 200 result = response.json() assert result["target_count"] == 10 assert result["target_parts_count"] == 50 @pytest.mark.asyncio @pytest.mark.integration async def test_update_project_target_parts_count(self, async_client: AsyncClient, project_factory, db_session): """Verify target_parts_count can be updated.""" project = await project_factory() response = await async_client.patch( f"/api/v1/projects/{project.id}", json={"target_parts_count": 100}, ) assert response.status_code == 200 assert response.json()["target_parts_count"] == 100 @pytest.mark.asyncio @pytest.mark.integration async def test_project_parts_progress_calculation( self, async_client: AsyncClient, project_factory, archive_factory, db_session ): """Verify parts progress is calculated from archive quantities.""" # Create project with target of 20 parts project = await project_factory(target_parts_count=20) # Create archives with different quantities await archive_factory(project_id=project.id, quantity=3, status="completed") # 3 parts await archive_factory(project_id=project.id, quantity=5, status="completed") # 5 parts await archive_factory(project_id=project.id, quantity=2, status="completed") # 2 parts # Total: 10 parts completed out of 20 = 50% response = await async_client.get(f"/api/v1/projects/{project.id}") assert response.status_code == 200 data = response.json() # Check stats assert data["stats"]["completed_prints"] == 10 # Sum of quantities assert data["stats"]["parts_progress_percent"] == 50.0 # 10/20 = 50% assert data["stats"]["remaining_parts"] == 10 # 20 - 10 = 10 @pytest.mark.asyncio @pytest.mark.integration async def test_project_list_shows_parts_count( self, async_client: AsyncClient, project_factory, archive_factory, db_session ): """Verify project list returns correct completed_count (parts sum).""" project = await project_factory(name="List Parts Project", target_parts_count=100) # Create archives with quantities await archive_factory(project_id=project.id, quantity=4, status="completed") await archive_factory(project_id=project.id, quantity=6, status="completed") # Total: 10 parts, 2 plates response = await async_client.get("/api/v1/projects/") assert response.status_code == 200 data = response.json() # Find our project our_project = next((p for p in data if p["name"] == "List Parts Project"), None) assert our_project is not None assert our_project["archive_count"] == 2 # 2 plates assert our_project["completed_count"] == 10 # 10 parts (sum of quantities) assert our_project["target_parts_count"] == 100 @pytest.mark.asyncio @pytest.mark.integration async def test_plates_vs_parts_progress( self, async_client: AsyncClient, project_factory, archive_factory, db_session ): """Verify plates and parts progress are calculated separately.""" # Project needs 5 plates producing 25 parts total (5 parts per plate) project = await project_factory(target_count=5, target_parts_count=25) # Complete 2 plates, each with 5 parts await archive_factory(project_id=project.id, quantity=5, status="completed") await archive_factory(project_id=project.id, quantity=5, status="completed") # Plates: 2/5 = 40%, Parts: 10/25 = 40% response = await async_client.get(f"/api/v1/projects/{project.id}") assert response.status_code == 200 data = response.json() assert data["stats"]["total_archives"] == 2 # 2 plates assert data["stats"]["completed_prints"] == 10 # 10 parts assert data["stats"]["progress_percent"] == 40.0 # plates: 2/5 assert data["stats"]["parts_progress_percent"] == 40.0 # parts: 10/25 class TestProjectArchivedStatusNotCounted: """Tests for bug #630: archived files added to a project should not count as printed.""" @pytest.fixture async def project_factory(self, db_session): """Factory to create test projects.""" async def _create_project(**kwargs): from backend.app.models.project import Project defaults = { "name": "Archived Status Test", "description": "Test project", "color": "#FF0000", } defaults.update(kwargs) project = Project(**defaults) db_session.add(project) await db_session.commit() await db_session.refresh(project) return project return _create_project @pytest.fixture async def archive_factory(self, db_session): """Factory to create a test archive plus a matching PrintLogEntry — see TestProjectPartsTracking.archive_factory for rationale (#1593).""" async def _create_archive(**kwargs): from backend.app.models.archive import PrintArchive from backend.app.models.print_log import PrintLogEntry defaults = { "filename": "test.3mf", "file_path": "test/test.3mf", "file_size": 1000, "print_name": "Test Print", "status": "completed", "quantity": 1, } defaults.update(kwargs) archive = PrintArchive(**defaults) db_session.add(archive) await db_session.commit() await db_session.refresh(archive) if archive.status != "archived": db_session.add( PrintLogEntry( archive_id=archive.id, print_name=archive.print_name, status=archive.status, ) ) await db_session.commit() return archive return _create_archive @pytest.mark.asyncio @pytest.mark.integration async def test_archived_files_not_counted_as_completed( self, async_client: AsyncClient, project_factory, archive_factory, db_session ): """Archived files added to a project should not count in completed_prints stats.""" project = await project_factory(target_parts_count=20) # 2 actually printed (completed), 3 just archived (not printed yet) await archive_factory(project_id=project.id, quantity=2, status="completed") await archive_factory(project_id=project.id, quantity=3, status="archived") await archive_factory(project_id=project.id, quantity=5, status="archived") response = await async_client.get(f"/api/v1/projects/{project.id}") assert response.status_code == 200 data = response.json() # Only the completed archive should count assert data["stats"]["completed_prints"] == 2 assert data["stats"]["parts_progress_percent"] == 10.0 # 2/20 = 10% assert data["stats"]["remaining_parts"] == 18 @pytest.mark.asyncio @pytest.mark.integration async def test_archived_files_not_counted_in_project_list( self, async_client: AsyncClient, project_factory, archive_factory, db_session ): """Project list endpoint should not count archived files as completed.""" project = await project_factory(name="List Archived Test", target_parts_count=50) await archive_factory(project_id=project.id, quantity=4, status="completed") await archive_factory(project_id=project.id, quantity=6, status="archived") response = await async_client.get("/api/v1/projects/") assert response.status_code == 200 data = response.json() our_project = next((p for p in data if p["name"] == "List Archived Test"), None) assert our_project is not None assert our_project["completed_count"] == 4 # Only completed, not archived # Post-#1593: archive_count is "print runs", not "files attached". An # ``archived``-status file (shelved without printing) has no # PrintLogEntry and doesn't count — only the actual printed run does. assert our_project["archive_count"] == 1 @pytest.mark.asyncio @pytest.mark.integration async def test_only_completed_status_counts( self, async_client: AsyncClient, project_factory, archive_factory, db_session ): """Only 'completed' status should count in stats, not archived/failed/etc.""" project = await project_factory(target_parts_count=100) await archive_factory(project_id=project.id, quantity=10, status="completed") await archive_factory(project_id=project.id, quantity=5, status="archived") await archive_factory(project_id=project.id, quantity=3, status="failed") await archive_factory(project_id=project.id, quantity=2, status="aborted") response = await async_client.get(f"/api/v1/projects/{project.id}") assert response.status_code == 200 data = response.json() assert data["stats"]["completed_prints"] == 10 # Only "completed" assert data["stats"]["failed_prints"] == 2 # failed + aborted (count of runs) # Post-#1593: total_archives counts runs from print_log_entries, not # files. The ``archived`` row is a shelved file with no run, so it # contributes 0; the other three (completed, failed, aborted) each # produced a run. assert data["stats"]["total_archives"] == 3 # total_items sums quantity per run: 10 (completed) + 3 (failed) + 2 (aborted) = 15 assert data["stats"]["total_items"] == 15 class TestProjectStatsPerRun: """Project stats aggregate per-run from ``print_log_entries`` so reprints and multi-plate prints count every run (#1593). Pre-fix the stats counted ``print_archives`` (one row per file), so 3 reprints of one file showed as 1 job with plate-1-only filament/time/cost. """ @pytest.fixture async def project_factory(self, db_session): async def _create_project(**kwargs): from backend.app.models.project import Project defaults = {"name": "Per-Run Stats Project", "color": "#FF0000"} defaults.update(kwargs) project = Project(**defaults) db_session.add(project) await db_session.commit() await db_session.refresh(project) return project return _create_project @pytest.fixture async def archive_with_runs(self, db_session): """Build a single archive + N PrintLogEntry rows. Models the reporter's case: one source file (archive) is reprinted N times, each run with its own duration / filament / cost. """ async def _create(*, project_id: int, runs: list[dict], archive_status: str = "completed", quantity: int = 1): from backend.app.models.archive import PrintArchive from backend.app.models.print_log import PrintLogEntry archive = PrintArchive( filename="reprinted.3mf", file_path="test/reprinted.3mf", file_size=1000, print_name="Reprinted Print", status=archive_status, quantity=quantity, project_id=project_id, ) db_session.add(archive) await db_session.commit() await db_session.refresh(archive) for run in runs: db_session.add( PrintLogEntry( archive_id=archive.id, print_name=archive.print_name, status=run.get("status", "completed"), duration_seconds=run.get("duration_seconds"), filament_used_grams=run.get("filament_used_grams"), cost=run.get("cost"), energy_kwh=run.get("energy_kwh"), energy_cost=run.get("energy_cost"), ) ) await db_session.commit() return archive return _create @pytest.mark.asyncio @pytest.mark.integration async def test_three_reprints_count_as_three_jobs_with_summed_totals( self, async_client: AsyncClient, project_factory, archive_with_runs ): """Reporter's case: 3 runs of one multi-plate file should report 3 jobs and summed time / filament / cost — pre-fix it reported 1 job with plate-1-only totals.""" project = await project_factory() await archive_with_runs( project_id=project.id, runs=[ {"duration_seconds": 7140, "filament_used_grams": 19.2, "cost": 0.40}, {"duration_seconds": 6000, "filament_used_grams": 20.0, "cost": 0.40}, {"duration_seconds": 6300, "filament_used_grams": 18.8, "cost": 0.40}, ], ) response = await async_client.get(f"/api/v1/projects/{project.id}") assert response.status_code == 200 stats = response.json()["stats"] assert stats["total_archives"] == 3, "3 runs must show as 3 jobs" assert stats["completed_prints"] == 3, "Each run with quantity=1 contributes 1 part" assert stats["total_filament_grams"] == round(19.2 + 20.0 + 18.8, 2) assert stats["total_print_time_hours"] == round((7140 + 6000 + 6300) / 3600, 2) # Cost rounds at 2 decimals — 3 * 0.40 = 1.20 assert stats["estimated_cost"] == 1.20 @pytest.mark.asyncio @pytest.mark.integration async def test_orphan_log_entries_do_not_bleed_into_projects( self, async_client: AsyncClient, project_factory, db_session ): """Log rows whose ``archive_id`` is NULL (archive deleted via ON DELETE SET NULL) must not leak into any project — the inner join filters them out by construction.""" from backend.app.models.print_log import PrintLogEntry project = await project_factory() # Orphan log entries — no archive_id. for _ in range(5): db_session.add( PrintLogEntry( archive_id=None, print_name="Orphan Run", status="completed", duration_seconds=3600, filament_used_grams=20.0, cost=0.5, ) ) await db_session.commit() response = await async_client.get(f"/api/v1/projects/{project.id}") assert response.status_code == 200 stats = response.json()["stats"] # None of the orphan rows are attributable to this project. assert stats["total_archives"] == 0 assert stats["completed_prints"] == 0 assert stats["total_filament_grams"] == 0 assert stats["total_print_time_hours"] == 0 assert stats["estimated_cost"] == 0 @pytest.mark.asyncio @pytest.mark.integration async def test_mixed_run_outcomes_split_completed_and_failed( self, async_client: AsyncClient, project_factory, archive_with_runs ): """A multi-run archive with mixed outcomes splits cleanly between completed_prints (per-quantity) and failed_prints (per-run).""" project = await project_factory() await archive_with_runs( project_id=project.id, quantity=2, runs=[ {"status": "completed", "filament_used_grams": 30.0}, {"status": "completed", "filament_used_grams": 30.0}, {"status": "failed", "filament_used_grams": 5.0}, {"status": "aborted", "filament_used_grams": 2.0}, ], ) response = await async_client.get(f"/api/v1/projects/{project.id}") stats = response.json()["stats"] assert stats["total_archives"] == 4 # 2 completed runs × quantity=2 each = 4 parts assert stats["completed_prints"] == 4 # 2 failure runs (failed + aborted) count as 2, not 2*quantity assert stats["failed_prints"] == 2 # All 4 runs contribute filament: 30 + 30 + 5 + 2 = 67 assert stats["total_filament_grams"] == 67.0 @pytest.mark.asyncio @pytest.mark.integration async def test_quick_stats_in_list_view_agree_with_per_project_stats( self, async_client: AsyncClient, project_factory, archive_with_runs ): """The /projects list view's quick stats must agree with /projects/{id}'s detailed stats — both come from the same per-run aggregation.""" project = await project_factory(name="Quick-Stats Alignment") await archive_with_runs( project_id=project.id, quantity=1, runs=[ {"status": "completed"}, {"status": "completed"}, {"status": "failed"}, ], ) list_resp = await async_client.get("/api/v1/projects/") ours = next(p for p in list_resp.json() if p["name"] == "Quick-Stats Alignment") assert ours["archive_count"] == 3 assert ours["completed_count"] == 2 assert ours["failed_count"] == 1 class TestProjectArchivesAPI: """Tests for project-archive relationships.""" @pytest.fixture async def project_factory(self, db_session): """Factory to create test projects.""" async def _create_project(**kwargs): from backend.app.models.project import Project defaults = { "name": "Archive Test Project", "description": "Test project", "color": "#0000FF", } defaults.update(kwargs) project = Project(**defaults) db_session.add(project) await db_session.commit() await db_session.refresh(project) return project return _create_project @pytest.mark.asyncio @pytest.mark.integration async def test_get_project_with_archives(self, async_client: AsyncClient, project_factory, db_session): """Verify project can be retrieved with archive count.""" project = await project_factory() response = await async_client.get(f"/api/v1/projects/{project.id}") assert response.status_code == 200 # Project should have an archive count (may be 0) data = response.json() assert "name" in data @pytest.mark.asyncio @pytest.mark.integration async def test_list_archives_in_project_returns_archives_with_creator( self, async_client: AsyncClient, project_factory, db_session ): """``GET /projects/{id}/archives`` must eagerly load both the project AND the creator User. Without selectinload(created_by) the response converter triggers a lazy attribute load on a closed async session and the request 500s with MissingGreenlet — exactly what was reported the moment a user with auth enabled (so archives carry created_by_id) opened a project view. """ from backend.app.models.archive import PrintArchive from backend.app.models.user import User # Seed: a user (the eventual creator) and a project owning two archives, # one with created_by_id set, one without. creator = User( username="archive-creator", password_hash="x", role="user", is_active=True, ) db_session.add(creator) await db_session.commit() await db_session.refresh(creator) project = await project_factory(name="Project Archives Smoke") attributed = PrintArchive( filename="attributed.3mf", file_path="x/attributed.3mf", file_size=2048, print_name="Attributed Print", status="completed", quantity=1, project_id=project.id, created_by_id=creator.id, ) anonymous = PrintArchive( filename="anon.3mf", file_path="x/anon.3mf", file_size=2048, print_name="Anonymous Print", status="completed", quantity=1, project_id=project.id, created_by_id=None, ) db_session.add_all([attributed, anonymous]) await db_session.commit() response = await async_client.get(f"/api/v1/projects/{project.id}/archives?limit=100&offset=0") assert response.status_code == 200, f"Expected 200, got {response.status_code} body={response.text}" rows = response.json() assert len(rows) == 2 # Both archive shapes serialise — the attributed one surfaces the # creator username (proving the eager-load worked) and the anonymous # one stays None without exploding. by_filename = {r["filename"]: r for r in rows} assert by_filename["attributed.3mf"]["created_by_username"] == "archive-creator" assert by_filename["attributed.3mf"]["created_by_id"] == creator.id assert by_filename["anon.3mf"]["created_by_username"] is None assert by_filename["anon.3mf"]["created_by_id"] is None class TestProjectExportImport: """Tests for project export/import functionality.""" @pytest.fixture async def project_factory(self, db_session): """Factory to create test projects.""" _counter = [0] async def _create_project(**kwargs): from backend.app.models.project import Project _counter[0] += 1 counter = _counter[0] defaults = { "name": f"Export Test Project {counter}", "description": "Test project for export", "color": "#00FF00", } defaults.update(kwargs) project = Project(**defaults) db_session.add(project) await db_session.commit() await db_session.refresh(project) return project return _create_project @pytest.fixture async def bom_item_factory(self, db_session): """Factory to create test BOM items.""" async def _create_bom_item(project_id: int, **kwargs): from backend.app.models.project_bom import ProjectBOMItem defaults = { "project_id": project_id, "name": "Test Part", "quantity_needed": 1, "quantity_acquired": 0, "sort_order": 0, } defaults.update(kwargs) item = ProjectBOMItem(**defaults) db_session.add(item) await db_session.commit() await db_session.refresh(item) return item return _create_bom_item @pytest.mark.asyncio @pytest.mark.integration async def test_export_project(self, async_client: AsyncClient, project_factory, bom_item_factory, db_session): """Verify project export includes BOM items.""" project = await project_factory( name="Export Me", description="A test project", target_count=10, target_parts_count=50, budget=100.0, ) # Add BOM items await bom_item_factory(project.id, name="M3x8 Screws", quantity_needed=20, unit_price=0.10) await bom_item_factory(project.id, name="Heat Inserts", quantity_needed=10, unit_price=0.25) # Test JSON format export response = await async_client.get(f"/api/v1/projects/{project.id}/export?format=json") assert response.status_code == 200 data = response.json() assert data["name"] == "Export Me" assert data["description"] == "A test project" assert data["target_count"] == 10 assert data["target_parts_count"] == 50 assert data["budget"] == 100.0 assert len(data["bom_items"]) == 2 # Check BOM items bom_names = [item["name"] for item in data["bom_items"]] assert "M3x8 Screws" in bom_names assert "Heat Inserts" in bom_names # Test ZIP format export (default) zip_response = await async_client.get(f"/api/v1/projects/{project.id}/export") assert zip_response.status_code == 200 assert zip_response.headers["content-type"] == "application/zip" @pytest.mark.asyncio @pytest.mark.integration async def test_import_project(self, async_client: AsyncClient): """Verify project can be imported with BOM items.""" import_data = { "name": "Imported Project", "description": "Imported from JSON", "color": "#FF00FF", "target_count": 5, "target_parts_count": 25, "budget": 50.0, "bom_items": [ { "name": "PTFE Tubes", "quantity_needed": 4, "quantity_acquired": 0, "unit_price": 2.50, "sourcing_url": "https://example.com", "stl_filename": None, "remarks": "Need 4mm ID", }, ], } response = await async_client.post("/api/v1/projects/import", json=import_data) assert response.status_code == 200 data = response.json() assert data["name"] == "Imported Project" assert data["description"] == "Imported from JSON" assert data["target_count"] == 5 assert data["target_parts_count"] == 25 assert data["budget"] == 50.0 assert data["id"] > 0 # Has a valid ID # BOM stats should show 1 item imported assert data["stats"]["bom_total_items"] == 1 @pytest.mark.asyncio @pytest.mark.integration async def test_export_project_with_linked_folder(self, async_client: AsyncClient, project_factory, db_session): """Verify project export includes linked folders.""" from backend.app.models.library import LibraryFolder project = await project_factory(name="Project With Folder") # Create a linked folder folder = LibraryFolder(name="Project Files", project_id=project.id) db_session.add(folder) await db_session.commit() response = await async_client.get(f"/api/v1/projects/{project.id}/export?format=json") assert response.status_code == 200 data = response.json() assert data["name"] == "Project With Folder" assert len(data["linked_folders"]) == 1 assert data["linked_folders"][0]["name"] == "Project Files" @pytest.mark.asyncio @pytest.mark.integration async def test_import_project_with_linked_folder(self, async_client: AsyncClient): """Verify project import accepts linked folders data.""" import_data = { "name": "Imported With Folders", "linked_folders": [ {"name": "STL Files"}, {"name": "Documentation"}, ], } # Import should succeed with linked_folders response = await async_client.post("/api/v1/projects/import", json=import_data) assert response.status_code == 200 data = response.json() assert data["name"] == "Imported With Folders" assert data["id"] > 0 @pytest.mark.asyncio @pytest.mark.integration async def test_import_project_from_json_file(self, async_client: AsyncClient): """Verify project can be imported from JSON file upload.""" import io import json project_data = { "name": "File Uploaded Project", "description": "Imported from JSON file", "color": "#123456", } # Create a file-like object file_content = json.dumps(project_data).encode() files = {"file": ("project.json", io.BytesIO(file_content), "application/json")} response = await async_client.post("/api/v1/projects/import/file", files=files) assert response.status_code == 200 data = response.json() assert data["name"] == "File Uploaded Project" assert data["description"] == "Imported from JSON file" @pytest.mark.asyncio @pytest.mark.integration async def test_import_project_from_zip_file(self, async_client: AsyncClient): """Verify project can be imported from ZIP file with files.""" import io import json import zipfile project_data = { "name": "ZIP Imported Project", "description": "Imported from ZIP", "linked_folders": [{"name": "TestFolder", "files": [{"filename": "test.txt"}]}], } # Create a ZIP file in memory zip_buffer = io.BytesIO() with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf: zf.writestr("project.json", json.dumps(project_data)) zf.writestr("files/TestFolder/test.txt", "Hello World") zip_buffer.seek(0) files = {"file": ("project.zip", zip_buffer, "application/zip")} response = await async_client.post("/api/v1/projects/import/file", files=files) assert response.status_code == 200 data = response.json() assert data["name"] == "ZIP Imported Project" assert data["description"] == "Imported from ZIP" @pytest.mark.asyncio @pytest.mark.integration async def test_export_zip_contains_files(self, async_client: AsyncClient, project_factory, db_session): """Verify ZIP export contains actual files from linked folders.""" import io import json import zipfile from pathlib import Path from backend.app.api.routes.library import get_library_dir from backend.app.models.library import LibraryFile, LibraryFolder project = await project_factory(name="Project With Files") # Create a linked folder with is_external fields folder = LibraryFolder( name="TestExportFolder", project_id=project.id, is_external=False, external_readonly=False, external_show_hidden=False, ) db_session.add(folder) await db_session.flush() # Create a test file on disk library_dir = get_library_dir() folder_path = library_dir / "TestExportFolder" folder_path.mkdir(parents=True, exist_ok=True) test_file_path = folder_path / "test_export.txt" test_file_path.write_text("Export test content") # Create library file record lib_file = LibraryFile( folder_id=folder.id, filename="test_export.txt", file_path="TestExportFolder/test_export.txt", file_type="other", file_size=19, is_external=False, ) db_session.add(lib_file) await db_session.commit() # Export as ZIP response = await async_client.get(f"/api/v1/projects/{project.id}/export") assert response.status_code == 200 assert response.headers["content-type"] == "application/zip" # Verify ZIP contents zip_buffer = io.BytesIO(response.content) with zipfile.ZipFile(zip_buffer, "r") as zf: assert "project.json" in zf.namelist() assert "files/TestExportFolder/test_export.txt" in zf.namelist() # Verify file content file_content = zf.read("files/TestExportFolder/test_export.txt").decode() assert file_content == "Export test content" # Verify project.json project_data = json.loads(zf.read("project.json")) assert project_data["name"] == "Project With Files" # Cleanup test_file_path.unlink(missing_ok=True) folder_path.rmdir() @pytest.mark.asyncio @pytest.mark.integration async def test_import_invalid_file_type(self, async_client: AsyncClient): """Verify import rejects invalid file types.""" import io files = {"file": ("project.txt", io.BytesIO(b"invalid"), "text/plain")} response = await async_client.post("/api/v1/projects/import/file", files=files) assert response.status_code == 400 assert "must be .zip or .json" in response.json()["detail"] @pytest.mark.asyncio @pytest.mark.integration async def test_import_zip_missing_project_json(self, async_client: AsyncClient): """Verify import rejects ZIP without project.json.""" import io import zipfile zip_buffer = io.BytesIO() with zipfile.ZipFile(zip_buffer, "w") as zf: zf.writestr("other.txt", "no project.json here") zip_buffer.seek(0) files = {"file": ("project.zip", zip_buffer, "application/zip")} response = await async_client.post("/api/v1/projects/import/file", files=files) assert response.status_code == 400 assert "project.json" in response.json()["detail"] @pytest.mark.asyncio @pytest.mark.integration async def test_import_invalid_json(self, async_client: AsyncClient): """Verify import rejects invalid JSON content.""" import io files = {"file": ("project.json", io.BytesIO(b"not valid json"), "application/json")} response = await async_client.post("/api/v1/projects/import/file", files=files) assert response.status_code == 400 assert "Invalid JSON" in response.json()["detail"] @pytest.mark.asyncio @pytest.mark.integration async def test_import_rejects_absolute_path_in_folder_name(self, async_client: AsyncClient, tmp_path): """Absolute paths in `linked_folders[*].name` must not escape library_dir. Verbatim shape from the upstream advisory: attacker sets folder name to an absolute path, expecting Python's ``Path("/lib") / "/anywhere"`` to collapse to ``Path("/anywhere")`` and let the next file write land outside the library directory. """ import io import json import zipfile target_outside = tmp_path / "outside" / "owned" # Build a ZIP whose folder name points outside library_dir entirely. zip_buffer = io.BytesIO() with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf: zf.writestr( "project.json", json.dumps( { "name": "innocent", "linked_folders": [{"name": str(target_outside)}], } ), ) zf.writestr(f"files/{target_outside}/evil.pth", b"import os; os.system('echo pwned > /tmp/owned')\n") zip_buffer.seek(0) files = {"file": ("evil.zip", zip_buffer, "application/zip")} response = await async_client.post("/api/v1/projects/import/file", files=files) assert response.status_code == 400, response.text assert not target_outside.exists(), "Attacker payload landed outside library_dir" @pytest.mark.asyncio @pytest.mark.integration async def test_import_rejects_dotdot_in_folder_name(self, async_client: AsyncClient): """`..` segments in folder name must be rejected.""" import io import json import zipfile zip_buffer = io.BytesIO() with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf: zf.writestr( "project.json", json.dumps( { "name": "innocent", "linked_folders": [{"name": "../../../etc"}], } ), ) zf.writestr("files/../../../etc/x.txt", b"x") zip_buffer.seek(0) files = {"file": ("evil.zip", zip_buffer, "application/zip")} response = await async_client.post("/api/v1/projects/import/file", files=files) assert response.status_code == 400, response.text @pytest.mark.asyncio @pytest.mark.integration async def test_import_rejects_dotdot_in_relative_path(self, async_client: AsyncClient): """`..` segments in the per-entry path (Vector B in the advisory) must be rejected even when the folder name itself is fine.""" import io import json import zipfile zip_buffer = io.BytesIO() with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf: zf.writestr( "project.json", json.dumps( { "name": "innocent", "linked_folders": [{"name": "ok"}], } ), ) # Folder name is benign, but the file path inside attempts to # escape via ``..``. zf.writestr("files/ok/../../../etc/x.txt", b"x") zip_buffer.seek(0) files = {"file": ("evil.zip", zip_buffer, "application/zip")} response = await async_client.post("/api/v1/projects/import/file", files=files) assert response.status_code == 400, response.text @pytest.mark.asyncio @pytest.mark.integration async def test_import_legit_nested_zip_still_works(self, async_client: AsyncClient): """A legitimate ZIP with a nested file path inside the folder must continue to import cleanly. Guards against the fix being over-strict.""" import io import json import zipfile zip_buffer = io.BytesIO() with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf: zf.writestr( "project.json", json.dumps( { "name": "nested-ok", "linked_folders": [{"name": "OkFolder"}], } ), ) zf.writestr("files/OkFolder/sub/dir/inside.txt", b"hello") zip_buffer.seek(0) files = {"file": ("nested.zip", zip_buffer, "application/zip")} response = await async_client.post("/api/v1/projects/import/file", files=files) assert response.status_code == 200, response.text data = response.json() assert data["name"] == "nested-ok" class TestProjectListEditableFields: """Tests for #2536 — the project list payload must carry every field the shared edit dialog renders. The dialog is opened from both the project list and the project detail page and seeds itself from whichever project object it is handed, so a field missing from the list payload shows up blank there and is saved back over the stored value.""" @pytest.fixture async def project_factory(self, db_session): async def _create(**kwargs): from backend.app.models.project import Project defaults = {"name": "Editable Fields Project", "color": "#123456"} defaults.update(kwargs) project = Project(**defaults) db_session.add(project) await db_session.commit() await db_session.refresh(project) return project return _create @pytest.mark.asyncio @pytest.mark.integration async def test_list_carries_the_fields_the_edit_dialog_renders(self, async_client: AsyncClient, project_factory): """The list view is where the reporter saw an empty tags field.""" from datetime import datetime await project_factory( name="Tagged Project", tags="prototype,client-work", due_date=datetime(2026, 8, 1, 12, 0, 0), priority="high", target_parts_count=7, ) response = await async_client.get("/api/v1/projects/") assert response.status_code == 200 item = next(p for p in response.json() if p["name"] == "Tagged Project") assert item["tags"] == "prototype,client-work" assert item["due_date"].startswith("2026-08-01") assert item["priority"] == "high" assert item["target_parts_count"] == 7 @pytest.mark.asyncio @pytest.mark.integration async def test_template_list_carries_them_too(self, async_client: AsyncClient, project_factory): """Templates feed the same dialog, so they need the same payload.""" await project_factory( name="Tagged Template", is_template=True, tags="reusable", priority="urgent", target_parts_count=3, ) response = await async_client.get("/api/v1/projects/templates") assert response.status_code == 200 item = next(p for p in response.json() if p["name"] == "Tagged Template") assert item["tags"] == "reusable" assert item["priority"] == "urgent" assert item["target_parts_count"] == 3 @pytest.mark.asyncio @pytest.mark.integration async def test_priority_survives_an_edit_that_does_not_touch_it(self, async_client: AsyncClient, project_factory): """A save from the list view used to submit the default priority over a stored 'high' — the dialog never received the real one.""" project = await project_factory(name="Important", priority="high", tags="keep-me") response = await async_client.patch(f"/api/v1/projects/{project.id}", json={"name": "Still Important"}) assert response.status_code == 200 result = response.json() assert result["priority"] == "high" assert result["tags"] == "keep-me" @pytest.mark.asyncio @pytest.mark.integration async def test_explicit_null_clears_tags_and_due_date(self, async_client: AsyncClient, project_factory): """Emptying the field in the dialog has to actually remove the value.""" from datetime import datetime project = await project_factory(name="Clearable", tags="obsolete", due_date=datetime(2026, 8, 1, 12, 0, 0)) response = await async_client.patch(f"/api/v1/projects/{project.id}", json={"tags": None, "due_date": None}) assert response.status_code == 200 result = response.json() assert result["tags"] is None assert result["due_date"] is None class TestProjectFileProgress: """Per-file print progress inside a project (#1897). Covers GET /projects/{id}/file-progress (attribution: library_file_id → content hash → filename, completed runs only, project-scoped), the target_sets field round-trip, and the add-to-queue project inheritance that feeds the attribution chain. """ @pytest.fixture async def project_factory(self, db_session): _counter = [0] async def _create_project(**kwargs): from backend.app.models.project import Project _counter[0] += 1 defaults = {"name": f"Progress Project {_counter[0]}"} defaults.update(kwargs) project = Project(**defaults) db_session.add(project) await db_session.commit() await db_session.refresh(project) return project return _create_project @pytest.fixture async def folder_factory(self, db_session): _counter = [0] async def _create_folder(**kwargs): from backend.app.models.library import LibraryFolder _counter[0] += 1 defaults = {"name": f"ProgressFolder {_counter[0]}"} defaults.update(kwargs) folder = LibraryFolder(**defaults) db_session.add(folder) await db_session.commit() await db_session.refresh(folder) return folder return _create_folder @pytest.fixture async def file_factory(self, db_session): _counter = [0] async def _create_file(**kwargs): from backend.app.models.library import LibraryFile _counter[0] += 1 counter = _counter[0] defaults = { "filename": f"plate_{counter}.gcode.3mf", "file_path": f"library/plate_{counter}.gcode.3mf", "file_size": 1024, "file_type": "3mf", } defaults.update(kwargs) lib_file = LibraryFile(**defaults) db_session.add(lib_file) await db_session.commit() await db_session.refresh(lib_file) return lib_file return _create_file @pytest.mark.asyncio @pytest.mark.integration async def test_counts_by_library_file_id( self, async_client: AsyncClient, project_factory, folder_factory, file_factory, printer_factory, archive_factory ): """Runs stamped with library_file_id count toward that file even when the archive's filename differs (rename after dispatch).""" project = await project_factory() folder = await folder_factory(project_id=project.id) file_a = await file_factory(folder_id=folder.id) file_b = await file_factory(folder_id=folder.id) printer = await printer_factory() for _ in range(2): await archive_factory( printer.id, project_id=project.id, library_file_id=file_a.id, filename="renamed_on_dispatch.gcode.3mf", ) await archive_factory(printer.id, project_id=project.id, library_file_id=file_b.id) response = await async_client.get(f"/api/v1/projects/{project.id}/file-progress") assert response.status_code == 200 counts = {row["file_id"]: row["completed_count"] for row in response.json()} assert counts == {file_a.id: 2, file_b.id: 1} @pytest.mark.asyncio @pytest.mark.integration async def test_hash_and_filename_fallback( self, async_client: AsyncClient, project_factory, folder_factory, file_factory, printer_factory, archive_factory ): """Historical archives without library_file_id match by content hash, then by filename.""" project = await project_factory() folder = await folder_factory(project_id=project.id) hashed_file = await file_factory(folder_id=folder.id, file_hash="a" * 64) named_file = await file_factory(folder_id=folder.id, filename="unique_name.gcode.3mf") printer = await printer_factory() # Hash match despite a different filename await archive_factory( printer.id, project_id=project.id, content_hash="a" * 64, filename="printer_copy.gcode.3mf" ) # Filename match with no hash on either side await archive_factory(printer.id, project_id=project.id, filename="unique_name.gcode.3mf") response = await async_client.get(f"/api/v1/projects/{project.id}/file-progress") counts = {row["file_id"]: row["completed_count"] for row in response.json()} assert counts == {hashed_file.id: 1, named_file.id: 1} @pytest.mark.asyncio @pytest.mark.integration async def test_only_completed_runs_count( self, async_client: AsyncClient, project_factory, folder_factory, file_factory, printer_factory, archive_factory ): """Failed runs and never-printed archives do not advance the count.""" project = await project_factory() folder = await folder_factory(project_id=project.id) lib_file = await file_factory(folder_id=folder.id) printer = await printer_factory() await archive_factory(printer.id, project_id=project.id, library_file_id=lib_file.id) await archive_factory( printer.id, project_id=project.id, library_file_id=lib_file.id, status="failed", run_status="failed" ) await archive_factory(printer.id, project_id=project.id, library_file_id=lib_file.id, with_run=False) response = await async_client.get(f"/api/v1/projects/{project.id}/file-progress") counts = {row["file_id"]: row["completed_count"] for row in response.json()} assert counts == {lib_file.id: 1} @pytest.mark.asyncio @pytest.mark.integration async def test_scoped_to_project( self, async_client: AsyncClient, project_factory, folder_factory, file_factory, printer_factory, archive_factory ): """Runs of the same file outside the project (no project / another project) are excluded.""" project = await project_factory() other_project = await project_factory() folder = await folder_factory(project_id=project.id) lib_file = await file_factory(folder_id=folder.id) printer = await printer_factory() await archive_factory(printer.id, project_id=None, library_file_id=lib_file.id) await archive_factory(printer.id, project_id=other_project.id, library_file_id=lib_file.id) response = await async_client.get(f"/api/v1/projects/{project.id}/file-progress") assert response.json() == [] @pytest.mark.asyncio @pytest.mark.integration async def test_unknown_project_404(self, async_client: AsyncClient): response = await async_client.get("/api/v1/projects/999999/file-progress") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_target_sets_roundtrip(self, async_client: AsyncClient): """target_sets survives create, update, and explicit-null clearing.""" create = await async_client.post("/api/v1/projects/", json={"name": "Sets Project", "target_sets": 10}) assert create.status_code == 200 project = create.json() assert project["target_sets"] == 10 update = await async_client.patch(f"/api/v1/projects/{project['id']}", json={"target_sets": 4}) assert update.status_code == 200, update.json() assert update.json()["target_sets"] == 4 cleared = await async_client.patch(f"/api/v1/projects/{project['id']}", json={"target_sets": None}) assert cleared.json()["target_sets"] is None untouched = await async_client.patch(f"/api/v1/projects/{project['id']}", json={"name": "Renamed"}) assert untouched.json()["target_sets"] is None @pytest.mark.asyncio @pytest.mark.integration async def test_add_to_queue_inherits_folder_project( self, async_client: AsyncClient, project_factory, folder_factory, file_factory, db_session, tmp_path ): """Queueing a file from a project-linked folder attributes the queue item (and thus the later archive) to that project; a root file stays unattributed.""" from sqlalchemy import select from backend.app.models.print_queue import PrintQueueItem project = await project_factory() folder = await folder_factory(project_id=project.id) on_disk = tmp_path / "linked.gcode.3mf" on_disk.write_bytes(b"fake sliced content") linked_file = await file_factory(folder_id=folder.id, file_path=str(on_disk)) root_disk = tmp_path / "root.gcode.3mf" root_disk.write_bytes(b"fake sliced content") root_file = await file_factory(folder_id=None, file_path=str(root_disk)) response = await async_client.post( "/api/v1/library/files/add-to-queue", json={"file_ids": [linked_file.id, root_file.id]} ) assert response.status_code == 200 assert len(response.json()["added"]) == 2 result = await db_session.execute( select(PrintQueueItem.library_file_id, PrintQueueItem.project_id).where( PrintQueueItem.library_file_id.in_([linked_file.id, root_file.id]) ) ) projects_by_file = dict(result.all()) assert projects_by_file[linked_file.id] == project.id assert projects_by_file[root_file.id] is None class TestSoftDeletedArchivesLeaveTheProject: """Deleting a print removes it from its project, everywhere (#2731). The default archive delete is soft (#1343): the files go, the row stays so global Quick Stats keeps counting its filament / time / cost. Nothing in the projects module filtered on that, so a deleted print stayed listed on the project with a thumbnail pointing at a file that no longer existed — and could not be unassigned, because the only unassign UI lives on the Archives page, which correctly hides it. Unlike Quick Stats, project *counts* exclude it too. A project is a piece of work with a definite membership, not a lifetime total, so a project that lists one print must not claim two. """ @pytest.fixture async def project_factory(self, db_session): async def _create_project(**kwargs): from backend.app.models.project import Project defaults = {"name": "Deleted Archive Project", "color": "#FF0000"} defaults.update(kwargs) project = Project(**defaults) db_session.add(project) await db_session.commit() await db_session.refresh(project) return project return _create_project @pytest.fixture async def archive_factory(self, db_session): """Archive + matching PrintLogEntry, as production always writes both.""" async def _create_archive(**kwargs): from backend.app.models.archive import PrintArchive from backend.app.models.print_log import PrintLogEntry defaults = { "filename": "test.3mf", "file_path": "test/test.3mf", "file_size": 1000, "print_name": "Test Print", "status": "completed", "quantity": 1, "thumbnail_path": "test/thumb.png", } defaults.update(kwargs) archive = PrintArchive(**defaults) db_session.add(archive) await db_session.commit() await db_session.refresh(archive) db_session.add( PrintLogEntry( archive_id=archive.id, print_name=archive.print_name, status=archive.status, filament_used_grams=10.0, ) ) await db_session.commit() return archive return _create_archive @staticmethod async def _soft_delete(db_session, archive) -> int: """Soft-delete *archive* and return its id. The commit expires the instance, so reading an attribute off it afterwards is lazy IO outside the greenlet context (MissingGreenlet). Callers take the id from here instead. """ from datetime import datetime, timezone archive_id = archive.id archive.deleted_at = datetime.now(timezone.utc) await db_session.commit() return archive_id @pytest.mark.asyncio @pytest.mark.integration async def test_deleted_archive_is_not_listed_on_the_project( self, async_client: AsyncClient, project_factory, archive_factory, db_session ): """The reported symptom: a card with a broken preview image.""" project = await project_factory() await archive_factory(project_id=project.id, print_name="Kept") gone = await archive_factory(project_id=project.id, print_name="Deleted") await self._soft_delete(db_session, gone) response = await async_client.get(f"/api/v1/projects/{project.id}/archives") assert response.status_code == 200 assert [a["print_name"] for a in response.json()] == ["Kept"] @pytest.mark.asyncio @pytest.mark.integration async def test_deleted_archive_is_not_a_preview_on_the_project_card( self, async_client: AsyncClient, project_factory, archive_factory, db_session ): """The overview page renders these as thumbnails too, so it broke there as well — not just on the detail page.""" project = await project_factory() gone = await archive_factory(project_id=project.id, print_name="Deleted") await self._soft_delete(db_session, gone) response = await async_client.get("/api/v1/projects/") assert response.status_code == 200 row = next(p for p in response.json() if p["id"] == project.id) assert row["archives"] == [] @pytest.mark.asyncio @pytest.mark.integration async def test_project_counts_exclude_the_deleted_archive( self, async_client: AsyncClient, project_factory, archive_factory, db_session ): """The list shows one print, so the count must say one.""" project = await project_factory() await archive_factory(project_id=project.id, print_name="Kept") gone = await archive_factory(project_id=project.id, print_name="Deleted") await self._soft_delete(db_session, gone) response = await async_client.get("/api/v1/projects/") row = next(p for p in response.json() if p["id"] == project.id) assert row["archive_count"] == 1 @pytest.mark.asyncio @pytest.mark.integration async def test_project_stats_exclude_the_deleted_archive( self, async_client: AsyncClient, project_factory, archive_factory, db_session ): """Deliberate divergence from #1343: the contribution leaves the project even though it stays in global Quick Stats.""" project = await project_factory() await archive_factory(project_id=project.id, print_name="Kept") gone = await archive_factory(project_id=project.id, print_name="Deleted") await self._soft_delete(db_session, gone) response = await async_client.get(f"/api/v1/projects/{project.id}") assert response.status_code == 200 stats = response.json()["stats"] assert stats["total_archives"] == 1 assert stats["total_filament_grams"] == pytest.approx(10.0) @pytest.mark.asyncio @pytest.mark.integration async def test_deleted_archive_is_not_in_the_project_timeline( self, async_client: AsyncClient, project_factory, archive_factory, db_session ): """A timeline entry for it links to an archive that 404s when clicked.""" project = await project_factory() gone = await archive_factory(project_id=project.id, print_name="Deleted") await self._soft_delete(db_session, gone) response = await async_client.get(f"/api/v1/projects/{project.id}/timeline") assert response.status_code == 200 assert not any(e.get("description") == "Deleted" for e in response.json()) @pytest.mark.asyncio @pytest.mark.integration async def test_a_live_archive_is_untouched_by_all_of_this( self, async_client: AsyncClient, project_factory, archive_factory ): """The filter must not cost a project its actual prints.""" project = await project_factory() await archive_factory(project_id=project.id, print_name="Kept") listing = await async_client.get(f"/api/v1/projects/{project.id}/archives") assert [a["print_name"] for a in listing.json()] == ["Kept"] stats = await async_client.get(f"/api/v1/projects/{project.id}") assert stats.json()["stats"]["total_archives"] == 1 row = next(p for p in (await async_client.get("/api/v1/projects/")).json() if p["id"] == project.id) assert row["archive_count"] == 1 assert len(row["archives"]) == 1 @pytest.mark.asyncio @pytest.mark.integration async def test_unassigning_an_already_orphaned_link_still_works( self, async_client: AsyncClient, project_factory, archive_factory, db_session ): """The listings hide it, but the API must still be able to clear the link — that is the repair path for rows written before this fix.""" from sqlalchemy import select from backend.app.models.archive import PrintArchive project = await project_factory() gone = await archive_factory(project_id=project.id, print_name="Deleted") gone_id = await self._soft_delete(db_session, gone) response = await async_client.post( f"/api/v1/projects/{project.id}/remove-archives", json={"archive_ids": [gone_id]} ) assert response.status_code == 200 db_session.expire_all() result = await db_session.execute(select(PrintArchive.project_id).where(PrintArchive.id == gone_id)) assert result.scalar_one() is None class TestSubProjectRollup: """Tests for #1264 — nesting projects and rolling their figures up. The parent/child columns predate this; what these cover is the roll-up, the cycle guard that a roll-up needs to terminate, and what a delete does to the branch hanging off it. """ @pytest.fixture async def project_factory(self, db_session): async def _create_project(**kwargs): from backend.app.models.project import Project defaults = {"name": "Rollup Project", "color": "#FF0000"} defaults.update(kwargs) project = Project(**defaults) db_session.add(project) await db_session.commit() await db_session.refresh(project) return project return _create_project @pytest.fixture async def run_factory(self, db_session): """One completed run against a project, with figures worth summing.""" async def _create_run(project_id, *, grams=100.0, cost=5.0, seconds=3600, status="completed", quantity=1): from backend.app.models.archive import PrintArchive from backend.app.models.print_log import PrintLogEntry archive = PrintArchive( filename="test.3mf", file_path="test/test.3mf", file_size=1000, print_name="Run", status=status, quantity=quantity, project_id=project_id, ) db_session.add(archive) await db_session.commit() await db_session.refresh(archive) db_session.add( PrintLogEntry( archive_id=archive.id, print_name=archive.print_name, status=status, duration_seconds=seconds, filament_used_grams=grams, cost=cost, ) ) await db_session.commit() return archive return _create_run @pytest.mark.asyncio @pytest.mark.integration async def test_a_master_project_rolls_up_every_sub_project( self, async_client: AsyncClient, project_factory, run_factory ): """The whole point of the feature: one number for the programme.""" master = await project_factory(name="Airframe") wing = await project_factory(name="Wing", parent_id=master.id) tail = await project_factory(name="Tail", parent_id=master.id) await run_factory(master.id, grams=10.0, cost=1.0, seconds=3600) await run_factory(wing.id, grams=20.0, cost=2.0, seconds=7200) await run_factory(tail.id, grams=30.0, cost=3.0, seconds=1800) body = (await async_client.get(f"/api/v1/projects/{master.id}")).json() assert body["descendant_count"] == 2 assert body["rollup_stats"]["total_archives"] == 3 assert body["rollup_stats"]["total_filament_grams"] == 60.0 assert body["rollup_stats"]["estimated_cost"] == 6.0 assert body["rollup_stats"]["total_print_time_hours"] == 3.5 @pytest.mark.asyncio @pytest.mark.integration async def test_the_masters_own_stats_still_mean_its_own_prints( self, async_client: AsyncClient, project_factory, run_factory ): """``stats`` keeps its existing meaning — anyone who nested projects over the API before this shipped must not see their figures restated.""" master = await project_factory(name="Airframe") wing = await project_factory(name="Wing", parent_id=master.id) await run_factory(master.id, grams=10.0) await run_factory(wing.id, grams=20.0) body = (await async_client.get(f"/api/v1/projects/{master.id}")).json() assert body["stats"]["total_archives"] == 1 assert body["stats"]["total_filament_grams"] == 10.0 @pytest.mark.asyncio @pytest.mark.integration async def test_the_roll_up_reaches_past_the_first_generation( self, async_client: AsyncClient, project_factory, run_factory ): """Nesting is arbitrary depth, so a grandchild has to count too.""" master = await project_factory(name="Airframe") wing = await project_factory(name="Wing", parent_id=master.id) spar = await project_factory(name="Spar", parent_id=wing.id) await run_factory(spar.id, grams=50.0) body = (await async_client.get(f"/api/v1/projects/{master.id}")).json() assert body["descendant_count"] == 2 assert body["rollup_stats"]["total_filament_grams"] == 50.0 @pytest.mark.asyncio @pytest.mark.integration async def test_a_childless_project_reports_no_roll_up_at_all( self, async_client: AsyncClient, project_factory, run_factory ): """Null, not a copy of ``stats`` — the page uses the absence to stay quiet rather than printing the same figures twice.""" lonely = await project_factory(name="Solo") await run_factory(lonely.id) body = (await async_client.get(f"/api/v1/projects/{lonely.id}")).json() assert body["rollup_stats"] is None assert body["descendant_count"] == 0 assert body["children"] == [] @pytest.mark.asyncio @pytest.mark.integration async def test_each_listed_child_carries_its_own_branch_total( self, async_client: AsyncClient, project_factory, run_factory ): """Otherwise the listed rows do not add up to the master's total and the page contradicts itself.""" master = await project_factory(name="Airframe") wing = await project_factory(name="Wing", parent_id=master.id) spar = await project_factory(name="Spar", parent_id=wing.id) await run_factory(wing.id, grams=20.0, cost=2.0) await run_factory(spar.id, grams=30.0, cost=3.0) body = (await async_client.get(f"/api/v1/projects/{master.id}")).json() assert len(body["children"]) == 1 row = body["children"][0] assert row["name"] == "Wing" assert row["descendant_count"] == 1 assert row["total_archives"] == 2 assert row["total_filament_grams"] == 50.0 assert row["total_cost"] == 5.0 @pytest.mark.asyncio @pytest.mark.integration async def test_roll_up_progress_measures_against_the_summed_targets( self, async_client: AsyncClient, project_factory, run_factory ): """A target on each part of the tree is a target for the whole.""" master = await project_factory(name="Airframe", target_count=2) wing = await project_factory(name="Wing", parent_id=master.id, target_count=2) await run_factory(master.id) await run_factory(wing.id) await run_factory(wing.id) body = (await async_client.get(f"/api/v1/projects/{master.id}")).json() assert body["stats"]["progress_percent"] == 50.0 # 1 of its own 2 assert body["rollup_stats"]["progress_percent"] == 75.0 # 3 of the tree's 4 assert body["rollup_stats"]["remaining_prints"] == 1 @pytest.mark.asyncio @pytest.mark.integration async def test_a_project_cannot_be_moved_under_its_own_sub_project( self, async_client: AsyncClient, project_factory ): """Rejecting only the direct self-parent left A -> B -> A reachable in two calls, and a cycle has no root to roll anything up to.""" master = await project_factory(name="Airframe") wing = await project_factory(name="Wing", parent_id=master.id) response = await async_client.patch(f"/api/v1/projects/{master.id}", json={"parent_id": wing.id}) assert response.status_code == 400 assert "sub-projects" in response.json()["detail"] @pytest.mark.asyncio @pytest.mark.integration async def test_the_guard_reaches_a_distant_descendant_too(self, async_client: AsyncClient, project_factory): """A three-deep loop is no more legal than a two-deep one.""" master = await project_factory(name="Airframe") wing = await project_factory(name="Wing", parent_id=master.id) spar = await project_factory(name="Spar", parent_id=wing.id) response = await async_client.patch(f"/api/v1/projects/{master.id}", json={"parent_id": spar.id}) assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration async def test_an_unrelated_project_is_still_a_legal_parent(self, async_client: AsyncClient, project_factory): """The guard must not refuse ordinary nesting.""" master = await project_factory(name="Airframe") wing = await project_factory(name="Wing", parent_id=master.id) other = await project_factory(name="Ground Station") response = await async_client.patch(f"/api/v1/projects/{other.id}", json={"parent_id": wing.id}) assert response.status_code == 200 assert response.json()["parent_id"] == wing.id @pytest.mark.asyncio @pytest.mark.integration async def test_a_cycle_already_in_the_database_does_not_hang_the_roll_up( self, async_client: AsyncClient, project_factory, db_session ): """Databases written before the guard was widened can hold A -> B -> A. Reading one has to terminate, not spin.""" from sqlalchemy import update as sa_update from backend.app.models.project import Project first = await project_factory(name="First") second = await project_factory(name="Second", parent_id=first.id) # Straight to the table: the API now refuses to write this. await db_session.execute(sa_update(Project).where(Project.id == first.id).values(parent_id=second.id)) await db_session.commit() response = await async_client.get(f"/api/v1/projects/{first.id}") assert response.status_code == 200 assert response.json()["descendant_count"] == 1 @pytest.mark.asyncio @pytest.mark.integration async def test_deleting_a_middle_layer_promotes_its_children( self, async_client: AsyncClient, project_factory, db_session ): """Collapse the tree by one rather than scattering the branch.""" from sqlalchemy import select from backend.app.models.project import Project master = await project_factory(name="Airframe") wing = await project_factory(name="Wing", parent_id=master.id) spar = await project_factory(name="Spar", parent_id=wing.id) # Read the ids out before expiring: an expired instance refreshes itself # on attribute access, which is a lazy load in a sync frame. master_id, spar_id = master.id, spar.id response = await async_client.delete(f"/api/v1/projects/{wing.id}") assert response.status_code == 200 db_session.expire_all() parent_id = (await db_session.execute(select(Project.parent_id).where(Project.id == spar_id))).scalar_one() assert parent_id == master_id @pytest.mark.asyncio @pytest.mark.integration async def test_deleting_a_top_level_project_frees_its_children( self, async_client: AsyncClient, project_factory, db_session ): """Nothing to promote to, so the child becomes top-level — and the delete has to succeed at all, which the bare FK would have refused.""" from sqlalchemy import select from backend.app.models.project import Project master = await project_factory(name="Airframe") wing = await project_factory(name="Wing", parent_id=master.id) wing_id = wing.id # See the sibling test: expiring invalidates the instance. response = await async_client.delete(f"/api/v1/projects/{master.id}") assert response.status_code == 200 db_session.expire_all() parent_id = (await db_session.execute(select(Project.parent_id).where(Project.id == wing_id))).scalar_one() assert parent_id is None @pytest.mark.asyncio @pytest.mark.integration async def test_the_grid_can_tell_a_sub_project_from_a_top_level_one( self, async_client: AsyncClient, project_factory ): """Without these the list view shows eight sub-projects as eight unrelated ones.""" master = await project_factory(name="Airframe") await project_factory(name="Wing", parent_id=master.id) rows = {p["name"]: p for p in (await async_client.get("/api/v1/projects/")).json()} assert rows["Airframe"]["parent_id"] is None assert rows["Airframe"]["child_count"] == 1 assert rows["Wing"]["parent_id"] == master.id assert rows["Wing"]["child_count"] == 0 @pytest.mark.asyncio @pytest.mark.integration async def test_a_filtered_listing_still_admits_to_its_hidden_children( self, async_client: AsyncClient, project_factory ): """A parent that claimed no children would invite deleting it as if nothing hung off it.""" master = await project_factory(name="Airframe", status="active") await project_factory(name="Wing", parent_id=master.id, status="completed") rows = {p["name"]: p for p in (await async_client.get("/api/v1/projects/?status=active")).json()} assert "Wing" not in rows assert rows["Airframe"]["child_count"] == 1