US08-05: Automate Container Deployment Acceptance (#100)
This commit was merged in pull request #100.
This commit is contained in:
147
tests/integration/test_container_gate.py
Normal file
147
tests/integration/test_container_gate.py
Normal file
@@ -0,0 +1,147 @@
|
||||
"""US08-05: the container acceptance gate's own contract, checked without a daemon.
|
||||
|
||||
The gate itself needs Docker, a browser, and several minutes; running it from inside
|
||||
the suite would be a fork bomb with better manners. What is checkable offline is what
|
||||
makes it a *gate* rather than a long test run:
|
||||
|
||||
* it selects the container journeys by marker, so adding one is enough to put it in
|
||||
front of a deploy;
|
||||
* it accepts no skip at all — every reason a check would skip here (no daemon, no
|
||||
compose plugin, no browser) means the deployed runtime was not proven;
|
||||
* it retains checksummed evidence per run;
|
||||
* and CI runs it on `main`, which is what the publish step waits for.
|
||||
|
||||
The running proof is ``tests/e2e/test_phase_h_container.py``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.services import release
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
SUITE = REPO / "tests" / "e2e" / "test_phase_h_container.py"
|
||||
TEST_WORKFLOW = yaml.safe_load((REPO / ".gitea" / "workflows" / "test.yml").read_text())
|
||||
|
||||
|
||||
def _config(tmp_path) -> Config:
|
||||
for name in ("data", "lib"):
|
||||
(tmp_path / name).mkdir(exist_ok=True)
|
||||
return Config.from_env(
|
||||
{
|
||||
"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
|
||||
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(tmp_path / "lib"),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
# ── what the gate runs ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_the_gate_selects_the_container_journeys_by_their_marker():
|
||||
assert release.CONTAINER_STAGES == (("container", ("tests/e2e", "-m", "phase_h")),)
|
||||
declared = [line for line in (REPO / "pyproject.toml").read_text().splitlines()
|
||||
if line.strip().startswith('"phase_h:')]
|
||||
assert declared, "an unregistered marker selects nothing and fails no gate"
|
||||
|
||||
|
||||
def test_every_journey_in_the_suite_carries_the_marker():
|
||||
"""A test in that file without the marker is a check the gate never runs."""
|
||||
source = SUITE.read_text()
|
||||
assert "pytestmark = [pytest.mark.phase_h, pytest.mark.container]" in source
|
||||
journeys = re.findall(r"^def (test_[a-z_]+)", source, re.MULTILINE)
|
||||
assert len(journeys) >= 4, journeys
|
||||
for required in ("browser", "upgrade", "kill", "secret"):
|
||||
assert any(required in name for name in journeys), (required, journeys)
|
||||
|
||||
|
||||
# ── no skip is an environment limit here ─────────────────────────────────────
|
||||
|
||||
|
||||
def test_the_gate_accepts_no_skipped_check_at_all():
|
||||
assert release.CONTAINER_ALLOWED_SKIP_REASONS == ()
|
||||
docker_missing = release.StageResult(
|
||||
"container", [], 0, 0.1, "1 skipped", ["SKIPPED [1] x.py:1: no Docker daemon available"]
|
||||
)
|
||||
# The release gate tolerates exactly this one, because the image definition is
|
||||
# still checked offline. The container gate cannot: it is the deployment's proof.
|
||||
assert release.unexpected_skips([docker_missing]) == []
|
||||
assert release.unexpected_skips(
|
||||
[docker_missing], release.CONTAINER_ALLOWED_SKIP_REASONS
|
||||
) == ["SKIPPED [1] x.py:1: no Docker daemon available"]
|
||||
|
||||
|
||||
def test_a_skipped_check_fails_the_run_and_the_evidence_says_so(tmp_path):
|
||||
skipping = tmp_path / "test_skipping.py"
|
||||
skipping.write_text(
|
||||
"import pytest\n\n"
|
||||
"def test_x():\n"
|
||||
" pytest.skip('no Docker daemon available')\n"
|
||||
)
|
||||
evidence = tmp_path / "evidence"
|
||||
|
||||
report = release.run_gate(
|
||||
_config(tmp_path),
|
||||
output=evidence,
|
||||
stages=(("container", (str(skipping),)),),
|
||||
allowed_skips=release.CONTAINER_ALLOWED_SKIP_REASONS,
|
||||
)
|
||||
|
||||
assert report["ok"] is False
|
||||
assert report["failures"] == [], "the stage passed; the skip is what fails the gate"
|
||||
assert report["unexpected_skips"], report
|
||||
written = json.loads((evidence / release.REPORT_NAME).read_text())
|
||||
assert written["ok"] is False
|
||||
assert (evidence / "logs" / "container.log").exists()
|
||||
for line in (evidence / release.CHECKSUMS_NAME).read_text().splitlines():
|
||||
digest, name = line.split(" ", 1)
|
||||
assert release.sha256_file(evidence / name) == digest
|
||||
|
||||
|
||||
def test_the_same_run_passes_when_nothing_skips(tmp_path):
|
||||
passing = tmp_path / "test_passing.py"
|
||||
passing.write_text("def test_x():\n assert True\n")
|
||||
|
||||
report = release.run_gate(
|
||||
_config(tmp_path),
|
||||
output=tmp_path / "evidence",
|
||||
stages=(("container", (str(passing),)),),
|
||||
allowed_skips=release.CONTAINER_ALLOWED_SKIP_REASONS,
|
||||
)
|
||||
|
||||
assert report["ok"] is True and report["unexpected_skips"] == []
|
||||
assert report["revision"], "the evidence must say which commit it covers"
|
||||
|
||||
|
||||
# ── the command, and CI ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_the_command_is_documented_and_wired():
|
||||
from photo_pipeline.__main__ import main # noqa: F401 (import proves it loads)
|
||||
|
||||
assert "container-gate" in (REPO / "photo_pipeline" / "__main__.py").read_text()
|
||||
assert "container-gate" in (REPO / "README.md").read_text()
|
||||
|
||||
|
||||
def test_ci_runs_the_gate_on_main_and_keeps_its_evidence():
|
||||
job = TEST_WORKFLOW["jobs"]["container"]
|
||||
assert job["if"] == "gitea.event_name == 'push'", "pull requests have nothing to upgrade from"
|
||||
script = "\n".join(step["run"] for step in job["steps"] if "run" in step)
|
||||
assert "photo_pipeline container-gate" in script
|
||||
evidence = next(step for step in job["steps"] if "upload-artifact" in str(step.get("uses")))
|
||||
assert evidence["if"] == "always()", "a failed gate's logs are the ones worth keeping"
|
||||
assert evidence["with"]["path"] == "gate-evidence"
|
||||
|
||||
|
||||
def test_the_upgrade_journey_can_still_reach_the_previous_version():
|
||||
"""Without depth, `git archive HEAD~1` has nothing to build."""
|
||||
checkout = next(
|
||||
step for step in TEST_WORKFLOW["jobs"]["container"]["steps"] if "checkout" in str(step.get("uses"))
|
||||
)
|
||||
assert checkout["with"]["fetch-depth"] >= 2
|
||||
Reference in New Issue
Block a user