116 lines
3.7 KiB
Python
116 lines
3.7 KiB
Python
"""Minimal CI gate for qihuaiflow.
|
|
|
|
This is the canonical, single-source-of-truth gate. It runs the two test
|
|
suites that guard production behavior, INSIDE the long-lived qihu-aiflow
|
|
container (our deployment model keeps the app running in a container, not
|
|
built fresh in CI):
|
|
|
|
1. backend pytest -> /app/backend/python/.venv/bin/python -m pytest -c pytest.ini
|
|
2. frontend vitest -> /app/frontend && npx vitest run
|
|
|
|
Exit codes:
|
|
0 all suites green -> deploy/push may proceed
|
|
1 a test suite failed -> BLOCK (do not deploy/push)
|
|
2 environment error -> docker missing or container not running
|
|
|
|
It auto-detects where it runs:
|
|
* If the qihu-aiflow container is reachable via LOCAL docker (i.e. on the
|
|
deploy host), it execs directly.
|
|
* Otherwise it falls back to ssh_helper (remote control from a laptop),
|
|
which reaches the host over SSH and execs into the container there.
|
|
|
|
Usage:
|
|
python scripts/ci_gate.py
|
|
"""
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
CONTAINER = "qihu-aiflow"
|
|
|
|
BACKEND_CMD = "cd /app/backend/python && .venv/bin/python -m pytest -c pytest.ini -q"
|
|
FRONTEND_CMD = "cd /app/frontend && npx vitest run --reporter=dot"
|
|
|
|
|
|
def shell_quote(s: str) -> str:
|
|
return "'" + s.replace("'", "'\\''") + "'"
|
|
|
|
|
|
def local_runner(cmd: str):
|
|
"""Run via local docker (deploy host)."""
|
|
p = subprocess.run(cmd, shell=True, capture_output=True, text=True)
|
|
return p.stdout, p.stderr, p.returncode
|
|
|
|
|
|
def get_runner():
|
|
# Prefer local docker ONLY if the actual container is reachable here
|
|
# (avoids a Windows-docker edge case where `docker ps` works but the
|
|
# qihu-aiflow container lives on a different host).
|
|
try:
|
|
r = subprocess.run(
|
|
"docker ps --format '{{.Names}}'",
|
|
shell=True, capture_output=True, text=True, timeout=10,
|
|
)
|
|
if r.returncode == 0 and CONTAINER in r.stdout.split():
|
|
return local_runner
|
|
except Exception:
|
|
pass
|
|
# Fall back to SSH control of the real host.
|
|
try:
|
|
import ssh_helper as s # type: ignore
|
|
return lambda cmd: s.run(cmd)
|
|
except Exception as e: # pragma: no cover
|
|
print(f"[CI Gate] ERROR: docker unavailable locally and ssh_helper missing: {e}",
|
|
file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
|
|
def run_in_container(runner, cmd: str):
|
|
return runner(f"docker exec {CONTAINER} bash -c {shell_quote(cmd)}")
|
|
|
|
|
|
def container_up(runner) -> bool:
|
|
out, _, _ = runner(f"docker ps --format {shell_quote('{{.Names}}')}")
|
|
return CONTAINER in out.split()
|
|
|
|
|
|
def main():
|
|
runner = get_runner()
|
|
if not container_up(runner):
|
|
print(f"[CI Gate] ERROR: container {CONTAINER} is not running", file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
failures = []
|
|
|
|
print("=== [CI Gate] backend pytest ===")
|
|
o, e, rc = run_in_container(runner, BACKEND_CMD)
|
|
sys.stdout.write(o)
|
|
if e.strip():
|
|
print("STDERR:", e.strip(), file=sys.stderr)
|
|
if rc != 0:
|
|
failures.append("backend pytest")
|
|
print(f"[CI Gate] backend pytest FAILED (rc={rc})", file=sys.stderr)
|
|
else:
|
|
print("[CI Gate] backend pytest OK")
|
|
|
|
print("=== [CI Gate] frontend vitest ===")
|
|
o, e, rc = run_in_container(runner, FRONTEND_CMD)
|
|
sys.stdout.write(o)
|
|
if e.strip():
|
|
print("STDERR:", e.strip(), file=sys.stderr)
|
|
if rc != 0:
|
|
failures.append("frontend vitest")
|
|
print(f"[CI Gate] frontend vitest FAILED (rc={rc})", file=sys.stderr)
|
|
else:
|
|
print("[CI Gate] frontend vitest OK")
|
|
|
|
if failures:
|
|
print(f"\n[CI Gate] BLOCKED - failed: {', '.join(failures)}", file=sys.stderr)
|
|
sys.exit(1)
|
|
print("\n[CI Gate] PASSED - all suites green")
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|