209 lines
8.2 KiB
Python
209 lines
8.2 KiB
Python
"""AI 讲解/知识点，以及用户 AI 配置。判分相关见 test_grading.py。"""
|
|
import json
|
|
from unittest.mock import patch
|
|
|
|
from app.services.llm.client import LLMResult
|
|
|
|
SUBJECTIVE = {
|
|
"type": "short_answer",
|
|
"stem_markdown": "简述牛顿第一定律",
|
|
"correct_answer": ["惯性定律"],
|
|
}
|
|
|
|
|
|
def _fake_text(content: str):
|
|
async def _inner(self, system, user, json_mode=False, model=None):
|
|
return LLMResult(content, "text-model", 10, 20)
|
|
|
|
return _inner
|
|
|
|
|
|
class TestSettings:
|
|
async def test_api_key_never_returned(self, client, auth_headers, configure_ai):
|
|
h = await auth_headers()
|
|
out = await configure_ai(h, api_key="sk-secret-value")
|
|
assert out["has_api_key"] is True
|
|
assert "sk-secret-value" not in json.dumps(out)
|
|
|
|
r = await client.get("/api/settings", headers=h)
|
|
assert "sk-secret-value" not in r.text
|
|
assert r.json()["has_api_key"] is True
|
|
|
|
async def test_api_key_stored_encrypted(self, client, auth_headers, configure_ai):
|
|
"""库里存的必须是密文，且能正确解密回原值。"""
|
|
from sqlalchemy import select
|
|
|
|
from app.db import SessionLocal
|
|
from app.models.user import UserSettings
|
|
from app.services import crypto
|
|
|
|
h = await auth_headers()
|
|
await configure_ai(h, api_key="sk-secret-value")
|
|
|
|
async with SessionLocal() as db:
|
|
row = (await db.scalars(select(UserSettings))).first()
|
|
assert row.llm_api_key_encrypted != "sk-secret-value"
|
|
assert crypto.decrypt(row.llm_api_key_encrypted) == "sk-secret-value"
|
|
|
|
async def test_empty_api_key_leaves_existing(self, client, auth_headers, configure_ai):
|
|
"""留空表示不修改，不该把已配置的 key 清掉。"""
|
|
h = await auth_headers()
|
|
await configure_ai(h)
|
|
r = await client.put(
|
|
"/api/settings", json={"llm_text_model": "new-model"}, headers=h
|
|
)
|
|
assert r.json()["has_api_key"] is True
|
|
assert r.json()["llm_text_model"] == "new-model"
|
|
|
|
async def test_settings_isolated_per_user(self, client, auth_headers, configure_ai):
|
|
ha = await auth_headers("alice")
|
|
hb = await auth_headers("bob")
|
|
await configure_ai(ha)
|
|
assert (
|
|
await client.get("/api/settings", headers=hb)
|
|
).json()["has_api_key"] is False
|
|
|
|
|
|
class TestAiExplain:
|
|
async def test_explain_persists(self, client, auth_headers, configure_ai):
|
|
h = await auth_headers()
|
|
await configure_ai(h)
|
|
qid = (await client.post("/api/questions", json=SUBJECTIVE, headers=h)).json()["id"]
|
|
|
|
with patch(
|
|
"app.services.llm.client.LLMClient.chat_text",
|
|
_fake_text("## 讲解\n\n物体在无外力时 $v$ 恒定。"),
|
|
):
|
|
r = await client.post(f"/api/questions/{qid}/ai/explain", headers=h)
|
|
assert r.status_code == 200
|
|
assert "$v$" in r.json()["explanation_markdown"]
|
|
|
|
q = (await client.get(f"/api/questions/{qid}", headers=h)).json()
|
|
assert "讲解" in q["explanation_markdown"]
|
|
|
|
async def test_explain_never_touches_my_note(self, client, auth_headers, configure_ai):
|
|
"""我自己写的笔记是我的，AI 讲解不得覆盖它。"""
|
|
h = await auth_headers()
|
|
await configure_ai(h)
|
|
qid = (
|
|
await client.post(
|
|
"/api/questions",
|
|
json={**SUBJECTIVE, "my_note_markdown": "我当时错在没考虑摩擦力"},
|
|
headers=h,
|
|
)
|
|
).json()["id"]
|
|
|
|
with patch(
|
|
"app.services.llm.client.LLMClient.chat_text", _fake_text("AI 的讲解内容")
|
|
):
|
|
await client.post(f"/api/questions/{qid}/ai/explain", headers=h)
|
|
|
|
q = (await client.get(f"/api/questions/{qid}", headers=h)).json()
|
|
assert q["my_note_markdown"] == "我当时错在没考虑摩擦力"
|
|
assert q["explanation_markdown"] == "AI 的讲解内容"
|
|
|
|
async def test_unconfigured_ai_returns_400(self, client, auth_headers):
|
|
h = await auth_headers()
|
|
qid = (await client.post("/api/questions", json=SUBJECTIVE, headers=h)).json()["id"]
|
|
r = await client.post(f"/api/questions/{qid}/ai/explain", headers=h)
|
|
assert r.status_code == 400
|
|
assert "设置" in r.json()["detail"]
|
|
|
|
async def test_cannot_explain_others_note(self, client, auth_headers, configure_ai):
|
|
ha = await auth_headers("alice")
|
|
hb = await auth_headers("bob")
|
|
await configure_ai(hb)
|
|
qid = (await client.post("/api/questions", json=SUBJECTIVE, headers=ha)).json()["id"]
|
|
r = await client.post(f"/api/questions/{qid}/ai/explain", headers=hb)
|
|
assert r.status_code == 404
|
|
|
|
|
|
class TestAiTags:
|
|
async def test_summarize_tags_merges(self, client, auth_headers, configure_ai):
|
|
h = await auth_headers()
|
|
await configure_ai(h)
|
|
qid = (
|
|
await client.post(
|
|
"/api/questions", json={**SUBJECTIVE, "tags": ["物理"]}, headers=h
|
|
)
|
|
).json()["id"]
|
|
|
|
with patch(
|
|
"app.services.llm.client.LLMClient.chat_text",
|
|
_fake_text(json.dumps({"tags": ["力学", "牛顿定律"]})),
|
|
):
|
|
r = await client.post(f"/api/questions/{qid}/ai/summarize-tags", headers=h)
|
|
assert r.status_code == 200
|
|
assert set(r.json()["tags"]) == {"物理", "力学", "牛顿定律"}
|
|
|
|
async def test_malformed_tags_json_yields_no_tags(
|
|
self, client, auth_headers, configure_ai
|
|
):
|
|
h = await auth_headers()
|
|
await configure_ai(h)
|
|
qid = (await client.post("/api/questions", json=SUBJECTIVE, headers=h)).json()["id"]
|
|
|
|
with patch("app.services.llm.client.LLMClient.chat_text", _fake_text("not json")):
|
|
r = await client.post(f"/api/questions/{qid}/ai/summarize-tags", headers=h)
|
|
assert r.status_code == 200
|
|
assert r.json()["tags"] == []
|
|
|
|
async def test_tags_on_unclassified_note(self, client, auth_headers, configure_ai):
|
|
"""拍照记下的未分类笔记，也能让 AI 归类知识点。"""
|
|
h = await auth_headers()
|
|
await configure_ai(h)
|
|
qid = (
|
|
await client.post(
|
|
"/api/questions",
|
|
json={"type": "unclassified", "stem_markdown": "求导数"},
|
|
headers=h,
|
|
)
|
|
).json()["id"]
|
|
|
|
with patch(
|
|
"app.services.llm.client.LLMClient.chat_text",
|
|
_fake_text(json.dumps({"tags": ["导数"]})),
|
|
):
|
|
r = await client.post(f"/api/questions/{qid}/ai/summarize-tags", headers=h)
|
|
assert r.json()["tags"] == ["导数"]
|
|
|
|
|
|
class TestAiAudit:
|
|
async def test_ai_generation_is_audited(self, client, auth_headers, configure_ai):
|
|
"""每次 AI 调用都要留审计记录，便于观察成本。"""
|
|
from sqlalchemy import select
|
|
|
|
from app.db import SessionLocal
|
|
from app.models.practice import AiGeneration
|
|
|
|
h = await auth_headers()
|
|
await configure_ai(h)
|
|
qid = (await client.post("/api/questions", json=SUBJECTIVE, headers=h)).json()["id"]
|
|
|
|
with patch("app.services.llm.client.LLMClient.chat_text", _fake_text("讲解内容")):
|
|
await client.post(f"/api/questions/{qid}/ai/explain", headers=h)
|
|
|
|
async with SessionLocal() as db:
|
|
rows = (await db.scalars(select(AiGeneration))).all()
|
|
assert len(rows) == 1
|
|
assert rows[0].task == "explain"
|
|
assert rows[0].model == "text-model"
|
|
assert rows[0].request_tokens == 10
|
|
assert rows[0].response_tokens == 20
|
|
|
|
async def test_recording_attempt_makes_no_ai_call(self, client, auth_headers):
|
|
"""记录作答不该产生任何 AI 调用（也就不消耗额度）。"""
|
|
from sqlalchemy import select
|
|
|
|
from app.db import SessionLocal
|
|
from app.models.practice import AiGeneration
|
|
|
|
h = await auth_headers()
|
|
qid = (await client.post("/api/questions", json=SUBJECTIVE, headers=h)).json()["id"]
|
|
await client.post(
|
|
f"/api/questions/{qid}/attempts", json={"user_answer": ["x"]}, headers=h
|
|
)
|
|
|
|
async with SessionLocal() as db:
|
|
assert (await db.scalars(select(AiGeneration))).all() == []
|