107 lines
5.5 KiB
Python
107 lines
5.5 KiB
Python
import pytest
|
|
from PIL import Image, ImageDraw
|
|
from sqlalchemy import select
|
|
|
|
from wisegrade.llm import DeepSeekClient, FactResult, ModelError, ScoreResult, validate_score
|
|
from wisegrade.models import AppSetting, KnowledgeChunk, KnowledgePage, Project, Student
|
|
from wisegrade.rag import ingest, local_encoder, retrieve, split_pages
|
|
from wisegrade.settings_store import model_config, save_model_config
|
|
|
|
|
|
def make_project(db, number):
|
|
student = Student(student_no=f"90000000{number}", name=f"测试学生{number}")
|
|
db.add(student)
|
|
db.flush()
|
|
project = Project(course_version_id=1, student_id=student.id, topic=f"测试项目{number}",
|
|
source_dir=f"tests/student-{number}")
|
|
db.add(project)
|
|
db.flush()
|
|
return project.id
|
|
|
|
|
|
def test_json_validation_and_evidence():
|
|
criterion = {"criterion_id": "stage1.problem", "max_score": 25}
|
|
chunks = [{"chunk_id": 7, "page": 2, "text": "工程需求是实时目标识别"}]
|
|
result = ScoreResult(criterion_id="stage1.problem", score=19, max_score=25, reason="满足需求",
|
|
evidence=[{"chunk_id": 7, "page": 2, "quote": "实时目标识别"}], confidence=.8)
|
|
assert validate_score(result, criterion, chunks).score == 19
|
|
with pytest.raises(ModelError):
|
|
validate_score(result, criterion, [{**chunks[0], "text": "其他内容"}])
|
|
with pytest.raises(ValueError):
|
|
ScoreResult(criterion_id="x", score=5, max_score=10, reason="无依据", evidence=[], confidence=.5)
|
|
|
|
|
|
def test_evidence_quote_resolves_pdf_line_breaks_and_page():
|
|
criterion = {"criterion_id": "stage1.problem", "max_score": 25}
|
|
chunks = [{"chunk_id": 7, "page": 2, "text": "机器人\n控制精度：2 cm。"},
|
|
{"chunk_id": 8, "page": 3, "text": "其他实验结果"}]
|
|
result = ScoreResult(criterion_id="stage1.problem", score=19, max_score=25, reason="有指标",
|
|
evidence=[{"chunk_id": 8, "page": 3, "quote": "机器人控制精度:2 cm"}], confidence=.8)
|
|
checked = validate_score(result, criterion, chunks)
|
|
assert checked.evidence[0].chunk_id == 7 and checked.evidence[0].page == 2
|
|
assert checked.evidence[0].quote in chunks[0]["text"]
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_deepseek_adapter_live_with_image(tmp_path):
|
|
image = tmp_path / "page.png"
|
|
picture = Image.new("RGB", (320, 100), "white")
|
|
ImageDraw.Draw(picture).text((20, 30), "TEST 123", fill="black")
|
|
picture.save(image)
|
|
result = DeepSeekClient().json_call(
|
|
"从用户提供的JSON中复制topic字段，原样返回到JSON对象的topic字段。",
|
|
{"topic": "真实模型连通测试"}, FactResult, images=[str(image)])
|
|
assert result.topic == "真实模型连通测试"
|
|
|
|
|
|
def test_project_isolation_and_chunk_metadata(db):
|
|
project_a = make_project(db, 1)
|
|
project_b = make_project(db, 2)
|
|
pages = [{"page_number": 3, "section": "实验", "text": "机器人实验数据和分析"}]
|
|
ingest(db, course_version_id=1, project_id=project_a, submission_file_id=None, title="学生A", category="stage3",
|
|
source_hash="a", pages=pages)
|
|
ingest(db, course_version_id=1, project_id=project_b, submission_file_id=None, title="学生B", category="stage3",
|
|
source_hash="b", pages=[{**pages[0], "text": "另一个学生的秘密"}])
|
|
hits = retrieve(db, course_version_id=1, project_id=project_a, category="stage3", query="实验数据")
|
|
assert hits and all(hit["project_id"] == project_a and hit["page"] == 3 for hit in hits)
|
|
assert "秘密" not in str(hits)
|
|
|
|
|
|
def test_short_paragraphs_share_a_page_chunk():
|
|
text = "\n\n".join(f"实验步骤{i}：采集机器人定位数据。" for i in range(12))
|
|
chunks = list(split_pages([{"page_number": 4, "section": "实验", "text": text}]))
|
|
assert len(chunks) == 1
|
|
assert chunks[0]["page"] == 4
|
|
assert "实验步骤0" in chunks[0]["text"] and "实验步骤11" in chunks[0]["text"]
|
|
|
|
|
|
def test_stale_knowledge_index_reuses_document_id(db):
|
|
project_id = make_project(db, 1)
|
|
pages = [{"page_number": 1, "text": "实验方案与测量结果"}]
|
|
document = ingest(db, course_version_id=1, project_id=project_id, submission_file_id=None,
|
|
title="报告", category="stage3", source_hash="same-source", pages=pages)
|
|
old_chunk = db.scalar(select(KnowledgeChunk).where(KnowledgeChunk.document_id == document.id))
|
|
old_chunk.text = "过时片段"
|
|
db.get(KnowledgePage, (document.id, 1)).text = "过时整页"
|
|
document.index_version = 1
|
|
db.commit()
|
|
reindexed = ingest(db, course_version_id=1, project_id=project_id, submission_file_id=None,
|
|
title="报告", category="stage3", source_hash="same-source", pages=pages)
|
|
assert reindexed.id == document.id and reindexed.index_version == 2
|
|
assert db.scalar(select(KnowledgeChunk.text).where(KnowledgeChunk.document_id == document.id)) == pages[0]["text"]
|
|
assert db.get(KnowledgePage, (document.id, 1)).text == pages[0]["text"]
|
|
|
|
|
|
def test_model_key_is_encrypted_and_hidden(db):
|
|
result = save_model_config(db, {"model": "deepseek-flash", "api_key": "test-private-key"})
|
|
assert result["api_key_configured"] is True
|
|
assert "api_key" not in result
|
|
stored = db.get(AppSetting, "api_key")
|
|
assert b"test-private-key" not in stored.encrypted_value
|
|
assert model_config(db)["model"] == "deepseek-flash"
|
|
|
|
|
|
def test_missing_offline_embedding_package_fails_clearly(tmp_path):
|
|
with pytest.raises(RuntimeError, match="离线嵌入模型不存在"):
|
|
local_encoder(str(tmp_path / "missing-model"))
|