feat(regression): add tests/regression/ E2E canary suite
Some checks failed
continuous-integration/drone/push Build is failing
Some checks failed
continuous-integration/drone/push Build is failing
Three canaries (@pytest.mark.canary) drive the real cold CI lifecycle:
- good-simple: custom-html-tiny @ main (435df8fc) — fast signal, expects GREEN
- good-significant: lasuite-docs @ main (290a8ad7) — multi-service, expects GREEN
- bad-false-green: custom-html @ v5-stale-docroot (71e7326a) — expects RED
Semantic teeth: beyond exit-code, each test asserts that specific named tests
ran in results.json stages (test_serving, test_serving_and_frontend, test_content_type).
If an assertion is removed, the named test disappears → regression test fails.
Includes conftest (run_recipe_ci helper + stage_has_{passing,failing}_test),
README (cadence policy, how to run, how to add), and phase state files.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
102
tests/regression/conftest.py
Normal file
102
tests/regression/conftest.py
Normal file
@ -0,0 +1,102 @@
|
||||
"""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.",
|
||||
)
|
||||
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user