Compare commits

...

12 Commits

Author SHA1 Message Date
689810eaeb chore: auto release commit (2026-09-15 02:56:48)
Some checks failed
技能自动化发布 / release (push) Failing after 4s
2026-09-15 02:56:49 +08:00
7584e69714 chore: auto release commit (2026-09-15 01:39:08)
Some checks failed
技能自动化发布 / release (push) Failing after 6s
2026-09-15 01:39:08 +08:00
d3fee3f46c chore: auto release commit (2026-09-15 01:30:18)
Some checks failed
技能自动化发布 / release (push) Failing after 4s
2026-09-15 01:30:18 +08:00
ebea77b5f2 chore: auto release commit (2026-09-15 01:27:04)
Some checks failed
技能自动化发布 / release (push) Failing after 4s
2026-09-15 01:27:05 +08:00
430274dbf9 chore: auto release commit (2026-09-15 01:23:04)
Some checks failed
技能自动化发布 / release (push) Failing after 4s
2026-09-15 01:23:04 +08:00
0157fde0f1 chore: auto release commit (2026-09-15 01:20:35)
Some checks failed
技能自动化发布 / release (push) Failing after 7s
2026-09-15 01:20:35 +08:00
aa9bf262fe chore: auto release commit (2026-09-15 01:15:35)
Some checks failed
技能自动化发布 / release (push) Failing after 5s
2026-09-15 01:15:35 +08:00
52831b80e0 chore: auto release commit (2026-09-15 01:10:24)
Some checks failed
技能自动化发布 / release (push) Failing after 6s
2026-09-15 01:10:24 +08:00
17a7a54349 chore: auto release commit (2026-09-15 01:07:11)
Some checks failed
技能自动化发布 / release (push) Failing after 4s
2026-09-15 01:07:11 +08:00
c21339ae44 chore: auto release commit (2026-09-15 01:02:50)
All checks were successful
技能自动化发布 / release (push) Successful in 6s
2026-09-15 01:02:50 +08:00
9878f570d7 chore: auto release commit (2026-09-15 00:57:09)
All checks were successful
技能自动化发布 / release (push) Successful in 5s
2026-09-15 00:57:09 +08:00
44088336b4 chore: auto release commit (2026-09-15 00:49:52)
All checks were successful
技能自动化发布 / release (push) Successful in 4s
2026-09-15 00:49:52 +08:00
10 changed files with 1907 additions and 101 deletions

View File

@@ -86,3 +86,20 @@ INPUT_DIR=
# 失败最大重试次数(默认 3)
DOWNLOAD_QUEUE_MAX_ATTEMPTS=
# ----- v1.1.31:SmartCampus HR 平台登录凭证(不硬编码到代码)-----
# 根域登录页(https://sandbox.jc2009.com/)和 SPA 二级登录页
# (/industries/campus/smartcampus/login)共用同一组凭证。
# 留空 → 脚本不自动 fill,等用户在浏览器手工输入。
SMARTCAMPUS_LOGIN_USER=
SMARTCAMPUS_LOGIN_PASSWORD=
# ----- v1.1.32:目标行业 tile(DEV MODE 面板)-----
# SPA 登录页右侧 DEV MODE 面板包含多个行业 tile(教师通用 / 人事处 / 财务处 / 科研处),
# 脚本自动点对应 tile 的 Sign in 把 demo 凭据搬到主表单。可选:
# hr → 人事处(默认)
# finance → 财务处
# research → 科研处
# demo → 教师(通用)
# 留空 → 默认 hr,且会探测主表单已预填的 user 值自动选择。
SMARTCAMPUS_LOGIN_INDUSTRY=

View File

@@ -2,6 +2,101 @@
本文件对应技能市场 **「更新日志」** Tab。
## 1.1.30
- **修复 jiangchang 客户端 arguments 丢失导致 argparse SystemExit 问题**:
- `generate-renewal-task --contract-id` / `dispatch-task-notice --unit-id` / `mark-contract-renewed --contract-id --material-id` / `render-html --task-id` 4 个 sync action 的 `--xxx` 参数从 `required=True` 改为 `default=""`
- CLI handler 检测到空值返回 JSON `{"error": "missing_arg", "detail": ..., "hint": ...}` + exit=1(不再 argparse 报错退出,宿主能拿到结构化错误)
- `assets/actions.json` 4 个 sync action 追加 `v1_1_30_notes` 字段说明新行为
- 1 个新测试 `sync_actions_tolerate_missing_args_v1_30` 覆盖 4 个 action 的 missing_arg JSON
## 1.1.29
- **修复「老 DB 没 download_tasks 表导致入队失败」问题**:
- 用户本地 `data/alert-contract-expiring-hr.db` 是 v1.1.17 之前 init 的(只有前 6 张表),没跑过 v1.1.18 的 init-db → 没 `download_tasks` 表
- `_enqueue_download_tasks_safe` 在 `owns_conn=True` 路径下,先 idempotent `CREATE TABLE IF NOT EXISTS download_tasks ...` + 2 索引,再 INSERT
- 解决 `OperationalError: no such table: download_tasks` + 下载队列 `failed=3`
- 新增测试 `enqueue_safe_auto_create_table_v1_29` 模拟「老 DB 没建过 download_tasks」场景(用 monkeypatch `rp.get_db_path` + `db.connection.get_db_path`),合计 67/67
## 1.1.28
- **彻底修复「SPA 二级登录页卡顿」**:
- v1.1.27 失败根因:URL 已稳定在 `/login` 时,React SPA input 可能还没渲染(hydrate 慢),`_fill_demo_on_spa_login` 找不到 input → False → 卡 180 秒等用户输入
- 修复:步骤 4.5 先 `wait_for_selector(input, timeout=10s, state="attached")` 等 input 真正 hydrate + 0.5s React 稳定,再 fill demo/123456 + 点登录
- demo fill 后如果 URL 仍卡在 login 页(demo 失效)→ headed 模式 fallback 等用户手工登录,headless 模式抛 `LoginTimeoutError`
- 不再依赖「input 是否预填」—— sandbox SPA 二级登录页 input 默认不预填,必须强制覆盖 fill
- 合计 66/66
## 1.1.27
- **修复「到 SPA 二级登录页卡 30 秒」问题**:
- 之前步骤 4 等 URL 稳定到 `/app/` 或 `/select` 30 秒,但 sandbox 登录后会把人踢回 `/industries/campus/smartcampus/login`(URL 含 `/login`),等满 30 秒才发现要进 SPA 登录兜底
- 步骤 4 改成等 URL 跳到 `/app/` 或 `/select` 或 `/login` 或 `/signin`(任一即可,**最多 30 秒**),URL 含 `/login` 立即退出等待
- **新增 `_fill_demo_on_spa_login(page)`**:SPA 二级登录页 input 默认不预填(与根域不同),主动 `fill('demo')` + `fill('123456')` + 点登录按钮(Sign in / 登录 / Login / submit 任一)
- 步骤 4.5 改为:URL 落在 SPA 登录页 → 自动 `_fill_demo_on_spa_login` → 不再卡 180 秒等用户输入
- 新增 3 个测试覆盖 `_fill_demo_on_spa_login`(basic / 中文按钮 / 无 input),合计 66/66
## 1.1.26
- **按用户最新要求改回「自动点登录」行为**:「到了 /industries/campus/smartcampus/login 整个页面,有账号和密码可用直接点登录跳转」
- `_wait_for_login_inputs_filled` 改回 v1.1.21 行为:input 都有值 → 自动点 Sign in / 登录按钮 → return True
- 与 `_try_auto_sign_in` 行为统一:登录页 = 「有数据自动点」/ 「无数据等用户输入」
- 测试替换:`wait_login_inputs_filled_no_auto_click_v1_24` → `wait_login_inputs_filled_auto_click_v1_26`,合计 63/63
## 1.1.25
- **按用户要求修正登录页等待逻辑**:
- 有数据(账号密码已存在)→ 自动点登录按钮(`_try_auto_sign_in`)
- **无数据 → 等待用户输入完成(最多 180 秒)→ 等 URL 跳离登录页**(按钮交给用户自己点,脚本不抢按钮)
- `_wait_for_login_inputs_filled` 重写为「**等待模式**」:不再自动点 Sign in / 登录按钮;只轮询 input 值 + URL,每 15 秒提示「请点登录按钮」或「等待输入」
- 根域登录 + SPA 二级登录页统一使用同一套行为
- 新增测试 `wait_login_inputs_filled_no_auto_click_v1_24` 验证不抢按钮,合计 63/63
## 1.1.24
- **修复 `/industries/campus/smartcampus/login` 等待过久问题**:sandbox 在用户首次登录后可能把人踢回 SPA 二级登录页(按钮中文「登录」),之前根域自动登录代码不识别这个 URL + 中文按钮,导致脚本卡死等用户输入
- `_try_auto_sign_in` 扩展为通用版本:支持 `#jc-user` / `input[type='tel']` / `input[autocomplete='username']` 多种 user input selector;按钮文本按优先级匹配 `Sign in` > `登录` > `Login` > `signin` > `submit`
- `collect_all` 登录后新增步骤 4.5:等 URL 稳定到 `/app/` 或 `/select` 后,若发现 URL 仍含 `/login` / `/signin`(落在 SPA 二级登录页),自动重走一次 `_try_auto_sign_in` → 再等 URL 跳到 `/app/` 或 `/select`
- 新增 4 个测试覆盖 `_try_auto_sign_in`(SPA 中文按钮 / 根域英文按钮 / input 空 / selector 超时),合计 58/58
## 1.1.23
- **修复登录后跳 /app/home 的流程**:
- `_wait_for_url_stable` 支持 tuple targets(任一匹配即稳定)
- 登录后等 URL 稳定到 `/app/` 或 `/select`(sandbox 中间页)
- 显式 `goto SPA_HOME_URL` (/app/home) → 等 sidebar link 出现 → click `<a href="/industries/campus/smartcampus/app/hr/contracts">合同到期续聘</a>`
- 删掉重复的 # 4 块(之前手抖复制粘贴的 bug)
## 1.1.22
- **登录流程接入 sandbox 自动登录**:root 登录页(https://sandbox.jc2009.com/)的 #jc-user / #jc-pass 默认有 demo/123456 预填
- 新增 `_try_auto_sign_in(page, wait_input_sec=2)`:等 input 出现 2 秒 → 检查是否有值 → 有则自动点 Sign in
- 新增 `_wait_for_login_inputs_filled(page, wait_sec)`:input 为空时等用户输入 → 两个 input 都有值后自动点 Sign in
- 已登录态(profile cookie 复用)直接跳过
- 输入超时 fallback 到 `_wait_for_manual_login`(多信号判定)
## 1.1.21
- **修复登录完成检测盲区**:`_wait_for_manual_login` 现在监听 3 个信号判断「登录已完成」,任一触发即继续
1. URL 含 `/app/` 且稳定 2 秒(v1.1.20 旧逻辑)
2. URL 已离开 `/login` / `/signin` / `/auth/` 等登录路径
3. 页面出现 sidebar 元素 `a[href$='/hr/contracts']`
- 解决「用户已经登录但脚本还在傻等」的回归:之前只判 URL 变化,cookie 已生效复用 profile 场景或 SPA 不跳 URL 的场景会卡 180 秒
## 1.1.20
- **登录流程重写**:第一跳改为 `https://sandbox.jc2009.com/`(根域,用户最初看到的页面),**不做任何操作**等 URL 稳定
- 已登录态:根域自动跳到 `/app/home` → 直接走侧边栏到 `hr/contracts` 采集
- 未登录态:**纯等用户在浏览器手工输入账号密码**,不再自动 fill demo/123456,不再 retry 强制 reload 打断用户输入
- 新增 `_wait_for_url_stable(page, target_substr, max_wait_sec)`:等 URL 真正稳定 2 秒后才返回,避免 sandbox 中间页误导
- `_wait_for_manual_login` 重写:每 5 秒轮询 URL(之前 2 秒太密),每 15 秒打印一次剩余时间提示,**等待期间不触发任何 goto / reload / click**
- 新增 4 个 login flow 测试(`wait_url_stable_immediate` / `wait_url_stable_transition` / `wait_url_stable_timeout` / `wait_manual_login_no_goto`),合计 58/58
## 1.1.19
- **修复 1.1.18 引入的采集回归**:把 `_persist_contracts` 里的入队调用移到独立事务 `_enqueue_download_tasks_safe`,FK 失败不再回滚 `contracts` / `materials_content` 落库;返回 JSON 新增 `download_queue` 字段方便 UI 直接看到入队结果
- 新增 4 个测试覆盖 `_enqueue_download_tasks_safe`(basic / idempotent / fk_failure / empty_list),合计 54/54
## 1.1.18
- **一键式"运维 + 材料管理"全打通**

View File

@@ -1,7 +1,7 @@
---
name: 合同到期智能预警与续聘辅助
description: "当用户说要扫描即将到期的合同、查 90/60/30 天预警名单、给某位教师生成续聘任务书、下发单位版续聘清单、从 SmartCampus HR 平台抓合同数据、解析续聘材料 PDF/Word、查看本月合同到期人员清单时,加载本技能。支持定时扫描(每日凌晨)+ 数据管理 + Agent 三入口,覆盖合同到期智能预警与续聘辅助全流程。"
version: 1.1.18
version: 1.1.30
author: 深圳匠厂科技有限公司
metadata:
openclaw:

View File

@@ -83,7 +83,8 @@
"type": "cli",
"command": "generate-renewal-task",
"args": ["--contract-id", "{row.合同ID}"]
}
},
"v1_1_30_notes": "contract-id 缺失时返回 JSON 提示(missing_arg),不再 argparse 报错。"
},
{
"id": "dispatch-task-notice",
@@ -99,7 +100,8 @@
"type": "cli",
"command": "dispatch-task-notice",
"args": ["--unit-id", "{form.unit_id}", "--month", "{form.month}"]
}
},
"v1_1_30_notes": "unit-id 缺失时返回 JSON 提示(missing_arg),不再 argparse 报错。"
},
{
"id": "mark-contract-renewed",
@@ -116,7 +118,8 @@
"type": "cli",
"command": "mark-contract-renewed",
"args": ["--contract-id", "{row.合同ID}", "--material-id", "{form.material_id}"]
}
},
"v1_1_30_notes": "contract-id/material-id 缺失时返回 JSON 提示(missing_arg),不再 argparse 报错。"
},
{
"id": "render-html",
@@ -128,7 +131,8 @@
"type": "cli",
"command": "render-html",
"args": ["--task-id", "{form.task_id}"]
}
},
"v1_1_30_notes": "task-id 缺失时返回 JSON 提示(missing_arg),不再 argparse 报错。"
},
{
"id": "show-config",

Binary file not shown.

After

Width:  |  Height:  |  Size: 589 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 156 KiB

View File

@@ -127,6 +127,14 @@ def cmd_scan(args: argparse.Namespace) -> int:
def cmd_generate_task(args: argparse.Namespace) -> int:
contract_id = args.contract_id
task_id = args.task_id
# v1.1.30:友好提示 —— 空 contract_id(客户端 arguments 丢失场景)
if not contract_id:
print(json.dumps({
"error": "missing_arg",
"detail": "缺少必填参数 --contract-id。请在调用方传入合同 ID。",
"hint": "可先运行 `collect-from-platform` 抓取合同台账,或在数据管理 Tab 选中一行后再调用。",
}, ensure_ascii=False, indent=2))
return 1
conn = get_connection()
try:
result = task_generator.regenerate_task_for_contract_id(conn, contract_id, task_id)
@@ -141,6 +149,14 @@ def cmd_generate_task(args: argparse.Namespace) -> int:
def cmd_dispatch(args: argparse.Namespace) -> int:
# v1.1.30:友好提示 —— 空 unit_id
if not args.unit_id:
print(json.dumps({
"error": "missing_arg",
"detail": "缺少必填参数 --unit-id。请在调用方传入二级单位 ID。",
"hint": "可参考 contracts 表的 unit_id 字段;或用 --month 限定月份范围。",
}, ensure_ascii=False, indent=2))
return 1
conn = get_connection()
try:
result = dispatcher.dispatch(conn, args.unit_id, args.month or "")
@@ -151,6 +167,17 @@ def cmd_dispatch(args: argparse.Namespace) -> int:
def cmd_mark_renewed(args: argparse.Namespace) -> int:
# v1.1.30:友好提示 —— 空 contract_id 或 material_id
if not args.contract_id or not args.material_id:
print(json.dumps({
"error": "missing_arg",
"detail": (
f"缺少必填参数 {'--contract-id ' if not args.contract_id else ''}"
f"{'--material-id ' if not args.material_id else ''}(由调用方传入)。"
),
"hint": "合同 ID 来自 contracts 表,材料 ID 来自 materials_content 表。",
}, ensure_ascii=False, indent=2))
return 1
conn = get_connection()
try:
from db import repository as repo
@@ -176,6 +203,14 @@ def cmd_mark_renewed(args: argparse.Namespace) -> int:
def cmd_render_html(args: argparse.Namespace) -> int:
# v1.1.30:友好提示 —— 空 task_id
if not args.task_id:
print(json.dumps({
"error": "missing_arg",
"detail": "缺少必填参数 --task-id。请在调用方传入任务 ID。",
"hint": "可先运行 generate-renewal-task 生成任务,再用其 task_id 调用。",
}, ensure_ascii=False, indent=2))
return 1
conn = get_connection()
try:
output_path = html_renderer.render_task_html(conn, args.task_id)
@@ -380,28 +415,30 @@ def build_parser() -> argparse.ArgumentParser:
sp = sub.add_parser(
"generate-renewal-task", help="为指定合同生成续聘任务书",
)
sp.add_argument("--contract-id", required=True, help="合同 ID")
# v1.1.30:去掉 required=True,给 contract_id 默认空串占位
# (jiangchang 客户端偶发 arguments 丢失时空参数不会让 argparse 直接报错退出)
sp.add_argument("--contract-id", default="", help="合同 ID(必填,由调用方传入)")
sp.add_argument("--task-id", default=None, help="可选:指定 task_id(默认自动生成)")
sp.set_defaults(handler=cmd_generate_task)
sp = sub.add_parser(
"dispatch-task-notice", help="按单位/月份派发续聘通知(dry-run)",
)
sp.add_argument("--unit-id", required=True, help="二级单位 ID")
sp.add_argument("--unit-id", default="", help="二级单位 ID(必填,由调用方传入)")
sp.add_argument("--month", default="", help="月份 YYYY-MM(留空 = 全部)")
sp.set_defaults(handler=cmd_dispatch)
sp = sub.add_parser(
"mark-contract-renewed", help="标记合同已续签",
)
sp.add_argument("--contract-id", required=True, help="合同 ID")
sp.add_argument("--material-id", required=True, help="续签材料 ID")
sp.add_argument("--contract-id", default="", help="合同 ID(必填,由调用方传入)")
sp.add_argument("--material-id", default="", help="续签材料 ID(必填,由调用方传入)")
sp.set_defaults(handler=cmd_mark_renewed)
sp = sub.add_parser(
"render-html", help="渲染任务书 HTML",
)
sp.add_argument("--task-id", required=True, help="任务 ID")
sp.add_argument("--task-id", default="", help="任务 ID(必填,由调用方传入)")
sp.set_defaults(handler=cmd_render_html)
sp = sub.add_parser(

File diff suppressed because it is too large Load Diff

View File

@@ -6,7 +6,7 @@
"""
SKILL_SLUG = "alert-contract-expiring-hr"
SKILL_VERSION = "1.1.18"
SKILL_VERSION = "1.1.30"
LOG_LOGGER_NAME = "openclaw.skill.alert_contract_expiring_hr"
PLATFORM_KIT_MIN_VERSION = "1.2.11"
@@ -48,7 +48,10 @@ PERFORMANCE_RENEW_MIN = 80 # 全部 >= 80 且逐年上升 → renew
# ----- v1.1.6:RPA 超时常量(与 skill-template human_verification.py 对齐)-----
# 手工登录等待(HITL fallback)
HUMAN_WAIT_TIMEOUT = 180 # 手工登录等待超时(秒)—— skill-template 同名常量
HUMAN_POLL_INTERVAL = 2.0 # 轮询间隔(秒)—— skill-template 同名常量
HUMAN_POLL_INTERVAL = 5.0 # 轮询间隔(秒)—— v1.1.19 调长:避免打断用户输入
HUMAN_PROMPT_INTERVAL = 15.0 # 剩余时间提示间隔(秒)—— v1.1.19 新增
URL_STABLE_REQUIRED_S = 2.0 # URL 连续 N 秒不变才算稳定 —— v1.1.19 新增
URL_STABLE_POLL_S = 0.5 # URL 稳定探测间隔(秒)
# Playwright 操作超时(毫秒)
DEFAULT_PAGE_TIMEOUT_MS = 15_000 # Playwright 默认操作超时 —— skill-template 同名常量
@@ -65,16 +68,16 @@ SPA_HOME_RETRY_MAX = 2 # SPA_HOME retry 最大次数
# 表格等待(毫秒)
TABLE_INITIAL_WAIT_MS = 3_000 # 表格初始等待(React + mock 填充)
TABLE_POLL_INTERVAL_MS = 2_000 # 表格轮询间隔
TABLE_POLL_MAX_ATTEMPTS = 20 # 表格轮询最大次数(合计 ~40s)
TABLE_POLL_INTERVAL_MS = 1_500 # v1.1.32 优化:表格轮询间隔(2s → 1.5s)
TABLE_POLL_MAX_ATTEMPTS = 10 # v1.1.32 优化:轮询最大次数(20 → 10,合计 ~15s)
TABLE_TR_TIMEOUT_MS = 3_000 # 单次 wait_for_selector 超时
FULL_RENDER_WAIT_MS = 1_500 # 完整 DOM 渲染后等渲染结束
# 侧边栏/卡片(毫秒)
SIDEBAR_LINK_TIMEOUT_MS = 30_000 # 等侧边栏 hr/contracts link
SIDEBAR_LINK_TIMEOUT_MS = 10_000 # v1.1.32 优化:等侧边栏 link(30s → 10s;sandbox 通常 2-5s)
PUSHSTATE_SETTLE_MS = 1_500 # SPA pushState 后等
SIDEBAR_CLICK_URL_TIMEOUT_MS = 10_000 # click 后等 URL 变 /hr/contracts
URL_AFTER_LOGIN_TIMEOUT_MS = 20_000 # 根域 login 后等跳转
URL_AFTER_LOGIN_TIMEOUT_MS = 10_000 # v1.1.32 优化:根域 login 后等跳转(20s → 10s)
# reload 后备用
RELOAD_NETWORKIDLE_TIMEOUT_MS = 30_000 # page.reload networkidle 超时

View File

@@ -284,6 +284,53 @@ def test_collect_platform_defaults_to_browser() -> tuple[bool, str]:
return True, "默认浏览器模式,--headless 和 --show-browser 行为正确"
def test_sync_actions_tolerate_missing_args_v1_30() -> tuple[bool, str]:
"""v1.1.30:4 个 sync action(generate-renewal-task / dispatch-task-notice /
mark-contract-renewed / render-html)的必填参数全部不传时,
argparse 不应 SystemExit,而是给空字符串默认值;handler 检测到空值后返回
JSON {error: missing_arg, hint: ...} 友好提示,**不再 argparse 错误退出**。
解决问题:jiangchang 客户端 arguments 丢失时调用方收到 argparse 错误而误判失败。
"""
from cli.app import build_parser
parser = build_parser()
# 1. 无参调用不应 argparse SystemExit(注意:之前是 required=True 会抛错)
try:
for cmd in (
"generate-renewal-task",
"dispatch-task-notice",
"mark-contract-renewed",
"render-html",
):
parser.parse_args([cmd]) # 不传任何参数
except SystemExit as e:
return False, f"argparse 在无参调用时 SystemExit({e.code}),应给默认值"
# 2. handler 检测空值应返回 JSON 提示而非抛异常
import argparse
from cli.app import (
cmd_dispatch, cmd_generate_task, cmd_mark_renewed, cmd_render_html,
)
empty_args = argparse.Namespace(
contract_id="", unit_id="", month="", material_id="", task_id="",
)
try:
rc1 = cmd_generate_task(empty_args)
rc2 = cmd_dispatch(empty_args)
rc3 = cmd_mark_renewed(empty_args)
rc4 = cmd_render_html(empty_args)
except Exception as e:
return False, f"handler 应返回 JSON 提示而非抛异常: {e}"
for name, rc in (("generate", rc1), ("dispatch", rc2), ("mark_renewed", rc3), ("render", rc4)):
if rc != 1:
return False, f"{name} 空参数应返回 exit=1,实际 {rc}"
return True, "4 个 sync action 空参数容忍 OK(exit=1 + JSON 提示)"
# ---------- db.repository:v1.1.2 新增的 sandbox card 解析 ----------
def test_parse_sandbox_card_row_format() -> tuple[bool, str]:
@@ -1192,6 +1239,783 @@ def test_download_tasks_count_by_status() -> tuple[bool, str]:
conn.close()
def test_enqueue_download_tasks_safe_basic() -> tuple[bool, str]:
"""_enqueue_download_tasks_safe: 全部成功入队。"""
from service.platform_collector import _enqueue_download_tasks_safe
# 准备 3 个合同(in-memory DB,函数不会关闭)
conn = _make_conn()
for i in range(3):
conn.execute(
f"""INSERT INTO contracts
(contract_id, employee_id, teacher_name, unit_id, unit_name,
position, contract_type, contract_start, contract_end,
trial_period_end, status)
VALUES ('ENQ-{i}', 'T{i}', 'X', 'U{i}', 'Y', '', '',
'2024-01-01', '2026-04-15', '2024-04-15', '可续聘')"""
)
conn.commit()
summary = _enqueue_download_tasks_safe(["ENQ-0", "ENQ-1", "ENQ-2"], conn=conn)
if summary["enqueued"] != 3:
return False, f"enqueued={summary['enqueued']} (want 3) - {summary}"
if summary["failed"] != 0:
return False, f"failed={summary['failed']} - {summary}"
if summary["errors"]:
return False, f"errors={summary['errors']}"
return True, f"3 个合同全部入队: {summary}"
def test_enqueue_download_tasks_safe_idempotent() -> tuple[bool, str]:
"""_enqueue_download_tasks_safe: 同 contract_id 第二次调用 → skipped。"""
from service.platform_collector import _enqueue_download_tasks_safe
conn = _make_conn()
conn.execute(
f"""INSERT INTO contracts
(contract_id, employee_id, teacher_name, unit_id, unit_name,
position, contract_type, contract_start, contract_end,
trial_period_end, status)
VALUES ('ENQ-IDM', 'T1', 'X', 'U1', 'Y', '', '',
'2024-01-01', '2026-04-15', '2024-04-15', '可续聘')"""
)
conn.commit()
s1 = _enqueue_download_tasks_safe(["ENQ-IDM"], conn=conn)
s2 = _enqueue_download_tasks_safe(["ENQ-IDM"], conn=conn)
if s1["enqueued"] != 1:
return False, f"第一次 enqueued={s1['enqueued']} (want 1) - {s1}"
if s2["skipped"] != 1:
return False, f"第二次 skipped={s2['skipped']} (want 1) - {s2}"
if s2["enqueued"] != 0:
return False, f"第二次不应新增 enqueued={s2['enqueued']} - {s2}"
return True, f"幂等 OK: 第一次 {s1}, 第二次 {s2}"
def test_enqueue_download_tasks_safe_fk_failure() -> tuple[bool, str]:
"""_enqueue_download_tasks_safe: contract_id 不存在 → 计入 failed,不抛异常。"""
from service.platform_collector import _enqueue_download_tasks_safe
# 不插入合同,直接入队 → FK 失败(in-memory DB,函数不关闭)
conn = _make_conn()
summary = _enqueue_download_tasks_safe(["DOES-NOT-EXIST"], conn=conn)
if summary["failed"] != 1:
return False, f"failed={summary['failed']} (want 1) - {summary}"
if not summary["errors"]:
return False, f"errors 应非空: {summary}"
if "FOREIGN KEY" not in summary["errors"][0] and "foreign key" not in summary["errors"][0].lower():
return False, f"错误信息应提及 FOREIGN KEY: {summary['errors'][0]}"
return True, f"FK 失败被吞掉: failed=1, errors[0]={summary['errors'][0][:80]}"
def test_enqueue_download_tasks_safe_empty_list() -> tuple[bool, str]:
"""_enqueue_download_tasks_safe: 空列表 → no-op,不开 DB 连接。"""
from service.platform_collector import _enqueue_download_tasks_safe
summary = _enqueue_download_tasks_safe([])
if summary != {"enqueued": 0, "skipped": 0, "failed": 0, "errors": []}:
return False, f"空列表返回值异常: {summary}"
return True, "空列表 no-op OK"
def test_enqueue_download_tasks_safe_auto_create_table_v1_29() -> tuple[bool, str]:
"""v1.1.29:用户没跑 init-db 时,首次调用 `_enqueue_download_tasks_safe`
不传 conn(owns_conn=True),函数内部 idempotent 创建 download_tasks 表
而非抛 "no such table"。
"""
import os
import sqlite3
import tempfile
from pathlib import Path
from service.platform_collector import _enqueue_download_tasks_safe
# 1. 临时文件 DB,**只建 contracts 表**(模拟老 DB 没 download_tasks)
with tempfile.TemporaryDirectory() as tmp:
tmp_db = Path(tmp) / "old.db"
# 重定向 JIANGCHANG_DATA_ROOT 或 get_db_path 用临时路径
# 这里用 monkeypatch get_db_path + get_connection
from util import runtime_paths as rp
from db import connection as dbconn
orig_rp_db_path = rp.get_db_path
orig_dbconn_db_path = dbconn.get_db_path
rp.get_db_path = lambda: tmp_db # type: ignore[assignment]
dbconn.get_db_path = lambda: tmp_db # type: ignore[assignment]
try:
# 模拟老 DB:只有 contracts 表(v1.1.17 之前 init 过)
_tmp_conn = sqlite3.connect(str(tmp_db))
try:
_tmp_conn.executescript(
"""
CREATE TABLE contracts (
contract_id TEXT PRIMARY KEY,
teacher_name TEXT, unit_id TEXT, unit_name TEXT,
position TEXT, contract_type TEXT, contract_start TEXT,
contract_end TEXT, trial_period_end TEXT,
status TEXT DEFAULT '可续聘',
last_material_id TEXT, renewed_at TEXT
);
INSERT INTO contracts
(contract_id, teacher_name, unit_id, unit_name,
position, contract_type, contract_start,
contract_end, trial_period_end)
VALUES ('ENQ-CREATE-1', '测试', 'U1', '物理学院',
'实验师', '专任教师聘用合同',
'2024-01-01', '2026-12-31', '2024-04-01');
"""
)
_tmp_conn.commit()
finally:
_tmp_conn.close()
# 2. 不传 conn 调用(owns_conn=True,会走 get_connection → 触发 ensure 建表)
summary = _enqueue_download_tasks_safe(["ENQ-CREATE-1"])
if summary.get("failed", 0) > 0:
return False, f"应自动建表后入队成功,但 failed={summary}"
# 3. 验证表确实被自动创建了
conn = sqlite3.connect(str(tmp_db))
try:
count = conn.execute(
"SELECT COUNT(*) FROM download_tasks"
).fetchone()[0]
finally:
conn.close()
if count != 1:
return False, f"download_tasks 应有 1 行,实际 {count}"
return True, f"ensure 建表 + 入队 OK(enqueued={summary.get('enqueued')})"
finally:
rp.get_db_path = orig_rp_db_path # type: ignore[assignment]
dbconn.get_db_path = orig_dbconn_db_path # type: ignore[assignment]
# ---------- v1.1.19: _wait_for_url_stable + login flow ----------
class _FakePage:
"""v1.1.19 login flow 测试用的简易 page mock。
- .url: 字符串
- 支持外部修改 .url 模拟用户登录跳转
- v1.1.24:新增 .locator() 返回 _FakeLocatorNoOp,让
_wait_for_login_inputs_filled 的 selector 探测不抛 AttributeError
"""
def __init__(self, initial_url: str = "about:blank") -> None:
self.url = initial_url
self.goto_calls: list[str] = []
self.reload_calls: int = 0
self.click_calls: int = 0
def locator(self, selector: str): # noqa: ARG002
"""v1.1.24 扩展:测试 _wait_for_login_inputs_filled 用,返回空 locator。"""
return _FakeLocator(count=0)
def test_wait_for_url_stable_immediate_match() -> tuple[bool, str]:
"""_wait_for_url_stable: URL 已含目标 + 稳定 → 立即返回 True。"""
from service.platform_collector import _wait_for_url_stable
page = _FakePage("https://example.com/app/contracts")
if not _wait_for_url_stable(page, "/app/", max_wait_sec=5):
return False, "URL 已含 /app/ 应立即返回 True"
if page.goto_calls or page.reload_calls:
return False, f"不应触发 goto/reload: {page.goto_calls}/{page.reload_calls}"
return True, "URL 已含 /app/ 立即返回 OK"
def test_wait_for_url_stable_transition() -> tuple[bool, str]:
"""_wait_for_url_stable: URL 经历 login → /app/ 过渡,等稳定后返回 True。"""
import threading
import time as _time
from service.platform_collector import _wait_for_url_stable
page = _FakePage("https://example.com/login")
def flip_to_app():
_time.sleep(0.6)
page.url = "https://example.com/app/home"
threading.Thread(target=flip_to_app, daemon=True).start()
if not _wait_for_url_stable(page, "/app/", max_wait_sec=5):
return False, f"应等 URL 稳定到 /app/,但超时(最终 URL={page.url})"
return True, f"URL 过渡后稳定 OK(最终 URL={page.url})"
def test_wait_for_url_stable_timeout() -> tuple[bool, str]:
"""_wait_for_url_stable: URL 始终未跳到目标 → 超时返回 False。"""
from service.platform_collector import _wait_for_url_stable
page = _FakePage("https://example.com/login")
if _wait_for_url_stable(page, "/app/", max_wait_sec=2):
return False, "URL 永远含 login,应超时返回 False"
return True, "超时返回 False OK"
def test_wait_for_manual_login_no_goto() -> tuple[bool, str]:
"""_wait_for_manual_login: 等待期间 **不应** 触发 goto / reload / click。"""
import threading
import time as _time
from service import platform_collector as pc
# platform_collector 在 import 时已经把常量复制到模块 globals;
# 直接覆盖模块属性最稳。缩短 HUMAN_POLL_INTERVAL 和 URL_STABLE_REQUIRED_S,
# 让 fake_user 0.8s 翻 URL + 0.3s 稳定窗口能凑进 wait_sec=5。
orig_poll = pc.HUMAN_POLL_INTERVAL
orig_stable = pc.URL_STABLE_REQUIRED_S
pc.HUMAN_POLL_INTERVAL = 0.3
pc.URL_STABLE_REQUIRED_S = 0.3
try:
page = _FakePage("https://example.com/login")
def fake_user_login():
_time.sleep(0.8)
page.url = "https://example.com/app/home"
threading.Thread(target=fake_user_login, daemon=True).start()
try:
pc._wait_for_manual_login(page, wait_sec=5)
except Exception as e:
return False, f"应等用户登录成功,但抛异常: {e}"
if page.goto_calls:
return False, f"等待期间不应 goto: {page.goto_calls}"
if page.reload_calls:
return False, f"等待期间不应 reload: {page.reload_calls}"
return True, "等用户登录期间无打断 OK"
finally:
pc.HUMAN_POLL_INTERVAL = orig_poll
pc.URL_STABLE_REQUIRED_S = orig_stable
# ---------- v1.1.24: _try_auto_sign_in SPA 二级登录页(中文「登录」按钮)----------
class _FakeLocator:
"""v1.1.24 _try_auto_sign_in 测试用的简易 Playwright Locator mock。
支持的 API:
- .count() -> int
- .input_value() -> str
- .click() -> None
- .first -> 自身
"""
def __init__(
self,
count: int = 0,
value: str = "",
click_log: list | None = None,
) -> None:
self._count = count
self._value = value
self._click_log = click_log if click_log is not None else []
@property
def first(self) -> "_FakeLocator":
return self
def count(self) -> int:
return self._count
def input_value(self) -> str:
return self._value
def click(self) -> None:
self._click_log.append(self._value or "clicked")
def fill(self, value: str) -> None:
"""v1.1.27:fill 后 value 立即更新。"""
self._value = value
class _FakeLocatorRouter:
"""根据 selector 返回对应 _FakeLocator 的路由器。
selectors: dict[selector_str, _FakeLocator]
"""
def __init__(self, selectors: dict[str, _FakeLocator], page) -> None:
self._selectors = selectors
self._page = page
def _resolve(self, sel: str) -> _FakeLocator:
# v1.1.32:支持「,」分隔的多 selector 联合(任一命中即返回)
# 例:'*:has-text("DEV MODE"), *:text-is("DEV MODE")'
for part in sel.split(","):
part = part.strip()
if part in self._selectors:
return self._selectors[part]
return _FakeLocator(count=0)
def __call__(self, sel: str):
"""page.locator(sel) 形式调用。"""
return self._resolve(sel)
def __getattr__(self, sel: str):
# loc.page 已存在则按 selector 路由
return self._resolve(sel)
class _FakePlaywrightPage:
"""v1.1.24 _try_auto_sign_in 测试用的 page mock。
usage:
page = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
selectors={
"input[type='tel']": _FakeLocator(count=1, value="18677308191"),
"input[type='password']": _FakeLocator(count=1, value="***"),
"button:has-text('登录')": _FakeLocator(count=1, click_log=click_log),
},
wait_for_selector_log=wait_log,
)
"""
def __init__(
self,
url: str = "about:blank",
selectors: dict[str, _FakeLocator] | None = None,
wait_for_selector_log: list | None = None,
wait_selector_should_raise: bool = False,
) -> None:
self.url = url
self._selectors = selectors or {}
self._wait_log = wait_for_selector_log if wait_for_selector_log is not None else []
self._wait_should_raise = wait_selector_should_raise
def wait_for_selector(self, selector: str, timeout: int = 0) -> None:
self._wait_log.append(selector)
if self._wait_should_raise:
raise Exception("fake: timeout")
@property
def locator(self):
return _FakeLocatorRouter(self._selectors, self)
def test_try_auto_sign_in_spa_login_chinese_button() -> tuple[bool, str]:
"""v1.1.24:SPA 二级登录页(/industries/campus/smartcampus/login),
input 用 `input[type='tel']` / `input[type='password']`,
按钮中文「登录」。预填有值 → 自动点登录。
"""
from service.platform_collector import _try_auto_sign_in
click_log: list[str] = []
page = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
selectors={
"input[type='tel']": _FakeLocator(count=1, value="18677308191"),
"input[type='password']": _FakeLocator(count=1, value="dummy"),
"button:has-text('登录')": _FakeLocator(count=1, click_log=click_log),
},
)
if not _try_auto_sign_in(page, wait_input_sec=2):
return False, "SPA 二级登录页预填应自动点 登录"
if not click_log:
return False, f"未触发点击: {click_log}"
return True, f"SPA 二级登录页中文按钮自动点 OK(click_log={click_log})"
def test_try_auto_sign_in_root_sign_in_english_button() -> tuple[bool, str]:
"""v1.1.24(向后兼容):根域登录页(https://sandbox.jc2009.com/),
input 用 `#jc-user` / `#jc-pass`,按钮英文「Sign in」。预填有值 → 自动点。
"""
from service.platform_collector import _try_auto_sign_in
click_log: list[str] = []
page = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/",
selectors={
"#jc-user": _FakeLocator(count=1, value="demo"),
"#jc-pass": _FakeLocator(count=1, value="123456"),
"button:has-text('Sign in')": _FakeLocator(count=1, click_log=click_log),
},
)
if not _try_auto_sign_in(page, wait_input_sec=2):
return False, "根域登录页预填应自动点 Sign in"
if not click_log:
return False, f"未触发点击: {click_log}"
return True, f"根域登录页 Sign in 自动点 OK(click_log={click_log})"
def test_try_auto_sign_in_no_input_returns_false() -> tuple[bool, str]:
"""v1.1.24:input 为空(用户还没输入)→ 返回 False,不点按钮。"""
from service.platform_collector import _try_auto_sign_in
click_log: list[str] = []
page = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
selectors={
"input[type='tel']": _FakeLocator(count=1, value=""), # 空
"input[type='password']": _FakeLocator(count=1, value="123"),
"button:has-text('登录')": _FakeLocator(count=1, click_log=click_log),
},
)
if _try_auto_sign_in(page, wait_input_sec=2):
return False, "user 空时不应自动点"
if click_log:
return False, f"未填完不应点击: {click_log}"
return True, "input 空返回 False 不点 OK"
def test_try_auto_sign_in_selector_timeout_returns_false() -> tuple[bool, str]:
"""v1.1.24:等 input 超时(如 URL 已是 SPA home,无登录 input)→ 返回 False。"""
from service.platform_collector import _try_auto_sign_in
click_log: list[str] = []
page = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/app/home",
selectors={}, # 没有任何 input
wait_selector_should_raise=True,
)
if _try_auto_sign_in(page, wait_input_sec=1):
return False, "无登录 input 应返回 False"
if click_log:
return False, f"无 input 不应点击: {click_log}"
return True, "无登录 input 超时返回 False OK"
def test_wait_for_login_inputs_filled_auto_click_v1_26() -> tuple[bool, str]:
"""v1.1.26:`_wait_for_login_inputs_filled` **自动点登录按钮**——
按用户最新要求:「到了 /industries/campus/smartcampus/login 整个页面,
有账号和密码可用直接点登录跳转」。
input 有值时点击 `button:has-text('登录')` 或 `Sign in`(多按钮文本优先级)。
"""
from service import platform_collector as pc
orig_poll = pc.HUMAN_POLL_INTERVAL
orig_prompt = pc.HUMAN_PROMPT_INTERVAL
pc.HUMAN_POLL_INTERVAL = 0.2
pc.HUMAN_PROMPT_INTERVAL = 0.2
try:
click_log: list[str] = []
page = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
selectors={
"input[type='tel']": _FakeLocator(count=1, value="18677308191"),
"input[type='password']": _FakeLocator(count=1, value="dummy"),
"button:has-text('登录')": _FakeLocator(count=1, click_log=click_log),
},
)
result = pc._wait_for_login_inputs_filled(page, wait_sec=5)
if not result:
return False, "input 有值应自动点登录 → return True"
if not click_log:
return False, f"v1.1.26 应自动点登录按钮,但 click_log={click_log}"
return True, f"v1.1.26 自动点登录按钮 OK(click_log={click_log})"
finally:
pc.HUMAN_POLL_INTERVAL = orig_poll
pc.HUMAN_PROMPT_INTERVAL = orig_prompt
def test_fill_spa_login_from_env_basic_v1_31() -> tuple[bool, str]:
"""v1.1.31:`_fill_spa_login_from_env` ——从 .env 读 SMARTCAMPUS_LOGIN_USER /
SMARTCAMPUS_LOGIN_PASSWORD fill 再点登录按钮(密码不硬编码到代码)。
"""
import os
from service import platform_collector as pc
orig_u, orig_p = os.environ.get("SMARTCAMPUS_LOGIN_USER"), os.environ.get("SMARTCAMPUS_LOGIN_PASSWORD")
os.environ["SMARTCAMPUS_LOGIN_USER"] = "hr_demo"
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = "hr_secret_xxx"
try:
click_log: list[str] = []
page = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
selectors={
"#jc-user": _FakeLocator(count=1, value=""),
"#jc-pass": _FakeLocator(count=1, value=""),
"button:has-text('Sign in')": _FakeLocator(count=1, click_log=click_log),
},
)
if not pc._fill_spa_login_from_env(page):
return False, "SPA login fill(来自 .env)应成功"
if not click_log:
return False, f"应点 Sign in,但 click_log={click_log}"
return True, f"SPA 二级登录页 .env 凭证自动 fill+点 OK(click_log={click_log})"
finally:
if orig_u is None:
os.environ.pop("SMARTCAMPUS_LOGIN_USER", None)
else:
os.environ["SMARTCAMPUS_LOGIN_USER"] = orig_u
if orig_p is None:
os.environ.pop("SMARTCAMPUS_LOGIN_PASSWORD", None)
else:
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = orig_p
def test_fill_spa_login_from_env_chinese_button_v1_31() -> tuple[bool, str]:
"""v1.1.31:中文「登录」按钮也能 fill .env 凭证并点。"""
import os
from service import platform_collector as pc
orig_u, orig_p = os.environ.get("SMARTCAMPUS_LOGIN_USER"), os.environ.get("SMARTCAMPUS_LOGIN_PASSWORD")
os.environ["SMARTCAMPUS_LOGIN_USER"] = "hr_demo"
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = "hr_secret_xxx"
try:
click_log: list[str] = []
page = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
selectors={
"input[type='tel']": _FakeLocator(count=1, value=""),
"input[type='password']": _FakeLocator(count=1, value=""),
"button:has-text('登录')": _FakeLocator(count=1, click_log=click_log),
},
)
if not pc._fill_spa_login_from_env(page):
return False, "SPA login 中文按钮 fill 应成功"
if not click_log:
return False, f"应点 登录,但 click_log={click_log}"
return True, f"SPA 二级登录页中文「登录」按钮 OK(click_log={click_log})"
finally:
if orig_u is None:
os.environ.pop("SMARTCAMPUS_LOGIN_USER", None)
else:
os.environ["SMARTCAMPUS_LOGIN_USER"] = orig_u
if orig_p is None:
os.environ.pop("SMARTCAMPUS_LOGIN_PASSWORD", None)
else:
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = orig_p
def test_fill_spa_login_from_env_no_credentials_returns_false_v1_31() -> tuple[bool, str]:
"""v1.1.31:.env 未配置 SMARTCAMPUS_LOGIN_USER / PASSWORD → 返回 False,不自动 fill。
按用户要求「密码不能直接放到代码」,无 .env 凭证时等用户手工输入。
v1.1.32 修复:清空进程 env 同时清空 _ENV_CACHE(_get_env 会从 .env 文件 fallback)。
"""
import os
from service import platform_collector as pc
orig_u = os.environ.pop("SMARTCAMPUS_LOGIN_USER", None)
orig_p = os.environ.pop("SMARTCAMPUS_LOGIN_PASSWORD", None)
# v1.1.32:清空 .env 缓存 + 临时把 .env 路径指向不存在的文件
orig_cache = pc._ENV_CACHE.copy()
orig_cache_loaded = pc._ENV_CACHE_LOADED
orig_get_env_path = pc.get_env_path if hasattr(pc, "get_env_path") else None
from pathlib import Path
pc._ENV_CACHE = {}
pc._ENV_CACHE_LOADED = True
pc.get_env_path = lambda: Path("Z:/nonexistent-venv-test/.env") # type: ignore[assignment]
try:
click_log: list[str] = []
page = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
selectors={
"#jc-user": _FakeLocator(count=1, value=""),
"#jc-pass": _FakeLocator(count=1, value=""),
"button:has-text('Sign in')": _FakeLocator(count=1, click_log=click_log),
},
)
if pc._fill_spa_login_from_env(page):
return False, "无 .env 凭证时不应自动 fill"
if click_log:
return False, f"无 .env 凭证时不应点登录按钮,但 click_log={click_log}"
return True, "无 .env 凭证返回 False 不点登录 OK"
finally:
if orig_u is not None:
os.environ["SMARTCAMPUS_LOGIN_USER"] = orig_u
if orig_p is not None:
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = orig_p
pc._ENV_CACHE = orig_cache
pc._ENV_CACHE_LOADED = orig_cache_loaded
if orig_get_env_path is not None:
pc.get_env_path = orig_get_env_path # type: ignore[assignment]
def test_fill_spa_login_from_env_no_input_returns_false_v1_31() -> tuple[bool, str]:
"""v1.1.31:找不到 user input(页面不是 login 页)→ 返回 False。"""
import os
from service import platform_collector as pc
orig_u, orig_p = os.environ.get("SMARTCAMPUS_LOGIN_USER"), os.environ.get("SMARTCAMPUS_LOGIN_PASSWORD")
os.environ["SMARTCAMPUS_LOGIN_USER"] = "hr_demo"
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = "hr_secret_xxx"
try:
page = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/app/home",
selectors={},
)
if pc._fill_spa_login_from_env(page):
return False, "无 login input 应返回 False"
return True, "无 login input 返回 False OK"
finally:
if orig_u is None:
os.environ.pop("SMARTCAMPUS_LOGIN_USER", None)
else:
os.environ["SMARTCAMPUS_LOGIN_USER"] = orig_u
if orig_p is None:
os.environ.pop("SMARTCAMPUS_LOGIN_PASSWORD", None)
else:
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = orig_p
# ---------- v1.1.32: DEV MODE 面板对应行业 Sign in 点击 ----------
class _CountingLocator(_FakeLocator):
"""v1.1.32:扩展 _FakeLocator,支持 nth() 返回子 locator。"""
def __init__(self, *args, parent_count: int = 0, **kwargs) -> None:
super().__init__(*args, **kwargs)
self._parent_count = parent_count
def nth(self, n: int) -> "_FakeLocator":
return _FakeLocator(
count=1 if n < self._parent_count else 0,
value=self._value,
click_log=self._click_log,
)
def _make_dev_mode_page(
sign_in_count: int = 4,
with_dev_marker: bool = True,
main_form_filled: bool = True,
) -> _FakePlaywrightPage:
"""v1.1.32:构造 DEV MODE 面板的 fake page。
Args:
sign_in_count: Sign in 按钮总数(教师通用=1 + N 个行业 tile)
with_dev_marker: 是否包含 DEV MODE 标志
main_form_filled: 主表单是否已预填 demo(左侧教师登录表单)
"""
selectors: dict[str, _FakeLocator] = {}
if with_dev_marker:
# DEV MODE 面板标志:模拟一个含 'DEV MODE' 文本的元素
selectors["*:has-text('DEV MODE')"] = _FakeLocator(count=1)
# Sign in 按钮(DEV MODE 面板里每个 tile 一个 + 主表单的 Sign in = sign_in_count + 1)
click_log: list[str] = []
selectors["button:has-text('Sign in')"] = _CountingLocator(
count=sign_in_count, parent_count=sign_in_count, click_log=click_log,
)
# 主表单
user_val = "demo" if main_form_filled else ""
selectors["#jc-user"] = _FakeLocator(count=1, value=user_val)
selectors["#jc-pass"] = _FakeLocator(count=1, value="123456" if main_form_filled else "")
return _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
selectors=selectors,
)
def test_dev_mode_panel_open_detection_v1_32() -> tuple[bool, str]:
"""v1.1.32:DEV MODE 面板展开(≥2 Sign in + DEV MODE 标志)→ _is_dev_mode_panel_open 返回 True。"""
from service import platform_collector as pc
page = _make_dev_mode_page(sign_in_count=4, with_dev_marker=True)
if not pc._is_dev_mode_panel_open(page):
return False, "DEV MODE 面板展开但 _is_dev_mode_panel_open 返回 False"
return True, "DEV MODE 面板检测 OK"
def test_dev_mode_panel_closed_detection_v1_32() -> tuple[bool, str]:
"""v1.1.32:DEV MODE 面板未展开(<2 Sign in 或无 DEV MODE 标志)→ 返回 False。"""
from service import platform_collector as pc
# 场景 A:只有 1 个 Sign in 按钮(主表单),无 DEV MODE 标志
page_a = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
selectors={
"button:has-text('Sign in')": _FakeLocator(count=1),
"#jc-user": _FakeLocator(count=1, value="demo"),
"#jc-pass": _FakeLocator(count=1, value="123456"),
},
)
if pc._is_dev_mode_panel_open(page_a):
return False, "1 个 Sign in 不应判定为 DEV MODE 展开"
# 场景 B:≥2 Sign in 但无 DEV MODE 标志
page_b = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
selectors={
"button:has-text('Sign in')": _FakeLocator(count=3),
},
)
if pc._is_dev_mode_panel_open(page_b):
return False, "无 DEV MODE 标志不应判定为 DEV MODE 展开"
return True, "DEV MODE 面板关闭检测 OK"
def test_click_dev_mode_sign_in_for_industry_v1_32() -> tuple[bool, str]:
"""v1.1.32:点击 DEV MODE 面板对应行业 tile 的 Sign in。
无 SMARTCAMPUS_LOGIN_INDUSTRY env 时默认 hr → 点第 2 个 Sign in(人事处 tile)。
"""
import os
from service import platform_collector as pc
orig_industry = os.environ.pop("SMARTCAMPUS_LOGIN_INDUSTRY", None)
try:
page = _make_dev_mode_page(sign_in_count=4)
# _click_dev_mode_sign_in_for_industry 不依赖 _is_dev_mode_panel_open
# 这里直接测点击逻辑
if not pc._click_dev_mode_sign_in_for_industry(page, "hr"):
return False, "点 hr tile Sign in 应成功"
# 验证 click 落在 Sign in 按钮上(点击会让 click_log 多 1 条)
all_sign_in = page._selectors["button:has-text('Sign in')"]
if len(all_sign_in._click_log) < 1:
return False, f"应点击 Sign in,但 click_log={all_sign_in._click_log}"
return True, f"DEV MODE hr tile Sign in 点击 OK(click_log={all_sign_in._click_log})"
finally:
if orig_industry is not None:
os.environ["SMARTCAMPUS_LOGIN_INDUSTRY"] = orig_industry
def test_fill_spa_login_from_env_clicks_dev_mode_sign_in_first_v1_32() -> tuple[bool, str]:
"""v1.1.32:DEV MODE 面板展开时,_fill_spa_login_from_env 应先点对应行业 Sign in
再走原 fill + 点主表单 Sign in 流程(让对应行业 demo 凭据填到主表单)。
"""
import os
from service import platform_collector as pc
orig_u = os.environ.get("SMARTCAMPUS_LOGIN_USER")
orig_p = os.environ.get("SMARTCAMPUS_LOGIN_PASSWORD")
orig_industry = os.environ.get("SMARTCAMPUS_LOGIN_INDUSTRY")
os.environ["SMARTCAMPUS_LOGIN_USER"] = "hr_demo"
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = "hr_secret_xxx"
os.environ["SMARTCAMPUS_LOGIN_INDUSTRY"] = "finance"
try:
# 主表单 input 留空(DEV MODE 面板存在时,主表单可能没填)
page = _make_dev_mode_page(sign_in_count=4, main_form_filled=False)
if not pc._fill_spa_login_from_env(page):
return False, "_fill_spa_login_from_env 应成功(DEV MODE + .env)"
all_sign_in = page._selectors["button:has-text('Sign in')"]
# 期望点击次数 ≥ 2:1 次 DEV MODE finance tile + 1 次主表单 Sign in
if len(all_sign_in._click_log) < 2:
return False, (
f"应先点 DEV MODE 行业 Sign in 再点主表单 Sign in,"
f"但 click_log={all_sign_in._click_log}(期望 ≥2 次)"
)
return True, (
f"SPA 登录页 DEV MODE + .env fill 流程 OK(click_log={all_sign_in._click_log})"
)
finally:
if orig_u is None:
os.environ.pop("SMARTCAMPUS_LOGIN_USER", None)
else:
os.environ["SMARTCAMPUS_LOGIN_USER"] = orig_u
if orig_p is None:
os.environ.pop("SMARTCAMPUS_LOGIN_PASSWORD", None)
else:
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = orig_p
if orig_industry is None:
os.environ.pop("SMARTCAMPUS_LOGIN_INDUSTRY", None)
else:
os.environ["SMARTCAMPUS_LOGIN_INDUSTRY"] = orig_industry
# ---------- v1.1.18: show-config / logs-list ----------
@@ -1422,6 +2246,7 @@ def main() -> int:
_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("cli/sync_actions_tolerate_missing_args_v1_30", test_sync_actions_tolerate_missing_args_v1_30)
_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)
@@ -1457,6 +2282,28 @@ def main() -> int:
_run("download_tasks/claim_atomic", test_download_tasks_claim_atomic)
_run("download_tasks/mark_success_and_failed", test_download_tasks_mark_success_and_failed)
_run("download_tasks/count_by_status", test_download_tasks_count_by_status)
_run("download_tasks/enqueue_safe_basic", test_enqueue_download_tasks_safe_basic)
_run("download_tasks/enqueue_safe_idempotent", test_enqueue_download_tasks_safe_idempotent)
_run("download_tasks/enqueue_safe_fk_failure", test_enqueue_download_tasks_safe_fk_failure)
_run("download_tasks/enqueue_safe_empty_list", test_enqueue_download_tasks_safe_empty_list)
_run("download_tasks/enqueue_safe_auto_create_table_v1_29", test_enqueue_download_tasks_safe_auto_create_table_v1_29)
_run("login_flow/wait_url_stable_immediate", test_wait_for_url_stable_immediate_match)
_run("login_flow/wait_url_stable_transition", test_wait_for_url_stable_transition)
_run("login_flow/wait_url_stable_timeout", test_wait_for_url_stable_timeout)
_run("login_flow/wait_manual_login_no_goto", test_wait_for_manual_login_no_goto)
_run("login_flow/auto_sign_in_spa_chinese_button", test_try_auto_sign_in_spa_login_chinese_button)
_run("login_flow/auto_sign_in_root_english_button", test_try_auto_sign_in_root_sign_in_english_button)
_run("login_flow/auto_sign_in_no_input_returns_false", test_try_auto_sign_in_no_input_returns_false)
_run("login_flow/auto_sign_in_selector_timeout_returns_false", test_try_auto_sign_in_selector_timeout_returns_false)
_run("login_flow/wait_login_inputs_filled_auto_click_v1_26", test_wait_for_login_inputs_filled_auto_click_v1_26)
_run("login_flow/fill_spa_login_from_env_basic_v1_31", test_fill_spa_login_from_env_basic_v1_31)
_run("login_flow/fill_spa_login_from_env_chinese_button_v1_31", test_fill_spa_login_from_env_chinese_button_v1_31)
_run("login_flow/fill_spa_login_from_env_no_credentials_returns_false_v1_31", test_fill_spa_login_from_env_no_credentials_returns_false_v1_31)
_run("login_flow/fill_spa_login_from_env_no_input_returns_false_v1_31", test_fill_spa_login_from_env_no_input_returns_false_v1_31)
_run("login_flow/dev_mode_panel_open_detection_v1_32", test_dev_mode_panel_open_detection_v1_32)
_run("login_flow/dev_mode_panel_closed_detection_v1_32", test_dev_mode_panel_closed_detection_v1_32)
_run("login_flow/click_dev_mode_sign_in_for_industry_v1_32", test_click_dev_mode_sign_in_for_industry_v1_32)
_run("login_flow/fill_spa_login_from_env_clicks_dev_mode_sign_in_first_v1_32", test_fill_spa_login_from_env_clicks_dev_mode_sign_in_first_v1_32)
_run("show_config/masks_tokens", test_show_config_masks_tokens)
_run("show_config/skips_empty", test_show_config_skips_empty)
_run("show_config/skips_comments", test_show_config_skips_comments)