cf405b4195
Some checks failed
continuous-integration/drone/push Build is failing
Four new per-tier RED canaries prove the server catches failure at every lifecycle tier: - bad-install: custom-html-tiny @ regression-bad-image (4ae88661) nonexistent image → prepull fails → install=fail STAGES=install → no prev-version lookup → chaos deploy of HEAD - bad-upgrade: same branch + SHA, STAGES=install,upgrade install uses prev-version (good image) → PASS upgrade chaos checks out HEAD (bad image) → prepull fails → FAIL - bad-backup: custom-html @ regression-bad-backup (e1e3c5fc) backupbot.backup.path=/nonexistent-path-cc-ci-canary-bad abra app backup create fails → backup=fail - bad-restore: custom-html @ regression-bad-restore (5a481cc1) backup targets .backup-data/ subdir (not where ci-marker.txt lives) backup succeeds; restore puts .backup-data back but NOT the marker marker stays "mutated" → test_restore_returns_state FAILS → restore=fail Each test asserts: rc!=0, failing_tier="fail", prior tiers="pass". Adds @pytest.mark.canary_fast for the fast subset. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
107 lines
3.4 KiB
Python
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
|