565 lines
21 KiB
Python
565 lines
21 KiB
Python
|
"""Additional DMS error-path tests to boost coverage above 80%."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import uuid
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from tests.conftest import ORIGIN_HEADER
|
||
|
|
|
||
|
|
PDF_CONTENT = b"%PDF-1.4\n1 0 obj<</Type/Catalog/Pages 2 0 R>>endobj\n%%EOF"
|
||
|
|
|
||
|
|
|
||
|
|
class TestFolderErrorPaths:
|
||
|
|
"""Error path tests for folder endpoints."""
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_create_folder_duplicate_name(self, authed_client):
|
||
|
|
"""POST /folders → 409 duplicate name."""
|
||
|
|
client, _ = authed_client
|
||
|
|
await client.post("/api/v1/dms/folders", json={"name": "Dup"}, headers=ORIGIN_HEADER)
|
||
|
|
resp = await client.post("/api/v1/dms/folders", json={"name": "Dup"}, headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 409
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_create_folder_parent_not_found(self, authed_client):
|
||
|
|
"""POST /folders → 404 parent not found."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/folders",
|
||
|
|
json={"name": "Child", "parent_id": str(uuid.uuid4())},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_create_folder_invalid_parent_id(self, authed_client):
|
||
|
|
"""POST /folders → 400 invalid parent_id."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/folders",
|
||
|
|
json={"name": "Child", "parent_id": "not-a-uuid"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 400
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_list_folders_with_parent_id(self, authed_client):
|
||
|
|
"""GET /folders?parent_id → children of specified parent."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/folders", json={"name": "Root"}, headers=ORIGIN_HEADER
|
||
|
|
)
|
||
|
|
root_id = resp.json()["id"]
|
||
|
|
await client.post(
|
||
|
|
"/api/v1/dms/folders",
|
||
|
|
json={"name": "Child1", "parent_id": root_id},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
await client.post(
|
||
|
|
"/api/v1/dms/folders",
|
||
|
|
json={"name": "Child2", "parent_id": root_id},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
|
||
|
|
resp = await client.get(f"/api/v1/dms/folders?parent_id={root_id}", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
children = resp.json()
|
||
|
|
assert len(children) == 2
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_list_folders_parent_not_found(self, authed_client):
|
||
|
|
"""GET /folders?parent_id=nonexistent → 404."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.get(
|
||
|
|
f"/api/v1/dms/folders?parent_id={uuid.uuid4()}", headers=ORIGIN_HEADER
|
||
|
|
)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_update_folder_not_found(self, authed_client):
|
||
|
|
"""PATCH /folders/{id} → 404 not found."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.patch(
|
||
|
|
f"/api/v1/dms/folders/{uuid.uuid4()}",
|
||
|
|
json={"name": "New"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_update_folder_duplicate_name(self, authed_client):
|
||
|
|
"""PATCH /folders/{id} → 409 duplicate name."""
|
||
|
|
client, _ = authed_client
|
||
|
|
await client.post("/api/v1/dms/folders", json={"name": "A"}, headers=ORIGIN_HEADER)
|
||
|
|
resp = await client.post("/api/v1/dms/folders", json={"name": "B"}, headers=ORIGIN_HEADER)
|
||
|
|
b_id = resp.json()["id"]
|
||
|
|
|
||
|
|
resp = await client.patch(
|
||
|
|
f"/api/v1/dms/folders/{b_id}",
|
||
|
|
json={"name": "A"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 409
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_update_folder_move_into_self(self, authed_client):
|
||
|
|
"""PATCH /folders/{id} → 400 cannot move into itself."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/folders", json={"name": "Self"}, headers=ORIGIN_HEADER
|
||
|
|
)
|
||
|
|
folder_id = resp.json()["id"]
|
||
|
|
|
||
|
|
resp = await client.patch(
|
||
|
|
f"/api/v1/dms/folders/{folder_id}",
|
||
|
|
json={"parent_id": folder_id},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 400
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_update_folder_move_parent_not_found(self, authed_client):
|
||
|
|
"""PATCH /folders/{id} → 404 parent not found."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/folders", json={"name": "Move"}, headers=ORIGIN_HEADER
|
||
|
|
)
|
||
|
|
folder_id = resp.json()["id"]
|
||
|
|
|
||
|
|
resp = await client.patch(
|
||
|
|
f"/api/v1/dms/folders/{folder_id}",
|
||
|
|
json={"parent_id": str(uuid.uuid4())},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_update_folder_cycle_detection(self, authed_client):
|
||
|
|
"""PATCH /folders/{id} → 400 cannot move into descendant."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post("/api/v1/dms/folders", json={"name": "P"}, headers=ORIGIN_HEADER)
|
||
|
|
p_id = resp.json()["id"]
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/folders",
|
||
|
|
json={"name": "C", "parent_id": p_id},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
c_id = resp.json()["id"]
|
||
|
|
|
||
|
|
# Try to move P into C (its own child)
|
||
|
|
resp = await client.patch(
|
||
|
|
f"/api/v1/dms/folders/{p_id}",
|
||
|
|
json={"parent_id": c_id},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 400
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_delete_folder_not_found(self, authed_client):
|
||
|
|
"""DELETE /folders/{id} → 404 not found."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.delete(f"/api/v1/dms/folders/{uuid.uuid4()}", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_delete_folder_invalid_id(self, authed_client):
|
||
|
|
"""DELETE /folders/{id} → 400 invalid id."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.delete("/api/v1/dms/folders/not-a-uuid", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 400
|
||
|
|
|
||
|
|
|
||
|
|
class TestFileErrorPaths:
|
||
|
|
"""Error path tests for file endpoints."""
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_upload_file_folder_not_found(self, authed_client):
|
||
|
|
"""POST /files/upload → 404 folder not found."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/upload",
|
||
|
|
files={"file": ("test.pdf", PDF_CONTENT, "application/pdf")},
|
||
|
|
data={"folder_id": str(uuid.uuid4())},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_get_file_not_found(self, authed_client):
|
||
|
|
"""GET /files/{id} → 404 not found."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.get(f"/api/v1/dms/files/{uuid.uuid4()}", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_get_file_invalid_id(self, authed_client):
|
||
|
|
"""GET /files/{id} → 400 invalid id."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.get("/api/v1/dms/files/bad-id", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 400
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_update_file_not_found(self, authed_client):
|
||
|
|
"""PATCH /files/{id} → 404 not found."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.patch(
|
||
|
|
f"/api/v1/dms/files/{uuid.uuid4()}",
|
||
|
|
json={"name": "new.pdf"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_update_file_move_folder_not_found(self, authed_client):
|
||
|
|
"""PATCH /files/{id} → 404 folder not found."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/upload",
|
||
|
|
files={"file": ("test.pdf", PDF_CONTENT, "application/pdf")},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
file_id = resp.json()["id"]
|
||
|
|
|
||
|
|
resp = await client.patch(
|
||
|
|
f"/api/v1/dms/files/{file_id}",
|
||
|
|
json={"folder_id": str(uuid.uuid4())},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_delete_file_not_found(self, authed_client):
|
||
|
|
"""DELETE /files/{id} → 404 not found."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.delete(f"/api/v1/dms/files/{uuid.uuid4()}", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_restore_file_not_found(self, authed_client):
|
||
|
|
"""POST /files/{id}/restore → 404 not found."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(f"/api/v1/dms/files/{uuid.uuid4()}/restore", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_restore_file_not_deleted(self, authed_client):
|
||
|
|
"""POST /files/{id}/restore → 404 file not in trash."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/upload",
|
||
|
|
files={"file": ("active.pdf", PDF_CONTENT, "application/pdf")},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
file_id = resp.json()["id"]
|
||
|
|
|
||
|
|
resp = await client.post(f"/api/v1/dms/files/{file_id}/restore", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_preview_file_not_found(self, authed_client):
|
||
|
|
"""GET /files/{id}/preview → 404 not found."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.get(f"/api/v1/dms/files/{uuid.uuid4()}/preview", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_edit_session_not_found(self, authed_client):
|
||
|
|
"""POST /files/{id}/edit-session → 404 not found."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
f"/api/v1/dms/files/{uuid.uuid4()}/edit-session", headers=ORIGIN_HEADER
|
||
|
|
)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_edit_session_xlsx(self, authed_client):
|
||
|
|
"""POST /files/{id}/edit-session → 200 for xlsx file."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/upload",
|
||
|
|
files={
|
||
|
|
"file": (
|
||
|
|
"sheet.xlsx",
|
||
|
|
b"PK\x03\x04",
|
||
|
|
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||
|
|
)
|
||
|
|
},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
file_id = resp.json()["id"]
|
||
|
|
|
||
|
|
resp = await client.post(f"/api/v1/dms/files/{file_id}/edit-session", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["document"]["fileType"] == "xlsx"
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_edit_session_pptx(self, authed_client):
|
||
|
|
"""POST /files/{id}/edit-session → 200 for pptx file."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/upload",
|
||
|
|
files={
|
||
|
|
"file": (
|
||
|
|
"slides.pptx",
|
||
|
|
b"PK\x03\x04",
|
||
|
|
"application/vnd.openxmlformats-officedocument.presentationml.presentation",
|
||
|
|
)
|
||
|
|
},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
file_id = resp.json()["id"]
|
||
|
|
|
||
|
|
resp = await client.post(f"/api/v1/dms/files/{file_id}/edit-session", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["document"]["fileType"] == "pptx"
|
||
|
|
|
||
|
|
|
||
|
|
class TestShareErrorPaths:
|
||
|
|
"""Error path tests for sharing endpoints."""
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_share_file_not_found(self, authed_client):
|
||
|
|
"""POST /files/{id}/share → 404 file not found."""
|
||
|
|
client, seed = authed_client
|
||
|
|
viewer_id = str(seed["viewer_a"].id)
|
||
|
|
resp = await client.post(
|
||
|
|
f"/api/v1/dms/files/{uuid.uuid4()}/share",
|
||
|
|
json={"user_ids": [viewer_id], "access_level": "read"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_share_with_invalid_user_id(self, authed_client):
|
||
|
|
"""POST /files/{id}/share → 400 invalid user_id."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/upload",
|
||
|
|
files={"file": ("share.pdf", PDF_CONTENT, "application/pdf")},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
file_id = resp.json()["id"]
|
||
|
|
|
||
|
|
resp = await client.post(
|
||
|
|
f"/api/v1/dms/files/{file_id}/share",
|
||
|
|
json={"user_ids": ["not-a-uuid"], "access_level": "read"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 400
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_share_with_group(self, authed_client):
|
||
|
|
"""POST /files/{id}/share → 200 with group_ids."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/upload",
|
||
|
|
files={"file": ("group.pdf", PDF_CONTENT, "application/pdf")},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
file_id = resp.json()["id"]
|
||
|
|
|
||
|
|
group_id = str(uuid.uuid4())
|
||
|
|
resp = await client.post(
|
||
|
|
f"/api/v1/dms/files/{file_id}/share",
|
||
|
|
json={"group_ids": [group_id], "access_level": "write"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["count"] == 1
|
||
|
|
assert resp.json()["shared_with"][0]["group_id"] == group_id
|
||
|
|
assert resp.json()["shared_with"][0]["access_level"] == "write"
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_share_duplicate_ignored(self, authed_client):
|
||
|
|
"""POST /files/{id}/share → 200, duplicate shares ignored."""
|
||
|
|
client, seed = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/upload",
|
||
|
|
files={"file": ("dup.pdf", PDF_CONTENT, "application/pdf")},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
file_id = resp.json()["id"]
|
||
|
|
viewer_id = str(seed["viewer_a"].id)
|
||
|
|
|
||
|
|
# Share once
|
||
|
|
await client.post(
|
||
|
|
f"/api/v1/dms/files/{file_id}/share",
|
||
|
|
json={"user_ids": [viewer_id], "access_level": "read"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
# Share again — should be ignored
|
||
|
|
resp = await client.post(
|
||
|
|
f"/api/v1/dms/files/{file_id}/share",
|
||
|
|
json={"user_ids": [viewer_id], "access_level": "read"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["count"] == 0
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_remove_share_group(self, authed_client):
|
||
|
|
"""DELETE /files/{id}/share → 204, group share removed."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/upload",
|
||
|
|
files={"file": ("grp.pdf", PDF_CONTENT, "application/pdf")},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
file_id = resp.json()["id"]
|
||
|
|
group_id = str(uuid.uuid4())
|
||
|
|
|
||
|
|
await client.post(
|
||
|
|
f"/api/v1/dms/files/{file_id}/share",
|
||
|
|
json={"group_ids": [group_id], "access_level": "read"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
|
||
|
|
resp = await client.request(
|
||
|
|
"DELETE",
|
||
|
|
f"/api/v1/dms/files/{file_id}/share",
|
||
|
|
json={"group_id": group_id},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 204
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_remove_share_invalid_id(self, authed_client):
|
||
|
|
"""DELETE /files/{id}/share → 400 invalid user_id."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/upload",
|
||
|
|
files={"file": ("bad.pdf", PDF_CONTENT, "application/pdf")},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
file_id = resp.json()["id"]
|
||
|
|
|
||
|
|
resp = await client.request(
|
||
|
|
"DELETE",
|
||
|
|
f"/api/v1/dms/files/{file_id}/share",
|
||
|
|
json={"user_id": "not-a-uuid"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 400
|
||
|
|
|
||
|
|
|
||
|
|
class TestBulkErrorPaths:
|
||
|
|
"""Error path tests for bulk operations."""
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_bulk_move_target_not_found(self, authed_client):
|
||
|
|
"""POST /files/bulk-move → 404 target folder not found."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/upload",
|
||
|
|
files={"file": ("f.pdf", PDF_CONTENT, "application/pdf")},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
file_id = resp.json()["id"]
|
||
|
|
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/bulk-move",
|
||
|
|
json={"file_ids": [file_id], "target_folder_id": str(uuid.uuid4())},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_bulk_move_to_root(self, authed_client):
|
||
|
|
"""POST /files/bulk-move → 200, move to root (null target)."""
|
||
|
|
client, _ = authed_client
|
||
|
|
# Create folder and upload file to it
|
||
|
|
resp = await client.post("/api/v1/dms/folders", json={"name": "Src"}, headers=ORIGIN_HEADER)
|
||
|
|
folder_id = resp.json()["id"]
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/upload",
|
||
|
|
files={"file": ("f.pdf", PDF_CONTENT, "application/pdf")},
|
||
|
|
data={"folder_id": folder_id},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
file_id = resp.json()["id"]
|
||
|
|
|
||
|
|
# Move to root (no target_folder_id)
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/bulk-move",
|
||
|
|
json={"file_ids": [file_id]},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["target_folder_id"] is None
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_bulk_move_invalid_file_id(self, authed_client):
|
||
|
|
"""POST /files/bulk-move → 400 invalid file_id."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/bulk-move",
|
||
|
|
json={"file_ids": ["not-a-uuid"]},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 400
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_bulk_delete_invalid_file_id(self, authed_client):
|
||
|
|
"""POST /files/bulk-delete → 400 invalid file_id."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/bulk-delete",
|
||
|
|
json={"file_ids": ["not-a-uuid"]},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 400
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_bulk_delete_nonexistent_files(self, authed_client):
|
||
|
|
"""POST /files/bulk-delete → 200, 0 deleted for nonexistent files."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/bulk-delete",
|
||
|
|
json={"file_ids": [str(uuid.uuid4())]},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["deleted"] == 0
|
||
|
|
|
||
|
|
|
||
|
|
class TestSearchEdgeCases:
|
||
|
|
"""Edge case tests for search endpoint."""
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_search_no_results(self, authed_client):
|
||
|
|
"""GET /search?q=nonexistent → 200 + empty list."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.get("/api/v1/dms/search?q=nonexistent", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json() == []
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_search_excludes_deleted(self, authed_client):
|
||
|
|
"""GET /search → excludes soft-deleted files."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/dms/files/upload",
|
||
|
|
files={"file": ("deleteme.pdf", PDF_CONTENT, "application/pdf")},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
file_id = resp.json()["id"]
|
||
|
|
await client.delete(f"/api/v1/dms/files/{file_id}", headers=ORIGIN_HEADER)
|
||
|
|
|
||
|
|
resp = await client.get("/api/v1/dms/search?q=deleteme", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json() == []
|
||
|
|
|
||
|
|
|
||
|
|
class TestSharedWithMeEdgeCases:
|
||
|
|
"""Edge case tests for shared-with-me endpoint."""
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_shared_with_me_empty(self, authed_client):
|
||
|
|
"""GET /shared-with-me → 200 + empty list for admin (no shares)."""
|
||
|
|
client, _ = authed_client
|
||
|
|
resp = await client.get("/api/v1/dms/shared-with-me", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json() == []
|