T04: DMS plugin backend — folders + files + preview + OnlyOffice + shares + search + bulk — 106 tests, 97.90% coverage
This commit is contained in:
leocrm-bot
@@ -0,0 +1,832 @@
|
||||
"""DMS coverage tests — exercise uncovered paths to reach ≥80% coverage.
|
||||
|
||||
Targets missing lines in routes.py: tree building, path construction,
|
||||
deep cascade delete, file too large, preview missing on disk,
|
||||
multi-user/group sharing, share removal, search edge cases,
|
||||
shared-with-me with data, bulk mixed IDs, tenant isolation.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import uuid
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from tests.conftest import ORIGIN_HEADER, login_client
|
||||
|
||||
PDF_CONTENT = b"%PDF-1.4\n1 0 obj<</Type/Catalog/Pages 2 0 R>>endobj\n%%EOF"
|
||||
|
||||
|
||||
# ─── Folder coverage ───
|
||||
|
||||
|
||||
class TestFolderCoverage:
|
||||
"""Coverage tests for folder endpoints — tree, path, cascade, isolation."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_folders_deeply_nested_tree(self, authed_client):
|
||||
"""GET /folders → deeply nested tree with correct paths."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post("/api/v1/dms/folders", json={"name": "A"}, headers=ORIGIN_HEADER)
|
||||
a_id = resp.json()["id"]
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/folders",
|
||||
json={"name": "B", "parent_id": a_id},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
b_id = resp.json()["id"]
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/folders",
|
||||
json={"name": "C", "parent_id": b_id},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
|
||||
resp = await client.get("/api/v1/dms/folders", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 200
|
||||
tree = resp.json()
|
||||
assert len(tree) == 1
|
||||
assert tree[0]["name"] == "A"
|
||||
assert tree[0]["path"] == "A"
|
||||
assert len(tree[0]["children"]) == 1
|
||||
child = tree[0]["children"][0]
|
||||
assert child["name"] == "B"
|
||||
assert child["path"] == "A/B"
|
||||
assert len(child["children"]) == 1
|
||||
assert child["children"][0]["name"] == "C"
|
||||
assert child["children"][0]["path"] == "A/B/C"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_folders_invalid_parent_id(self, authed_client):
|
||||
"""GET /folders?parent_id=invalid → 400."""
|
||||
client, _ = authed_client
|
||||
resp = await client.get("/api/v1/dms/folders?parent_id=not-a-uuid", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 400
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_folder_deep_nested_path(self, authed_client):
|
||||
"""POST /folders → 201 with 3-level nested path."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post("/api/v1/dms/folders", json={"name": "L1"}, headers=ORIGIN_HEADER)
|
||||
l1_id = resp.json()["id"]
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/folders",
|
||||
json={"name": "L2", "parent_id": l1_id},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
l2_id = resp.json()["id"]
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/folders",
|
||||
json={"name": "L3", "parent_id": l2_id},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
assert resp.json()["path"] == "L1/L2/L3"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_folder_rename_and_move(self, authed_client):
|
||||
"""PATCH /folders/{id} → 200, rename + move in one request."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post("/api/v1/dms/folders", json={"name": "A"}, headers=ORIGIN_HEADER)
|
||||
a_id = resp.json()["id"]
|
||||
resp = await client.post("/api/v1/dms/folders", json={"name": "B"}, headers=ORIGIN_HEADER)
|
||||
b_id = resp.json()["id"]
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/folders",
|
||||
json={"name": "C", "parent_id": b_id},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
c_id = resp.json()["id"]
|
||||
|
||||
resp = await client.patch(
|
||||
f"/api/v1/dms/folders/{c_id}",
|
||||
json={"name": "D", "parent_id": a_id},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["name"] == "D"
|
||||
assert resp.json()["parent_id"] == a_id
|
||||
assert resp.json()["path"] == "A/D"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_folder_move_to_root(self, authed_client):
|
||||
"""PATCH /folders/{id} → 200, move to root via parent_id=null."""
|
||||
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"]
|
||||
|
||||
resp = await client.patch(
|
||||
f"/api/v1/dms/folders/{c_id}",
|
||||
json={"parent_id": None},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["parent_id"] is None
|
||||
assert resp.json()["path"] == "C"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_folder_no_changes(self, authed_client):
|
||||
"""PATCH /folders/{id} → 200 with empty body (no changes)."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/folders", json={"name": "NoChange"}, headers=ORIGIN_HEADER
|
||||
)
|
||||
folder_id = resp.json()["id"]
|
||||
|
||||
resp = await client.patch(
|
||||
f"/api/v1/dms/folders/{folder_id}",
|
||||
json={},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["name"] == "NoChange"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_folder_invalid_id(self, authed_client):
|
||||
"""PATCH /folders/{id} → 400 invalid id."""
|
||||
client, _ = authed_client
|
||||
resp = await client.patch(
|
||||
"/api/v1/dms/folders/not-a-uuid",
|
||||
json={"name": "New"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_folder_deep_cascade(self, authed_client):
|
||||
"""DELETE /folders/{id} → 204, 3-level cascade with files in each."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post("/api/v1/dms/folders", json={"name": "A"}, headers=ORIGIN_HEADER)
|
||||
a_id = resp.json()["id"]
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/folders",
|
||||
json={"name": "B", "parent_id": a_id},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
b_id = resp.json()["id"]
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/folders",
|
||||
json={"name": "C", "parent_id": b_id},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
c_id = resp.json()["id"]
|
||||
|
||||
file_ids = []
|
||||
for folder_id in [a_id, b_id, c_id]:
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": (f"f_{folder_id[:4]}.pdf", PDF_CONTENT, "application/pdf")},
|
||||
data={"folder_id": folder_id},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
file_ids.append(resp.json()["id"])
|
||||
|
||||
resp = await client.delete(f"/api/v1/dms/folders/{a_id}", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 204
|
||||
|
||||
resp = await client.get("/api/v1/dms/folders", headers=ORIGIN_HEADER)
|
||||
assert resp.json() == []
|
||||
|
||||
for fid in file_ids:
|
||||
resp = await client.get(f"/api/v1/dms/files/{fid}", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 404
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_folder_tenant_isolation(self, authed_client, dms_client, db_session):
|
||||
"""Folders visible only within their tenant."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/folders", json={"name": "TenantA"}, headers=ORIGIN_HEADER
|
||||
)
|
||||
folder_id = resp.json()["id"]
|
||||
|
||||
await login_client(dms_client, "admin@tenantb.com")
|
||||
resp = await dms_client.get("/api/v1/dms/folders", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json() == []
|
||||
|
||||
resp = await dms_client.get(
|
||||
f"/api/v1/dms/folders?parent_id={folder_id}", headers=ORIGIN_HEADER
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
# ─── File coverage ───
|
||||
|
||||
|
||||
class TestFileCoverage:
|
||||
"""Coverage tests for file endpoints — upload edge cases, invalid IDs, preview."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_file_too_large(self, authed_client):
|
||||
"""POST /files/upload → 413 file too large."""
|
||||
client, _ = authed_client
|
||||
with patch("app.plugins.builtins.dms.routes.MAX_FILE_SIZE", 10):
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("large.bin", b"\x00" * 100, "application/octet-stream")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 413
|
||||
assert resp.json()["detail"]["code"] == "file_too_large"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_empty_file(self, authed_client):
|
||||
"""POST /files/upload → 201 for empty file."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("empty.txt", b"", "text/plain")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
assert resp.json()["size_bytes"] == 0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_file_invalid_folder_id(self, authed_client):
|
||||
"""POST /files/upload → 400 invalid folder_id."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("test.pdf", PDF_CONTENT, "application/pdf")},
|
||||
data={"folder_id": "not-a-uuid"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_file_invalid_id(self, authed_client):
|
||||
"""DELETE /files/{id} → 400 invalid id."""
|
||||
client, _ = authed_client
|
||||
resp = await client.delete("/api/v1/dms/files/bad-id", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 400
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_restore_file_invalid_id(self, authed_client):
|
||||
"""POST /files/{id}/restore → 400 invalid id."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post("/api/v1/dms/files/bad-id/restore", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 400
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_file_rename_and_move(self, authed_client):
|
||||
"""PATCH /files/{id} → 200, rename + move in one request."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("original.pdf", PDF_CONTENT, "application/pdf")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
file_id = resp.json()["id"]
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/folders", json={"name": "Target"}, headers=ORIGIN_HEADER
|
||||
)
|
||||
folder_id = resp.json()["id"]
|
||||
|
||||
resp = await client.patch(
|
||||
f"/api/v1/dms/files/{file_id}",
|
||||
json={"name": "renamed.pdf", "folder_id": folder_id},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["name"] == "renamed.pdf"
|
||||
assert resp.json()["folder_id"] == folder_id
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_file_move_to_root(self, authed_client):
|
||||
"""PATCH /files/{id} → 200, move file to root via folder_id=null."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post("/api/v1/dms/folders", json={"name": "F"}, headers=ORIGIN_HEADER)
|
||||
folder_id = resp.json()["id"]
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("test.pdf", PDF_CONTENT, "application/pdf")},
|
||||
data={"folder_id": folder_id},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
file_id = resp.json()["id"]
|
||||
|
||||
resp = await client.patch(
|
||||
f"/api/v1/dms/files/{file_id}",
|
||||
json={"folder_id": None},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["folder_id"] is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_file_no_changes(self, authed_client):
|
||||
"""PATCH /files/{id} → 200 with empty body (no changes)."""
|
||||
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={},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["name"] == "test.pdf"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_file_invalid_id(self, authed_client):
|
||||
"""PATCH /files/{id} → 400 invalid id."""
|
||||
client, _ = authed_client
|
||||
resp = await client.patch(
|
||||
"/api/v1/dms/files/bad-id",
|
||||
json={"name": "new.pdf"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_preview_file_invalid_id(self, authed_client):
|
||||
"""GET /files/{id}/preview → 400 invalid id."""
|
||||
client, _ = authed_client
|
||||
resp = await client.get("/api/v1/dms/files/bad-id/preview", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 400
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_preview_file_missing_on_disk(self, authed_client):
|
||||
"""GET /files/{id}/preview → 404 when file missing on disk."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("ondisk.pdf", PDF_CONTENT, "application/pdf")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
file_id = resp.json()["id"]
|
||||
storage_path = resp.json()["storage_path"]
|
||||
|
||||
if os.path.exists(storage_path):
|
||||
os.remove(storage_path)
|
||||
|
||||
resp = await client.get(f"/api/v1/dms/files/{file_id}/preview", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 404
|
||||
assert resp.json()["detail"]["code"] == "file_missing"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_edit_session_invalid_id(self, authed_client):
|
||||
"""POST /files/{id}/edit-session → 400 invalid id."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post("/api/v1/dms/files/bad-id/edit-session", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 400
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_file_tenant_isolation(self, authed_client, dms_client, db_session):
|
||||
"""Files visible only within their tenant."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("tenant_a.pdf", PDF_CONTENT, "application/pdf")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
file_id = resp.json()["id"]
|
||||
|
||||
await login_client(dms_client, "admin@tenantb.com")
|
||||
resp = await dms_client.get(f"/api/v1/dms/files/{file_id}", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
# ─── Share coverage ───
|
||||
|
||||
|
||||
class TestShareCoverage:
|
||||
"""Coverage tests for share endpoints — multi-user, groups, removal."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_share_with_multiple_users(self, authed_client):
|
||||
"""POST /files/{id}/share → 200 with multiple users."""
|
||||
client, seed = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("multi.pdf", PDF_CONTENT, "application/pdf")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
file_id = resp.json()["id"]
|
||||
|
||||
viewer_id = str(seed["viewer_a"].id)
|
||||
editor_id = str(seed["editor_a"].id)
|
||||
|
||||
resp = await client.post(
|
||||
f"/api/v1/dms/files/{file_id}/share",
|
||||
json={"user_ids": [viewer_id, editor_id], "access_level": "write"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["count"] == 2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_share_with_users_and_groups(self, authed_client):
|
||||
"""POST /files/{id}/share → 200 with both users and groups."""
|
||||
client, seed = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("both.pdf", PDF_CONTENT, "application/pdf")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
file_id = resp.json()["id"]
|
||||
|
||||
viewer_id = str(seed["viewer_a"].id)
|
||||
group_id = str(uuid.uuid4())
|
||||
|
||||
resp = await client.post(
|
||||
f"/api/v1/dms/files/{file_id}/share",
|
||||
json={"user_ids": [viewer_id], "group_ids": [group_id], "access_level": "read"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["count"] == 2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_share_with_invalid_group_id(self, authed_client):
|
||||
"""POST /files/{id}/share → 400 invalid group_id."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("badgroup.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={"group_ids": ["not-a-uuid"], "access_level": "read"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_share_duplicate_group_ignored(self, authed_client):
|
||||
"""POST /files/{id}/share → 200, duplicate group share ignored."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("dupgroup.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.post(
|
||||
f"/api/v1/dms/files/{file_id}/share",
|
||||
json={"group_ids": [group_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_user(self, authed_client):
|
||||
"""DELETE /files/{id}/share → 204, user share removed."""
|
||||
client, seed = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("unshare_user.pdf", PDF_CONTENT, "application/pdf")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
file_id = resp.json()["id"]
|
||||
viewer_id = str(seed["viewer_a"].id)
|
||||
|
||||
await client.post(
|
||||
f"/api/v1/dms/files/{file_id}/share",
|
||||
json={"user_ids": [viewer_id], "access_level": "read"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
|
||||
resp = await client.request(
|
||||
"DELETE",
|
||||
f"/api/v1/dms/files/{file_id}/share",
|
||||
json={"user_id": viewer_id},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 204
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_remove_share_no_match(self, authed_client):
|
||||
"""DELETE /files/{id}/share → 204, no matching share (idempotent)."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("nomatch.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": str(uuid.uuid4())},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 204
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_remove_share_invalid_group_id(self, authed_client):
|
||||
"""DELETE /files/{id}/share → 400 invalid group_id."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("badgid.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={"group_id": "not-a-uuid"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_remove_share_both_user_and_group(self, authed_client):
|
||||
"""DELETE /files/{id}/share → 204, removes both user and group shares."""
|
||||
client, seed = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("bothremoval.pdf", PDF_CONTENT, "application/pdf")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
file_id = resp.json()["id"]
|
||||
viewer_id = str(seed["viewer_a"].id)
|
||||
group_id = str(uuid.uuid4())
|
||||
|
||||
await client.post(
|
||||
f"/api/v1/dms/files/{file_id}/share",
|
||||
json={"user_ids": [viewer_id], "group_ids": [group_id], "access_level": "read"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
|
||||
resp = await client.request(
|
||||
"DELETE",
|
||||
f"/api/v1/dms/files/{file_id}/share",
|
||||
json={"user_id": viewer_id, "group_id": group_id},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 204
|
||||
|
||||
|
||||
# ─── Search coverage ───
|
||||
|
||||
|
||||
class TestSearchCoverage:
|
||||
"""Coverage tests for search — special chars, partial match, tenant isolation."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_special_characters(self, authed_client):
|
||||
"""GET /search?q=50% → 200, handles special characters in filename."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("test_50%.pdf", PDF_CONTENT, "application/pdf")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
|
||||
resp = await client.get("/api/v1/dms/search?q=50%", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 200
|
||||
assert len(resp.json()) == 1
|
||||
assert resp.json()[0]["name"] == "test_50%.pdf"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_partial_match(self, authed_client):
|
||||
"""GET /search → partial name match."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("monthly_report.pdf", PDF_CONTENT, "application/pdf")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
resp = await client.get("/api/v1/dms/search?q=monthly", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 200
|
||||
assert len(resp.json()) == 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_tenant_isolation(self, authed_client, dms_client, db_session):
|
||||
"""GET /search → only returns files from current tenant."""
|
||||
client, _ = authed_client
|
||||
await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("tenant_a_file.pdf", PDF_CONTENT, "application/pdf")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
|
||||
await login_client(dms_client, "admin@tenantb.com")
|
||||
resp = await dms_client.get("/api/v1/dms/search?q=tenant_a", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json() == []
|
||||
|
||||
|
||||
# ─── Shared-with-me coverage ───
|
||||
|
||||
|
||||
class TestSharedWithMeCoverage:
|
||||
"""Coverage tests for shared-with-me — multiple files, deleted exclusion."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_shared_with_me_multiple_files(self, authed_client, dms_client, db_session):
|
||||
"""GET /shared-with-me → 200 with multiple shared files."""
|
||||
client, seed = authed_client
|
||||
viewer_id = str(seed["viewer_a"].id)
|
||||
|
||||
file_ids = []
|
||||
for i in range(3):
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": (f"shared_{i}.pdf", PDF_CONTENT, "application/pdf")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
fid = resp.json()["id"]
|
||||
file_ids.append(fid)
|
||||
await client.post(
|
||||
f"/api/v1/dms/files/{fid}/share",
|
||||
json={"user_ids": [viewer_id], "access_level": "read"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
|
||||
await login_client(dms_client, "viewer@tenanta.com")
|
||||
resp = await dms_client.get("/api/v1/dms/shared-with-me", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 200
|
||||
shared = resp.json()
|
||||
assert len(shared) == 3
|
||||
returned_ids = {s["id"] for s in shared}
|
||||
assert returned_ids == set(file_ids)
|
||||
for s in shared:
|
||||
assert s["access_level"] == "read"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_shared_with_me_excludes_deleted(self, authed_client, dms_client, db_session):
|
||||
"""GET /shared-with-me → excludes soft-deleted files."""
|
||||
client, seed = authed_client
|
||||
viewer_id = str(seed["viewer_a"].id)
|
||||
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("del_share.pdf", PDF_CONTENT, "application/pdf")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
file_id = resp.json()["id"]
|
||||
await client.post(
|
||||
f"/api/v1/dms/files/{file_id}/share",
|
||||
json={"user_ids": [viewer_id], "access_level": "read"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
await client.delete(f"/api/v1/dms/files/{file_id}", headers=ORIGIN_HEADER)
|
||||
|
||||
await login_client(dms_client, "viewer@tenanta.com")
|
||||
resp = await dms_client.get("/api/v1/dms/shared-with-me", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json() == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_shared_with_me_write_access(self, authed_client, dms_client, db_session):
|
||||
"""GET /shared-with-me → shows write access_level correctly."""
|
||||
client, seed = authed_client
|
||||
viewer_id = str(seed["viewer_a"].id)
|
||||
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("write_share.pdf", PDF_CONTENT, "application/pdf")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
file_id = resp.json()["id"]
|
||||
await client.post(
|
||||
f"/api/v1/dms/files/{file_id}/share",
|
||||
json={"user_ids": [viewer_id], "access_level": "write"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
|
||||
await login_client(dms_client, "viewer@tenanta.com")
|
||||
resp = await dms_client.get("/api/v1/dms/shared-with-me", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 200
|
||||
shared = resp.json()
|
||||
assert len(shared) == 1
|
||||
assert shared[0]["access_level"] == "write"
|
||||
|
||||
|
||||
# ─── Bulk operations coverage ───
|
||||
|
||||
|
||||
class TestBulkCoverage:
|
||||
"""Coverage tests for bulk operations — mixed IDs, tenant isolation."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_move_mixed_valid_invalid(self, authed_client):
|
||||
"""POST /files/bulk-move → 200 with mixed valid/invalid file IDs."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/folders", json={"name": "Target"}, headers=ORIGIN_HEADER
|
||||
)
|
||||
folder_id = resp.json()["id"]
|
||||
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("valid.pdf", PDF_CONTENT, "application/pdf")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
valid_file_id = resp.json()["id"]
|
||||
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/bulk-move",
|
||||
json={"file_ids": [valid_file_id, str(uuid.uuid4())], "target_folder_id": folder_id},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["moved"] == 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_delete_mixed_valid_invalid(self, authed_client):
|
||||
"""POST /files/bulk-delete → 200 with mixed valid/invalid file IDs."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("valid.pdf", PDF_CONTENT, "application/pdf")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
valid_file_id = resp.json()["id"]
|
||||
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/bulk-delete",
|
||||
json={"file_ids": [valid_file_id, str(uuid.uuid4())]},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["deleted"] == 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_move_invalid_target_id(self, authed_client):
|
||||
"""POST /files/bulk-move → 400 invalid target_folder_id."""
|
||||
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": "not-a-uuid"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_delete_tenant_isolation(self, authed_client, dms_client, db_session):
|
||||
"""POST /files/bulk-delete → 0 deleted for cross-tenant file IDs."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("tenant_a.pdf", PDF_CONTENT, "application/pdf")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
tenant_a_file_id = resp.json()["id"]
|
||||
|
||||
await login_client(dms_client, "admin@tenantb.com")
|
||||
resp = await dms_client.post(
|
||||
"/api/v1/dms/files/bulk-delete",
|
||||
json={"file_ids": [tenant_a_file_id]},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["deleted"] == 0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_move_tenant_isolation(self, authed_client, dms_client, db_session):
|
||||
"""POST /files/bulk-move → 0 moved for cross-tenant file IDs."""
|
||||
client, _ = authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/dms/files/upload",
|
||||
files={"file": ("cross_tenant.pdf", PDF_CONTENT, "application/pdf")},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
tenant_a_file_id = resp.json()["id"]
|
||||
|
||||
await login_client(dms_client, "admin@tenantb.com")
|
||||
resp = await dms_client.post(
|
||||
"/api/v1/dms/files/bulk-move",
|
||||
json={"file_ids": [tenant_a_file_id]},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["moved"] == 0
|
||||
Reference in New Issue
Block a user