|
|
|
@@ -0,0 +1,233 @@
|
|
|
|
|
#!/usr/bin/env python3
|
|
|
|
|
"""Run project verification and retain machine-readable test evidence."""
|
|
|
|
|
|
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
|
|
|
|
import argparse
|
|
|
|
|
import json
|
|
|
|
|
import os
|
|
|
|
|
import shutil
|
|
|
|
|
import subprocess
|
|
|
|
|
import sys
|
|
|
|
|
import time
|
|
|
|
|
import urllib.request
|
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
|
from pathlib import Path
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
BACKEND = ROOT / "backend"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def executable(name: str) -> str:
|
|
|
|
|
found = shutil.which(name)
|
|
|
|
|
if not found:
|
|
|
|
|
raise RuntimeError(f"Required executable is unavailable: {name}")
|
|
|
|
|
return found
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def project_python() -> str:
|
|
|
|
|
candidates = [
|
|
|
|
|
BACKEND / "venv" / "Scripts" / "python.exe",
|
|
|
|
|
BACKEND / "venv" / "bin" / "python",
|
|
|
|
|
]
|
|
|
|
|
return str(next((p for p in candidates if p.exists()), Path(sys.executable)))
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def git_output(*args: str) -> str:
|
|
|
|
|
result = subprocess.run(
|
|
|
|
|
[executable("git"), *args], cwd=ROOT, text=True,
|
|
|
|
|
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, check=False,
|
|
|
|
|
)
|
|
|
|
|
return result.stdout.strip()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def changed_files(base: str | None) -> list[str]:
|
|
|
|
|
command = ["diff", "--name-only"]
|
|
|
|
|
if base:
|
|
|
|
|
command.append(f"{base}...HEAD")
|
|
|
|
|
files = set(filter(None, git_output(*command).splitlines()))
|
|
|
|
|
files.update(filter(None, git_output("ls-files", "--others", "--exclude-standard").splitlines()))
|
|
|
|
|
return sorted(files)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def make_plan(files: list[str], docker_runtime: bool) -> list[dict[str, str]]:
|
|
|
|
|
plan = [
|
|
|
|
|
{"case": "repository-diff", "reason": "Reject whitespace and patch formatting errors"},
|
|
|
|
|
{"case": "backend-unit", "reason": "Run all deterministic API and agent unit tests"},
|
|
|
|
|
{"case": "frontend-build", "reason": "Type-check and create the production Next.js bundle"},
|
|
|
|
|
{"case": "compose-config", "reason": "Validate local and Alibaba Cloud Compose manifests"},
|
|
|
|
|
]
|
|
|
|
|
if any(path.startswith("backend/") for path in files):
|
|
|
|
|
plan.append({"case": "backend-change-coverage", "reason": "Backend files changed; full backend suite is mandatory"})
|
|
|
|
|
if any(path.startswith("src/") for path in files):
|
|
|
|
|
plan.append({"case": "frontend-change-coverage", "reason": "Frontend files changed; production build is mandatory"})
|
|
|
|
|
if any("docker" in path.lower() or path.startswith("deploy/") for path in files):
|
|
|
|
|
plan.append({"case": "deployment-change-coverage", "reason": "Deployment files changed; both manifests are validated"})
|
|
|
|
|
if docker_runtime:
|
|
|
|
|
plan.append({"case": "docker-runtime-smoke", "reason": "Build, start and probe the real container"})
|
|
|
|
|
return plan
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def run_case(name: str, command: list[str], cwd: Path, evidence_dir: Path) -> dict:
|
|
|
|
|
started = time.monotonic()
|
|
|
|
|
result = subprocess.run(
|
|
|
|
|
command, cwd=cwd, text=True, encoding="utf-8", errors="replace",
|
|
|
|
|
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False,
|
|
|
|
|
)
|
|
|
|
|
duration = round(time.monotonic() - started, 3)
|
|
|
|
|
log_path = evidence_dir / f"{name}.log"
|
|
|
|
|
log_path.write_text(result.stdout, encoding="utf-8")
|
|
|
|
|
print(f"[{name}] {'PASS' if result.returncode == 0 else 'FAIL'} ({duration}s)")
|
|
|
|
|
return {
|
|
|
|
|
"name": name,
|
|
|
|
|
"status": "passed" if result.returncode == 0 else "failed",
|
|
|
|
|
"exit_code": result.returncode,
|
|
|
|
|
"duration_seconds": duration,
|
|
|
|
|
"command": command,
|
|
|
|
|
"working_directory": str(cwd),
|
|
|
|
|
"log": log_path.name,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def wait_for(url: str, timeout: int = 150) -> None:
|
|
|
|
|
deadline = time.monotonic() + timeout
|
|
|
|
|
last_error = "not attempted"
|
|
|
|
|
while time.monotonic() < deadline:
|
|
|
|
|
try:
|
|
|
|
|
with urllib.request.urlopen(url, timeout=5) as response:
|
|
|
|
|
if 200 <= response.status < 400:
|
|
|
|
|
return
|
|
|
|
|
except Exception as exc: # evidence captures the eventual timeout
|
|
|
|
|
last_error = str(exc)
|
|
|
|
|
time.sleep(2)
|
|
|
|
|
raise RuntimeError(f"Timed out waiting for {url}: {last_error}")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def docker_smoke(docker: str, evidence_dir: Path, keep: bool) -> dict:
|
|
|
|
|
started = time.monotonic()
|
|
|
|
|
output: list[str] = []
|
|
|
|
|
status = "passed"
|
|
|
|
|
exit_code = 0
|
|
|
|
|
compose = [docker, "compose", "-f", str(ROOT / "docker-compose.yml")]
|
|
|
|
|
try:
|
|
|
|
|
for args in (["build"], ["up", "-d", "--wait"]):
|
|
|
|
|
result = subprocess.run(compose + args, cwd=ROOT, text=True, encoding="utf-8",
|
|
|
|
|
errors="replace", stdout=subprocess.PIPE,
|
|
|
|
|
stderr=subprocess.STDOUT, check=False)
|
|
|
|
|
output.append(f"$ {' '.join(compose + args)}\n{result.stdout}")
|
|
|
|
|
if result.returncode:
|
|
|
|
|
raise RuntimeError(f"Docker Compose {' '.join(args)} failed with {result.returncode}")
|
|
|
|
|
wait_for("http://127.0.0.1:3200/")
|
|
|
|
|
wait_for("http://127.0.0.1:8002/api/xingbang/health")
|
|
|
|
|
output.append("HTTP smoke checks passed for frontend and backend")
|
|
|
|
|
except Exception as exc:
|
|
|
|
|
status, exit_code = "failed", 1
|
|
|
|
|
output.append(f"ERROR: {exc}")
|
|
|
|
|
finally:
|
|
|
|
|
ps = subprocess.run(compose + ["ps"], cwd=ROOT, text=True, encoding="utf-8",
|
|
|
|
|
errors="replace", stdout=subprocess.PIPE,
|
|
|
|
|
stderr=subprocess.STDOUT, check=False)
|
|
|
|
|
output.append(f"$ {' '.join(compose + ['ps'])}\n{ps.stdout}")
|
|
|
|
|
if not keep:
|
|
|
|
|
down = subprocess.run(compose + ["down"], cwd=ROOT, text=True, encoding="utf-8",
|
|
|
|
|
errors="replace", stdout=subprocess.PIPE,
|
|
|
|
|
stderr=subprocess.STDOUT, check=False)
|
|
|
|
|
output.append(f"$ {' '.join(compose + ['down'])}\n{down.stdout}")
|
|
|
|
|
log_path = evidence_dir / "docker-runtime-smoke.log"
|
|
|
|
|
log_path.write_text("\n\n".join(output), encoding="utf-8")
|
|
|
|
|
duration = round(time.monotonic() - started, 3)
|
|
|
|
|
print(f"[docker-runtime-smoke] {status.upper()} ({duration}s)")
|
|
|
|
|
return {"name": "docker-runtime-smoke", "status": status, "exit_code": exit_code,
|
|
|
|
|
"duration_seconds": duration, "command": compose, "working_directory": str(ROOT),
|
|
|
|
|
"log": log_path.name}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def write_report(evidence_dir: Path, summary: dict) -> None:
|
|
|
|
|
(evidence_dir / "summary.json").write_text(
|
|
|
|
|
json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8"
|
|
|
|
|
)
|
|
|
|
|
rows = "\n".join(
|
|
|
|
|
f"| {item['name']} | {item['status']} | {item['duration_seconds']}s | [{item['log']}]({item['log']}) |"
|
|
|
|
|
for item in summary["results"]
|
|
|
|
|
)
|
|
|
|
|
report = f"""# Verification Evidence
|
|
|
|
|
|
|
|
|
|
- Run ID: `{summary['run_id']}`
|
|
|
|
|
- Commit: `{summary['commit']}`
|
|
|
|
|
- Branch: `{summary['branch']}`
|
|
|
|
|
- Started: `{summary['started_at']}`
|
|
|
|
|
- Result: **{summary['status'].upper()}**
|
|
|
|
|
|
|
|
|
|
## Generated test plan
|
|
|
|
|
|
|
|
|
|
```json
|
|
|
|
|
{json.dumps(summary['test_plan'], ensure_ascii=False, indent=2)}
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
## Results
|
|
|
|
|
|
|
|
|
|
| Check | Result | Duration | Evidence |
|
|
|
|
|
| --- | --- | ---: | --- |
|
|
|
|
|
{rows}
|
|
|
|
|
|
|
|
|
|
## Changed files used to generate the plan
|
|
|
|
|
|
|
|
|
|
```text
|
|
|
|
|
{os.linesep.join(summary['changed_files']) or '(clean working tree)'}
|
|
|
|
|
```
|
|
|
|
|
"""
|
|
|
|
|
(evidence_dir / "REPORT.md").write_text(report, encoding="utf-8")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def main() -> int:
|
|
|
|
|
parser = argparse.ArgumentParser()
|
|
|
|
|
parser.add_argument("--base", help="Git base revision used to generate the change-aware test plan")
|
|
|
|
|
parser.add_argument("--docker", action="store_true", help="Build and smoke-test the local Docker deployment")
|
|
|
|
|
parser.add_argument("--keep-docker", action="store_true", help="Keep containers running after Docker smoke tests")
|
|
|
|
|
args = parser.parse_args()
|
|
|
|
|
|
|
|
|
|
run_id = datetime.now().strftime("%Y%m%d-%H%M%S")
|
|
|
|
|
evidence_dir = ROOT / "test-evidence" / f"{run_id}-{git_output('rev-parse', '--short', 'HEAD') or 'uncommitted'}"
|
|
|
|
|
evidence_dir.mkdir(parents=True, exist_ok=False)
|
|
|
|
|
files = changed_files(args.base)
|
|
|
|
|
summary = {
|
|
|
|
|
"run_id": run_id,
|
|
|
|
|
"started_at": datetime.now(timezone.utc).isoformat(),
|
|
|
|
|
"commit": git_output("rev-parse", "HEAD"),
|
|
|
|
|
"branch": git_output("branch", "--show-current"),
|
|
|
|
|
"changed_files": files,
|
|
|
|
|
"test_plan": make_plan(files, args.docker),
|
|
|
|
|
"results": [],
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
git = executable("git")
|
|
|
|
|
npm = executable("npm")
|
|
|
|
|
docker = executable("docker")
|
|
|
|
|
checks = [
|
|
|
|
|
("repository-diff", [git, "diff", "--check"], ROOT),
|
|
|
|
|
("backend-unit", [project_python(), "-m", "unittest", "discover", "-s", ".", "-p", "test_*.py", "-v"], BACKEND),
|
|
|
|
|
("frontend-build", [npm, "run", "build"], ROOT),
|
|
|
|
|
("compose-local-config", [docker, "compose", "-f", "docker-compose.yml", "config", "--quiet"], ROOT),
|
|
|
|
|
("compose-aliyun-config", [docker, "compose", "--env-file", "deploy/aliyun.env.example", "-f", "deploy/docker-compose.aliyun.yml", "config", "--quiet"], ROOT),
|
|
|
|
|
]
|
|
|
|
|
for name, command, cwd in checks:
|
|
|
|
|
summary["results"].append(run_case(name, command, cwd, evidence_dir))
|
|
|
|
|
if args.docker:
|
|
|
|
|
summary["results"].append(docker_smoke(docker, evidence_dir, args.keep_docker))
|
|
|
|
|
except Exception as exc:
|
|
|
|
|
error_log = evidence_dir / "runner-error.log"
|
|
|
|
|
error_log.write_text(str(exc), encoding="utf-8")
|
|
|
|
|
summary["results"].append({"name": "runner", "status": "failed", "exit_code": 1,
|
|
|
|
|
"duration_seconds": 0, "command": [], "working_directory": str(ROOT),
|
|
|
|
|
"log": error_log.name})
|
|
|
|
|
|
|
|
|
|
summary["status"] = "passed" if all(x["status"] == "passed" for x in summary["results"]) else "failed"
|
|
|
|
|
write_report(evidence_dir, summary)
|
|
|
|
|
print(f"Evidence: {evidence_dir / 'REPORT.md'}")
|
|
|
|
|
return 0 if summary["status"] == "passed" else 1
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
|
raise SystemExit(main())
|
NathanAlex7