65 lines
1.7 KiB
Python
65 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
|
|
import httpx
|
|
|
|
|
|
REQUIRED_SERVICES = [
|
|
"auth",
|
|
"audit",
|
|
"customer",
|
|
"interaction",
|
|
"routing",
|
|
"voice",
|
|
"telegram",
|
|
"kb",
|
|
"reporting",
|
|
"supervisor",
|
|
]
|
|
|
|
|
|
async def check_gateway(base_url: str) -> bool:
|
|
async with httpx.AsyncClient(base_url=base_url, timeout=5) as client:
|
|
try:
|
|
health = await client.get("/health")
|
|
if health.status_code != 200:
|
|
print(f"[FAIL] gateway /health status={health.status_code}")
|
|
return False
|
|
|
|
registry = await client.get("/registry")
|
|
if registry.status_code != 200:
|
|
print(f"[FAIL] gateway /registry status={registry.status_code}")
|
|
return False
|
|
data = registry.json()
|
|
services = data.get("services", {})
|
|
missing = [s for s in REQUIRED_SERVICES if s not in services]
|
|
if missing:
|
|
print(f"[FAIL] registry missing services: {missing}")
|
|
return False
|
|
|
|
login = await client.post(
|
|
"/proxy/auth/auth/login",
|
|
json={"username": "admin", "password": "admin123"},
|
|
)
|
|
if login.status_code != 200:
|
|
print(f"[FAIL] auth login status={login.status_code}")
|
|
return False
|
|
|
|
print("[PASS] Stage 1 gateway/auth health checks passed")
|
|
return True
|
|
except Exception as exc:
|
|
print(f"[FAIL] exception: {exc}")
|
|
return False
|
|
|
|
|
|
async def main() -> None:
|
|
base_url = "http://localhost:8080"
|
|
ok = await check_gateway(base_url)
|
|
sys.exit(0 if ok else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|