| """Test all Bee server endpoints.""" |
|
|
| import json |
| import httpx |
|
|
| BASE = "http://localhost:8000" |
| client = httpx.Client(timeout=30) |
|
|
|
|
| def test(method, path, body=None, expected=200): |
| try: |
| if method == "GET": |
| r = client.get(f"{BASE}{path}") |
| else: |
| r = client.post(f"{BASE}{path}", json=body) |
| status = "OK" if r.status_code == expected else f"FAIL({r.status_code})" |
| return status, r.json() if r.status_code < 500 else {} |
| except Exception as e: |
| return f"ERR({e})", {} |
|
|
|
|
| print("=" * 60) |
| print("BEE SERVER — ENDPOINT TESTS") |
| print("=" * 60) |
|
|
| endpoints = [ |
| ("GET", "/health", None), |
| ("GET", "/v1/models", None), |
| ("GET", "/v1/router/stats", None), |
| ("GET", "/v1/community/stats", None), |
| ("GET", "/v1/interactions", None), |
| ("GET", "/v1/evolution/status", None), |
| ("POST", "/v1/chat/completions", { |
| "messages": [{"role": "user", "content": "What is 2+2?"}], |
| "max_tokens": 50, |
| }), |
| ("POST", "/v1/domain/switch", {"domain": "programming"}), |
| ("POST", "/v1/domain/switch", {"domain": "quantum"}), |
| ("POST", "/v1/domain/switch", {"domain": "cybersecurity"}), |
| ("POST", "/v1/domain/switch", {"domain": "fintech"}), |
| ("POST", "/v1/domain/switch", {"domain": "general"}), |
| ] |
|
|
| passed = 0 |
| total = len(endpoints) |
| for method, path, body in endpoints: |
| status, data = test(method, path, body) |
| ok = status == "OK" |
| if ok: |
| passed += 1 |
| icon = "PASS" if ok else "FAIL" |
| print(f" [{icon}] {method:4s} {path}") |
|
|
| print(f"\n{passed}/{total} endpoints passed") |
| print("=" * 60) |
|
|