diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py new file mode 100644 index 0000000..3523fc8 --- /dev/null +++ b/backend/tests/conftest.py @@ -0,0 +1,40 @@ +import os +from pathlib import Path + +import pytest +from sqlalchemy.orm import sessionmaker + +from wisegrade.config import get_settings +from wisegrade.db import Base, engine_for +from wisegrade.seed import seed_initial +from wisegrade import models # noqa: F401 + + +@pytest.fixture +def environment(tmp_path, monkeypatch): + inbox = tmp_path / "inbox" + inbox.mkdir() + monkeypatch.setenv("INBOX_PATH", str(inbox)) + monkeypatch.setenv("PROCESSED_PATH", str(tmp_path / "processed")) + monkeypatch.setenv("EXPORTS_PATH", str(tmp_path / "exports")) + monkeypatch.setenv("WISEGRADE_SECRET_KEY", "test-only-secret-key-of-at-least-32-characters") + get_settings.cache_clear() + yield tmp_path + get_settings.cache_clear() + + +@pytest.fixture +def session_factory(environment): + engine = engine_for(f"sqlite:///{environment / 'test.db'}") + Base.metadata.create_all(engine) + factory = sessionmaker(bind=engine, expire_on_commit=False) + with factory() as db: + seed_initial(db) + yield factory + engine.dispose() + + +@pytest.fixture +def db(session_factory): + with session_factory() as session: + yield session diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py new file mode 100644 index 0000000..333a879 --- /dev/null +++ b/backend/tests/test_api.py @@ -0,0 +1,71 @@ +from fastapi.testclient import TestClient +from sqlalchemy import select + +from wisegrade.db import get_db +from wisegrade.main import app +from wisegrade.models import CourseVersion, CriterionScore, Project, RubricVersion, Student +from wisegrade.scoring import run_batch + + +def client_for(session_factory): + def override_db(): + with session_factory() as db: + yield db + app.dependency_overrides[get_db] = override_db + return TestClient(app) + + +def test_course_and_rubric_versions_are_immutable(session_factory): + client = client_for(session_factory) + try: + new = client.post("/api/courses/1/versions", json={"title": "新学期规则", "source_version_id": 1}) + assert new.status_code == 200 + version = new.json() + version["objectives"][0]["name"] = "修订后的目标" + edited = client.put(f"/api/course-versions/{version['id']}", json={key: version[key] for key in + ("title", "objectives", "indicators", "directions", "prompts", "templates")}) + assert edited.status_code == 200 + assert client.post(f"/api/course-versions/{version['id']}/publish").status_code == 200 + assert client.put(f"/api/course-versions/{version['id']}", json={key: version[key] for key in + ("title", "objectives", "indicators", "directions", "prompts", "templates")}).status_code == 409 + rubric = client.get(f"/api/course-versions/{version['id']}/rubrics").json()[0] + rubric["criteria"][0]["name"] = "修订后的评分项" + assert client.put(f"/api/rubrics/{rubric['id']}", json={"criteria": rubric["criteria"], "weights": rubric["weights"]}).status_code == 200 + assert client.post(f"/api/rubrics/{rubric['id']}/publish").status_code == 200 + batch = client.post("/api/scoring-batches", json={"course_version_id": version["id"], "rubric_version_id": rubric["id"]}) + assert batch.status_code == 200 + with session_factory() as db: + from wisegrade.models import ScoringBatch + snapshot = db.get(ScoringBatch, batch.json()["id"]) + assert snapshot.course_snapshot["objectives"][0]["name"] == "修订后的目标" + assert snapshot.rubric_snapshot["criteria"][0]["name"] == "修订后的评分项" + assert db.get(CourseVersion, 1).objectives[0]["name"] != "修订后的目标" + assert db.get(RubricVersion, 1).criteria[0]["name"] != "修订后的评分项" + finally: + app.dependency_overrides.clear() + + +def test_missing_material_is_explicit_and_reviewable(session_factory): + with session_factory() as db: + student = Student(student_no="1023009007", name="测试学生") + db.add(student) + db.flush() + db.add(Project(course_version_id=1, student_id=student.id, topic="没有材料", source_dir="empty/student")) + db.commit() + client = client_for(session_factory) + try: + batch_id = client.post("/api/scoring-batches", json={"course_version_id": 1, "rubric_version_id": 1}).json()["id"] + run_batch(batch_id, session_factory=session_factory, client_factory=lambda db, batch: object()) + status = client.get(f"/api/scoring-batches/{batch_id}/progress").json() + assert status["status"] == "partial" and status["progress_done"] == status["progress_total"] == 22 + result = client.get(f"/api/scoring-batches/{batch_id}/scores").json()[0] + assert result["totals"]["overall"] is None + assert all(row["status"] == "unscorable" and row["error"] for row in result["rows"]) + first_id = result["rows"][0]["score_id"] + response = client.put(f"/api/criterion-scores/{first_id}/review", json={"score": 3, "comment": "人工核定", "reason": "教师确认"}) + assert response.status_code == 200 and response.json()["model_score"] is None + with session_factory() as db: + score = db.get(CriterionScore, first_id) + assert score.teacher_score == 3 and score.model_score is None + finally: + app.dependency_overrides.clear() diff --git a/backend/tests/test_foundation.py b/backend/tests/test_foundation.py new file mode 100644 index 0000000..c4b4807 --- /dev/null +++ b/backend/tests/test_foundation.py @@ -0,0 +1,61 @@ +from pathlib import Path + +import pytest +from sqlalchemy import select + +from wisegrade.models import CourseObjective, CourseVersion, GraduationIndicator, RubricCriterion, RubricScoreLevel, RubricVersion, TopicDirection +from wisegrade.rules import RubricData, totals +from wisegrade.storage import classify, extract_identity, parse_student_folder, secure_path, sha256 + + +def test_seeded_course_and_weights(db): + version = db.scalar(select(CourseVersion)) + rubric = db.scalar(select(RubricVersion)) + assert version.published and rubric.published + assert {item["code"] for item in version.indicators} == {"2.3", "2.4", "4.1", "4.2", "4.3", "4.4", "11.1", "11.2"} + assert len(RubricData(criteria=rubric.criteria, weights=rubric.weights).criteria) == 22 + assert rubric.weights == {"stage1": .1, "stage2": .1, "stage3": .1, "final": .7} + assert len(db.scalars(select(CourseObjective)).all()) == 4 + assert len(db.scalars(select(GraduationIndicator)).all()) == 8 + assert len(db.scalars(select(TopicDirection)).all()) == 3 + assert len(db.scalars(select(RubricCriterion)).all()) == 22 + assert len(db.scalars(select(RubricScoreLevel)).all()) == 66 + + +def test_score_totals_preserve_missing(db): + rubric = db.scalar(select(RubricVersion)) + scores = {item["criterion_id"]: item["max_score"] for item in rubric.criteria} + assert totals(scores, rubric.criteria, rubric.weights)["overall"] == 100 + scores.pop("stage2.hardware") + result = totals(scores, rubric.criteria, rubric.weights) + assert result["stage2"] is None and result["overall"] is None + + +def test_invalid_rubric_weights(db): + rubric = db.scalar(select(RubricVersion)) + with pytest.raises(ValueError): + RubricData(criteria=rubric.criteria, weights={"stage1": .2, "stage2": .1, "stage3": .1, "final": .7}) + + +def test_path_security_and_hash(environment): + root = environment / "inbox" + target = root / "student" / "report.pdf" + target.parent.mkdir() + target.write_bytes(b"unchanged") + assert secure_path("student/report.pdf", root=root) == target + assert sha256(target) == "aaa8d3c8d74ad3e8f6b1772aa9c7e0eaa528cb42fc93599ce2f125b00d4c424c" + for bad in ("../outside", "/etc/passwd", "student/../../outside", "student\\report.pdf"): + with pytest.raises(ValueError): + secure_path(bad, root=root) + (root / "link").symlink_to(environment) + with pytest.raises(ValueError): + secure_path("link/test.db", root=root, must_exist=False) + + +def test_student_folder_and_file_recognition(): + assert parse_student_folder("1023009006_李傲琦_基于ROS2系统的自瞄移动机器人")["topic"].startswith("基于ROS2") + assert parse_student_folder("bad_name")["status"] == "needs_confirmation" + assert classify(Path("汇报PPT.pdf")) == ("presentation", "needs_confirmation") + assert classify(Path("阶段报告2.docx")) == ("stage2", "recognized") + assert classify(Path("legacy.doc")) == ("unknown", "unsupported") + assert extract_identity("学号:1023009006\n姓名:李傲琦\n题目:机器人系统")["topic"] == "机器人系统" diff --git a/backend/tests/test_llm_and_rag.py b/backend/tests/test_llm_and_rag.py new file mode 100644 index 0000000..5c838b4 --- /dev/null +++ b/backend/tests/test_llm_and_rag.py @@ -0,0 +1,55 @@ +import json + +import httpx +import pytest +from sqlalchemy import select + +from wisegrade.llm import DeepSeekClient, ModelError, ScoreResult, validate_score +from wisegrade.models import KnowledgeChunk +from wisegrade.rag import ingest, retrieve, split_pages + + +def fake_embed(texts): + return [[1.0] + [0.0] * 1023 for _ in texts] + + +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_deepseek_adapter_mock_and_key_not_in_error(tmp_path): + image = tmp_path / "page.png" + image.write_bytes(b"png fixture") + def handler(request): + assert request.url.path == "/chat/completions" + assert request.headers["Authorization"] == "Bearer fake-secret" + content = json.loads(request.content)["messages"][1]["content"] + assert content[1]["type"] == "image_url" and content[1]["image_url"]["url"].startswith("data:image/png;base64,") + return httpx.Response(200, json={"choices": [{"message": {"content": json.dumps({"criterion_id": "x", "score": 0, + "max_score": 10, "reason": "缺失", "evidence": [], "confidence": .2})}}]}) + client = DeepSeekClient(base_url="https://example.test", api_key="fake-secret", retries=0, transport=httpx.MockTransport(handler)) + assert client.json_call("prompt", {}, ScoreResult, images=[str(image)]).score == 0 + bad = DeepSeekClient(base_url="https://example.test", api_key="fake-secret", retries=0, + transport=httpx.MockTransport(lambda _: httpx.Response(500, text="fake-secret"))) + with pytest.raises(ModelError) as error: + bad.json_call("prompt", {}, ScoreResult) + assert "fake-secret" not in str(error.value) + + +def test_project_isolation_and_chunk_metadata(db): + pages = [{"page_number": 3, "section": "实验", "text": "机器人实验数据和分析"}] + ingest(db, course_version_id=1, project_id=101, submission_file_id=None, title="学生A", category="stage3", + source_hash="a", pages=pages, embedder=fake_embed) + ingest(db, course_version_id=1, project_id=102, submission_file_id=None, title="学生B", category="stage3", + source_hash="b", pages=[{**pages[0], "text": "另一个学生的秘密"}], embedder=fake_embed) + hits = retrieve(db, course_version_id=1, project_id=101, category="stage3", query="实验数据", embedder=fake_embed) + assert hits and all(hit["project_id"] == 101 and hit["page"] == 3 for hit in hits) + assert "秘密" not in str(hits) diff --git a/backend/tests/test_parsing.py b/backend/tests/test_parsing.py new file mode 100644 index 0000000..8dd2195 --- /dev/null +++ b/backend/tests/test_parsing.py @@ -0,0 +1,45 @@ +import shutil + +import pymupdf as fitz +import pytest +from docx import Document +from pptx import Presentation +from pptx.util import Inches + +from wisegrade.parsing import parse_document + + +def test_pdf_pages_and_images(tmp_path): + pdf = fitz.open() + pdf.new_page().insert_text((72, 72), "A sample engineering report with sufficiently long searchable text.") + path = tmp_path / "report.pdf" + pdf.save(path) + parsed = parse_document(path, "final_report", tmp_path / "assets", ocr=False) + assert parsed["pages"][0]["page_number"] == 1 + assert "engineering" in parsed["pages"][0]["text"] + assert (tmp_path / "assets" / "page-1.png").exists() + + +@pytest.mark.skipif(not shutil.which("libreoffice"), reason="LibreOffice is installed in the deployment image") +def test_docx_and_pptx_render(tmp_path): + word = Document() + word.add_heading("Automation experiment", 1) + word.add_paragraph("The system measured repeatable engineering results.") + table = word.add_table(rows=1, cols=2) + table.cell(0, 0).text, table.cell(0, 1).text = "metric", "accuracy" + docx = tmp_path / "stage.docx" + word.save(docx) + parsed = parse_document(docx, "stage2", tmp_path / "docx-assets", ocr=False) + assert parsed["headings"] and parsed["tables"] + assert parsed["pages"][0]["rendered_image_path"] + + deck = Presentation() + slide = deck.slides.add_slide(deck.slide_layouts[5]) + slide.shapes.title.text = "Robot presentation" + box = slide.shapes.add_textbox(Inches(1), Inches(2), Inches(4), Inches(1)) + box.text = "Results and analysis" + pptx = tmp_path / "slides.pptx" + deck.save(pptx) + slides = parse_document(pptx, "presentation", tmp_path / "pptx-assets", ocr=False)["pages"] + assert len(slides) == 1 and "Robot presentation" in slides[0]["text"] + assert slides[0]["rendered_image_path"] diff --git a/backend/tests/test_workflow.py b/backend/tests/test_workflow.py new file mode 100644 index 0000000..7eb8467 --- /dev/null +++ b/backend/tests/test_workflow.py @@ -0,0 +1,90 @@ +import shutil +from pathlib import Path + +import pymupdf as fitz +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import select + +from wisegrade.config import get_settings +from wisegrade.db import get_db +from wisegrade.llm import ConsistencyResult, FactResult, ScoreResult +from wisegrade.main import app +from wisegrade.models import CourseVersion, CriterionScore, RubricVersion, ScoringBatch, TeacherRevision +from wisegrade.rag import retrieve +from wisegrade.scoring import run_batch +from wisegrade.storage import scan + + +def fake_embed(texts): + return [[1.0] + [0.0] * 1023 for _ in texts] + + +class FakeModel: + def json_call(self, _prompt, payload, schema, images=None): + if schema is FactResult: + return FactResult(topic="自动化实践示例") + if schema is ConsistencyResult: + return ConsistencyResult() + criterion = payload["criterion"] + evidence = payload["evidence"][0] + return ScoreResult(criterion_id=criterion["criterion_id"], score=criterion["max_score"] / 2, + max_score=criterion["max_score"], reason="示例证据支持部分要求", + evidence=[{"chunk_id": evidence["chunk_id"], "page": evidence["page"], + "quote": evidence["text"][:min(12, len(evidence["text"]))]}], + missing_evidence=[], suggestion="补充量化论证", confidence=.7) + + +def make_pdf(path, title): + pdf = fitz.open() + page = pdf.new_page() + page.insert_text((72, 72), title + " results and analysis for robot control system. " * 4) + pdf.save(path) + + +@pytest.mark.integration +def test_docs_examples_end_to_end(environment, session_factory): + root = environment / "inbox" / "2026春季第1批" / "1023009006_李傲琦_基于ROS2系统的自瞄移动机器人" + root.mkdir(parents=True) + docs = Path(__file__).resolve().parents[2] / "docs" + for filename in ("阶段报告1.pdf", "阶段报告2.pdf", "阶段报告3.pdf", "自动化专业综合实践报告.pdf"): + shutil.copy2(docs / filename, root / filename) + make_pdf(root / "汇报PPT.pdf", "Presentation") + data = scan("2026春季第1批") + assert len(data["projects"]) == 1 + scanned = data["projects"][0] + assert scanned["missing"] == [] + assert next(f for f in scanned["files"] if f["category"] == "presentation")["status"] == "needs_confirmation" + + def override_db(): + with session_factory() as session: + yield session + app.dependency_overrides[get_db] = override_db + client = TestClient(app) + try: + files = [{"relative_path": f["relative_path"], "category": f["category"], "sha256": f["sha256"], "status": "recognized"} for f in scanned["files"]] + response = client.post("/api/import/confirm", json={"course_version_id": 1, "projects": [{"relative_dir": scanned["relative_dir"], + "student_no": scanned["student_no"], "name": scanned["name"], "topic": scanned["topic"], "files": files}]}) + assert response.status_code == 200, response.text + project_id = response.json()["project_ids"][0] + response = client.post("/api/scoring-batches", json={"course_version_id": 1, "rubric_version_id": 1}) + assert response.status_code == 200, response.text + batch_id = response.json()["id"] + run_batch(batch_id, session_factory=session_factory, client_factory=lambda db, batch: FakeModel(), embedder=fake_embed, + retriever=lambda db, **kwargs: retrieve(db, embedder=fake_embed, **kwargs)) + result = client.get(f"/api/projects/{project_id}/scores?batch_id={batch_id}").json() + assert result["totals"]["overall"] == 50 + assert all(row["evidence"] and row["evidence"][0]["page"] >= 1 for row in result["rows"]) + assert client.get(result["rows"][0]["evidence"][0]["image_url"]).content[:8] == b"\x89PNG\r\n\x1a\n" + score_id = result["rows"][0]["score_id"] + reviewed = client.put(f"/api/criterion-scores/{score_id}/review", json={"score": 20, "comment": "教师核定", "reason": "核查原文"}) + assert reviewed.status_code == 200 + assert reviewed.json()["model_score"] != reviewed.json()["teacher_score"] + assert len(client.get(f"/api/criterion-scores/{score_id}/revisions").json()) == 1 + excel = client.post("/api/exports/excel", json={"batch_id": batch_id}) + pdf = client.post("/api/exports/project-pdf", json={"batch_id": batch_id, "project_id": project_id}) + assert excel.status_code == pdf.status_code == 200 + assert client.get(f"/api/exports/{excel.json()['id']}/download").content[:2] == b"PK" + assert client.get(f"/api/exports/{pdf.json()['id']}/download").content[:4] == b"%PDF" + finally: + app.dependency_overrides.clear()