140 lines
4.5 KiB
Python
140 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import time
|
|
|
|
import httpx
|
|
from sqlalchemy import create_engine, text
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from services.shared.security import issue_app_token
|
|
|
|
|
|
def _load_env_file(path: str) -> None:
|
|
env_path = Path(path).expanduser().resolve()
|
|
if not env_path.exists() or not env_path.is_file():
|
|
return
|
|
for line in env_path.read_text(encoding="utf-8").splitlines():
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith("#") or "=" not in stripped:
|
|
continue
|
|
key, value = stripped.split("=", 1)
|
|
key = key.strip()
|
|
if key and key not in os.environ:
|
|
os.environ[key] = value.strip()
|
|
|
|
|
|
def _bool_env(name: str, default: bool) -> bool:
|
|
raw = os.getenv(name)
|
|
if raw is None:
|
|
return default
|
|
return raw.strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def _ops_headers() -> dict[str, str]:
|
|
if _bool_env("ALLOW_LEGACY_HEADER_AUTH", True):
|
|
return {"X-User": "admin", "X-Role": "admin"}
|
|
token = issue_app_token(
|
|
subject="ops:track9-smoke",
|
|
username="track9-smoke",
|
|
role="admin",
|
|
auth_source="service",
|
|
provider="track9-script",
|
|
ttl_seconds=300,
|
|
)
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def _get_json(url: str) -> dict:
|
|
with httpx.Client(timeout=10, trust_env=False) as client:
|
|
response = client.get(url, headers=_ops_headers())
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
def _wait_for_ami_connected(base_url: str, retries: int = 8, delay_seconds: float = 1.0) -> dict:
|
|
last_status: dict = {}
|
|
for _ in range(retries):
|
|
status = _get_json(f"{base_url.rstrip('/')}/proxy/asterisk-bridge/asterisk/status")
|
|
last_status = status
|
|
if status.get("ami_connected"):
|
|
return status
|
|
time.sleep(delay_seconds)
|
|
return last_status
|
|
|
|
|
|
def _scalar(conn, sql: str) -> int:
|
|
value = conn.execute(text(sql)).scalar()
|
|
return int(value or 0)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Track 9 Asterisk bridge smoke check")
|
|
parser.add_argument("--base-url", default="http://127.0.0.1:8080")
|
|
parser.add_argument("--env-file", default=".env.production")
|
|
parser.add_argument("--database-url", required=True)
|
|
parser.add_argument("--require-recording", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
_load_env_file(args.env_file)
|
|
|
|
health = _get_json(f"{args.base_url.rstrip('/')}/proxy/asterisk-bridge/health")
|
|
status = _wait_for_ami_connected(args.base_url)
|
|
|
|
failures: list[str] = []
|
|
if health.get("status") != "ok":
|
|
failures.append("asterisk-bridge health is not ok")
|
|
if not status.get("ami_connected"):
|
|
failures.append("AMI is not connected")
|
|
if not status.get("queue_codes_loaded"):
|
|
failures.append("Queue mapping is empty")
|
|
|
|
engine = create_engine(args.database_url, future=True)
|
|
with engine.begin() as conn:
|
|
started = _scalar(
|
|
conn,
|
|
"SELECT COUNT(*) FROM asterisk_event_log WHERE ami_event_name = 'MVPCCCallStarted' AND forward_status = 'forwarded'",
|
|
)
|
|
ended = _scalar(
|
|
conn,
|
|
"SELECT COUNT(*) FROM asterisk_event_log WHERE ami_event_name = 'MVPCCCallEnded' AND forward_status = 'forwarded'",
|
|
)
|
|
links = _scalar(conn, "SELECT COUNT(*) FROM asterisk_call_links")
|
|
recordings = _scalar(
|
|
conn,
|
|
"SELECT COUNT(*) FROM asterisk_event_log WHERE ami_event_name = 'MVPCCRecordingReady' AND recording_id IS NOT NULL",
|
|
)
|
|
|
|
if started < 1:
|
|
failures.append("No forwarded MVPCCCallStarted events")
|
|
if ended < 1:
|
|
failures.append("No forwarded MVPCCCallEnded events")
|
|
if links < 1:
|
|
failures.append("No asterisk_call_links rows")
|
|
if args.require_recording and recordings < 1:
|
|
failures.append("No uploaded recording linked to an Asterisk event")
|
|
|
|
if failures:
|
|
print("[FAIL] Asterisk lab smoke failed")
|
|
for item in failures:
|
|
print(f"- {item}")
|
|
return 1
|
|
|
|
print("[PASS] Asterisk lab smoke passed")
|
|
print(f"- started_events: {started}")
|
|
print(f"- ended_events: {ended}")
|
|
print(f"- call_links: {links}")
|
|
print(f"- recording_events_with_upload: {recordings}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|