From a03067720b5681afd87b1c3924c2d30b23693cca Mon Sep 17 00:00:00 2001 From: sinohqb Date: Thu, 13 Aug 2026 02:56:32 +0800 Subject: [PATCH] test(intelligent-eval): add openclaw_client + webhook tests (#5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit T5 OpenClawClient subprocess args: docker exec cmd + token-last placement, cron add/rm/list params, JSON parse success + failure paths. T6 webhook: marked sent on 2xx, failure does not raise; xfail guards expose missing retry and missing dedupe (.scratch/v111-architecture-scan.md §6.2). --- .scratch/v111-architecture-scan.md | 152 +++++++++++++++ .../unit/test_openclaw_client_and_webhook.py | 178 ++++++++++++++++++ 2 files changed, 330 insertions(+) create mode 100644 .scratch/v111-architecture-scan.md create mode 100644 tests/unit/test_openclaw_client_and_webhook.py diff --git a/.scratch/v111-architecture-scan.md b/.scratch/v111-architecture-scan.md new file mode 100644 index 0000000..22199d1 --- /dev/null +++ b/.scratch/v111-architecture-scan.md @@ -0,0 +1,152 @@ +# v1.1.1 智能评估模块架构扫描 + +> 生成：2026-08-13 ｜ 用途：v1.1.1（收尾 + 测试加固 + 架构深化）的输入文档，驱动 Gitea tickets 拆解 +> 范围：`backend/agenteval/intelligent_eval/` 为核心 + 外部边界（web routers、openclaw_client、前端、Worker Skill） +> 依据：grilling 共识（测试先行为锁定 → 分阶段深化 → 统一发布 v1.1.1） + +--- + +## 1. 模块现状 + +`intelligent_eval/` 共 13 文件 / 2883 行（v1.1.0 新增功能模块，生产单进程 FastAPI + SQLite）： + +| 文件 | 行数 | 职责 | +|---|---|---| +| repository.py | 459 | 3 个 Repository 类（Eval/Session/Message）+ CAS 条件写 | +| lifecycle.py | 354 | 智能评估生命周期（create/plan/approve/reject/cancel/report + session 会话） | +| cron_pool.py | 310 | Cron 池管理（initialize/scale/auto_scale/sync/stuck 检测） | +| decision.py | 276 | 自主决策（时段/欠账/严重度 → execute/wait/analyze） | +| task_queue.py | 247 | 任务队列（扫描入队/优先级/next/assign/complete） | +| alerts.py | 216 | 告警规则与历史 | +| fault_tolerance.py | 204 | 故障恢复（stuck 处理/对账/双端重启恢复） | +| read_model.py | 196 | 读模型投影（list/detail/report） | +| openclaw_client.py | 177 | OpenClaw 容器 CLI 客户端（docker exec） | +| report.py | 130 | 报告渲染 | +| metrics.py | 117 | 指标聚合 | +| config_snapshot.py | 109 | 配置快照 | +| models.py | 87 | 领域模型 | + +**依赖关系要点**：`lifecycle`→repository；`read_model`→repository；`cron_pool`→openclaw_client + **task_queue（267 行运行时 import complete_task）**；`fault_tolerance`→cron_pool+task_queue+openclaw_client；`decision`/`task_queue` 直接 `session.exec` 查库。 + +**外部边界**： +- **Web**：`web/routers/intelligent_evals.py`（前缀 `/api/intelligent-evals`）+ `web/routers/openclaw_cron_pool.py`（前缀 `/api/openclaw`） +- **OpenClaw 客户端**：docker exec `openclaw` CLI 管理容器内 cron；**硬编码** `gateway_token="agenteval-openclaw-token-2026"` +- **Worker Skill**：OpenClaw 容器内 cron worker 经 docker 网络直连后端 `tasks/next|assign|complete` + `heartbeat`，**不经前端** +- **前端**：`IntelligentEvals.tsx`（读模型 `useIntelligentEvalRead`，统一轮询 + stale 过滤）；`CronPoolMonitor.tsx`（**自带 setInterval 5s**，无统一 hook） + +--- + +## 2. Seam 泄漏清单（深化候选） + +| # | 位置 | 泄漏 | 影响 | 深化方向 | +|---|---|---|---|---| +| **S1** | `task_queue.py`（`_is_slot_due`/`_calculate_session_deficit`/`_calculate_priority`）vs `decision.py`（`_parse_time_slot`/`_get_current_slot`/`_count_sessions_in_slot`/`_has_high_severity_issues`） | **时段/欠账/严重度领域知识重复**："8-10h" 解析两份、时段内 session 统计两份、优先级/欠账各自实现 | 改口径需改两处，易漂移 | 收敛为 shared 领域模块，两模块消费同一实现 | +| **S2** | `web/routers/intelligent_evals.py` 后段（tasks/next 内联拼装、assign/complete、decision-logs POST/GET 直接 ORM at L332/360）+ `openclaw_cron_pool.py`（scale 决策混入、heartbeat ORM at L86/87） | **HTTP 层直接拼装 ORM/领域逻辑**；前段干净 HTTP→lifecycle，后段落差 | decision-logs（Worker 审计面）回归风险高 | 收进 service/repository，router 只做参数校验+委托 | +| **S3** | `decision.py`、`task_queue.py` 直接 `session.exec` 查 `IntelligentEvalSessionDB` | **纯决策/队列逻辑与 SQLite 直接耦合**，绕过 repository/read_model | 可测试性受限（需真实 DB）；逻辑难单测 | 注入读模型/查询接口，逻辑层纯化 | +| **S4** | `cron_pool.py:267` `handle_stuck_cron` 内 `from ... import complete_task` | **stuck→任务结算横跨两模块**（运行时 import） | 卡死恢复的结算顺序分散 | 把 stuck→requeue/complete 结算收进 task_queue/统一作业模块 | +| **S5** | `openclaw_client.py` 硬编码 `gateway_token`/`container_name` + subprocess | **外部系统边界配置硬编码**；token 明文进进程列表；无独立测试（全 mock） | 换 token/环境需改代码；边界无守护 | token/容器走 settings 注入；subprocess 参数/解析加契约测试 | +| **S6** | `pages/CronPoolMonitor.tsx` 自带 setInterval | **前端轮询不一致**：Cron 池不走统一 usePolling，无 stale 过滤；无测试 | 与智能评估读模型行为漂移；卸载/不可见不清理 | 抽统一轮询 hook + 页面测试 | +| **S7** | Worker Skill 回路（tasks/next/assign/complete + heartbeat） | **关键链路无端到端守护**（heartbeat→busy→kill→requeue 故障链路缺失） | 深化动代码时最怕破坏的回环 | 补 e2e 测试 + router 下沉后契约守卫 | + +> 非候选：`repository.py` 单文件多类——v2 方法论明确"不因文件长度拆分 Repository，文件移动不算 deepening"，不处理。 + +--- + +## 3. 测试覆盖矩阵 + +### 现有覆盖（智能评估相关：10 单元 + 9 集成 + 1 前端） + +| 主题 | 单元 | 集成 | 前端 | 评价 | +|---|---|---|---|---| +| 生命周期状态机 | ✅ model 测试 | ✅ evals_api | — | 完善 | +| 决策逻辑 | ✅ decision 测试 | — | — | 分支覆盖好 | +| 任务队列 | ✅ task_queue 测试 | ✅ task_queue_api | — | happy path 完整 | +| Cron 池管理 | ✅ cron_pool 测试 | ✅ cron_pool_api | — | 扩缩容路径完整 | +| 故障恢复 | ✅ fault_tolerance | ✅ fault_tolerance_e2e | — | 双端恢复均有 | +| 告警 | ✅ alerts 测试 | ✅ metrics_alerts_api | — | 规则覆盖好 | +| 快照 | ✅ config_snapshot | ✅ config_snapshot_api | — | 完善 | +| 读模型 | ✅ read_model 测试 | — | ✅ intelligentEval.test | 有 stale 过滤 | +| Worker Skill | — | ✅ worker_skill_api | ❌ | CRUD 有，回路无 e2e | +| **Cron 池监控前端** | — | — | ❌ **0 测试** | 5s 轮询无守卫 | + +## 6. 实施期间发现的新问题（v1.1.1 P0 期间） + +> 这里记录在按本扫描实施时**新暴露**的真实问题（不在原 seam/T 列表里），是测试防护网的首要价值。 + +### 6.1 `assign_task` 缺原子 CAS（**真 bug**，T1 暴露） + +**症状**：`tests/integration/test_worker_task_resilience.py::test_concurrent_assign_via_two_sessions` 失败。 +两个独立 session（两个 worker 进程）并发调用 `assign_task` 同一 pending task 时，**两个都返回 True**——任务被双认领。 + +**根因**：`backend/agenteval/intelligent_eval/task_queue.py` 的 `assign_task` 是非原子的 *读-检查-写*： +```python +task = session.get(...); if task.status != "pending": return False +task.status = "assigned"; session.commit() +``` +SQLite（`check_same_thread=False` + StaticPool）下两个 session 各自读到 `pending`、各自检查通过、各自 commit，无行级锁/CAS。 + +**影响**：Worker Skill 端点 `POST /tasks/{id}/assign` 可被双认领 → 两个 cron 同时执行同一任务 → 重复执行、可能双写 report/decision log。当前顺序 assign（单 session）通过只是因为 session 内 serializable。 + +**修复方向**（待开独立 issue，不在 #3 范围）： +- `assign_task` 改为 `UPDATE ... WHERE id=? AND status='pending'`，按 affected_rows 判定（真正的 CAS）。 +- 同样审计 `complete_task` 的并发（虽然 converge 测试通过，但非原子写）。 +- `get_next_task` 的 `SELECT ... LIMIT 1` 也非原子，应改 `UPDATE ... WHERE status='pending' ... RETURNING` 抢占式。 + +### 6.2 Webhook 缺重试 + 缺去重（**两个真缺口**，T6 暴露） + +**症状**：`test_webhook_retries_on_failure` 与 `test_webhook_dedupes_repeat_triggers` xfail。 +`backend/agenteval/intelligent_eval/alerts.py::AlertManager._send_webhook`： +```python +try: + httpx.post(self.webhook_url, json=payload, timeout=5.0); ... +except Exception as e: + _logger.error(f"Failed to send webhook: {e}") # 吞掉，不重试 +``` +- **无重试**：首次失败仅 log，`webhook_sent` 保持 `False`，无后续尝试。 +- **无去重**：`check_alerts` 每次触发都会调 `_send_webhook(alert)`；同一 alert 在恢复前可被重复发送。 + +**影响**：瞬时网络抖动丢告警；告警风暴期间 webhook 被反复打。 + +**修复方向**（待开独立 issue，不在 #5 范围）： +- 重试：失败后排队 N 次重试（指数退避），或记入 outbox 表后台重发。 +- 去重：`alert.webhook_sent=True` 后跳过；或加 `webhook_in_flight` 锁。 + +**当前状态**：两个 xfail 守卫在 `test_openclaw_client_and_webhook.py`，CI 不阻塞；修复后移除 xfail 即转绿。 + +### 薄弱/缺失点（加固清单 T） + +### 薄弱/缺失点（加固清单 T） + +| # | 缺口 | 位置 | +|---|---|---| +| T1 | Worker Skill 三端点**并发/竞争**（双 worker 拉 next 不得同 task；complete 与 stuck 同时发生不丢任务） | task_queue / worker 回路 | +| T2 | **CronPoolMonitor 页面/hook 测试**（unmount 清理、不可见暂停轮询） | 前端 | +| T3 | **router 内嵌 ORM 契约测试**（decision-logs、heartbeat 一旦下沉的守卫） | 集成 | +| T4 | **auto-scale 振荡**（连续调度不抖动翻转） | cron_pool | +| T5 | **OpenClawClient 单元测试**（subprocess 参数、token 位置、JSON 解析失败路径） | openclaw_client | +| T6 | **webhook 失败重试/去重**（已发送但失败、重发不重复建 alert） | alerts | +| T7 | **告警→扩缩容端到端联动**（check_alerts → auto_scale 贯穿链路） | 集成 | +| T8 | **decision-logs 不可变性/去重**（同 eval+type+context 不重复；context 修改追加非覆盖） | worker_skill | +| T9 | **stuck 检测临界值边界**（last_active_at 临界） | cron_pool | +| T10 | **e2e 故障链路**（heartbeat→busy→kill→requeue 完整闭环） | 集成 e2e | + +--- + +## 4. 深化 Phase 框架（v2 风格，内容待 tickets 细化） + +| Phase | 内容 | 对应 seam | +|---|---|---| +| **P0 测试加固（防护网）** | 覆盖矩阵清零 + T1–T10 补齐；不改生产代码 | — | +| **P1 收敛领域知识** | 时段/欠账/严重度 → shared 模块；stuck 结算收敛 | S1, S4 | +| **P2 解耦 DB** | decision/task_queue 走读模型/repository，逻辑纯化 | S3 | +| **P3 下沉 Router 逻辑** | decision-logs/heartbeat/tasks 收进 service | S2 | +| **P4 边界收尾 + ADR** | openclaw_client 配置化、前端轮询统一、删旧、记录 ADR | S5, S6, S7 | + +> 顺序依赖：P0 先行（锁行为）→ P1（收敛）→ P2（解耦，为 P3 提供 repository 能力）→ P3（router 下沉）→ P4（边界+ADR）。每阶段 CI 全绿、可独立合入。 + +--- + +## 5. 收尾发布（v1.1.1） + +- 测试/深化全部合入 main、CI 全绿后统一发布：`bump 1.1.0 → 1.1.1` → 完整回归（后端全套 + 前端 + tsc + 生产构建 + ruff + 迁移往返）→ 生产部署演练（副本验证迁移/恢复/读取）→ **CD 自动部署**到生产（1.0.0→1.1.1）→ 健康检查 + 关键路径 smoke +- 附带收尾：CI 镜像 push 到私有 registry（持久化） diff --git a/tests/unit/test_openclaw_client_and_webhook.py b/tests/unit/test_openclaw_client_and_webhook.py new file mode 100644 index 0000000..4d43f6b --- /dev/null +++ b/tests/unit/test_openclaw_client_and_webhook.py @@ -0,0 +1,178 @@ +"""OpenClaw client + alert webhook tests (Gitea issue #5 / P0). + +T5 — OpenClawClient subprocess args (token placement, cron subcommand params, JSON parse). +T6 — Alert webhook send/skip/retry/dedupe semantics. +""" +import asyncio +import json +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from sqlmodel import Session + +from agenteval.intelligent_eval import openclaw_client as oc_mod +from agenteval.intelligent_eval.alerts import AlertHistoryDB, AlertManager +from agenteval.intelligent_eval.openclaw_client import OpenClawClient +from agenteval.storage.db import utc_now + + +# --------------------------------------------------------------------------- +# T5 — OpenClawClient subprocess args +# --------------------------------------------------------------------------- + + +def _make_process(stdout: bytes, stderr: bytes = b"", returncode: int = 0): + p = MagicMock() + p.communicate = AsyncMock(return_value=(stdout, stderr)) + p.returncode = returncode + return p + + +def test_create_cron_subprocess_args(monkeypatch: pytest.MonkeyPatch): + """create_cron builds the documented docker exec invocation. + + Asserts: `docker exec openclaw cron add --cron --name + --message --json [--description state:...] --token=`. + Token is appended LAST (documented requirement). + """ + captured = {} + + async def fake_exec(*cmd, **kwargs): + captured["cmd"] = list(cmd) + return _make_process(b'{"id":"cron-1"}') + + monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_exec) + + client = OpenClawClient(gateway_token="tok-xyz") + asyncio.run( + client.create_cron( + name="n", schedule="* * * * *", skill="the-skill", + state={"k": "v"}, + ) + ) + + cmd = captured["cmd"] + # prefix + assert cmd[0] == "docker" + assert cmd[1:4] == ["exec", "openclaw-eval", "openclaw"] + # subcommand + add + assert "cron" in cmd and "add" in cmd + # key flags + assert any(s.startswith("--cron=") or cmd[cmd.index("--cron") + 1] == "* * * * *" for s in cmd) + # token is last + assert cmd[-1] == "--token=tok-xyz" + # --message holds the skill (current implementation quirk: message, not --skill) + assert any(s.startswith("--message=the-skill") for s in cmd) + # description carries state + assert any(s.startswith("--description=state:") for s in cmd) + + +def test_list_crons_parses_json(monkeypatch: pytest.MonkeyPatch): + async def fake_exec(*cmd, **kwargs): + payload = json.dumps([{"id": "a", "name": "n", "schedule": "* * * * *", "enabled": True}]).encode() + return _make_process(payload) + + monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_exec) + crons = asyncio.run(OpenClawClient().list_crons()) + assert [c.id for c in crons] == ["a"] + + +def test_list_crons_invalid_json_raises(monkeypatch: pytest.MonkeyPatch): + async def fake_exec(*cmd, **kwargs): + return _make_process(b"not json {") + + monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_exec) + with pytest.raises(RuntimeError, match="parse"): + asyncio.run(OpenClawClient().list_crons()) + + +def test_delete_cron_command_uses_rm(monkeypatch: pytest.MonkeyPatch): + captured = {} + + async def fake_exec(*cmd, **kwargs): + captured["cmd"] = list(cmd) + return _make_process(b"") + + monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_exec) + asyncio.run(OpenClawClient().delete_cron("cron-99")) + assert "cron" in captured["cmd"] + assert "rm" in captured["cmd"] + assert "cron-99" in captured["cmd"] + assert captured["cmd"][-1].startswith("--token=") + + +# --------------------------------------------------------------------------- +# T6 — Alert webhook semantics +# --------------------------------------------------------------------------- + + +def _make_alert(db_session: Session) -> AlertHistoryDB: + import uuid + a = AlertHistoryDB( + id=str(uuid.uuid4()), + alert_type="test", severity="warning", rule_name="r", + message="m", metric_value=1.0, threshold=0.5, + created_at=utc_now(), + ) + db_session.add(a) + db_session.commit() + db_session.refresh(a) + return a + + +def test_webhook_marked_sent_on_2xx(db_session: Session): + alert = _make_alert(db_session) + fake_resp = MagicMock(); fake_resp.raise_for_status = MagicMock() + with patch("agenteval.intelligent_eval.alerts.httpx.post", return_value=fake_resp) as p: + AlertManager(db_session, webhook_url="http://x")._send_webhook(alert) + assert p.called + db_session.refresh(alert) + assert alert.webhook_sent is True + + +def test_webhook_failure_does_not_raise_or_mark_sent(db_session: Session): + alert = _make_alert(db_session) + with patch("agenteval.intelligent_eval.alerts.httpx.post", side_effect=RuntimeError("boom")): + # Must not raise. + AlertManager(db_session, webhook_url="http://x")._send_webhook(alert) + db_session.refresh(alert) + assert alert.webhook_sent is False + + +@pytest.mark.xfail( + reason=( + "Known gap: webhook has no retry. _send_webhook swallows the first " + "failure and leaves webhook_sent=False; no follow-up attempt is made. " + "Tracked in .scratch/v111-architecture-scan.md." + ), + strict=False, +) +def test_webhook_retries_on_failure(db_session: Session): + alert = _make_alert(db_session) + fake_resp = MagicMock(); fake_resp.raise_for_status = MagicMock() + with patch( + "agenteval.intelligent_eval.alerts.httpx.post", + side_effect=[RuntimeError("transient"), fake_resp], + ) as p: + AlertManager(db_session, webhook_url="http://x")._send_webhook(alert) + assert p.call_count >= 2 + db_session.refresh(alert) + assert alert.webhook_sent is True + + +@pytest.mark.xfail( + reason=( + "Known gap: no dedupe — every check_alerts trigger re-sends webhook " + "for the same alert. Tracked in .scratch/v111-architecture-scan.md." + ), + strict=False, +) +def test_webhook_dedupes_repeat_triggers(db_session: Session): + """Calling _send_webhook twice on the same alert must POST at most once.""" + alert = _make_alert(db_session) + fake_resp = MagicMock(); fake_resp.raise_for_status = MagicMock() + with patch("agenteval.intelligent_eval.alerts.httpx.post", return_value=fake_resp) as p: + mgr = AlertManager(db_session, webhook_url="http://x") + mgr._send_webhook(alert) + mgr._send_webhook(alert) + assert p.call_count == 1