304 lines
12 KiB
Python
304 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DATA_ROOT = ROOT / ".data_gate3"
|
|
|
|
SERVICE_SPECS = [
|
|
{"name": "auth", "module": "services.auth_service.app:app", "port": 28001},
|
|
{"name": "audit", "module": "services.audit_service.app:app", "port": 28002},
|
|
{"name": "customer", "module": "services.customer_service.app:app", "port": 28003},
|
|
{"name": "interaction", "module": "services.interaction_service.app:app", "port": 28004},
|
|
{"name": "routing", "module": "services.routing_service.app:app", "port": 28005},
|
|
{"name": "voice", "module": "services.voice_adapter_service.app:app", "port": 28006},
|
|
{"name": "telegram", "module": "services.telegram_adapter_service.app:app", "port": 28007},
|
|
{"name": "kb", "module": "services.kb_service.app:app", "port": 28008},
|
|
{"name": "reporting", "module": "services.reporting_service.app:app", "port": 28009},
|
|
{"name": "supervisor", "module": "services.supervisor_service.app:app", "port": 28010},
|
|
{"name": "gateway", "module": "gateway.app:app", "port": 28080},
|
|
]
|
|
|
|
REQUIRED_STAGE3_SERVICES = ["kb", "reporting", "supervisor"]
|
|
|
|
|
|
async def wait_for_health(base_url: str, retries: int = 80, delay: float = 0.25) -> None:
|
|
async with httpx.AsyncClient(timeout=2) as client:
|
|
for _ in range(retries):
|
|
try:
|
|
response = await client.get(f"{base_url}/health")
|
|
if response.status_code == 200:
|
|
return
|
|
except Exception:
|
|
pass
|
|
await asyncio.sleep(delay)
|
|
raise RuntimeError(f"Service not healthy: {base_url}/health")
|
|
|
|
|
|
def start_one_service(spec: dict[str, Any], database_url: str | None, data_dir: Path) -> subprocess.Popen:
|
|
env = os.environ.copy()
|
|
env["CC_DATA_DIR"] = str(data_dir)
|
|
if database_url:
|
|
env["DATABASE_URL"] = database_url
|
|
|
|
if spec["name"] == "gateway":
|
|
env["AUTH_SERVICE_URL"] = "http://127.0.0.1:28001"
|
|
env["AUDIT_SERVICE_URL"] = "http://127.0.0.1:28002"
|
|
env["CUSTOMER_SERVICE_URL"] = "http://127.0.0.1:28003"
|
|
env["INTERACTION_SERVICE_URL"] = "http://127.0.0.1:28004"
|
|
env["ROUTING_SERVICE_URL"] = "http://127.0.0.1:28005"
|
|
env["VOICE_ADAPTER_SERVICE_URL"] = "http://127.0.0.1:28006"
|
|
env["TELEGRAM_ADAPTER_SERVICE_URL"] = "http://127.0.0.1:28007"
|
|
env["KB_SERVICE_URL"] = "http://127.0.0.1:28008"
|
|
env["REPORTING_SERVICE_URL"] = "http://127.0.0.1:28009"
|
|
env["SUPERVISOR_SERVICE_URL"] = "http://127.0.0.1:28010"
|
|
|
|
cmd = [
|
|
sys.executable,
|
|
"-m",
|
|
"uvicorn",
|
|
spec["module"],
|
|
"--host",
|
|
"127.0.0.1",
|
|
"--port",
|
|
str(spec["port"]),
|
|
]
|
|
return subprocess.Popen(
|
|
cmd,
|
|
cwd=str(ROOT),
|
|
env=env,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
|
|
|
|
async def start_services_sequential(database_url: str | None, data_dir: Path) -> list[subprocess.Popen]:
|
|
data_dir.mkdir(parents=True, exist_ok=True)
|
|
processes: list[subprocess.Popen] = []
|
|
for spec in SERVICE_SPECS:
|
|
proc = start_one_service(spec, database_url, data_dir)
|
|
processes.append(proc)
|
|
await wait_for_health(f"http://127.0.0.1:{spec['port']}", retries=60, delay=0.2)
|
|
return processes
|
|
|
|
|
|
def stop_services(processes: list[subprocess.Popen]) -> None:
|
|
for proc in processes:
|
|
if proc.poll() is not None:
|
|
continue
|
|
try:
|
|
proc.terminate()
|
|
except Exception:
|
|
pass
|
|
|
|
for proc in processes:
|
|
try:
|
|
proc.wait(timeout=2)
|
|
except Exception:
|
|
try:
|
|
proc.kill()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
async def run_gate3_checks(base_url: str, reporting_url: str | None = None) -> bool:
|
|
contract_path = ROOT / "contracts" / "openapi" / "stage3-kb-reporting-supervisor.yaml"
|
|
if not contract_path.exists():
|
|
print(f"[FAIL] Missing Stage 3 OpenAPI: {contract_path}")
|
|
return False
|
|
print(f"[PASS] Stage 3 OpenAPI present: {contract_path.name}")
|
|
|
|
operator = {"X-User": "operator", "X-Role": "operator"}
|
|
analyst = {"X-User": "analyst", "X-Role": "analyst"}
|
|
|
|
async with httpx.AsyncClient(base_url=base_url, timeout=10) as client:
|
|
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
|
|
services = registry.json().get("services", {})
|
|
missing = [name for name in REQUIRED_STAGE3_SERVICES if name not in services]
|
|
if missing:
|
|
print(f"[FAIL] registry missing Stage 3 services: {missing}")
|
|
return False
|
|
print("[PASS] Gateway and registry checks")
|
|
|
|
denied = await client.post(
|
|
"/proxy/kb/knowledge/categories",
|
|
headers=operator,
|
|
json={"name": "Forbidden", "description": "Role check"},
|
|
)
|
|
if denied.status_code != 403:
|
|
print(f"[FAIL] KB role check expected 403, got {denied.status_code}")
|
|
return False
|
|
|
|
category = await client.post(
|
|
"/proxy/kb/knowledge/categories",
|
|
headers=analyst,
|
|
json={"name": "Stage 3 KB", "description": "Gate 3"},
|
|
)
|
|
if category.status_code != 200:
|
|
print(f"[FAIL] KB category create status={category.status_code}")
|
|
return False
|
|
category_id = category.json()["category_id"]
|
|
|
|
article = await client.post(
|
|
"/proxy/kb/knowledge/articles",
|
|
headers=analyst,
|
|
json={
|
|
"category_id": category_id,
|
|
"title": "Install helper",
|
|
"body": "Step-by-step instruction for operator",
|
|
"tags": ["stage3", "kb"],
|
|
},
|
|
)
|
|
if article.status_code != 200:
|
|
print(f"[FAIL] KB article create status={article.status_code}")
|
|
return False
|
|
|
|
search = await client.get("/proxy/kb/knowledge/search?q=helper")
|
|
if search.status_code != 200 or len(search.json()) < 1:
|
|
print(
|
|
f"[FAIL] KB search failed status={search.status_code} count={len(search.json()) if search.status_code == 200 else 0}"
|
|
)
|
|
return False
|
|
print("[PASS] KB checks")
|
|
|
|
rows = [
|
|
{
|
|
"queue_id": "q_gate3",
|
|
"answered": True,
|
|
"wait_seconds": 15,
|
|
"handle_seconds": 95,
|
|
"abandoned": False,
|
|
"resolved_first_contact": True,
|
|
},
|
|
{
|
|
"queue_id": "q_gate3",
|
|
"answered": False,
|
|
"wait_seconds": 20,
|
|
"handle_seconds": 0,
|
|
"abandoned": True,
|
|
"resolved_first_contact": False,
|
|
},
|
|
]
|
|
for row in rows:
|
|
ingested = await client.post("/proxy/reporting/reports/events", json=row)
|
|
if ingested.status_code != 200:
|
|
print(f"[FAIL] reporting ingest status={ingested.status_code}")
|
|
return False
|
|
|
|
kpi = await client.get("/proxy/reporting/reports/kpi?queue_id=q_gate3&sl_threshold_seconds=30")
|
|
if kpi.status_code != 200:
|
|
print(f"[FAIL] reporting KPI status={kpi.status_code}")
|
|
return False
|
|
kpi_body = kpi.json()
|
|
if "kpi" not in kpi_body or not {"SL", "ASA", "AHT", "Abandon", "FCR"}.issubset(set(kpi_body["kpi"])):
|
|
print("[FAIL] KPI payload missing required keys")
|
|
return False
|
|
print("[PASS] Reporting KPI checks")
|
|
|
|
report_service_url = reporting_url or services.get("reporting")
|
|
if not report_service_url:
|
|
print("[FAIL] reporting service URL not found")
|
|
return False
|
|
|
|
async with httpx.AsyncClient(base_url=report_service_url, timeout=10) as reporting_client:
|
|
exported = await reporting_client.get("/reports/export")
|
|
if exported.status_code != 200:
|
|
print(f"[FAIL] reporting export status={exported.status_code}")
|
|
return False
|
|
if "queue_id,answered,wait_seconds,handle_seconds,abandoned,resolved_first_contact,created_at" not in exported.text:
|
|
print("[FAIL] reporting export missing CSV header")
|
|
return False
|
|
if "q_gate3" not in exported.text:
|
|
print("[FAIL] reporting export missing inserted data")
|
|
return False
|
|
print("[PASS] Reporting CSV export checks")
|
|
|
|
up1 = await client.post(
|
|
"/proxy/supervisor/supervisor/agent-states",
|
|
json={"agent_id": "a_gate3_1", "state": "READY", "queue_id": "q_gate3"},
|
|
)
|
|
up2 = await client.post(
|
|
"/proxy/supervisor/supervisor/agent-states",
|
|
json={"agent_id": "a_gate3_2", "state": "BUSY", "queue_id": "q_gate3"},
|
|
)
|
|
queue = await client.post(
|
|
"/proxy/supervisor/supervisor/queue-metrics?queue_id=q_gate3&in_queue=3&avg_wait_seconds=21"
|
|
)
|
|
if up1.status_code != 200 or up2.status_code != 200 or queue.status_code != 200:
|
|
print(
|
|
f"[FAIL] supervisor updates status: agent1={up1.status_code}, agent2={up2.status_code}, queue={queue.status_code}"
|
|
)
|
|
return False
|
|
|
|
realtime = await client.get("/proxy/supervisor/supervisor/realtime")
|
|
if realtime.status_code != 200:
|
|
print(f"[FAIL] supervisor realtime status={realtime.status_code}")
|
|
return False
|
|
body: dict[str, Any] = realtime.json()
|
|
if body.get("agents", {}).get("total", 0) < 2:
|
|
print("[FAIL] supervisor realtime has less than 2 agents")
|
|
return False
|
|
queues = body.get("queues", [])
|
|
if not any(item.get("queue_id") == "q_gate3" for item in queues):
|
|
print("[FAIL] supervisor realtime missing q_gate3 queue snapshot")
|
|
return False
|
|
print("[PASS] Supervisor realtime checks")
|
|
|
|
print("[PASS] Gate 3 checks completed")
|
|
return True
|
|
|
|
|
|
async def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Gate 3 checker")
|
|
parser.add_argument("--base-url", default="http://localhost:8080", help="Gateway base URL")
|
|
parser.add_argument("--auto-start", action="store_true", help="Auto-start local services for the check")
|
|
parser.add_argument(
|
|
"--database-url",
|
|
default=os.getenv("DATABASE_URL", ""),
|
|
help="Optional DB URL for auto-start mode",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
if not args.auto_start:
|
|
ok = await run_gate3_checks(args.base_url)
|
|
sys.exit(0 if ok else 1)
|
|
|
|
db_url = args.database_url.strip() or None
|
|
run_data_dir = DATA_ROOT / f"run_{int(time.time() * 1000)}"
|
|
processes: list[subprocess.Popen] = []
|
|
try:
|
|
processes = await start_services_sequential(db_url, run_data_dir)
|
|
ok = await run_gate3_checks("http://127.0.0.1:28080", reporting_url="http://127.0.0.1:28009")
|
|
if db_url:
|
|
print(f"[INFO] DB mode: {db_url}")
|
|
sys.exit(0 if ok else 1)
|
|
finally:
|
|
stop_services(processes)
|
|
try:
|
|
shutil.rmtree(run_data_dir)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|