Files
cc-ci/tests/regression/conftest.py
autonomic-bot 73427d6e36
Some checks failed
continuous-integration/drone/push Build is failing
feat(regression): add E2E canary regression suite (tests/regression/)
Seven canaries prove both halves of the server's job:
- GREEN: good apps are reported healthy (good-simple + good-significant)
- RED: broken apps are caught at intended tier (false-green guard + 4 per-tier)

Fixtures: custom-html-bkp-bad (backup tier RED) + custom-html-rst-bad (restore tier RED).
All 7 canaries verified on live server (see STATUS-regression.md for artifacts).

Not wired to per-commit CI — run on-demand: pytest -m canary tests/regression/

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-02 03:33:24 +00:00

107 lines
3.4 KiB
Python

"""Shared fixtures and helpers for E2E canary regression tests.
The regression tests call the real cc-ci harness (run_recipe_ci.py) as a subprocess and assert on
its outputs (exit code, results.json). They run ON the cc-ci server, not the orchestrator — abra,
Docker, and Swarm must be present.
Invoke: cc-ci-run python -m pytest tests/regression/ -m canary -v
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
import time
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
def pytest_configure(config):
config.addinivalue_line(
"markers",
"canary: slow E2E canary test — drives the full cold CI lifecycle; run on-demand only.",
)
config.addinivalue_line(
"markers",
"canary_fast: fast per-tier RED canary (still tagged canary); subset for quick pre-merge checks.",
)
def run_recipe_ci(
recipe: str,
src: str,
ref: str,
pr: str = "0",
stages: str = "install,upgrade,backup,restore,custom",
runs_dir: str | None = None,
run_id_prefix: str = "regression",
timeout: int = 3600,
) -> tuple[int, dict | None, str]:
"""Invoke run_recipe_ci.py with the given canary params.
Returns (rc, results_dict_or_None, run_artifact_dir).
Stdout/stderr stream live so a human can follow progress.
"""
ts = int(time.time())
run_id = f"{run_id_prefix}-{recipe}-{ref[:12]}-{ts}"
if runs_dir is None:
runs_dir = "/var/lib/cc-ci-runs"
env = dict(os.environ)
env.update(
{
"RECIPE": recipe,
"REF": ref,
"SRC": src,
"PR": pr,
"STAGES": stages,
"CCCI_RUN_ID": run_id,
"CCCI_RUNS_DIR": runs_dir,
"HOME": "/root",
}
)
# Keep PLAYWRIGHT env from the outer cc-ci-run wrapper (already in os.environ if running under it)
script = os.path.join(ROOT, "runner", "run_recipe_ci.py")
result = subprocess.run(
[sys.executable, script],
env=env,
timeout=timeout,
)
rc = result.returncode
artifact_dir = os.path.join(runs_dir, run_id)
results_path = os.path.join(artifact_dir, "results.json")
results_data: dict | None = None
if os.path.exists(results_path):
with open(results_path) as f:
results_data = json.load(f)
return rc, results_data, artifact_dir
def find_stage_tests(results: dict, stage_name: str) -> list[dict]:
"""Return the per-test list for a named stage from results.json, or []."""
for stage in results.get("stages", []):
if stage.get("name") == stage_name:
return stage.get("tests", [])
return []
def stage_has_passing_test(results: dict, stage_name: str, test_name_substr: str) -> bool:
"""True if the named stage contains a passing test whose name includes test_name_substr."""
for t in find_stage_tests(results, stage_name):
if test_name_substr in t.get("name", "") and t.get("status") == "pass":
return True
return False
def stage_has_failing_test(results: dict, stage_name: str, test_name_substr: str) -> bool:
"""True if the named stage contains a failing test whose name includes test_name_substr."""
for t in find_stage_tests(results, stage_name):
if test_name_substr in t.get("name", "") and t.get("status") in ("fail", "error"):
return True
return False