tests/test_documents_delete.py

from __future__ import annotations

import os

from app.core.database import SessionLocal
from app.models.chat_session import ChatMessageRecord, ChatSession
from app.models.document import Document
from app.models.document_chunk import DocumentChunk
from app.models.flashcard import FlashcardSet
from app.models.generation import Generation
from app.models.generation_cache import GenerationCache
from app.models.learning_state import GeneratedResource
from app.models.previous_paper import PreviousPaper
from app.models.previous_question import PreviousQuestion
from app.models.quiz import Quiz
from app.models.study_profile import StudyProfile
from app.models.syllabus_item import SyllabusItem
from app.models.video_render_job import VideoRenderJob


def _signup(client, *, email: str, password: str = "Pass123!beta", name: str = "Test User") -> str:
    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:
    resp = client.post(
        "/documents/upload",
        headers=_auth(token),
        data={"title": title, "subject": "Physics", "chapter": "Sound Waves"},
        files={
            "file": (
                "physics_notes.txt",
                b"Sound travels as a longitudinal wave through a medium.",
                "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:
    assert resp.status_code == 404, f"expected 404, got {resp.status_code}: {resp.text}"
    data = resp.json()
    assert data["success"] is False
    assert data["error"]["code"] == "RESOURCE_NOT_FOUND"


class TestDocumentDeletion:
    def test_delete_requires_authentication(self, auth_client) -> None:
        resp = auth_client.delete("/documents/doc_does_not_matter")
        assert resp.status_code in (401, 403)

    def test_delete_nonexistent_document_returns_standard_404(self, auth_client) -> None:
        token = _signup(auth_client, email="delete_missing@docs.test", name="Missing User")
        resp = auth_client.delete("/documents/doc_not_real_id", headers=_auth(token))
        _assert_standard_not_found(resp)

    def test_delete_requires_ownership(self, auth_client) -> None:
        token_a = _signup(auth_client, email="alice_delete@docs.test", name="Alice")
        token_b = _signup(auth_client, email="bob_delete@docs.test", name="Bob")
        doc_id = _create_document(auth_client, token_a, "Alice Notes")

        resp = auth_client.delete(f"/documents/{doc_id}", headers=_auth(token_b))
        _assert_standard_not_found(resp)

        # Alice's document must still be reachable by Alice — Bob's attempt did nothing.
        still_there = auth_client.get(f"/documents/{doc_id}", headers=_auth(token_a))
        assert still_there.status_code == 200

    def test_delete_removes_document_chunks_and_file_from_disk(self, auth_client) -> None:
        token = _signup(auth_client, email="delete_full@docs.test", name="Delete User")
        doc_id = _create_document(auth_client, token, "Full Delete Notes")

        with SessionLocal() as db:
            doc = db.get(Document, doc_id)
            assert doc is not None
            file_path = doc.file_path
            assert os.path.exists(file_path)
            chunk_count_before = db.query(DocumentChunk).filter(DocumentChunk.document_id == doc_id).count()
            assert chunk_count_before > 0

        resp = auth_client.delete(f"/documents/{doc_id}", headers=_auth(token))
        assert resp.status_code == 204, f"expected 204, got {resp.status_code}: {resp.text}"

        # Row, chunks, and the physical file must all be gone.
        with SessionLocal() as db:
            assert db.get(Document, doc_id) is None
            chunk_count_after = db.query(DocumentChunk).filter(DocumentChunk.document_id == doc_id).count()
            assert chunk_count_after == 0
        assert not os.path.exists(file_path)

        # A second delete (or any subsequent access) is an honest 404, not a 500.
        _assert_standard_not_found(auth_client.delete(f"/documents/{doc_id}", headers=_auth(token)))
        _assert_standard_not_found(auth_client.get(f"/documents/{doc_id}", headers=_auth(token)))

    def test_delete_removes_derived_generations_quizzes_and_flashcards(self, auth_client) -> None:
        token = _signup(auth_client, email="delete_derived@docs.test", name="Derived User")
        doc_id = _create_document(auth_client, token, "Derived Notes")

        with SessionLocal() as db:
            user_id = db.query(Document).filter(Document.id == doc_id).one().user_id
            db.add(Generation(
                user_id=user_id,
                document_id=doc_id,
                type="notes",
                output_json={"notes": ["a"]},
                model_used="test-model",
            ))
            db.add(Quiz(user_id=user_id, document_id=doc_id, questions_json={"questions": []}))
            db.add(FlashcardSet(user_id=user_id, document_id=doc_id, cards_json={"cards": []}))
            db.commit()

            assert db.query(Generation).filter(Generation.document_id == doc_id).count() == 1
            assert db.query(Quiz).filter(Quiz.document_id == doc_id).count() == 1
            assert db.query(FlashcardSet).filter(FlashcardSet.document_id == doc_id).count() == 1

        resp = auth_client.delete(f"/documents/{doc_id}", headers=_auth(token))
        assert resp.status_code == 204

        with SessionLocal() as db:
            assert db.query(Generation).filter(Generation.document_id == doc_id).count() == 0
            assert db.query(Quiz).filter(Quiz.document_id == doc_id).count() == 0
            assert db.query(FlashcardSet).filter(FlashcardSet.document_id == doc_id).count() == 0

    def test_delete_detaches_video_render_job_and_study_profile_without_deleting_them(self, auth_client) -> None:
        token = _signup(auth_client, email="delete_detach@docs.test", name="Detach User")
        doc_id = _create_document(auth_client, token, "Detach Notes")

        with SessionLocal() as db:
            user_id = db.query(Document).filter(Document.id == doc_id).one().user_id
            job = VideoRenderJob(user_id=user_id, title="A rendered video", source_document_id=doc_id)
            db.add(job)
            profile = db.query(StudyProfile).filter(StudyProfile.user_id == user_id).one_or_none()
            if profile is None:
                profile = StudyProfile(user_id=user_id, uploaded_document_id=doc_id)
                db.add(profile)
            else:
                profile.uploaded_document_id = doc_id
            db.commit()
            job_id = job.id
            profile_id = profile.id

        resp = auth_client.delete(f"/documents/{doc_id}", headers=_auth(token))
        assert resp.status_code == 204

        with SessionLocal() as db:
            kept_job = db.get(VideoRenderJob, job_id)
            kept_profile = db.get(StudyProfile, profile_id)
            # The video and the profile survive deletion — only the dangling
            # reference to the removed document is cleared.
            assert kept_job is not None
            assert kept_job.source_document_id is None
            assert kept_profile is not None
            assert kept_profile.uploaded_document_id is None

    def test_delete_removes_all_source_derived_account_data(self, auth_client) -> None:
        token = _signup(auth_client, email="delete_all_derived@docs.test", name="Privacy User")
        doc_id = _create_document(auth_client, token, "Private Sound Waves Notes")

        with SessionLocal() as db:
            document = db.get(Document, doc_id)
            assert document is not None
            user_id = document.user_id

            paper = PreviousPaper(
                user_id=user_id,
                title="Derived Sound Waves PYQ",
                subject="Physics",
                file_name=document.file_name,
                file_type=document.file_type,
                file_path=document.file_path,
                status="ready",
                verification_status="verified",
            )
            paper.questions.append(
                PreviousQuestion(
                    question_number="1",
                    question_text="Define frequency.",
                    subject="Physics",
                    source_origin="user_uploaded",
                ),
            )
            db.add(paper)
            db.add(
                SyllabusItem(
                    user_id=user_id,
                    source_id=doc_id,
                    board="Kerala State Board",
                    class_level="SSLC / 10th",
                    subject="Physics",
                    chapter="Sound Waves",
                    topic="Frequency",
                ),
            )
            db.add(
                GeneratedResource(
                    user_id=user_id,
                    source_id=doc_id,
                    resource_type="notes",
                    title="Generated Sound Waves notes",
                    resource_data={"body": "private derived content"},
                ),
            )
            session = ChatSession(
                user_id=user_id,
                source_id=doc_id,
                subject="Physics",
                title="Chat from private notes",
            )
            session.messages.append(
                ChatMessageRecord(
                    role="assistant",
                    content="Answer grounded in the private notes.",
                ),
            )
            db.add(session)
            db.add(
                GenerationCache(
                    cache_key=f"legacy:{doc_id}",
                    task_type="notes",
                    provider="test",
                    input_hash="private-source-hash",
                    output_text="cached private output",
                    metadata_json={"document_id": doc_id},
                ),
            )
            db.commit()
            paper_id = paper.id
            session_id = session.id
            assert db.query(PreviousQuestion).filter(
                PreviousQuestion.previous_paper_id == paper_id,
            ).count() == 1
            assert db.query(ChatMessageRecord).filter(
                ChatMessageRecord.session_id == session_id,
            ).count() == 1

        resp = auth_client.delete(f"/documents/{doc_id}", headers=_auth(token))
        assert resp.status_code == 204

        with SessionLocal() as db:
            assert db.get(PreviousPaper, paper_id) is None
            assert db.query(PreviousQuestion).filter(
                PreviousQuestion.previous_paper_id == paper_id,
            ).count() == 0
            assert db.query(SyllabusItem).filter(
                SyllabusItem.source_id == doc_id,
            ).count() == 0
            assert db.query(GeneratedResource).filter(
                GeneratedResource.source_id == doc_id,
            ).count() == 0
            preserved_session = db.get(ChatSession, session_id)
            assert preserved_session is not None
            assert preserved_session.source_id is None
            assert preserved_session.active_source_ids == []
            assert db.query(ChatMessageRecord).filter(
                ChatMessageRecord.session_id == session_id,
            ).count() == 1
            assert db.get(GenerationCache, f"legacy:{doc_id}") is None

    def test_legacy_source_delete_uses_the_same_complete_cleanup(self, auth_client) -> None:
        token = _signup(auth_client, email="delete_source_alias@docs.test", name="Source User")
        doc_id = _create_document(auth_client, token, "Legacy Source Delete")

        with SessionLocal() as db:
            document = db.get(Document, doc_id)
            assert document is not None
            file_path = document.file_path
            db.add(
                GeneratedResource(
                    user_id=document.user_id,
                    source_id=doc_id,
                    resource_type="quiz",
                    title="Source-linked quiz",
                    resource_data={},
                ),
            )
            db.commit()

        resp = auth_client.delete(f"/sources/{doc_id}", headers=_auth(token))
        assert resp.status_code == 204

        with SessionLocal() as db:
            assert db.get(Document, doc_id) is None
            assert db.query(GeneratedResource).filter(
                GeneratedResource.source_id == doc_id,
            ).count() == 0
        assert not os.path.exists(file_path)