from __future__ import annotations import argparse import asyncio import os from pathlib import Path import shutil import statistics import subprocess import sys import time from typing import Any import httpx ROOT = Path(__file__).resolve().parents[1] DATA_ROOT = ROOT / ".data_gate4" SERVICE_SPECS = [ {"name": "auth", "module": "services.auth_service.app:app", "port": 38001}, {"name": "audit", "module": "services.audit_service.app:app", "port": 38002}, {"name": "customer", "module": "services.customer_service.app:app", "port": 38003}, {"name": "interaction", "module": "services.interaction_service.app:app", "port": 38004}, {"name": "routing", "module": "services.routing_service.app:app", "port": 38005}, {"name": "voice", "module": "services.voice_adapter_service.app:app", "port": 38006}, {"name": "telegram", "module": "services.telegram_adapter_service.app:app", "port": 38007}, {"name": "kb", "module": "services.kb_service.app:app", "port": 38008}, {"name": "reporting", "module": "services.reporting_service.app:app", "port": 38009}, {"name": "supervisor", "module": "services.supervisor_service.app:app", "port": 38010}, {"name": "gateway", "module": "gateway.app:app", "port": 38080}, ] async def wait_for_health(base_url: str, retries: int = 60, delay: float = 0.2) -> 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], data_dir: Path) -> subprocess.Popen: env = os.environ.copy() env["CC_DATA_DIR"] = str(data_dir) if spec["name"] == "gateway": env["AUTH_SERVICE_URL"] = "http://127.0.0.1:38001" env["AUDIT_SERVICE_URL"] = "http://127.0.0.1:38002" env["CUSTOMER_SERVICE_URL"] = "http://127.0.0.1:38003" env["INTERACTION_SERVICE_URL"] = "http://127.0.0.1:38004" env["ROUTING_SERVICE_URL"] = "http://127.0.0.1:38005" env["VOICE_ADAPTER_SERVICE_URL"] = "http://127.0.0.1:38006" env["TELEGRAM_ADAPTER_SERVICE_URL"] = "http://127.0.0.1:38007" env["KB_SERVICE_URL"] = "http://127.0.0.1:38008" env["REPORTING_SERVICE_URL"] = "http://127.0.0.1:38009" env["SUPERVISOR_SERVICE_URL"] = "http://127.0.0.1:38010" 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(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, data_dir) processes.append(proc) await wait_for_health(f"http://127.0.0.1:{spec['port']}") 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 _create_interaction(client: httpx.AsyncClient, channel: str, idx: int) -> tuple[bool, float]: headers = {"X-User": "operator", "X-Role": "operator"} payload = { "channel": channel, "subject": f"Gate4 load {channel} #{idx}", "customer_id": None, "queue_id": "q_gate4", "priority": 3, } started = time.perf_counter() try: resp = await client.post("/proxy/interaction/interactions", headers=headers, json=payload, timeout=10) elapsed = time.perf_counter() - started return resp.status_code == 200, elapsed except Exception: elapsed = time.perf_counter() - started return False, elapsed async def run_load_wave(base_url: str, voice: int, digital: int) -> dict[str, float | int]: async with httpx.AsyncClient(base_url=base_url) as client: tasks = [] for i in range(voice): tasks.append(_create_interaction(client, "voice", i)) for i in range(digital): tasks.append(_create_interaction(client, "telegram", i)) results = await asyncio.gather(*tasks) ok_times = [t for ok, t in results if ok] failed = len(results) - len(ok_times) if ok_times: p95 = sorted(ok_times)[max(0, int(len(ok_times) * 0.95) - 1)] avg = statistics.mean(ok_times) else: p95 = 0.0 avg = 0.0 return { "total": len(results), "success": len(ok_times), "failed": failed, "avg_seconds": round(avg, 4), "p95_seconds": round(p95, 4), } async def get_interaction_count(base_url: str, limit: int = 1000) -> int: async with httpx.AsyncClient(base_url=base_url, timeout=10) as client: response = await client.get(f"/proxy/interaction/interactions?limit={limit}") response.raise_for_status() items = response.json() return len(items) async def run_security_smoke(base_url: str) -> bool: operator = {"X-User": "operator", "X-Role": "operator"} async with httpx.AsyncClient(base_url=base_url, timeout=10) as client: create_user = await client.post( "/proxy/auth/users", headers=operator, json={ "username": "sec_denied_user", "password": "secret123", "full_name": "Security Denied", "role": "operator", }, ) if create_user.status_code != 403: print(f"[FAIL] Security RBAC auth/users expected 403, got {create_user.status_code}") return False create_queue = await client.post( "/proxy/routing/queues", json={ "name": "Denied queue", "description": "Role check", "rules": [{"channel": "voice", "priority": 3, "strategy": "round_robin", "sla_seconds": 30}], }, ) if create_queue.status_code != 403: print(f"[FAIL] Security RBAC routing/queues expected 403, got {create_queue.status_code}") return False kb_create = await client.post( "/proxy/kb/knowledge/categories", headers=operator, json={"name": "Denied KB", "description": "Role check"}, ) if kb_create.status_code != 403: print(f"[FAIL] Security RBAC kb category expected 403, got {kb_create.status_code}") return False print("[PASS] Security smoke checks") return True def run_backup_restore(source_dir: Path) -> tuple[Path, Path]: backup_script = ROOT / "scripts" / "backup_data.ps1" restore_script = ROOT / "scripts" / "restore_data.ps1" backup_output = DATA_ROOT / f"backups_{int(time.time() * 1000)}" backup_output.mkdir(parents=True, exist_ok=True) backup_cmd = [ "powershell", "-ExecutionPolicy", "Bypass", "-File", str(backup_script), "-SourceDir", str(source_dir), "-OutputDir", str(backup_output), ] result = subprocess.run(backup_cmd, cwd=str(ROOT), capture_output=True, text=True, check=False) if result.returncode != 0: raise RuntimeError(f"Backup failed: {result.stdout}\n{result.stderr}") archives = sorted(backup_output.glob("*.zip"), key=lambda p: p.stat().st_mtime, reverse=True) if not archives: raise RuntimeError("Backup archive not created") archive = archives[0] if source_dir.exists(): shutil.rmtree(source_dir) restore_cmd = [ "powershell", "-ExecutionPolicy", "Bypass", "-File", str(restore_script), "-BackupZip", str(archive), "-TargetDir", str(source_dir), ] restore = subprocess.run(restore_cmd, cwd=str(ROOT), capture_output=True, text=True, check=False) if restore.returncode != 0: raise RuntimeError(f"Restore failed: {restore.stdout}\n{restore.stderr}") db_file = source_dir / "mvp_cc.db" if not db_file.exists(): raise RuntimeError(f"Restored DB file not found: {db_file}") return backup_output, archive def check_docs_bundle() -> bool: required = [ ROOT / "docs" / "security" / "checklist.md", ROOT / "docs" / "releases" / "v1.0.0-mvp.md", ROOT / "docs" / "gates" / "mvp-pilot-baseline.md", ROOT / "docs" / "gates" / "p1-p2-defects.md", ROOT / "docs" / "roadmap" / "05-wave2-backlog.md", ROOT / "docs" / "runbooks" / "backup-restore.md", ROOT / "docs" / "runbooks" / "load-test-plan.md", ROOT / "docs" / "runbooks" / "pilot-uat.md", ROOT / "docs" / "uat" / "README.md", ROOT / "docs" / "uat" / "scenario-checklist.md", ROOT / "docs" / "uat" / "session-template.md", ROOT / "docs" / "uat" / "signoff-template.md", ROOT / "docs" / "uat" / "defect-log-template.csv", ] missing = [str(path) for path in required if not path.exists()] if missing: print(f"[FAIL] Missing required Stage 4 docs: {missing}") return False print("[PASS] Stage 4 docs bundle present") return True async def main() -> None: parser = argparse.ArgumentParser(description="Gate 4 checker: load, backup/restore, security smoke") parser.add_argument("--voice", type=int, default=100, help="Concurrent voice requests") parser.add_argument("--digital", type=int, default=100, help="Concurrent digital requests") parser.add_argument("--keep-artifacts", action="store_true", help="Keep .data_gate4 run and backup artifacts") args = parser.parse_args() run_data_dir = DATA_ROOT / f"run_{int(time.time() * 1000)}" backup_output: Path | None = None processes: list[subprocess.Popen] = [] try: processes = await start_services_sequential(run_data_dir) base_url = "http://127.0.0.1:38080" load_result = await run_load_wave(base_url, args.voice, args.digital) print("[INFO] Load result:", load_result) if load_result["failed"] != 0 or load_result["success"] != load_result["total"]: print("[FAIL] Load target not reached with zero errors") sys.exit(1) print("[PASS] Load target reached") count_before = await get_interaction_count(base_url, limit=max(500, args.voice + args.digital + 100)) if count_before < (args.voice + args.digital): print( f"[FAIL] Interaction count before backup too low: {count_before} < {args.voice + args.digital}" ) sys.exit(1) print(f"[PASS] Interaction count before backup: {count_before}") security_ok = await run_security_smoke(base_url) if not security_ok: sys.exit(1) stop_services(processes) processes = [] backup_output, archive = run_backup_restore(run_data_dir) print(f"[PASS] Backup/restore scripts completed, archive: {archive}") processes = await start_services_sequential(run_data_dir) count_after = await get_interaction_count(base_url, limit=max(500, args.voice + args.digital + 100)) if count_after != count_before: print(f"[FAIL] Restored interaction count mismatch: before={count_before}, after={count_after}") sys.exit(1) print(f"[PASS] Restore data integrity validated: {count_after} interactions") if not check_docs_bundle(): sys.exit(1) print("[PASS] Gate 4 automated checks completed") finally: stop_services(processes) if not args.keep_artifacts: try: if run_data_dir.exists(): shutil.rmtree(run_data_dir) except Exception: pass try: if backup_output and backup_output.exists(): shutil.rmtree(backup_output) except Exception: pass if __name__ == "__main__": asyncio.run(main())