c22e7f9547
Asked for: no tutor on a question outside a session. The tutor is handed the correct answer and told it may explain it, so it is answer-side content — and once that rule is written down, the same rule catches two bigger holes: - `GET /questions/bank` returned `correct_answer`, `explanation`, `option_explanations`, `key_points` and `attending_tip` for every question in the bank, to any signed-in learner. It is the question manager's listing, but nothing stopped anyone calling it: the whole answer key, one request away from the questions it answers. Stems are still listed to everyone; the answer side now goes only to whoever writes that question. - The explanation image behind a question was readable by the same rule, with no attempt behind it. "Whoever writes it" is one function now — `may_edit_question` — and it means moderation, authorship, or an editorial grant that reaches where the question is filed. Everyone else earns the answer by sitting the question, which is what an attempt is. The bank browse, the search, the session and the review are all unchanged; the frontend already sends `attempt_id` everywhere it shows an answer. The question manager was reachable by a learner with no grant, and would now load as a bank of stems with every answer field blanked — a broken page rather than a door that is not theirs. It says so instead. Also, while looking at where cards surface: the answer review showed neither the topic reading nor the cards written against a question, though the player has shown both under the answer for a while — and the review is the one place a learner goes through everything they got wrong. The list form of that component fetched its cards and then dropped them on the floor. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
272 lines
12 KiB
Python
272 lines
12 KiB
Python
"""What happens to an image when the model doing the job cannot see one.
|
|
|
|
No network and no models: the proxy's catalogue and the completions client are
|
|
both stubbed, because what is worth testing is not what a model says about a
|
|
picture but which model is asked, how often, and what the caller is told when
|
|
nobody can be.
|
|
"""
|
|
import os
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
os.environ.setdefault("DATABASE_URL", "sqlite://")
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from app.config import settings
|
|
from app.models.ai_model_config import AIModelConfig
|
|
from app.services import vision_service
|
|
from app.services.ai_service import ProxyError
|
|
from app.services.vision_service import Image, VisionUnavailable
|
|
|
|
|
|
def catalogue_rows(*pairs):
|
|
"""A /model/info body, in the shape the live proxy answers with."""
|
|
return {"data": [{"model_name": name, "model_info": {"supports_vision": says}}
|
|
for name, says in pairs]}
|
|
|
|
|
|
def stub_chat(reply="A chest radiograph with a right lower lobe opacity."):
|
|
"""`ai_service.chat` hands back the message text itself, so the stub is the
|
|
text — there is no envelope left to imitate."""
|
|
return Mock(return_value=reply)
|
|
|
|
|
|
class VisionFallbackTests(unittest.TestCase):
|
|
def setUp(self):
|
|
engine = create_engine("sqlite://")
|
|
AIModelConfig.__table__.create(engine)
|
|
self.db = sessionmaker(bind=engine)()
|
|
self.image = Image(data=b"\x89PNG-not-really", media_type="image/png",
|
|
caption="the figure printed with the question stem")
|
|
|
|
# Redis is not part of what is being tested, and a cache that answers
|
|
# would hide the calls these tests are counting.
|
|
patch.object(vision_service, "_redis", return_value=None).start()
|
|
self.reset_caches()
|
|
|
|
def tearDown(self):
|
|
patch.stopall()
|
|
self.db.close()
|
|
self.reset_caches()
|
|
|
|
def reset_caches(self):
|
|
vision_service._catalogue = {}
|
|
vision_service._catalogue_at = 0.0
|
|
vision_service._probes.clear()
|
|
|
|
def catalogue(self, *pairs):
|
|
vision_service._catalogue = dict(pairs)
|
|
vision_service._catalogue_at = vision_service.time.monotonic()
|
|
|
|
def tool_model(self, model_id="tool-vision"):
|
|
self.db.add(AIModelConfig(name=model_id, model_id=model_id, task="tool",
|
|
is_active=True, is_default=True))
|
|
self.db.commit()
|
|
|
|
def test_a_model_that_can_see_is_handed_the_image_itself(self):
|
|
self.catalogue(("seeing-model", True))
|
|
create = stub_chat()
|
|
with patch.object(vision_service, "chat", create):
|
|
parts, handoff = vision_service.image_context(
|
|
self.db, [self.image], model_id="seeing-model")
|
|
|
|
self.assertEqual([part["type"] for part in parts], ["image_url"])
|
|
self.assertTrue(parts[0]["image_url"]["url"].startswith("data:image/png;base64,"))
|
|
self.assertFalse(handoff.delegated)
|
|
self.assertIsNone(handoff.tool_model)
|
|
create.assert_not_called()
|
|
|
|
def test_a_model_that_cannot_see_gets_the_tool_model_s_description(self):
|
|
self.catalogue(("blind-model", False), ("tool-vision", True))
|
|
self.tool_model()
|
|
create = stub_chat()
|
|
with patch.object(vision_service, "chat", create):
|
|
parts, handoff = vision_service.image_context(
|
|
self.db, [self.image], model_id="blind-model", context="A 4-year-old…")
|
|
|
|
self.assertTrue(handoff.delegated)
|
|
self.assertEqual(handoff.tool_model, "tool-vision")
|
|
self.assertEqual(handoff.as_dict()["primary_model"], "blind-model")
|
|
|
|
# The tool model saw the picture; the primary is given words.
|
|
create.assert_called_once()
|
|
sent = create.call_args.kwargs
|
|
self.assertEqual(sent["model"], "tool-vision")
|
|
self.assertEqual([part["type"] for part in sent["messages"][0]["content"]],
|
|
["image_url", "text"])
|
|
self.assertEqual([part["type"] for part in parts], ["text"])
|
|
self.assertIn("right lower lobe opacity", parts[0]["text"])
|
|
self.assertIn("tool-vision", parts[0]["text"])
|
|
|
|
def test_no_tool_model_is_an_error_naming_the_fix(self):
|
|
self.catalogue(("blind-model", False))
|
|
with self.assertRaises(VisionUnavailable) as raised:
|
|
vision_service.image_context(self.db, [self.image], model_id="blind-model")
|
|
message = str(raised.exception)
|
|
self.assertIn("cannot read images", message)
|
|
self.assertIn("Settings → AI models", message)
|
|
|
|
def test_a_tool_model_that_cannot_see_either_is_refused_before_it_is_called(self):
|
|
self.catalogue(("blind-model", False), ("also-blind", False))
|
|
self.tool_model("also-blind")
|
|
create = stub_chat()
|
|
with patch.object(vision_service, "chat", create):
|
|
with self.assertRaises(VisionUnavailable) as raised:
|
|
vision_service.image_context(self.db, [self.image], model_id="blind-model")
|
|
self.assertIn("also-blind", str(raised.exception))
|
|
create.assert_not_called()
|
|
|
|
def test_a_failing_tool_model_is_an_error_rather_than_a_text_only_answer(self):
|
|
self.catalogue(("blind-model", False), ("tool-vision", True))
|
|
self.tool_model()
|
|
create = Mock(side_effect=RuntimeError("upstream 500"))
|
|
with patch.object(vision_service, "chat", create):
|
|
with self.assertRaises(VisionUnavailable) as raised:
|
|
vision_service.image_context(self.db, [self.image], model_id="blind-model")
|
|
self.assertIn("upstream 500", str(raised.exception))
|
|
|
|
|
|
class CapabilityLookupTests(unittest.TestCase):
|
|
def setUp(self):
|
|
patch.object(vision_service, "_redis", return_value=None).start()
|
|
self.base = settings.LITELLM_API_BASE
|
|
settings.LITELLM_API_BASE = "http://proxy.invalid"
|
|
vision_service._catalogue = {}
|
|
vision_service._catalogue_at = 0.0
|
|
vision_service._probes.clear()
|
|
|
|
def tearDown(self):
|
|
patch.stopall()
|
|
settings.LITELLM_API_BASE = self.base
|
|
vision_service._catalogue = {}
|
|
vision_service._catalogue_at = 0.0
|
|
vision_service._probes.clear()
|
|
|
|
def test_the_catalogue_is_read_once_for_many_questions(self):
|
|
response = Mock(json=Mock(return_value=catalogue_rows(
|
|
("sees", True), ("blind", False), ("unlisted-in-the-cost-map", None))))
|
|
with patch("httpx.get", return_value=response) as fetch:
|
|
for _ in range(5):
|
|
self.assertTrue(vision_service.can_see("sees"))
|
|
self.assertFalse(vision_service.can_see("blind"))
|
|
fetch.assert_called_once()
|
|
|
|
def test_one_deployment_that_cannot_see_decides_the_alias(self):
|
|
# `best-chat` fans out to six deployments and a request lands on any of
|
|
# them, so a single false is the answer for the name.
|
|
response = Mock(json=Mock(return_value=catalogue_rows(
|
|
("best-chat", True), ("best-chat", None), ("best-chat", False))))
|
|
with patch("httpx.get", return_value=response):
|
|
self.assertFalse(vision_service.can_see("best-chat"))
|
|
|
|
def test_a_model_the_catalogue_says_nothing_about_is_probed_once(self):
|
|
response = Mock(json=Mock(return_value=catalogue_rows(("quiet", None))))
|
|
create = stub_chat("ok")
|
|
with patch("httpx.get", return_value=response), \
|
|
patch.object(vision_service, "chat", create):
|
|
self.assertTrue(vision_service.can_see("quiet"))
|
|
self.assertTrue(vision_service.can_see("quiet"))
|
|
create.assert_called_once()
|
|
self.assertEqual(create.call_args.kwargs["max_tokens"], 1)
|
|
|
|
def test_a_refused_probe_is_remembered_and_a_broken_one_is_not(self):
|
|
# The real error the proxy layer raises, not a stand-in: the probe sorts
|
|
# refusals by `status_code`, and that ProxyError carries one is the
|
|
# whole reason it exists.
|
|
response = Mock(json=Mock(return_value=catalogue_rows(("quiet", None))))
|
|
create = Mock(side_effect=ProxyError(400, "image input not supported"))
|
|
with patch("httpx.get", return_value=response), \
|
|
patch.object(vision_service, "chat", create):
|
|
self.assertFalse(vision_service.can_see("quiet"))
|
|
self.assertFalse(vision_service.can_see("quiet"))
|
|
create.assert_called_once()
|
|
|
|
# A timeout says nothing about the model, so nothing is kept: asking
|
|
# again asks the model again.
|
|
vision_service._probes.clear()
|
|
create.reset_mock()
|
|
create.side_effect = TimeoutError("proxy is down")
|
|
with patch("httpx.get", return_value=response), \
|
|
patch.object(vision_service, "chat", create):
|
|
self.assertFalse(vision_service.can_see("quiet"))
|
|
self.assertFalse(vision_service.can_see("quiet"))
|
|
self.assertEqual(create.call_count, 2)
|
|
|
|
def test_a_5xx_probe_is_not_remembered_either(self):
|
|
# Same shape as a refusal, opposite meaning: the proxy broke, the model
|
|
# said nothing, so the verdict must not be cached.
|
|
response = Mock(json=Mock(return_value=catalogue_rows(("quiet", None))))
|
|
create = Mock(side_effect=ProxyError(503, "upstream unavailable"))
|
|
with patch("httpx.get", return_value=response), \
|
|
patch.object(vision_service, "chat", create):
|
|
self.assertFalse(vision_service.can_see("quiet"))
|
|
self.assertFalse(vision_service.can_see("quiet"))
|
|
self.assertEqual(create.call_count, 2)
|
|
|
|
|
|
class TutorFigureTests(unittest.TestCase):
|
|
"""The tutor is the first job in the app where a picture reaches a model.
|
|
|
|
Borrowed fixtures: every question in them is printed with a stem figure and
|
|
an explanation figure, which is exactly the case being tested.
|
|
"""
|
|
|
|
def setUp(self):
|
|
import test_related_privacy as fixtures
|
|
from app.routers import teach
|
|
|
|
self.teach = teach
|
|
self.privacy = fixtures.PrivacyTests()
|
|
self.privacy.setUp()
|
|
self.privacy.can_see.return_value = False
|
|
patch.object(vision_service, "_redis", return_value=None).start()
|
|
# The tool model is checked against the catalogue before it is trusted,
|
|
# and this suite reaches no proxy: an empty catalogue is "nothing known
|
|
# against it", which is the case being tested.
|
|
patch.object(vision_service, "catalogue", return_value={}).start()
|
|
vision_service._catalogue, vision_service._catalogue_at = {}, 0.0
|
|
# The tutor is answer-side content, so outside a session it opens only
|
|
# for whoever writes the question. What is under test here is what a
|
|
# model is shown once it is open, so the caller is an educator.
|
|
self.privacy.login(self.privacy.mod)
|
|
|
|
def tearDown(self):
|
|
self.privacy.tearDown()
|
|
vision_service._catalogue, vision_service._catalogue_at = {}, 0.0
|
|
|
|
def chat(self):
|
|
return self.privacy.client.post("/teach/chat", json={
|
|
"question_id": 1, "messages": [{"role": "user", "content": "Explain"}]})
|
|
|
|
def sent_to_the_tutor(self):
|
|
return self.privacy.ai.call_args.kwargs["messages"]
|
|
|
|
def test_a_blind_tutor_is_given_the_tool_model_s_description(self):
|
|
self.privacy.db.add(AIModelConfig(name="tool-vision", model_id="tool-vision",
|
|
task="tool", is_active=True, is_default=True))
|
|
self.privacy.db.commit()
|
|
create = stub_chat("A frontal chest radiograph.")
|
|
with patch.object(vision_service, "chat", create):
|
|
response = self.chat()
|
|
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertTrue(response.json()["vision"]["delegated"])
|
|
self.assertEqual(response.json()["vision"]["tool_model"], "tool-vision")
|
|
# Both figures described, and the descriptions are what the tutor read.
|
|
self.assertEqual(create.call_count, 2)
|
|
figures = self.sent_to_the_tutor()[1]["content"]
|
|
self.assertEqual([part["type"] for part in figures], ["text", "text"])
|
|
self.assertIn("A frontal chest radiograph.", figures[0]["text"])
|
|
|
|
def test_with_nothing_able_to_read_it_the_tutor_is_told_so(self):
|
|
response = self.chat()
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertNotIn("vision", response.json())
|
|
told = self.sent_to_the_tutor()[1]["content"][0]["text"]
|
|
self.assertIn("neither describe nor infer", told)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|