#!/usr/bin/env python3 """TaskPilot 端到端自检(仅标准库)。 用法: uv run tools/verify.py uv run tools/verify.py --base-url http://127.0.0.1:3000 --keep 它会真实地:创建任务 → 立即运行 → 读回日志 → 校验中文输出与虚拟环境注入 → 校验调度与统计接口 → 删除任务(--keep 可保留)。 """ from __future__ import annotations import argparse import json import sys import time import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass, field from pathlib import Path TASK_ID = "verify_python_demo" @dataclass class Check: name: str ok: bool detail: str = "" def render(self) -> str: marker = "OK " if self.ok else "FAIL" suffix = f" {self.detail}" if self.detail else "" return f"[{marker}] {self.name}{suffix}" @dataclass class Api: base_url: str token: str | None = None checks: list[Check] = field(default_factory=list) def request( self, method: str, path: str, body: dict | None = None, authenticated: bool = False, ) -> tuple[int, object]: url = f"{self.base_url}{path}" data = None headers = {"Accept": "application/json"} if body is not None: data = json.dumps(body).encode("utf-8") headers["Content-Type"] = "application/json" if authenticated: headers["X-TaskPilot-Token"] = self.token or "" request = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(request, timeout=15) as response: raw = response.read() status = response.status except urllib.error.HTTPError as error: raw = error.read() status = error.code except urllib.error.URLError as error: raise SystemExit(f"无法连接 {url}: {error.reason}") from error if not raw: return status, None try: return status, json.loads(raw.decode("utf-8")) except json.JSONDecodeError: return status, raw.decode("utf-8", errors="replace") def check(self, name: str, ok: bool, detail: str = "") -> bool: self.checks.append(Check(name, ok, detail)) return ok def session(self) -> str: status, body = self.request("GET", "/api/v1/session") if status != 200 or not isinstance(body, dict): raise SystemExit(f"获取会话令牌失败: HTTP {status}") self.token = str(body["token"]) return self.token def poll_run(self, run_id: int, timeout: float) -> dict: deadline = time.monotonic() + timeout terminal = {"succeeded", "failed", "timed_out", "cancelled", "skipped"} while time.monotonic() < deadline: status, body = self.request("GET", f"/api/v1/runs/{run_id}") if status != 200 or not isinstance(body, dict): raise SystemExit(f"读取运行记录失败: HTTP {status}") if body.get("status") in terminal: return body time.sleep(0.2) raise SystemExit(f"运行 {run_id} 在 {timeout} 秒内未结束") def read_logs(self, run_id: int, timeout: float) -> str: deadline = time.monotonic() + timeout content = "" offset = 0 while time.monotonic() < deadline: status, body = self.request( "GET", f"/api/v1/runs/{run_id}/logs?offset={offset}" ) if status != 200 or not isinstance(body, dict): raise SystemExit(f"读取运行日志失败: HTTP {status}") content += str(body.get("content", "")) offset = int(body.get("next_offset", offset)) if body.get("complete"): return content time.sleep(0.1) return content def main() -> int: parser = argparse.ArgumentParser(description="TaskPilot end-to-end self check") parser.add_argument("--base-url", default="http://127.0.0.1:3000") parser.add_argument( "--working-dir", default=str(Path(__file__).resolve().parent.parent / "samples" / "python-task"), help="包含 .venv 的演示任务目录", ) parser.add_argument("--keep", action="store_true", help="检查完成后保留任务") parser.add_argument("--timeout", type=float, default=60.0) args = parser.parse_args() working_dir = Path(args.working_dir).resolve() api = Api(args.base_url.rstrip("/")) # 1. 健康检查 status, health = api.request("GET", "/api/v1/health") api.check( "health endpoint", status == 200 and isinstance(health, dict) and health.get("status") == "ok", f"HTTP {status}", ) # 2. 会话令牌 token = api.session() api.check("session token", bool(token), f"{len(token)} chars") # 3. 目录列举能看到演示目录下的 .venv query = urllib.parse.urlencode({"path": str(working_dir)}) status, listing = api.request("GET", f"/api/v1/system/directories?{query}") names = [entry["name"] for entry in (listing or {}).get("entries", [])] api.check( "directory listing sees .venv", status == 200 and ".venv" in names, f"entries={names}", ) # 4. 虚拟环境探测识别出 .venv\Scripts\python.exe status, probe = api.request( "POST", "/api/v1/python/venv", {"working_dir": str(working_dir), "interpreter": "python.exe", "venv": None}, authenticated=True, ) probe = probe if isinstance(probe, dict) else {} display_path = str(probe.get("display_path") or "") api.check( "venv probe links the demo environment", status == 200 and probe.get("status") == "linked" and display_path.endswith(r"Scripts\python.exe"), f"status={probe.get('status')} path={display_path}", ) # 5. 创建任务(启用 → 自动注册调度) payload = { "id": TASK_ID, "name": "Python 演示任务", "cron": "0 0 3 * * *", "timezone": "Asia/Shanghai", "enabled": True, "timeout_secs": 30, "action": { "type": "script", "working_dir": str(working_dir), "interpreter": "python.exe", "script": "main.py", "args": ["--lines", "2"], "env": {}, }, } api.request("DELETE", f"/api/v1/tasks/{TASK_ID}", authenticated=True) status, created = api.request( "POST", "/api/v1/tasks", payload, authenticated=True ) api.check("create task", status == 201, f"HTTP {status}") # 6. 调度接口给出下次触发时间 status, schedule = api.request("GET", "/api/v1/schedule") entries = schedule if isinstance(schedule, list) else [] entry = next((item for item in entries if item.get("task_id") == TASK_ID), None) api.check( "task is registered in the schedule", status == 200 and entry is not None and bool(entry.get("next_run_at")), f"next_run_at={(entry or {}).get('next_run_at')}", ) # 7. 立即运行并等待结束 status, run = api.request( "POST", f"/api/v1/tasks/{TASK_ID}/run", authenticated=True ) run = run if isinstance(run, dict) else {} run_id = int(run.get("id", 0)) api.check("run accepted", status == 202 and run_id > 0, f"run={run_id}") finished = api.poll_run(run_id, args.timeout) api.check( "run succeeded", finished.get("status") == "succeeded" and finished.get("exit_code") == 0, f"status={finished.get('status')} exit={finished.get('exit_code')}", ) # 8. 日志里能看到中文、虚拟环境注入与真实解释器路径 log = api.read_logs(run_id, args.timeout) api.check("log contains Chinese output", "中文输出" in log) api.check( "log shows the venv interpreter", ".venv" in log.lower() or "Scripts" in log, ) api.check( "log shows VIRTUAL_ENV injection", "VIRTUAL_ENV" in log and "<未设置>" not in log, ) api.check("log has no replacement characters", "\ufffd" not in log) # 9. 统计接口口径正确 status, stats = api.request("GET", "/api/v1/stats") stats = stats if isinstance(stats, dict) else {} api.check( "stats endpoint reports the task", status == 200 and int(stats.get("total", 0)) >= 1 and int(stats.get("running", -1)) == 0, f"total={stats.get('total')} enabled={stats.get('enabled')} running={stats.get('running')}", ) # 10. 清理 if args.keep: api.check("cleanup skipped (--keep)", True) else: status, _ = api.request( "DELETE", f"/api/v1/tasks/{TASK_ID}", authenticated=True ) api.check("delete task", status == 204, f"HTTP {status}") failures = [check for check in api.checks if not check.ok] print(f"\nTaskPilot self check against {args.base_url}\n") for check in api.checks: print(check.render()) print(f"\n{len(api.checks) - len(failures)}/{len(api.checks)} checks passed") if failures: print("\n失败的检查:") for check in failures: print(f" - {check.name}: {check.detail}") return 1 return 0 if __name__ == "__main__": sys.exit(main())