mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-08-11 00:30:12 -04:00
afa0ba0dc0
Projects were flat. The parent_id column and the sub-project list existed but nothing could set a parent outside the API, and a master project's stats only ever covered its own prints. The project dialog gets a parent picker, and a project with sub-projects gets a second card covering the whole tree -- jobs, parts, time, filament, cost, and progress against every target in the tree added together. That card is separate from the project's own stats, which keep their existing meaning; widening them would have restated the figures of anyone who had already nested projects over the API. Each listed sub-project carries its own branch's roll-up, so the rows add up to the card above them. On the Projects page a sub-project is drawn inside its parent's group rather than as another card in the grid -- two cards columns apart cannot show that they belong together, whatever the caption says. A sub-project whose parent the status filter has hidden stays put and names its parent instead. compute_project_stats now goes through the same grouped aggregation as the roll-up rather than its own copy of the SQL, since the two must agree. Three things the interface made reachable: - PATCH refused only a project as its own direct parent, so A -> B -> A was two calls away. A cycle has no root to roll up to, and the walk keeps its seen-set for databases that already contain one. - A sub-project's percentage was completed quantities against the plate target, disagreeing with the page it linked to. - Deleting a mid-tree project orphaned its children at top level; they now move up to its own parent.
2134 lines
86 KiB
Python
2134 lines
86 KiB
Python
"""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):
|
||
# `<a href>` 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,<script>alert(1)</script>"},
|
||
)
|
||
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 ``<img src>`` 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
|