3279e14bb2
There will be no courses. What was there: one draft called "jk" with two empty lessons, and 4,000 lines of code around it — courses, modules, lessons, enrolments, per-lesson progress, SCORM, BigBlueButton, completion certificates, three React pages, a router, two models. Its real cost was everywhere else. Every query that measured practice had to remember `Quiz.course_id.is_(None)`, and forgetting it in one place would have silently mixed course attempts into a learner's analytics; the bank predicate carried a subquery to exclude a course's own questions from every search, recommendation and share; quiz access had a second, parallel rule about enrolment. All of that is gone, so the remaining rules say what they mean. `quizzes.allow_review` goes with it. It was only ever enforced for a course quiz, so it had become a promise nothing keeps — the public session page was still offering "no answer review" about sessions that review fine. The fixtures' question 5 lived in a course quiz and stood for "a question that exists but is not in your bank". There is no such thing now — a question is in the bank unless it is deleted — so the counts it kept out of the numbers are back in, and the tests that turned on it now turn on deletion or on the attempt that actually holds a question. Files the LMS uploaded stay on disk and stay protected: LEGACY_LMS_PREFIXES in app/utils/upload_access.py is what keeps them unreachable. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
225 lines
10 KiB
Python
225 lines
10 KiB
Python
"""What the cross-encoder is allowed to change, and what happens when it cannot.
|
|
|
|
No network: the point of these is not whether a reranker ranks well — that is
|
|
measured against the real corpus, in docs/reranking.md — but that every way it
|
|
can fail leaves the same rows on the page, in some order, with no error.
|
|
|
|
Run: DATABASE_URL=sqlite:// PYTHONPATH=backend python -m unittest discover -s backend/tests
|
|
"""
|
|
import os
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
os.environ.setdefault("DATABASE_URL", "sqlite://")
|
|
|
|
import httpx
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.database import Base
|
|
from app.models.article import Article, ArticleSectionIndex # noqa — mapper registration.
|
|
from app.models.media import MediaAsset # noqa — media is an embeddable corpus.
|
|
from app.models.question import Question
|
|
from app.models.user import User
|
|
from app.services import rerank_service, search_service
|
|
|
|
|
|
def response(payload, status=200):
|
|
"""An httpx reply in the shape the proxy sends one."""
|
|
return httpx.Response(status, json=payload, request=httpx.Request("POST", "http://proxy/v1/rerank"))
|
|
|
|
|
|
def scored(*pairs):
|
|
"""A rerank body: (index, score) pairs, deliberately not in index order."""
|
|
return {"id": "test", "results": [{"index": i, "relevance_score": s} for i, s in pairs]}
|
|
|
|
|
|
class ConfigurationTests(unittest.TestCase):
|
|
def test_no_model_means_no_call_and_no_opinion(self):
|
|
with patch.object(rerank_service, "rerank_model", return_value=""), \
|
|
patch("httpx.post", side_effect=AssertionError("must not reach the proxy")):
|
|
self.assertIsNone(rerank_service.rerank("croup", ["a", "b"]))
|
|
|
|
def test_a_model_without_a_proxy_is_not_configured(self):
|
|
with patch.object(rerank_service, "rerank_model", return_value="a-reranker"), \
|
|
patch.object(rerank_service.settings, "LITELLM_API_BASE", ""):
|
|
self.assertFalse(rerank_service.is_configured())
|
|
|
|
|
|
class DegradationTests(unittest.TestCase):
|
|
"""Every one of these must return None, which every caller reads as "keep it"."""
|
|
|
|
def setUp(self):
|
|
self.configured = patch.multiple(
|
|
rerank_service.settings, LITELLM_API_BASE="http://proxy",
|
|
LITELLM_API_KEY="k", LITELLM_RERANK_MODEL="a-reranker")
|
|
self.configured.start()
|
|
self.addCleanup(self.configured.stop)
|
|
# Redis is absent in tests; the cache must be optional, not required.
|
|
self.no_cache = patch.object(rerank_service, "_redis", return_value=None)
|
|
self.no_cache.start()
|
|
self.addCleanup(self.no_cache.stop)
|
|
|
|
def none_for(self, **post):
|
|
with patch("httpx.post", **post):
|
|
return rerank_service.rerank("croup", ["a", "b", "c"])
|
|
|
|
def test_unreachable_proxy(self):
|
|
self.assertIsNone(self.none_for(side_effect=httpx.ConnectError("refused")))
|
|
|
|
def test_a_bare_502_while_the_proxy_restarts(self):
|
|
self.assertIsNone(self.none_for(return_value=response({"error": "bad gateway"}, status=502)))
|
|
|
|
def test_a_timeout(self):
|
|
self.assertIsNone(self.none_for(side_effect=httpx.ReadTimeout("slow")))
|
|
|
|
def test_a_body_that_is_not_json(self):
|
|
self.assertIsNone(self.none_for(return_value=httpx.Response(
|
|
200, text="<html>gateway</html>",
|
|
request=httpx.Request("POST", "http://proxy/v1/rerank"))))
|
|
|
|
def test_a_reply_missing_a_document(self):
|
|
# Two scores for three documents: the third would have to be guessed,
|
|
# and a guess here is a reordering nobody asked for.
|
|
self.assertIsNone(self.none_for(return_value=response(scored((0, 0.1), (1, 0.9)))))
|
|
|
|
def test_a_reply_with_an_index_out_of_range(self):
|
|
self.assertIsNone(self.none_for(return_value=response(
|
|
scored((0, 0.1), (1, 0.9), (7, 0.5)))))
|
|
|
|
def test_a_reply_with_a_duplicated_index(self):
|
|
self.assertIsNone(self.none_for(return_value=response(
|
|
scored((0, 0.1), (0, 0.9), (2, 0.5)))))
|
|
|
|
def test_a_score_that_is_not_a_number(self):
|
|
self.assertIsNone(self.none_for(return_value=response(
|
|
{"results": [{"index": 0, "relevance_score": "high"},
|
|
{"index": 1, "relevance_score": 0.2},
|
|
{"index": 2, "relevance_score": 0.3}]})))
|
|
|
|
def test_scores_are_placed_by_the_index_the_server_echoed(self):
|
|
# The proxy answers best-first, so reading positionally would give the
|
|
# first document the best document's score.
|
|
with patch("httpx.post", return_value=response(scored((2, 0.9), (0, 0.1), (1, 0.4)))):
|
|
self.assertEqual(rerank_service.rerank("croup", ["a", "b", "c"]), [0.1, 0.4, 0.9])
|
|
|
|
|
|
class ReorderTests(unittest.TestCase):
|
|
def test_the_result_is_always_a_permutation(self):
|
|
items = list(range(8))
|
|
with patch.object(rerank_service, "rerank", return_value=[0.5] * 8):
|
|
self.assertCountEqual(rerank_service.reorder("q", items, str), items)
|
|
|
|
def test_no_opinion_leaves_the_order_alone(self):
|
|
items = [3, 1, 2]
|
|
with patch.object(rerank_service, "rerank", return_value=None):
|
|
self.assertEqual(rerank_service.reorder("q", items, str), items)
|
|
|
|
def test_ties_keep_the_order_fusion_gave_them(self):
|
|
with patch.object(rerank_service, "rerank", return_value=[0.4, 0.4, 0.9]):
|
|
self.assertEqual(rerank_service.reorder("q", [10, 20, 30], str), [30, 10, 20])
|
|
|
|
def test_only_the_head_is_scored_and_the_tail_is_kept(self):
|
|
items = list(range(rerank_service.MAX_CANDIDATES + 5))
|
|
seen = {}
|
|
|
|
def fake(query, documents):
|
|
seen["count"] = len(documents)
|
|
return list(range(len(documents))) # exactly reverses the head
|
|
|
|
with patch.object(rerank_service, "rerank", side_effect=fake):
|
|
out = rerank_service.reorder("q", items, str)
|
|
self.assertEqual(seen["count"], rerank_service.MAX_CANDIDATES)
|
|
self.assertEqual(out[0], rerank_service.MAX_CANDIDATES - 1)
|
|
self.assertEqual(out[-5:], items[-5:])
|
|
self.assertCountEqual(out, items)
|
|
|
|
|
|
class CorpusReorderTests(unittest.TestCase):
|
|
"""`rerank_ids` against a real (SQLite) corpus, where the text comes from."""
|
|
|
|
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.db.add(User(id=1, name="Mod", email="mod@example.test",
|
|
hashed_password="unused", role="moderator"))
|
|
for question_id, text in [
|
|
(1, "A child with fever and a rash"),
|
|
(2, "A toddler with a barking cough and stridor"),
|
|
(3, "An infant with jaundice on day three"),
|
|
]:
|
|
self.db.add(Question(id=question_id, user_id=1, question_text=text,
|
|
question_type="mcq", options=["yes", "no"], correct_answer="yes"))
|
|
self.db.commit()
|
|
self.addCleanup(self.engine.dispose)
|
|
self.addCleanup(self.db.close)
|
|
|
|
def test_the_reranker_sees_the_row_text_and_reorders_by_it(self):
|
|
seen = {}
|
|
|
|
def fake(query, documents):
|
|
seen["documents"] = documents
|
|
return [0.1, 0.9, 0.2]
|
|
|
|
with patch.object(rerank_service, "is_configured", return_value=True), \
|
|
patch.object(rerank_service, "rerank", side_effect=fake):
|
|
out = search_service.rerank_ids(self.db, "croup", "question", [1, 2, 3])
|
|
self.assertEqual(out, [2, 3, 1])
|
|
self.assertIn("barking cough", " ".join(seen["documents"]))
|
|
|
|
def test_an_unconfigured_reranker_costs_neither_a_query_nor_the_order(self):
|
|
with patch.object(rerank_service, "is_configured", return_value=False), \
|
|
patch.object(rerank_service, "rerank", side_effect=AssertionError("no call")):
|
|
self.assertEqual(search_service.rerank_ids(self.db, "croup", "question", [3, 1, 2]),
|
|
[3, 1, 2])
|
|
|
|
def test_ids_that_are_no_longer_in_the_table_are_still_returned(self):
|
|
# Retrieval decided these exist; a row deleted between the two queries
|
|
# is scored on empty text, and is still on the page afterwards.
|
|
with patch.object(rerank_service, "is_configured", return_value=True), \
|
|
patch.object(rerank_service, "rerank", return_value=[0.2, 0.8, 0.5]):
|
|
out = search_service.rerank_ids(self.db, "croup", "question", [1, 2, 999])
|
|
self.assertEqual(out, [2, 999, 1])
|
|
|
|
def test_a_score_list_that_does_not_match_the_shortlist_is_refused(self):
|
|
with patch.object(rerank_service, "is_configured", return_value=True), \
|
|
patch.object(rerank_service, "rerank", return_value=[0.2, 0.8]):
|
|
self.assertEqual(search_service.rerank_ids(self.db, "croup", "question", [1, 2, 3]),
|
|
[1, 2, 3])
|
|
|
|
|
|
class AiModeShortlistTests(unittest.TestCase):
|
|
"""The shortlist is the answer's evidence, so membership matters most."""
|
|
|
|
def sources(self):
|
|
return [
|
|
{"kind": "section", "id": 1, "title": "Croup", "text": "barking cough", "score": 1.0},
|
|
{"kind": "section", "id": 2, "title": "Asthma", "text": "wheeze", "score": 0.5},
|
|
{"kind": "question", "id": 9, "title": "A toddler", "text": "stridor", "score": 1.0},
|
|
]
|
|
|
|
def test_one_scale_replaces_four_incomparable_ones(self):
|
|
from app.services import ai_mode_service
|
|
|
|
sources = self.sources()
|
|
with patch.object(ai_mode_service.rerank_service, "rerank",
|
|
return_value=[0.2, 0.1, 0.9]):
|
|
ai_mode_service._cross_encode("croup", sources)
|
|
# The best section and the best question both scored 1.0 before; now
|
|
# every source sits on one ranking, so the sort can mean something.
|
|
self.assertEqual([s["score"] for s in sources], [0.5, 1 / 3, 1.0])
|
|
|
|
def test_silence_leaves_every_score_exactly_as_it_was(self):
|
|
from app.services import ai_mode_service
|
|
|
|
sources = self.sources()
|
|
with patch.object(ai_mode_service.rerank_service, "rerank", return_value=None):
|
|
ai_mode_service._cross_encode("croup", sources)
|
|
self.assertEqual([s["score"] for s in sources], [1.0, 0.5, 1.0])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|