SkyLiCoder/WiseGrade
1
0
Fork 1
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

test: use PostgreSQL and live model integrations

Browse Source
This commit is contained in:
李傲琦
2026-09-25 20:30:28 +08:00
parent dd48e13cf2
commit 8bebcdb470
9 changed files with 144 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Dockerfile
+2
View File
@@ -5,5 +5,7 @@ WORKDIR /app
COPY --from=ghcr.io/astral-sh/uv:0.11.0 /uv /uvx /bin/
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev --extra ocr
ARG INCLUDE_DEV=false
RUN if [ "$INCLUDE_DEV" = "true" ]; then uv sync --frozen --group dev --extra ocr; fi
COPY . .
ENV PATH=/app/.venv/bin:$PATH PYTHONPATH=/app
backend/tests/conftest.py
+46 -10
View File
@@ -1,13 +1,28 @@
import os
from pathlib import Path
from uuid import uuid4
import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import create_engine
from sqlalchemy.engine import make_url
from sqlalchemy.orm import sessionmaker
from wisegrade.config import get_settings
from wisegrade.db import Base, engine_for
from wisegrade.db import engine_for
from wisegrade.seed import seed_initial
from wisegrade import models # noqa: F401
@pytest.fixture(scope="session")
def test_database_url():
value = os.environ.get("TEST_DATABASE_URL")
if not value:
pytest.fail("Set TEST_DATABASE_URL to a dedicated PostgreSQL database named wisegrade_tests")
url = make_url(value)
if url.drivername != "postgresql+psycopg" or url.database != "wisegrade_tests":
pytest.fail("TEST_DATABASE_URL must use postgresql+psycopg and the wisegrade_tests database")
return url
@pytest.fixture
@@ -24,14 +39,35 @@ def environment(tmp_path, monkeypatch):
@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()
def session_factory(environment, monkeypatch, test_database_url):
admin_engine = create_engine(test_database_url, isolation_level="AUTOCOMMIT")
name = f"wisegrade_test_{uuid4().hex}"
engine = None
created = False
try:
with admin_engine.connect() as connection:
connection.exec_driver_sql(f'CREATE DATABASE "{name}"')
created = True
url = test_database_url.set(database=name).render_as_string(hide_password=False)
monkeypatch.setenv("DATABASE_URL", url)
get_settings.cache_clear()
backend_root = Path(__file__).resolve().parents[1]
migration_config = Config(str(backend_root / "alembic.ini"))
migration_config.set_main_option("script_location", str(backend_root / "migrations"))
command.upgrade(migration_config, "head")
engine = engine_for(url)
factory = sessionmaker(bind=engine, expire_on_commit=False)
with factory() as db:
seed_initial(db)
yield factory
finally:
if engine is not None:
engine.dispose()
get_settings.cache_clear()
if created:
with admin_engine.connect() as connection:
connection.exec_driver_sql(f'DROP DATABASE "{name}" WITH (FORCE)')
admin_engine.dispose()
@pytest.fixture
backend/tests/test_api.py
+1 -1
View File
@@ -55,7 +55,7 @@ def test_missing_material_is_explicit_and_reviewable(session_factory):
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())
run_batch(batch_id, session_factory=session_factory)
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]
backend/tests/test_foundation.py
+7 -1
View File
@@ -1,13 +1,19 @@
from pathlib import Path
import pytest
from sqlalchemy import select
from sqlalchemy import select, text
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_postgresql_migrations_and_vector_extension(db):
assert db.bind.dialect.name == "postgresql"
assert db.scalar(text("SELECT version_num FROM alembic_version")) == "0002"
assert db.scalar(text("SELECT extname FROM pg_extension WHERE extname = 'vector'")) == "vector"
def test_seeded_course_and_weights(db):
version = db.scalar(select(CourseVersion))
rubric = db.scalar(select(RubricVersion))
backend/tests/test_llm_and_rag.py
+34 -33
View File
@@ -1,17 +1,22 @@
import json
import httpx
import pytest
from PIL import Image, ImageDraw
from sqlalchemy import select
from wisegrade.llm import DeepSeekClient, ModelError, ScoreResult, validate_score
from wisegrade.models import AppSetting, KnowledgeChunk
from wisegrade.llm import DeepSeekClient, FactResult, ModelError, ScoreResult, validate_score
from wisegrade.models import AppSetting, KnowledgeChunk, Project, Student
from wisegrade.rag import ingest, local_encoder, retrieve, split_pages
from wisegrade.settings_store import model_config, save_model_config
def fake_embed(texts):
return [[1.0] + [0.0] * 1023 for _ in texts]
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():
@@ -37,33 +42,28 @@ def test_evidence_quote_resolves_pdf_line_breaks_and_page():
assert checked.evidence[0].quote in chunks[0]["text"]
def test_deepseek_adapter_mock_and_key_not_in_error(tmp_path):
@pytest.mark.integration
def test_deepseek_adapter_live_with_image(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)
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=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)
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)
@@ -76,15 +76,16 @@ def test_short_paragraphs_share_a_page_chunk():
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=101, submission_file_id=None,
title="报告", category="stage3", source_hash="same-source", pages=pages, embedder=fake_embed)
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 = "过时片段"
document.index_version = 1
db.commit()
reindexed = ingest(db, course_version_id=1, project_id=101, submission_file_id=None,
title="报告", category="stage3", source_hash="same-source", pages=pages, embedder=fake_embed)
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"]
backend/tests/test_workflow.py
+12 -30
View File
@@ -4,37 +4,13 @@ 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()
@@ -70,14 +46,20 @@ def test_docs_examples_end_to_end(environment, session_factory):
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))
run_batch(batch_id, session_factory=session_factory)
progress = client.get(f"/api/scoring-batches/{batch_id}/progress").json()
assert progress["status"] == "completed", progress
assert progress["progress_done"] == progress["progress_total"] == 22
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"
assert result["totals"]["overall"] is not None
assert all(row["status"] == "scored" and
(row["model_score"] == 0 or row["evidence"] and row["evidence"][0]["page"] >= 1)
for row in result["rows"]), result["rows"]
evidence_row = next(row for row in result["rows"] if row["evidence"] and row["evidence"][0]["image_url"])
assert client.get(evidence_row["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": "核查原文"})
teacher_score = 0 if result["rows"][0]["model_score"] != 0 else 1
reviewed = client.put(f"/api/criterion-scores/{score_id}/review", json={"score": teacher_score, "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