test: add end to end coverage

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

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()

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"] == "机器人系统"

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)

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"]

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()