交付物: 测试套件完成(15/15通过) + 清单更新

- 创建tests/test_health_checks.py: 15个测试覆盖Tier1/Tier2/Registry/Watchdog/Executor
- 交付物清单: 测试套件❌→✅(已完成)
- 测试内容: pid/port/http/todo/service_registry/watchdog各维度
- 运行方式: pytest tests/test_health_checks.py -v

tests/test_health_checks.py

@@ -0,0 +1,118 @@
+"""
+测试套件 — 健康检查组件
+验证 agents_health_check.py (Tier1) 和 agents_daily_health.py (Tier2) 的核心函数。
+
+运行: pytest tests/test_health_checks.py -v
+"""
+import os, sys, json, subprocess, tempfile
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "gateway", "scripts"))
+
+# ─── Tier 1 健康检查 ─────────────────────────────────────
+def test_tier1_imports():
+    """可以导入 agents_health_check"""
+    import agents_health_check
+    assert hasattr(agents_health_check, "pid_alive")
+    assert hasattr(agents_health_check, "port_open")
+    assert hasattr(agents_health_check, "http_check")
+
+def test_tier1_pid_alive():
+    """pid_alive 对当前进程应返回 True"""
+    import agents_health_check as hc
+    assert hc.pid_alive(os.getpid()) == True
+
+def test_tier1_pid_dead():
+    """pid_alive 对不存在 PID 应返回 False"""
+    import agents_health_check as hc
+    assert hc.pid_alive(999999999) == False
+
+def test_tier1_port_open_5803():
+    """port_open 对 dashboard 端口(5803)应返回 True"""
+    import agents_health_check as hc
+    assert hc.port_open(5803) == True
+
+def test_tier1_port_closed():
+    """port_open 对未使用端口应返回 False"""
+    import agents_health_check as hc
+    assert hc.port_open(51999) == False
+
+def test_tier1_http_xmpp():
+    """http_check 对 xmpp_bot /health 应返回 200"""
+    import agents_health_check as hc
+    ok, msg = hc.http_check("http://127.0.0.1:5802/health")
+    assert ok, f"xmpp_bot /health 应返回 200, 实际: {msg}"
+
+def test_tier1_http_ap():
+    """http_check 对 article_processor /health 应返回 200"""
+    import agents_health_check as hc
+    ok, msg = hc.http_check("http://127.0.0.1:5810/health")
+    assert ok, f"AP /health 应返回 200, 实际: {msg}"
+
+def test_tier1_http_dashboard():
+    """http_check 对 dashboard /health 应返回 200"""
+    import agents_health_check as hc
+    ok, msg = hc.http_check("http://127.0.0.1:5803/api/health")
+    assert ok, f"Dashboard /health 应返回 200, 实际: {msg}"
+
+def test_tier1_write_todo():
+    """write_todo 应写入 JSONL 文件"""
+    import agents_health_check as hc
+    test_file = os.path.join(os.path.dirname(hc.TODO_FILE), "test_todo.jsonl")
+    # 使用临时的 TODO 文件路径
+    hc.TODO_FILE = test_file
+    try:
+        hc.write_todo("test_service", "test issue", "/tmp/test.py", [], "/tmp")
+        assert os.path.exists(test_file)
+        with open(test_file, "r") as f:
+            line = f.readline().strip()
+            entry = json.loads(line)
+            assert entry["service"] == "test_service"
+            assert entry["status"] == "pending"
+    finally:
+        if os.path.exists(test_file):
+            os.remove(test_file)
+
+# ─── Tier 2 每日健康检查 ─────────────────────────────────
+def test_tier2_imports():
+    """可以导入 agents_daily_health"""
+    import agents_daily_health
+    assert hasattr(agents_daily_health, "pid_alive")
+    assert hasattr(agents_daily_health, "port_open")
+
+def test_tier2_pid_alive():
+    """pid_alive 对当前进程应返回 True"""
+    import agents_daily_health as hc
+    assert hc.pid_alive(os.getpid()) == True
+
+def test_tier2_http_check():
+    """http_check 应正确处理 200 和 401"""
+    import agents_daily_health as hc
+    ok, msg = hc.http_check("http://127.0.0.1:5802/health", accept_401=False)
+    assert ok, f"xmpp_bot /health 应返回 200, 实际: {msg}"
+
+# ─── Service Registry ────────────────────────────────────
+def test_service_registry():
+    """service_registry 应包含所有注册服务"""
+    import service_registry
+    names = [s["name"] for s in service_registry.SERVICES]
+    assert "xmpp_bot" in names
+    assert "article_processor" in names
+    assert "dashboard" in names
+    assert len(service_registry.SERVICES) >= 3
+
+# ─── 看门狗 ─────────────────────────────────────────────
+def test_watchdog_helpers():
+    """watchdog 的辅助函数应可用"""
+    import xmpp_watchdog as wd
+    assert hasattr(wd, "is_process_alive")
+    assert hasattr(wd, "is_port_listening")
+    assert hasattr(wd, "check_http_health")
+    # is_port_listening 对已知端口应返回 True
+    assert wd.is_port_listening(5803) == True
+
+# ─── TODO Executor ───────────────────────────────────────
+def test_executor_imports():
+    """self_todo_executor 应可导入"""
+    import self_todo_executor
+    assert hasattr(self_todo_executor, "read_pending_todos")
+    assert hasattr(self_todo_executor, "mark_todo")
+    assert hasattr(self_todo_executor, "execute_fix")