Files
verify-promotion-metric-def…/tests/test_verify_rules.py

239 lines
11 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
"""职称硬指标核验内核金标准用例(离线、确定性)。"""
from __future__ import annotations
import json
import os
import tempfile
import unittest
from _support import IsolatedDataRoot
def _metrics(rank="副教授", **overrides) -> dict:
base = {
"teacherName": "张老师",
"department": "化材学院",
"applyRank": rank,
"presidingProvincial": 1,
"corePapers": 4,
"teachingHours": 180,
"currentTitleYears": 6,
}
base.update(overrides)
return base
class VerifyRuleTestBase(unittest.TestCase):
def setUp(self) -> None:
self._root = IsolatedDataRoot()
self._root.__enter__()
from db.connection import init_db
init_db()
def tearDown(self) -> None:
self._root.__exit__(None, None, None)
def _verify(self, metrics=None, rank="副教授"):
from service.verify_service import load_thresholds, verify_metrics
return verify_metrics(metrics if metrics is not None else _metrics(rank=rank), load_thresholds(rank))
class TestMatrixColors(VerifyRuleTestBase):
def test_all_safe(self) -> None:
result = self._verify(_metrics(presidingProvincial=2, corePapers=5, teachingHours=200, currentTitleYears=8))
self.assertEqual(result["verdict"], "安全")
colors = {m["metric_key"]: m["color"] for m in result["matrix"]}
self.assertTrue(all(c == "green" for c in colors.values()))
def test_red_hard_fail(self) -> None:
result = self._verify(_metrics(corePapers=1))
self.assertEqual(result["verdict"], "硬伤")
red = [m for m in result["matrix"] if m["color"] == "red"]
self.assertEqual(red[0]["metric_key"], "core_papers")
self.assertIn("依据", red[0]["advice"])
def test_yellow_border_below_10_percent(self) -> None:
result = self._verify(_metrics(presidingProvincial=1, corePapers=3, teachingHours=165, currentTitleYears=6))
self.assertEqual(result["verdict"], "边缘")
colors = {m["metric_key"]: m["color"] for m in result["matrix"]}
self.assertEqual(colors["teaching_hours"], "yellow") # 165/160 = 3.1% 余量
self.assertEqual(colors["core_papers"], "yellow") # 3/3 = 0 余量
def test_green_boundary_at_exactly_10_percent(self) -> None:
# 恰好 10% 余量(176 = 160×1.10)属绿
result = self._verify(_metrics(teachingHours=176))
colors = {m["metric_key"]: m["color"] for m in result["matrix"]}
self.assertEqual(colors["teaching_hours"], "green")
def test_yellow_boundary_at_109_percent(self) -> None:
# 1.09 倍(174.4 = 160×1.09,余量 9%)属黄
result = self._verify(_metrics(teachingHours=174.4))
colors = {m["metric_key"]: m["color"] for m in result["matrix"]}
self.assertEqual(colors["teaching_hours"], "yellow")
def test_alias_metric_without_threshold_gets_v1(self) -> None:
# 副教授种子未配置 awards:提供该指标须出 V1 灰行,不得当作已核验跳过
result = self._verify(_metrics(awards=2))
v1 = [m for m in result["matrix"] if m["metric_key"] == "awards"]
self.assertEqual(len(v1), 1)
self.assertIn("V1", v1[0]["advice"])
def test_empty_thresholds_never_safe(self) -> None:
"""该级别门槛库为空 → 一律不判"安全"(V1 灰行 + 边缘结论)。"""
from service.verify_service import verify_metrics
result = verify_metrics(_metrics(), {})
self.assertEqual(result["verdict"], "边缘")
self.assertIn("门槛库未配置", result["summary"])
provided = [m for m in result["matrix"] if m["metric_key"] in ("presiding_provincial", "core_papers")]
self.assertTrue(provided and all(m["color"] == "gray" for m in provided))
bare = verify_metrics({"teacherName": "某人", "applyRank": "副教授"}, {})
self.assertEqual(bare["verdict"], "边缘")
self.assertIn("门槛库未配置", bare["summary"])
def test_r4_missing_metric_pending(self) -> None:
result = self._verify(_metrics(corePapers=None))
self.assertEqual(result["verdict"], "边缘")
colors = {m["metric_key"]: m["color"] for m in result["matrix"]}
self.assertEqual(colors["core_papers"], "gray")
pending = next(m for m in result["matrix"] if m["metric_key"] == "core_papers")
self.assertIn("R4", pending["advice"])
def test_v1_unconfigured_metric(self) -> None:
result = self._verify(_metrics(someExtraAward=3))
v1 = [m for m in result["matrix"] if m["metric_key"] == "someExtraAward"]
self.assertTrue(v1)
self.assertIn("V1", v1[0]["advice"])
class TestVerifyKernel(VerifyRuleTestBase):
def _run(self, cases):
from service.verify_service import run_verify
return run_verify(cases)
def test_run_verify_writes_reports_and_brief(self) -> None:
result = self._run([_metrics(presidingProvincial=2), _metrics(teacherName="李老师", corePapers=1)])
self.assertEqual(result["count"], 2)
self.assertEqual(result["verdict_counts"], {"安全": 1, "硬伤": 1})
self.assertTrue(result["brief_file"].endswith(".html"))
for report in result["reports"]:
self.assertTrue(report["report_file"].endswith(".html"))
with open(report["report_file"], encoding="utf-8") as f:
html_text = f.read()
self.assertIn("职称申报核验报告", html_text)
self.assertIn("不构成评审结论", html_text)
def test_empty_cases_raises(self) -> None:
with self.assertRaises(ValueError):
self._run([])
def test_professor_rank_thresholds(self) -> None:
result = self._run([_metrics(rank="教授", presidingProvincial=2, corePapers=7, monographs=2)])
self.assertEqual(result["reports"][0]["verdict"], "边缘") # core 7/6 余量 16.7% 绿、presiding 2/2 黄
def test_rank_invalid_raises(self) -> None:
with self.assertRaises(ValueError) as ctx:
self._run([_metrics(rank="院士", presidingProvincial=0)])
self.assertIn("RANK_INVALID", str(ctx.exception))
def test_professor_first_author_threshold_enforced(self) -> None:
"""教授"一作/通讯 ≥3"为独立门槛:缺数据判待核(边缘),提供后参与判定。"""
from service.verify_service import load_thresholds, verify_metrics
base = _metrics(rank="教授", presidingProvincial=3, corePapers=7, monographs=2, currentTitleYears=6)
without = verify_metrics(base, load_thresholds("教授"))
self.assertEqual(without["verdict"], "边缘")
pending = next(m for m in without["matrix"] if m["metric_key"] == "first_author_core_papers")
self.assertEqual(pending["color"], "gray")
full = verify_metrics(dict(base, firstAuthorCorePapers=4), load_thresholds("教授"))
fa = next(m for m in full["matrix"] if m["metric_key"] == "first_author_core_papers")
self.assertEqual(fa["color"], "green")
self.assertEqual(fa["required"], 3.0)
def test_set_threshold_changes_verdict(self) -> None:
from service.threshold_service import upsert_threshold
upsert_threshold({"rank": "副教授", "metric_key": "core_papers", "metric_name": "核心期刊论文", "min_value": 5})
result = self._verify(_metrics(corePapers=4))
self.assertEqual(result["verdict"], "硬伤")
class TestThresholdService(VerifyRuleTestBase):
def test_upsert_threshold_create_update(self) -> None:
from service.threshold_service import list_thresholds, upsert_threshold
payload = upsert_threshold({"rank": "副教授", "metric_key": "awards", "metric_name": "教学/科研获奖", "min_value": 1})
self.assertEqual(payload["action"], "created")
payload = upsert_threshold({"rank": "副教授", "metric_key": "awards", "min_value": 2})
self.assertEqual(payload["action"], "updated")
rows = [t for t in list_thresholds("副教授") if t["metric_key"] == "awards"]
self.assertEqual(rows[0]["min_value"], 2.0)
def test_upsert_threshold_validations(self) -> None:
from service.threshold_service import upsert_threshold
with self.assertRaises(ValueError):
upsert_threshold({"rank": "院士", "metric_key": "awards", "min_value": 1})
with self.assertRaises(ValueError):
upsert_threshold({"rank": "副教授", "metric_key": "", "min_value": 1})
with self.assertRaises(ValueError):
upsert_threshold({"rank": "副教授", "metric_key": "awards", "min_value": -1})
def test_seeded_thresholds(self) -> None:
from service.threshold_service import list_thresholds
self.assertGreaterEqual(len(list_thresholds("副教授")), 4)
self.assertGreaterEqual(len(list_thresholds("教授")), 4)
def test_stats(self) -> None:
from service.threshold_service import stats
from service.verify_service import run_verify
run_verify([_metrics(presidingProvincial=2), _metrics(teacherName="李老师", corePapers=1)])
payload = stats()
self.assertEqual(payload["case_total"], 2)
self.assertEqual(payload["by_verdict"].get("硬伤"), 1)
self.assertTrue(payload["red_cases"])
class TestBatchInputCollection(VerifyRuleTestBase):
"""input/ 目录收集:坏文件跳过并记录原因,不中断整批。"""
def test_collect_input_cases_skips_bad_files(self) -> None:
from service.verify_service import collect_input_cases
with tempfile.TemporaryDirectory() as tmp:
with open(os.path.join(tmp, "a.json"), "w", encoding="utf-8") as f:
json.dump([_metrics(), _metrics(teacherName="李老师")], f, ensure_ascii=False)
with open(os.path.join(tmp, "single.json"), "w", encoding="utf-8") as f:
json.dump({"teacherName": "王老师", "applyRank": "副教授"}, f, ensure_ascii=False)
with open(os.path.join(tmp, "bad.json"), "w", encoding="utf-8") as f:
f.write("{not-json")
with open(os.path.join(tmp, "scalar.json"), "w", encoding="utf-8") as f:
json.dump("just-a-string", f, ensure_ascii=False)
cases, skipped = collect_input_cases(tmp)
names = [c.get("teacherName") for c in cases]
self.assertEqual(names, ["张老师", "李老师", "王老师"])
self.assertEqual({e["file"] for e in skipped}, {"bad.json", "scalar.json"})
self.assertTrue(all(e["reason"] for e in skipped))
def test_collect_input_cases_empty_dir(self) -> None:
from service.verify_service import collect_input_cases
with tempfile.TemporaryDirectory() as tmp:
cases, skipped = collect_input_cases(tmp)
self.assertEqual(cases, [])
self.assertEqual(skipped, [])
if __name__ == "__main__":
unittest.main()