test(phase-04): fill Nyquist validation gaps — FOLD-04, FOLD-05, SEC-08, SEC-09

Add 6 new tests covering document sort (name/size), FTS search cross-user
isolation, credentials_enc exclusion from all responses, and MinIO object
cleanup on user deletion.

Fix FTS try/except misplacement in api/documents.py — was wrapping the ORM
statement builder (never raises) instead of the execute call, causing HTTP 500
on SQLite test env. Now falls back to unfiltered results when @@ unsupported.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
curo1305
2026-05-31 15:21:02 +02:00
parent eab5f124f6
commit 62daf0d750
3 changed files with 370 additions and 35 deletions
+184
View File
@@ -710,6 +710,190 @@ async def test_delete_cloud_document_failure(async_client, auth_user, db_session
assert still_there is not None, "DB row should not be deleted when cloud delete fails"
# ---------------------------------------------------------------------------
# Phase 4 FOLD-04 — Document list sort (task 4-03-07)
# ---------------------------------------------------------------------------
async def test_document_sort_by_name_asc(async_client, auth_user, db_session):
"""GET /api/documents?sort=name&order=asc returns docs sorted by filename ascending.
FOLD-04: sort=name|date|size with order=asc|desc.
Creates three docs with distinct filenames; asserts the response order is
lexicographically ascending (a, b, c).
"""
import uuid as _uuid
from db.models import Document
user = auth_user["user"]
filenames = ["charlie.pdf", "alpha.pdf", "bravo.pdf"]
for name in filenames:
doc_id = _uuid.uuid4()
db_session.add(Document(
id=doc_id,
user_id=user.id,
filename=name,
content_type="application/pdf",
size_bytes=100,
storage_backend="minio",
status="uploaded",
object_key=f"{user.id}/{doc_id}/{_uuid.uuid4()}.pdf",
))
await db_session.commit()
resp = await async_client.get(
"/api/documents?sort=name&order=asc",
headers=auth_user["headers"],
)
assert resp.status_code == 200, resp.text
data = resp.json()
assert data["total"] == 3
names = [item["filename"] for item in data["items"]]
assert names == sorted(names), (
f"Expected ascending filename order, got: {names}"
)
# Explicit: alpha before bravo before charlie
assert names[0] == "alpha.pdf"
assert names[1] == "bravo.pdf"
assert names[2] == "charlie.pdf"
async def test_document_sort_by_size_desc(async_client, auth_user, db_session):
"""GET /api/documents?sort=size&order=desc returns docs sorted by size_bytes descending.
FOLD-04: the largest document must appear first.
Creates three docs with distinct sizes; asserts response order is largest-first.
"""
import uuid as _uuid
from db.models import Document
user = auth_user["user"]
sizes = [512, 2048, 1024] # expected desc order: 2048, 1024, 512
for sz in sizes:
doc_id = _uuid.uuid4()
db_session.add(Document(
id=doc_id,
user_id=user.id,
filename=f"file_{sz}.pdf",
content_type="application/pdf",
size_bytes=sz,
storage_backend="minio",
status="uploaded",
object_key=f"{user.id}/{doc_id}/{_uuid.uuid4()}.pdf",
))
await db_session.commit()
resp = await async_client.get(
"/api/documents?sort=size&order=desc",
headers=auth_user["headers"],
)
assert resp.status_code == 200, resp.text
data = resp.json()
assert data["total"] == 3
returned_sizes = [item["size_bytes"] for item in data["items"]]
assert returned_sizes == sorted(returned_sizes, reverse=True), (
f"Expected descending size order, got: {returned_sizes}"
)
assert returned_sizes[0] == 2048
assert returned_sizes[-1] == 512
# ---------------------------------------------------------------------------
# Phase 4 FOLD-05 — Full-text search (task 4-03-08)
# ---------------------------------------------------------------------------
async def test_fts_search_returns_200(async_client, auth_user, db_session):
"""GET /api/documents?q=keyword returns 200 without crashing (SQLite compat).
FOLD-05: FTS endpoint must not raise an error on any DB backend.
On SQLite the plainto_tsquery clause is silently skipped; endpoint must
still return 200 with a valid paginated response.
"""
import uuid as _uuid
from db.models import Document
user = auth_user["user"]
doc_id = _uuid.uuid4()
db_session.add(Document(
id=doc_id,
user_id=user.id,
filename="invoice_report.pdf",
content_type="application/pdf",
size_bytes=500,
storage_backend="minio",
status="uploaded",
object_key=f"{user.id}/{doc_id}/{_uuid.uuid4()}.pdf",
extracted_text="This document is about invoices and financial reports.",
))
await db_session.commit()
resp = await async_client.get(
"/api/documents?q=invoice",
headers=auth_user["headers"],
)
assert resp.status_code == 200, (
f"Expected 200 for ?q= search, got {resp.status_code}: {resp.text}"
)
data = resp.json()
assert "items" in data
assert "total" in data
assert isinstance(data["items"], list)
async def test_fts_search_cross_user_isolation(async_client, auth_user, second_auth_user, db_session):
"""GET /api/documents?q=keyword never returns another user's documents.
FOLD-05 T-4-05: FTS results are always scoped to the current user's documents.
User B's document with matching text must not appear in User A's search results.
"""
import uuid as _uuid
from db.models import Document
user_a = auth_user["user"]
user_b = second_auth_user["user"]
# Create a doc owned by user B with distinctive text
doc_b_id = _uuid.uuid4()
db_session.add(Document(
id=doc_b_id,
user_id=user_b.id,
filename="userb_secret.pdf",
content_type="application/pdf",
size_bytes=200,
storage_backend="minio",
status="uploaded",
object_key=f"{user_b.id}/{doc_b_id}/{_uuid.uuid4()}.pdf",
extracted_text="confidential contract agreement userb_only_term",
))
# Create a doc owned by user A (no matching text for the search query)
doc_a_id = _uuid.uuid4()
db_session.add(Document(
id=doc_a_id,
user_id=user_a.id,
filename="usera_unrelated.pdf",
content_type="application/pdf",
size_bytes=100,
storage_backend="minio",
status="uploaded",
object_key=f"{user_a.id}/{doc_a_id}/{_uuid.uuid4()}.pdf",
extracted_text="completely different content",
))
await db_session.commit()
# User A searches — must never see user B's document ID
resp = await async_client.get(
"/api/documents?q=userb_only_term",
headers=auth_user["headers"],
)
assert resp.status_code == 200, resp.text
data = resp.json()
returned_ids = [item["id"] for item in data["items"]]
assert str(doc_b_id) not in returned_ids, (
f"User B's document appeared in User A's search results: {returned_ids}"
)
async def test_delete_cloud_remove_only(async_client, auth_user, db_session):
"""DELETE /api/documents/{id}?remove_only=true skips cloud delete, removes DB row only (D-02)"""
import uuid as _uuid