968 lines
36 KiB
Python
968 lines
36 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import csv
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
import json
|
|
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_uat_dry_run"
|
|
DEFAULT_OUTPUT_ROOT = ROOT / "docs" / "uat" / "evidence"
|
|
|
|
SCENARIO_SERVICE_MAP = {
|
|
"SYS": "platform",
|
|
"S0": "auth-service/gateway",
|
|
"S1": "customer-service",
|
|
"S2": "interaction-service",
|
|
"S3": "interaction-service/routing-service",
|
|
"S4": "routing-service/interaction-service",
|
|
"S5": "voice-adapter-service",
|
|
"S6": "telegram-adapter-service",
|
|
"S7": "kb-service",
|
|
"S8": "supervisor-service",
|
|
"S9": "reporting-service",
|
|
}
|
|
|
|
SEVERITY_RISK_MAP = {
|
|
"P1": "High",
|
|
"P2": "Medium",
|
|
"P3": "Medium",
|
|
"P4": "Low",
|
|
}
|
|
|
|
SERVICE_SPECS = [
|
|
{"name": "auth", "module": "services.auth_service.app:app", "port": 58001},
|
|
{"name": "audit", "module": "services.audit_service.app:app", "port": 58002},
|
|
{"name": "customer", "module": "services.customer_service.app:app", "port": 58003},
|
|
{"name": "interaction", "module": "services.interaction_service.app:app", "port": 58004},
|
|
{"name": "routing", "module": "services.routing_service.app:app", "port": 58005},
|
|
{"name": "voice", "module": "services.voice_adapter_service.app:app", "port": 58006},
|
|
{"name": "telegram", "module": "services.telegram_adapter_service.app:app", "port": 58007},
|
|
{"name": "kb", "module": "services.kb_service.app:app", "port": 58008},
|
|
{"name": "reporting", "module": "services.reporting_service.app:app", "port": 58009},
|
|
{"name": "supervisor", "module": "services.supervisor_service.app:app", "port": 58010},
|
|
{"name": "gateway", "module": "gateway.app:app", "port": 58080},
|
|
]
|
|
|
|
|
|
@dataclass
|
|
class ScenarioResult:
|
|
scenario_id: str
|
|
title: str
|
|
passed: bool
|
|
severity_on_fail: str
|
|
details: str
|
|
evidence: dict[str, Any]
|
|
|
|
|
|
def utc_now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def now_compact() -> str:
|
|
return utc_now().strftime("%Y%m%d_%H%M%S")
|
|
|
|
|
|
def iso_now() -> str:
|
|
return utc_now().isoformat()
|
|
|
|
|
|
async def wait_for_health(base_url: str, retries: int = 80, 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:58001"
|
|
env["AUDIT_SERVICE_URL"] = "http://127.0.0.1:58002"
|
|
env["CUSTOMER_SERVICE_URL"] = "http://127.0.0.1:58003"
|
|
env["INTERACTION_SERVICE_URL"] = "http://127.0.0.1:58004"
|
|
env["ROUTING_SERVICE_URL"] = "http://127.0.0.1:58005"
|
|
env["VOICE_ADAPTER_SERVICE_URL"] = "http://127.0.0.1:58006"
|
|
env["TELEGRAM_ADAPTER_SERVICE_URL"] = "http://127.0.0.1:58007"
|
|
env["KB_SERVICE_URL"] = "http://127.0.0.1:58008"
|
|
env["REPORTING_SERVICE_URL"] = "http://127.0.0.1:58009"
|
|
env["SUPERVISOR_SERVICE_URL"] = "http://127.0.0.1:58010"
|
|
|
|
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
|
|
|
|
|
|
def scenario_result(
|
|
scenario_id: str,
|
|
title: str,
|
|
passed: bool,
|
|
severity_on_fail: str,
|
|
details: str,
|
|
evidence: dict[str, Any] | None = None,
|
|
) -> ScenarioResult:
|
|
return ScenarioResult(
|
|
scenario_id=scenario_id,
|
|
title=title,
|
|
passed=passed,
|
|
severity_on_fail=severity_on_fail,
|
|
details=details,
|
|
evidence=evidence or {},
|
|
)
|
|
|
|
|
|
async def run_scenarios(base_url: str) -> tuple[list[ScenarioResult], dict[str, Any]]:
|
|
admin = {"X-User": "admin", "X-Role": "admin"}
|
|
supervisor = {"X-User": "supervisor", "X-Role": "supervisor"}
|
|
operator = {"X-User": "operator", "X-Role": "operator"}
|
|
analyst = {"X-User": "analyst", "X-Role": "analyst"}
|
|
results: list[ScenarioResult] = []
|
|
context: dict[str, Any] = {
|
|
"started_at": iso_now(),
|
|
"baseline_doc": "docs/gates/mvp-pilot-baseline.md",
|
|
"wave2_backlog": "docs/roadmap/05-wave2-backlog.md",
|
|
}
|
|
|
|
async with httpx.AsyncClient(base_url=base_url, timeout=12) as client:
|
|
health = await client.get("/health")
|
|
registry = await client.get("/registry")
|
|
if health.status_code != 200 or registry.status_code != 200:
|
|
details = f"health={health.status_code}, registry={registry.status_code}"
|
|
return [scenario_result("SYS", "System precheck", False, "P1", details)], context
|
|
services = registry.json().get("services", {})
|
|
context["reporting_url"] = services.get("reporting", "")
|
|
|
|
login = await client.post("/proxy/auth/auth/login", json={"username": "admin", "password": "admin123"})
|
|
login_ok = False
|
|
if login.status_code == 200:
|
|
login_ok = bool(login.json().get("access_token"))
|
|
denied = await client.post(
|
|
"/proxy/auth/users",
|
|
headers=operator,
|
|
json={
|
|
"username": f"uat_denied_{now_compact()}",
|
|
"password": "secret123",
|
|
"full_name": "UAT Denied",
|
|
"role": "operator",
|
|
},
|
|
)
|
|
results.append(
|
|
scenario_result(
|
|
"S0",
|
|
"Login and RBAC deny",
|
|
login_ok and denied.status_code == 403,
|
|
"P1",
|
|
f"login={login.status_code}, denied={denied.status_code}",
|
|
{"user": "admin", "denied_status": denied.status_code},
|
|
)
|
|
)
|
|
|
|
queue = await client.post(
|
|
"/proxy/routing/queues",
|
|
headers=admin,
|
|
json={
|
|
"name": f"UAT Dry Queue {now_compact()}",
|
|
"description": "UAT dry-run queue",
|
|
"rules": [
|
|
{"channel": "voice", "priority": 3, "strategy": "round_robin", "sla_seconds": 30},
|
|
{"channel": "telegram", "priority": 3, "strategy": "round_robin", "sla_seconds": 45},
|
|
],
|
|
},
|
|
)
|
|
if queue.status_code != 200:
|
|
results.append(
|
|
scenario_result(
|
|
"SYS",
|
|
"System precheck",
|
|
False,
|
|
"P1",
|
|
f"queue create failed status={queue.status_code}",
|
|
)
|
|
)
|
|
context["completed_at"] = iso_now()
|
|
return results, context
|
|
queue_id = queue.json()["queue_id"]
|
|
context["queue_id"] = queue_id
|
|
|
|
customer = await client.post(
|
|
"/proxy/customer/customers",
|
|
json={
|
|
"display_name": "UAT Dry Customer",
|
|
"phones": ["+77017770000"],
|
|
"preferred_phone": "+77017770000",
|
|
"tags": ["uat", "dry-run"],
|
|
},
|
|
)
|
|
customer_id = ""
|
|
search_ok = False
|
|
if customer.status_code == 200:
|
|
customer_id = customer.json()["customer_id"]
|
|
search = await client.get("/proxy/customer/customers?query=dry-run")
|
|
if search.status_code == 200:
|
|
search_ok = any(item.get("customer_id") == customer_id for item in search.json())
|
|
context["customer_id"] = customer_id
|
|
results.append(
|
|
scenario_result(
|
|
"S1",
|
|
"Customer create and search",
|
|
customer.status_code == 200 and search_ok,
|
|
"P1",
|
|
f"create={customer.status_code}, search={search_ok}",
|
|
{"customer_id": customer_id, "query": "dry-run"},
|
|
)
|
|
)
|
|
|
|
voice_interaction_id = ""
|
|
interaction = await client.post(
|
|
"/proxy/interaction/interactions",
|
|
headers=operator,
|
|
json={
|
|
"channel": "voice",
|
|
"subject": "UAT dry-run S2 voice",
|
|
"customer_id": customer_id or None,
|
|
"queue_id": queue_id,
|
|
"priority": 3,
|
|
},
|
|
)
|
|
if interaction.status_code != 200:
|
|
results.append(
|
|
scenario_result(
|
|
"S2",
|
|
"Voice interaction lifecycle",
|
|
False,
|
|
"P1",
|
|
f"create status={interaction.status_code}",
|
|
)
|
|
)
|
|
else:
|
|
voice_interaction_id = interaction.json()["interaction_id"]
|
|
context["voice_interaction_id"] = voice_interaction_id
|
|
assign = await client.patch(
|
|
f"/proxy/interaction/interactions/{voice_interaction_id}/assign",
|
|
headers=supervisor,
|
|
json={"assignee": "operator_a"},
|
|
)
|
|
close = await client.patch(
|
|
f"/proxy/interaction/interactions/{voice_interaction_id}/status",
|
|
headers=operator,
|
|
json={"status": "closed"},
|
|
)
|
|
timeline = await client.get(f"/proxy/interaction/interactions/{voice_interaction_id}/timeline")
|
|
actions: set[str] = set()
|
|
if timeline.status_code == 200:
|
|
actions = {event.get("action", "") for event in timeline.json().get("events", [])}
|
|
expected = {"interaction.created", "interaction.assigned", "interaction.status_changed"}
|
|
results.append(
|
|
scenario_result(
|
|
"S2",
|
|
"Voice interaction lifecycle",
|
|
assign.status_code == 200 and close.status_code == 200 and expected.issubset(actions),
|
|
"P1",
|
|
f"assign={assign.status_code}, close={close.status_code}, actions={sorted(actions)}",
|
|
{"interaction_id": voice_interaction_id},
|
|
)
|
|
)
|
|
|
|
esc_create = await client.post(
|
|
"/proxy/interaction/interactions",
|
|
headers=operator,
|
|
json={
|
|
"channel": "voice",
|
|
"subject": "UAT dry-run S3 escalation",
|
|
"customer_id": customer_id or None,
|
|
"queue_id": queue_id,
|
|
"priority": 3,
|
|
},
|
|
)
|
|
if esc_create.status_code != 200:
|
|
results.append(
|
|
scenario_result(
|
|
"S3",
|
|
"Assignment and escalation to second line",
|
|
False,
|
|
"P1",
|
|
f"create status={esc_create.status_code}",
|
|
)
|
|
)
|
|
else:
|
|
esc_interaction_id = esc_create.json()["interaction_id"]
|
|
assign = await client.patch(
|
|
f"/proxy/interaction/interactions/{esc_interaction_id}/assign",
|
|
headers=supervisor,
|
|
json={"assignee": "operator_b"},
|
|
)
|
|
escalated = await client.post(
|
|
f"/proxy/interaction/interactions/{esc_interaction_id}/escalate",
|
|
headers=operator,
|
|
json={"target_queue_id": "line2"},
|
|
)
|
|
timeline = await client.get(f"/proxy/interaction/interactions/{esc_interaction_id}/timeline")
|
|
actions = []
|
|
if timeline.status_code == 200:
|
|
actions = [event.get("action", "") for event in timeline.json().get("events", [])]
|
|
results.append(
|
|
scenario_result(
|
|
"S3",
|
|
"Assignment and escalation to second line",
|
|
assign.status_code == 200
|
|
and escalated.status_code == 200
|
|
and escalated.json().get("status") == "escalated"
|
|
and escalated.json().get("queue_id") == "line2"
|
|
and "interaction.escalated" in actions,
|
|
"P1",
|
|
f"assign={assign.status_code}, escalate={escalated.status_code}, queue={escalated.json().get('queue_id') if escalated.status_code == 200 else ''}",
|
|
{"interaction_id": esc_interaction_id, "target_queue": "line2"},
|
|
)
|
|
)
|
|
|
|
route = await client.post(f"/proxy/routing/queues/{queue_id}/route?channel=voice&priority=3")
|
|
timeline = (
|
|
await client.get(f"/proxy/interaction/interactions/{voice_interaction_id}/timeline")
|
|
if voice_interaction_id
|
|
else None
|
|
)
|
|
timeline_ok = False
|
|
timeline_count = 0
|
|
if timeline and timeline.status_code == 200:
|
|
timeline_count = len(timeline.json().get("events", []))
|
|
timeline_ok = timeline_count >= 3
|
|
route_ok = False
|
|
if route.status_code == 200:
|
|
route_ok = bool(route.json().get("assignee"))
|
|
results.append(
|
|
scenario_result(
|
|
"S4",
|
|
"Routing and timeline verification",
|
|
route_ok and timeline_ok,
|
|
"P2",
|
|
f"route={route.status_code}, timeline_ok={timeline_ok}, timeline_events={timeline_count}",
|
|
{
|
|
"queue_id": queue_id,
|
|
"routed_assignee": route.json().get("assignee") if route.status_code == 200 else "",
|
|
"interaction_id": voice_interaction_id,
|
|
},
|
|
)
|
|
)
|
|
|
|
voice_event = await client.post(
|
|
"/proxy/voice/integrations/voice/events",
|
|
headers=operator,
|
|
json={
|
|
"event_type": "call.started",
|
|
"call_id": f"uat_call_{now_compact()}",
|
|
"interaction_id": voice_interaction_id or None,
|
|
"payload": {"source": "uat_dry_run"},
|
|
},
|
|
)
|
|
voice_event_id = voice_event.json().get("event_id") if voice_event.status_code == 200 else ""
|
|
voice_list_ok = False
|
|
if voice_event.status_code == 200:
|
|
voice_events = await client.get(
|
|
"/proxy/voice/integrations/voice/events?limit=10",
|
|
headers=supervisor,
|
|
)
|
|
if voice_events.status_code == 200:
|
|
voice_list_ok = any(item.get("event_id") == voice_event_id for item in voice_events.json())
|
|
results.append(
|
|
scenario_result(
|
|
"S5",
|
|
"Voice event intake",
|
|
voice_event.status_code == 200 and voice_list_ok,
|
|
"P2",
|
|
f"create={voice_event.status_code}, listed={voice_list_ok}",
|
|
{"event_id": voice_event_id},
|
|
)
|
|
)
|
|
|
|
telegram = await client.post(
|
|
"/proxy/telegram/integrations/telegram/webhook",
|
|
json={
|
|
"chat_id": f"uat_dry_chat_{now_compact()}",
|
|
"text": "UAT dry-run telegram",
|
|
"payload": {"source": "uat_dry_run"},
|
|
},
|
|
)
|
|
telegram_id = telegram.json().get("message_id") if telegram.status_code == 200 else ""
|
|
telegram_list_ok = False
|
|
if telegram.status_code == 200:
|
|
messages = await client.get("/proxy/telegram/integrations/telegram/messages?limit=10")
|
|
if messages.status_code == 200:
|
|
telegram_list_ok = any(item.get("message_id") == telegram_id for item in messages.json())
|
|
results.append(
|
|
scenario_result(
|
|
"S6",
|
|
"Telegram interaction lifecycle",
|
|
telegram.status_code == 200 and telegram_list_ok,
|
|
"P1",
|
|
f"create={telegram.status_code}, listed={telegram_list_ok}",
|
|
{"message_id": telegram_id},
|
|
)
|
|
)
|
|
|
|
category = await client.post(
|
|
"/proxy/kb/knowledge/categories",
|
|
headers=analyst,
|
|
json={"name": f"UAT Dry {now_compact()}", "description": "dry-run"},
|
|
)
|
|
if category.status_code != 200:
|
|
results.append(
|
|
scenario_result(
|
|
"S7",
|
|
"KB usage in active handling",
|
|
False,
|
|
"P2",
|
|
f"category create status={category.status_code}",
|
|
)
|
|
)
|
|
else:
|
|
category_id = category.json()["category_id"]
|
|
article = await client.post(
|
|
"/proxy/kb/knowledge/articles",
|
|
headers=analyst,
|
|
json={
|
|
"category_id": category_id,
|
|
"title": "UAT dry KB article",
|
|
"body": "KB article body for dry run",
|
|
"tags": ["uat", "kb", "dry"],
|
|
},
|
|
)
|
|
search = await client.get("/proxy/kb/knowledge/search?q=dry")
|
|
article_id = article.json().get("article_id") if article.status_code == 200 else ""
|
|
found = False
|
|
if search.status_code == 200:
|
|
found = any(item.get("article_id") == article_id for item in search.json())
|
|
results.append(
|
|
scenario_result(
|
|
"S7",
|
|
"KB usage in active handling",
|
|
article.status_code == 200 and search.status_code == 200 and found,
|
|
"P2",
|
|
f"article={article.status_code}, search={search.status_code}, found={found}",
|
|
{"article_id": article_id, "keyword": "dry"},
|
|
)
|
|
)
|
|
|
|
up1 = await client.post(
|
|
"/proxy/supervisor/supervisor/agent-states",
|
|
json={"agent_id": "uat_dry_a1", "state": "READY", "queue_id": queue_id},
|
|
)
|
|
up2 = await client.post(
|
|
"/proxy/supervisor/supervisor/agent-states",
|
|
json={"agent_id": "uat_dry_a2", "state": "BUSY", "queue_id": queue_id},
|
|
)
|
|
metrics = await client.post(
|
|
f"/proxy/supervisor/supervisor/queue-metrics?queue_id={queue_id}&in_queue=2&avg_wait_seconds=19"
|
|
)
|
|
realtime = await client.get("/proxy/supervisor/supervisor/realtime")
|
|
queue_seen = False
|
|
agent_total = 0
|
|
if realtime.status_code == 200:
|
|
body = realtime.json()
|
|
agent_total = int(body.get("agents", {}).get("total", 0))
|
|
queue_seen = any(row.get("queue_id") == queue_id for row in body.get("queues", []))
|
|
results.append(
|
|
scenario_result(
|
|
"S8",
|
|
"Supervisor realtime",
|
|
up1.status_code == 200
|
|
and up2.status_code == 200
|
|
and metrics.status_code == 200
|
|
and realtime.status_code == 200
|
|
and agent_total >= 2
|
|
and queue_seen,
|
|
"P2",
|
|
f"state_updates=({up1.status_code},{up2.status_code}), metrics={metrics.status_code}, realtime={realtime.status_code}",
|
|
{"queue_id": queue_id, "agent_total": agent_total},
|
|
)
|
|
)
|
|
|
|
kpi_rows = [
|
|
{
|
|
"queue_id": queue_id,
|
|
"answered": True,
|
|
"wait_seconds": 15,
|
|
"handle_seconds": 90,
|
|
"abandoned": False,
|
|
"resolved_first_contact": True,
|
|
},
|
|
{
|
|
"queue_id": queue_id,
|
|
"answered": True,
|
|
"wait_seconds": 33,
|
|
"handle_seconds": 110,
|
|
"abandoned": False,
|
|
"resolved_first_contact": False,
|
|
},
|
|
{
|
|
"queue_id": queue_id,
|
|
"answered": False,
|
|
"wait_seconds": 10,
|
|
"handle_seconds": 0,
|
|
"abandoned": True,
|
|
"resolved_first_contact": False,
|
|
},
|
|
]
|
|
ingest_statuses: list[int] = []
|
|
for row in kpi_rows:
|
|
ingested = await client.post("/proxy/reporting/reports/events", json=row)
|
|
ingest_statuses.append(ingested.status_code)
|
|
kpi = await client.get(f"/proxy/reporting/reports/kpi?queue_id={queue_id}&sl_threshold_seconds=30")
|
|
has_kpi = False
|
|
if kpi.status_code == 200:
|
|
payload = kpi.json().get("kpi", {})
|
|
has_kpi = {"SL", "ASA", "AHT", "Abandon", "FCR"}.issubset(set(payload.keys()))
|
|
exported = await client.get("/proxy/reporting/reports/export")
|
|
export_raw = exported.json().get("raw", "") if exported.status_code == 200 else ""
|
|
csv_ok = (
|
|
exported.status_code == 200
|
|
and "queue_id,answered,wait_seconds,handle_seconds,abandoned,resolved_first_contact,created_at" in export_raw
|
|
and queue_id in export_raw
|
|
)
|
|
results.append(
|
|
scenario_result(
|
|
"S9",
|
|
"KPI report and export validation",
|
|
all(code == 200 for code in ingest_statuses) and kpi.status_code == 200 and has_kpi and csv_ok,
|
|
"P2",
|
|
f"ingest={ingest_statuses}, kpi={kpi.status_code}, csv={csv_ok}",
|
|
{"queue_id": queue_id},
|
|
)
|
|
)
|
|
|
|
context["completed_at"] = iso_now()
|
|
return results, context
|
|
|
|
|
|
def build_defects(results: list[ScenarioResult]) -> list[dict[str, str]]:
|
|
defects: list[dict[str, str]] = []
|
|
stamp = iso_now()
|
|
counter = 1
|
|
for row in results:
|
|
if row.passed:
|
|
continue
|
|
defect_id = f"UAT-DRY-{counter:03d}"
|
|
counter += 1
|
|
severity = row.severity_on_fail
|
|
defects.append(
|
|
{
|
|
"defect_id": defect_id,
|
|
"severity": severity,
|
|
"status": "Open",
|
|
"scenario": row.scenario_id,
|
|
"service": SCENARIO_SERVICE_MAP.get(row.scenario_id, "platform"),
|
|
"risk_level": SEVERITY_RISK_MAP.get(severity, "Medium"),
|
|
"summary": row.title,
|
|
"repro_steps": "See scenario checklist and automated dry-run logs",
|
|
"actual_result": row.details,
|
|
"expected_result": "Scenario should pass",
|
|
"verification_step": f"Re-run scenario {row.scenario_id} after fix",
|
|
"owner": "TBD",
|
|
"opened_at": stamp,
|
|
"closed_at": "",
|
|
"backlog_bucket": "MVP" if severity in {"P1", "P2"} else "Wave2",
|
|
"notes": "Generated by scripts/uat_dry_run.py",
|
|
}
|
|
)
|
|
return defects
|
|
|
|
|
|
def write_scenario_json(output_dir: Path, results: list[ScenarioResult], context: dict[str, Any]) -> Path:
|
|
payload = {
|
|
"generated_at": iso_now(),
|
|
"context": context,
|
|
"scenarios": [
|
|
{
|
|
"scenario_id": row.scenario_id,
|
|
"title": row.title,
|
|
"passed": row.passed,
|
|
"severity_on_fail": row.severity_on_fail,
|
|
"details": row.details,
|
|
"evidence": row.evidence,
|
|
}
|
|
for row in results
|
|
],
|
|
}
|
|
path = output_dir / "scenario-results.json"
|
|
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def write_defect_csv(output_dir: Path, defects: list[dict[str, str]]) -> Path:
|
|
path = output_dir / "defect-log.csv"
|
|
fieldnames = [
|
|
"defect_id",
|
|
"severity",
|
|
"status",
|
|
"scenario",
|
|
"service",
|
|
"risk_level",
|
|
"summary",
|
|
"repro_steps",
|
|
"actual_result",
|
|
"expected_result",
|
|
"verification_step",
|
|
"owner",
|
|
"opened_at",
|
|
"closed_at",
|
|
"backlog_bucket",
|
|
"notes",
|
|
]
|
|
with path.open("w", newline="", encoding="utf-8") as fp:
|
|
writer = csv.DictWriter(fp, fieldnames=fieldnames)
|
|
writer.writeheader()
|
|
for row in defects:
|
|
writer.writerow(row)
|
|
return path
|
|
|
|
|
|
def write_session_protocol(
|
|
output_dir: Path,
|
|
session_id: str,
|
|
base_url: str,
|
|
results: list[ScenarioResult],
|
|
defects: list[dict[str, str]],
|
|
build_version: str,
|
|
deployment_date: str,
|
|
cluster_namespace: str,
|
|
) -> Path:
|
|
passed = sum(1 for row in results if row.passed)
|
|
failed = sum(1 for row in results if not row.passed)
|
|
p1 = sum(1 for row in defects if row["severity"] == "P1")
|
|
p2 = sum(1 for row in defects if row["severity"] == "P2")
|
|
p3 = sum(1 for row in defects if row["severity"] == "P3")
|
|
p4 = sum(1 for row in defects if row["severity"] == "P4")
|
|
|
|
lines: list[str] = []
|
|
lines.append("# UAT Session Protocol (Automated Dry Run)")
|
|
lines.append("")
|
|
lines.append("## Session Metadata")
|
|
lines.append("")
|
|
lines.append(f"- Session ID: {session_id}")
|
|
lines.append(f"- Date: {utc_now().date().isoformat()}")
|
|
lines.append(f"- Start time (UTC): {iso_now()}")
|
|
lines.append(f"- End time (UTC): {iso_now()}")
|
|
lines.append(f"- Environment URL: {base_url}")
|
|
lines.append(f"- Build/version: {build_version}")
|
|
lines.append(f"- Deployment date: {deployment_date}")
|
|
lines.append(f"- Cluster/namespace: {cluster_namespace}")
|
|
lines.append("")
|
|
lines.append("## Scope Confirmation")
|
|
lines.append("")
|
|
lines.append("- [x] MVP pilot scope reviewed against `docs/gates/mvp-pilot-baseline.md`")
|
|
lines.append("- [x] Out-of-scope requests remain deferred to `docs/roadmap/05-wave2-backlog.md`")
|
|
lines.append("- [x] API stabilization rules remain in effect (backward-compatible fixes only)")
|
|
lines.append("")
|
|
lines.append("## Participants")
|
|
lines.append("")
|
|
lines.append("- Operators: automated dry-run actor")
|
|
lines.append("- Supervisor: automated dry-run actor")
|
|
lines.append("- Analyst: automated dry-run actor")
|
|
lines.append("- Admin: automated dry-run actor")
|
|
lines.append("- Business owner: pending manual UAT")
|
|
lines.append("- IT owner: pending manual UAT")
|
|
lines.append("")
|
|
lines.append("## Preconditions")
|
|
lines.append("")
|
|
lines.append("- [x] Environment is reachable")
|
|
lines.append("- [x] Test accounts are active")
|
|
lines.append("- [x] Test queue is configured")
|
|
lines.append("- [x] Audit and reporting endpoints are reachable")
|
|
lines.append("")
|
|
lines.append("## Execution Summary")
|
|
lines.append("")
|
|
lines.append(f"- Mandatory scenarios executed: {len(results)}")
|
|
lines.append(f"- Passed: {passed}")
|
|
lines.append(f"- Failed: {failed}")
|
|
lines.append("- Blocked: 0")
|
|
lines.append("")
|
|
lines.append("## Scenario Results")
|
|
lines.append("")
|
|
for row in results:
|
|
mark = "PASS" if row.passed else "FAIL"
|
|
lines.append(f"- [{mark}] {row.scenario_id} {row.title}: {row.details}")
|
|
lines.append("")
|
|
lines.append("## Defect Summary")
|
|
lines.append("")
|
|
lines.append(f"- P1: {p1}")
|
|
lines.append(f"- P2: {p2}")
|
|
lines.append(f"- P3: {p3}")
|
|
lines.append(f"- P4: {p4}")
|
|
lines.append("- Defect log attachment: defect-log.csv")
|
|
lines.append("")
|
|
lines.append("## Decision")
|
|
lines.append("")
|
|
if failed == 0:
|
|
lines.append("- [x] Accepted with conditions")
|
|
lines.append("- [ ] Accepted for pilot completion")
|
|
lines.append("- [ ] Not accepted")
|
|
else:
|
|
lines.append("- [ ] Accepted with conditions")
|
|
lines.append("- [ ] Accepted for pilot completion")
|
|
lines.append("- [x] Not accepted")
|
|
lines.append("")
|
|
lines.append("## Comments")
|
|
lines.append("")
|
|
lines.append("- Automated dry-run completed.")
|
|
lines.append("- Manual UAT with real operators/supervisor is still required for final sign-off.")
|
|
lines.append("- Any future scope additions must be recorded in `docs/roadmap/05-wave2-backlog.md`.")
|
|
lines.append("")
|
|
lines.append("## Signatures")
|
|
lines.append("")
|
|
lines.append("- Business owner: pending")
|
|
lines.append("- IT owner: pending")
|
|
lines.append("- Supervisor representative: pending")
|
|
lines.append(f"- Date: {utc_now().date().isoformat()}")
|
|
lines.append("")
|
|
path = output_dir / "session-protocol.md"
|
|
path.write_text("\n".join(lines), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def write_signoff_draft(output_dir: Path, session_id: str, failed: int) -> Path:
|
|
lines: list[str] = []
|
|
lines.append("# UAT Sign-off Sheet (Draft)")
|
|
lines.append("")
|
|
lines.append(f"- Session ID: {session_id}")
|
|
lines.append(f"- Date: {utc_now().date().isoformat()}")
|
|
lines.append(f"- Scenario failures: {failed}")
|
|
lines.append("")
|
|
lines.append("## Scope Confirmation")
|
|
lines.append("")
|
|
lines.append("- MVP pilot scope confirmed against `docs/gates/mvp-pilot-baseline.md`")
|
|
lines.append("- Out-of-scope requests remain deferred to `docs/roadmap/05-wave2-backlog.md`")
|
|
lines.append("")
|
|
lines.append("## Acceptance Statement")
|
|
lines.append("")
|
|
lines.append("This draft is generated by automated dry-run and cannot replace real UAT signatures.")
|
|
lines.append("")
|
|
lines.append("## Required Manual Signatures")
|
|
lines.append("")
|
|
lines.append("- Business owner: pending")
|
|
lines.append("- IT owner: pending")
|
|
lines.append("- Supervisor representative: pending")
|
|
lines.append("")
|
|
path = output_dir / "signoff-draft.md"
|
|
path.write_text("\n".join(lines), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def write_summary(
|
|
output_dir: Path,
|
|
session_id: str,
|
|
base_url: str,
|
|
scenario_file: Path,
|
|
defect_file: Path,
|
|
protocol_file: Path,
|
|
signoff_file: Path,
|
|
results: list[ScenarioResult],
|
|
defects: list[dict[str, str]],
|
|
) -> Path:
|
|
passed = sum(1 for row in results if row.passed)
|
|
failed = sum(1 for row in results if not row.passed)
|
|
p1 = sum(1 for row in defects if row["severity"] == "P1")
|
|
p2 = sum(1 for row in defects if row["severity"] == "P2")
|
|
|
|
lines: list[str] = []
|
|
lines.append("# UAT Dry-Run Summary")
|
|
lines.append("")
|
|
lines.append(f"- Session ID: {session_id}")
|
|
lines.append(f"- Generated at (UTC): {iso_now()}")
|
|
lines.append(f"- Environment URL: {base_url}")
|
|
lines.append(f"- Scenarios: {len(results)} total, {passed} passed, {failed} failed")
|
|
lines.append(f"- Defects: P1={p1}, P2={p2}")
|
|
lines.append("")
|
|
lines.append("## Files")
|
|
lines.append("")
|
|
lines.append(f"- Scenario results: `{scenario_file.name}`")
|
|
lines.append(f"- Session protocol: `{protocol_file.name}`")
|
|
lines.append(f"- Defect log: `{defect_file.name}`")
|
|
lines.append(f"- Sign-off draft: `{signoff_file.name}`")
|
|
lines.append("")
|
|
lines.append("## Next Step")
|
|
lines.append("")
|
|
lines.append("Run manual UAT with real participants, fix only P1/P2 items, and collect signatures.")
|
|
lines.append("Route feature requests and deferred scope to `docs/roadmap/05-wave2-backlog.md`.")
|
|
lines.append("")
|
|
path = output_dir / "summary.md"
|
|
path.write_text("\n".join(lines), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def update_p1p2_register(defects: list[dict[str, str]], session_id: str) -> None:
|
|
p1 = sum(1 for row in defects if row["severity"] == "P1" and row["status"] == "Open")
|
|
p2 = sum(1 for row in defects if row["severity"] == "P2" and row["status"] == "Open")
|
|
path = ROOT / "docs" / "gates" / "p1-p2-defects.md"
|
|
lines: list[str] = []
|
|
lines.append("# P1/P2 Defect Register (Pilot)")
|
|
lines.append("")
|
|
lines.append(f"Last update: {utc_now().date().isoformat()} ({session_id})")
|
|
lines.append("")
|
|
lines.append("## Current Status")
|
|
lines.append(f"- Open P1: {p1}")
|
|
lines.append(f"- Open P2: {p2}")
|
|
lines.append("- Source: `scripts/uat_dry_run.py` and automated checks")
|
|
lines.append(f"- Latest dry-run evidence: `docs/uat/evidence/dry_run_{session_id}/`")
|
|
lines.append("")
|
|
lines.append("## Triage Policy")
|
|
lines.append("- Only `P1` and `P2` issues belong to MVP remediation.")
|
|
lines.append("- `P3` and `P4` issues move to `docs/roadmap/05-wave2-backlog.md` unless they block sign-off.")
|
|
lines.append("")
|
|
lines.append("## Pilot Note")
|
|
lines.append("- Final P1/P2 closure is confirmed only after the real-operator UAT cycle and sign-off.")
|
|
path.write_text("\n".join(lines), encoding="utf-8")
|
|
|
|
|
|
async def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Run automated UAT dry-run and generate session artifacts")
|
|
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 dry-run")
|
|
parser.add_argument("--keep-data", action="store_true", help="Keep temporary auto-start data directory")
|
|
parser.add_argument(
|
|
"--output-dir",
|
|
default="",
|
|
help="Output root directory (default: docs/uat/evidence)",
|
|
)
|
|
parser.add_argument("--session-id", default="", help="Optional session ID override")
|
|
parser.add_argument("--build-version", default="v1.0.0-mvp", help="Build/version for protocol")
|
|
parser.add_argument(
|
|
"--deployment-date",
|
|
default=utc_now().date().isoformat(),
|
|
help="Deployment date for protocol",
|
|
)
|
|
parser.add_argument("--cluster-namespace", default="local/standalone", help="Cluster/namespace for protocol")
|
|
parser.add_argument(
|
|
"--update-defect-register",
|
|
action="store_true",
|
|
help="Update docs/gates/p1-p2-defects.md using dry-run result",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
output_root = Path(args.output_dir) if args.output_dir else DEFAULT_OUTPUT_ROOT
|
|
output_root = output_root if output_root.is_absolute() else (ROOT / output_root).resolve()
|
|
session_id = args.session_id.strip() or f"UAT-DRY-{now_compact()}"
|
|
session_dir = output_root / f"dry_run_{session_id}"
|
|
session_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
run_data_dir = DATA_ROOT / f"run_{int(time.time() * 1000)}"
|
|
processes: list[subprocess.Popen] = []
|
|
base_url = args.base_url
|
|
|
|
try:
|
|
if args.auto_start:
|
|
processes = await start_services_sequential(run_data_dir)
|
|
base_url = "http://127.0.0.1:58080"
|
|
|
|
results, context = await run_scenarios(base_url)
|
|
defects = build_defects(results)
|
|
|
|
scenario_file = write_scenario_json(session_dir, results, context)
|
|
defect_file = write_defect_csv(session_dir, defects)
|
|
protocol_file = write_session_protocol(
|
|
session_dir,
|
|
session_id,
|
|
base_url,
|
|
results,
|
|
defects,
|
|
args.build_version,
|
|
args.deployment_date,
|
|
args.cluster_namespace,
|
|
)
|
|
signoff_file = write_signoff_draft(session_dir, session_id, sum(1 for row in results if not row.passed))
|
|
summary_file = write_summary(
|
|
session_dir,
|
|
session_id,
|
|
base_url,
|
|
scenario_file,
|
|
defect_file,
|
|
protocol_file,
|
|
signoff_file,
|
|
results,
|
|
defects,
|
|
)
|
|
|
|
if args.update_defect_register:
|
|
update_p1p2_register(defects, session_id)
|
|
|
|
passed = sum(1 for row in results if row.passed)
|
|
failed = sum(1 for row in results if not row.passed)
|
|
print(f"UAT dry-run session: {session_id}")
|
|
print(f"Output directory: {session_dir}")
|
|
print(f"Scenarios: total={len(results)} passed={passed} failed={failed}")
|
|
print(f"Summary: {summary_file}")
|
|
sys.exit(0 if failed == 0 else 1)
|
|
finally:
|
|
stop_services(processes)
|
|
if args.auto_start and not args.keep_data:
|
|
try:
|
|
if run_data_dir.exists():
|
|
shutil.rmtree(run_data_dir)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|