leocrm-bot
741 lines
23 KiB
Python
741 lines
23 KiB
Python
"""Tests for DMS plugin — folders, files, preview, OnlyOffice, sharing, search, bulk.
|
|
|
|
Tests all 19 acceptance criteria including AC14/AC15 which verify the existing
|
|
permissions plugin public share endpoints.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
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\n2 0 obj<</Type/Pages/Kids[3 0 R]/Count 1>>endobj\n3 0 obj<</Type/Page/MediaBox[0 0 612 792]>>endobj\nxref\n0 4\n0000000000 65535 f\n0000000009 00000 n\n0000000058 00000 n\n0000000115 00000 n\ntrailer<</Size 4/Root 1 0 R>>\nstartxref\n190\n%%EOF"
|
|
|
|
|
|
# ─── AC1: List folders (tree) ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac1_list_folders_empty(authed_client):
|
|
"""AC1: GET /api/v1/dms/folders → 200 + folder tree (empty)."""
|
|
client, _ = authed_client
|
|
resp = await client.get("/api/v1/dms/folders", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
assert resp.json() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac1_list_folders_tree(authed_client):
|
|
"""AC1: GET /api/v1/dms/folders → 200 + folder tree with nested children."""
|
|
client, _ = authed_client
|
|
# Create root folder
|
|
resp = await client.post(
|
|
"/api/v1/dms/folders",
|
|
json={"name": "Root"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
root_id = resp.json()["id"]
|
|
|
|
# Create child folder
|
|
resp = await client.post(
|
|
"/api/v1/dms/folders",
|
|
json={"name": "Child", "parent_id": root_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
|
|
# Get tree
|
|
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"] == "Root"
|
|
assert len(tree[0]["children"]) == 1
|
|
assert tree[0]["children"][0]["name"] == "Child"
|
|
|
|
|
|
# ─── AC2: Create folder ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac2_create_folder(authed_client):
|
|
"""AC2: POST /api/v1/dms/folders → 201, folder created with path."""
|
|
client, _ = authed_client
|
|
resp = await client.post(
|
|
"/api/v1/dms/folders",
|
|
json={"name": "Documents"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["name"] == "Documents"
|
|
assert data["path"] == "Documents"
|
|
assert data["parent_id"] is None
|
|
assert "id" in data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac2_create_folder_with_parent(authed_client):
|
|
"""AC2: POST /api/v1/dms/folders → 201, nested folder with full path."""
|
|
client, _ = authed_client
|
|
resp = await client.post(
|
|
"/api/v1/dms/folders",
|
|
json={"name": "Parent"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
parent_id = resp.json()["id"]
|
|
|
|
resp = await client.post(
|
|
"/api/v1/dms/folders",
|
|
json={"name": "Sub", "parent_id": parent_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["name"] == "Sub"
|
|
assert data["path"] == "Parent/Sub"
|
|
assert data["parent_id"] == parent_id
|
|
|
|
|
|
# ─── AC3: Update folder (rename/move) ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac3_rename_folder(authed_client):
|
|
"""AC3: PATCH /api/v1/dms/folders/{id} → 200, renamed."""
|
|
client, _ = authed_client
|
|
resp = await client.post(
|
|
"/api/v1/dms/folders",
|
|
json={"name": "OldName"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
folder_id = resp.json()["id"]
|
|
|
|
resp = await client.patch(
|
|
f"/api/v1/dms/folders/{folder_id}",
|
|
json={"name": "NewName"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["name"] == "NewName"
|
|
assert resp.json()["path"] == "NewName"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac3_move_folder(authed_client):
|
|
"""AC3: PATCH /api/v1/dms/folders/{id} → 200, moved."""
|
|
client, _ = authed_client
|
|
# Create two root folders
|
|
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"]
|
|
|
|
# Move B into A
|
|
resp = await client.patch(
|
|
f"/api/v1/dms/folders/{b_id}",
|
|
json={"parent_id": a_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["parent_id"] == a_id
|
|
assert resp.json()["path"] == "A/B"
|
|
|
|
|
|
# ─── AC4: Delete folder (soft-delete, cascade) ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac4_delete_folder_cascade(authed_client):
|
|
"""AC4: DELETE /api/v1/dms/folders/{id} → 204, soft-delete with cascade."""
|
|
client, _ = authed_client
|
|
# Create parent with child and file in child
|
|
resp = await client.post("/api/v1/dms/folders", json={"name": "Parent"}, headers=ORIGIN_HEADER)
|
|
parent_id = resp.json()["id"]
|
|
resp = await client.post(
|
|
"/api/v1/dms/folders",
|
|
json={"name": "Child", "parent_id": parent_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
child_id = resp.json()["id"]
|
|
|
|
# Upload a file into child folder
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("test.pdf", PDF_CONTENT, "application/pdf")},
|
|
data={"folder_id": child_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
file_id = resp.json()["id"]
|
|
|
|
# Delete parent folder
|
|
resp = await client.delete(f"/api/v1/dms/folders/{parent_id}", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 204
|
|
|
|
# Verify parent folder is gone from tree
|
|
resp = await client.get("/api/v1/dms/folders", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
assert resp.json() == []
|
|
|
|
# Verify file is soft-deleted (GET should 404)
|
|
resp = await client.get(f"/api/v1/dms/files/{file_id}", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# ─── AC5: Upload file (multipart) ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac5_upload_file(authed_client):
|
|
"""AC5: POST /api/v1/dms/files/upload → 201, file stored + metadata."""
|
|
client, _ = authed_client
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("doc.pdf", PDF_CONTENT, "application/pdf")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["name"] == "doc.pdf"
|
|
assert data["mime_type"] == "application/pdf"
|
|
assert data["size_bytes"] == len(PDF_CONTENT)
|
|
assert data["folder_id"] is None
|
|
assert "id" in data
|
|
assert "storage_path" in data
|
|
|
|
# Verify file exists on disk
|
|
assert os.path.exists(data["storage_path"])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac5_upload_file_to_folder(authed_client):
|
|
"""AC5: POST /api/v1/dms/files/upload → 201, file stored in folder."""
|
|
client, _ = authed_client
|
|
# Create folder
|
|
resp = await client.post("/api/v1/dms/folders", json={"name": "Docs"}, headers=ORIGIN_HEADER)
|
|
folder_id = resp.json()["id"]
|
|
|
|
# Upload file to folder
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("doc.pdf", PDF_CONTENT, "application/pdf")},
|
|
data={"folder_id": folder_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
assert resp.json()["folder_id"] == folder_id
|
|
|
|
|
|
# ─── AC6: Get file metadata ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac6_get_file_metadata(authed_client):
|
|
"""AC6: GET /api/v1/dms/files/{id} → 200 + file metadata."""
|
|
client, _ = authed_client
|
|
# Upload file first
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("report.pdf", PDF_CONTENT, "application/pdf")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
file_id = resp.json()["id"]
|
|
|
|
# Get metadata
|
|
resp = await client.get(f"/api/v1/dms/files/{file_id}", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["id"] == file_id
|
|
assert data["name"] == "report.pdf"
|
|
assert data["mime_type"] == "application/pdf"
|
|
assert data["size_bytes"] == len(PDF_CONTENT)
|
|
|
|
|
|
# ─── AC7: Rename/move file ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac7_rename_file(authed_client):
|
|
"""AC7: PATCH /api/v1/dms/files/{id} → 200, renamed."""
|
|
client, _ = authed_client
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("old.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={"name": "new.pdf"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["name"] == "new.pdf"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac7_move_file(authed_client):
|
|
"""AC7: PATCH /api/v1/dms/files/{id} → 200, moved to folder."""
|
|
client, _ = authed_client
|
|
# Upload file at root
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("file.pdf", PDF_CONTENT, "application/pdf")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
file_id = resp.json()["id"]
|
|
|
|
# Create folder
|
|
resp = await client.post("/api/v1/dms/folders", json={"name": "Archive"}, headers=ORIGIN_HEADER)
|
|
folder_id = resp.json()["id"]
|
|
|
|
# Move file
|
|
resp = await client.patch(
|
|
f"/api/v1/dms/files/{file_id}",
|
|
json={"folder_id": folder_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["folder_id"] == folder_id
|
|
|
|
|
|
# ─── AC8: Delete file (soft-delete) ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac8_delete_file(authed_client):
|
|
"""AC8: DELETE /api/v1/dms/files/{id} → 204, soft-delete."""
|
|
client, _ = authed_client
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("delete.pdf", PDF_CONTENT, "application/pdf")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
file_id = resp.json()["id"]
|
|
|
|
resp = await client.delete(f"/api/v1/dms/files/{file_id}", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 204
|
|
|
|
# Verify file is gone from list
|
|
resp = await client.get(f"/api/v1/dms/files/{file_id}", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# ─── AC9: Restore file ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac9_restore_file(authed_client):
|
|
"""AC9: POST /api/v1/dms/files/{id}/restore → 200, restored from trash."""
|
|
client, _ = authed_client
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("restore.pdf", PDF_CONTENT, "application/pdf")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
file_id = resp.json()["id"]
|
|
|
|
# Delete file
|
|
resp = await client.delete(f"/api/v1/dms/files/{file_id}", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 204
|
|
|
|
# Restore file
|
|
resp = await client.post(f"/api/v1/dms/files/{file_id}/restore", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["id"] == file_id
|
|
assert data["deleted_at"] is None
|
|
|
|
# Verify file is accessible again
|
|
resp = await client.get(f"/api/v1/dms/files/{file_id}", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
|
|
|
|
# ─── AC10: PDF preview ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac10_pdf_preview(authed_client):
|
|
"""AC10: GET /api/v1/dms/files/{id}/preview → 200 + PDF stream."""
|
|
client, _ = authed_client
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("preview.pdf", PDF_CONTENT, "application/pdf")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
file_id = resp.json()["id"]
|
|
|
|
resp = await client.get(f"/api/v1/dms/files/{file_id}/preview", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
assert resp.headers["content-type"] == "application/pdf"
|
|
assert resp.content == PDF_CONTENT
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac10_non_pdf_preview_400(authed_client):
|
|
"""AC10: GET /api/v1/dms/files/{id}/preview → 400 for non-PDF files."""
|
|
client, _ = authed_client
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("doc.txt", b"hello world", "text/plain")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
file_id = resp.json()["id"]
|
|
|
|
resp = await client.get(f"/api/v1/dms/files/{file_id}/preview", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
# ─── AC11: OnlyOffice edit session ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac11_edit_session_docx(authed_client):
|
|
"""AC11: POST /api/v1/dms/files/{id}/edit-session → 200 + OnlyOffice config."""
|
|
client, _ = authed_client
|
|
docx_content = b"PK\x03\x04" + b"\x00" * 100 # minimal zip-like header
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={
|
|
"file": (
|
|
"report.docx",
|
|
docx_content,
|
|
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
)
|
|
},
|
|
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
|
|
config = resp.json()
|
|
assert "document" in config
|
|
assert "editorConfig" in config
|
|
assert config["document"]["fileType"] == "docx"
|
|
assert config["document"]["title"] == "report.docx"
|
|
assert "key" in config["document"]
|
|
assert "url" in config["document"]
|
|
assert config["editorConfig"]["mode"] == "edit"
|
|
assert "callbackUrl" in config["editorConfig"]
|
|
assert "user" in config["editorConfig"]
|
|
assert "id" in config["editorConfig"]["user"]
|
|
assert "name" in config["editorConfig"]["user"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac11_edit_session_non_office_400(authed_client):
|
|
"""AC11: POST /api/v1/dms/files/{id}/edit-session → 400 for non-Office files."""
|
|
client, _ = authed_client
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("image.png", b"\x89PNG", "image/png")},
|
|
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 == 400
|
|
|
|
|
|
# ─── AC12: Internal share ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac12_internal_share(authed_client):
|
|
"""AC12: POST /api/v1/dms/files/{id}/share → 200, internal share created."""
|
|
client, seed = authed_client
|
|
# Upload file
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("shared.pdf", PDF_CONTENT, "application/pdf")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
file_id = resp.json()["id"]
|
|
|
|
# Share with viewer_a
|
|
viewer_id = str(seed["viewer_a"].id)
|
|
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
|
|
data = resp.json()
|
|
assert data["file_id"] == file_id
|
|
assert data["count"] == 1
|
|
assert data["shared_with"][0]["user_id"] == viewer_id
|
|
assert data["shared_with"][0]["access_level"] == "read"
|
|
|
|
|
|
# ─── AC13: Remove share ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac13_remove_share(authed_client):
|
|
"""AC13: DELETE /api/v1/dms/files/{id}/share → 204, share removed."""
|
|
client, seed = authed_client
|
|
# Upload file
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("unshare.pdf", PDF_CONTENT, "application/pdf")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
file_id = resp.json()["id"]
|
|
|
|
# Share with viewer_a
|
|
viewer_id = str(seed["viewer_a"].id)
|
|
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
|
|
|
|
# Remove share
|
|
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
|
|
|
|
|
|
# ─── AC14: Public share access (no auth) — via T11 permissions plugin ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac14_public_share_access(authed_client):
|
|
"""AC14: GET /api/public/share/{token} → 200 (no auth, public access)."""
|
|
client, _ = authed_client
|
|
# Upload file
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("public.pdf", PDF_CONTENT, "application/pdf")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
file_id = resp.json()["id"]
|
|
|
|
# Create share link (no password)
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/share-link",
|
|
json={},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
token = resp.json()["token"]
|
|
|
|
# Access publicly without auth
|
|
resp = await client.get(f"/api/public/share/{token}")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["file_id"] == file_id
|
|
|
|
|
|
# ─── AC15: Public share with password → 401 without password ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac15_public_share_password_required(authed_client):
|
|
"""AC15: GET /api/public/share/{token} with password → 401 without password."""
|
|
client, _ = authed_client
|
|
# Upload file
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("protected.pdf", PDF_CONTENT, "application/pdf")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
file_id = resp.json()["id"]
|
|
|
|
# Create share link WITH password
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/share-link",
|
|
json={"password": "Secret123"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
token = resp.json()["token"]
|
|
|
|
# Access without password → 401
|
|
resp = await client.get(f"/api/public/share/{token}")
|
|
assert resp.status_code == 401
|
|
assert resp.json()["detail"]["code"] == "password_required"
|
|
|
|
# Access WITH password via POST → 200
|
|
resp = await client.post(
|
|
f"/api/public/share/{token}",
|
|
json={"password": "Secret123"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["file_id"] == file_id
|
|
|
|
|
|
# ─── AC16: Search files ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac16_search_files(authed_client):
|
|
"""AC16: GET /api/v1/dms/search?q=text → 200 + matching files (ILIKE)."""
|
|
client, _ = authed_client
|
|
# Upload multiple files
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("invoice_2024.pdf", PDF_CONTENT, "application/pdf")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("report_2024.pdf", PDF_CONTENT, "application/pdf")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("photo.png", b"\x89PNG", "image/png")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
|
|
# Search for "2024"
|
|
resp = await client.get("/api/v1/dms/search?q=2024", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
results = resp.json()
|
|
assert len(results) == 2
|
|
names = {r["name"] for r in results}
|
|
assert "invoice_2024.pdf" in names
|
|
assert "report_2024.pdf" in names
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac16_search_case_insensitive(authed_client):
|
|
"""AC16: GET /api/v1/dms/search?q=INVOICE → 200, case-insensitive match."""
|
|
client, _ = authed_client
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("Invoice.pdf", PDF_CONTENT, "application/pdf")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
|
|
resp = await client.get("/api/v1/dms/search?q=invoice", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()) == 1
|
|
assert resp.json()[0]["name"] == "Invoice.pdf"
|
|
|
|
|
|
# ─── AC17: Shared-with-me ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac17_shared_with_me(authed_client, dms_client, db_session):
|
|
"""AC17: GET /api/v1/dms/shared-with-me → 200 + shared files list."""
|
|
client, seed = authed_client
|
|
# Upload file as admin
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": ("shared_doc.pdf", PDF_CONTENT, "application/pdf")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
file_id = resp.json()["id"]
|
|
|
|
# Share with viewer_a
|
|
viewer_id = str(seed["viewer_a"].id)
|
|
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
|
|
|
|
# Login as viewer_a and check shared-with-me
|
|
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]["id"] == file_id
|
|
assert shared[0]["name"] == "shared_doc.pdf"
|
|
assert shared[0]["access_level"] == "read"
|
|
|
|
|
|
# ─── AC18: Bulk move ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac18_bulk_move(authed_client):
|
|
"""AC18: POST /api/v1/dms/files/bulk-move → 200, files moved."""
|
|
client, _ = authed_client
|
|
# Upload 3 files at root
|
|
file_ids = []
|
|
for i in range(3):
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": (f"file{i}.pdf", PDF_CONTENT, "application/pdf")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
file_ids.append(resp.json()["id"])
|
|
|
|
# Create target folder
|
|
resp = await client.post(
|
|
"/api/v1/dms/folders", json={"name": "BulkTarget"}, headers=ORIGIN_HEADER
|
|
)
|
|
folder_id = resp.json()["id"]
|
|
|
|
# Bulk move
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/bulk-move",
|
|
json={"file_ids": file_ids, "target_folder_id": folder_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["moved"] == 3
|
|
assert data["target_folder_id"] == folder_id
|
|
|
|
# Verify files are in folder
|
|
for fid in file_ids:
|
|
resp = await client.get(f"/api/v1/dms/files/{fid}", headers=ORIGIN_HEADER)
|
|
assert resp.json()["folder_id"] == folder_id
|
|
|
|
|
|
# ─── AC19: Bulk delete ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac19_bulk_delete(authed_client):
|
|
"""AC19: POST /api/v1/dms/files/bulk-delete → 200, files soft-deleted."""
|
|
client, _ = authed_client
|
|
# Upload 3 files
|
|
file_ids = []
|
|
for i in range(3):
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/upload",
|
|
files={"file": (f"del{i}.pdf", PDF_CONTENT, "application/pdf")},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
file_ids.append(resp.json()["id"])
|
|
|
|
# Bulk delete
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/bulk-delete",
|
|
json={"file_ids": file_ids},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["deleted"] == 3
|
|
|
|
# Verify files are soft-deleted
|
|
for fid in file_ids:
|
|
resp = await client.get(f"/api/v1/dms/files/{fid}", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 404
|