""" Folder API tests — FOLD-01 through FOLD-05. Covers: POST /api/folders — create (FOLD-01) GET /api/folders — list root / children (FOLD-02) GET /api/folders/{id} — get + breadcrumb (FOLD-02, FOLD-05) PATCH /api/folders/{id} — rename (FOLD-03) DELETE /api/folders/{id} — delete cascade (FOLD-03) PATCH /api/documents/{id}/folder — move document (FOLD-04) Security invariants (T-04-03-xx) tested throughout. """ from __future__ import annotations import uuid as _uuid import pytest import pytest_asyncio from sqlalchemy.ext.asyncio import AsyncSession from db.models import Document, Folder, Quota, User # ── Helpers ─────────────────────────────────────────────────────────────────── async def _create_folder(db: AsyncSession, user: User, name: str, parent=None) -> Folder: """Create a Folder row directly via ORM.""" f = Folder( user_id=user.id, name=name, parent_id=parent.id if parent else None, ) db.add(f) await db.commit() await db.refresh(f) return f async def _create_document(db: AsyncSession, user: User, *, folder=None) -> Document: """Create a minimal Document row via ORM (no MinIO object needed).""" doc_id = _uuid.uuid4() doc = Document( id=doc_id, user_id=user.id, filename="test.pdf", content_type="application/pdf", size_bytes=1024, storage_backend="minio", status="uploaded", object_key=f"{user.id}/{doc_id}/{_uuid.uuid4()}.pdf", folder_id=folder.id if folder else None, ) db.add(doc) await db.commit() await db.refresh(doc) return doc # ── FOLD-01: Create folder ──────────────────────────────────────────────────── async def test_create_root_folder(async_client, auth_user): """POST /api/folders creates a root folder, returns 201 with id/name.""" resp = await async_client.post( "/api/folders", json={"name": "MyFolder"}, headers=auth_user["headers"], ) assert resp.status_code == 201 data = resp.json() assert data["name"] == "MyFolder" assert data["parent_id"] is None assert "id" in data async def test_create_subfolder(async_client, auth_user, db_session): """POST /api/folders with parent_id creates a child folder.""" parent = await _create_folder(db_session, auth_user["user"], "Parent") resp = await async_client.post( "/api/folders", json={"name": "Child", "parent_id": str(parent.id)}, headers=auth_user["headers"], ) assert resp.status_code == 201 data = resp.json() assert data["name"] == "Child" assert data["parent_id"] == str(parent.id) async def test_create_folder_duplicate_name_409(async_client, auth_user, db_session): """POST /api/folders with same name under same parent returns 409.""" await _create_folder(db_session, auth_user["user"], "DupFolder") resp = await async_client.post( "/api/folders", json={"name": "DupFolder"}, headers=auth_user["headers"], ) assert resp.status_code == 409 async def test_create_folder_invalid_parent_404(async_client, auth_user): """POST /api/folders with non-existent parent_id returns 404.""" resp = await async_client.post( "/api/folders", json={"name": "Orphan", "parent_id": str(_uuid.uuid4())}, headers=auth_user["headers"], ) assert resp.status_code == 404 async def test_create_folder_other_users_parent_404(async_client, auth_user, admin_user, db_session): """POST /api/folders with parent owned by another user returns 404 (IDOR).""" other_parent = await _create_folder(db_session, admin_user["user"], "AdminParent") resp = await async_client.post( "/api/folders", json={"name": "Steal", "parent_id": str(other_parent.id)}, headers=auth_user["headers"], ) assert resp.status_code == 404 async def test_create_folder_requires_auth(async_client): """POST /api/folders without auth returns 401 or 403.""" resp = await async_client.post("/api/folders", json={"name": "Unauth"}) assert resp.status_code in (401, 403) # ── FOLD-02: List folders ───────────────────────────────────────────────────── async def test_list_root_folders(async_client, auth_user, db_session): """GET /api/folders returns only root folders for current user.""" await _create_folder(db_session, auth_user["user"], "RootA") await _create_folder(db_session, auth_user["user"], "RootB") resp = await async_client.get("/api/folders", headers=auth_user["headers"]) assert resp.status_code == 200 items = resp.json()["items"] names = {f["name"] for f in items} assert "RootA" in names assert "RootB" in names async def test_list_root_folders_excludes_other_users(async_client, auth_user, admin_user, db_session): """GET /api/folders does not return other users' folders.""" await _create_folder(db_session, admin_user["user"], "AdminRoot") resp = await async_client.get("/api/folders", headers=auth_user["headers"]) names = {f["name"] for f in resp.json()["items"]} assert "AdminRoot" not in names async def test_list_root_folders_excludes_subfolders(async_client, auth_user, db_session): """GET /api/folders (no parent_id) does not return nested folders.""" root = await _create_folder(db_session, auth_user["user"], "Root") await _create_folder(db_session, auth_user["user"], "Child", parent=root) resp = await async_client.get("/api/folders", headers=auth_user["headers"]) items = resp.json()["items"] names = [f["name"] for f in items] assert "Root" in names assert "Child" not in names async def test_list_children(async_client, auth_user, db_session): """GET /api/folders?parent_id=X returns direct children of X.""" root = await _create_folder(db_session, auth_user["user"], "Root") child = await _create_folder(db_session, auth_user["user"], "Child", parent=root) resp = await async_client.get( f"/api/folders?parent_id={root.id}", headers=auth_user["headers"] ) assert resp.status_code == 200 items = resp.json()["items"] assert len(items) == 1 assert items[0]["id"] == str(child.id) async def test_list_children_other_user_parent_404(async_client, auth_user, admin_user, db_session): """GET /api/folders?parent_id= with other user's folder returns 404 (T-04-03-04).""" other_folder = await _create_folder(db_session, admin_user["user"], "OtherFolder") resp = await async_client.get( f"/api/folders?parent_id={other_folder.id}", headers=auth_user["headers"] ) assert resp.status_code == 404 async def test_list_folders_has_children_field(async_client, auth_user, db_session): """GET /api/folders includes has_children: true for folders with sub-folders.""" root = await _create_folder(db_session, auth_user["user"], "WithChild") await _create_folder(db_session, auth_user["user"], "Leaf", parent=root) await _create_folder(db_session, auth_user["user"], "NoChild") resp = await async_client.get("/api/folders", headers=auth_user["headers"]) items = {f["name"]: f for f in resp.json()["items"]} assert items["WithChild"]["has_children"] is True assert items["NoChild"]["has_children"] is False # ── FOLD-02 / FOLD-05: Get folder + breadcrumb ──────────────────────────────── async def test_get_folder(async_client, auth_user, db_session): """GET /api/folders/{id} returns folder metadata.""" folder = await _create_folder(db_session, auth_user["user"], "GetMe") resp = await async_client.get(f"/api/folders/{folder.id}", headers=auth_user["headers"]) assert resp.status_code == 200 data = resp.json() assert data["name"] == "GetMe" assert data["id"] == str(folder.id) async def test_get_folder_breadcrumb_single(async_client, auth_user, db_session): """GET /api/folders/{id} breadcrumb for root folder = [{id, name}].""" folder = await _create_folder(db_session, auth_user["user"], "RootFolder") resp = await async_client.get(f"/api/folders/{folder.id}", headers=auth_user["headers"]) crumbs = resp.json()["breadcrumb"] assert len(crumbs) == 1 assert crumbs[0]["name"] == "RootFolder" async def test_get_folder_breadcrumb_deep(async_client, auth_user, db_session): """GET /api/folders/{id} breadcrumb is root-first for 3-level hierarchy.""" root = await _create_folder(db_session, auth_user["user"], "Root") mid = await _create_folder(db_session, auth_user["user"], "Mid", parent=root) leaf = await _create_folder(db_session, auth_user["user"], "Leaf", parent=mid) resp = await async_client.get(f"/api/folders/{leaf.id}", headers=auth_user["headers"]) assert resp.status_code == 200 crumbs = resp.json()["breadcrumb"] assert len(crumbs) == 3 assert crumbs[0]["name"] == "Root" assert crumbs[1]["name"] == "Mid" assert crumbs[2]["name"] == "Leaf" async def test_get_folder_wrong_owner_404(async_client, auth_user, admin_user, db_session): """GET /api/folders/{id} for another user's folder returns 404 (T-04-03-04).""" other = await _create_folder(db_session, admin_user["user"], "OtherFolder") resp = await async_client.get(f"/api/folders/{other.id}", headers=auth_user["headers"]) assert resp.status_code == 404 async def test_get_folder_not_found_404(async_client, auth_user): """GET /api/folders/{id} with non-existent id returns 404.""" resp = await async_client.get(f"/api/folders/{_uuid.uuid4()}", headers=auth_user["headers"]) assert resp.status_code == 404 # ── FOLD-03: Rename folder ──────────────────────────────────────────────────── async def test_rename_folder(async_client, auth_user, db_session): """PATCH /api/folders/{id} changes folder name, returns 200.""" folder = await _create_folder(db_session, auth_user["user"], "OldName") resp = await async_client.patch( f"/api/folders/{folder.id}", json={"name": "NewName"}, headers=auth_user["headers"], ) assert resp.status_code == 200 assert resp.json()["name"] == "NewName" async def test_rename_folder_duplicate_409(async_client, auth_user, db_session): """PATCH /api/folders/{id} to a name that already exists at same level → 409.""" await _create_folder(db_session, auth_user["user"], "Existing") to_rename = await _create_folder(db_session, auth_user["user"], "ToRename") resp = await async_client.patch( f"/api/folders/{to_rename.id}", json={"name": "Existing"}, headers=auth_user["headers"], ) assert resp.status_code == 409 async def test_rename_folder_wrong_owner_404(async_client, auth_user, admin_user, db_session): """PATCH /api/folders/{id} for another user's folder returns 404 (T-04-03-04).""" other = await _create_folder(db_session, admin_user["user"], "AdminFolder") resp = await async_client.patch( f"/api/folders/{other.id}", json={"name": "Hijacked"}, headers=auth_user["headers"], ) assert resp.status_code == 404 # ── FOLD-03: Delete folder ──────────────────────────────────────────────────── async def test_delete_empty_folder(async_client, auth_user, db_session): """DELETE /api/folders/{id} on empty folder returns 204.""" folder = await _create_folder(db_session, auth_user["user"], "Empty") resp = await async_client.delete(f"/api/folders/{folder.id}", headers=auth_user["headers"]) assert resp.status_code == 204 # Verify it's gone resp2 = await async_client.get(f"/api/folders/{folder.id}", headers=auth_user["headers"]) assert resp2.status_code == 404 async def test_delete_folder_cascade_documents(async_client, auth_user, db_session): """DELETE /api/folders/{id} cascades to documents inside it.""" folder = await _create_folder(db_session, auth_user["user"], "WithDocs") doc = await _create_document(db_session, auth_user["user"], folder=folder) resp = await async_client.delete(f"/api/folders/{folder.id}", headers=auth_user["headers"]) assert resp.status_code == 204 # Document should be deleted doc_resp = await async_client.get(f"/api/documents/{doc.id}", headers=auth_user["headers"]) assert doc_resp.status_code == 404 async def test_delete_folder_cascade_quota(async_client, auth_user, db_session): """DELETE /api/folders/{id} decrements quota by sum of deleted doc sizes.""" folder = await _create_folder(db_session, auth_user["user"], "ForQuota") # Manually set quota used_bytes to 2048 (2 x 1024) quota = await db_session.get(Quota, auth_user["user"].id) quota.used_bytes = 2048 await db_session.commit() await _create_document(db_session, auth_user["user"], folder=folder) # 1024 bytes resp = await async_client.delete(f"/api/folders/{folder.id}", headers=auth_user["headers"]) assert resp.status_code == 204 await db_session.refresh(quota) assert quota.used_bytes == 1024 # decremented by 1024 async def test_delete_subfolder_not_in_parent_list(async_client, auth_user, db_session): """DELETE /api/folders/{id} on subfolder — parent folder still exists after.""" root = await _create_folder(db_session, auth_user["user"], "Root") sub = await _create_folder(db_session, auth_user["user"], "Sub", parent=root) resp = await async_client.delete(f"/api/folders/{sub.id}", headers=auth_user["headers"]) assert resp.status_code == 204 # Root still exists root_resp = await async_client.get(f"/api/folders/{root.id}", headers=auth_user["headers"]) assert root_resp.status_code == 200 async def test_delete_folder_wrong_owner_404(async_client, auth_user, admin_user, db_session): """DELETE /api/folders/{id} for another user's folder returns 404 (T-04-03-04).""" other = await _create_folder(db_session, admin_user["user"], "AdminFolder") resp = await async_client.delete(f"/api/folders/{other.id}", headers=auth_user["headers"]) assert resp.status_code == 404 async def test_delete_folder_not_found_404(async_client, auth_user): """DELETE /api/folders/{id} with non-existent id returns 404.""" resp = await async_client.delete( f"/api/folders/{_uuid.uuid4()}", headers=auth_user["headers"] ) assert resp.status_code == 404 # ── FOLD-04: Move document ──────────────────────────────────────────────────── async def test_move_document_to_folder(async_client, auth_user, db_session): """PATCH /api/documents/{id}/folder moves doc to target folder, returns 200.""" folder = await _create_folder(db_session, auth_user["user"], "Destination") doc = await _create_document(db_session, auth_user["user"]) resp = await async_client.patch( f"/api/documents/{doc.id}/folder", json={"folder_id": str(folder.id)}, headers=auth_user["headers"], ) assert resp.status_code == 200 assert resp.json()["folder_id"] == str(folder.id) async def test_move_document_to_root(async_client, auth_user, db_session): """PATCH /api/documents/{id}/folder with folder_id: null moves doc to root.""" folder = await _create_folder(db_session, auth_user["user"], "Source") doc = await _create_document(db_session, auth_user["user"], folder=folder) resp = await async_client.patch( f"/api/documents/{doc.id}/folder", json={"folder_id": None}, headers=auth_user["headers"], ) assert resp.status_code == 200 assert resp.json()["folder_id"] is None async def test_move_document_wrong_owner_404(async_client, auth_user, admin_user, db_session): """PATCH /api/documents/{id}/folder for another user's doc returns 404 (IDOR).""" other_doc = await _create_document(db_session, admin_user["user"]) folder = await _create_folder(db_session, auth_user["user"], "MyFolder") resp = await async_client.patch( f"/api/documents/{other_doc.id}/folder", json={"folder_id": str(folder.id)}, headers=auth_user["headers"], ) assert resp.status_code == 404 async def test_move_document_to_other_users_folder_404(async_client, auth_user, admin_user, db_session): """PATCH /api/documents/{id}/folder with target folder owned by another user → 404 (T-04-03-05).""" doc = await _create_document(db_session, auth_user["user"]) other_folder = await _create_folder(db_session, admin_user["user"], "AdminFolder") resp = await async_client.patch( f"/api/documents/{doc.id}/folder", json={"folder_id": str(other_folder.id)}, headers=auth_user["headers"], ) assert resp.status_code == 404 async def test_move_document_invalid_folder_404(async_client, auth_user, db_session): """PATCH /api/documents/{id}/folder with non-existent folder_id returns 404.""" doc = await _create_document(db_session, auth_user["user"]) resp = await async_client.patch( f"/api/documents/{doc.id}/folder", json={"folder_id": str(_uuid.uuid4())}, headers=auth_user["headers"], ) assert resp.status_code == 404 # ── FOLD-05: Sort and breadcrumb edge cases ─────────────────────────────────── async def test_list_folders_sorted_by_name(async_client, auth_user, db_session): """GET /api/folders returns folders ordered alphabetically by name.""" for name in ("Zebra", "Apple", "Mango"): await _create_folder(db_session, auth_user["user"], name) resp = await async_client.get("/api/folders", headers=auth_user["headers"]) names = [f["name"] for f in resp.json()["items"]] assert names == sorted(names) async def test_breadcrumb_two_levels(async_client, auth_user, db_session): """GET /api/folders/{id} breadcrumb for two-level tree is [root, child].""" root = await _create_folder(db_session, auth_user["user"], "Root") child = await _create_folder(db_session, auth_user["user"], "Child", parent=root) resp = await async_client.get(f"/api/folders/{child.id}", headers=auth_user["headers"]) crumbs = resp.json()["breadcrumb"] assert len(crumbs) == 2 assert crumbs[0]["id"] == str(root.id) assert crumbs[1]["id"] == str(child.id) # ── Security: admin blocked from folder endpoints ──────────────────────────── async def test_admin_cannot_access_folder_endpoints(async_client, admin_user, db_session): """Admin role is blocked from folder endpoints (T-04-03-01: get_regular_user only).""" resp = await async_client.post( "/api/folders", json={"name": "AdminFolder"}, headers=admin_user["headers"] ) assert resp.status_code == 403 async def test_list_documents_scoped_to_folder(async_client, auth_user, db_session): """GET /api/documents?folder_id=X returns only docs in that folder.""" folder_a = await _create_folder(db_session, auth_user["user"], "FolderA") folder_b = await _create_folder(db_session, auth_user["user"], "FolderB") doc_a = await _create_document(db_session, auth_user["user"], folder=folder_a) await _create_document(db_session, auth_user["user"], folder=folder_b) resp = await async_client.get( f"/api/documents?folder_id={folder_a.id}", headers=auth_user["headers"] ) assert resp.status_code == 200 data = resp.json() assert data["total"] == 1 assert data["items"][0]["id"] == str(doc_a.id)