pdf-quiz-generator/backend/tests/test_ai_mode_matching.py

"""AI-mode entry points build tests from existing bank questions, never new ones.

Run: DATABASE_URL=sqlite:///:memory: PYTHONPATH=backend python -m unittest discover -s backend/tests
"""
import os
os.environ["DATABASE_URL"] = "sqlite:///:memory:"

import io
import zipfile
from datetime import datetime
import unittest
from unittest.mock import patch

from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool

from app.database import Base, get_db
from app.models.question import Question
from app.models.quiz import Quiz
from app.models.user import User
from app.routers import questions
from app.utils.auth import get_current_user


class AiModeMatchingTests(unittest.TestCase):
    def setUp(self):
        self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
        Base.metadata.create_all(self.engine)
        self.db = Session(self.engine)

        self.user = User(id=1, name="Learner", email="learner@example.test", hashed_password="unused")
        self.other = User(id=2, name="Other", email="other@example.test", hashed_password="unused")
        self.db.add_all([self.user, self.other])
        # A question that nobody may match against is a deleted one; the flag
        # an author could set to keep one back no longer exists.
        for qid, text, owner, deleted in [
            (1, "A child with fever and a seizure", 1, None),
            (2, "An infant with jaundice", 1, None),
            (3, "A question of someone else, since removed", 2, datetime(2026, 1, 1)),
        ]:
            self.db.add(Question(id=qid, user_id=owner, deleted_at=deleted, question_text=text,
                                 question_type="mcq", options=["yes", "no"], correct_answer="yes"))
        self.db.commit()

        app = FastAPI()
        app.include_router(questions.router, prefix="/questions")
        app.dependency_overrides[get_db] = lambda: self.db
        app.dependency_overrides[get_current_user] = lambda: self.user
        self.client = TestClient(app)

    def tearDown(self):
        self.client.close()
        self.db.close()
        self.engine.dispose()

    def describe(self, **body):
        # Wording overlaps the stems literally: the SQLite fallback has no
        # stemming, so "seizures" would not reach "seizure" as Postgres would.
        return self.client.post("/questions/builder/describe",
                                json={"text": "I want to study fever and seizure in a child", **body})

    def test_a_description_builds_a_test_from_existing_bank_questions(self):
        before = self.db.query(Question).count()
        response = self.describe(count=5)
        self.assertEqual(response.status_code, 200, response.text)
        body = response.json()
        self.assertGreater(body["questions_count"], 0)
        # Matching must never invent questions.
        self.assertEqual(self.db.query(Question).count(), before)
        quiz = self.db.get(Quiz, body["id"])
        self.assertIsNotNone(quiz)

    def test_matches_exclude_questions_the_user_cannot_see(self):
        with patch.object(questions, "hybrid_ids", return_value=([3, 1], set())):
            response = self.describe(count=5)
        self.assertEqual(response.status_code, 200, response.text)
        # Question 3 belongs to someone else and is private, so only 1 survives.
        self.assertEqual(response.json()["matched"], 1)

    def test_no_match_is_reported_rather_than_returning_an_empty_test(self):
        with patch.object(questions, "hybrid_ids", return_value=([], set())):
            response = self.describe()
        self.assertEqual(response.status_code, 400)
        self.assertIn("match", response.json()["detail"].lower())

    def test_description_and_count_are_bounded(self):
        self.assertEqual(self.describe(count=0).status_code, 422)
        self.assertEqual(self.describe(count=999).status_code, 422)
        self.assertEqual(self.client.post("/questions/builder/describe",
                                          json={"text": "short"}).status_code, 422)

    NOTES = "febrile seizure in a toddler with fever, and jaundice in a newborn infant"

    @staticmethod
    def as_docx(text: str) -> bytes:
        """The smallest thing that is really a Word document.

        Built rather than fixtured because what is being tested is that the
        endpoint reads the *bytes*: a .docx name over a text file is refused,
        and this is what makes the difference visible.
        """
        buffer = io.BytesIO()
        with zipfile.ZipFile(buffer, "w") as bundle:
            bundle.writestr("[Content_Types].xml", "<Types/>")
            bundle.writestr("word/document.xml",
                            f"<w:document><w:body><w:p><w:r><w:t>{text}</w:t>"
                            "</w:r></w:p></w:body></w:document>")
        return buffer.getvalue()

    def upload(self, content=None, filename="handout.docx", **data):
        if content is None:
            content = self.as_docx(self.NOTES)
        return self.client.post("/questions/builder/from-upload",
                                files={"file": (filename, io.BytesIO(content),
                                                "application/octet-stream")},
                                data={"count": "5", "mode": "learning", **data})

    def test_an_upload_is_matched_against_the_bank_and_not_stored(self):
        response = self.upload()
        self.assertEqual(response.status_code, 200, response.text)
        self.assertGreater(response.json()["questions_count"], 0)
        # The document is a query, so nothing new is persisted.
        self.assertEqual(self.db.query(Question).count(), 3)

    def test_upload_limits_are_enforced(self):
        oversized = b"x" * (questions.MAX_UPLOAD_BYTES + 10)
        self.assertEqual(self.upload(content=oversized).status_code, 413)
        self.assertEqual(self.upload(content=b"").status_code, 400)
        self.assertEqual(self.upload(content=self.as_docx("too short")).status_code, 400)
        self.assertEqual(self.upload(count="99").status_code, 400)
        self.assertEqual(self.upload(mode="nonsense").status_code, 400)

    def test_only_the_three_kinds_are_taken_and_the_name_is_not_evidence(self):
        """A file is what its bytes say it is.

        The endpoint used to decide by extension and fall through to "decode
        whatever this is as UTF-8" for everything else, so a shell script, an
        HTML page or a CSV all became a search query. Each is now refused with
        415, and renaming one to .pdf does not change that.
        """
        for content in [b"#!/bin/sh\nrm -rf /", b"<html><script>alert(1)</script></html>",
                        b"id,name\n1,two\n", self.NOTES.encode()]:
            self.assertEqual(self.upload(content=content, filename="notes.pdf").status_code, 415)
        # And an image with no tool model configured is told why, not 500'd.
        png = b"\x89PNG\r\n\x1a\n" + b"0" * 64
        answer = self.upload(content=png, filename="slide.png")
        self.assertEqual(answer.status_code, 400)
        self.assertIn("tool model", answer.json()["detail"])

    def test_a_matched_test_is_capped(self):
        self.assertLessEqual(questions.MAX_MATCHED_QUESTIONS, 30)


if __name__ == "__main__":
    unittest.main()