#!/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 json
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}"


def test_data_management_actions() -> tuple[bool, str]:
    """数据管理动作必须绑定业务表，并映射到实际 CLI 命令。"""
    actions_path = _REPO_ROOT / "assets" / "actions.json"
    actions = json.loads(actions_path.read_text(encoding="utf-8"))["actions"]
    by_id = {action["id"]: action for action in actions}

    expected = {
        "scan-expiring-contracts": "toolbar",
        "collect-from-platform": "toolbar",
        "generate-renewal-task": "row",
        "dispatch-task-notice": "batch",
        "mark-contract-renewed": "row",
    }
    for action_id, placement in expected.items():
        action = by_id.get(action_id)
        if not action:
            return False, f"缺少数据管理动作 {action_id}"
        if placement not in action.get("placements", []):
            return False, f"{action_id} 缺少 {placement} placement"
        if "合同台账" not in action.get("bind", {}).get("tables", []):
            return False, f"{action_id} 未绑定合同台账"
        if action.get("entrypoint", {}).get("command") != action_id:
            return False, f"{action_id} CLI command 映射错误"
    return True, "5 个数据管理动作均绑定合同台账并映射到 CLI"


def test_collect_platform_defaults_to_browser() -> tuple[bool, str]:
    """平台采集默认显示浏览器；--headless 仅用于已登录自动化。"""
    from cli.app import build_parser

    parser = build_parser()
    default_args = parser.parse_args(["collect-from-platform"])
    if default_args.headless:
        return False, "collect-from-platform 默认不应使用无头模式"
    headless_args = parser.parse_args(["collect-from-platform", "--headless"])
    if not headless_args.headless:
        return False, "--headless 未启用无头模式"
    legacy_args = parser.parse_args(["collect-from-platform", "--show-browser"])
    if legacy_args.headless:
        return False, "--show-browser 兼容别名未保留浏览器模式"
    return True, "默认浏览器模式，--headless 和 --show-browser 行为正确"


# ---------- 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_equal_scores_not_renew() -> tuple[bool, str]:
    """持平不算逐年上升；应回落到 observe"""
    from service.task_generator import suggest
    rows = [
        _fake_perf(2023, 80, moral=85),
        _fake_perf(2024, 80, moral=85),
        _fake_perf(2025, 90, moral=85),
    ]
    s, r = suggest(rows)
    if s != "observe":
        return False, f"suggest(80,80,90) = {s}, want observe"
    return True, "观察：持平不等于逐年上升"


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，实际未抛"


# ---------- v1.1.6：service.platform_collector 纯函数 ----------

def test_gen_contract_id_emp_id() -> tuple[bool, str]:
    """_gen_contract_id 工号派生（v1.1.1 起用工号作主键）"""
    from service import platform_collector as pc

    cases = [
        (("T20190033", "2026-04-15"), "RPA-T20190033-20260415"),
        (("T20200055", "2026-06-30"), "RPA-T20200055-20260630"),
        (("T20180021", "2026-08-31"), "RPA-T20180021-20260831"),
    ]
    for (emp_id, end_date), expected in cases:
        got = pc._gen_contract_id(emp_id, end_date)
        if got != expected:
            return False, f"({emp_id!r}, {end_date!r}) → {got!r}，want {expected!r}"
    return True, f"3 工号派生正确"


def test_gen_contract_id_special_chars() -> tuple[bool, str]:
    """_gen_contract_id 工号含特殊字符时正确清洗（保留前 10 字符）"""
    from service import platform_collector as pc

    # 工号含非 \w 字符 → 清洗；超过 10 字符截断
    got = pc._gen_contract_id("T-2019.0033", "2026-04-15")
    # 清洗后保留 \w，"T-2019.0033" → "T20190033"（8 字符，未截断）
    if "RPA-" not in got or "20260415" not in got:
        return False, f"清洗后格式错：{got!r}"
    # 长工号 → 截断到 10 字符
    got2 = pc._gen_contract_id("ABCDEFGHIJKLMNO", "2026-01-01")
    # "ABCDEFGHIJ"（10 字符）
    if not got2.startswith("RPA-ABCDEFGHIJ-"):
        return False, f"长工号未截断到 10 字符：{got2!r}"
    return True, "清洗 + 截断 OK"


def test_normalize_date_iso() -> tuple[bool, str]:
    """_normalize_date ISO 日期直通"""
    from service import platform_collector as pc

    cases = [
        ("2026-04-15", "2026-04-15"),
        ("2024-01-01", "2024-01-01"),
    ]
    for inp, want in cases:
        got = pc._normalize_date(inp)
        if got != want:
            return False, f"{inp!r} → {got!r}，want {want!r}"
    return True, "ISO 日期直通"


def test_normalize_date_fallback() -> tuple[bool, str]:
    """_normalize_date 非标准格式截前 10 字符"""
    from service import platform_collector as pc

    # 非标准但带 ISO 前缀
    got = pc._normalize_date("2026-04-15T00:00:00")
    if got != "2026-04-15":
        return False, f"ISO 前缀应保留前 10 字符 → {got!r}"
    # 完全非日期 → 截前 10 字符
    got2 = pc._normalize_date("20260415X")
    if got2 != "20260415X":  # 10 字符
        return False, f"非日期应截前 10 字符 → {got2!r}"
    # 带空白
    got3 = pc._normalize_date("  2026-04-15  ")
    if got3 != "2026-04-15":
        return False, f"带空白应去除 → {got3!r}"
    return True, "非标准格式 fallback OK"


def test_parse_days_positive() -> tuple[bool, str]:
    """_parse_days 正天数"""
    from service import platform_collector as pc

    cases = [
        ("30", 30),
        ("90", 90),
        ("还剩 45 天", 45),
        ("0", 0),
    ]
    for inp, want in cases:
        got = pc._parse_days(inp)
        if got != want:
            return False, f"{inp!r} → {got!r}，want {want!r}"
    return True, "正天数解析 OK"


def test_parse_days_negative() -> tuple[bool, str]:
    """_parse_days 负数 / 兜底"""
    from service import platform_collector as pc

    # 含负号
    if pc._parse_days("-15") != -15:
        return False, "负天数未解析"
    # 已过期（带 "已过期" 文本）
    if pc._parse_days("已过期 7 天") != 7:
        return False, "已过期文本未解析数字部分"
    # 无法解析
    if pc._parse_days("") is not None:
        return False, "空字符串应返回 None"
    if pc._parse_days("无数字") is not None:
        return False, "纯文本应返回 None"
    return True, "负数 + 兜底 OK"


def test_parse_alert_red_orange_yellow_unknown() -> tuple[bool, str]:
    """_parse_alert 红/橙/黄/未知四级映射"""
    from service import platform_collector as pc

    cases = [
        ("红色预警", "red"),
        ("红牌", "red"),
        ("橙色预警", "orange"),
        ("黄灯", "yellow"),
        ("黄色预警", "yellow"),
        ("无", "unknown"),
        ("", "unknown"),
        ("blue", "unknown"),
    ]
    for inp, want in cases:
        got = pc._parse_alert(inp)
        if got != want:
            return False, f"{inp!r} → {got!r}，want {want!r}"
    return True, "四级映射 OK"


# ---------- 入口 ----------

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("assets/data_management_actions",       test_data_management_actions)
    _run("cli/collect_platform_browser_default", test_collect_platform_defaults_to_browser)
    _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_equal_scores",   test_suggest_observe_equal_scores_not_renew)
    _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)
    _run("platform_collector/gen_contract_id_emp_id",  test_gen_contract_id_emp_id)
    _run("platform_collector/gen_contract_id_special", test_gen_contract_id_special_chars)
    _run("platform_collector/normalize_date_iso",      test_normalize_date_iso)
    _run("platform_collector/normalize_date_fallback", test_normalize_date_fallback)
    _run("platform_collector/parse_days_positive",    test_parse_days_positive)
    _run("platform_collector/parse_days_negative",    test_parse_days_negative)
    _run("platform_collector/parse_alert_levels",     test_parse_alert_red_orange_yellow_unknown)

    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())