71 lines
3.7 KiB
Python
71 lines
3.7 KiB
Python
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from wisegrade.llm import DeepSeekClient, ModelError, ScoreResult, validate_score
|
|
from wisegrade.models import AppSetting, KnowledgeChunk
|
|
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 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)
|
|
|
|
|
|
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"))
|