import pytest import os from pathlib import Path from unittest.mock import patch from app.models.document import Document from app.models.document_chunk import DocumentChunk from app.core.database import SessionLocal from app.core.config import Settings # ── Helpers ─────────────────────────────────────────────────────────────────── def _signup(client, *, email: str, password: str = "Pass123!beta", name: str = "Test User") -> str: """Register a new user and return their JWT access token.""" resp = client.post( "/auth/signup", json={"name": name, "email": email, "password": password}, ) assert resp.status_code == 201, f"signup failed: {resp.status_code} {resp.text}" return resp.json()["access_token"] def _auth(token: str) -> dict: return {"Authorization": f"Bearer {token}"} def _create_document(client, token: str, title: str = "User Notes") -> str: """Create a document using the upload endpoint.""" resp = client.post( "/documents/upload", headers=_auth(token), data={ "title": title, "subject": "Physics", "chapter": "Electromagnetic Induction", }, files={ "file": ( "physics_notes.txt", b"Faraday discovered electromagnetic induction using coils and magnets.", "text/plain", ), }, ) assert resp.status_code == 201, f"upload failed: {resp.status_code} {resp.text}" return resp.json()["id"] def _assert_standard_not_found(resp) -> None: """Helper to verify response status and standardized error shape.""" assert resp.status_code == 404, f"expected 404, got {resp.status_code}: {resp.text}" data = resp.json() assert data["success"] is False assert "error" in data assert data["error"]["code"] == "RESOURCE_NOT_FOUND" assert data["error"]["message"] == "Resource not found." assert isinstance(data["error"]["details"], dict) # ── Test Suite ─────────────────────────────────────────────────────────────── class TestPhase4Checkpoint3Documents: def test_upload_extraction_success_sets_ready_status(self, auth_client) -> None: token = _signup(auth_client, email="success@checkpoint3.test", name="Success User") resp = auth_client.post( "/documents/upload", headers=_auth(token), data={ "title": "Good Notes", "subject": "Biology", "chapter": "Cell Division", }, files={ "file": ( "bio_notes.txt", b"Mitosis is a process of cell division that results in two genetically identical daughter cells.", "text/plain", ), }, ) assert resp.status_code == 201 data = resp.json() assert data["status"] == "ready" assert data["extracted_text_length"] == len("Mitosis is a process of cell division that results in two genetically identical daughter cells.") assert data["chunk_count"] > 0 # extraction_error is intentionally not exposed in API responses (security fix) assert data["processing_started_at"] is not None assert data["processing_completed_at"] is not None def test_upload_extraction_failure_sets_failed_status_and_error(self, auth_client) -> None: token = _signup(auth_client, email="failure@checkpoint3.test", name="Failure User") # To trigger an extraction failure under upload without hitting MIME validation 415, # we upload a text/plain file with fewer than 10 characters (extracted text too short). resp = auth_client.post( "/documents/upload", headers=_auth(token), data={ "title": "Short Notes", "subject": "Chemistry", "chapter": "Acids", }, files={ "file": ( "short.txt", b"Too short", "text/plain", ), }, ) assert resp.status_code == 201 data = resp.json() assert data["status"] == "failed" # extraction_error is intentionally not exposed in API responses (security fix) assert data["chunk_count"] == 0 assert data["processing_started_at"] is not None assert data["processing_completed_at"] is not None def test_retry_processing_requires_ownership(self, auth_client) -> None: token_a = _signup(auth_client, email="alice_retry@checkpoint3.test", name="Alice") token_b = _signup(auth_client, email="bob_retry@checkpoint3.test", name="Bob") doc_id_a = _create_document(auth_client, token_a, "Alice Notes") # Bob tries to retry Alice's document -> 404 RESOURCE_NOT_FOUND resp = auth_client.post( f"/documents/{doc_id_a}/retry-processing", headers=_auth(token_b), ) _assert_standard_not_found(resp) def test_retry_processing_clears_old_error(self, auth_client) -> None: token = _signup(auth_client, email="retry_clear@checkpoint3.test", name="Retry Clear User") # 1. Upload a short document so it fails resp = auth_client.post( "/documents/upload", headers=_auth(token), data={ "title": "Initial Fail", "subject": "History", "chapter": "World War II", }, files={ "file": ( "history.txt", b"Too short", "text/plain", ), }, ) doc_id = resp.json()["id"] with SessionLocal() as db: doc = db.get(Document, doc_id) assert doc.status == "failed" assert doc.extraction_error is not None # Replace the file content on disk to be valid so the retry succeeds Path(doc.file_path).write_text("World War II started in 1939 and ended in 1945.", encoding="utf-8") db.commit() # 2. Call retry-processing resp = auth_client.post( f"/documents/{doc_id}/retry-processing", headers=_auth(token), ) assert resp.status_code == 200 data = resp.json() assert data["status"] == "ready" # extraction_error is intentionally not exposed in API responses (security fix) assert data["extracted_text_length"] > 0 assert data["chunk_count"] > 0 assert data["processing_started_at"] is not None assert data["processing_completed_at"] is not None def test_retry_processing_fails_if_processing(self, auth_client) -> None: token = _signup(auth_client, email="retry_proc@checkpoint3.test", name="Retry Proc User") doc_id = _create_document(auth_client, token, "Proc Notes") with SessionLocal() as db: doc = db.get(Document, doc_id) doc.status = "processing" db.add(doc) db.commit() resp = auth_client.post( f"/documents/{doc_id}/retry-processing", headers=_auth(token), ) assert resp.status_code == 400 assert resp.json()["success"] is False assert "already being processed" in resp.json()["error"]["message"] def test_retry_processing_fails_if_file_missing(self, auth_client) -> None: token = _signup(auth_client, email="retry_missing@checkpoint3.test", name="Retry Missing User") doc_id = _create_document(auth_client, token, "Missing Notes") with SessionLocal() as db: doc = db.get(Document, doc_id) # Remove file from disk if os.path.exists(doc.file_path): os.remove(doc.file_path) doc.status = "failed" db.add(doc) db.commit() resp = auth_client.post( f"/documents/{doc_id}/retry-processing", headers=_auth(token), ) assert resp.status_code == 400 assert resp.json()["success"] is False assert "file was not found" in resp.json()["error"]["message"] def test_retrieval_requires_ownership(self, auth_client) -> None: token_a = _signup(auth_client, email="alice_ret@checkpoint3.test", name="Alice") token_b = _signup(auth_client, email="bob_ret@checkpoint3.test", name="Bob") doc_id_a = _create_document(auth_client, token_a, "Alice Notes") resp = auth_client.post( f"/documents/{doc_id_a}/retrieve", headers=_auth(token_b), json={"query": "induction", "limit": 5}, ) _assert_standard_not_found(resp) def test_retrieval_rejects_empty_query(self, auth_client) -> None: token = _signup(auth_client, email="empty_ret@checkpoint3.test", name="Empty User") doc_id = _create_document(auth_client, token, "Doc Notes") resp = auth_client.post( f"/documents/{doc_id}/retrieve", headers=_auth(token), json={"query": "", "limit": 5}, ) assert resp.status_code == 422 assert resp.json()["success"] is False assert resp.json()["error"]["code"] == "UNPROCESSABLE_ENTITY" def test_retrieval_rejects_too_large_limit(self, auth_client) -> None: token = _signup(auth_client, email="limit_ret@checkpoint3.test", name="Limit User") doc_id = _create_document(auth_client, token, "Doc Notes") resp = auth_client.post( f"/documents/{doc_id}/retrieve", headers=_auth(token), json={"query": "induction", "limit": 25}, ) # Fastapi validation error because limit > 20 in RetrievalRequest assert resp.status_code == 422 assert resp.json()["success"] is False def test_retrieval_returns_only_chunks_from_requested_document(self, auth_client) -> None: token = _signup(auth_client, email="multi_ret@checkpoint3.test", name="Multi User") doc_id_1 = _create_document(auth_client, token, "Doc 1") doc_id_2 = _create_document(auth_client, token, "Doc 2") resp = auth_client.post( f"/documents/{doc_id_1}/retrieve", headers=_auth(token), json={"query": "Faraday discover induction", "limit": 5}, ) assert resp.status_code == 200 chunks = resp.json()["chunks"] assert len(chunks) > 0 for chunk in chunks: assert chunk["document_id"] == doc_id_1 def test_retrieval_debug_does_not_leak_in_production(self, auth_client) -> None: token = _signup(auth_client, email="prod_ret@checkpoint3.test", name="Prod User") doc_id = _create_document(auth_client, token, "Doc Notes") # Mock settings.environment as "production" with patch("app.core.config.get_settings") as mock_settings: mock_set = Settings(environment="production") mock_settings.return_value = mock_set resp = auth_client.post( f"/documents/{doc_id}/retrieve", headers=_auth(token), json={"query": "Faraday", "limit": 5, "debug": True}, ) assert resp.status_code == 200 chunks = resp.json()["chunks"] assert len(chunks) > 0 for chunk in chunks: assert chunk["debug_info"] is None def test_retrieval_debug_exposed_in_development(self, auth_client) -> None: token = _signup(auth_client, email="dev_ret@checkpoint3.test", name="Dev User") doc_id = _create_document(auth_client, token, "Doc Notes") # Mock settings.environment as "development" with patch("app.core.config.get_settings") as mock_settings: mock_set = Settings(environment="development") mock_settings.return_value = mock_set resp = auth_client.post( f"/documents/{doc_id}/retrieve", headers=_auth(token), json={"query": "Faraday", "limit": 5, "debug": True}, ) assert resp.status_code == 200 chunks = resp.json()["chunks"] assert len(chunks) > 0 for chunk in chunks: assert chunk["debug_info"] is not None assert "tf_idf_overlap_score" in chunk["debug_info"] assert "exact_phrase_boost" in chunk["debug_info"] def test_unsupported_file_type_returns_standard_error_shape(self, auth_client) -> None: token = _signup(auth_client, email="unsupported@checkpoint3.test", name="Unsupported User") resp = auth_client.post( "/documents/upload", headers=_auth(token), data={ "title": "Bad Suffix Notes", "subject": "Chemistry", "chapter": "Acids", }, files={ "file": ( "notes.zip", b"Fake archive data", "application/zip", ), }, ) # Bypassed to 415 immediately by MIME check assert resp.status_code == 415 assert resp.json()["success"] is False assert "is not supported" in resp.json()["error"]["message"]