Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 1707b453c8 | |||
| 01ba253bfe | |||
| 4af2b4fbf3 |
21
CHANGELOG.md
21
CHANGELOG.md
@@ -2,6 +2,27 @@
|
||||
|
||||
本文件对应技能市场 **「更新日志」** Tab。
|
||||
|
||||
## 1.1.3
|
||||
|
||||
- **版本对齐修复**:v1.1.2 tag(commit `4af2b4f`)发布时 `SKILL.md version` / `scripts/util/constants.py SKILL_VERSION` 未同步升级(仍为 1.1.1),本次发布将 manifest 升至 1.1.3 并与代码现状对齐
|
||||
- **内容同 v1.1.2**:sandbox 卡片采集功能(详见 1.1.2 条目)
|
||||
|
||||
> **说明**:v1.1.2 与 v1.1.3 实质内容相同(均含 sandbox 卡片采集)。v1.1.2 tag 视为已发布但 manifest 错位的历史遗留;v1.1.3 为首次 manifest 版本号与代码对齐的 release tag。
|
||||
|
||||
## 1.1.2
|
||||
|
||||
- **sandbox 卡片采集**:表格 10 列(工号/姓名/单位/岗位/合同类型/起止日期/剩余天数/预警/建议/材料包按钮),进入 hr/contracts 页面后 `page.reload()` 刷新触发 mock 数据加载
|
||||
- **「续聘材料包」按钮行为**:不触发浏览器 download event,而是展开页面下方「续聘准备任务书」+「单位清单」卡片(含 checklist + 近三年考核表)。RPA 改为抓卡片 inner_text 存 `materials_content`(parse_method='sandbox_card')
|
||||
- **卡片文本解析进 performance_records**:新增 `repository.parse_sandbox_card_performance`(双模式正则:行格式 `2024\t称职` + 叙述式 `2024年:等次「称职」`)+ `repository.upsert_performance_record`(等次→分数:优秀=90 称职=75 基本称职=60 不称职=30)
|
||||
- **contract_id 改用工号**(稳定,不随姓名变化):`_gen_contract_id(emp_id, end_date)`
|
||||
- **实测**(sandbox 3 合同):
|
||||
|
||||
| contract_id | 教师 | 解析 perf 条数 | AI 决策 |
|
||||
|---|---|---|---|
|
||||
| RPA-T20190033-20260415 | 钱实验 | 4 | observe |
|
||||
| RPA-T20200055-20260630 | 孙助理 | 6 | observe |
|
||||
| RPA-T20180021-20260831 | 王明远 | 6 | observe |
|
||||
|
||||
## 1.1.1
|
||||
|
||||
- **AI 续聘建议升级**:本地规则版 → LLM 优先 + 规则降级
|
||||
|
||||
2
SKILL.md
2
SKILL.md
@@ -1,7 +1,7 @@
|
||||
---
|
||||
name: 合同到期智能预警与续聘辅助
|
||||
description: "每日凌晨扫描全校教职工合同电子台账,对距离到期日 90/60/30 天的合同分级预警,并自动生成《续聘准备任务书》和《本月合同到期人员清单》。"
|
||||
version: 1.1.1
|
||||
version: 1.1.3
|
||||
author: 深圳匠厂科技有限公司
|
||||
metadata:
|
||||
openclaw:
|
||||
|
||||
@@ -59,6 +59,82 @@ def list_performance_by_contract(
|
||||
).fetchall()
|
||||
|
||||
|
||||
# 等次 → 分数映射(v1.1.1 sandbox card 解析)
|
||||
_GRADE_TO_SCORE = {
|
||||
"优秀": 90.0,
|
||||
"称职": 75.0,
|
||||
"基本称职": 60.0,
|
||||
"不称职": 30.0,
|
||||
"不合格": 30.0,
|
||||
}
|
||||
|
||||
|
||||
def upsert_performance_record(
|
||||
conn: sqlite3.Connection,
|
||||
contract_id: str,
|
||||
year: int,
|
||||
grade: str,
|
||||
) -> bool:
|
||||
"""upsert 一条业绩记录(用 grade 映射分数)。
|
||||
|
||||
sandbox card 文本不含真分数时使用。返回是否实际写入。
|
||||
"""
|
||||
score = _GRADE_TO_SCORE.get(grade)
|
||||
if score is None:
|
||||
return False
|
||||
record_id = f"P-{contract_id}-{year}"
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO performance_records
|
||||
(record_id, contract_id, year, performance_score,
|
||||
moral_teaching_score, research_score, teaching_score)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(record_id) DO UPDATE SET
|
||||
performance_score = excluded.performance_score
|
||||
""",
|
||||
(record_id, contract_id, year, score, score, score, score),
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
def parse_sandbox_card_performance(
|
||||
conn: sqlite3.Connection,
|
||||
contract_id: str,
|
||||
card_text: str,
|
||||
) -> int:
|
||||
"""从 sandbox 卡片文本抽 (year, 等次) → upsert 到 performance_records。
|
||||
|
||||
文本片段形如:
|
||||
year result
|
||||
2024 称职
|
||||
2025 优秀
|
||||
2024年:等次「称职」。
|
||||
返回写入条数。
|
||||
"""
|
||||
import re
|
||||
if not card_text:
|
||||
return 0
|
||||
|
||||
written = 0
|
||||
# 模式 1:行格式 "2024\t称职"
|
||||
for m in re.finditer(r"^(\d{4})\s*[ \t]+(优秀|称职|基本称职|不称职|不合格)\s*$",
|
||||
card_text, re.MULTILINE):
|
||||
year = int(m.group(1))
|
||||
grade = m.group(2)
|
||||
if upsert_performance_record(conn, contract_id, year, grade):
|
||||
written += 1
|
||||
|
||||
# 模式 2:叙述式 "2024年:等次「称职」"
|
||||
for m in re.finditer(r"(\d{4})\s*年[::]\s*等次[「\"']\s*(优秀|称职|基本称职|不称职|不合格)\s*[」\"']",
|
||||
card_text):
|
||||
year = int(m.group(1))
|
||||
grade = m.group(2)
|
||||
if upsert_performance_record(conn, contract_id, year, grade):
|
||||
written += 1
|
||||
|
||||
return written
|
||||
|
||||
|
||||
# ---------- alerts ----------
|
||||
|
||||
def insert_alert(
|
||||
|
||||
@@ -243,6 +243,17 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
|
||||
|
||||
logger.info("platform_collector: 当前 URL=%s", page.url)
|
||||
|
||||
# 5.5. 进入页面后刷新(SPA 进入新路由时 mock 数据可能未加载,需刷新触发)
|
||||
try:
|
||||
logger.info("platform_collector: 进入 hr/contracts 后刷新页面")
|
||||
page.reload(wait_until="networkidle", timeout=20_000)
|
||||
except Exception as e:
|
||||
logger.warning("刷新 networkidle 失败: %s;尝试 domcontentloaded", e)
|
||||
try:
|
||||
page.reload(wait_until="domcontentloaded", timeout=20_000)
|
||||
except Exception as e2:
|
||||
logger.warning("刷新 domcontentloaded 也失败: %s", e2)
|
||||
|
||||
# 7. dump 当前 hr/contracts 页面(用于调试)
|
||||
try:
|
||||
html = page.content()
|
||||
@@ -284,6 +295,23 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
|
||||
logger.warning("dump 失败: %s", e)
|
||||
return _error_result("table_not_found_after_40s", 0)
|
||||
|
||||
# 表格已找到,等 tbody 完全填充 + dump 一份完整 DOM(用于诊断按钮 selector)
|
||||
try:
|
||||
page.wait_for_function(
|
||||
"document.querySelectorAll('table tbody tr').length > 0",
|
||||
timeout=10_000,
|
||||
)
|
||||
page.wait_for_timeout(1500)
|
||||
html = page.content()
|
||||
full_dump = download_dir.parent / "hr_contracts_full_dump.html"
|
||||
full_dump.write_text(html, encoding="utf-8")
|
||||
logger.info(
|
||||
"platform_collector: 完整 hr/contracts dump → %s (%d 字节)",
|
||||
full_dump, len(html),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning("等完整渲染超时: %s", e)
|
||||
|
||||
rows = page.eval_on_selector_all(
|
||||
"table tbody tr",
|
||||
"rs => rs.map(r => Array.from(r.querySelectorAll('td')).map(td => td.innerText))",
|
||||
@@ -302,21 +330,33 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
|
||||
except Exception:
|
||||
mat_btn_count = 0
|
||||
|
||||
# 抓「材料包」card 区段的 DOM(用于后续解析每行 click 后的卡片文本)
|
||||
# sandbox 行为:click 按钮后会在表格下方追加「续聘准备任务书」+「单位清单」卡片
|
||||
|
||||
for i, r in enumerate(rows):
|
||||
if limit and i >= limit:
|
||||
break
|
||||
if len(r) < 6:
|
||||
logger.warning("行 %d 列数 < 6,跳过: %s", i, r)
|
||||
if len(r) < 7:
|
||||
logger.warning("行 %d 列数 < 7(新列序工号+姓名+单位+岗位+合同类型+起止+剩余+预警+建议),跳过: %s", i, r)
|
||||
continue
|
||||
|
||||
teacher = r[0].strip()
|
||||
unit = r[1].strip()
|
||||
contract_end = r[2].strip()
|
||||
days_raw = r[3].strip()
|
||||
alert_label = r[4].strip()
|
||||
suggestion_hint = r[5].strip()
|
||||
# 新列序(sandbox 加了工号 / 岗位 / 合同类型 / 起止日期):
|
||||
# 0: 工号 1: 姓名 2: 单位 3: 岗位 4: 合同类型
|
||||
# 5: 起止日期 (含 ~ ) 6: 剩余天数 7: 预警 8: 建议(含原因)
|
||||
emp_id = r[0].strip()
|
||||
teacher = r[1].strip()
|
||||
unit = r[2].strip()
|
||||
position = r[3].strip()
|
||||
contract_type = r[4].strip()
|
||||
period = r[5].strip()
|
||||
# 起止日期形如 "2024-03-01 ~ 2026-04-15" → 取 ~ 后
|
||||
contract_end = period.split("~")[-1].strip() if "~" in period else period
|
||||
days_raw = r[6].strip()
|
||||
alert_label = r[7].strip()
|
||||
suggestion_hint = r[8].strip() if len(r) > 8 else ""
|
||||
|
||||
contract_id = _gen_contract_id(teacher, contract_end)
|
||||
# 用工号生成 contract_id(稳定,不会随姓名变化)
|
||||
contract_id = _gen_contract_id(emp_id, contract_end)
|
||||
contract = {
|
||||
"contract_id": contract_id,
|
||||
"teacher_name": teacher,
|
||||
@@ -328,32 +368,50 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
|
||||
"materials": [],
|
||||
}
|
||||
|
||||
# 只有在按钮存在时才尝试下载(sandbox mock 当前未部署此按钮)
|
||||
# 点「续聘材料包」按钮 → 等卡片展开 → 抓卡片文本(含 checklist)
|
||||
if mat_btn_count > 0:
|
||||
try:
|
||||
btn = page.locator("table tbody tr").nth(i).locator(
|
||||
"button:has-text('续聘材料包')"
|
||||
)
|
||||
with page.expect_download(timeout=10_000) as dl_info:
|
||||
btn.click()
|
||||
dl = dl_info.value
|
||||
fname = dl.suggested_filename or "materials.zip"
|
||||
dest = download_dir / f"{contract_id}_{fname}"
|
||||
dl.save_as(str(dest))
|
||||
contract["materials"].append(str(dest))
|
||||
downloaded += 1
|
||||
btn.click()
|
||||
# 等卡片出现(页面下方追加 h2「续聘准备任务书」/「单位清单」)
|
||||
page.wait_for_timeout(1500)
|
||||
# 抓表格下方所有 card 文本
|
||||
cards_text = page.locator(
|
||||
"main .grid > div, main > div > div > div"
|
||||
).all_inner_texts()
|
||||
materials_text = "\n\n".join(
|
||||
t.strip() for t in cards_text if t.strip()
|
||||
)
|
||||
if materials_text:
|
||||
# 存到 materials_content(sandbox mock 卡片文本作为材料包内容)
|
||||
contract["materials"].append({
|
||||
"kind": "sandbox_card",
|
||||
"file_name": f"{contract_id}_materials.txt",
|
||||
"text": materials_text,
|
||||
})
|
||||
downloaded += 1
|
||||
logger.info(
|
||||
"platform_collector: 抓到卡片文本 %d 字 (%s)",
|
||||
len(materials_text), contract_id,
|
||||
)
|
||||
# dump 一次(i==0)方便调试
|
||||
if i == 0:
|
||||
html_after = page.content()
|
||||
click_dump = download_dir.parent / "after_click_dump.html"
|
||||
click_dump.write_text(html_after, encoding="utf-8")
|
||||
logger.info(
|
||||
"platform_collector: 首行 click 后 dump → %s (%d 字节)",
|
||||
click_dump, len(html_after),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning("材料下载失败 %s: %s", contract_id, e)
|
||||
logger.warning("抓卡片失败 %s: %s", contract_id, e)
|
||||
else:
|
||||
logger.info(
|
||||
"platform_collector: sandbox 未部署「续聘材料包」按钮,"
|
||||
"跳过材料下载(仅采集表格列)",
|
||||
)
|
||||
# 仅警告一次
|
||||
if i == 0:
|
||||
logger.warning(
|
||||
"platform_collector: 提示:材料包缺失时,"
|
||||
"parse-materials CLI 将标记为「无材料」"
|
||||
logger.info(
|
||||
"platform_collector: sandbox 未部署「续聘材料包」按钮,"
|
||||
"跳过材料抓取(仅采集表格列)",
|
||||
)
|
||||
|
||||
contracts.append(contract)
|
||||
@@ -413,27 +471,56 @@ def _persist_contracts(contracts: list[dict]) -> int:
|
||||
),
|
||||
)
|
||||
|
||||
# 写 materials_content(占位:file_path 写入,extracted_text 待 parse-materials)
|
||||
# 写 materials_content(sandbox card 文本 + parse-materials 解析后文件路径)
|
||||
from db.migrations import ensure_materials_content_schema
|
||||
|
||||
ensure_materials_content_schema(conn)
|
||||
|
||||
for fp in c["materials"]:
|
||||
fname = Path(fp).name
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT OR REPLACE INTO materials_content
|
||||
(contract_id, file_name, file_path, extracted_text,
|
||||
parse_method, parsed_at)
|
||||
VALUES (?, ?, ?, NULL, 'pending', ?)
|
||||
""",
|
||||
(
|
||||
c["contract_id"],
|
||||
fname,
|
||||
fp,
|
||||
datetime.now().isoformat(timespec="seconds"),
|
||||
),
|
||||
)
|
||||
for m in c["materials"]:
|
||||
if isinstance(m, dict) and m.get("kind") == "sandbox_card":
|
||||
# sandbox mock 卡片文本,直接写入 extracted_text
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT OR REPLACE INTO materials_content
|
||||
(contract_id, file_name, file_path, extracted_text,
|
||||
parse_method, parsed_at)
|
||||
VALUES (?, ?, ?, ?, 'sandbox_card', ?)
|
||||
""",
|
||||
(
|
||||
c["contract_id"],
|
||||
m["file_name"],
|
||||
f"sandbox://hr/contracts/{c['contract_id']}",
|
||||
m["text"],
|
||||
datetime.now().isoformat(timespec="seconds"),
|
||||
),
|
||||
)
|
||||
# 同步解析 → 抽 year + 等次 → 写入 performance_records
|
||||
perf_n = repo.parse_sandbox_card_performance(
|
||||
conn, c["contract_id"], m["text"],
|
||||
)
|
||||
if perf_n > 0:
|
||||
logger.info(
|
||||
"platform_collector: %s 解析卡片 → performance %d 条",
|
||||
c["contract_id"], perf_n,
|
||||
)
|
||||
else:
|
||||
# 本地文件路径(parse-materials 解析后写入)
|
||||
fp = m if isinstance(m, str) else m.get("file_path")
|
||||
fname = Path(fp).name if fp else "unknown"
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT OR REPLACE INTO materials_content
|
||||
(contract_id, file_name, file_path, extracted_text,
|
||||
parse_method, parsed_at)
|
||||
VALUES (?, ?, ?, NULL, 'pending', ?)
|
||||
""",
|
||||
(
|
||||
c["contract_id"],
|
||||
fname,
|
||||
fp,
|
||||
datetime.now().isoformat(timespec="seconds"),
|
||||
),
|
||||
)
|
||||
return len(contracts)
|
||||
finally:
|
||||
conn.close()
|
||||
@@ -450,11 +537,14 @@ def _error_result(reason: str, fetched: int) -> dict:
|
||||
}
|
||||
|
||||
|
||||
def _gen_contract_id(teacher: str, end_date: str) -> str:
|
||||
"""从 (姓名, 到期日) 派 contract_id,确保幂等。"""
|
||||
safe_teacher = re.sub(r"[^\w]", "", teacher)[:8] or "T"
|
||||
def _gen_contract_id(emp_id: str, end_date: str) -> str:
|
||||
"""从 (工号, 到期日) 派 contract_id,确保幂等。
|
||||
|
||||
sandbox v1.1.1 起改用「工号」作主键(之前用姓名)。
|
||||
"""
|
||||
safe_emp = re.sub(r"[^\w]", "", emp_id)[:10] or "T"
|
||||
safe_end = end_date.replace("-", "")[:8]
|
||||
return f"RPA-{safe_teacher}-{safe_end}"
|
||||
return f"RPA-{safe_emp}-{safe_end}"
|
||||
|
||||
|
||||
def _normalize_date(s: str) -> str:
|
||||
|
||||
@@ -6,7 +6,7 @@
|
||||
"""
|
||||
|
||||
SKILL_SLUG = "alert-contract-expiring-hr"
|
||||
SKILL_VERSION = "1.1.1"
|
||||
SKILL_VERSION = "1.1.3"
|
||||
LOG_LOGGER_NAME = "openclaw.skill.alert_contract_expiring_hr"
|
||||
PLATFORM_KIT_MIN_VERSION = "1.2.11"
|
||||
|
||||
|
||||
612
tests/run_tests.py
Normal file
612
tests/run_tests.py
Normal file
@@ -0,0 +1,612 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user