curo1305
87a32b7ee8
Adds the unified file manager view (Windows Explorer-style), collapsible folder tree sidebar item, full vitest test suite (55 tests, 4 files), and commits all Phase 4 backend/frontend fixes that were staged but uncommitted. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
495 lines
20 KiB
Python
495 lines
20 KiB
Python
"""
|
|
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)
|