stack/tests/dev/test_gen_config_deploy.py

"""Tests for the generated Deploy workflow's change detection.

``dev/scripts/backends/gitea.py::_gen_deploy`` used to gate every image
job on ``contains(github.event.head_commit.modified, ...)``. That list
only describes the tip commit of a push and is empty for merge commits,
so a PR landed with a merge commit never rebuilt any image. The workflow
now has a ``changes`` job that diffs the pushed range (``before`` → the
new tip) and the image jobs gate on its outputs.

The shape tests guard the generated YAML; the behavioural tests run the
generated detector script in a scratch git repository so the shell logic
is exercised for real (merge commit, direct push, unknown previous tip).
"""

from __future__ import annotations

import os
import subprocess
import sys
from pathlib import Path

import pytest
import yaml

_DEV_SCRIPTS = Path(__file__).resolve().parents[2] / "dev" / "scripts"
if str(_DEV_SCRIPTS) not in sys.path:
    sys.path.insert(0, str(_DEV_SCRIPTS))

from backends import gitea  # noqa: E402

_IMAGES = [
    {
        "name": "api",
        "dockerfile": "infra/images/api.Dockerfile",
        "context": ".",
        "path_filter": ["infra/images/api.Dockerfile", "src/**", "pyproject.toml"],
    },
    {
        "name": "mc",
        "dockerfile": "infra/images/mc.Dockerfile",
        "context": "infra/rustfs/",
        "path_filter": ["infra/rustfs/**", "infra/images/mc.Dockerfile"],
    },
    {
        "name": "always",
        "dockerfile": "infra/images/always.Dockerfile",
        "context": ".",
    },
]


def _gen() -> tuple[str, dict]:
    path, content = gitea._gen_deploy(
        _IMAGES,
        [_IMAGES[0]],
        "git:3000",
        "homelab/stack",
        "ubuntu-latest",
        "latest",
    )
    return path, yaml.safe_load(content)


def test_case_glob_collapses_double_star():
    assert gitea._case_glob("src/**") == "src/*"
    assert gitea._case_glob("infra/rustfs/**") == "infra/rustfs/*"
    assert gitea._case_glob("pyproject.toml") == "pyproject.toml"
    assert gitea._case_glob("a/**/**/b") == "a/*/*/b"


def test_changes_job_exposes_one_output_per_image():
    path, wf = _gen()
    assert path == ".gitea/workflows/deploy.yml"
    changes = wf["jobs"]["changes"]
    assert changes["outputs"] == {
        "api": "${{ steps.filter.outputs.api }}",
        "mc": "${{ steps.filter.outputs.mc }}",
        "always": "${{ steps.filter.outputs.always }}",
    }
    step = next(s for s in changes["steps"] if s.get("id") == "filter")
    assert step["env"] == {"BEFORE": "${{ github.event.before }}"}
    assert (
        "match api 'infra/images/api.Dockerfile' 'src/*' 'pyproject.toml'"
        in (step["run"])
    )
    assert "match mc 'infra/rustfs/*' 'infra/images/mc.Dockerfile'" in step["run"]
    # An image without path filters follows every push.
    assert 'echo "always=true" >> "$GITHUB_OUTPUT"' in step["run"]


def test_image_jobs_gate_on_changes_outputs_not_head_commit():
    _, wf = _gen()
    for name in ("api", "mc", "always"):
        job = wf["jobs"][name]
        assert job["needs"] == ["changes"]
        assert job["if"] == f"needs.changes.outputs.{name} == 'true'"
    assert wf["jobs"]["report"]["needs"] == ["changes", "api", "mc", "always"]
    _, content = gitea._gen_deploy(
        _IMAGES, [], "git:3000", "homelab/stack", "ubuntu-latest", "latest"
    )
    assert "contains(github.event.head_commit" not in content


# ── Behavioural: run the detector script against real git history ──────


def _git(*args: str, cwd: Path) -> str:
    env = {
        **os.environ,
        "GIT_AUTHOR_NAME": "t",
        "GIT_AUTHOR_EMAIL": "t@example.invalid",
        "GIT_COMMITTER_NAME": "t",
        "GIT_COMMITTER_EMAIL": "t@example.invalid",
    }
    return subprocess.run(
        ["git", *args], cwd=cwd, env=env, check=True, capture_output=True, text=True
    ).stdout.strip()


def _commit_file(repo: Path, rel: str, msg: str) -> str:
    target = repo / rel
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(f"{msg}\n")
    _git("add", "-A", cwd=repo)
    _git("commit", "-q", "-m", msg, cwd=repo)
    return _git("rev-parse", "HEAD", cwd=repo)


@pytest.fixture
def history(tmp_path: Path) -> dict[str, object]:
    """A bare 'origin' whose main has: base → (direct push) → merge commit.

    Returns the SHAs a push event would carry for each push plus a
    shallow clone of the final main, as the runner's checkout leaves it.
    """
    origin = tmp_path / "origin.git"
    _git("init", "-q", "--bare", "-b", "main", str(origin), cwd=tmp_path)
    # Shallow fetches by raw SHA need the server's consent; Gitea grants
    # it via protocol v2, the local file transport must be told.
    _git("config", "uploadpack.allowAnySHA1InWant", "true", cwd=origin)

    work = tmp_path / "work"
    _git("clone", "-q", str(origin), str(work), cwd=tmp_path)
    base = _commit_file(work, "README.md", "base")
    _commit_file(work, "docs/index.md", "docs only")
    direct = _commit_file(work, "infra/rustfs/policy.json", "mc config")
    _git("push", "-q", "origin", "main", cwd=work)

    _git("checkout", "-q", "-b", "feature", cwd=work)
    _commit_file(work, "src/llm/rag.py", "feature: rag")
    _commit_file(work, "tests/llm/test_rag.py", "feature: test")
    _git("checkout", "-q", "main", cwd=work)
    _git("merge", "-q", "--no-ff", "-m", "merge feature", "feature", cwd=work)
    merge = _git("rev-parse", "HEAD", cwd=work)
    _git("push", "-q", "origin", "main", cwd=work)

    shallow = tmp_path / "shallow"
    _git("clone", "-q", "--depth", "1", str(origin), str(shallow), cwd=tmp_path)
    assert _git("rev-parse", "HEAD", cwd=shallow) == merge

    return {"base": base, "direct": direct, "merge": merge, "shallow": shallow}


def _run_detector(checkout: Path, before: str, sha: str, tmp_path: Path) -> dict:
    _, wf = _gen()
    step = next(s for s in wf["jobs"]["changes"]["steps"] if s.get("id") == "filter")
    out_file = tmp_path / "github_output"
    out_file.write_text("")
    proc = subprocess.run(
        ["bash", "-e", "-c", step["run"]],
        cwd=checkout,
        env={
            **os.environ,
            "BEFORE": before,
            "GITHUB_SHA": sha,
            "GITHUB_OUTPUT": str(out_file),
        },
        capture_output=True,
        text=True,
    )
    assert proc.returncode == 0, proc.stdout + proc.stderr
    outputs = dict(
        line.split("=", 1) for line in out_file.read_text().splitlines() if line
    )
    return {"outputs": outputs, "log": proc.stdout}


def test_merge_commit_push_rebuilds_images_the_branch_touched(history, tmp_path):
    """The push that lands a merge commit must see every file the branch
    brought in, which `head_commit.modified` never did."""
    res = _run_detector(
        history["shallow"], history["direct"], history["merge"], tmp_path
    )
    assert res["outputs"] == {"api": "true", "mc": "false", "always": "true"}
    assert "src/llm/rag.py" in res["log"]
    assert "tests/llm/test_rag.py" in res["log"]


def test_multi_commit_direct_push_covers_the_whole_range(history, tmp_path):
    """A push of several plain commits is judged on all of them, not
    only the tip (the tip here touched infra/rustfs, an earlier one docs)."""
    shallow = history["shallow"]
    res = _run_detector(shallow, history["base"], history["direct"], tmp_path)
    assert res["outputs"] == {"api": "false", "mc": "true", "always": "true"}
    assert "docs/index.md" in res["log"]


def test_unknown_previous_tip_builds_everything(history, tmp_path):
    for before in ("", "0" * 40, "f" * 40):
        res = _run_detector(history["shallow"], before, history["merge"], tmp_path)
        assert res["outputs"] == {"api": "true", "mc": "true", "always": "true"}, before