613 lines
24 KiB
Python
613 lines
24 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""默认测试套件：合同到期智能预警与续聘辅助。
|
||
|
||
设计原则：
|
||
- 仅用标准库（不依赖 pytest / jinja2），CI 与本地均能直接 `python tests/run_tests.py` 跑
|
||
- 用 sqlite3 ":memory:" 隔离 DB，不污染仓库 data/ 目录
|
||
- 覆盖：util.date_util 纯函数 / util.constants 版本对齐 / repository 新增功能
|
||
（parse_sandbox_card_performance + upsert_performance_record）/ scanner 分级 / task_generator
|
||
规则版 / html_renderer 降级路径
|
||
- 退出码：0 = 全部通过，1 = 有失败
|
||
- -v 开关：详细输出每个用例的状态
|
||
|
||
release.ps1 在打 tag 前会自动跑 `python tests/run_tests.py -v`，任何一个
|
||
test_* 返回 (False, msg) 即 abort 发布。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import os
|
||
import re
|
||
import shutil
|
||
import sqlite3
|
||
import sys
|
||
import tempfile
|
||
import traceback
|
||
from datetime import date
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from typing import Any, Callable
|
||
|
||
# 让 scripts/ 下的模块可导入
|
||
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
||
_SCRIPTS_DIR = _REPO_ROOT / "scripts"
|
||
sys.path.insert(0, str(_SCRIPTS_DIR))
|
||
|
||
|
||
# ---------- 测试基础设施 ----------
|
||
|
||
# 测试结果收集器
|
||
_RESULTS: list[tuple[str, bool, str]] = []
|
||
_VERBOSE = False
|
||
|
||
|
||
def _run(name: str, fn: Callable[[], tuple[bool, str]]) -> None:
|
||
"""运行单个 test_*，记录结果。"""
|
||
try:
|
||
ok, msg = fn()
|
||
except Exception as e: # 测试本身抛异常也算失败
|
||
tb = traceback.format_exc(limit=3)
|
||
ok, msg = False, f"exception: {e}\n{tb}"
|
||
_RESULTS.append((name, ok, msg))
|
||
if _VERBOSE or not ok:
|
||
marker = "PASS" if ok else "FAIL"
|
||
print(f" [{marker}] {name}: {msg}", flush=True)
|
||
|
||
|
||
def _make_conn() -> sqlite3.Connection:
|
||
"""建一张 in-memory DB，应用所有 DDL + 迁移（不加载 seed）。"""
|
||
from db.connection import DDL_STATEMENTS
|
||
from db.migrations import apply_all
|
||
|
||
conn = sqlite3.connect(":memory:")
|
||
conn.row_factory = sqlite3.Row
|
||
conn.execute("PRAGMA foreign_keys = ON;")
|
||
for ddl in DDL_STATEMENTS:
|
||
conn.execute(ddl)
|
||
apply_all(conn)
|
||
conn.commit()
|
||
return conn
|
||
|
||
|
||
class _FakeRow:
|
||
"""Mock 一行 sqlite3.Row：同时支持下标访问 (r["x"]) 和属性访问 (r.x)。
|
||
|
||
production 代码里 task_generator.suggest() 用的是 r["moral_teaching_score"]
|
||
下标语法，所以 fake row 必须也支持。
|
||
"""
|
||
|
||
__slots__ = ("year", "performance_score", "moral_teaching_score",
|
||
"research_score", "teaching_score")
|
||
|
||
def __init__(self, year: int, score: float, moral: float):
|
||
self.year = year
|
||
self.performance_score = score
|
||
self.moral_teaching_score = moral
|
||
self.research_score = score
|
||
self.teaching_score = score
|
||
|
||
def __getitem__(self, key: str) -> Any:
|
||
return getattr(self, key)
|
||
|
||
|
||
def _fake_perf(year: int, score: float, moral: float | None = None) -> _FakeRow:
|
||
"""构造一个 performance_records-like 行（用于纯函数测试，不入 DB）。"""
|
||
m = moral if moral is not None else score
|
||
return _FakeRow(year=year, score=score, moral=m)
|
||
|
||
|
||
def _insert_stub_contract(conn: sqlite3.Connection, contract_id: str) -> None:
|
||
"""插入一份占位合同（performance_records 的 FK 目标）。"""
|
||
conn.execute(
|
||
"""INSERT INTO contracts
|
||
(contract_id, employee_id, teacher_name, unit_id, unit_name,
|
||
position, contract_type, contract_start, contract_end,
|
||
trial_period_end, status, last_material_id, renewed_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '可续聘', NULL, NULL)""",
|
||
(contract_id, f"E-{contract_id}", f"占位-{contract_id}", "U-TEST",
|
||
"测试系", "讲师", "正式", "2022-01-01", "2099-12-31", "2022-07-01"),
|
||
)
|
||
conn.commit()
|
||
|
||
|
||
# ---------- util.date_util ----------
|
||
|
||
def test_parse_iso_date() -> tuple[bool, str]:
|
||
from util.date_util import parse_iso_date
|
||
cases = [
|
||
("2024-03-15", date(2024, 3, 15)),
|
||
("2024-03-15T08:30:00", date(2024, 3, 15)),
|
||
(" 2024-03-15 ", date(2024, 3, 15)),
|
||
(date(2024, 3, 15), date(2024, 3, 15)), # 已为 date
|
||
]
|
||
for inp, want in cases:
|
||
got = parse_iso_date(inp)
|
||
if got != want:
|
||
return False, f"parse_iso_date({inp!r}) = {got}, want {want}"
|
||
return True, "4 cases ok"
|
||
|
||
|
||
def test_days_to() -> tuple[bool, str]:
|
||
from util.date_util import days_to
|
||
if days_to(date(2024, 3, 15), today=date(2024, 1, 1)) != 74:
|
||
return False, "days_to 正向计算错"
|
||
if days_to(date(2023, 12, 31), today=date(2024, 1, 1)) != -1:
|
||
return False, "days_to 已过期返回负数错"
|
||
# 默认 today=今天
|
||
if days_to(date(2099, 1, 1)) <= 0:
|
||
return False, "days_to 默认 today 应为今天"
|
||
return True, "正/负/默认 today 三种场景"
|
||
|
||
|
||
def test_classify_alert() -> tuple[bool, str]:
|
||
from util.date_util import classify_alert
|
||
t = (90, 60, 30)
|
||
cases = [
|
||
(100, None), # 超出预警范围
|
||
(91, None), # 边界外
|
||
(90, "yellow"),
|
||
(61, "yellow"),
|
||
(60, "orange"),
|
||
(31, "orange"),
|
||
(30, "red"),
|
||
(1, "red"),
|
||
(0, "red"),
|
||
(-5, "red"), # 已过期仍属 red
|
||
]
|
||
for days, want in cases:
|
||
got = classify_alert(days, t)
|
||
if got != want:
|
||
return False, f"classify_alert({days}) = {got}, want {want}"
|
||
return True, "10 边界 + 越界场景"
|
||
|
||
|
||
def test_month_match() -> tuple[bool, str]:
|
||
from util.date_util import month_match
|
||
d = date(2024, 3, 15)
|
||
if not month_match(d, "2024-03"):
|
||
return False, "同月应命中"
|
||
if month_match(d, "2024-04"):
|
||
return False, "不同月不应命中"
|
||
if not month_match(d, ""):
|
||
return False, "空 month_str 应命中全部"
|
||
if month_match(d, "garbage"):
|
||
return False, "非法 month_str 不应命中"
|
||
return True, "命中 / 未命中 / 空 / 非法 四种"
|
||
|
||
|
||
# ---------- util.constants 版本对齐（POLICY-RELEASE-001） ----------
|
||
|
||
def test_skill_md_version_alignment() -> tuple[bool, str]:
|
||
"""POLICY-RELEASE-001：SKILL.md version == SKILL_VERSION == CHANGELOG 顶部版本"""
|
||
from util.constants import SKILL_VERSION
|
||
|
||
skill_md = (_REPO_ROOT / "SKILL.md").read_text(encoding="utf-8")
|
||
m = re.search(r'(?m)^version:\s*["\']?(\d+\.\d+\.\d+)', skill_md)
|
||
if not m:
|
||
return False, "SKILL.md 缺顶层 version: X.Y.Z"
|
||
md_ver = m.group(1)
|
||
if md_ver != SKILL_VERSION:
|
||
return False, f"SKILL.md={md_ver} != SKILL_VERSION={SKILL_VERSION}"
|
||
|
||
# CHANGELOG 顶部应有 ## X.Y.Z
|
||
cl = (_REPO_ROOT / "CHANGELOG.md").read_text(encoding="utf-8")
|
||
cl_m = re.search(r'(?m)^##\s+(\d+\.\d+\.\d+)', cl)
|
||
if not cl_m:
|
||
return False, "CHANGELOG.md 缺顶层 ## X.Y.Z"
|
||
cl_ver = cl_m.group(1)
|
||
if cl_ver != SKILL_VERSION:
|
||
return False, f"CHANGELOG 顶部={cl_ver} != SKILL_VERSION={SKILL_VERSION}"
|
||
|
||
return True, f"SKILL.md={md_ver} == CHANGELOG={cl_ver} == SKILL_VERSION={SKILL_VERSION}"
|
||
|
||
|
||
def test_skill_md_slug_alignment() -> tuple[bool, str]:
|
||
"""SKILL.md metadata.openclaw.slug == SKILL_SLUG"""
|
||
from util.constants import SKILL_SLUG
|
||
|
||
skill_md = (_REPO_ROOT / "SKILL.md").read_text(encoding="utf-8")
|
||
m = re.search(r'(?m)^\s*slug:\s*["\']?([a-z0-9_-]+)', skill_md)
|
||
if not m:
|
||
return False, "SKILL.md 缺 metadata.openclaw.slug"
|
||
md_slug = m.group(1)
|
||
if md_slug != SKILL_SLUG:
|
||
return False, f"SKILL.md slug={md_slug} != SKILL_SLUG={SKILL_SLUG}"
|
||
return True, f"slug={md_slug}"
|
||
|
||
|
||
# ---------- db.repository：v1.1.2 新增的 sandbox card 解析 ----------
|
||
|
||
def test_parse_sandbox_card_row_format() -> tuple[bool, str]:
|
||
"""v1.1.2：行格式 '2024\\t称职'"""
|
||
from db import repository as repo
|
||
|
||
conn = _make_conn()
|
||
_insert_stub_contract(conn, "C001")
|
||
card_text = "year\tresult\n2023\t优秀\n2024\t称职\n2025\t基本称职\n"
|
||
n = repo.parse_sandbox_card_performance(conn, "C001", card_text)
|
||
if n != 3:
|
||
return False, f"写入 {n} 条，期望 3"
|
||
|
||
rows = repo.list_performance_by_contract(conn, "C001")
|
||
if len(rows) != 3:
|
||
return False, f"DB 行数 {len(rows)}，期望 3"
|
||
by_year = {r["year"]: r["performance_score"] for r in rows}
|
||
expected = {2023: 90.0, 2024: 75.0, 2025: 60.0}
|
||
for y, want in expected.items():
|
||
if by_year.get(y) != want:
|
||
return False, f"{y} 年分数={by_year.get(y)}，期望 {want}"
|
||
return True, "3 行（优秀/称职/基本称职）解析 + 写入正确"
|
||
|
||
|
||
def test_parse_sandbox_card_narrative_format() -> tuple[bool, str]:
|
||
"""v1.1.2：叙述式 '2024年：等次「称职」'"""
|
||
from db import repository as repo
|
||
|
||
conn = _make_conn()
|
||
_insert_stub_contract(conn, "C002")
|
||
card_text = "考核情况：\n2023年：等次「优秀」。\n2024年：等次「称职」。\n2025年：等次「不称职」。\n"
|
||
n = repo.parse_sandbox_card_performance(conn, "C002", card_text)
|
||
if n != 3:
|
||
return False, f"写入 {n} 条，期望 3"
|
||
|
||
rows = repo.list_performance_by_contract(conn, "C002")
|
||
by_year = {r["year"]: r["performance_score"] for r in rows}
|
||
expected = {2023: 90.0, 2024: 75.0, 2025: 30.0}
|
||
for y, want in expected.items():
|
||
if by_year.get(y) != want:
|
||
return False, f"{y} 年分数={by_year.get(y)}，期望 {want}"
|
||
return True, "3 行（优秀/称职/不称职）叙述式解析正确"
|
||
|
||
|
||
def test_parse_sandbox_card_empty_and_invalid() -> tuple[bool, str]:
|
||
"""空字符串 / 无匹配应返回 0，不报错"""
|
||
from db import repository as repo
|
||
|
||
conn = _make_conn()
|
||
_insert_stub_contract(conn, "C001")
|
||
if repo.parse_sandbox_card_performance(conn, "C001", "") != 0:
|
||
return False, "空字符串应返回 0"
|
||
if repo.parse_sandbox_card_performance(conn, "C001", "没有任何年份/等次信息") != 0:
|
||
return False, "无匹配应返回 0"
|
||
rows = repo.list_performance_by_contract(conn, "C001")
|
||
if rows:
|
||
return False, "无匹配不应写入任何行"
|
||
return True, "空 / 无匹配场景"
|
||
|
||
|
||
def test_parse_sandbox_card_mixed_formats() -> tuple[bool, str]:
|
||
"""同卡片混用两种格式时按行去重（PK = P-<contract>-<year>）"""
|
||
from db import repository as repo
|
||
|
||
conn = _make_conn()
|
||
_insert_stub_contract(conn, "C003")
|
||
card_text = (
|
||
"2024年：等次「称职」。\n" # 叙述式
|
||
"2025\t优秀\n" # 行格式
|
||
"2026年：等次「基本称职」\n"
|
||
)
|
||
n = repo.parse_sandbox_card_performance(conn, "C003", card_text)
|
||
if n != 3:
|
||
return False, f"混合解析写入 {n} 条，期望 3"
|
||
return True, "叙述式 + 行格式混合解析"
|
||
|
||
|
||
def test_upsert_performance_record_grade_mapping() -> tuple[bool, str]:
|
||
"""grade → score 映射 + ON CONFLICT upsert"""
|
||
from db import repository as repo
|
||
|
||
conn = _make_conn()
|
||
_insert_stub_contract(conn, "C001")
|
||
if not repo.upsert_performance_record(conn, "C001", 2024, "优秀"):
|
||
return False, "优秀应返回 True"
|
||
if not repo.upsert_performance_record(conn, "C001", 2024, "优秀"):
|
||
return False, "upsert 已存在应仍返回 True"
|
||
if not repo.upsert_performance_record(conn, "C001", 2025, "称职"):
|
||
return False, "称职应返回 True"
|
||
if not repo.upsert_performance_record(conn, "C001", 2026, "基本称职"):
|
||
return False, "基本称职应返回 True"
|
||
if not repo.upsert_performance_record(conn, "C001", 2027, "不称职"):
|
||
return False, "不称职应返回 True"
|
||
if not repo.upsert_performance_record(conn, "C001", 2028, "不合格"):
|
||
return False, "不合格应返回 True"
|
||
if repo.upsert_performance_record(conn, "C001", 2029, "未知等次"):
|
||
return False, "未知等次应返回 False"
|
||
|
||
rows = repo.list_performance_by_contract(conn, "C001")
|
||
# 期望 5 行：2024/2025/2026/2027/2028
|
||
# 第二次 (2024, 优秀) 是 upsert 覆盖不新增；2029 未知等次不入库
|
||
if len(rows) != 5:
|
||
return False, f"DB 行数 {len(rows)}，期望 5（upsert 同年不新增；未知等次不入库）"
|
||
by_year = {r["year"]: r["performance_score"] for r in rows}
|
||
expected = {2024: 90.0, 2025: 75.0, 2026: 60.0, 2027: 30.0, 2028: 30.0}
|
||
for y, want in expected.items():
|
||
if by_year.get(y) != want:
|
||
return False, f"{y} 年分数={by_year.get(y)}，期望 {want}"
|
||
# 验证 2024 的分数确实被 upsert 覆盖（不是 30/60/75 等其他值）
|
||
if by_year[2024] != 90.0:
|
||
return False, f"upsert 未保持分数：2024={by_year[2024]}，want 90.0"
|
||
return True, "grade 映射 + upsert 幂等 + 未知等次返回 False"
|
||
|
||
|
||
# ---------- service.task_generator.suggest 规则版 ----------
|
||
|
||
def test_suggest_empty_records() -> tuple[bool, str]:
|
||
"""无业绩记录 → observe（数据缺失）"""
|
||
from service.task_generator import suggest
|
||
s, r = suggest([])
|
||
if s != "observe":
|
||
return False, f"suggest([]) = {s}, want observe"
|
||
if "缺失" not in r:
|
||
return False, f"reason 应提示缺失，实际 {r!r}"
|
||
return True, "observe + '缺失'"
|
||
|
||
|
||
def test_suggest_low_moral() -> tuple[bool, str]:
|
||
"""师德 < 60 → non_renew"""
|
||
from service.task_generator import suggest
|
||
rows = [_fake_perf(2024, 85, moral=50)]
|
||
s, r = suggest(rows)
|
||
if s != "non_renew":
|
||
return False, f"suggest(师德 50) = {s}, want non_renew"
|
||
if "师德" not in r:
|
||
return False, f"reason 应含「师德」，实际 {r!r}"
|
||
return True, "non_renew + '师德'"
|
||
|
||
|
||
def test_suggest_fail_year() -> tuple[bool, str]:
|
||
"""任意一年考核 < 60 → observe（复核）"""
|
||
from service.task_generator import suggest
|
||
rows = [
|
||
_fake_perf(2023, 80, moral=85),
|
||
_fake_perf(2024, 50, moral=85), # 不及格
|
||
_fake_perf(2025, 85, moral=85),
|
||
]
|
||
s, r = suggest(rows)
|
||
if s != "observe":
|
||
return False, f"suggest(有不及格年) = {s}, want observe"
|
||
if "复核" not in r and "不合格" not in r:
|
||
return False, f"reason 应提示不合格/复核，实际 {r!r}"
|
||
return True, "observe + '不合格'"
|
||
|
||
|
||
def test_suggest_renew_upward() -> tuple[bool, str]:
|
||
"""全部 >= 80 且逐年上升 → renew"""
|
||
from service.task_generator import suggest
|
||
rows = [
|
||
_fake_perf(2023, 80, moral=85),
|
||
_fake_perf(2024, 85, moral=85),
|
||
_fake_perf(2025, 90, moral=85),
|
||
]
|
||
s, r = suggest(rows)
|
||
if s != "renew":
|
||
return False, f"suggest(全部 80+ 上升) = {s}, want renew"
|
||
return True, "renew + 上升趋势"
|
||
|
||
|
||
def test_suggest_observe_default() -> tuple[bool, str]:
|
||
"""达标但非逐年上升 → observe（常规评审）"""
|
||
from service.task_generator import suggest
|
||
rows = [
|
||
_fake_perf(2023, 85, moral=85),
|
||
_fake_perf(2024, 80, moral=85), # 回落
|
||
_fake_perf(2025, 85, moral=85),
|
||
]
|
||
s, r = suggest(rows)
|
||
if s != "observe":
|
||
return False, f"suggest(回落) = {s}, want observe"
|
||
return True, "observe（常规评审）"
|
||
|
||
|
||
# ---------- service.scanner 端到端 ----------
|
||
|
||
def test_scanner_no_alerts() -> tuple[bool, str]:
|
||
"""所有合同都在 90 天外 → 0 预警"""
|
||
from service import scanner
|
||
|
||
conn = _make_conn()
|
||
conn.executemany(
|
||
"""INSERT INTO contracts
|
||
(contract_id, employee_id, teacher_name, unit_id, unit_name,
|
||
position, contract_type, contract_start, contract_end,
|
||
trial_period_end, status, last_material_id, renewed_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '可续聘', NULL, NULL)""",
|
||
[
|
||
("C-FAR-1", "T001", "张三", "U1", "数学系", "教授", "正式",
|
||
"2022-01-01", "2099-12-31", "2022-07-01"),
|
||
],
|
||
)
|
||
conn.commit()
|
||
result = scanner.scan(conn, today=date(2024, 1, 15))
|
||
if result["scanned_count"] != 1:
|
||
return False, f"scanned={result['scanned_count']}, want 1"
|
||
if result["yellow_count"] + result["orange_count"] + result["red_count"] != 0:
|
||
return False, f"应无预警，实测 yellow={result['yellow_count']} orange={result['orange_count']} red={result['red_count']}"
|
||
return True, "1 合同全在 90 天外 → 0 预警"
|
||
|
||
|
||
def test_scanner_three_levels() -> tuple[bool, str]:
|
||
"""3 份合同分别命中黄/橙/红"""
|
||
from service import scanner
|
||
|
||
conn = _make_conn()
|
||
conn.executemany(
|
||
"""INSERT INTO contracts
|
||
(contract_id, employee_id, teacher_name, unit_id, unit_name,
|
||
position, contract_type, contract_start, contract_end,
|
||
trial_period_end, status, last_material_id, renewed_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '可续聘', NULL, NULL)""",
|
||
[
|
||
("C-YEL", "T001", "黄老师", "U1", "数学系", "教授", "正式",
|
||
"2022-01-01", "2024-05-15", "2022-07-01"), # 距 2024-02-15 = 90 天 → yellow
|
||
("C-ORA", "T002", "橙老师", "U1", "物理系", "讲师", "正式",
|
||
"2022-01-01", "2024-04-15", "2022-07-01"), # 距 2024-02-15 = 60 天 → orange
|
||
("C-RED", "T003", "红老师", "U1", "化学系", "副教授", "正式",
|
||
"2022-01-01", "2024-03-15", "2022-07-01"), # 距 2024-02-15 = 29 天 → red
|
||
],
|
||
)
|
||
conn.commit()
|
||
result = scanner.scan(conn, today=date(2024, 2, 15))
|
||
if result["scanned_count"] != 3:
|
||
return False, f"scanned={result['scanned_count']}, want 3"
|
||
if result["yellow_count"] != 1 or result["orange_count"] != 1 or result["red_count"] != 1:
|
||
return False, (f"分级数 yellow={result['yellow_count']} "
|
||
f"orange={result['orange_count']} red={result['red_count']}, "
|
||
f"want 1/1/1")
|
||
|
||
# 验证 alerts 表实际写入
|
||
alerts = conn.execute("SELECT contract_id, alert_level FROM alerts").fetchall()
|
||
by_cid = {a["contract_id"]: a["alert_level"] for a in alerts}
|
||
if by_cid.get("C-YEL") != "yellow":
|
||
return False, f"C-YEL 写入等级 {by_cid.get('C-YEL')}, want yellow"
|
||
if by_cid.get("C-ORA") != "orange":
|
||
return False, f"C-ORA 写入等级 {by_cid.get('C-ORA')}, want orange"
|
||
if by_cid.get("C-RED") != "red":
|
||
return False, f"C-RED 写入等级 {by_cid.get('C-RED')}, want red"
|
||
return True, "3 合同 → 1 黄 + 1 橙 + 1 红，alerts 表写入正确"
|
||
|
||
|
||
def test_scanner_skips_non_active_contracts() -> tuple[bool, str]:
|
||
"""status != '可续聘' 的合同不应被扫描"""
|
||
from service import scanner
|
||
|
||
conn = _make_conn()
|
||
conn.execute(
|
||
"""INSERT INTO contracts
|
||
(contract_id, employee_id, teacher_name, unit_id, unit_name,
|
||
position, contract_type, contract_start, contract_end,
|
||
trial_period_end, status, last_material_id, renewed_at)
|
||
VALUES ('C-DONE', 'T001', '已续签老师', 'U1', '数学系', '教授', '正式',
|
||
'2022-01-01', '2024-03-15', '2022-07-01', '已续签', NULL, NULL)"""
|
||
)
|
||
conn.commit()
|
||
result = scanner.scan(conn, today=date(2024, 2, 15))
|
||
if result["scanned_count"] != 0:
|
||
return False, f"已续签合同不应被扫描，scanned={result['scanned_count']}"
|
||
alerts = conn.execute("SELECT COUNT(*) AS n FROM alerts").fetchone()["n"]
|
||
if alerts != 0:
|
||
return False, f"应无 alerts 写入，实测 {alerts}"
|
||
return True, "已续签合同被跳过"
|
||
|
||
|
||
# ---------- service.html_renderer 降级路径 ----------
|
||
|
||
def test_html_render_basic() -> tuple[bool, str]:
|
||
"""HTML 渲染（无 jinja2 走 str.replace 降级路径）"""
|
||
from service import html_renderer
|
||
|
||
tmpdir = Path(tempfile.mkdtemp(prefix="alert-hr-test-"))
|
||
saved = os.environ.get("HTML_OUTPUT_DIR")
|
||
os.environ["HTML_OUTPUT_DIR"] = str(tmpdir)
|
||
try:
|
||
conn = _make_conn()
|
||
conn.execute(
|
||
"""INSERT INTO contracts
|
||
(contract_id, employee_id, teacher_name, unit_id, unit_name,
|
||
position, contract_type, contract_start, contract_end,
|
||
trial_period_end, status, last_material_id, renewed_at)
|
||
VALUES ('C-HTML', 'T100', '渲染测试', 'U9', '测试系', '教授', '正式',
|
||
'2022-01-01', '2024-06-30', '2022-07-01', '可续聘', NULL, NULL)"""
|
||
)
|
||
conn.execute(
|
||
"""INSERT INTO alerts
|
||
VALUES ('A-HTML', 'C-HTML', 'orange', 60, 'test',
|
||
'2024-04-01T00:00:00', 'open')"""
|
||
)
|
||
conn.execute(
|
||
"""INSERT INTO renewal_tasks
|
||
VALUES ('T-HTML', 'C-HTML', 'A-HTML', 'renew', '近 3 年考核良好',
|
||
'业绩良好', '["材料 A", "材料 B"]',
|
||
'2024-04-01T00:00:00', 'pending',
|
||
'rules', '')"""
|
||
)
|
||
conn.commit()
|
||
|
||
path = html_renderer.render_task_html(conn, "T-HTML")
|
||
if not path.exists():
|
||
return False, f"输出文件不存在 {path}"
|
||
|
||
text = path.read_text(encoding="utf-8")
|
||
if "渲染测试" not in text:
|
||
return False, "HTML 应含教师姓名"
|
||
if "续聘准备任务书" not in text:
|
||
return False, "HTML 应含 TITLE"
|
||
if "橙色" not in text:
|
||
return False, "HTML 应含「橙色」预警等级标签"
|
||
if "建议续聘" not in text:
|
||
return False, "HTML 应含「建议续聘」建议标签"
|
||
return True, f"HTML 渲染 OK → {path.name}"
|
||
finally:
|
||
if saved is None:
|
||
os.environ.pop("HTML_OUTPUT_DIR", None)
|
||
else:
|
||
os.environ["HTML_OUTPUT_DIR"] = saved
|
||
shutil.rmtree(tmpdir, ignore_errors=True)
|
||
|
||
|
||
def test_html_render_unknown_task_raises() -> tuple[bool, str]:
|
||
"""渲染不存在的 task_id 应抛 ValueError"""
|
||
from service import html_renderer
|
||
|
||
conn = _make_conn()
|
||
try:
|
||
html_renderer.render_task_html(conn, "T-NONEXISTENT")
|
||
except ValueError as e:
|
||
return True, f"ValueError OK: {e}"
|
||
return False, "应抛 ValueError，实际未抛"
|
||
|
||
|
||
# ---------- 入口 ----------
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="默认 release 测试套件")
|
||
parser.add_argument("-v", "--verbose", action="store_true",
|
||
help="详细输出每个用例结果")
|
||
args = parser.parse_args()
|
||
|
||
global _VERBOSE
|
||
_VERBOSE = args.verbose
|
||
|
||
print("=== alert-contract-expiring-hr release tests ===", flush=True)
|
||
|
||
_run("util/parse_iso_date", test_parse_iso_date)
|
||
_run("util/days_to", test_days_to)
|
||
_run("util/classify_alert", test_classify_alert)
|
||
_run("util/month_match", test_month_match)
|
||
_run("constants/skill_md_version_alignment", test_skill_md_version_alignment)
|
||
_run("constants/skill_md_slug_alignment", test_skill_md_slug_alignment)
|
||
_run("db/parse_sandbox_card_row", test_parse_sandbox_card_row_format)
|
||
_run("db/parse_sandbox_card_narrative", test_parse_sandbox_card_narrative_format)
|
||
_run("db/parse_sandbox_card_empty_invalid", test_parse_sandbox_card_empty_and_invalid)
|
||
_run("db/parse_sandbox_card_mixed", test_parse_sandbox_card_mixed_formats)
|
||
_run("db/upsert_performance_record", test_upsert_performance_record_grade_mapping)
|
||
_run("task_generator/suggest_empty", test_suggest_empty_records)
|
||
_run("task_generator/suggest_low_moral", test_suggest_low_moral)
|
||
_run("task_generator/suggest_fail_year", test_suggest_fail_year)
|
||
_run("task_generator/suggest_renew", test_suggest_renew_upward)
|
||
_run("task_generator/suggest_observe", test_suggest_observe_default)
|
||
_run("scanner/no_alerts", test_scanner_no_alerts)
|
||
_run("scanner/three_levels", test_scanner_three_levels)
|
||
_run("scanner/skips_non_active", test_scanner_skips_non_active_contracts)
|
||
_run("html_renderer/basic", test_html_render_basic)
|
||
_run("html_renderer/unknown_task_raises", test_html_render_unknown_task_raises)
|
||
|
||
total = len(_RESULTS)
|
||
passed = sum(1 for _, ok, _ in _RESULTS if ok)
|
||
failed = total - passed
|
||
|
||
print()
|
||
print(f"=== Summary: {passed}/{total} passed, {failed} failed ===", flush=True)
|
||
if failed:
|
||
print("\nFailures:")
|
||
for name, ok, msg in _RESULTS:
|
||
if not ok:
|
||
print(f" - {name}: {msg}")
|
||
return 1
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|