313 lines
16 KiB
Python
313 lines
16 KiB
Python
"""中心服务端：预审规则、跨校判重合并、发布范围与版本号、凭证、桌面端落地。
|
||
|
||
数据库用例连的是中心库 school_assistant_hub，跑完把自己造的数据删干净。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import uuid
|
||
|
||
import pytest
|
||
|
||
from server import review
|
||
from server.db import database as hub
|
||
|
||
|
||
class TestRuleSuggestion:
|
||
@staticmethod
|
||
def entry(**over) -> dict:
|
||
return {"value": "班级是下拉框，要先点开再选", "category": "control_tip",
|
||
"school_count": 1, "observations": 1, **over}
|
||
|
||
def test_too_short_is_rejected(self):
|
||
verdict, _ = review.rule_suggestion(self.entry(value="太短"))
|
||
assert verdict == "reject"
|
||
|
||
def test_two_schools_beats_everything(self):
|
||
# 回归：这条含「计科2301班」举例，早先被一次性数据的正则抢先判成 review
|
||
value = "班级是下拉框，要先点开再从选项里选，选项形如「计科2301班（计算机科学与技术）」"
|
||
verdict, reason = review.rule_suggestion(self.entry(value=value, school_count=2))
|
||
assert verdict == "adopt" and "2 所学校" in reason
|
||
|
||
@pytest.mark.parametrize("value", [
|
||
"计科2301班 2026-09-17",
|
||
"听课时间 2026-09-17，授课教师 张老师",
|
||
"2026/9/17 高等数学",
|
||
])
|
||
def test_short_values_that_are_just_data_are_rejected(self, value):
|
||
verdict, _ = review.rule_suggestion(self.entry(value=value, category="note"))
|
||
assert verdict == "reject"
|
||
|
||
def test_an_operation_tip_quoting_a_class_is_not_one_off_data(self):
|
||
value = "听课时间是日期框，点进去直接键入 2026-09-17 再回车，不要用日历面板点选"
|
||
verdict, _ = review.rule_suggestion(self.entry(value=value, category="control_tip"))
|
||
assert verdict != "reject"
|
||
|
||
def test_control_tip_with_enough_observations(self):
|
||
verdict, reason = review.rule_suggestion(self.entry(observations=3))
|
||
assert verdict == "adopt" and "控件技巧" in reason
|
||
|
||
def test_rules_always_need_a_human(self):
|
||
value = "同一天同一班级不能重复提交查课记录，提交前要先看历史列表"
|
||
verdict, reason = review.rule_suggestion(self.entry(value=value, category="rule", observations=9))
|
||
assert verdict == "review" and "人工确认" in reason
|
||
|
||
def test_thin_evidence_goes_to_review(self):
|
||
verdict, _ = review.rule_suggestion(self.entry(category="path"))
|
||
assert verdict == "review"
|
||
|
||
|
||
@pytest.fixture
|
||
def hub_scope():
|
||
"""一次性的 knowledge_id，把本次用例造的数据和真实数据隔开。"""
|
||
knowledge_id = f"pytest_{uuid.uuid4().hex[:8]}"
|
||
yield knowledge_id
|
||
with hub.get_conn() as conn:
|
||
hub._execute(conn, "DELETE FROM hub_entries WHERE knowledge_id = %s", (knowledge_id,))
|
||
hub._execute(conn, "DELETE FROM schools WHERE id LIKE 'pytest_%%'")
|
||
hub._execute(conn, "DELETE FROM sync_log WHERE school_id LIKE 'pytest_%%'")
|
||
|
||
|
||
def draft(knowledge_id: str, key: str, value: str, category: str = "control_tip", observations: int = 1) -> dict:
|
||
return {"knowledge_id": knowledge_id, "business_id": "b1", "key": key, "value": value,
|
||
"category": category, "observations": observations, "source_task_id": "t1"}
|
||
|
||
|
||
@pytest.mark.db
|
||
class TestSchoolTokens:
|
||
def test_token_is_returned_once_and_stored_hashed(self, hub_scope):
|
||
school = f"pytest_{uuid.uuid4().hex[:6]}"
|
||
issued = hub.register_school(school, "测试学校")
|
||
assert issued["token"].startswith("sa_")
|
||
rows = [s for s in hub.list_schools() if s["id"] == school]
|
||
assert rows and "token" not in rows[0]
|
||
assert rows[0]["token_hash"] != issued["token"]
|
||
|
||
def test_authenticate_accepts_only_the_real_token(self, hub_scope):
|
||
school = f"pytest_{uuid.uuid4().hex[:6]}"
|
||
issued = hub.register_school(school)
|
||
assert hub.authenticate(school, issued["token"]) is True
|
||
assert hub.authenticate(school, "sa_wrong") is False
|
||
assert hub.authenticate("no_such_school", issued["token"]) is False
|
||
assert hub.authenticate(school, "") is False
|
||
|
||
def test_regenerating_invalidates_the_old_token(self, hub_scope):
|
||
school = f"pytest_{uuid.uuid4().hex[:6]}"
|
||
old = hub.register_school(school)["token"]
|
||
new = hub.register_school(school)["token"]
|
||
assert hub.authenticate(school, new) is True
|
||
assert hub.authenticate(school, old) is False
|
||
|
||
|
||
@pytest.mark.db
|
||
class TestReceiveDrafts:
|
||
VALUE = "班级是下拉框，要先点开再从选项里选，选项形如「计科2301班（计算机科学与技术）」"
|
||
|
||
def test_first_report_creates(self, hub_scope):
|
||
stats = hub.receive_drafts("pytest_a", [draft(hub_scope, "班级下拉框", self.VALUE)])
|
||
assert stats == {"created": 1, "merged": 0, "skipped": 0, "revived": 0}
|
||
|
||
def test_same_knowledge_from_another_school_merges_and_counts_schools(self, hub_scope):
|
||
hub.receive_drafts("pytest_a", [draft(hub_scope, "班级下拉框", self.VALUE)])
|
||
similar = "班级是下拉框，要先点开再从选项中选择，选项形如「计科2301班（计算机科学与技术）」"
|
||
stats = hub.receive_drafts("pytest_b", [draft(hub_scope, "班级选择控件", similar)])
|
||
assert stats["merged"] == 1 and stats["created"] == 0
|
||
row = hub.list_entries(knowledge_id=hub_scope)[0]
|
||
assert row["school_count"] == 2
|
||
assert set((row["schools"] or "").split(",")) == {"pytest_a", "pytest_b"}
|
||
|
||
def test_same_school_reporting_twice_does_not_inflate_school_count(self, hub_scope):
|
||
hub.receive_drafts("pytest_a", [draft(hub_scope, "班级下拉框", self.VALUE)])
|
||
hub.receive_drafts("pytest_a", [draft(hub_scope, "班级下拉框", self.VALUE)])
|
||
assert hub.list_entries(knowledge_id=hub_scope)[0]["school_count"] == 1
|
||
|
||
def test_merge_keeps_the_longer_value(self, hub_scope):
|
||
hub.receive_drafts("pytest_a", [draft(hub_scope, "提交入口", "提交按钮在表单底部")])
|
||
longer = "提交按钮在表单底部，叫「提交填报」，不要点旁边的「保存草稿」"
|
||
hub.receive_drafts("pytest_b", [draft(hub_scope, "提交入口", longer)])
|
||
assert hub.list_entries(knowledge_id=hub_scope)[0]["value"] == longer
|
||
|
||
def test_rejected_knowledge_does_not_come_back(self, hub_scope):
|
||
hub.receive_drafts("pytest_a", [draft(hub_scope, "错的经验", "这条被中心驳回过，不该再回到队列里")])
|
||
entry = hub.list_entries(knowledge_id=hub_scope)[0]
|
||
hub.review_entry(entry["id"], "reject")
|
||
stats = hub.receive_drafts("pytest_a", [draft(hub_scope, "错的经验", "这条被中心驳回过，不该再回到队列里")])
|
||
assert stats["skipped"] == 1
|
||
assert hub.list_entries(knowledge_id=hub_scope, status="draft") == []
|
||
|
||
def test_entries_without_key_or_value_are_skipped(self, hub_scope):
|
||
stats = hub.receive_drafts("pytest_a", [
|
||
{"knowledge_id": hub_scope, "business_id": "b1", "key": "", "value": "有值没名"},
|
||
{"knowledge_id": hub_scope, "business_id": "b1", "key": "有名没值", "value": " "},
|
||
])
|
||
assert stats == {"created": 0, "merged": 0, "skipped": 2, "revived": 0}
|
||
|
||
def test_unknown_category_degrades_to_note(self, hub_scope):
|
||
hub.receive_drafts("pytest_a", [draft(hub_scope, "k", "模型编了个类别出来", category="模型编的")])
|
||
assert hub.list_entries(knowledge_id=hub_scope)[0]["category"] == "note"
|
||
|
||
|
||
@pytest.mark.db
|
||
class TestPublishing:
|
||
def test_verified_entry_reaches_its_origin_school(self, hub_scope):
|
||
hub.receive_drafts("pytest_a", [draft(hub_scope, "k1", "这条只发回来源学校，别的学校拿不到")])
|
||
entry = hub.list_entries(knowledge_id=hub_scope)[0]
|
||
hub.review_entry(entry["id"], "verify", scope="origin")
|
||
mine = hub.published_for("pytest_a")["entries"]
|
||
assert any(e["id"] == entry["id"] for e in mine)
|
||
|
||
def test_origin_scope_does_not_leak_to_other_schools(self, hub_scope):
|
||
hub.receive_drafts("pytest_a", [draft(hub_scope, "k1", "这条只发回来源学校，别的学校拿不到")])
|
||
entry = hub.list_entries(knowledge_id=hub_scope)[0]
|
||
hub.review_entry(entry["id"], "verify", scope="origin")
|
||
assert not any(e["id"] == entry["id"] for e in hub.published_for("pytest_c")["entries"])
|
||
|
||
def test_scope_all_reaches_a_school_that_never_reported_it(self, hub_scope):
|
||
hub.receive_drafts("pytest_a", [draft(hub_scope, "k2", "这条发给所有学校，没报过的也能拿到")])
|
||
entry = hub.list_entries(knowledge_id=hub_scope)[0]
|
||
hub.review_entry(entry["id"], "verify", scope="all")
|
||
assert any(e["id"] == entry["id"] for e in hub.published_for("pytest_c")["entries"])
|
||
|
||
def test_rejections_are_published_too(self, hub_scope):
|
||
# 本地可能已经自动晋升过这条，必须把驳回结果也送回去
|
||
hub.receive_drafts("pytest_a", [draft(hub_scope, "k3", "本地自动生效过但中心判定是错的")])
|
||
entry = hub.list_entries(knowledge_id=hub_scope)[0]
|
||
hub.review_entry(entry["id"], "reject")
|
||
got = [e for e in hub.published_for("pytest_a")["entries"] if e["id"] == entry["id"]]
|
||
assert got and got[0]["status"] == "rejected"
|
||
|
||
def test_drafts_are_never_published(self, hub_scope):
|
||
hub.receive_drafts("pytest_a", [draft(hub_scope, "k4", "还在草稿状态，不该发给任何人")])
|
||
entry = hub.list_entries(knowledge_id=hub_scope)[0]
|
||
assert not any(e["id"] == entry["id"] for e in hub.published_for("pytest_a")["entries"])
|
||
|
||
def test_pull_is_incremental(self, hub_scope):
|
||
hub.receive_drafts("pytest_a", [draft(hub_scope, "k5", "第一条发布出去的经验，用来验证增量拉取")])
|
||
entry = hub.list_entries(knowledge_id=hub_scope)[0]
|
||
hub.review_entry(entry["id"], "verify", scope="origin")
|
||
first = hub.published_for("pytest_a")
|
||
assert any(e["id"] == entry["id"] for e in first["entries"])
|
||
again = hub.published_for("pytest_a", since=first["version"])
|
||
assert not any(e["id"] == entry["id"] for e in again["entries"])
|
||
|
||
def test_review_can_rewrite_the_value(self, hub_scope):
|
||
hub.receive_drafts("pytest_a", [draft(hub_scope, "k6", "原始表述比较含糊，审核时会被改写")])
|
||
entry = hub.list_entries(knowledge_id=hub_scope)[0]
|
||
row = hub.review_entry(entry["id"], "verify", value="审核改写后的清晰表述", scope="origin")
|
||
assert row["value"] == "审核改写后的清晰表述"
|
||
|
||
def test_reset_sends_an_entry_back_to_the_queue(self, hub_scope):
|
||
hub.receive_drafts("pytest_a", [draft(hub_scope, "k7", "审错了要能退回待审重新处理")])
|
||
entry = hub.list_entries(knowledge_id=hub_scope)[0]
|
||
hub.review_entry(entry["id"], "verify", scope="origin")
|
||
row = hub.review_entry(entry["id"], "reset")
|
||
assert row["status"] == "draft" and row["version"] == 0 and row["reviewed_by"] is None
|
||
|
||
def test_unknown_action_is_refused(self, hub_scope):
|
||
with pytest.raises(ValueError, match="未知的审核动作"):
|
||
hub.review_entry(1, "delete")
|
||
|
||
|
||
@pytest.mark.db
|
||
class TestDesktopApply:
|
||
"""中心发布的结果落到桌面端本地库。"""
|
||
|
||
KID = "pytest_apply"
|
||
|
||
def test_verified_entry_becomes_locally_active(self, test_school):
|
||
from backend.db import database as db
|
||
|
||
entries = [{"knowledge_id": self.KID, "business_id": "b1", "key": "中心采纳的经验",
|
||
"value": "这条经过中心审核，应该直接在本地生效并注入提示词", "category": "control_tip",
|
||
"status": "verified", "version": 1}]
|
||
assert db.apply_hub_entries(entries)["created"] == 1
|
||
rows = db.list_learned(knowledge_id=self.KID, business_id="b1", status="verified")
|
||
assert len(rows) == 1 and rows[0]["verified_by"] == "hub"
|
||
assert "中心采纳的经验" in db.verified_learned(self.KID, "b1")
|
||
|
||
def test_rewritten_value_replaces_the_local_one(self, test_school):
|
||
from backend.db import database as db
|
||
|
||
original = "班级是下拉框，要先点开再选"
|
||
db.add_learned_drafts(self.KID, "b1", [{"key": "班级下拉框", "value": original, "category": "control_tip"}], "t1")
|
||
rewritten = "班级是下拉框，要先点开再选；选项可能要滚动才能看到目标班级"
|
||
db.apply_hub_entries([{"knowledge_id": self.KID, "business_id": "b1", "key": "班级下拉框",
|
||
"value": rewritten, "category": "control_tip", "status": "verified", "version": 2}])
|
||
rows = db.list_learned(knowledge_id=self.KID, business_id="b1")
|
||
assert len(rows) == 1 and rows[0]["value"] == rewritten and rows[0]["status"] == "verified"
|
||
|
||
def test_central_rejection_disables_a_locally_promoted_entry(self, test_school):
|
||
from backend.db import database as db
|
||
|
||
value = "这条本地自动晋升过，但中心判定它是错的"
|
||
db.add_learned_drafts(self.KID, "b1", [{"key": "错的经验", "value": value, "category": "control_tip"}], "t1")
|
||
entry = db.list_learned(knowledge_id=self.KID, business_id="b1")[0]
|
||
db.set_learned_status(entry["id"], "verified")
|
||
assert db.verified_learned(self.KID, "b1")
|
||
|
||
db.apply_hub_entries([{"knowledge_id": self.KID, "business_id": "b1", "key": "错的经验",
|
||
"value": value, "category": "control_tip", "status": "rejected", "version": 3}])
|
||
assert db.verified_learned(self.KID, "b1") == {}
|
||
|
||
def test_rejection_without_a_local_match_is_ignored(self, test_school):
|
||
from backend.db import database as db
|
||
|
||
stats = db.apply_hub_entries([{"knowledge_id": self.KID, "business_id": "b1", "key": "本地没有这条",
|
||
"value": "本地从来没有过这条经验，驳回它没有意义", "status": "rejected", "version": 4}])
|
||
assert stats["skipped"] == 1 and stats["created"] == 0
|
||
|
||
def test_malformed_entries_are_skipped(self, test_school):
|
||
from backend.db import database as db
|
||
|
||
stats = db.apply_hub_entries([
|
||
{"knowledge_id": self.KID, "key": "", "value": "没有 key", "status": "verified"},
|
||
{"knowledge_id": self.KID, "key": "没有状态", "value": "状态字段缺失", "status": ""},
|
||
])
|
||
assert stats["skipped"] == 2
|
||
|
||
def test_empty_input(self, test_school):
|
||
from backend.db import database as db
|
||
|
||
assert db.apply_hub_entries([]) == {"verified": 0, "rejected": 0, "created": 0, "skipped": 0}
|
||
|
||
|
||
class TestSyncDisabled:
|
||
"""没配中心服务端时，同步必须安静地不做事，绝不能影响办事。"""
|
||
|
||
def test_disabled_by_default(self, monkeypatch):
|
||
from backend import hub_sync
|
||
from backend.config import settings
|
||
|
||
monkeypatch.setattr(settings, "hub_url", "")
|
||
monkeypatch.setattr(settings, "hub_token", "")
|
||
assert hub_sync.enabled() is False
|
||
|
||
def test_needs_both_url_and_token(self, monkeypatch):
|
||
from backend import hub_sync
|
||
from backend.config import settings
|
||
|
||
monkeypatch.setattr(settings, "hub_url", "http://127.0.0.1:8801")
|
||
monkeypatch.setattr(settings, "hub_token", "")
|
||
assert hub_sync.enabled() is False
|
||
|
||
async def test_report_and_pull_are_no_ops_when_disabled(self, monkeypatch):
|
||
from backend import hub_sync
|
||
from backend.config import settings
|
||
|
||
monkeypatch.setattr(settings, "hub_url", "")
|
||
monkeypatch.setattr(settings, "hub_token", "")
|
||
assert "skipped" in await hub_sync.report_drafts()
|
||
assert "skipped" in await hub_sync.pull_published()
|
||
|
||
async def test_unreachable_hub_returns_an_error_instead_of_raising(self, monkeypatch):
|
||
from backend import hub_sync
|
||
from backend.config import settings
|
||
|
||
monkeypatch.setattr(settings, "hub_url", "http://127.0.0.1:9") # 关着的端口
|
||
monkeypatch.setattr(settings, "hub_token", "sa_whatever")
|
||
monkeypatch.setattr(settings, "hub_timeout_seconds", 2.0)
|
||
monkeypatch.setattr(hub_sync.db, "set_setting", lambda *a: None)
|
||
monkeypatch.setattr(hub_sync.db, "get_setting", lambda *a, **k: "0")
|
||
result = await hub_sync.pull_published()
|
||
assert "error" in result
|