mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
maziggy
70b42d1c8d
POST /library/files/add-to-queue reported every per-file rejection in an errors array and returned 200 regardless. A caller that checks the status code saw a successful request, no visible failure, and no queue item. That is a 400 now when nothing at all was added, with the same reasons in the body. A call that created some items still succeeds, because it did. The items it created were aimed at nothing. The route always wrote printer_id=None with no target_model, and the scheduler dispatches on one or the other -- so those rows matched neither branch and could never be picked up by anything. They sat in Unassigned until someone opened each one by hand. The request takes an optional printer_id or target_model for the batch, and with neither it aims each file at the model its own G-code declares. Only when a printer of that model is active: owning no H2D is the user's situation rather than their mistake, so the file still queues as the unassigned row it has always been, rather than gaining a target nothing can answer. Three gates POST /queue/ has applied for a while now apply here too, because an item reaching the scheduler through this route has to be as printable as one reaching it through that one: the cross-model check that stops a file sliced for one printer being dispatched to another (#2578), the filename check that would otherwise surface as a failed upload hours later (#1540), and the filament requirements the scheduler matches before handing a model-based item to hardware. Nothing inside Bambuddy calls this endpoint -- the Library's own Print action goes through the queue API with a printer already chosen -- which is how it came to drift this far from it. ----- fix(library): scope add-to-queue file reads to the caller The bulk add resolved its files by raw id. Every other read in this module goes through the ownership gate, and so does the single-item queue path; this one did not. Invisible rows are dropped before the loop, so they report as the plain "File not found" an unknown id already gets.
1940 lines
82 KiB
Python
1940 lines
82 KiB
Python
"""Integration tests for Library API endpoints."""
|
|
|
|
import io
|
|
import tempfile
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import select
|
|
|
|
from backend.app.core.config import settings as app_settings
|
|
from backend.app.models.print_queue import PrintQueueItem
|
|
|
|
|
|
async def _read_queue_item(db_session, item_id: int) -> PrintQueueItem:
|
|
"""Re-read a queue row the route just committed through its own session."""
|
|
db_session.expire_all()
|
|
return (await db_session.execute(select(PrintQueueItem).where(PrintQueueItem.id == item_id))).scalar_one()
|
|
|
|
|
|
class TestLibraryFoldersAPI:
|
|
"""Integration tests for library folders endpoints."""
|
|
|
|
@pytest.fixture
|
|
async def folder_factory(self, db_session):
|
|
"""Factory to create test folders."""
|
|
_counter = [0]
|
|
|
|
async def _create_folder(**kwargs):
|
|
from backend.app.models.library import LibraryFolder
|
|
|
|
_counter[0] += 1
|
|
counter = _counter[0]
|
|
|
|
defaults = {
|
|
"name": f"Test Folder {counter}",
|
|
}
|
|
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.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_list_folders_empty(self, async_client: AsyncClient, db_session):
|
|
"""Verify empty folder list returns empty array."""
|
|
response = await async_client.get("/api/v1/library/folders")
|
|
assert response.status_code == 200
|
|
assert response.json() == []
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_folder_tree_exposes_latest_activity_at_from_files(
|
|
self, async_client: AsyncClient, folder_factory, db_session
|
|
):
|
|
"""#1770: folder list returns latest_activity_at = MAX(folder.updated_at,
|
|
MAX(immediate-child file.updated_at)) so the frontend can sort by
|
|
recent activity. Adding a file with a later updated_at must bubble it.
|
|
"""
|
|
from datetime import datetime, timedelta
|
|
|
|
from backend.app.models.library import LibraryFile
|
|
|
|
folder = await folder_factory(name="Active Folder")
|
|
# File whose updated_at is well after the folder's. Activity should
|
|
# surface this timestamp, not the folder's stale one.
|
|
future = datetime.utcnow() + timedelta(hours=24)
|
|
db_session.add(
|
|
LibraryFile(
|
|
folder_id=folder.id,
|
|
filename="model.3mf",
|
|
file_path="library/model.3mf",
|
|
file_type="3mf",
|
|
file_size=123,
|
|
updated_at=future,
|
|
)
|
|
)
|
|
await db_session.commit()
|
|
|
|
response = await async_client.get("/api/v1/library/folders")
|
|
assert response.status_code == 200
|
|
items = response.json()
|
|
assert len(items) == 1
|
|
item = items[0]
|
|
assert item["id"] == folder.id
|
|
assert item["latest_activity_at"] is not None
|
|
# latest_activity_at should be at least the future stamp we set.
|
|
assert item["latest_activity_at"] >= future.isoformat()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_folder_tree_latest_activity_at_falls_back_to_folder_updated_at(
|
|
self, async_client: AsyncClient, folder_factory, db_session
|
|
):
|
|
"""#1770: a folder with no files reports its own updated_at, not null —
|
|
otherwise the activity sort would dump every empty folder to one end."""
|
|
await folder_factory(name="Empty Folder")
|
|
response = await async_client.get("/api/v1/library/folders")
|
|
assert response.status_code == 200
|
|
items = response.json()
|
|
assert len(items) == 1
|
|
item = items[0]
|
|
# latest_activity_at == folder.updated_at when there are no files
|
|
assert item["latest_activity_at"] is not None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_folder(self, async_client: AsyncClient, db_session):
|
|
"""Verify folder can be created."""
|
|
data = {"name": "New Folder"}
|
|
response = await async_client.post("/api/v1/library/folders", json=data)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["name"] == "New Folder"
|
|
assert result["id"] is not None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_nested_folder(self, async_client: AsyncClient, folder_factory, db_session):
|
|
"""Verify nested folder can be created."""
|
|
parent = await folder_factory(name="Parent")
|
|
data = {"name": "Child", "parent_id": parent.id}
|
|
response = await async_client.post("/api/v1/library/folders", json=data)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["name"] == "Child"
|
|
assert result["parent_id"] == parent.id
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_folder(self, async_client: AsyncClient, folder_factory, db_session):
|
|
"""Verify single folder can be retrieved."""
|
|
folder = await folder_factory(name="Test Folder")
|
|
response = await async_client.get(f"/api/v1/library/folders/{folder.id}")
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["id"] == folder.id
|
|
assert result["name"] == "Test Folder"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_folder_not_found(self, async_client: AsyncClient, db_session):
|
|
"""Verify 404 for non-existent folder."""
|
|
response = await async_client.get("/api/v1/library/folders/9999")
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_folder(self, async_client: AsyncClient, folder_factory, db_session):
|
|
"""Verify folder can be updated."""
|
|
folder = await folder_factory(name="Old Name")
|
|
data = {"name": "New Name"}
|
|
response = await async_client.put(f"/api/v1/library/folders/{folder.id}", json=data)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["name"] == "New Name"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_folder(self, async_client: AsyncClient, folder_factory, db_session):
|
|
"""Verify folder can be deleted."""
|
|
folder = await folder_factory()
|
|
response = await async_client.delete(f"/api/v1/library/folders/{folder.id}")
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result.get("message") or result.get("success", True)
|
|
|
|
|
|
class TestLibraryFilesAPI:
|
|
"""Integration tests for library files endpoints."""
|
|
|
|
@pytest.fixture
|
|
async def folder_factory(self, db_session):
|
|
"""Factory to create test folders."""
|
|
_counter = [0]
|
|
|
|
async def _create_folder(**kwargs):
|
|
from backend.app.models.library import LibraryFolder
|
|
|
|
_counter[0] += 1
|
|
counter = _counter[0]
|
|
|
|
defaults = {"name": f"Test Folder {counter}"}
|
|
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):
|
|
"""Factory to create test files."""
|
|
_counter = [0]
|
|
|
|
async def _create_file(**kwargs):
|
|
from backend.app.models.library import LibraryFile
|
|
|
|
_counter[0] += 1
|
|
counter = _counter[0]
|
|
|
|
defaults = {
|
|
"filename": f"test_file_{counter}.3mf",
|
|
"file_path": f"/test/path/test_file_{counter}.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_list_files_empty(self, async_client: AsyncClient, db_session):
|
|
"""Verify empty file list returns empty array."""
|
|
response = await async_client.get("/api/v1/library/files")
|
|
assert response.status_code == 200
|
|
assert response.json() == []
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_list_files_in_folder(self, async_client: AsyncClient, folder_factory, file_factory, db_session):
|
|
"""Verify files can be filtered by folder."""
|
|
folder = await folder_factory()
|
|
file1 = await file_factory(folder_id=folder.id)
|
|
await file_factory() # File in root (no folder)
|
|
|
|
response = await async_client.get(f"/api/v1/library/files?folder_id={folder.id}")
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert len(result) == 1
|
|
assert result[0]["id"] == file1.id
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_list_files_by_project_id(self, async_client: AsyncClient, folder_factory, file_factory, db_session):
|
|
"""#932: project_id filter returns files across all folders linked to the project.
|
|
|
|
Replaces the prior N+1 pattern where the frontend fired one request per
|
|
linked folder. A single JOIN query must return every file in folders whose
|
|
project_id matches, while excluding files from unlinked folders.
|
|
"""
|
|
from backend.app.models.project import Project
|
|
|
|
project = Project(name="Test Project for Files", color="#00ff00")
|
|
db_session.add(project)
|
|
await db_session.commit()
|
|
await db_session.refresh(project)
|
|
|
|
folder_a = await folder_factory(name="Folder A", project_id=project.id)
|
|
folder_b = await folder_factory(name="Folder B", project_id=project.id)
|
|
other_folder = await folder_factory(name="Unlinked")
|
|
|
|
linked_a = await file_factory(folder_id=folder_a.id, filename="a.3mf")
|
|
linked_b = await file_factory(folder_id=folder_b.id, filename="b.3mf")
|
|
await file_factory(folder_id=other_folder.id, filename="unlinked.3mf")
|
|
await file_factory(filename="root.3mf") # no folder → not part of any project
|
|
|
|
response = await async_client.get(f"/api/v1/library/files?project_id={project.id}")
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
ids = {f["id"] for f in result}
|
|
assert ids == {linked_a.id, linked_b.id}
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_list_files_folder_id_takes_precedence_over_project_id(
|
|
self, async_client: AsyncClient, folder_factory, file_factory, db_session
|
|
):
|
|
"""When both folder_id and project_id are passed, folder_id wins.
|
|
|
|
Documented precedence in list_files(): folder_id > project_id > include_root.
|
|
This guards the behavior so a future refactor can't silently flip it.
|
|
"""
|
|
from backend.app.models.project import Project
|
|
|
|
project = Project(name="Precedence Project")
|
|
db_session.add(project)
|
|
await db_session.commit()
|
|
await db_session.refresh(project)
|
|
|
|
folder_linked = await folder_factory(name="Linked", project_id=project.id)
|
|
folder_other = await folder_factory(name="Other")
|
|
|
|
await file_factory(folder_id=folder_linked.id, filename="linked.3mf")
|
|
other_file = await file_factory(folder_id=folder_other.id, filename="other.3mf")
|
|
|
|
# folder_id points at a folder that is NOT in the project — must return
|
|
# that folder's contents and ignore project_id entirely.
|
|
response = await async_client.get(f"/api/v1/library/files?folder_id={folder_other.id}&project_id={project.id}")
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert len(result) == 1
|
|
assert result[0]["id"] == other_file.id
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_list_files_internal_only(self, async_client: AsyncClient, folder_factory, file_factory, db_session):
|
|
"""#1621: `internal_only=true` restricts the listing to files in managed
|
|
storage (`is_external=False`) so a linked NAS with hundreds of files
|
|
doesn't drown the user's own uploads in the "All Files" sidebar view."""
|
|
internal_folder = await folder_factory(name="My uploads")
|
|
external_folder = await folder_factory(name="NAS", is_external=True, external_path="/mnt/nas")
|
|
|
|
internal_file = await file_factory(folder_id=internal_folder.id, filename="mine.3mf", is_external=False)
|
|
await file_factory(folder_id=external_folder.id, filename="nas.3mf", is_external=True)
|
|
root_file = await file_factory(filename="root.3mf", is_external=False) # Root-uploaded is always internal.
|
|
|
|
response = await async_client.get("/api/v1/library/files?include_root=false&internal_only=true")
|
|
assert response.status_code == 200
|
|
ids = {f["id"] for f in response.json()}
|
|
assert ids == {internal_file.id, root_file.id}
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_list_files_external_only(self, async_client: AsyncClient, folder_factory, file_factory, db_session):
|
|
"""#1621 symmetric: `external_only=true` returns the combined view
|
|
across every linked external folder so users with several mounts can
|
|
see all external content in one place without clicking each folder."""
|
|
internal_folder = await folder_factory(name="My uploads")
|
|
nas_a = await folder_factory(name="NAS A", is_external=True, external_path="/mnt/a")
|
|
nas_b = await folder_factory(name="NAS B", is_external=True, external_path="/mnt/b")
|
|
|
|
await file_factory(folder_id=internal_folder.id, filename="mine.3mf", is_external=False)
|
|
ext_a = await file_factory(folder_id=nas_a.id, filename="a.3mf", is_external=True)
|
|
ext_b = await file_factory(folder_id=nas_b.id, filename="b.3mf", is_external=True)
|
|
|
|
response = await async_client.get("/api/v1/library/files?include_root=false&external_only=true")
|
|
assert response.status_code == 200
|
|
ids = {f["id"] for f in response.json()}
|
|
assert ids == {ext_a.id, ext_b.id}
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_list_files_internal_and_external_mutually_exclusive(self, async_client: AsyncClient, db_session):
|
|
"""Both flags together is a caller bug — fail loud (400) rather than
|
|
silently picking one, so a frontend regression is caught immediately."""
|
|
response = await async_client.get("/api/v1/library/files?internal_only=true&external_only=true")
|
|
assert response.status_code == 400
|
|
assert "mutually exclusive" in response.json()["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_file(self, async_client: AsyncClient, file_factory, db_session):
|
|
"""Verify single file can be retrieved."""
|
|
lib_file = await file_factory(filename="test.3mf")
|
|
response = await async_client.get(f"/api/v1/library/files/{lib_file.id}")
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["id"] == lib_file.id
|
|
assert result["filename"] == "test.3mf"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_file_not_found(self, async_client: AsyncClient, db_session):
|
|
"""Verify 404 for non-existent file."""
|
|
response = await async_client.get("/api/v1/library/files/9999")
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_file(self, async_client: AsyncClient, file_factory, db_session):
|
|
"""Verify file can be deleted."""
|
|
lib_file = await file_factory()
|
|
response = await async_client.delete(f"/api/v1/library/files/{lib_file.id}")
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result.get("message") or result.get("success", True)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_rename_file(self, async_client: AsyncClient, file_factory, db_session):
|
|
"""Verify file can be renamed."""
|
|
lib_file = await file_factory(filename="old_name.3mf")
|
|
data = {"filename": "new_name.3mf"}
|
|
response = await async_client.put(f"/api/v1/library/files/{lib_file.id}", json=data)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["filename"] == "new_name.3mf"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_rename_file_invalid_path_separator(self, async_client: AsyncClient, file_factory, db_session):
|
|
"""Verify file rename fails with a forward slash (FAT32-illegal, #1540)."""
|
|
lib_file = await file_factory(filename="test.3mf")
|
|
data = {"filename": "path/to/file.3mf"}
|
|
response = await async_client.put(f"/api/v1/library/files/{lib_file.id}", json=data)
|
|
assert response.status_code == 400
|
|
assert "invalid character" in response.json()["detail"].lower()
|
|
assert "/" in response.json()["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_rename_file_invalid_backslash(self, async_client: AsyncClient, file_factory, db_session):
|
|
"""Verify file rename fails with a backslash (FAT32-illegal, #1540)."""
|
|
lib_file = await file_factory(filename="test.3mf")
|
|
data = {"filename": "path\\to\\file.3mf"}
|
|
response = await async_client.put(f"/api/v1/library/files/{lib_file.id}", json=data)
|
|
assert response.status_code == 400
|
|
assert "invalid character" in response.json()["detail"].lower()
|
|
assert "\\" in response.json()["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_library_stats(self, async_client: AsyncClient, folder_factory, file_factory, db_session):
|
|
"""Verify library stats endpoint returns counts."""
|
|
await folder_factory()
|
|
await folder_factory()
|
|
await file_factory()
|
|
|
|
response = await async_client.get("/api/v1/library/stats")
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["total_folders"] == 2
|
|
assert result["total_files"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_file_list_includes_user_tracking_fields(self, async_client: AsyncClient, file_factory, db_session):
|
|
"""Verify file list response includes user tracking fields (Issue #206)."""
|
|
lib_file = await file_factory(filename="test.3mf")
|
|
response = await async_client.get("/api/v1/library/files?include_root=false")
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert len(result) >= 1
|
|
# Find our test file
|
|
test_file = next((f for f in result if f["id"] == lib_file.id), None)
|
|
assert test_file is not None
|
|
# User tracking fields should be present (even if null)
|
|
assert "created_by_id" in test_file
|
|
assert "created_by_username" in test_file
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_file_detail_includes_user_tracking_fields(self, async_client: AsyncClient, file_factory, db_session):
|
|
"""Verify file detail response includes user tracking fields (Issue #206)."""
|
|
lib_file = await file_factory(filename="test_detail.3mf")
|
|
response = await async_client.get(f"/api/v1/library/files/{lib_file.id}")
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
# User tracking fields should be present (even if null)
|
|
assert "created_by_id" in result
|
|
assert "created_by_username" in result
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_file_with_user_tracking(self, async_client: AsyncClient, db_session):
|
|
"""Verify file created with user shows username in response (Issue #206)."""
|
|
from backend.app.models.library import LibraryFile
|
|
from backend.app.models.user import User
|
|
|
|
# Create a test user
|
|
user = User(username="testuploader", password_hash="fakehash", role="user")
|
|
db_session.add(user)
|
|
await db_session.flush()
|
|
|
|
# Create a file with created_by_id set
|
|
lib_file = LibraryFile(
|
|
filename="user_uploaded.3mf",
|
|
file_path="/test/user_uploaded.3mf",
|
|
file_size=2048,
|
|
file_type="3mf",
|
|
created_by_id=user.id,
|
|
)
|
|
db_session.add(lib_file)
|
|
await db_session.commit()
|
|
await db_session.refresh(lib_file)
|
|
|
|
# Verify file detail shows username
|
|
response = await async_client.get(f"/api/v1/library/files/{lib_file.id}")
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["created_by_id"] == user.id
|
|
assert result["created_by_username"] == "testuploader"
|
|
|
|
# Verify file list also shows username
|
|
response = await async_client.get("/api/v1/library/files?include_root=false")
|
|
assert response.status_code == 200
|
|
files = response.json()
|
|
test_file = next((f for f in files if f["id"] == lib_file.id), None)
|
|
assert test_file is not None
|
|
assert test_file["created_by_id"] == user.id
|
|
assert test_file["created_by_username"] == "testuploader"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_list_files_recursive_includes_subfolders(
|
|
self, async_client: AsyncClient, folder_factory, file_factory
|
|
):
|
|
"""#1268: ?recursive=true with folder_id must include every descendant.
|
|
|
|
Tree:
|
|
toys/ ← f_toys, direct file "robot_top.3mf"
|
|
cars/ ← child of toys, file "robot_car.3mf"
|
|
race/ ← grandchild, file "robot_race.3mf"
|
|
other/ ← unrelated, file "robot_other.3mf" (must NOT appear)
|
|
"""
|
|
toys = await folder_factory(name="toys")
|
|
cars = await folder_factory(name="cars", parent_id=toys.id)
|
|
race = await folder_factory(name="race", parent_id=cars.id)
|
|
other = await folder_factory(name="other")
|
|
|
|
top = await file_factory(folder_id=toys.id, filename="robot_top.3mf")
|
|
mid = await file_factory(folder_id=cars.id, filename="robot_car.3mf")
|
|
deep = await file_factory(folder_id=race.id, filename="robot_race.3mf")
|
|
await file_factory(folder_id=other.id, filename="robot_other.3mf")
|
|
|
|
# Non-recursive: only the file directly under toys.
|
|
r = await async_client.get(f"/api/v1/library/files?folder_id={toys.id}")
|
|
assert r.status_code == 200
|
|
assert {f["id"] for f in r.json()} == {top.id}
|
|
|
|
# Recursive: toys + cars + race files, but NOT other/.
|
|
r = await async_client.get(f"/api/v1/library/files?folder_id={toys.id}&recursive=true")
|
|
assert r.status_code == 200
|
|
assert {f["id"] for f in r.json()} == {top.id, mid.id, deep.id}
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_list_files_recursive_without_folder_id_is_noop(
|
|
self, async_client: AsyncClient, folder_factory, file_factory
|
|
):
|
|
"""recursive=true is meaningful only with folder_id — without it the
|
|
existing include_root branch handles scoping. Just confirming the new
|
|
param doesn't shadow that path."""
|
|
folder = await folder_factory()
|
|
f_in = await file_factory(folder_id=folder.id)
|
|
f_root = await file_factory()
|
|
|
|
r = await async_client.get("/api/v1/library/files?include_root=false&recursive=true")
|
|
assert r.status_code == 200
|
|
assert {f["id"] for f in r.json()} == {f_in.id, f_root.id}
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_folder_readme_returns_first_markdown(
|
|
self, async_client: AsyncClient, folder_factory, file_factory
|
|
):
|
|
"""#1268: /folders/{id}/readme reads on-disk content of the first .md."""
|
|
folder = await folder_factory()
|
|
with tempfile.NamedTemporaryFile(suffix=".md", delete=False, mode="w", encoding="utf-8") as f:
|
|
f.write("# Robot\n\nA cute little robot.")
|
|
md_path = f.name
|
|
try:
|
|
await file_factory(
|
|
folder_id=folder.id,
|
|
filename="README.md",
|
|
file_path=md_path,
|
|
file_type="md",
|
|
file_size=Path(md_path).stat().st_size,
|
|
)
|
|
r = await async_client.get(f"/api/v1/library/folders/{folder.id}/readme")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["filename"] == "README.md"
|
|
assert body["content"] == "# Robot\n\nA cute little robot."
|
|
assert body["truncated"] is False
|
|
finally:
|
|
import os
|
|
|
|
os.unlink(md_path)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_folder_readme_prefers_readme_over_other_md(
|
|
self, async_client: AsyncClient, folder_factory, file_factory
|
|
):
|
|
"""When the folder has multiple .md files, README.md / description.md
|
|
wins regardless of insertion order or filename case."""
|
|
folder = await folder_factory()
|
|
with tempfile.NamedTemporaryFile(suffix=".md", delete=False, mode="w", encoding="utf-8") as f:
|
|
f.write("notes notes notes")
|
|
notes_path = f.name
|
|
with tempfile.NamedTemporaryFile(suffix=".md", delete=False, mode="w", encoding="utf-8") as f:
|
|
f.write("the real one")
|
|
readme_path = f.name
|
|
try:
|
|
# notes.md inserted FIRST — naive ordering would pick this one.
|
|
await file_factory(
|
|
folder_id=folder.id,
|
|
filename="notes.md",
|
|
file_path=notes_path,
|
|
file_type="md",
|
|
)
|
|
await file_factory(
|
|
folder_id=folder.id,
|
|
filename="readme.md", # lowercase to confirm case-insensitive match
|
|
file_path=readme_path,
|
|
file_type="md",
|
|
)
|
|
r = await async_client.get(f"/api/v1/library/folders/{folder.id}/readme")
|
|
assert r.status_code == 200
|
|
assert r.json()["filename"] == "readme.md"
|
|
assert r.json()["content"] == "the real one"
|
|
finally:
|
|
import os
|
|
|
|
os.unlink(notes_path)
|
|
os.unlink(readme_path)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_folder_readme_404_when_no_markdown(
|
|
self, async_client: AsyncClient, folder_factory, file_factory
|
|
):
|
|
"""No .md in the folder → 404 so the FE can hide the side panel."""
|
|
folder = await folder_factory()
|
|
await file_factory(folder_id=folder.id, filename="model.3mf", file_type="3mf")
|
|
r = await async_client.get(f"/api/v1/library/folders/{folder.id}/readme")
|
|
assert r.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_folder_readme_404_when_folder_missing(self, async_client: AsyncClient):
|
|
r = await async_client.get("/api/v1/library/folders/999999/readme")
|
|
assert r.status_code == 404
|
|
|
|
|
|
class TestLibraryAddToQueueAPI:
|
|
"""Integration tests for /api/v1/library/files/add-to-queue endpoint."""
|
|
|
|
@pytest.fixture
|
|
async def printer_factory(self, db_session):
|
|
"""Factory to create test printers."""
|
|
_counter = [0]
|
|
|
|
async def _create_printer(**kwargs):
|
|
from backend.app.models.printer import Printer
|
|
|
|
_counter[0] += 1
|
|
counter = _counter[0]
|
|
|
|
defaults = {
|
|
"name": f"Test Printer {counter}",
|
|
"ip_address": f"192.168.1.{100 + counter}",
|
|
"serial_number": f"TESTSERIAL{counter:04d}",
|
|
"access_code": "12345678",
|
|
"model": "X1C",
|
|
}
|
|
defaults.update(kwargs)
|
|
|
|
printer = Printer(**defaults)
|
|
db_session.add(printer)
|
|
await db_session.commit()
|
|
await db_session.refresh(printer)
|
|
return printer
|
|
|
|
return _create_printer
|
|
|
|
@pytest.fixture
|
|
async def library_file_factory(self, db_session):
|
|
"""Factory to create test library files."""
|
|
_counter = [0]
|
|
|
|
async def _create_library_file(**kwargs):
|
|
from backend.app.models.library import LibraryFile
|
|
|
|
_counter[0] += 1
|
|
counter = _counter[0]
|
|
|
|
defaults = {
|
|
"filename": f"test_file_{counter}.gcode.3mf",
|
|
"file_path": f"/test/path/test_file_{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_library_file
|
|
|
|
@pytest.fixture
|
|
async def on_disk_file_factory(self, library_file_factory):
|
|
"""A library file whose bytes exist, so the route gets past its disk check."""
|
|
written: list[Path] = []
|
|
|
|
async def _create(**kwargs):
|
|
counter = len(written) + 1
|
|
rel_path = kwargs.pop("file_path", f"archive/library/files/queue_probe_{counter}.gcode.3mf")
|
|
abs_path = Path(app_settings.base_dir) / rel_path
|
|
abs_path.parent.mkdir(parents=True, exist_ok=True)
|
|
abs_path.write_bytes(b"probe")
|
|
written.append(abs_path)
|
|
kwargs.setdefault("filename", f"queue_probe_{counter}.gcode.3mf")
|
|
return await library_file_factory(file_path=rel_path, **kwargs)
|
|
|
|
yield _create
|
|
|
|
for path in written:
|
|
path.unlink(missing_ok=True)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_add_to_queue_file_not_found(self, async_client: AsyncClient, printer_factory, db_session):
|
|
"""Nothing queued is not a success (#3112).
|
|
|
|
This used to assert 200: the caller got an OK for a call that created
|
|
nothing, with the reason in a body it had no cause to read. The reason
|
|
is still reported, now where a failed call puts it.
|
|
"""
|
|
await printer_factory()
|
|
|
|
data = {"file_ids": [9999]}
|
|
response = await async_client.post("/api/v1/library/files/add-to-queue", json=data)
|
|
assert response.status_code == 400
|
|
errors = response.json()["detail"]["errors"]
|
|
assert len(errors) == 1
|
|
assert errors[0]["file_id"] == 9999
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_add_non_sliced_file_to_queue_fails(
|
|
self, async_client: AsyncClient, printer_factory, library_file_factory, db_session
|
|
):
|
|
"""Verify non-sliced file cannot be added to queue."""
|
|
await printer_factory()
|
|
lib_file = await library_file_factory(
|
|
filename="model.stl",
|
|
file_path="/test/path/model.stl",
|
|
file_type="stl",
|
|
)
|
|
|
|
data = {"file_ids": [lib_file.id]}
|
|
response = await async_client.post("/api/v1/library/files/add-to-queue", json=data)
|
|
assert response.status_code == 400
|
|
errors = response.json()["detail"]["errors"]
|
|
assert len(errors) == 1
|
|
assert "sliced" in errors[0]["error"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_partial_success_still_returns_200(
|
|
self, async_client: AsyncClient, printer_factory, library_file_factory, on_disk_file_factory, db_session
|
|
):
|
|
"""Items really were created, so the call succeeded (#3112).
|
|
|
|
The per-file errors ride along with them, which is the whole point of a
|
|
bulk endpoint. Only a call that produced nothing is a failed call.
|
|
"""
|
|
await printer_factory()
|
|
good = await on_disk_file_factory()
|
|
bad = await library_file_factory(filename="model.stl", file_path="/test/path/model.stl", file_type="stl")
|
|
|
|
response = await async_client.post("/api/v1/library/files/add-to-queue", json={"file_ids": [good.id, bad.id]})
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert [a["file_id"] for a in result["added"]] == [good.id]
|
|
assert [e["file_id"] for e in result["errors"]] == [bad.id]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_target_model_is_inferred_so_the_item_can_be_dispatched(
|
|
self, async_client: AsyncClient, printer_factory, on_disk_file_factory, db_session
|
|
):
|
|
"""#3112: an item with no printer and no target model is inert.
|
|
|
|
The scheduler dispatches on `item.printer_id` or on
|
|
`item.target_model or item.variants`; a row with neither matches no
|
|
branch and waits forever. With an active X1C present, a file that says
|
|
it was sliced for one is aimed at it.
|
|
"""
|
|
await printer_factory(model="X1C")
|
|
lib_file = await on_disk_file_factory(file_metadata={"sliced_for_model": "X1C"})
|
|
|
|
response = await async_client.post("/api/v1/library/files/add-to-queue", json={"file_ids": [lib_file.id]})
|
|
assert response.status_code == 200
|
|
item = await _read_queue_item(db_session, response.json()["added"][0]["queue_item_id"])
|
|
assert item.printer_id is None
|
|
assert item.target_model == "X1C"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_no_printer_of_that_model_leaves_the_item_unassigned(
|
|
self, async_client: AsyncClient, printer_factory, on_disk_file_factory, db_session
|
|
):
|
|
"""Aiming an item at hardware nobody owns would only look like progress.
|
|
|
|
Having no H2D is the user's situation, not their mistake, so the file
|
|
is still queued -- as the unassigned row it has always been.
|
|
"""
|
|
await printer_factory(model="X1C")
|
|
lib_file = await on_disk_file_factory(file_metadata={"sliced_for_model": "H2D"})
|
|
|
|
response = await async_client.post("/api/v1/library/files/add-to-queue", json={"file_ids": [lib_file.id]})
|
|
assert response.status_code == 200
|
|
item = await _read_queue_item(db_session, response.json()["added"][0]["queue_item_id"])
|
|
assert item.printer_id is None
|
|
assert item.target_model is None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_explicit_printer_wins_over_the_files_own_model(
|
|
self, async_client: AsyncClient, printer_factory, on_disk_file_factory, db_session
|
|
):
|
|
printer = await printer_factory(model="X1C")
|
|
# Read before the queue row is re-read: that expires the session, and
|
|
# a lazy refresh of this row would then happen outside the greenlet.
|
|
printer_id = printer.id
|
|
lib_file = await on_disk_file_factory(file_metadata={"sliced_for_model": "X1C"})
|
|
|
|
response = await async_client.post(
|
|
"/api/v1/library/files/add-to-queue",
|
|
json={"file_ids": [lib_file.id], "printer_id": printer_id},
|
|
)
|
|
assert response.status_code == 200
|
|
item = await _read_queue_item(db_session, response.json()["added"][0]["queue_item_id"])
|
|
assert item.printer_id == printer_id
|
|
assert item.target_model is None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_incompatible_target_model_is_refused_per_file(
|
|
self, async_client: AsyncClient, printer_factory, on_disk_file_factory, db_session
|
|
):
|
|
"""The same cross-model gate POST /queue/ applies (#2578).
|
|
|
|
The scheduler hands model-based items to hardware with no human in the
|
|
loop, so a file sliced for one model must not be aimed at another.
|
|
"""
|
|
await printer_factory(model="A1")
|
|
lib_file = await on_disk_file_factory(file_metadata={"sliced_for_model": "X1C"})
|
|
|
|
response = await async_client.post(
|
|
"/api/v1/library/files/add-to-queue",
|
|
json={"file_ids": [lib_file.id], "target_model": "A1"},
|
|
)
|
|
assert response.status_code == 400
|
|
assert "cannot be dispatched" in response.json()["detail"]["errors"][0]["error"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_target_model_without_an_active_printer_is_refused(
|
|
self, async_client: AsyncClient, printer_factory, on_disk_file_factory, db_session
|
|
):
|
|
await printer_factory(model="X1C")
|
|
lib_file = await on_disk_file_factory(file_metadata={"sliced_for_model": "H2D"})
|
|
|
|
response = await async_client.post(
|
|
"/api/v1/library/files/add-to-queue",
|
|
json={"file_ids": [lib_file.id], "target_model": "H2D"},
|
|
)
|
|
assert response.status_code == 400
|
|
assert "No active printers" in response.json()["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_printer_and_target_model_together_are_refused(
|
|
self, async_client: AsyncClient, printer_factory, on_disk_file_factory, db_session
|
|
):
|
|
printer = await printer_factory(model="X1C")
|
|
lib_file = await on_disk_file_factory()
|
|
|
|
response = await async_client.post(
|
|
"/api/v1/library/files/add-to-queue",
|
|
json={"file_ids": [lib_file.id], "printer_id": printer.id, "target_model": "X1C"},
|
|
)
|
|
assert response.status_code == 400
|
|
assert "both" in response.json()["detail"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_unknown_printer_is_refused(
|
|
self, async_client: AsyncClient, printer_factory, on_disk_file_factory, db_session
|
|
):
|
|
await printer_factory(model="X1C")
|
|
lib_file = await on_disk_file_factory()
|
|
|
|
response = await async_client.post(
|
|
"/api/v1/library/files/add-to-queue",
|
|
json={"file_ids": [lib_file.id], "printer_id": 999999},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"] == "Printer not found"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_a_filename_the_printer_cannot_store_is_refused(
|
|
self, async_client: AsyncClient, printer_factory, on_disk_file_factory, db_session
|
|
):
|
|
"""The Bambu SD card is FAT32; an illegal character 553s at upload.
|
|
|
|
POST /queue/ has rejected these at queue time since #1540. This route
|
|
did not, so the mistake surfaced as a print that failed later.
|
|
"""
|
|
await printer_factory(model="X1C")
|
|
lib_file = await on_disk_file_factory(filename="bad:name?.gcode.3mf")
|
|
|
|
response = await async_client.post("/api/v1/library/files/add-to-queue", json={"file_ids": [lib_file.id]})
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"]["errors"][0]["file_id"] == lib_file.id
|
|
|
|
|
|
class TestLibraryZipExtractAPI:
|
|
"""Integration tests for ZIP extraction endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_extract_zip_invalid_file_type(self, async_client: AsyncClient, db_session):
|
|
"""Verify non-ZIP files are rejected."""
|
|
# Create a fake file that's not a ZIP
|
|
files = {"file": ("test.txt", b"This is not a zip file", "text/plain")}
|
|
response = await async_client.post("/api/v1/library/files/extract-zip", files=files)
|
|
assert response.status_code == 400
|
|
assert "ZIP" in response.json()["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_extract_zip_basic(self, async_client: AsyncClient, db_session):
|
|
"""Verify basic ZIP extraction works."""
|
|
import io
|
|
|
|
# Create a simple ZIP file in memory
|
|
zip_buffer = io.BytesIO()
|
|
with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr("test1.txt", "Content of file 1")
|
|
zf.writestr("test2.txt", "Content of file 2")
|
|
zip_buffer.seek(0)
|
|
|
|
files = {"file": ("test.zip", zip_buffer.read(), "application/zip")}
|
|
response = await async_client.post("/api/v1/library/files/extract-zip", files=files)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["extracted"] == 2
|
|
assert len(result["files"]) == 2
|
|
assert len(result["errors"]) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_extract_zip_with_folders(self, async_client: AsyncClient, db_session):
|
|
"""Verify ZIP extraction preserves folder structure."""
|
|
import io
|
|
|
|
# Create a ZIP file with folder structure
|
|
zip_buffer = io.BytesIO()
|
|
with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr("folder1/file1.txt", "Content 1")
|
|
zf.writestr("folder1/subfolder/file2.txt", "Content 2")
|
|
zf.writestr("folder2/file3.txt", "Content 3")
|
|
zip_buffer.seek(0)
|
|
|
|
files = {"file": ("test.zip", zip_buffer.read(), "application/zip")}
|
|
params = {"preserve_structure": "true"}
|
|
response = await async_client.post("/api/v1/library/files/extract-zip", files=files, params=params)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["extracted"] == 3
|
|
assert result["folders_created"] >= 3 # folder1, folder1/subfolder, folder2
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_extract_zip_flat(self, async_client: AsyncClient, db_session):
|
|
"""Verify ZIP extraction can extract flat (no folders)."""
|
|
import io
|
|
|
|
# Create a ZIP file with folder structure
|
|
zip_buffer = io.BytesIO()
|
|
with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr("folder/file1.txt", "Content 1")
|
|
zf.writestr("folder/file2.txt", "Content 2")
|
|
zip_buffer.seek(0)
|
|
|
|
files = {"file": ("test.zip", zip_buffer.read(), "application/zip")}
|
|
params = {"preserve_structure": "false"}
|
|
response = await async_client.post("/api/v1/library/files/extract-zip", files=files, params=params)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["extracted"] == 2
|
|
assert result["folders_created"] == 0 # No folders created when flat
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_extract_zip_skips_macos_files(self, async_client: AsyncClient, db_session):
|
|
"""Verify ZIP extraction skips __MACOSX and hidden files."""
|
|
import io
|
|
|
|
# Create a ZIP file with macOS junk files
|
|
zip_buffer = io.BytesIO()
|
|
with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr("real_file.txt", "Real content")
|
|
zf.writestr("__MACOSX/._real_file.txt", "macOS metadata")
|
|
zf.writestr(".hidden_file", "Hidden content")
|
|
zip_buffer.seek(0)
|
|
|
|
files = {"file": ("test.zip", zip_buffer.read(), "application/zip")}
|
|
response = await async_client.post("/api/v1/library/files/extract-zip", files=files)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["extracted"] == 1 # Only real_file.txt
|
|
assert result["files"][0]["filename"] == "real_file.txt"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_extract_zip_create_folder_from_zip(self, async_client: AsyncClient, db_session):
|
|
"""Verify ZIP extraction creates a folder from the ZIP filename."""
|
|
import io
|
|
|
|
# Create a ZIP file with some files
|
|
zip_buffer = io.BytesIO()
|
|
with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr("file1.txt", "Content 1")
|
|
zf.writestr("file2.txt", "Content 2")
|
|
zip_buffer.seek(0)
|
|
|
|
files = {"file": ("MyProject.zip", zip_buffer.read(), "application/zip")}
|
|
params = {"create_folder_from_zip": "true", "preserve_structure": "false"}
|
|
response = await async_client.post("/api/v1/library/files/extract-zip", files=files, params=params)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["extracted"] == 2
|
|
assert result["folders_created"] == 1 # MyProject folder created
|
|
|
|
# Verify the files are in a folder
|
|
assert result["files"][0]["folder_id"] is not None
|
|
assert result["files"][1]["folder_id"] is not None
|
|
# Both files should be in the same folder
|
|
assert result["files"][0]["folder_id"] == result["files"][1]["folder_id"]
|
|
|
|
# Verify the folder was created with the right name
|
|
folder_response = await async_client.get(f"/api/v1/library/folders/{result['files'][0]['folder_id']}")
|
|
assert folder_response.status_code == 200
|
|
folder = folder_response.json()
|
|
assert folder["name"] == "MyProject"
|
|
|
|
|
|
class TestLibraryStlThumbnailAPI:
|
|
"""Integration tests for STL thumbnail generation endpoints."""
|
|
|
|
@pytest.fixture
|
|
async def file_factory(self, db_session):
|
|
"""Factory to create test files."""
|
|
_counter = [0]
|
|
|
|
async def _create_file(**kwargs):
|
|
from backend.app.models.library import LibraryFile
|
|
|
|
_counter[0] += 1
|
|
counter = _counter[0]
|
|
|
|
defaults = {
|
|
"filename": f"test_model_{counter}.stl",
|
|
"file_path": f"/test/path/test_model_{counter}.stl",
|
|
"file_size": 1024,
|
|
"file_type": "stl",
|
|
}
|
|
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_batch_generate_thumbnails_empty(self, async_client: AsyncClient, db_session):
|
|
"""Verify batch thumbnail generation with no files."""
|
|
data = {"all_missing": True}
|
|
response = await async_client.post("/api/v1/library/generate-stl-thumbnails", json=data)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["processed"] == 0
|
|
assert result["succeeded"] == 0
|
|
assert result["failed"] == 0
|
|
assert result["results"] == []
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_batch_generate_thumbnails_no_criteria(self, async_client: AsyncClient, db_session):
|
|
"""Verify batch thumbnail generation with no criteria returns empty."""
|
|
data = {}
|
|
response = await async_client.post("/api/v1/library/generate-stl-thumbnails", json=data)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["processed"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_batch_generate_thumbnails_file_not_on_disk(
|
|
self, async_client: AsyncClient, file_factory, db_session
|
|
):
|
|
"""Verify batch thumbnail generation handles missing files gracefully."""
|
|
# Create a file in DB but not on disk
|
|
stl_file = await file_factory(
|
|
filename="missing.stl",
|
|
file_path="/nonexistent/path/missing.stl",
|
|
thumbnail_path=None,
|
|
)
|
|
|
|
data = {"file_ids": [stl_file.id]}
|
|
response = await async_client.post("/api/v1/library/generate-stl-thumbnails", json=data)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["processed"] == 1
|
|
assert result["succeeded"] == 0
|
|
assert result["failed"] == 1
|
|
assert result["results"][0]["success"] is False
|
|
assert "not found" in result["results"][0]["error"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_batch_generate_thumbnails_with_real_stl(self, async_client: AsyncClient, db_session):
|
|
"""Verify batch thumbnail generation with a real STL file."""
|
|
from backend.app.models.library import LibraryFile
|
|
|
|
# Create a simple ASCII STL cube
|
|
stl_content = """solid cube
|
|
facet normal 0 0 -1
|
|
outer loop
|
|
vertex 0 0 0
|
|
vertex 1 0 0
|
|
vertex 1 1 0
|
|
endloop
|
|
endfacet
|
|
facet normal 0 0 1
|
|
outer loop
|
|
vertex 0 0 1
|
|
vertex 1 1 1
|
|
vertex 1 0 1
|
|
endloop
|
|
endfacet
|
|
endsolid cube"""
|
|
|
|
with tempfile.NamedTemporaryFile(suffix=".stl", delete=False, mode="w") as f:
|
|
f.write(stl_content)
|
|
stl_path = f.name
|
|
|
|
try:
|
|
# Create file in DB pointing to real STL
|
|
lib_file = LibraryFile(
|
|
filename="test_cube.stl",
|
|
file_path=stl_path,
|
|
file_size=len(stl_content),
|
|
file_type="stl",
|
|
thumbnail_path=None,
|
|
)
|
|
db_session.add(lib_file)
|
|
await db_session.commit()
|
|
await db_session.refresh(lib_file)
|
|
|
|
data = {"file_ids": [lib_file.id]}
|
|
response = await async_client.post("/api/v1/library/generate-stl-thumbnails", json=data)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["processed"] == 1
|
|
# Result depends on whether trimesh/matplotlib are installed
|
|
# Either succeeds or fails gracefully
|
|
assert result["succeeded"] + result["failed"] == 1
|
|
finally:
|
|
import os
|
|
|
|
if os.path.exists(stl_path):
|
|
os.unlink(stl_path)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_upload_file_with_stl_thumbnail_param(self, async_client: AsyncClient, db_session):
|
|
"""Verify file upload accepts generate_stl_thumbnails parameter."""
|
|
# Create a simple STL file
|
|
stl_content = b"solid test\nendsolid test"
|
|
|
|
files = {"file": ("test.stl", stl_content, "application/octet-stream")}
|
|
params = {"generate_stl_thumbnails": "false"}
|
|
response = await async_client.post("/api/v1/library/files", files=files, params=params)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["filename"] == "test.stl"
|
|
assert result["file_type"] == "stl"
|
|
# No thumbnail should be generated when disabled
|
|
assert result["thumbnail_path"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_extract_zip_with_stl_thumbnail_param(self, async_client: AsyncClient, db_session):
|
|
"""Verify ZIP extraction accepts generate_stl_thumbnails parameter."""
|
|
# Create a ZIP file containing an STL
|
|
stl_content = b"solid test\nendsolid test"
|
|
zip_buffer = io.BytesIO()
|
|
with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr("model.stl", stl_content)
|
|
zip_buffer.seek(0)
|
|
|
|
files = {"file": ("test.zip", zip_buffer.read(), "application/zip")}
|
|
params = {"generate_stl_thumbnails": "false"}
|
|
response = await async_client.post("/api/v1/library/files/extract-zip", files=files, params=params)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["extracted"] == 1
|
|
assert result["files"][0]["filename"] == "model.stl"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_batch_generate_thumbnails_by_folder(self, async_client: AsyncClient, file_factory, db_session):
|
|
"""Verify batch thumbnail generation can filter by folder."""
|
|
from backend.app.models.library import LibraryFolder
|
|
|
|
# Create a folder
|
|
folder = LibraryFolder(name="STL Folder")
|
|
db_session.add(folder)
|
|
await db_session.commit()
|
|
await db_session.refresh(folder)
|
|
|
|
# Create STL file in folder (no thumbnail)
|
|
stl_in_folder = await file_factory(
|
|
filename="in_folder.stl",
|
|
folder_id=folder.id,
|
|
thumbnail_path=None,
|
|
)
|
|
|
|
# Create STL file at root (no thumbnail)
|
|
_stl_at_root = await file_factory(
|
|
filename="at_root.stl",
|
|
folder_id=None,
|
|
thumbnail_path=None,
|
|
)
|
|
|
|
# Request thumbnails only for files in folder
|
|
data = {"folder_id": folder.id, "all_missing": True}
|
|
response = await async_client.post("/api/v1/library/generate-stl-thumbnails", json=data)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
# Should only process the file in the folder
|
|
assert result["processed"] == 1
|
|
assert result["results"][0]["file_id"] == stl_in_folder.id
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_batch_generate_thumbnails_all_missing(self, async_client: AsyncClient, file_factory, db_session):
|
|
"""Verify batch thumbnail generation finds all STL files missing thumbnails."""
|
|
# Create files with and without thumbnails
|
|
_stl_with_thumb = await file_factory(
|
|
filename="with_thumb.stl",
|
|
thumbnail_path="/some/path/thumb.png",
|
|
)
|
|
stl_without_thumb1 = await file_factory(
|
|
filename="without_thumb1.stl",
|
|
thumbnail_path=None,
|
|
)
|
|
stl_without_thumb2 = await file_factory(
|
|
filename="without_thumb2.stl",
|
|
thumbnail_path=None,
|
|
)
|
|
|
|
data = {"all_missing": True}
|
|
response = await async_client.post("/api/v1/library/generate-stl-thumbnails", json=data)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
# Should only process files without thumbnails
|
|
assert result["processed"] == 2
|
|
file_ids = {r["file_id"] for r in result["results"]}
|
|
assert stl_without_thumb1.id in file_ids
|
|
assert stl_without_thumb2.id in file_ids
|
|
|
|
|
|
class TestLibraryPathHelpers:
|
|
"""Tests for path handling utilities used for backup portability."""
|
|
|
|
def test_to_relative_path_converts_absolute(self):
|
|
"""Verify absolute paths are converted to relative paths."""
|
|
from backend.app.api.routes.library import to_relative_path
|
|
from backend.app.core.config import settings
|
|
|
|
base_dir = str(settings.base_dir)
|
|
abs_path = f"{base_dir}/archive/library/files/test.3mf"
|
|
rel_path = to_relative_path(abs_path)
|
|
|
|
assert not rel_path.startswith("/")
|
|
assert rel_path == "archive/library/files/test.3mf"
|
|
|
|
def test_to_relative_path_handles_path_object(self):
|
|
"""Verify Path objects are handled correctly."""
|
|
from pathlib import Path
|
|
|
|
from backend.app.api.routes.library import to_relative_path
|
|
from backend.app.core.config import settings
|
|
|
|
abs_path = Path(settings.base_dir) / "archive" / "test.3mf"
|
|
rel_path = to_relative_path(abs_path)
|
|
|
|
assert not rel_path.startswith("/")
|
|
assert rel_path == "archive/test.3mf"
|
|
|
|
def test_to_relative_path_returns_empty_for_empty_input(self):
|
|
"""Verify empty input returns empty string."""
|
|
from backend.app.api.routes.library import to_relative_path
|
|
|
|
assert to_relative_path("") == ""
|
|
assert to_relative_path(None) == ""
|
|
|
|
def test_to_absolute_path_converts_relative(self):
|
|
"""Verify relative paths are converted to absolute paths."""
|
|
from backend.app.api.routes.library import to_absolute_path
|
|
from backend.app.core.config import settings
|
|
|
|
rel_path = "archive/library/files/test.3mf"
|
|
abs_path = to_absolute_path(rel_path)
|
|
|
|
assert abs_path is not None
|
|
assert abs_path.is_absolute()
|
|
assert str(abs_path) == f"{settings.base_dir}/archive/library/files/test.3mf"
|
|
|
|
def test_to_absolute_path_handles_already_absolute(self):
|
|
"""Verify already absolute paths are returned as-is (for backwards compatibility)."""
|
|
from backend.app.api.routes.library import to_absolute_path
|
|
|
|
abs_path_str = "/data/archive/test.3mf"
|
|
result = to_absolute_path(abs_path_str)
|
|
|
|
assert result is not None
|
|
assert str(result) == abs_path_str
|
|
|
|
def test_to_absolute_path_returns_none_for_empty(self):
|
|
"""Verify None/empty input returns None."""
|
|
from backend.app.api.routes.library import to_absolute_path
|
|
|
|
assert to_absolute_path(None) is None
|
|
assert to_absolute_path("") is None
|
|
|
|
|
|
class TestLibraryPermissions:
|
|
"""Tests for library permission enforcement."""
|
|
|
|
@pytest.fixture
|
|
async def auth_setup(self, db_session):
|
|
"""Set up auth with users of different permission levels."""
|
|
from backend.app.core.auth import create_access_token, get_password_hash
|
|
from backend.app.models.group import Group
|
|
from backend.app.models.settings import Settings
|
|
from backend.app.models.user import User
|
|
|
|
# Enable auth
|
|
settings = Settings(key="auth_enabled", value="true")
|
|
db_session.add(settings)
|
|
await db_session.commit()
|
|
|
|
# Groups are auto-seeded during db init, but we need to commit them
|
|
await db_session.commit()
|
|
|
|
# Get groups
|
|
from sqlalchemy import select
|
|
|
|
admin_group = (await db_session.execute(select(Group).where(Group.name == "Administrators"))).scalar_one()
|
|
operator_group = (await db_session.execute(select(Group).where(Group.name == "Operators"))).scalar_one()
|
|
viewer_group = (await db_session.execute(select(Group).where(Group.name == "Viewers"))).scalar_one()
|
|
|
|
password_hash = get_password_hash("password")
|
|
|
|
# Create users
|
|
admin_user = User(username="admin_lib", password_hash=password_hash, role="admin", is_active=True)
|
|
admin_user.groups.append(admin_group)
|
|
|
|
operator_user = User(username="operator_lib", password_hash=password_hash, is_active=True)
|
|
operator_user.groups.append(operator_group)
|
|
|
|
viewer_user = User(username="viewer_lib", password_hash=password_hash, is_active=True)
|
|
viewer_user.groups.append(viewer_group)
|
|
|
|
db_session.add_all([admin_user, operator_user, viewer_user])
|
|
await db_session.commit()
|
|
|
|
# Create tokens
|
|
admin_token = create_access_token(data={"sub": admin_user.username})
|
|
operator_token = create_access_token(data={"sub": operator_user.username})
|
|
viewer_token = create_access_token(data={"sub": viewer_user.username})
|
|
|
|
return {
|
|
"admin_user": admin_user,
|
|
"operator_user": operator_user,
|
|
"viewer_user": viewer_user,
|
|
"admin_token": admin_token,
|
|
"operator_token": operator_token,
|
|
"viewer_token": viewer_token,
|
|
}
|
|
|
|
@pytest.fixture
|
|
async def test_file(self, db_session, auth_setup):
|
|
"""Create a test file owned by the operator user."""
|
|
from backend.app.models.library import LibraryFile
|
|
|
|
operator_user = auth_setup["operator_user"]
|
|
lib_file = LibraryFile(
|
|
filename="test.txt",
|
|
file_path="data/archive/library/files/test.txt",
|
|
file_type="txt",
|
|
file_size=100,
|
|
created_by_id=operator_user.id,
|
|
)
|
|
db_session.add(lib_file)
|
|
await db_session.commit()
|
|
await db_session.refresh(lib_file)
|
|
return lib_file
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_list_files_requires_library_read(self, async_client: AsyncClient, db_session, auth_setup):
|
|
"""Verify list_files requires library:read permission."""
|
|
viewer_token = auth_setup["viewer_token"]
|
|
|
|
# Viewers have library:read, should succeed
|
|
response = await async_client.get("/api/v1/library/files", headers={"Authorization": f"Bearer {viewer_token}"})
|
|
assert response.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_list_files_denied_without_permission(self, async_client: AsyncClient, db_session):
|
|
"""Verify list_files denied without auth when auth is enabled."""
|
|
from backend.app.models.settings import Settings
|
|
|
|
# Enable auth
|
|
settings = Settings(key="auth_enabled", value="true")
|
|
db_session.add(settings)
|
|
await db_session.commit()
|
|
|
|
# Request without token should fail
|
|
response = await async_client.get("/api/v1/library/files")
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_file_own_by_owner(self, async_client: AsyncClient, db_session, auth_setup, test_file):
|
|
"""Verify operator can delete their own files."""
|
|
from pathlib import Path
|
|
|
|
# Create actual file on disk so delete doesn't fail
|
|
from backend.app.core.config import settings as app_settings
|
|
|
|
file_path = Path(app_settings.base_dir) / test_file.file_path
|
|
file_path.parent.mkdir(parents=True, exist_ok=True)
|
|
file_path.write_text("test content")
|
|
|
|
operator_token = auth_setup["operator_token"]
|
|
|
|
response = await async_client.delete(
|
|
f"/api/v1/library/files/{test_file.id}", headers={"Authorization": f"Bearer {operator_token}"}
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_file_own_denied_for_others_file(self, async_client: AsyncClient, db_session, auth_setup):
|
|
"""Verify operator cannot delete files owned by others."""
|
|
# Create another operator user with a file
|
|
from sqlalchemy import select
|
|
|
|
from backend.app.core.auth import create_access_token
|
|
from backend.app.models.group import Group
|
|
from backend.app.models.library import LibraryFile
|
|
from backend.app.models.user import User
|
|
|
|
operator_group = (await db_session.execute(select(Group).where(Group.name == "Operators"))).scalar_one()
|
|
|
|
from backend.app.core.auth import get_password_hash as get_pw_hash
|
|
|
|
other_user = User(username="other_op", password_hash=get_pw_hash("password"), is_active=True)
|
|
other_user.groups.append(operator_group)
|
|
db_session.add(other_user)
|
|
await db_session.commit()
|
|
await db_session.refresh(other_user)
|
|
|
|
# Create file owned by other user
|
|
other_file = LibraryFile(
|
|
filename="other.txt",
|
|
file_path="data/archive/library/files/other.txt",
|
|
file_type="txt",
|
|
file_size=100,
|
|
created_by_id=other_user.id,
|
|
)
|
|
db_session.add(other_file)
|
|
await db_session.commit()
|
|
await db_session.refresh(other_file)
|
|
|
|
# Original operator should not be able to delete it
|
|
operator_token = auth_setup["operator_token"]
|
|
response = await async_client.delete(
|
|
f"/api/v1/library/files/{other_file.id}", headers={"Authorization": f"Bearer {operator_token}"}
|
|
)
|
|
assert response.status_code == 403
|
|
assert "your own files" in response.json()["detail"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_file_admin_can_delete_any(self, async_client: AsyncClient, db_session, auth_setup):
|
|
"""Verify admin can delete any file."""
|
|
from pathlib import Path
|
|
|
|
from backend.app.core.config import settings as app_settings
|
|
from backend.app.models.library import LibraryFile
|
|
|
|
# Create file owned by operator
|
|
operator_user = auth_setup["operator_user"]
|
|
lib_file = LibraryFile(
|
|
filename="admin_can_delete.txt",
|
|
file_path="data/archive/library/files/admin_can_delete.txt",
|
|
file_type="txt",
|
|
file_size=100,
|
|
created_by_id=operator_user.id,
|
|
)
|
|
db_session.add(lib_file)
|
|
await db_session.commit()
|
|
await db_session.refresh(lib_file)
|
|
|
|
# Create actual file on disk
|
|
file_path = Path(app_settings.base_dir) / lib_file.file_path
|
|
file_path.parent.mkdir(parents=True, exist_ok=True)
|
|
file_path.write_text("test content")
|
|
|
|
# Admin should be able to delete it
|
|
admin_token = auth_setup["admin_token"]
|
|
response = await async_client.delete(
|
|
f"/api/v1/library/files/{lib_file.id}", headers={"Authorization": f"Bearer {admin_token}"}
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_viewer_cannot_delete_files(self, async_client: AsyncClient, db_session, auth_setup, test_file):
|
|
"""Verify viewer cannot delete any files."""
|
|
viewer_token = auth_setup["viewer_token"]
|
|
|
|
response = await async_client.delete(
|
|
f"/api/v1/library/files/{test_file.id}", headers={"Authorization": f"Bearer {viewer_token}"}
|
|
)
|
|
# Viewers don't have delete_own or delete_all permissions
|
|
assert response.status_code == 403
|
|
|
|
# ---------- #1832: API-key curation under can_manage_library ----------
|
|
#
|
|
# require_ownership_permission gates API keys on `all_perm`, but the
|
|
# library deliberately split UPDATE_OWN/DELETE_OWN (allowed under
|
|
# can_manage_library) from UPDATE_ALL/DELETE_ALL (previously denied).
|
|
# That made the entire curation surface (DELETE, PUT rename, POST move)
|
|
# unreachable for API keys, including for files the key's owner uploaded.
|
|
# The fix folds UPDATE_ALL/DELETE_ALL into can_manage_library so the
|
|
# checker passes; LIBRARY_PURGE stays admin-only.
|
|
|
|
@pytest.fixture
|
|
async def manage_library_key(self, db_session, auth_setup):
|
|
"""Mint an API key owned by the admin user with can_manage_library."""
|
|
from backend.app.core.auth import generate_api_key
|
|
from backend.app.models.api_key import APIKey
|
|
|
|
admin = auth_setup["admin_user"]
|
|
full_key, key_hash, key_prefix = generate_api_key()
|
|
row = APIKey(
|
|
name="lib-curation",
|
|
key_hash=key_hash,
|
|
key_prefix=key_prefix,
|
|
user_id=admin.id,
|
|
can_manage_library=True,
|
|
)
|
|
db_session.add(row)
|
|
await db_session.commit()
|
|
return full_key
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_apikey_with_manage_library_can_delete_file(
|
|
self, async_client: AsyncClient, db_session, auth_setup, test_file, manage_library_key
|
|
):
|
|
"""Pre-#1832 this 403'd with "administrative operations" because
|
|
LIBRARY_DELETE_ALL wasn't in _APIKEY_SCOPE_BY_PERMISSION."""
|
|
from pathlib import Path
|
|
|
|
from backend.app.core.config import settings as app_settings
|
|
|
|
# Materialise the file on disk so the delete handler doesn't 500 on
|
|
# the path it tries to unlink.
|
|
file_path = Path(app_settings.base_dir) / test_file.file_path
|
|
file_path.parent.mkdir(parents=True, exist_ok=True)
|
|
file_path.write_text("test content")
|
|
|
|
response = await async_client.delete(
|
|
f"/api/v1/library/files/{test_file.id}",
|
|
headers={"X-API-Key": manage_library_key},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_apikey_with_manage_library_can_rename_file(
|
|
self, async_client: AsyncClient, db_session, auth_setup, test_file, manage_library_key
|
|
):
|
|
"""PUT /library/files/{id} is gated on LIBRARY_UPDATE_ALL/OWN. Same
|
|
#1832 path as delete."""
|
|
response = await async_client.put(
|
|
f"/api/v1/library/files/{test_file.id}",
|
|
headers={"X-API-Key": manage_library_key},
|
|
json={"filename": "renamed.txt"},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["filename"] == "renamed.txt"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_apikey_with_manage_library_can_move_file(
|
|
self, async_client: AsyncClient, db_session, auth_setup, test_file, manage_library_key
|
|
):
|
|
"""POST /library/files/move (bulk) is gated on LIBRARY_UPDATE_ALL/OWN
|
|
— same checker, same #1832 path."""
|
|
# Create a target folder the move can land in.
|
|
from backend.app.models.library import LibraryFolder
|
|
|
|
folder = LibraryFolder(name="target")
|
|
db_session.add(folder)
|
|
await db_session.commit()
|
|
await db_session.refresh(folder)
|
|
|
|
response = await async_client.post(
|
|
"/api/v1/library/files/move",
|
|
headers={"X-API-Key": manage_library_key},
|
|
json={"file_ids": [test_file.id], "folder_id": folder.id},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_apikey_without_manage_library_still_blocked(
|
|
self, async_client: AsyncClient, db_session, auth_setup, test_file
|
|
):
|
|
"""Regression guard: a key WITHOUT can_manage_library must still get
|
|
403 — the fix widens the allowed-permission set, it doesn't bypass
|
|
the per-key scope check."""
|
|
from backend.app.core.auth import generate_api_key
|
|
from backend.app.models.api_key import APIKey
|
|
|
|
admin = auth_setup["admin_user"]
|
|
full_key, key_hash, key_prefix = generate_api_key()
|
|
row = APIKey(
|
|
name="read-only",
|
|
key_hash=key_hash,
|
|
key_prefix=key_prefix,
|
|
user_id=admin.id,
|
|
can_read_status=True,
|
|
can_manage_library=False,
|
|
)
|
|
db_session.add(row)
|
|
await db_session.commit()
|
|
|
|
response = await async_client.delete(
|
|
f"/api/v1/library/files/{test_file.id}",
|
|
headers={"X-API-Key": full_key},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_apikey_with_manage_library_still_cannot_purge(
|
|
self, async_client: AsyncClient, db_session, auth_setup, manage_library_key
|
|
):
|
|
"""LIBRARY_PURGE deliberately stays in _APIKEY_DENIED_PERMISSIONS as
|
|
a genuinely destructive op that bypasses the soft-delete window.
|
|
can_manage_library does NOT grant it."""
|
|
response = await async_client.post(
|
|
"/api/v1/library/purge",
|
|
headers={"X-API-Key": manage_library_key},
|
|
json={"days_in_trash": 30},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
class TestPrintFileUploadValidation:
|
|
"""#1401: pre-flight rejection of unprintable uploads at the library +
|
|
archive routes. Smoke tests the shared ``validate_print_file_upload``
|
|
helper through both surfaces a user can reach with a drag-drop."""
|
|
|
|
def _valid_3mf_bytes(self, name: str = "Metadata/plate_1.gcode") -> bytes:
|
|
"""Build a minimal-but-real zip with the gcode-3mf magic in it so
|
|
the validator's ``startswith(b"PK\\x03\\x04")`` check passes."""
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr(name, "; G-code\nG28\n")
|
|
return buf.getvalue()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_library_rejects_raw_gcode_upload(self, async_client: AsyncClient, db_session):
|
|
"""``Foo.gcode`` direct uploads are blocked at the library route —
|
|
the dispatcher would otherwise append ``.3mf`` and ship raw gcode
|
|
to the printer as a fake 3MF."""
|
|
files = {"file": ("plate_1.gcode", b"; raw gcode\nG28\n", "application/octet-stream")}
|
|
response = await async_client.post("/api/v1/library/files", files=files)
|
|
assert response.status_code == 400
|
|
# Error message must name the actual remedy, not just say "invalid".
|
|
assert "gcode.3mf" in response.json()["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_library_rejects_non_zip_3mf_upload(self, async_client: AsyncClient, db_session):
|
|
"""A ``.3mf`` upload whose body isn't a zip is rejected — covers
|
|
raw gcode renamed to .3mf, corrupted downloads, etc."""
|
|
files = {"file": ("model.3mf", b"; raw gcode\nG28\n", "application/octet-stream")}
|
|
response = await async_client.post("/api/v1/library/files", files=files)
|
|
assert response.status_code == 400
|
|
assert "ZIP container" in response.json()["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_library_rejects_non_zip_gcode_3mf_upload(self, async_client: AsyncClient, db_session):
|
|
"""The compound-extension ``.gcode.3mf`` case is gated by the same
|
|
zip-magic check — splitext returns just ``.3mf``, but the suffix
|
|
match covers both."""
|
|
files = {"file": ("plate_1.gcode.3mf", b"; raw gcode\nG28\n", "application/octet-stream")}
|
|
response = await async_client.post("/api/v1/library/files", files=files)
|
|
assert response.status_code == 400
|
|
assert "ZIP container" in response.json()["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_library_accepts_valid_gcode_3mf_upload(self, async_client: AsyncClient, db_session):
|
|
"""A real ``.gcode.3mf`` zip uploads successfully — the existing
|
|
happy path is not regressed by the new validation."""
|
|
files = {
|
|
"file": (
|
|
"plate_1.gcode.3mf",
|
|
self._valid_3mf_bytes(),
|
|
"application/zip",
|
|
)
|
|
}
|
|
response = await async_client.post("/api/v1/library/files", files=files)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["filename"] == "plate_1.gcode.3mf"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_library_upload_classifies_gcode_3mf_as_compound(self, async_client: AsyncClient, db_session):
|
|
"""#1600 follow-up: upload path used to strip to the trailing
|
|
extension and store ``file_type='3mf'`` for sliced outputs, while
|
|
the external-folder scan stored ``file_type='gcode.3mf'``. Now
|
|
every ingest path goes through ``classify_file_type`` and
|
|
produces the canonical compound name."""
|
|
files = {
|
|
"file": (
|
|
"sliced.gcode.3mf",
|
|
self._valid_3mf_bytes(),
|
|
"application/zip",
|
|
)
|
|
}
|
|
response = await async_client.post("/api/v1/library/files", files=files)
|
|
assert response.status_code == 200
|
|
assert response.json()["file_type"] == "gcode.3mf"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_library_get_gcode_endpoint_accepts_compound_file_type(self, async_client: AsyncClient, db_session):
|
|
"""#1600 follow-up: pre-fix, ``GET /files/{id}/gcode`` only handled
|
|
``file_type`` of ``gcode`` or ``3mf`` and 400'd on a row whose
|
|
``file_type`` was ``gcode.3mf`` — exactly the rows the external-
|
|
folder scan was creating. The gate now treats both as 3MF and
|
|
unzips the embedded gcode the same way."""
|
|
from backend.app.models.library import LibraryFile
|
|
|
|
# Persist a real `.gcode.3mf` zip under file_type='gcode.3mf' so
|
|
# the endpoint hits the new branch.
|
|
with tempfile.NamedTemporaryFile(suffix=".gcode.3mf", delete=False) as tmp:
|
|
tmp.write(self._valid_3mf_bytes(name="Metadata/plate_1.gcode"))
|
|
tmp_path = tmp.name
|
|
|
|
lib_file = LibraryFile(
|
|
filename="sliced.gcode.3mf",
|
|
file_path=tmp_path,
|
|
file_type="gcode.3mf",
|
|
file_size=Path(tmp_path).stat().st_size,
|
|
)
|
|
db_session.add(lib_file)
|
|
await db_session.commit()
|
|
await db_session.refresh(lib_file)
|
|
|
|
response = await async_client.get(f"/api/v1/library/files/{lib_file.id}/gcode")
|
|
assert response.status_code == 200
|
|
assert b"G28" in response.content
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_library_get_gcode_recovers_legacy_gcode_type_for_3mf(self, async_client: AsyncClient, db_session):
|
|
"""#1709 regression guard. Before the fix, ``slice_and_persist``
|
|
wrote a `.gcode.3mf` ZIP container to disk but stored the row with
|
|
``file_type='gcode'`` — the preview endpoint then streamed the
|
|
ZIP body as ``text/plain`` and the embedded G-code viewer saw
|
|
``PK\\x03\\x04...`` instead of the toolpath. New sliced rows now
|
|
store ``file_type='gcode.3mf'``; rows already written under the
|
|
bug self-heal because the endpoint also detects the ZIP via the
|
|
``.gcode.3mf`` filename suffix when the column is still legacy."""
|
|
from backend.app.models.library import LibraryFile
|
|
|
|
with tempfile.NamedTemporaryFile(suffix=".gcode.3mf", delete=False) as tmp:
|
|
tmp.write(self._valid_3mf_bytes(name="Metadata/plate_1.gcode"))
|
|
tmp_path = tmp.name
|
|
|
|
lib_file = LibraryFile(
|
|
filename="legacy-sliced.gcode.3mf",
|
|
file_path=tmp_path,
|
|
file_type="gcode",
|
|
file_size=Path(tmp_path).stat().st_size,
|
|
)
|
|
db_session.add(lib_file)
|
|
await db_session.commit()
|
|
await db_session.refresh(lib_file)
|
|
|
|
response = await async_client.get(f"/api/v1/library/files/{lib_file.id}/gcode")
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"].startswith("text/plain")
|
|
assert b"G28" in response.content
|
|
# The whole point of #1709: must NOT be ZIP bytes shoved at the viewer.
|
|
assert not response.content.startswith(b"PK")
|
|
|
|
async def _multi_plate_file(self, db_session):
|
|
"""A two-plate `.gcode.3mf` written plate 2 first, as Bambu Studio does.
|
|
|
|
The member order is copied from the file this was reported on — taking
|
|
the first `.gcode` in the zip opened plate 2.
|
|
"""
|
|
from backend.app.models.library import LibraryFile
|
|
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr("Metadata/plate_2.gcode", "; plate two\nG28\n")
|
|
zf.writestr("Metadata/plate_1.gcode", "; plate one\nG28\n")
|
|
with tempfile.NamedTemporaryFile(suffix=".gcode.3mf", delete=False) as tmp:
|
|
tmp.write(buf.getvalue())
|
|
tmp_path = tmp.name
|
|
|
|
lib_file = LibraryFile(
|
|
filename="two-plates.gcode.3mf",
|
|
file_path=tmp_path,
|
|
file_type="gcode.3mf",
|
|
file_size=Path(tmp_path).stat().st_size,
|
|
)
|
|
db_session.add(lib_file)
|
|
await db_session.commit()
|
|
await db_session.refresh(lib_file)
|
|
return lib_file
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_library_gcode_serves_the_requested_plate(self, async_client: AsyncClient, db_session):
|
|
"""The viewer has always sent ``?plate=``; this route took no such
|
|
parameter, and FastAPI drops unknown query parameters without a word —
|
|
so picking a plate did nothing at all."""
|
|
lib_file = await self._multi_plate_file(db_session)
|
|
|
|
first = await async_client.get(f"/api/v1/library/files/{lib_file.id}/gcode?plate=1")
|
|
second = await async_client.get(f"/api/v1/library/files/{lib_file.id}/gcode?plate=2")
|
|
|
|
assert first.status_code == 200
|
|
assert b"plate one" in first.content
|
|
assert second.status_code == 200
|
|
assert b"plate two" in second.content
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_library_gcode_without_a_plate_serves_the_first_plate(self, async_client: AsyncClient, db_session):
|
|
"""Not the first member in the zip — that is plate 2 in this file, and
|
|
opening a multi-plate file from the File Manager passes no plate."""
|
|
lib_file = await self._multi_plate_file(db_session)
|
|
|
|
response = await async_client.get(f"/api/v1/library/files/{lib_file.id}/gcode")
|
|
|
|
assert response.status_code == 200
|
|
assert b"plate one" in response.content
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_library_gcode_rejects_a_plate_the_file_does_not_hold(self, async_client: AsyncClient, db_session):
|
|
"""404 rather than quietly rendering some other plate."""
|
|
lib_file = await self._multi_plate_file(db_session)
|
|
|
|
missing = await async_client.get(f"/api/v1/library/files/{lib_file.id}/gcode?plate=3")
|
|
zeroth = await async_client.get(f"/api/v1/library/files/{lib_file.id}/gcode?plate=0")
|
|
|
|
assert missing.status_code == 404
|
|
assert zeroth.status_code == 400
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_library_still_accepts_non_print_extensions(self, async_client: AsyncClient, db_session):
|
|
"""STL / image / other non-print uploads bypass the validator
|
|
entirely — Bambuddy is also a library, not just a print dispatcher."""
|
|
files = {"file": ("model.stl", b"solid test\nendsolid test", "application/octet-stream")}
|
|
response = await async_client.post(
|
|
"/api/v1/library/files", files=files, params={"generate_stl_thumbnails": "false"}
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_archive_upload_rejects_non_zip(self, async_client: AsyncClient, db_session):
|
|
"""``POST /archives/upload`` shares the same validator — covers the
|
|
manual archive-upload entry point too."""
|
|
files = {"file": ("model.3mf", b"; raw gcode\nG28\n", "application/octet-stream")}
|
|
response = await async_client.post("/api/v1/archives/upload", files=files)
|
|
assert response.status_code == 400
|
|
assert "ZIP container" in response.json()["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_archive_bulk_upload_collects_per_file_errors(self, async_client: AsyncClient, db_session):
|
|
"""The bulk-archive route reports validation failures per file and
|
|
continues processing the remaining items — one bad upload in a
|
|
10-file drag-drop must not abort the whole batch."""
|
|
good = self._valid_3mf_bytes()
|
|
bad = b"; raw gcode\nG28\n"
|
|
# httpx multipart with a list-of-tuples preserves order + same field name.
|
|
files = [
|
|
("files", ("good.3mf", good, "application/zip")),
|
|
("files", ("bad.3mf", bad, "application/octet-stream")),
|
|
]
|
|
response = await async_client.post("/api/v1/archives/upload-bulk", files=files)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
# The bulk route's archive_print may still reject the "good" file
|
|
# downstream (no printer match, etc.) — we don't care about that
|
|
# here; what matters is the bad file lands in `errors` with the
|
|
# validator's message and the route didn't 500.
|
|
assert body["failed"] >= 1
|
|
bad_errors = [e for e in body["errors"] if e["filename"] == "bad.3mf"]
|
|
assert bad_errors, body
|
|
assert "ZIP container" in bad_errors[0]["error"]
|