Spaces:
Running
Running
| """Offline tests for the Creighton Anatomy Learning Platform (Gemini, Supabase, and the web are stubbed). | |
| Run from the app folder, with the lecture PDFs available: | |
| LECTURES_DIR=/path/to/Lectures GEMINI_API_KEY=test python3 -m unittest -v test_app | |
| """ | |
| from __future__ import annotations | |
| import io | |
| import json | |
| import os | |
| import unittest | |
| import urllib.error | |
| import urllib.request | |
| from types import SimpleNamespace | |
| from unittest import mock | |
| os.environ.setdefault("GEMINI_API_KEY", "test") | |
| os.environ.pop("SUPABASE_URL", None) | |
| os.environ.pop("SUPABASE_KEY", None) | |
| import app # noqa: E402 (loads the lecture PDFs) | |
| import supabase_store # noqa: E402 | |
| import web_sources # noqa: E402 | |
| from app import Outcome, SlideRef, TutorReply, WebReply # noqa: E402 | |
| class FakeStore: | |
| """Records every call the app makes; accounts and messages live in memory.""" | |
| def __init__(self, enabled: bool = True) -> None: | |
| self.enabled = enabled | |
| self.users: dict[str, dict] = {} | |
| self.messages: list[dict] = [] | |
| self.calls: list[tuple] = [] | |
| def config_hint(self): | |
| return "" if self.enabled else "Missing Space secret(s): SUPABASE_URL" | |
| def get_user(self, user_id): | |
| return self.users.get(user_id), None | |
| def create_user(self, user_id, email): | |
| if user_id in self.users: | |
| return None, supabase_store.EXISTS | |
| self.users[user_id] = {"user_id": user_id, "email": email} | |
| return self.users[user_id], None | |
| def load_messages(self, user_id, limit=200): | |
| return [m for m in self.messages if m["user_id"] == user_id][-limit:], None | |
| def enqueue(self, method, *args): | |
| self.calls.append((method,) + args) | |
| if method == "append_messages": | |
| user_id, session_id, rows = args | |
| self.messages += [dict(row, user_id=user_id) for row in rows] | |
| def events(self): | |
| return [call[3] for call in self.calls if call[0] == "log_event"] | |
| def flush(self, timeout=3.0): | |
| self.calls.append(("flush",)) | |
| return True | |
| def first_text_slide_with_figure() -> tuple[app.Slide, app.Slide]: | |
| """A slide with text followed by a diagram or image-only slide (exists in every lecture set).""" | |
| for lecture in app.LECTURES.values(): | |
| for slide, following in zip(lecture.slides, lecture.slides[1:]): | |
| if not slide.diagram and not slide.image_only and not slide.section and (following.diagram or following.image_only): | |
| return slide, following | |
| raise AssertionError("no text slide followed by a figure slide") | |
| def signed_in_session(store: FakeStore, user_id: str = "jdoe") -> app.Session: | |
| session = app.Session() | |
| session.user_id, session.email, session.session_id = user_id, f"{user_id}@example.com", "sess1" | |
| return session | |
| def drain(generator) -> list[tuple]: | |
| return list(generator) | |
| class CitationTests(unittest.TestCase): | |
| def test_cited_slides_expands_ranges_and_drops_missing(self): | |
| lecture = next(iter(app.LECTURES.values())) | |
| answer = f"Text (Lecture {lecture.number}, slides 1-3 and 5). More (Lecture 999, slide 1)." | |
| self.assertEqual(app._cited_slides(answer), [(lecture.number, 1), (lecture.number, 2), (lecture.number, 3), (lecture.number, 5)]) | |
| def test_relevant_slides_keeps_cited_slides_and_their_figures(self): | |
| slide, figure = first_text_slide_with_figure() | |
| other = app.LECTURES[slide.lecture].slides[-1] | |
| answer = f"Fact (Lecture {slide.lecture}, slide {slide.number})." | |
| refs = [SlideRef(lecture=slide.lecture, slide=slide.number), SlideRef(lecture=figure.lecture, slide=figure.number), | |
| SlideRef(lecture=other.lecture, slide=other.number)] | |
| kept = app._relevant_slides(refs, answer) | |
| self.assertEqual([(r.lecture, r.slide) for r in kept], [slide.key, figure.key] if other.key != figure.key else [slide.key, figure.key]) | |
| def test_relevant_slides_is_empty_without_citations(self): | |
| slide, _ = first_text_slide_with_figure() | |
| self.assertEqual(app._relevant_slides([SlideRef(lecture=slide.lecture, slide=slide.number)], "No citations here."), []) | |
| def test_slides_note_groups_by_lecture(self): | |
| first, second = list(app.LECTURES.values())[:2] | |
| answer = f"(Lecture {first.number}, slide 2) and (Lecture {first.number}, slide 1) and (Lecture {second.number}, slide 1)" | |
| note = app._slides_note(answer) | |
| self.assertIn(f"Lecture {first.number} ({first.title}), slides 1, 2", note) | |
| self.assertIn(f"Lecture {second.number} ({second.title}), slide 1", note) | |
| self.assertEqual(app._slides_note("nothing cited"), "Course lecture slides.") | |
| def test_source_note_round_trip(self): | |
| content = app._with_source("Answer text", "Course lecture slides.") | |
| self.assertEqual(app._split_source(content), ("Answer text", "Course lecture slides.")) | |
| self.assertEqual(app._split_source("plain"), ("plain", "")) | |
| def test_none_note_variants(self): | |
| page = web_sources.Page("Kenhub", "https://www.kenhub.com/en/library/anatomy/retina", "Retina") | |
| self.assertIn("(Kenhub: [Retina](https://www.kenhub.com/en/library/anatomy/retina)) do not answer it either", app._none_note([page])) | |
| self.assertIn("no matching article was found", app._none_note([])) | |
| self.assertTrue(app._none_note([], "Kenhub could not be reached").endswith("(Kenhub could not be reached.)")) | |
| class TurnTests(unittest.TestCase): | |
| def setUp(self): | |
| self.store = FakeStore() | |
| self.patches = [mock.patch.object(app, "STORE", self.store)] | |
| for patch in self.patches: | |
| patch.start() | |
| self.slide, self.figure = first_text_slide_with_figure() | |
| def tearDown(self): | |
| for patch in self.patches: | |
| patch.stop() | |
| def slides_reply(self, covered=True, extra_slide=None): | |
| answer = f"## Answer\nA fact (Lecture {self.slide.lecture}, slide {self.slide.number})." | |
| refs = [SlideRef(lecture=self.slide.lecture, slide=self.slide.number), SlideRef(lecture=self.figure.lecture, slide=self.figure.number)] | |
| if extra_slide is not None: | |
| refs.append(SlideRef(lecture=extra_slide.lecture, slide=extra_slide.number)) | |
| return TutorReply(answer=answer if covered else "The slides do not cover this.", slides=refs, covered_by_slides=covered, search_topics=["retina"]) | |
| def test_thinking_message_then_answer_from_slides(self): | |
| other = app.LECTURES[self.slide.lecture].slides[-1] | |
| session = signed_in_session(self.store) | |
| with mock.patch.object(app, "gemini_reply", return_value=self.slides_reply(extra_slide=other)): | |
| outputs = drain(app.run_turn("ask", "What is it?", session)) | |
| self.assertGreaterEqual(len(outputs), 2) | |
| pending = outputs[0][0][-1] | |
| self.assertEqual(pending["metadata"], {"title": "Thinking", "status": "pending"}) | |
| self.assertEqual(outputs[0][0][-2], {"role": "user", "content": "What is it?"}) | |
| self.assertFalse(outputs[0][-3]["interactive"]) # Ask disabled while thinking | |
| self.assertEqual(len(outputs[0]), len(app.turn_outputs)) | |
| final = outputs[-1][0][-1] | |
| self.assertNotIn("metadata", final) | |
| self.assertTrue(final["content"].endswith(f"slide {self.slide.number}.")) | |
| self.assertIn("**Source:** Course lecture slides:", final["content"]) | |
| self.assertEqual(outputs[-1][1], "") # question box cleared | |
| self.assertEqual(outputs[-1][-1], session.version) # the page learns the chat version it now shows | |
| self.assertEqual(session.state["slides"], [[self.slide.lecture, self.slide.number], [self.figure.lecture, self.figure.number]]) | |
| self.assertEqual(session.state["topic"], "What is it?") | |
| self.assertFalse(session.busy) | |
| saved = [c for c in self.store.calls if c[0] == "append_messages"] | |
| self.assertEqual(len(saved), 1) | |
| rows = saved[0][3] | |
| self.assertEqual([r["role"] for r in rows], ["user", "assistant"]) | |
| self.assertEqual(rows[1]["source"], "slides") | |
| self.assertEqual(self.store.events(), ["ask"]) | |
| def test_falls_back_to_reference_pages(self): | |
| session = signed_in_session(self.store) | |
| page = web_sources.Page("TeachMeAnatomy", "https://teachmeanatomy.info/head/organs/eye/retina/", "The Retina", "text") | |
| web = WebReply(answer="Ten layers (TeachMeAnatomy: The Retina).", covered=True, used_urls=[page.url]) | |
| with mock.patch.object(app, "gemini_reply", return_value=self.slides_reply(covered=False)), \ | |
| mock.patch.object(web_sources, "find_pages", return_value=([page], "")) as find, \ | |
| mock.patch.object(app, "gemini_web_reply", return_value=web) as web_call: | |
| outputs = drain(app.run_turn("ask", "What are the layers of the retina?", session)) | |
| find.assert_called_once_with(["retina"]) | |
| self.assertEqual(web_call.call_args[0][0], "Student question: What are the layers of the retina?") | |
| self.assertEqual(web_call.call_args[0][2], [page]) | |
| progress = [o[0][-1]["content"] for o in outputs[:-1]] | |
| self.assertEqual(progress, ["Searching the lecture slides...", | |
| "The lecture slides do not cover this request. Checking TeachMeAnatomy and Kenhub..."]) | |
| final = outputs[-1][0][-1]["content"] | |
| self.assertTrue(final.startswith("Ten layers (TeachMeAnatomy: The Retina).")) | |
| self.assertIn("Answered from TeachMeAnatomy: [The Retina](https://teachmeanatomy.info/head/organs/eye/retina/).", final) | |
| self.assertEqual(session.state["slides"], []) | |
| self.assertIsNone(session.state["view"]) | |
| self.assertEqual(session.state["history"][-1]["text"], web.answer) | |
| rows = [c for c in self.store.calls if c[0] == "append_messages"][0][3] | |
| self.assertEqual(rows[1]["source"], "web") | |
| def test_reports_when_no_reference_page_matches(self): | |
| session = signed_in_session(self.store) | |
| with mock.patch.object(app, "gemini_reply", return_value=self.slides_reply(covered=False)), \ | |
| mock.patch.object(web_sources, "find_pages", return_value=([], "Kenhub could not be reached")), \ | |
| mock.patch.object(app, "gemini_web_reply") as web_call: | |
| outputs = drain(app.run_turn("ask", "Unknown thing?", session)) | |
| web_call.assert_not_called() | |
| final = outputs[-1][0][-1]["content"] | |
| self.assertIn("The slides do not cover this.", final) | |
| self.assertIn("no matching article was found on TeachMeAnatomy or Kenhub. (Kenhub could not be reached.)", final) | |
| rows = [c for c in self.store.calls if c[0] == "append_messages"][0][3] | |
| self.assertEqual(rows[1]["source"], "none") | |
| def test_reports_when_reference_pages_do_not_answer(self): | |
| session = signed_in_session(self.store) | |
| page = web_sources.Page("Kenhub", "https://www.kenhub.com/en/library/anatomy/retina", "Retina", "text") | |
| with mock.patch.object(app, "gemini_reply", return_value=self.slides_reply(covered=False)), \ | |
| mock.patch.object(web_sources, "find_pages", return_value=([page], "")), \ | |
| mock.patch.object(app, "gemini_web_reply", return_value=WebReply(answer="The pages do not answer this.", covered=False)): | |
| outputs = drain(app.run_turn("ask", "Unknown thing?", session)) | |
| final = outputs[-1][0][-1]["content"] | |
| self.assertTrue(final.startswith("The pages do not answer this.")) | |
| self.assertIn("(Kenhub: [Retina](https://www.kenhub.com/en/library/anatomy/retina)) do not answer it either", final) | |
| def test_relevant_slides_falls_back_to_cited_slides(self): | |
| other = app.LECTURES[self.slide.lecture].slides[-1] | |
| answer = f"Fact (Lecture {self.slide.lecture}, slide {self.slide.number})." | |
| kept = app._relevant_slides([SlideRef(lecture=other.lecture, slide=other.number)], answer) | |
| self.assertEqual([(r.lecture, r.slide) for r in kept], [self.slide.key]) | |
| def test_citation_groups_with_repeated_slide_word(self): | |
| lecture = self.slide.lecture | |
| self.assertEqual(app._cited_slides(f"(Lecture {lecture}, slide 1, slide 2)"), [(lecture, 1), (lecture, 2)]) | |
| self.assertEqual(app._verify_citations(f"(Lecture {lecture}, slide 1, slide 999)"), | |
| f"(Lecture {lecture}, slide 1, slide 999{app.NOT_FOUND_TAG})") | |
| def clinical(self, slide_reply, pages, web_reply): | |
| session = signed_in_session(self.store) | |
| session.state["topic"] = "knee and ankle" | |
| with mock.patch.object(app, "gemini_reply", return_value=slide_reply) as slide_call, \ | |
| mock.patch.object(web_sources, "find_pages", return_value=(pages, "")) as find, \ | |
| mock.patch.object(app, "gemini_web_reply", return_value=web_reply) as web_call: | |
| outputs = drain(app.run_turn("clinical", "", session)) | |
| rows = [c for c in self.store.calls if c[0] == "append_messages"][0][3] | |
| return session, outputs, rows, slide_call, find, web_call | |
| def test_clinical_answer_names_each_source_and_never_the_tutor(self): | |
| page = web_sources.Page("TeachMeAnatomy", "https://teachmeanatomy.info/lower-limb/joints/knee-joint/", "The Knee Joint", "text") | |
| slide_reply = TutorReply(answer=f"- An unhappy triad point (Lecture {self.slide.lecture}, slide {self.slide.number}).", | |
| slides=[], covered_by_slides=True, search_topics=["knee joint", "ankle joint"]) | |
| web_reply = WebReply(answer="- Bursitis (TeachMeAnatomy: The Knee Joint).", covered=True, used_urls=[page.url.rstrip("/")]) | |
| session, outputs, rows, slide_call, find, web_call = self.clinical(slide_reply, [page], web_reply) | |
| find.assert_called_once_with(["knee joint", "ankle joint"]) | |
| self.assertIn("List the clinical correlations for this topic that the slides state", slide_call.call_args[0][0]) | |
| self.assertIn("that the reference pages state: knee and ankle", web_call.call_args[0][0]) | |
| progress = [o[0][-1]["content"] for o in outputs[:-1]] | |
| self.assertEqual(progress, ["Searching the lecture slides for clinical notes...", | |
| "Checking TeachMeAnatomy and Kenhub for clinical notes..."]) | |
| body, note = app._split_source(outputs[-1][0][-1]["content"]) | |
| self.assertEqual(body, f"### From the lecture slides\n\n- An unhappy triad point (Lecture {self.slide.lecture}, slide {self.slide.number}).\n\n" | |
| "### From TeachMeAnatomy\n\n- Bursitis (TeachMeAnatomy: The Knee Joint).") | |
| self.assertTrue(note.startswith(f"Course lecture slides: Lecture {self.slide.lecture} (")) | |
| self.assertTrue(note.endswith("; TeachMeAnatomy: [The Knee Joint](https://teachmeanatomy.info/lower-limb/joints/knee-joint/).")) | |
| for phrase in ("beyond", "tutor", "standard clinical anatomy"): | |
| self.assertNotIn(phrase, outputs[-1][0][-1]["content"].lower()) | |
| self.assertEqual(rows[1]["source"], "slides+web") | |
| self.assertEqual(session.state["last_source"], "slides+web") | |
| def test_clinical_from_the_sites_when_the_slides_have_none(self): | |
| page = web_sources.Page("Kenhub", "https://www.kenhub.com/en/library/anatomy/the-ankle-joint", "Ankle joint", "text") | |
| slide_reply = TutorReply(answer="The lecture slides do not include clinical notes on this topic.", slides=[], | |
| covered_by_slides=False, search_topics=["ankle joint"]) | |
| web_reply = WebReply(answer="### Sprains\n- Inversion sprain (Kenhub: Ankle joint).", covered=True, used_urls=[page.url]) | |
| session, outputs, rows, *_ = self.clinical(slide_reply, [page], web_reply) | |
| body, note = app._split_source(outputs[-1][0][-1]["content"]) | |
| self.assertEqual(body, "### From the lecture slides\n\nThe lecture slides do not include clinical notes on this topic.\n\n" | |
| "### From Kenhub\n\n#### Sprains\n- Inversion sprain (Kenhub: Ankle joint).") | |
| self.assertEqual(note, "The lecture slides do not include clinical notes on this topic. Answered from " | |
| "Kenhub: [Ankle joint](https://www.kenhub.com/en/library/anatomy/the-ankle-joint).") | |
| self.assertEqual(rows[1]["source"], "web") | |
| def test_clinical_with_nothing_found_says_so(self): | |
| slide_reply = TutorReply(answer="The lecture slides do not include clinical notes on this topic.", slides=[], | |
| covered_by_slides=False, search_topics=["retina"]) | |
| session, outputs, rows, _, _, web_call = self.clinical(slide_reply, [], None) | |
| web_call.assert_not_called() | |
| body, note = app._split_source(outputs[-1][0][-1]["content"]) | |
| self.assertEqual(body, "Neither the lecture slides nor the TeachMeAnatomy and Kenhub pages checked include clinical notes on this topic.") | |
| self.assertEqual(rows[1]["source"], "none") | |
| def test_clinical_reads_the_pages_while_searching_the_slides_when_the_topic_is_known(self): | |
| session = signed_in_session(self.store) | |
| session.state.update(topic="knee", search_topics=["knee joint"]) | |
| page = web_sources.Page("Kenhub", "https://www.kenhub.com/en/library/anatomy/the-knee-joint", "Knee joint", "text") | |
| slide_reply = TutorReply(answer=f"- A point (Lecture {self.slide.lecture}, slide {self.slide.number}).", slides=[], | |
| covered_by_slides=True, search_topics=["something else"]) | |
| with mock.patch.object(app, "gemini_reply", return_value=slide_reply), \ | |
| mock.patch.object(web_sources, "find_pages", return_value=([page], "")) as find, \ | |
| mock.patch.object(app, "gemini_web_reply", side_effect=app.TutorError("busy")): | |
| outputs = drain(app.run_turn("clinical", "", session)) | |
| find.assert_called_once_with(["knee joint"]) # the structures of the answer that set the topic | |
| self.assertEqual([o[0][-1]["content"] for o in outputs[:-1]], | |
| ["Searching the lecture slides, TeachMeAnatomy, and Kenhub for clinical notes..."]) | |
| body, note = app._split_source(outputs[-1][0][-1]["content"]) # the page step failed: the slide part still shows | |
| self.assertIn("### From the lecture slides\n\n- A point", body) | |
| self.assertIn("The TeachMeAnatomy and Kenhub pages could not be read right now.", body) | |
| self.assertIn("could not be read right now", note) | |
| self.assertNotIn("add no clinical notes", note) | |
| self.assertEqual(session.state["search_topics"], ["knee joint"]) | |
| def test_ask_remembers_the_structures_for_clinical_correlation(self): | |
| session = signed_in_session(self.store) | |
| reply = self.slides_reply() | |
| reply.search_topics = ["knee joint", "ankle joint"] | |
| with mock.patch.object(app, "gemini_reply", return_value=reply): | |
| drain(app.run_turn("ask", "Knee and ankle?", session)) | |
| self.assertEqual(session.state["search_topics"], ["knee joint", "ankle joint"]) | |
| def test_beyond_guard_catches_heading_variants(self): | |
| for heading in ["### **Beyond the slides**", "### 2. Beyond the slides", "Beyond the slides:", "*Beyond the slides*", | |
| "### Additional clinical points (beyond the slides)", "**Beyond the lecture slides**"]: | |
| text = f"### Ligaments\n- Kept (Lecture 1, slide 2).\n{heading}\n- Tutor point\n#### Detail\n- More tutor\n### Summary\nKept." | |
| self.assertEqual(app._drop_beyond_sections(text), "### Ligaments\n- Kept (Lecture 1, slide 2).\n### Summary\nKept.", heading) | |
| self.assertEqual(app._drop_beyond_sections("**Beyond doubt, the ACL limits translation** (Lecture 1, slide 2)."), | |
| "**Beyond doubt, the ACL limits translation** (Lecture 1, slide 2).") | |
| def test_simplify_never_brings_back_old_beyond_sections_or_the_tutor_note(self): | |
| session = signed_in_session(self.store) | |
| old = "### From the reference pages\n- Point.\n\n### Beyond the reference pages\n- Tutor point." | |
| session.state.update(topic="knee", last_answer=old, last_source="web", last_note="Answered from Kenhub: [Knee](u).") | |
| rewritten = TutorReply(answer="### From the reference pages\n- Simple point.\n### Beyond the reference pages\n- Tutor.", | |
| slides=[], covered_by_slides=False) | |
| with mock.patch.object(app, "_generate", return_value=rewritten.model_dump_json()) as generate: | |
| outputs = drain(app.run_turn("simplify", "", session)) | |
| sent = generate.call_args[0][0][-1].parts[0].text # the request sent to Gemini | |
| self.assertIn("Previous answer:", sent) | |
| self.assertNotIn("Beyond", sent) # the old section is not sent ... | |
| self.assertNotIn("Beyond", outputs[-1][0][-1]["content"]) # ... nor shown if Gemini adds one | |
| _, state = app._restore([{"role": "assistant", "mode": "clinical", "topic": "t", "source": "slides", | |
| "content": "A" + app.SOURCE_MARK + "Course lecture slides." + app.OLD_TUTOR_NOTE}]) | |
| self.assertEqual(state["last_note"], "Course lecture slides.") | |
| def test_simplify_of_a_clinical_answer_from_both_sources_checks_slide_citations(self): | |
| session = signed_in_session(self.store) | |
| session.state.update(topic="knee", last_answer="prev", last_source="slides+web", last_note="note") | |
| rewritten = TutorReply(answer="Point (Lecture 1, slide 999) and (Lecture 1, slide 2).", slides=[], covered_by_slides=False) | |
| with mock.patch.object(app, "gemini_rewrite", return_value=rewritten): | |
| drain(app.run_turn("simplify", "", session)) | |
| self.assertIn(app.NOT_FOUND_TAG, session.messages[-1]["content"]) | |
| self.assertEqual(session.state["slides"], [[1, 2]]) | |
| def test_clinical_says_when_no_article_exists(self): | |
| slide_reply = TutorReply(answer=f"- A point (Lecture {self.slide.lecture}, slide {self.slide.number}).", slides=[], | |
| covered_by_slides=True, search_topics=["superior thoracic aperture"]) | |
| session, outputs, rows, *_ = self.clinical(slide_reply, [], None) | |
| body, note = app._split_source(outputs[-1][0][-1]["content"]) | |
| self.assertIn("No TeachMeAnatomy or Kenhub article about this topic was found.", body) | |
| self.assertTrue(note.endswith("No TeachMeAnatomy or Kenhub article about this topic was found.")) | |
| def test_follow_ups_and_quick_actions_search_the_slides_of_the_known_structures(self): | |
| state = {"search_topics": ["knee joint"]} | |
| self.assertEqual(app.slide_query("ask", "How is it injured?", state), "How is it injured? knee joint") | |
| self.assertEqual(app.slide_query("ask", "What muscles form the rotator cuff?", state), "What muscles form the rotator cuff?") | |
| self.assertEqual(app.slide_query("quiz", "How is it injured?", state), "How is it injured? knee joint") | |
| self.assertEqual(app.slide_query("ask", "How is it injured?", {}), "How is it injured?") | |
| upper_limb = {n for n, lec in app.LECTURES.items() if any(w in lec.title.lower() for w in ("shoulder", "brachial", "arm"))} | |
| lectures = {s.lecture for s in app.retrieve_slides(app.slide_query("ask", "How is it injured?", state))} | |
| self.assertFalse(lectures & upper_limb, lectures) | |
| def test_beyond_sections_are_removed(self): | |
| text = ("### From the reference pages\n- Point (TeachMeAnatomy: The Knee Joint).\n\n" | |
| "### Beyond the reference pages\nThe following points are standard clinical anatomy.\n- Own knowledge\n" | |
| "#### Sub point\n- more\n### Summary\nKept.\n\n**Beyond the slides**\n- tutor point\n### Next\nKept too.") | |
| self.assertEqual(app._drop_beyond_sections(text), | |
| "### From the reference pages\n- Point (TeachMeAnatomy: The Knee Joint).\n\n### Summary\nKept.\n\n### Next\nKept too.") | |
| self.assertEqual(app._drop_beyond_sections("The rotator cuff lies beyond the glenoid."), "The rotator cuff lies beyond the glenoid.") | |
| def test_prompts_never_ask_for_outside_knowledge(self): | |
| for text in [app.TASK_PROMPTS["clinical"], app.WEB_TASK_PROMPTS["clinical"], app.WEB_SYSTEM_PROMPT, app.TASK_PROMPTS["simplify"]]: | |
| self.assertNotIn("Beyond the slides'", text) | |
| self.assertNotIn("standard clinical anatomy", text) | |
| self.assertNotIn('write "the reference pages" instead of "the slides"', app.WEB_SYSTEM_PROMPT) | |
| self.assertIn("Never answer from outside knowledge", app.SYSTEM_PROMPT) | |
| def test_ignored_clicks_are_still_counted(self): | |
| session = signed_in_session(self.store) | |
| with mock.patch.object(app, "gemini_reply") as call: | |
| drain(app.run_turn("ask", " ", session)) | |
| call.assert_not_called() | |
| self.assertEqual(self.store.events(), ["ask"]) | |
| self.assertEqual(session.messages, []) | |
| def test_show_image_keeps_only_cited_slides(self): | |
| session = signed_in_session(self.store) | |
| session.state["topic"] = "topic" | |
| other = app.LECTURES[self.slide.lecture].slides[-1] | |
| reply = TutorReply(answer=f"Slide (Lecture {self.slide.lecture}, slide {self.slide.number}) shows it.", | |
| slides=[SlideRef(lecture=other.lecture, slide=other.number), SlideRef(lecture=self.slide.lecture, slide=self.slide.number), | |
| SlideRef(lecture=self.figure.lecture, slide=self.figure.number)], | |
| covered_by_slides=True) | |
| with mock.patch.object(app, "gemini_reply", return_value=reply): | |
| self._show(session) | |
| self.assertEqual(session.state["slides"], [[self.slide.lecture, self.slide.number], [self.figure.lecture, self.figure.number]]) | |
| def test_simplify_and_show_do_not_fall_back(self): | |
| session = signed_in_session(self.store) | |
| session.state.update(topic="retina", last_answer="prev", last_source="web", | |
| last_note="The lecture slides do not cover this request. Answered from Kenhub: [Retina](u).") | |
| with mock.patch.object(app, "gemini_rewrite", return_value=TutorReply(answer="Simpler.", slides=[], covered_by_slides=False)) as rewrite, \ | |
| mock.patch.object(app, "gemini_reply") as slides_call, \ | |
| mock.patch.object(web_sources, "find_pages") as find: | |
| outputs = drain(app.run_turn("simplify", "", session)) | |
| find.assert_not_called() | |
| slides_call.assert_not_called() # a web answer is rewritten without the slide prompt | |
| self.assertIn("Previous answer:\nprev", rewrite.call_args[0][0]) | |
| self.assertEqual(outputs[0][0][-1]["content"], "Rewriting the previous answer...") | |
| self.assertTrue(outputs[-1][0][-1]["content"].endswith("Answered from Kenhub: [Retina](u).")) | |
| rows = [c for c in self.store.calls if c[0] == "append_messages"][0][3] | |
| self.assertEqual(rows[1]["source"], "web") | |
| def test_quiz_answer_key_is_collapsed_before_the_source_note(self): | |
| session = signed_in_session(self.store) | |
| session.state["topic"] = "brachial plexus" | |
| reply = TutorReply(answer=f"1. Q?\n\n### Answer key\n1. A (Lecture {self.slide.lecture}, slide {self.slide.number})", slides=[], covered_by_slides=True) | |
| with mock.patch.object(app, "gemini_reply", return_value=reply): | |
| outputs = drain(app.run_turn("quiz", "", session)) | |
| content = outputs[-1][0][-1]["content"] | |
| self.assertLess(content.index("</details>"), content.index("**Source:**")) | |
| self.assertEqual(session.state["last_answer"], "") # quizzes never become the answer to simplify | |
| def test_error_is_shown_without_source_note(self): | |
| session = signed_in_session(self.store) | |
| with mock.patch.object(app, "gemini_reply", side_effect=app.TutorError("The tutor is busy right now.")): | |
| outputs = drain(app.run_turn("ask", "Q?", session)) | |
| final = outputs[-1][0][-1] | |
| self.assertEqual(final, {"role": "assistant", "content": "The tutor is busy right now."}) | |
| self.assertEqual(session.state["history"], []) | |
| self.assertFalse(session.busy) | |
| rows = [c for c in self.store.calls if c[0] == "append_messages"][0][3] | |
| self.assertEqual(rows[1]["source"], "error") | |
| def test_turns_need_a_signed_in_user(self): | |
| session = app.Session() | |
| with mock.patch.object(app, "gemini_reply") as call, mock.patch.object(app.gr, "Warning") as warning: | |
| outputs = drain(app.run_turn("ask", "Q?", session)) | |
| call.assert_not_called() | |
| self.assertEqual(outputs, [app.NOOP_TURN]) # the visible chat is left as it is | |
| warning.assert_called_once_with(app.SIGNED_OUT_TEXT) | |
| self.assertEqual(session.messages, []) | |
| with mock.patch.object(app, "get_session", return_value=session), mock.patch.object(app.gr, "Warning"): | |
| shown = list(app.on_show_image("", "", 0, SimpleNamespace(session_hash="x"))) | |
| self.assertEqual(shown, [app.NOOP_SHOW]) | |
| def test_turn_signs_the_student_back_in_after_a_restart(self): | |
| self.store.users["jdoe"] = {"user_id": "jdoe", "email": "jdoe@example.com"} | |
| self.store.messages = [{"user_id": "jdoe", "role": "user", "content": "Q1", "mode": "ask", "topic": "t1", "source": None}, | |
| {"user_id": "jdoe", "role": "assistant", "content": "A1", "mode": "ask", "topic": "t1", "source": "slides"}] | |
| session = app.Session() # what a restarted server knows about the tab | |
| with mock.patch.object(app, "gemini_reply", return_value=self.slides_reply()): | |
| outputs = drain(app.run_turn("ask", "Q2?", session, ("jdoe", "JDoe@example.com", 1791000000000))) | |
| self.assertEqual(session.user_id, "jdoe") | |
| self.assertEqual([m["content"] for m in outputs[-1][0]][:3], ["Q1", "A1", "Q2?"]) | |
| self.assertIn("resume", self.store.events()) | |
| self.assertEqual([c[0] for c in self.store.calls][0], "start_session") # no other tab: no waiting | |
| def test_no_automatic_sign_in_from_the_sign_in_form_or_a_wrong_email(self): | |
| self.store.users["jdoe"] = {"user_id": "jdoe", "email": "jdoe@example.com"} | |
| for creds in [("jdoe", "jdoe@example.com", 0), ("jdoe", "other@example.com", 5), ("nobody", "jdoe@example.com", 5)]: | |
| session = app.Session() | |
| with mock.patch.object(app, "gemini_reply") as call, mock.patch.object(app.gr, "Warning"): | |
| drain(app.run_turn("ask", "Q?", session, creds)) | |
| call.assert_not_called() | |
| self.assertIsNone(session.user_id, creds) | |
| def test_show_image_shows_thinking_while_choosing(self): | |
| session = signed_in_session(self.store) | |
| session.state["topic"] = "topic" | |
| reply = TutorReply(answer=f"Slide (Lecture {self.slide.lecture}, slide {self.slide.number}) shows it.", | |
| slides=[SlideRef(lecture=self.slide.lecture, slide=self.slide.number)], covered_by_slides=True) | |
| with mock.patch.object(app, "gemini_reply", return_value=reply): | |
| outputs = self._show(session) | |
| self.assertEqual(outputs[0][0][-1]["metadata"]["status"], "pending") | |
| self.assertEqual(session.state["view"], [self.slide.lecture, self.slide.number]) | |
| self.assertIn("**Source:** Course lecture slides:", outputs[-1][0][-1]["content"]) | |
| self.assertEqual(self.store.events(), ["show_image"]) | |
| def _show(self, session): | |
| request = SimpleNamespace(session_hash="tab-show") | |
| with mock.patch.object(app, "get_session", return_value=session): | |
| return list(app.on_show_image("", "", 0, request)) | |
| class RestoreTests(unittest.TestCase): | |
| def test_restore_rebuilds_chat_and_state(self): | |
| rows = [ | |
| {"role": "user", "content": "Q1", "mode": "ask", "topic": "t1", "source": None}, | |
| {"role": "assistant", "content": "A1" + app.SOURCE_MARK + "Course lecture slides.", "mode": "ask", "topic": "t1", "source": "slides"}, | |
| {"role": "user", "content": "Quiz me on: t1", "mode": "quiz", "topic": "t1", "source": None}, | |
| {"role": "assistant", "content": "Quiz" + app.SOURCE_MARK + "Course lecture slides.", "mode": "quiz", "topic": "t1", "source": "slides"}, | |
| ] | |
| messages, state = app._restore(rows) | |
| self.assertEqual(messages[1], {"role": "assistant", "content": rows[1]["content"]}) | |
| self.assertEqual([h["role"] for h in state["history"]], ["user", "model", "user", "model"]) | |
| self.assertEqual(state["history"][1]["text"], "A1") | |
| self.assertEqual(state["topic"], "t1") | |
| self.assertEqual(state["last_answer"], "A1") # the quiz is skipped | |
| self.assertEqual(state["last_note"], "Course lecture slides.") | |
| self.assertEqual(state["last_source"], "slides") | |
| self.assertIsNone(state["view"]) | |
| def test_restore_skips_failed_turns_and_show_notes(self): | |
| rows = [ | |
| {"role": "user", "content": "Q1", "mode": "ask", "topic": "t1", "source": None}, | |
| {"role": "assistant", "content": "A1" + app.SOURCE_MARK + "Not found in the lecture slides, and no matching article was found on TeachMeAnatomy or Kenhub.", | |
| "mode": "ask", "topic": "t1", "source": "none"}, | |
| {"role": "assistant", "content": "No slide in the lecture files illustrates this topic.", "mode": "show", "topic": "t1", "source": "none"}, | |
| {"role": "user", "content": "Q2", "mode": "ask", "topic": "t2", "source": None}, | |
| {"role": "assistant", "content": "The tutor is busy right now.", "mode": "ask", "topic": "t2", "source": "error"}, | |
| ] | |
| messages, state = app._restore(rows) | |
| self.assertEqual(len(messages), 5) # everything is shown again | |
| self.assertEqual([h["text"] for h in state["history"]], ["Q1", "A1"]) | |
| self.assertEqual(state["topic"], "t1") | |
| self.assertEqual((state["last_answer"], state["last_source"]), ("A1", "none")) | |
| class SignInTests(unittest.TestCase): | |
| def setUp(self): | |
| self.store = FakeStore() | |
| self.patch = mock.patch.object(app, "STORE", self.store) | |
| self.patch.start() | |
| self.request = SimpleNamespace(session_hash="tab-1") | |
| app.SESSIONS.clear() | |
| def tearDown(self): | |
| self.patch.stop() | |
| app.SESSIONS.clear() | |
| def test_create_account_then_sign_in_restores_history(self): | |
| out = app.on_create_account("JDoe", "JDoe@Example.com", self.request) | |
| self.assertFalse(out[0]["visible"]) # login panel hidden | |
| self.assertTrue(out[1]["visible"]) # app shown | |
| self.assertIn("Signed in as **jdoe** (jdoe@example.com).", out[3]) | |
| self.assertEqual(self.store.users["jdoe"]["email"], "jdoe@example.com") | |
| self.assertEqual(self.store.events(), ["create_account"]) | |
| self.assertEqual([c[0] for c in self.store.calls][:2], ["start_session", "log_event"]) | |
| again = app.on_create_account("jdoe", "other@example.com", self.request) | |
| self.assertTrue(again[0]["visible"]) | |
| self.assertIn("already taken", again[2]) | |
| session = app.get_session(self.request) | |
| session.messages = [{"role": "user", "content": "Q"}, {"role": "assistant", "content": "A" + app.SOURCE_MARK + "Course lecture slides."}] | |
| app._save(session, [{"role": "user", "content": "Q", "mode": "ask", "topic": "Q", "source": None}, | |
| {"role": "assistant", "content": session.messages[1]["content"], "mode": "ask", "topic": "Q", "source": "slides"}]) | |
| stored = session.messages[1]["content"] | |
| out = app.on_sign_out(self.request) | |
| self.assertTrue(out[0]["visible"]) | |
| self.assertEqual(out[5], []) | |
| self.assertEqual(out[-3:], ("", "", 0)) # user ID and email fields cleared, page shows sign-in | |
| self.assertEqual(len(out), len(app.login_outputs)) | |
| self.assertIsNone(session.user_id) | |
| self.assertEqual([c[0] for c in self.store.calls][-2:], ["log_event", "end_session"]) | |
| self.assertEqual(self.store.events()[-1], "sign_out") | |
| wrong = app.on_sign_in("jdoe", "wrong@example.com", self.request) | |
| self.assertIn("does not match", wrong[2]) | |
| unknown = app.on_sign_in("nobody", "a@b.co", self.request) | |
| self.assertIn("No account has this user ID", unknown[2]) | |
| out = app.on_sign_in("jdoe", "jdoe@example.com", self.request) | |
| self.assertFalse(out[0]["visible"]) | |
| self.assertIn("2 earlier messages restored", out[3]) | |
| self.assertEqual([m["content"] for m in out[5]], ["Q", stored]) | |
| self.assertEqual(session.state["topic"], "Q") | |
| self.assertTrue(out[-5]["interactive"]) # Ask enabled | |
| self.assertTrue(out[-9]["interactive"]) # quick actions enabled: a topic exists | |
| self.assertEqual(out[-1], session.version) | |
| def test_validation_messages(self): | |
| self.assertIn("user ID", app.on_sign_in("x", "a@b.co", self.request)[2]) | |
| self.assertIn("valid email", app.on_sign_in("jdoe", "not-an-email", self.request)[2]) | |
| with mock.patch.object(app, "STORE", FakeStore(enabled=False)): | |
| self.assertIn("not configured", app.on_sign_in("jdoe", "a@b.co", self.request)[2]) | |
| def sync(self, version, user="jdoe", email="jdoe@example.com"): | |
| return app.on_sync(version, user, email, self.request) | |
| def test_sync_does_nothing_for_the_sign_in_form_or_an_up_to_date_page(self): | |
| self.store.users["jdoe"] = {"user_id": "jdoe", "email": "jdoe@example.com"} | |
| self.assertEqual(self.sync(0), app.NOOP_SYNC) # page shows the sign-in form: never signs in | |
| self.assertIsNone(app.get_session(self.request).user_id) | |
| out = app.on_sign_in("jdoe", "jdoe@example.com", self.request) | |
| self.assertEqual(self.sync(out[-1]), app.NOOP_SYNC) | |
| def test_sync_delivers_an_answer_the_page_missed(self): | |
| import threading | |
| app.on_create_account("jdoe", "jdoe@example.com", self.request) | |
| session = app.get_session(self.request) | |
| shown = session.version # what the page has before it asks | |
| slide, figure = first_text_slide_with_figure() | |
| reply = TutorReply(answer=f"Fact (Lecture {slide.lecture}, slide {slide.number}).", slides=[], covered_by_slides=True) | |
| release = threading.Event() | |
| def slow_reply(*args): | |
| release.wait(5) | |
| return reply | |
| with mock.patch.object(app, "gemini_reply", side_effect=slow_reply): | |
| turn = app.run_turn("ask", "Q?", session, ("jdoe", "jdoe@example.com", shown)) | |
| pending = next(turn) # the page receives the Thinking message ... | |
| self.assertFalse(pending[-3]["interactive"]) | |
| self.assertEqual(self.sync(shown), app.NOOP_SYNC) # ... the answer is being written: nothing to send | |
| turn.close() # ... then its connection drops: Gradio drops the event | |
| release.set() | |
| for _ in range(100): # the answer is still finished and saved | |
| if not session.busy: | |
| break | |
| app.time.sleep(0.02) | |
| self.assertFalse(session.busy) | |
| self.assertIn("append_messages", [c[0] for c in self.store.calls]) | |
| out = self.sync(shown) | |
| self.assertEqual(len(out), len(app.sync_outputs)) | |
| self.assertEqual([m["content"].split(app.SOURCE_MARK)[0] for m in out[0]], ["Q?", reply.answer]) | |
| self.assertEqual(out[2], session.version) | |
| self.assertTrue(all(update["interactive"] for update in out[-6:])) # buttons usable again | |
| self.assertEqual(self.sync(out[2]), app.NOOP_SYNC) # nothing more to send | |
| def test_a_multi_step_answer_finishes_after_the_page_is_gone(self): | |
| session = app.get_session(self.request) | |
| app.on_create_account("jdoe", "jdoe@example.com", self.request) | |
| page = web_sources.Page("Kenhub", "https://www.kenhub.com/en/library/anatomy/retina", "Retina", "text") | |
| slides_reply = TutorReply(answer="The slides do not cover this.", slides=[], covered_by_slides=False, search_topics=["retina"]) | |
| with mock.patch.object(app, "gemini_reply", return_value=slides_reply), \ | |
| mock.patch.object(web_sources, "find_pages", return_value=([page], "")), \ | |
| mock.patch.object(app, "gemini_web_reply", return_value=WebReply(answer="Ten layers (Kenhub: Retina).", covered=True, used_urls=[page.url])): | |
| turn = app.run_turn("ask", "Layers of the retina?", session, ()) | |
| next(turn) # first progress step reaches the page | |
| turn.close() # the page is gone before the second step | |
| for _ in range(100): | |
| if not session.busy: | |
| break | |
| app.time.sleep(0.02) | |
| saved = [c for c in self.store.calls if c[0] == "append_messages"] | |
| self.assertEqual(len(saved), 1) | |
| self.assertEqual(saved[0][3][1]["source"], "web") | |
| self.assertTrue(session.messages[-1]["content"].startswith("Ten layers (Kenhub: Retina).")) | |
| def test_sync_restores_the_student_after_a_server_restart(self): | |
| self.store.users["jdoe"] = {"user_id": "jdoe", "email": "jdoe@example.com"} | |
| self.store.messages = [{"user_id": "jdoe", "role": "user", "content": "Q1", "mode": "ask", "topic": "t1", "source": None}, | |
| {"user_id": "jdoe", "role": "assistant", "content": "A1", "mode": "ask", "topic": "t1", "source": "slides"}] | |
| self.assertEqual(self.sync(1791000000000, email="wrong@example.com"), app.NOOP_SYNC) | |
| out = self.sync(1791000000000) | |
| session = app.get_session(self.request) | |
| self.assertEqual(session.user_id, "jdoe") | |
| self.assertEqual([m["content"] for m in out[0]], ["Q1", "A1"]) | |
| self.assertIn("The site restarted; your chat was restored.", out[3]) | |
| self.assertEqual(self.store.events(), ["resume"]) | |
| def test_reloaded_tab_signs_itself_back_in_and_keeps_the_draft(self): | |
| self.store.users["jdoe"] = {"user_id": "jdoe", "email": "jdoe@example.com"} | |
| self.store.messages = [{"user_id": "jdoe", "role": "user", "content": "Q1", "mode": "ask", "topic": "t1", "source": None}, | |
| {"user_id": "jdoe", "role": "assistant", "content": "A1", "mode": "ask", "topic": "t1", "source": "slides"}] | |
| steps = list(app.on_page_load("JDoe", "jdoe@example.com", "half-typed question", self.request)) | |
| self.assertEqual(steps[0][2], "Reconnecting and restoring your chat...") | |
| final = steps[-1] | |
| self.assertEqual(len(final), len(app.login_outputs)) | |
| self.assertFalse(final[0]["visible"]) # sign-in form hidden | |
| self.assertTrue(final[1]["visible"]) # chat shown | |
| self.assertIn("Reconnected; 2 messages restored.", final[3]) | |
| self.assertEqual([m["content"] for m in final[5]], ["Q1", "A1"]) | |
| self.assertEqual(final[6], "half-typed question") | |
| self.assertEqual(final[-3:], ("jdoe", "jdoe@example.com", app.get_session(self.request).version)) | |
| self.assertEqual(self.store.events(), ["resume"]) | |
| def test_reloaded_tab_drops_a_draft_that_was_already_answered(self): | |
| self.store.users["jdoe"] = {"user_id": "jdoe", "email": "jdoe@example.com"} | |
| self.store.messages = [{"user_id": "jdoe", "role": "user", "content": "Q1", "mode": "ask", "topic": "t1", "source": None}, | |
| {"user_id": "jdoe", "role": "assistant", "content": "A1", "mode": "ask", "topic": "t1", "source": "slides"}] | |
| final = list(app.on_page_load("jdoe", "jdoe@example.com", "Q1", self.request))[-1] | |
| self.assertEqual(final[6], "") | |
| def test_page_load_without_a_remembered_sign_in_shows_the_form(self): | |
| self.assertEqual(list(app.on_page_load("", "", "", self.request)), [app.NOOP_LOGIN]) | |
| steps = list(app.on_page_load("jdoe", "wrong@example.com", "", self.request)) | |
| self.assertTrue(steps[-1][0]["visible"]) | |
| self.assertIn("could not be restored automatically", steps[-1][2]) | |
| self.assertIsNone(app.get_session(self.request).user_id) | |
| def test_reloaded_page_waits_for_the_old_tab_and_ordinary_sign_ins_do_not(self): | |
| import threading | |
| self.store.users["jdoe"] = {"user_id": "jdoe", "email": "jdoe@example.com"} | |
| old = app.get_session(SimpleNamespace(session_hash="old-tab")) | |
| old.user_id, old.busy = "jdoe", True | |
| old.messages = [{"role": "user", "content": "Pending question?"}] | |
| def finish(): | |
| app.time.sleep(1.5) | |
| old.saved_at, old.busy = app.time.monotonic(), False | |
| threading.Thread(target=finish).start() | |
| started = app.time.monotonic() | |
| steps = list(app.on_page_load("jdoe", "jdoe@example.com", "Pending question?", self.request)) | |
| self.assertGreaterEqual(app.time.monotonic() - started, 1.0) # waited for the old tab | |
| self.assertGreater(len(steps), 2) # by polling, one step per second | |
| names = [c[0] for c in self.store.calls] | |
| self.assertLess(names.index("flush"), names.index("start_session")) # its answer reaches Supabase first | |
| # A plain sign-in on another tab does not wait for the shared write queue. | |
| self.store.calls.clear() | |
| app.on_sign_in("jdoe", "jdoe@example.com", SimpleNamespace(session_hash="tab-2")) | |
| self.assertIn("flush", [c[0] for c in self.store.calls]) # the old tab saved just now | |
| old.saved_at = 0.0 | |
| self.store.calls.clear() | |
| app.on_sign_in("jdoe", "jdoe@example.com", SimpleNamespace(session_hash="tab-3")) | |
| self.assertNotIn("flush", [c[0] for c in self.store.calls]) | |
| def test_an_answer_still_being_written_reaches_the_reloaded_page_later(self): | |
| self.store.users["jdoe"] = {"user_id": "jdoe", "email": "jdoe@example.com"} | |
| old = app.get_session(SimpleNamespace(session_hash="old-tab")) | |
| old.user_id, old.busy = "jdoe", True | |
| old.messages = [{"role": "user", "content": "Slow question?"}] | |
| with mock.patch.object(app, "ANSWER_WAIT_SECONDS", 0.2): | |
| final = list(app.on_page_load("jdoe", "jdoe@example.com", "Slow question?", self.request))[-1] | |
| self.assertEqual(final[6], "") # not put back: still being answered | |
| session = app.get_session(self.request) | |
| self.assertTrue(session.awaiting) | |
| self.assertEqual(self.sync(final[-1]), app.NOOP_SYNC) # still being written | |
| self.store.messages = [{"user_id": "jdoe", "role": "user", "content": "Slow question?", "mode": "ask", "topic": "s", "source": None}, | |
| {"user_id": "jdoe", "role": "assistant", "content": "Slow answer.", "mode": "ask", "topic": "s", "source": "slides"}] | |
| old.busy, old.saved_at = False, app.time.monotonic() | |
| out = self.sync(final[-1]) | |
| self.assertEqual([m["content"] for m in out[0]], ["Slow question?", "Slow answer."]) | |
| self.assertFalse(session.awaiting) | |
| def test_a_sign_in_made_while_reconnecting_is_not_overwritten(self): | |
| self.store.users["jdoe"] = {"user_id": "jdoe", "email": "jdoe@example.com"} | |
| self.store.users["bob"] = {"user_id": "bob", "email": "bob@example.com"} | |
| old = app.get_session(SimpleNamespace(session_hash="old-tab")) | |
| old.user_id, old.busy = "jdoe", True | |
| load = app.on_page_load("jdoe", "jdoe@example.com", "", self.request) | |
| next(load) # "Reconnecting..." on the sign-in form | |
| app.on_sign_in("bob", "bob@example.com", self.request) # bob signs in on this tab meanwhile | |
| old.busy = False | |
| steps = list(load) | |
| self.assertEqual(steps[-1], app.NOOP_LOGIN) | |
| self.assertEqual(app.get_session(self.request).user_id, "bob") | |
| def test_sign_out_is_final_for_requests_already_in_flight(self): | |
| out = app.on_sign_in("jdoe", "jdoe@example.com", self.request) if "jdoe" in self.store.users else None | |
| self.store.users["jdoe"] = {"user_id": "jdoe", "email": "jdoe@example.com"} | |
| out = app.on_sign_in("jdoe", "jdoe@example.com", self.request) | |
| version = out[-1] | |
| app.on_sign_out(self.request) | |
| self.assertEqual(self.sync(version), app.NOOP_SYNC) # a tick sent before Sign out | |
| with mock.patch.object(app, "gemini_reply") as call, mock.patch.object(app.gr, "Warning"): | |
| drain(app.run_turn("ask", "Q?", app.get_session(self.request), ("jdoe", "jdoe@example.com", version))) | |
| call.assert_not_called() | |
| self.assertIsNone(app.get_session(self.request).user_id) | |
| app.on_sign_in("jdoe", "jdoe@example.com", self.request) # signing in again works | |
| self.assertEqual(app.get_session(self.request).user_id, "jdoe") | |
| def test_sign_out_while_an_answer_is_written_says_so(self): | |
| app.on_create_account("jdoe", "jdoe@example.com", self.request) | |
| session = app.get_session(self.request) | |
| session.busy = True | |
| with mock.patch.object(app.gr, "Warning") as warning: | |
| self.assertEqual(app.on_sign_out(self.request), app.NOOP_LOGIN) | |
| warning.assert_called_once() | |
| self.assertEqual(session.user_id, "jdoe") | |
| def test_evicting_a_closed_visit_does_not_close_it_again(self): | |
| app.on_create_account("jdoe", "jdoe@example.com", self.request) | |
| app.forget_session(self.request) | |
| with mock.patch.object(app, "MAX_SESSIONS", 1): | |
| app.get_session(SimpleNamespace(session_hash="someone-else")) | |
| self.assertEqual([c[0] for c in self.store.calls].count("end_session"), 1) | |
| def test_time_on_site_and_new_visit_after_the_connection_drops(self): | |
| app.on_create_account("jdoe", "jdoe@example.com", self.request) | |
| session = app.get_session(self.request) | |
| first = session.session_id | |
| self.sync(session.version) | |
| self.assertNotIn("heartbeat", [c[0] for c in self.store.calls]) # written at most every HEARTBEAT_SECONDS | |
| session.last_beat -= app.HEARTBEAT_SECONDS | |
| self.sync(session.version) | |
| self.assertEqual(self.store.calls[-1], ("heartbeat", first)) | |
| app.forget_session(self.request) # laptop asleep: the visit is closed | |
| self.assertEqual(self.store.calls[-1], ("end_session", first)) | |
| app.forget_session(self.request) # closed once only | |
| self.assertEqual([c[0] for c in self.store.calls].count("end_session"), 1) | |
| self.assertEqual(session.user_id, "jdoe") # still signed in on the server | |
| self.sync(session.version) # back: a new visit starts | |
| self.assertNotEqual(session.session_id, first) | |
| self.assertEqual(self.store.calls[-1], ("start_session", session.session_id, "jdoe")) | |
| class WebSourceTests(unittest.TestCase): | |
| def test_article_filters(self): | |
| tma, kenhub = web_sources.SITES | |
| self.assertTrue(tma.article.match("https://teachmeanatomy.info/upper-limb/nerves/brachial-plexus/")) | |
| self.assertFalse(tma.article.match("https://teachmeanatomy.info/upper-limb/nerves/")) # category index | |
| self.assertFalse(tma.article.match("https://teachmeanatomy.info/encyclopaedia/d/")) | |
| self.assertFalse(tma.article.match("https://teachmeanatomy.info/shop/")) | |
| self.assertFalse(tma.article.match("https://teachmeanatomy.info/es/miembro-superior/nervios/plexo-braquial/")) | |
| self.assertFalse(tma.article.match("https://teachmeanatomy.info/3d-model/upper-limb/")) | |
| self.assertTrue(kenhub.article.match("https://www.kenhub.com/en/library/anatomy/brachial-plexus")) | |
| self.assertTrue(kenhub.article.match("https://www.kenhub.com/en/library/education/the-human-anatomy")) | |
| self.assertFalse(kenhub.article.match("https://www.kenhub.com/en/library/learning-strategies/how-to-study")) | |
| self.assertFalse(kenhub.article.match("https://www.kenhub.com/de/library/anatomy/plexus-brachialis")) | |
| def test_headings_stay_on_one_line_and_clinical_sections_survive_the_cut(self): | |
| self.assertEqual(web_sources.article_text("<article><h2>\r\n Clinical\r\n significance </h2><p>x</p></article>"), | |
| "## Clinical significance\n\nx") | |
| text = "## Anatomy\n" + "a" * 20000 + "\n## Clinical significance\nACL injury.\n### Terrible triad\nThree.\n## Sources\nRefs." | |
| cut = web_sources.excerpt(text, 1000) | |
| self.assertLessEqual(len(cut), 1000) | |
| self.assertTrue(cut.startswith("## Anatomy\naaa")) | |
| self.assertTrue(cut.endswith("[...]\n\n## Clinical significance\nACL injury.\n### Terrible triad\nThree.")) | |
| self.assertEqual(web_sources.excerpt("short", 1000), "short") | |
| def test_article_text_keeps_the_article_only(self): | |
| page = ("<html><head><title>Retina | Kenhub</title><script>x()</script></head><body><nav>Menu</nav>" | |
| "<article><h1>Retina</h1><p>The retina has <b>ten</b> layers.</p><ul><li>Layer one</li><li>Layer two</li></ul>" | |
| "<figure>caption</figure></article><footer>Footer</footer></body></html>") | |
| text = web_sources.article_text(page) | |
| self.assertEqual(text, "# Retina\n\nThe retina has ten layers.\n\n- Layer one\n- Layer two") | |
| self.assertEqual(web_sources.page_title(page, "Kenhub"), "Retina") | |
| self.assertEqual(web_sources.page_title("<title>Eye | Kenhub</title>", "Kenhub"), "Eye") | |
| def catalog(urls): | |
| entries = [("TeachMeAnatomy" if "teachmeanatomy" in u else "Kenhub", u) for u in urls] | |
| return web_sources.Catalog( | |
| data=(entries, web_sources.BM25([web_sources.url_tokens(u) for _, u in entries]), | |
| [frozenset(web_sources.words(web_sources.url_path(u))) for _, u in entries]), | |
| next_load=float("inf")) | |
| def test_find_pages_matches_each_structure_and_never_a_generic_word(self): | |
| tma, kh = "https://teachmeanatomy.info/", "https://www.kenhub.com/en/library/anatomy/" | |
| catalog = self.catalog([ | |
| tma + "lower-limb/joints/knee-joint/", tma + "lower-limb/joints/ankle-joint/", tma + "upper-limb/joints/shoulder/", | |
| tma + "lower-limb/areas/tarsal-tunnel/", tma + "upper-limb/areas/carpal-tunnel/", | |
| kh + "the-knee-joint", kh + "the-ankle-joint", kh + "upper-and-lower-brachial-plexus-injury", kh + "colles-fracture", | |
| kh + "carpal-tunnel", kh + "superior-thoracic-artery", kh + "retina", kh + "central-retinal-artery", kh + "medial-meniscus", | |
| ]) | |
| def find(topics): | |
| def load(page): # the page text names the structure in its URL | |
| return web_sources.Page(page.site, page.url, "T", web_sources.url_path(page.url)) | |
| with mock.patch.object(web_sources, "CATALOG", catalog), mock.patch.object(web_sources, "load_page", side_effect=load): | |
| pages, error = web_sources.find_pages(topics) | |
| self.assertEqual(error, "") | |
| return [p.url.replace(tma, "TMA:").replace(kh, "KH:") for p in pages] | |
| self.assertEqual(find(["knee joint", "ankle joint"]), | |
| ["TMA:lower-limb/joints/knee-joint/", "TMA:lower-limb/joints/ankle-joint/", "KH:the-knee-joint", "KH:the-ankle-joint"]) | |
| self.assertEqual(find(["knee joint injuries", "ankle fracture"]), # injury/fracture never select a page | |
| ["TMA:lower-limb/joints/knee-joint/", "TMA:lower-limb/joints/ankle-joint/", "KH:the-knee-joint", "KH:the-ankle-joint"]) | |
| self.assertEqual(find(["injury"]), []) | |
| self.assertEqual(find(["tarsal tunnel"]), ["TMA:lower-limb/areas/tarsal-tunnel/"]) # not the carpal tunnel | |
| self.assertEqual(find(["superior thoracic aperture"]), []) # not the superior thoracic artery | |
| self.assertEqual(find(["retinal layers"]), ["KH:retina"]) # the retina before the artery | |
| self.assertEqual(find(["menisci"]), ["KH:medial-meniscus"]) | |
| self.assertEqual(find("knee"), ["TMA:lower-limb/joints/knee-joint/", "KH:the-knee-joint"]) | |
| self.assertLessEqual(len(find(["knee", "ankle", "shoulder", "retina"])), web_sources.MAX_PAGES) | |
| with mock.patch.object(web_sources, "CATALOG", catalog), mock.patch.object(web_sources, "load_page", return_value=None): | |
| pages, error = web_sources.find_pages(["retina"]) | |
| self.assertEqual(pages, []) | |
| self.assertEqual(error, "the matching TeachMeAnatomy and Kenhub pages could not be downloaded") | |
| class WordTests(unittest.TestCase): | |
| def test_singular_and_joint_names(self): | |
| self.assertEqual(web_sources.words("arches sinuses menisci vertebrae trapezius trapezium pelvis processes talocrural"), | |
| ["arch", "sinus", "meniscus", "vertebra", "trapezius", "trapezium", "pelvis", "process", "ankle"]) | |
| self.assertFalse(web_sources.same_word("trapezius", "trapezium")) | |
| self.assertTrue(web_sources.same_word("retina", "retinal")) | |
| def test_type_word_must_agree_and_be_mentioned(self): | |
| kh = "https://www.kenhub.com/en/library/anatomy/" | |
| catalog = WebSourceTests.catalog([kh + "the-deltoid-muscle", kh + "the-ankle-joint"]) | |
| entries, index, slug_words = catalog.data | |
| self.assertEqual(web_sources.topic_matches("deltoid ligament", entries, index, slug_words), []) | |
| self.assertEqual(web_sources.topic_matches("deltoid muscle", entries, index, slug_words), [0]) | |
| page = web_sources.Page("Kenhub", kh + "x", "Deltoid", "The deltoid is a muscle of the shoulder.") | |
| self.assertFalse(web_sources.names_structure("deltoid ligament", page)) | |
| page.text = "The medial (deltoid) ligament of the ankle." | |
| self.assertTrue(web_sources.names_structure("deltoid ligament", page)) | |
| self.assertTrue(web_sources.names_structure("brachial plexus", page)) # no type word: always accepted | |
| def test_slide_names_reach_the_article_names(self): | |
| kh = "https://www.kenhub.com/en/library/anatomy/" | |
| catalog = WebSourceTests.catalog([kh + "achilles-tendon", kh + "calcaneal-tuberosity"]) | |
| load = lambda page: web_sources.Page(page.site, page.url, "Achilles tendon", "The achilles tendon (calcaneal tendon) ...") | |
| with mock.patch.object(web_sources, "CATALOG", catalog), mock.patch.object(web_sources, "load_page", side_effect=load): | |
| pages, _ = web_sources.find_pages(["calcaneal tendon"]) | |
| self.assertEqual([p.url for p in pages], [kh + "achilles-tendon"]) | |
| def test_question_fallback_splits_structures(self): | |
| kh = "https://www.kenhub.com/en/library/anatomy/" | |
| catalog = WebSourceTests.catalog([kh + "the-knee-joint", kh + "the-ankle-joint", kh + "carpal-tunnel"]) | |
| with mock.patch.object(web_sources, "CATALOG", catalog): | |
| self.assertEqual(web_sources.question_topics("What are the ligaments of the knee and ankle?"), ["knee", "ankle"]) | |
| self.assertEqual(web_sources.question_topics("Tell me about the carpal tunnel"), ["Tell me about the carpal tunnel"]) | |
| class RetrievalTests(unittest.TestCase): | |
| def test_question_words_do_not_match_slides(self): | |
| self.assertEqual(app.tokenize("What are the ligaments of the knees?"), ["ligament", "knee"]) | |
| def test_knee_and_ankle_questions_attach_no_upper_limb_slides(self): | |
| knee = next(n for n, lec in app.LECTURES.items() if "knee" in lec.title.lower()) | |
| ankle = next(n for n, lec in app.LECTURES.items() if "ankle" in lec.title.lower()) | |
| upper_limb = {n for n, lec in app.LECTURES.items() | |
| if any(w in lec.title.lower() for w in ("shoulder", "brachial", "arm", "forearm", "hand", "axillary"))} | |
| self.assertTrue(upper_limb) | |
| for question in ["What are the ligaments of the knee and ankle?", "What about the knees and ankles?", | |
| "What are the bones of the knee joint?", "What is the Ottawa ankle rule?", | |
| "How is the ankle joint stabilized?", "What causes knee and ankle injuries?"]: | |
| lectures = [slide.lecture for slide in app.retrieve_slides(question)] | |
| self.assertFalse(set(lectures) & upper_limb, (question, lectures)) | |
| self.assertGreaterEqual(sum(n in (knee, ankle) for n in lectures), len(lectures) / 2, (question, lectures)) | |
| class StoreTests(unittest.TestCase): | |
| def test_validators(self): | |
| self.assertTrue(supabase_store.is_valid_user_id(" JDoe_01 ")) | |
| self.assertFalse(supabase_store.is_valid_user_id("j")) | |
| self.assertFalse(supabase_store.is_valid_user_id("j doe")) | |
| self.assertTrue(supabase_store.is_valid_email("A.B@Example.co")) | |
| self.assertFalse(supabase_store.is_valid_email("a@b")) | |
| self.assertEqual(supabase_store.normalize_user_id(" JDoe "), "jdoe") | |
| def test_requests_use_the_supabase_rest_layout(self): | |
| store = supabase_store.SupabaseStore(url="https://abc.supabase.co/", key="k") | |
| seen = [] | |
| def fake_urlopen(request, timeout=None): | |
| seen.append(request) | |
| return io.BytesIO(b'[{"user_id": "jdoe", "email": "j@d.co"}]') | |
| with mock.patch.object(urllib.request, "urlopen", fake_urlopen): | |
| row, error = store.get_user("JDoe") | |
| self.assertEqual((row["email"], error), ("j@d.co", None)) | |
| self.assertEqual(seen[0].full_url, "https://abc.supabase.co/rest/v1/anatomy_users?select=user_id%2Cemail%2Ccreated_at%2Clast_login_at&user_id=eq.jdoe&limit=1") | |
| self.assertEqual(seen[0].get_header("Apikey"), "k") | |
| self.assertEqual(seen[0].get_header("Authorization"), "Bearer k") | |
| with mock.patch.object(urllib.request, "urlopen", fake_urlopen): | |
| ok, error = store.append_messages("jdoe", "s1", [{"role": "user", "content": "Q", "mode": "ask", "topic": "Q"}, {"role": "bad"}]) | |
| self.assertTrue(ok) | |
| body = json.loads(seen[-1].data) | |
| self.assertEqual(len(body), 1) | |
| self.assertEqual(set(body[0]), {"user_id", "session_id", "role", "content", "mode", "topic", "source", "created_at"}) | |
| self.assertEqual(seen[-1].get_header("Prefer"), "return=minimal") | |
| def test_duplicate_user_id_is_reported_as_exists(self): | |
| store = supabase_store.SupabaseStore(url="https://abc.supabase.co", key="k") | |
| error = urllib.error.HTTPError("u", 409, "Conflict", {}, io.BytesIO(b'{"code":"23505"}')) | |
| with mock.patch.object(urllib.request, "urlopen", side_effect=error): | |
| row, err = store.create_user("jdoe", "j@d.co") | |
| self.assertEqual((row, err), (None, supabase_store.EXISTS)) | |
| def test_inserts_are_not_retried_but_reads_are(self): | |
| store = supabase_store.SupabaseStore(url="https://abc.supabase.co", key="k", timeout=1) | |
| calls = [] | |
| def flaky(request, timeout=None): | |
| calls.append(request.get_method()) | |
| raise TimeoutError("timed out") | |
| with mock.patch.object(urllib.request, "urlopen", side_effect=flaky), mock.patch.object(supabase_store.time, "sleep"): | |
| ok, error = store.log_event("jdoe", "s1", "ask", {}) | |
| self.assertFalse(ok) | |
| self.assertIn("TimeoutError", error) | |
| row, error = store.get_user("jdoe") | |
| self.assertEqual(calls, ["POST"] + ["GET"] * supabase_store.RETRIES) | |
| def test_disabled_store_is_harmless(self): | |
| store = supabase_store.SupabaseStore(url="", key="") | |
| self.assertFalse(store.enabled) | |
| self.assertEqual(store.get_user("x"), (None, "Missing Space secret(s): SUPABASE_URL, SUPABASE_KEY")) | |
| store.enqueue("log_event", "x", "s", "ask", {}) | |
| self.assertTrue(store.flush(0.1)) | |
| if __name__ == "__main__": | |
| unittest.main() | |