Spaces:
Running
Running
File size: 9,454 Bytes
d49faba | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 | """Regression coverage for the chat context pipeline and sanitized model output.
Guards the behaviours introduced with scoped RAG:
* both chat endpoints share one system-prompt builder, so streaming gets
curriculum grounding and cross-session memory is kept alongside working memory;
* the posted transcript is budgeted (turn count + chars) instead of forwarded whole;
* small-talk turns never hit the vector store;
* model prose returned to the client is cleaned (think tags, fences, TeX delimiters).
"""
from __future__ import annotations
import json
from typing import Any
import pytest
import main
@pytest.fixture(autouse=True)
def isolate(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(main, "HAS_FIREBASE_ADMIN", False)
monkeypatch.setattr(main, "route_student_intent", None, raising=False)
monkeypatch.setattr(main, "_enforce_chat_rate_limit", lambda *_a, **_k: None, raising=False)
def _history(n: int, size: int = 10) -> list[main.ChatMessage]:
return [
main.ChatMessage(role="user" if i % 2 == 0 else "assistant", content=f"turn{i} " + ("x" * size))
for i in range(n)
]
class TestHistoryBudget:
def test_keeps_recent_tail_only(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(main, "CHAT_HISTORY_MAX_TURNS", 4)
out = main._budgeted_chat_history(_history(10))
assert [m["content"].split()[0] for m in out] == ["turn6", "turn7", "turn8", "turn9"]
def test_long_turn_is_truncated_from_head(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(main, "CHAT_HISTORY_MAX_CHARS_PER_TURN", 200)
out = main._budgeted_chat_history([main.ChatMessage(role="user", content="P" * 1000)])
assert out[0]["content"].startswith("P" * 200)
assert len(out[0]["content"]) < 210
def test_total_budget_drops_oldest(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(main, "CHAT_HISTORY_MAX_TOTAL_CHARS", 1000)
monkeypatch.setattr(main, "CHAT_HISTORY_MAX_CHARS_PER_TURN", 2500)
out = main._budgeted_chat_history(_history(6, size=400))
assert len(out) < 6
assert out[-1]["content"].startswith("turn5")
def test_blank_turns_skipped(self) -> None:
out = main._budgeted_chat_history([main.ChatMessage(role="user", content=" ")])
assert out == []
class TestSystemPromptLayers:
@pytest.mark.asyncio
async def test_memory_cross_session_and_rag_all_present(self, monkeypatch: pytest.MonkeyPatch) -> None:
async def fake_memory(**kwargs: Any) -> str:
assert kwargs["include_recent_turns"] is False
return "MEMORY CONTEXT — profile"
monkeypatch.setattr(main, "collect_memory_context", fake_memory)
monkeypatch.setattr(main, "retrieve_chat_context", lambda _m: "RELEVANT CURRICULUM REFERENCE: slope")
req = main.ChatRequest(
message="How do I find the slope?",
history=_history(2),
userId="u1",
sessionId="s1",
crossSessionMemory="Conversation: earlier chat about functions",
)
prompt = await main._chat_system_prompt(req, include_recent_turns=not req.history)
assert prompt.index("MEMORY CONTEXT") < prompt.index("PRIOR CONVERSATIONS") < prompt.index("RELEVANT CURRICULUM") < prompt.index(main.MATH_TUTOR_SYSTEM_PROMPT[:40])
assert "earlier chat about functions" in prompt
@pytest.mark.asyncio
async def test_rag_failure_is_non_fatal(self, monkeypatch: pytest.MonkeyPatch) -> None:
def boom(_m: str) -> str:
raise RuntimeError("chroma down")
monkeypatch.setattr(main, "collect_memory_context", None)
monkeypatch.setattr(main, "retrieve_chat_context", boom)
prompt = await main._chat_system_prompt(main.ChatRequest(message="What is a function?"), include_recent_turns=True)
assert prompt == main.MATH_TUTOR_SYSTEM_PROMPT
@pytest.mark.asyncio
async def test_small_talk_never_queries_store(self, monkeypatch: pytest.MonkeyPatch) -> None:
calls: list[str] = []
monkeypatch.setattr(main, "collect_memory_context", None)
monkeypatch.setattr(
"rag.curriculum_rag.retrieve_curriculum_context",
lambda **kw: calls.append(kw["query"]) or [],
)
await main._chat_system_prompt(main.ChatRequest(message="thanks!"), include_recent_turns=True)
assert calls == []
class TestChatEndpoints:
@pytest.mark.asyncio
async def test_non_stream_uses_budgeted_history_and_cleans_answer(self, monkeypatch: pytest.MonkeyPatch) -> None:
observed: list[dict[str, str]] = []
async def llm(messages: list[dict[str, str]], **_k: Any) -> str:
observed.extend(messages)
return "<think>plan</think>\nThe slope is \\(m = 2\\)."
monkeypatch.setattr(main, "HAS_MEMORY_SERVICE", False)
monkeypatch.setattr(main, "CHAT_HISTORY_MAX_TURNS", 3)
monkeypatch.setattr(main, "call_hf_chat_async", llm)
monkeypatch.setattr(main, "retrieve_chat_context", lambda _m: "RELEVANT CURRICULUM REFERENCE: slope")
res = await main.chat_tutor(main.ChatRequest(message="How do I find the slope of a line?", history=_history(8)))
assert res.response == "The slope is $m = 2$."
assert observed[0]["role"] == "system" and "RELEVANT CURRICULUM REFERENCE" in observed[0]["content"]
# system + 3 budgeted turns + current message
assert len(observed) == 5
assert observed[-1]["content"] == "How do I find the slope of a line?"
@pytest.mark.asyncio
async def test_stream_endpoint_injects_rag_and_budgets_history(self, monkeypatch: pytest.MonkeyPatch) -> None:
observed: list[dict[str, str]] = []
async def llm_stream(messages: list[dict[str, str]], **_k: Any):
observed.extend(messages)
yield "The slope "
yield "is 2."
monkeypatch.setattr(main, "HAS_MEMORY_SERVICE", False)
monkeypatch.setattr(main, "CHAT_HISTORY_MAX_TURNS", 2)
monkeypatch.setattr(main, "call_hf_chat_stream_async", llm_stream)
monkeypatch.setattr(main, "retrieve_chat_context", lambda _m: "RELEVANT CURRICULUM REFERENCE: slope")
response = await main.chat_tutor_stream(
main.ChatRequest(message="How do I find the slope of a line?", history=_history(6))
)
body = "".join([chunk.decode() if isinstance(chunk, bytes) else chunk async for chunk in response.body_iterator])
chunks = [json.loads(line[len("data: "):])["chunk"] for line in body.splitlines() if line.startswith("data: {")]
assert "".join(chunks) == "The slope is 2."
assert "RELEVANT CURRICULUM REFERENCE" in observed[0]["content"]
assert len(observed) == 4 # system + 2 turns + current
@pytest.mark.asyncio
async def test_stream_scope_boundary_skips_context_build(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(main, "HAS_MEMORY_SERVICE", False)
monkeypatch.setattr(main, "get_scope_boundary_response", lambda *_a, **_k: "I only help with math.")
async def must_not_run(*_a: Any, **_k: Any) -> str:
raise AssertionError("context build must be skipped for boundary replies")
monkeypatch.setattr(main, "_chat_system_prompt", must_not_run)
response = await main.chat_tutor_stream(main.ChatRequest(message="write me a poem"))
body = "".join([chunk.decode() if isinstance(chunk, bytes) else chunk async for chunk in response.body_iterator])
assert "I only help with math." in body
class TestQuizOutputSanitized:
def test_validate_quiz_questions_strips_markup(self) -> None:
raw = [{
"question": "<think>x</think>Find **the** slope of \\(y = 2x\\).",
"correctAnswer": "B. **2**",
"options": ["A. **1**", "B. **2**", "C) 3", "(d) 4"],
"explanation": "```\nThe slope is the coefficient.\n```",
"questionType": "multiple_choice",
"difficulty": "easy",
"bloomLevel": "understand",
}]
[q] = main._validate_quiz_questions(raw, [{}], topics=["Linear"])
assert q.question == "Find **the** slope of $y = 2x$."
assert q.options == ["1", "2", "3", "4"]
assert q.correctAnswer == "2"
assert q.explanation == "The slope is the coefficient."
assert q.correctAnswer in q.options
def test_quizforge_parser_strips_markup(self) -> None:
from routes.quiz_generation_routes import _parse_quiz_response
payload = json.dumps([
{
"question_text": "What is \\(2^3\\)?",
"type": "multiple_choice",
"options": [{"key": "A", "text": "**6**"}, {"key": "B", "text": "B. 8"}, {"key": "C", "text": "9"}],
"correct_answer": "B",
"explanation": "2 × 2 × 2 = 8\\nSo **8**.",
"hints": ["<think>t</think>Multiply 2 by itself three times", ""],
}
] * 3)
qs = _parse_quiz_response(payload, expected_count=3)
q = qs[0]
assert q["question"] == "What is $2^3$?"
assert q["options"] == ["6", "8", "9"]
assert q["correctAnswer"] == "8"
assert q["explanation"] == "2 × 2 × 2 = 8\nSo **8**."
assert q["hints"] == ["Multiply 2 by itself three times"]
|