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 INDEX_VERSION, 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": "工程需求是实时目标识别", "page_type": "body"}] 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。", "page_type": "body"}, {"chunk_id": 8, "page": 3, "text": "其他实验结果", "page_type": "body"}] 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"] def test_content_evidence_rejects_navigation_and_heading_only_sources(): """A contents page, an abstract or a bare heading can never prove content.""" criterion = {"criterion_id": "stage3.data", "max_score": 20, "description": "实验数据和初步分析", "required_evidence": ["实验数据", "分析"]} body = {"chunk_id": 1, "page": 11, "page_type": "body", "text": "在自主巡线测试中，底盘在标准黑色轨迹线场地内运行，记录了直线段与弯道的时间数据。"} good = ScoreResult(criterion_id="stage3.data", score=16, max_score=20, reason="有实测数据", evidence=[{"chunk_id": 1, "page": 11, "quote": "在自主巡线测试中", "target_id": "t0"}], confidence=.8) assert validate_score(good, criterion, [body]).score == 16 toc = {"chunk_id": 2, "page": 2, "page_type": "toc", "text": "第3 章系统测试与验证 ......... 8"} bad = ScoreResult(criterion_id="stage3.data", score=16, max_score=20, reason="看目录", evidence=[{"chunk_id": 2, "page": 2, "quote": "系统测试与验证", "target_id": "t0"}], confidence=.8) # The rejection is what matters; either guard may fire depending on the target's # allowed page types. with pytest.raises(ModelError, match="内容证据|不允许的页面类型"): validate_score(bad, criterion, [toc]) abstract = {"chunk_id": 3, "page": 2, "page_type": "abstract", "text": "摘要 本文实现了自瞄移动机器人系统，采用多处理器分布式架构。"} bad.evidence[0].chunk_id, bad.evidence[0].page, bad.evidence[0].quote = 3, 2, "本文实现了自瞄移动机器人系统" with pytest.raises(ModelError, match="内容证据|不允许的页面类型"): validate_score(bad, criterion, [abstract]) heading_only = {"chunk_id": 4, "page": 11, "page_type": "body", "text": "第3 章 系统测试与验证"} bad.evidence[0].chunk_id, bad.evidence[0].page, bad.evidence[0].quote = 4, 11, "第3 章 系统测试与验证" with pytest.raises(ModelError, match="章节标题"): validate_score(bad, criterion, [heading_only]) def test_structural_targets_may_use_the_contents_page(): """A structure target legitimately reads the contents list; a content one may not.""" criterion = {"criterion_id": "stage1.format", "max_score": 5, "description": "报告完整性和规范性", "required_evidence": ["标题", "章节", "参考文献"]} toc = {"chunk_id": 5, "page": 2, "page_type": "toc", "text": "第1 章项目背景与需求分析 .......... 1\n第2 章文献调研与技术现状 .......... 3"} result = ScoreResult(criterion_id="stage1.format", score=4, max_score=5, reason="章节完整", evidence=[{"chunk_id": 5, "page": 2, "quote": "第1 章项目背景与需求分析", "target_id": "t1"}], confidence=.8) checked = validate_score(result, criterion, [toc]) assert checked.evidence[0].page == 2 def test_model_may_return_a_null_score_when_content_evidence_is_missing(): result = ScoreResult(criterion_id="stage3.data", score=None, max_score=20, reason="仅检索到目录页，无法判断测试内容", evidence=[], missing_evidence=["实验数据"], confidence=.2) assert result.score is None @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_preserves_history_with_a_new_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 == INDEX_VERSION assert document.status == "superseded" assert db.scalar(select(KnowledgeChunk.text).where(KnowledgeChunk.document_id == document.id)) == "过时片段" assert db.get(KnowledgePage, (document.id, 1)).text == "过时整页" assert db.scalar(select(KnowledgeChunk.text).where(KnowledgeChunk.document_id == reindexed.id)) == pages[0]["text"] assert db.get(KnowledgePage, (reindexed.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"))