Agent/tests/test_router_api.py

"""测试 Router 核心接口

用法：
    uv run python tests/test_router_api.py                               # 只跑 health check
    uv run python tests/test_router_api.py --search                      # 搜索工具列表
    uv run python tests/test_router_api.py --search image                # 关键词搜索
    uv run python tests/test_router_api.py --status                      # 工具运行状态
    uv run python tests/test_router_api.py --select image_stitcher       # 调用指定工具
    uv run python tests/test_router_api.py --stitch                      # 测试图片拼接
    uv run python tests/test_router_api.py --create                      # 默认任务
    uv run python tests/test_router_api.py --create image_stitcher       # 指定任务文件
"""

import argparse
import base64
import json
import sys
import time
from pathlib import Path

import httpx

BASE_URL = "http://43.106.118.91:8001"
TASKS_DIR = Path(__file__).parent / "tasks"
TEST_IMAGES_DIR = TASKS_DIR / "stitcher_images"
OUTPUT_DIR = Path(__file__).parent / "output"


def check_connection():
    try:
        httpx.get(f"{BASE_URL}/health", timeout=3)
    except httpx.ConnectError:
        print(f"ERROR: Cannot connect to {BASE_URL}")
        print("Please start the service first:")
        print("  uv run python -m tool_agent")
        sys.exit(1)


def test_health():
    print("=== Health Check ===")
    resp = httpx.get(f"{BASE_URL}/health")
    print(f"  Status : {resp.status_code}")
    print(f"  Body   : {json.dumps(resp.json(), ensure_ascii=False, indent=4)}")
    assert resp.status_code == 200
    print("  [PASS]")


def test_search_tools(keyword: str = None):
    print(f"=== Search Tools{f' (keyword={keyword!r})' if keyword else ''} ===")
    payload = {"keyword": keyword} if keyword else {}
    resp = httpx.post(f"{BASE_URL}/search_tools", json=payload)
    print(f"  Status : {resp.status_code}")
    if resp.status_code != 200:
        print(f"  Body   : {resp.text}")
        print("  [FAIL]")
        return
    data = resp.json()
    print(f"  Total  : {data['total']}")
    for t in data["tools"]:
        print(f"\n  [{t['tool_id']}]")
        print(f"    name          : {t['name']}")
        print(f"    category      : {t.get('category', '')}")
        print(f"    state         : {t['state']}")
        print(f"    runtime       : {t.get('runtime_type', '')}  host_dir={t.get('host_dir', '')}")
        print(f"    endpoint      : {t.get('http_method', '')} {t.get('endpoint_path', '')}  port={t.get('port')}")
        print(f"    stream_support: {t.get('stream_support', False)}")
        print(f"    description   : {t.get('description', '')}")
        print(f"    params ({len(t.get('params', []))}):")
        for p in t.get("params", []):
            req_mark = "*" if p["required"] else " "
            default_str = f"  default={p['default']}" if p.get("default") is not None else ""
            enum_str = f"  enum={p['enum']}" if p.get("enum") else ""
            print(f"      {req_mark} {p['name']:<25} {p['type']:<12} {p.get('description', '')}{default_str}{enum_str}")
        if t.get("output_schema"):
            out_props = t["output_schema"].get("properties", {})
            print(f"    output ({len(out_props)}):")
            for oname, odef in out_props.items():
                print(f"        {oname:<25} {odef.get('type', ''):<12} {odef.get('description', '')}")
    print("\n  [PASS]")


def test_tools_status():
    print("=== Tools Status ===")
    resp = httpx.get(f"{BASE_URL}/tools/status")
    print(f"  Status : {resp.status_code}")
    data = resp.json()
    print(f"  Total  : {len(data['tools'])}")
    for t in data["tools"]:
        print(f"    - {t['tool_id']}")
        print(f"        state  : {t['state']}")
        print(f"        port   : {t.get('port')}")
        print(f"        pid    : {t.get('pid')}")
        print(f"        sources: {[s['type'] for s in t.get('sources', [])]}")
        if t.get("last_error"):
            print(f"        error  : {t['last_error']}")
    print("  [PASS]")


def test_select_tool(tool_id: str):
    print(f"=== Select Tool (tool_id={tool_id!r}) ===")
    resp = httpx.post(f"{BASE_URL}/select_tool", json={
        "tool_id": tool_id,
        "params": {}
    }, timeout=30)
    print(f"  Status : {resp.status_code}")
    data = resp.json()
    print(f"  Result :")
    print(f"    status: {data.get('status')}")
    if data.get("error"):
        print(f"    error : {data['error']}")
    else:
        result_str = json.dumps(data.get("result"), ensure_ascii=False, indent=6)
        print(f"    result: {result_str[:500]}")
    print("  [PASS]")


def test_stitch_images():
    print("=== Test Image Stitcher ===")
    if not TEST_IMAGES_DIR.exists():
        print(f"  ERROR: Test images directory not found: {TEST_IMAGES_DIR}")
        print("  [SKIP]")
        return

    image_files = sorted(TEST_IMAGES_DIR.glob("*.png"))
    if len(image_files) < 2:
        print(f"  ERROR: Need at least 2 images, found {len(image_files)}")
        print("  [SKIP]")
        return

    print(f"  Images : {len(image_files)} found")
    images_b64 = []
    for img_path in image_files[:6]:
        with open(img_path, "rb") as f:
            images_b64.append(base64.b64encode(f.read()).decode())
        print(f"    - {img_path.name}")

    print(f"  Calling image_stitcher (grid, 2 columns)...")
    try:
        resp = httpx.post(f"{BASE_URL}/select_tool", json={
            "tool_id": "image_stitcher",
            "params": {
                "images": images_b64,
                "direction": "grid",
                "columns": 2,
                "spacing": 10,
                "background_color": "#FFFFFF",
            }
        }, timeout=60)
        print(f"  Status : {resp.status_code}")
        data = resp.json()
        if data["status"] == "success":
            result = data["result"]
            print(f"  Result :")
            print(f"    width : {result['width']}")
            print(f"    height: {result['height']}")
            OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
            output_path = OUTPUT_DIR / "stitched_result.png"
            with open(output_path, "wb") as f:
                f.write(base64.b64decode(result["image"]))
            print(f"    saved : {output_path}")
            print("  [PASS]")
        else:
            print(f"  ERROR : {data.get('error', 'unknown')}")
            print("  [FAIL]")
    except httpx.TimeoutException:
        print("  ERROR : Request timeout")
        print("  [FAIL]")
    except Exception as e:
        print(f"  ERROR : {e}")
        print("  [FAIL]")


def load_task_spec(task_name: str) -> dict:
    task_file = TASKS_DIR / f"{task_name}.json"
    if not task_file.exists():
        print(f"  ERROR: Task file not found: {task_file}")
        print("  Available tasks:")
        if TASKS_DIR.exists():
            for f in TASKS_DIR.glob("*.json"):
                print(f"    - {f.stem}")
        sys.exit(1)
    with open(task_file, "r", encoding="utf-8") as f:
        return json.load(f)


def test_create_tool(task_name: str = None):
    print(f"=== Create Tool{f' (task={task_name!r})' if task_name else ''} ===")
    if task_name:
        task_data = load_task_spec(task_name)
        print(f"  File       : tests/tasks/{task_name}.json")
        print(f"  Description: {task_data['description'][:80]}")
    else:
        task_data = {"description": "创建一个简单的文本计数工具，输入文本，返回字数和字符数"}
        print(f"  Description: {task_data['description']}")

    resp = httpx.post(f"{BASE_URL}/create_tool", json=task_data)
    data = resp.json()
    task_id = data["task_id"]
    print(f"  Task ID    : {task_id}")
    print(f"  Status     : {data['status']}")
    assert data["status"] == "pending"
    print("  [SUBMITTED]")

    print(f"\n  Polling task {task_id} (timeout 10min)...")
    for i in range(120):
        time.sleep(5)
        resp = httpx.get(f"{BASE_URL}/tasks/{task_id}", timeout=30)
        task = resp.json()
        status = task["status"]

        if i % 6 == 0:
            print(f"    [{i*5}s] status={status}")

        if status == "completed":
            print(f"\n  Completed!")
            print(f"  Result     : {str(task.get('result', ''))[:300]}")
            resp2 = httpx.post(f"{BASE_URL}/search_tools", json={})
            tools = resp2.json()["tools"]
            print(f"  Registered : {[t['tool_id'] for t in tools]}")
            print("  [PASS]")
            return

        if status == "failed":
            print(f"\n  Failed!")
            print(f"  Error      : {task.get('error', 'unknown')}")
            print("  [FAIL]")
            return

    print(f"\n  Timeout after 600s")
    print("  [TIMEOUT]")


def main():
    parser = argparse.ArgumentParser(description="Router API Test")
    parser.add_argument("--search", nargs="?", const="", metavar="KEYWORD",
                        help="search tools, optional keyword")
    parser.add_argument("--status", action="store_true",
                        help="show tools status")
    parser.add_argument("--select", metavar="TOOL_ID",
                        help="call a tool by tool_id")
    parser.add_argument("--stitch", action="store_true",
                        help="test image stitcher with sample images")
    parser.add_argument("--create", nargs="?", const="", metavar="TASK_NAME",
                        help="create tool, optional task file name")
    args = parser.parse_args()

    print(f"Target: {BASE_URL}\n")
    check_connection()

    # 始终跑 health check
    test_health()

    ran_any = False

    if args.search is not None:
        print()
        test_search_tools(args.search or None)
        ran_any = True

    if args.status:
        print()
        test_tools_status()
        ran_any = True

    if args.select:
        print()
        test_select_tool(args.select)
        ran_any = True

    if args.stitch:
        print()
        test_stitch_images()
        ran_any = True

    if args.create is not None:
        print()
        test_create_tool(args.create or None)
        ran_any = True

    if not ran_any:
        print()
        print("No test specified. Available options:")
        parser.print_help()

    print("\n=== DONE ===")


if __name__ == "__main__":
    main()