yuqei
c24f7a4f1e
Build and Test Quality Gate / Contract Web typecheck and build (pull_request) Successful in 2m58s
Build and Test Quality Gate / Python tests and Contract API migration (pull_request) Failing after 15m25s
Security Quality Gate / npm audit, Bandit, pip-audit, detect-secrets (pull_request) Successful in 5m14s
645 lines
23 KiB
Python
645 lines
23 KiB
Python
#!/usr/bin/env python3
|
|
"""Verify the V2 release quality gate from runtime checks and QA summaries."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
DEFAULT_ARTIFACT_ROOT = REPO_ROOT / "project-docs" / "qa-artifacts" / "v2-release-quality-gate"
|
|
SMOKE_ARTIFACT_ROOT = REPO_ROOT / "project-docs" / "qa-artifacts" / "v2-smoke-docker-deploy"
|
|
BENCHMARK_ARTIFACT_ROOT = REPO_ROOT / "project-docs" / "qa-artifacts" / "v2-interface-benchmark"
|
|
BENCHMARK_SCRIPT = REPO_ROOT / "yuqei-infra" / "scripts" / "run-interface-benchmark.py"
|
|
REQUIRED_SMOKE_CHECKS = (
|
|
"remote_health",
|
|
"ai_golden_sample_gate",
|
|
"remote_db_check",
|
|
"review_worker_smoke",
|
|
)
|
|
DEFAULT_RELEASE_BENCHMARK_REQUESTS_PER_SCENARIO = 15
|
|
DEFAULT_RELEASE_BENCHMARK_CONCURRENCY = 3
|
|
DEFAULT_RELEASE_BENCHMARK_WARMUP = 2
|
|
|
|
|
|
@dataclass
|
|
class GateCheck:
|
|
key: str
|
|
label: str
|
|
status: str
|
|
detail: str
|
|
required: bool = True
|
|
meta: dict[str, Any] | None = None
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"key": self.key,
|
|
"label": self.label,
|
|
"status": self.status,
|
|
"required": self.required,
|
|
"detail": self.detail,
|
|
"meta": self.meta or {},
|
|
}
|
|
|
|
|
|
def now_stamp() -> str:
|
|
return datetime.now().strftime("%Y-%m-%dT%H-%M-%S")
|
|
|
|
|
|
def checked_at() -> str:
|
|
return datetime.now(timezone.utc).astimezone().isoformat()
|
|
|
|
|
|
def normalize_base_url(value: str) -> str:
|
|
return value.rstrip("/")
|
|
|
|
|
|
def resolve_path(value: str) -> Path:
|
|
path = Path(value)
|
|
return path if path.is_absolute() else REPO_ROOT / path
|
|
|
|
|
|
def latest_summary(root: Path, summary_name: str) -> Path | None:
|
|
if not root.exists():
|
|
return None
|
|
candidates = [path for path in root.glob(f"*/{summary_name}") if path.is_file()]
|
|
if not candidates:
|
|
return None
|
|
return max(candidates, key=lambda path: path.stat().st_mtime)
|
|
|
|
|
|
def read_json(path: Path) -> Any:
|
|
return json.loads(path.read_text(encoding="utf-8-sig"))
|
|
|
|
|
|
def parse_timestamp(value: Any) -> datetime | None:
|
|
if not isinstance(value, str) or not value.strip():
|
|
return None
|
|
normalized = value.strip().replace("Z", "+00:00")
|
|
try:
|
|
parsed = datetime.fromisoformat(normalized)
|
|
except ValueError:
|
|
return None
|
|
if parsed.tzinfo is None:
|
|
return parsed.replace(tzinfo=timezone.utc)
|
|
return parsed.astimezone(timezone.utc)
|
|
|
|
|
|
def age_hours(value: Any) -> float | None:
|
|
parsed = parse_timestamp(value)
|
|
if parsed is None:
|
|
return None
|
|
return (datetime.now(timezone.utc) - parsed).total_seconds() / 3600
|
|
|
|
|
|
def age_status(
|
|
label: str,
|
|
timestamp: Any,
|
|
max_age_hours: float,
|
|
*,
|
|
missing_is_failure: bool = True,
|
|
) -> tuple[str, str, dict[str, Any]]:
|
|
if max_age_hours <= 0:
|
|
return "pass", "", {}
|
|
actual_age = age_hours(timestamp)
|
|
meta = {"checked_at": timestamp, "max_age_hours": max_age_hours, "age_hours": actual_age}
|
|
if actual_age is None:
|
|
status = "fail" if missing_is_failure else "pass"
|
|
return status, f"{label} timestamp is missing or invalid.", meta
|
|
if actual_age > max_age_hours:
|
|
return "fail", f"{label} is stale: age_hours={actual_age:.2f} exceeds max_age_hours={max_age_hours}.", meta
|
|
return "pass", f"{label} age_hours={actual_age:.2f} is within max_age_hours={max_age_hours}.", meta
|
|
|
|
|
|
def fetch_json(
|
|
url: str,
|
|
timeout_seconds: float,
|
|
headers: dict[str, str] | None = None,
|
|
) -> tuple[Any | None, str, int]:
|
|
started = time.perf_counter()
|
|
request = Request(url, headers={"Accept": "application/json", **(headers or {})}, method="GET")
|
|
try:
|
|
with urlopen(request, timeout=timeout_seconds) as response:
|
|
payload = json.loads(response.read().decode("utf-8"))
|
|
elapsed_ms = int((time.perf_counter() - started) * 1000)
|
|
return payload, "", elapsed_ms
|
|
except HTTPError as exc:
|
|
elapsed_ms = int((time.perf_counter() - started) * 1000)
|
|
return None, f"HTTPError {exc.code}: {exc.reason}", elapsed_ms
|
|
except URLError as exc:
|
|
elapsed_ms = int((time.perf_counter() - started) * 1000)
|
|
return None, f"URLError: {exc.reason}", elapsed_ms
|
|
except TimeoutError as exc:
|
|
elapsed_ms = int((time.perf_counter() - started) * 1000)
|
|
return None, f"TimeoutError: {exc}", elapsed_ms
|
|
except Exception as exc: # pragma: no cover - defensive release logging.
|
|
elapsed_ms = int((time.perf_counter() - started) * 1000)
|
|
return None, f"{type(exc).__name__}: {exc}", elapsed_ms
|
|
|
|
|
|
def validate_schema_check(
|
|
service: str,
|
|
base_url: str,
|
|
timeout_seconds: float,
|
|
output_dir: Path,
|
|
*,
|
|
headers: dict[str, str] | None = None,
|
|
allowed_warning_checks: set[str] | None = None,
|
|
) -> GateCheck:
|
|
url = f"{normalize_base_url(base_url)}/api/v1/runtime/schema-check"
|
|
payload, error, elapsed_ms = fetch_json(url, timeout_seconds, headers)
|
|
artifact = output_dir / f"{service}-schema-check.json"
|
|
artifact.write_text(json.dumps(payload or {"error": error}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
if payload is None:
|
|
return GateCheck(
|
|
key=f"{service}_schema_check",
|
|
label=f"{service} schema-check",
|
|
status="fail",
|
|
detail=error,
|
|
meta={"url": url, "elapsed_ms": elapsed_ms, "artifact": str(artifact)},
|
|
)
|
|
failed_checks = [
|
|
item.get("name", "unknown")
|
|
for item in payload.get("checks", [])
|
|
if item.get("status") == "error"
|
|
]
|
|
warning_checks = {
|
|
item.get("name", "unknown")
|
|
for item in payload.get("checks", [])
|
|
if item.get("status") == "warning"
|
|
}
|
|
allowed_warnings = allowed_warning_checks or set()
|
|
if payload.get("status") == "ok" and not failed_checks:
|
|
return GateCheck(
|
|
key=f"{service}_schema_check",
|
|
label=f"{service} schema-check",
|
|
status="pass",
|
|
detail="Runtime schema-check returned status=ok.",
|
|
meta={"url": url, "elapsed_ms": elapsed_ms, "artifact": str(artifact)},
|
|
)
|
|
if (
|
|
payload.get("status") == "warning"
|
|
and not failed_checks
|
|
and warning_checks
|
|
and warning_checks <= allowed_warnings
|
|
):
|
|
return GateCheck(
|
|
key=f"{service}_schema_check",
|
|
label=f"{service} schema-check",
|
|
status="pass",
|
|
detail=f"Runtime schema-check returned only allowed warnings: {','.join(sorted(warning_checks))}.",
|
|
meta={
|
|
"url": url,
|
|
"elapsed_ms": elapsed_ms,
|
|
"artifact": str(artifact),
|
|
"allowed_warning_checks": sorted(allowed_warnings),
|
|
},
|
|
)
|
|
detail = f"Runtime schema-check status={payload.get('status')}; failed checks={','.join(failed_checks) or '-'}."
|
|
return GateCheck(
|
|
key=f"{service}_schema_check",
|
|
label=f"{service} schema-check",
|
|
status="fail",
|
|
detail=detail,
|
|
meta={"url": url, "elapsed_ms": elapsed_ms, "artifact": str(artifact)},
|
|
)
|
|
|
|
|
|
def validate_smoke_summary(path: Path | None, *, skip: bool, max_age_hours: float) -> GateCheck:
|
|
if skip:
|
|
return GateCheck(
|
|
key="docker_smoke_summary",
|
|
label="Docker smoke summary",
|
|
status="skip",
|
|
detail="Docker smoke summary validation was skipped.",
|
|
required=False,
|
|
)
|
|
if path is None:
|
|
return GateCheck(
|
|
key="docker_smoke_summary",
|
|
label="Docker smoke summary",
|
|
status="fail",
|
|
detail="No Docker smoke summary path was provided and no latest summary was found.",
|
|
)
|
|
try:
|
|
payload = read_json(path)
|
|
except Exception as exc:
|
|
return GateCheck(
|
|
key="docker_smoke_summary",
|
|
label="Docker smoke summary",
|
|
status="fail",
|
|
detail=f"Unable to read Docker smoke summary: {exc}",
|
|
meta={"path": str(path)},
|
|
)
|
|
failed_required = [
|
|
item.get("key") or item.get("label", "unknown")
|
|
for item in payload.get("checks", [])
|
|
if item.get("required", True) and item.get("status") != "pass"
|
|
]
|
|
checks_by_key = {item.get("key"): item for item in payload.get("checks", [])}
|
|
missing_required = [key for key in REQUIRED_SMOKE_CHECKS if key not in checks_by_key]
|
|
failed_named = [
|
|
key
|
|
for key in REQUIRED_SMOKE_CHECKS
|
|
if key in checks_by_key and checks_by_key[key].get("status") != "pass"
|
|
]
|
|
age_check_status, age_detail, age_meta = age_status(
|
|
"Docker smoke summary",
|
|
payload.get("checked_at"),
|
|
max_age_hours,
|
|
)
|
|
if (
|
|
payload.get("status") == "pass"
|
|
and not failed_required
|
|
and not missing_required
|
|
and not failed_named
|
|
and age_check_status == "pass"
|
|
):
|
|
return GateCheck(
|
|
key="docker_smoke_summary",
|
|
label="Docker smoke summary",
|
|
status="pass",
|
|
detail="Docker smoke deploy summary passed required checks."
|
|
+ (f" {age_detail}" if age_detail else ""),
|
|
meta={
|
|
"path": str(path),
|
|
"checked_at": payload.get("checked_at"),
|
|
"server_host": payload.get("server_host"),
|
|
"compose_project_name": payload.get("compose_project_name"),
|
|
**age_meta,
|
|
},
|
|
)
|
|
detail = (
|
|
f"Smoke status={payload.get('status')}; "
|
|
f"failed_required={','.join(failed_required + failed_named) or '-'}; "
|
|
f"missing_required={','.join(missing_required) or '-'}."
|
|
)
|
|
if age_check_status != "pass":
|
|
detail = f"{detail} {age_detail}"
|
|
return GateCheck(
|
|
key="docker_smoke_summary",
|
|
label="Docker smoke summary",
|
|
status="fail",
|
|
detail=detail,
|
|
meta={"path": str(path), **age_meta},
|
|
)
|
|
|
|
|
|
def validate_benchmark_summary(path: Path | None, *, skip: bool, max_age_hours: float) -> GateCheck:
|
|
if skip:
|
|
return GateCheck(
|
|
key="interface_benchmark_summary",
|
|
label="Interface benchmark summary",
|
|
status="skip",
|
|
detail="Interface benchmark summary validation was skipped.",
|
|
required=False,
|
|
)
|
|
if path is None:
|
|
return GateCheck(
|
|
key="interface_benchmark_summary",
|
|
label="Interface benchmark summary",
|
|
status="fail",
|
|
detail="No interface benchmark summary path was provided and no latest summary was found.",
|
|
)
|
|
try:
|
|
payload = read_json(path)
|
|
except Exception as exc:
|
|
return GateCheck(
|
|
key="interface_benchmark_summary",
|
|
label="Interface benchmark summary",
|
|
status="fail",
|
|
detail=f"Unable to read interface benchmark summary: {exc}",
|
|
meta={"path": str(path)},
|
|
)
|
|
scenarios = payload.get("scenarios", [])
|
|
failed = [
|
|
item.get("key") or item.get("label", "unknown")
|
|
for item in scenarios
|
|
if not item.get("passed") or item.get("errors", 0) != 0
|
|
]
|
|
benchmark_checked_at = payload.get("run", {}).get("checked_at")
|
|
age_check_status, age_detail, age_meta = age_status(
|
|
"Interface benchmark summary",
|
|
benchmark_checked_at,
|
|
max_age_hours,
|
|
)
|
|
if payload.get("status") == "pass" and scenarios and not failed and age_check_status == "pass":
|
|
return GateCheck(
|
|
key="interface_benchmark_summary",
|
|
label="Interface benchmark summary",
|
|
status="pass",
|
|
detail=f"Interface benchmark passed {len(scenarios)} scenarios."
|
|
+ (f" {age_detail}" if age_detail else ""),
|
|
meta={
|
|
"path": str(path),
|
|
"checked_at": benchmark_checked_at,
|
|
"requests_per_scenario": payload.get("run", {}).get("requests_per_scenario"),
|
|
"concurrency": payload.get("run", {}).get("concurrency"),
|
|
**age_meta,
|
|
},
|
|
)
|
|
detail = (
|
|
f"Benchmark status={payload.get('status')}; "
|
|
f"failed scenarios={','.join(failed) or '-'}; scenarios={len(scenarios)}."
|
|
)
|
|
if age_check_status != "pass":
|
|
detail = f"{detail} {age_detail}"
|
|
return GateCheck(
|
|
key="interface_benchmark_summary",
|
|
label="Interface benchmark summary",
|
|
status="fail",
|
|
detail=detail,
|
|
meta={"path": str(path), **age_meta},
|
|
)
|
|
|
|
|
|
def run_benchmark(args: argparse.Namespace, artifact_dir: Path) -> tuple[Path | None, GateCheck]:
|
|
benchmark_dir = artifact_dir / "interface-benchmark"
|
|
command = [
|
|
sys.executable,
|
|
str(BENCHMARK_SCRIPT),
|
|
"--contract-api-base",
|
|
args.contract_api_base,
|
|
"--ai-api-base",
|
|
args.ai_api_base,
|
|
"--requests-per-scenario",
|
|
str(args.benchmark_requests_per_scenario),
|
|
"--concurrency",
|
|
str(args.benchmark_concurrency),
|
|
"--warmup",
|
|
str(args.benchmark_warmup),
|
|
"--timeout-seconds",
|
|
str(args.timeout_seconds),
|
|
"--artifact-dir",
|
|
str(benchmark_dir),
|
|
"--tenant-id",
|
|
args.tenant_id,
|
|
"--user-id",
|
|
args.user_id,
|
|
"--contract-proxy-key-env",
|
|
args.contract_proxy_key_env,
|
|
"--ai-service-key-env",
|
|
args.ai_service_key_env,
|
|
]
|
|
if args.include_heavy_explain:
|
|
command.append("--include-heavy-explain")
|
|
log_path = artifact_dir / "run-interface-benchmark.log"
|
|
try:
|
|
completed = subprocess.run(
|
|
command,
|
|
cwd=REPO_ROOT,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
except Exception as exc:
|
|
log_path.write_text(f"Failed to start benchmark: {exc}\n", encoding="utf-8")
|
|
return None, GateCheck(
|
|
key="run_interface_benchmark",
|
|
label="Run interface benchmark",
|
|
status="fail",
|
|
detail=f"Failed to start benchmark: {exc}",
|
|
meta={"log": str(log_path)},
|
|
)
|
|
log_path.write_text(
|
|
"\n".join(
|
|
[
|
|
"$ " + " ".join(command),
|
|
"",
|
|
"## stdout",
|
|
completed.stdout,
|
|
"",
|
|
"## stderr",
|
|
completed.stderr,
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
summary_path = benchmark_dir / "interface-benchmark-summary.json"
|
|
if completed.returncode == 0 and summary_path.exists():
|
|
return summary_path, GateCheck(
|
|
key="run_interface_benchmark",
|
|
label="Run interface benchmark",
|
|
status="pass",
|
|
detail="Interface benchmark command completed.",
|
|
meta={"log": str(log_path), "summary": str(summary_path)},
|
|
)
|
|
return (summary_path if summary_path.exists() else None), GateCheck(
|
|
key="run_interface_benchmark",
|
|
label="Run interface benchmark",
|
|
status="fail",
|
|
detail=f"Interface benchmark command failed with exit code {completed.returncode}.",
|
|
meta={"log": str(log_path), "summary": str(summary_path) if summary_path.exists() else None},
|
|
)
|
|
|
|
|
|
def write_markdown(path: Path, summary: dict[str, Any]) -> None:
|
|
lines: list[str] = []
|
|
lines.append("# V2 Release Quality Gate Summary")
|
|
lines.append("")
|
|
lines.append(f"- Status: {summary['status']}")
|
|
lines.append(f"- Checked at: {summary['checked_at']}")
|
|
lines.append(f"- Artifact dir: {summary['artifact_dir']}")
|
|
lines.append(f"- Web base: {summary['inputs']['web_base_url']}")
|
|
lines.append(f"- Contract API base: {summary['inputs']['contract_api_base']}")
|
|
lines.append(f"- AI Platform API base: {summary['inputs']['ai_api_base']}")
|
|
lines.append("")
|
|
lines.append("## Checks")
|
|
lines.append("| Check | Status | Required | Detail |")
|
|
lines.append("| --- | --- | --- | --- |")
|
|
for check in summary["checks"]:
|
|
lines.append(
|
|
f"| {check['label']} | {check['status']} | {str(check['required']).lower()} | {check['detail']} |"
|
|
)
|
|
lines.append("")
|
|
lines.append("## Sources")
|
|
for key, value in summary["sources"].items():
|
|
lines.append(f"- {key}: {value or '-'}")
|
|
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Run the V2 release quality gate.")
|
|
parser.add_argument("--web-base-url", default="http://localhost:3000")
|
|
parser.add_argument("--contract-api-base", default="http://localhost:8100")
|
|
parser.add_argument("--ai-api-base", default="http://localhost:8101")
|
|
parser.add_argument("--timeout-seconds", type=float, default=20.0)
|
|
parser.add_argument("--artifact-dir", default="")
|
|
parser.add_argument("--smoke-summary", default="")
|
|
parser.add_argument("--benchmark-summary", default="")
|
|
parser.add_argument("--skip-smoke-summary", action="store_true")
|
|
parser.add_argument("--skip-benchmark-summary", action="store_true")
|
|
parser.add_argument("--skip-schema-check", action="store_true")
|
|
parser.add_argument("--max-smoke-summary-age-hours", type=float, default=0)
|
|
parser.add_argument("--max-benchmark-summary-age-hours", type=float, default=0)
|
|
parser.add_argument("--run-benchmark", action="store_true")
|
|
parser.add_argument(
|
|
"--benchmark-requests-per-scenario",
|
|
type=int,
|
|
default=DEFAULT_RELEASE_BENCHMARK_REQUESTS_PER_SCENARIO,
|
|
)
|
|
parser.add_argument(
|
|
"--benchmark-concurrency",
|
|
type=int,
|
|
default=DEFAULT_RELEASE_BENCHMARK_CONCURRENCY,
|
|
)
|
|
parser.add_argument(
|
|
"--benchmark-warmup",
|
|
type=int,
|
|
default=DEFAULT_RELEASE_BENCHMARK_WARMUP,
|
|
)
|
|
parser.add_argument("--include-heavy-explain", action="store_true")
|
|
parser.add_argument("--tenant-id", default="default")
|
|
parser.add_argument("--user-id", default="user-current-admin")
|
|
parser.add_argument("--contract-proxy-key-env", default="CONTRACT_API_WEB_PROXY_SHARED_SECRET")
|
|
parser.add_argument("--ai-service-key-env", default="AI_PLATFORM_SERVICE_SHARED_SECRET")
|
|
parser.add_argument(
|
|
"--allow-ai-provider-warning",
|
|
action="store_true",
|
|
help="Allow only the AI provider_runtime warning for an intentional smoke fallback environment.",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def request_headers(tenant_id: str, user_id: str, auth_header: str, secret: str) -> dict[str, str]:
|
|
headers: dict[str, str] = {}
|
|
if tenant_id.strip():
|
|
headers["X-Tenant-Id"] = tenant_id.strip()
|
|
if user_id.strip():
|
|
headers["X-User-Id"] = user_id.strip()
|
|
if secret:
|
|
headers[auth_header] = secret
|
|
return headers
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
contract_proxy_key = os.getenv(args.contract_proxy_key_env, "").strip()
|
|
ai_service_key = os.getenv(args.ai_service_key_env, "").strip()
|
|
contract_headers = request_headers(
|
|
args.tenant_id,
|
|
args.user_id,
|
|
"X-Contract-Proxy-Key",
|
|
contract_proxy_key,
|
|
)
|
|
ai_headers = request_headers(
|
|
args.tenant_id,
|
|
args.user_id,
|
|
"X-AI-Platform-Service-Key",
|
|
ai_service_key,
|
|
)
|
|
artifact_dir = resolve_path(args.artifact_dir) if args.artifact_dir else DEFAULT_ARTIFACT_ROOT / now_stamp()
|
|
artifact_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
checks: list[GateCheck] = []
|
|
smoke_summary = resolve_path(args.smoke_summary) if args.smoke_summary else latest_summary(
|
|
SMOKE_ARTIFACT_ROOT,
|
|
"v2-smoke-docker-deploy-summary.json",
|
|
)
|
|
benchmark_summary = resolve_path(args.benchmark_summary) if args.benchmark_summary else latest_summary(
|
|
BENCHMARK_ARTIFACT_ROOT,
|
|
"interface-benchmark-summary.json",
|
|
)
|
|
|
|
if not args.skip_schema_check:
|
|
checks.append(
|
|
validate_schema_check(
|
|
"contract_api",
|
|
args.contract_api_base,
|
|
args.timeout_seconds,
|
|
artifact_dir,
|
|
headers=contract_headers,
|
|
)
|
|
)
|
|
checks.append(
|
|
validate_schema_check(
|
|
"ai_platform_api",
|
|
args.ai_api_base,
|
|
args.timeout_seconds,
|
|
artifact_dir,
|
|
headers=ai_headers,
|
|
allowed_warning_checks={"provider_runtime"} if args.allow_ai_provider_warning else set(),
|
|
)
|
|
)
|
|
else:
|
|
checks.append(
|
|
GateCheck(
|
|
key="schema_check",
|
|
label="Runtime schema-check",
|
|
status="skip",
|
|
detail="Runtime schema-check validation was skipped.",
|
|
required=False,
|
|
)
|
|
)
|
|
|
|
checks.append(
|
|
validate_smoke_summary(
|
|
smoke_summary,
|
|
skip=args.skip_smoke_summary,
|
|
max_age_hours=args.max_smoke_summary_age_hours,
|
|
)
|
|
)
|
|
|
|
if args.run_benchmark and not args.skip_benchmark_summary:
|
|
benchmark_summary, benchmark_run_check = run_benchmark(args, artifact_dir)
|
|
checks.append(benchmark_run_check)
|
|
|
|
checks.append(
|
|
validate_benchmark_summary(
|
|
benchmark_summary,
|
|
skip=args.skip_benchmark_summary,
|
|
max_age_hours=args.max_benchmark_summary_age_hours,
|
|
)
|
|
)
|
|
|
|
failed_required = [check for check in checks if check.required and check.status != "pass"]
|
|
status = "pass" if not failed_required else "fail"
|
|
summary = {
|
|
"status": status,
|
|
"checked_at": checked_at(),
|
|
"artifact_dir": str(artifact_dir),
|
|
"inputs": {
|
|
"web_base_url": args.web_base_url,
|
|
"contract_api_base": args.contract_api_base,
|
|
"ai_api_base": args.ai_api_base,
|
|
"run_benchmark": args.run_benchmark,
|
|
"benchmark_requests_per_scenario": args.benchmark_requests_per_scenario,
|
|
"benchmark_concurrency": args.benchmark_concurrency,
|
|
"benchmark_warmup": args.benchmark_warmup,
|
|
"max_smoke_summary_age_hours": args.max_smoke_summary_age_hours,
|
|
"max_benchmark_summary_age_hours": args.max_benchmark_summary_age_hours,
|
|
"tenant_id": args.tenant_id,
|
|
"user_id": args.user_id,
|
|
"contract_proxy_auth_configured": bool(contract_proxy_key),
|
|
"ai_service_auth_configured": bool(ai_service_key),
|
|
"allow_ai_provider_warning": args.allow_ai_provider_warning,
|
|
},
|
|
"sources": {
|
|
"smoke_summary": str(smoke_summary) if smoke_summary else None,
|
|
"benchmark_summary": str(benchmark_summary) if benchmark_summary else None,
|
|
},
|
|
"checks": [check.as_dict() for check in checks],
|
|
}
|
|
summary_json = artifact_dir / "release-quality-gate-summary.json"
|
|
summary_md = artifact_dir / "release-quality-gate-summary.md"
|
|
summary_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
write_markdown(summary_md, summary)
|
|
|
|
print(f"V2_RELEASE_QUALITY_GATE_{status.upper()} {artifact_dir}")
|
|
return 0 if status == "pass" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|