multiAgentTry/test_swarm_frontend.py

"""
Swarm 前端完整测试
测试所有页面功能和 API
"""
from playwright.sync_api import sync_playwright
import json
import time

BASE_URL = "http://localhost:3000"
API_BASE = "http://localhost:8000/api"

def test_api_endpoints():
    """测试后端 API 端点"""
    print("\n=== 测试后端 API ===")
    import urllib.request

    endpoints = [
        "/health",
        "/agents/",
        "/locks/",
        "/heartbeats/",
        "/workflows/list",
        "/workflows/files",
        "/status",
        "/humans/summary",
    ]

    results = {}
    for endpoint in endpoints:
        try:
            url = f"{API_BASE}{endpoint}"
            with urllib.request.urlopen(url, timeout=5) as response:
                data = response.read().decode('utf-8')
                results[endpoint] = {"status": response.status, "data": data[:100]}
                print(f"  ✓ {endpoint}: {response.status}")
        except Exception as e:
            results[endpoint] = {"status": "error", "error": str(e)}
            print(f"  ✗ {endpoint}: {e}")

    # 测试 agents_control 端点
    control_endpoints = [
        "/api/agents/control/summary",
        "/api/agents/control/list",
    ]

    print("\n=== 测试 agents_control API ===")
    for endpoint in control_endpoints:
        try:
            url = f"http://localhost:8000{endpoint}"
            with urllib.request.urlopen(url, timeout=5) as response:
                data = response.read().decode('utf-8')
                results[endpoint] = {"status": response.status}
                print(f"  ✓ {endpoint}: {response.status}")
        except Exception as e:
            results[endpoint] = {"status": "error", "error": str(e)}
            print(f"  ✗ {endpoint}: {e}")

    return results

def test_frontend_pages():
    """测试前端页面"""
    print("\n=== 测试前端页面 ===")

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=False)
        context = browser.new_context()
        page = context.new_page()

        # 收集控制台错误
        console_errors = []
        def on_console(msg):
            if msg.type == "error":
                console_errors.append(msg.text)

        page.on("console", on_console)

        issues = {}

        try:
            # 1. Dashboard 页面
            print("\n1. 测试 Dashboard 页面...")
            page.goto(f"{BASE_URL}/")
            page.wait_for_load_state("networkidle", timeout=10000)
            page.screenshot(path="D:/ccProg/swarm/test_screenshots/dashboard.png")

            dashboard_errors = [e for e in console_errors if "dashboard" in e.lower()]
            issues["dashboard"] = {
                "url": page.url,
                "title": page.title(),
                "console_errors": dashboard_errors,
                "has_content": len(page.content()) > 1000
            }
            print(f"  标题: {page.title()}")
            print(f"  控制台错误数: {len(dashboard_errors)}")

            # 2. Agents 页面
            print("\n2. 测试 Agents 页面...")
            page.goto(f"{BASE_URL}/#/agents")
            page.wait_for_load_state("networkidle", timeout=10000)
            time.sleep(2)  # 等待动态内容加载
            page.screenshot(path="D:/ccProg/swarm/test_screenshots/agents.png")

            # 检查 agents/control API 调用
            agents_errors = [e for e in console_errors if "agent" in e.lower()]
            issues["agents"] = {
                "url": page.url,
                "console_errors": agents_errors,
                "has_agent_cards": page.locator('[class*="agent"]').count() > 0
            }
            print(f"  控制台错误数: {len(agents_errors)}")

            # 3. Meetings 页面
            print("\n3. 测试 Meetings 页面...")
            page.goto(f"{BASE_URL}/#/meetings")
            page.wait_for_load_state("networkidle", timeout=10000)
            time.sleep(2)
            page.screenshot(path="D:/ccProg/swarm/test_screenshots/meetings.png")

            meeting_errors = [e for e in console_errors if "meeting" in e.lower()]
            issues["meetings"] = {
                "url": page.url,
                "console_errors": meeting_errors
            }
            print(f"  控制台错误数: {len(meeting_errors)}")

            # 4. Resources 页面
            print("\n4. 测试 Resources 页面...")
            page.goto(f"{BASE_URL}/#/resources")
            page.wait_for_load_state("networkidle", timeout=10000)
            time.sleep(2)
            page.screenshot(path="D:/ccProg/swarm/test_screenshots/resources.png")

            resource_errors = [e for e in console_errors if "resource" in e.lower()]
            issues["resources"] = {
                "url": page.url,
                "console_errors": resource_errors
            }
            print(f"  控制台错误数: {len(resource_errors)}")

            # 5. Workflow 页面
            print("\n5. 测试 Workflow 页面...")
            page.goto(f"{BASE_URL}/#/workflow")
            page.wait_for_load_state("networkidle", timeout=10000)
            time.sleep(2)
            page.screenshot(path="D:/ccProg/swarm/test_screenshots/workflow.png")

            workflow_errors = [e for e in console_errors if "workflow" in e.lower()]
            issues["workflow"] = {
                "url": page.url,
                "console_errors": workflow_errors
            }
            print(f"  控制台错误数: {len(workflow_errors)}")

        except Exception as e:
            print(f"  错误: {e}")
            issues["error"] = str(e)

        # 输出所有控制台错误
        print("\n=== 所有控制台错误 ===")
        for error in console_errors:
            print(f"  - {error}")

        browser.close()

    return issues

if __name__ == "__main__":
    import os
    os.makedirs("D:/ccProg/swarm/test_screenshots", exist_ok=True)

    api_results = test_api_endpoints()
    frontend_issues = test_frontend_pages()

    # 保存结果
    with open("D:/ccProg/swarm/test_results.json", "w") as f:
        json.dump({
            "api": api_results,
            "frontend": frontend_issues
        }, f, indent=2)

    print("\n=== 测试结果已保存 ===")
    print("  截图: D:/ccProg/swarm/test_screenshots/")
    print("  结果: D:/ccProg/swarm/test_results.json")