diff --git a/backend/tests/test_cloud.py b/backend/tests/test_cloud.py index 1e765be..aea810f 100644 --- a/backend/tests/test_cloud.py +++ b/backend/tests/test_cloud.py @@ -563,3 +563,125 @@ async def test_cross_user_idor( ) assert resp.status_code == 404 + + +# ── Plan 09 tests: PATCH /documents/{id} and cloud-aware re-analyze ────────── + +async def test_patch_document_filename(async_client, db_session): + """PATCH /api/documents/{id} with {filename} returns 200 with updated filename. + + Covers T-05-09-01: ownership enforced via get_regular_user. + """ + from db.models import Document + + auth = await _create_user_and_token(db_session, role="user") + + # Create a document owned by this user + doc_id = _uuid.uuid4() + doc = Document( + id=doc_id, + user_id=auth["user"].id, + filename="original.pdf", + content_type="application/pdf", + size_bytes=1024, + storage_backend="minio", + status="uploaded", + object_key=f"{auth['user'].id}/{doc_id}/some-uuid.pdf", + ) + db_session.add(doc) + await db_session.commit() + + resp = await async_client.patch( + f"/api/documents/{doc_id}", + json={"filename": "renamed.pdf"}, + headers=auth["headers"], + ) + + assert resp.status_code == 200 + data = resp.json() + assert data["filename"] == "renamed.pdf" or data.get("original_name") == "renamed.pdf" + + +async def test_patch_document_wrong_owner(async_client, db_session): + """PATCH /api/documents/{id} by a non-owner returns 404 (IDOR protection). + + Covers T-05-09-01: cross-user access returns 404, not 403, to avoid leaking + which document IDs exist for other users (D-16, T-03-11). + """ + from db.models import Document + + auth1 = await _create_user_and_token(db_session, role="user") + auth2 = await _create_user_and_token(db_session, role="user") + + # Create a document owned by user1 + doc_id = _uuid.uuid4() + doc = Document( + id=doc_id, + user_id=auth1["user"].id, + filename="private.pdf", + content_type="application/pdf", + size_bytes=512, + storage_backend="minio", + status="uploaded", + object_key=f"{auth1['user'].id}/{doc_id}/some-uuid.pdf", + ) + db_session.add(doc) + await db_session.commit() + + # User2 tries to rename user1's document + resp = await async_client.patch( + f"/api/documents/{doc_id}", + json={"filename": "hacked.pdf"}, + headers=auth2["headers"], + ) + + assert resp.status_code == 404 + + +async def test_reanalyze_cloud_document_routes_to_cloud_backend(db_session): + """Re-analyze task calls get_storage_backend_for_document for cloud documents. + + Verifies that doc.storage_backend != 'minio' causes _run() to use the cloud + backend path instead of the MinIO path (Plan 09, requirement CLOUD-07). + """ + from db.models import Document + from tasks.document_tasks import _run + from unittest.mock import AsyncMock, patch, MagicMock + + auth = await _create_user_and_token(db_session, role="user") + + # Create a nextcloud document + doc_id = _uuid.uuid4() + doc = Document( + id=doc_id, + user_id=auth["user"].id, + filename="cloud.pdf", + content_type="application/pdf", + size_bytes=2048, + storage_backend="nextcloud", + status="uploaded", + object_key="nc_file_id_xyz", + ) + db_session.add(doc) + await db_session.commit() + + # Mock cloud backend: returns file bytes, enabling extraction to proceed + mock_cloud_backend = AsyncMock() + mock_cloud_backend.get_object = AsyncMock(return_value=b"%PDF-1.4 fake content") + + # Mock MinIO backend to verify it is NOT called + mock_minio_backend = AsyncMock() + mock_minio_backend.get_object = AsyncMock(return_value=b"should not be called") + + with patch("tasks.document_tasks.get_storage_backend_for_document", return_value=mock_cloud_backend) as mock_gsb_doc, \ + patch("tasks.document_tasks.get_storage_backend", return_value=mock_minio_backend) as mock_gsb: + result = await _run(str(doc_id)) + + # Cloud backend's get_object must have been called + mock_cloud_backend.get_object.assert_called_once_with("nc_file_id_xyz") + + # MinIO backend's get_object must NOT have been called + mock_minio_backend.get_object.assert_not_called() + + # Result must not be an error from MinIO path + assert result.get("status") != "extract_failed" or "MinIO" not in result.get("error", "")