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")
    # The range starts at the last successful deploy, not the previous push.
    assert step["env"] == {"BEFORE": "${{ steps.base.outputs.sha }}"}
    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_base_step_runs_before_the_detector_and_reads_gitea():
    _, wf = _gen()
    ids = [st.get("id") for st in wf["jobs"]["changes"]["steps"]]
    assert ids.index("base") < ids.index("filter")
    base = wf["jobs"]["changes"]["steps"][ids.index("base")]
    assert base["env"] == {
        "GITEA_TOKEN": "${{ secrets.DEPLOY_TOKEN }}",
        "API": "http://git:3000/api/v1",
        "REPO": "${{ github.repository }}",
    }
    assert "actions/runs?status=success&branch=main" in base["run"]


def test_deploy_can_be_dispatched_to_catch_up():
    _, wf = _gen()
    # PyYAML reads the bare `on:` key as True.
    assert "workflow_dispatch" in wf[True]
    assert wf[True]["push"] == {"branches": ["main"]}


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 _clean_env() -> dict[str, str]:
    """os.environ without GIT_* — under a pre-commit hook GIT_DIR and
    GIT_INDEX_FILE point at the real repo, and every scratch-repo command
    here would commit into it instead."""
    return {k: v for k, v in os.environ.items() if not k.startswith("GIT_")}


def _git(*args: str, cwd: Path) -> str:
    # Drop GIT_DIR / GIT_INDEX_FILE / GIT_WORK_TREE: the pre-commit hook runs
    # this file with them pointing at the real repo, which would redirect
    # every call away from the temp repo under `cwd`.
    env = {
        **_clean_env(),
        "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={
            **_clean_env(),
            "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


# ── Behavioural: the last-successful-deploy lookup ──────────────────────


def _gitea_run(run_id: int, path: str, sha: str, conclusion: str) -> str:
    """A run object in Gitea 1.25's field order: nested actor/repository
    objects carry their own ids, and `path` precedes `head_sha`."""
    return (
        f'{{"id":{run_id},"url":"u","html_url":"h","display_title":"t",'
        f'"path":"{path}@refs/heads/main","event":"push","run_attempt":1,'
        f'"run_number":{run_id},"repository_id":1,"head_sha":"{sha}",'
        f'"head_branch":"main","status":"completed",'
        f'"actor":{{"id":2,"login":"kert"}},"trigger_actor":{{"id":2}},'
        f'"repository":{{"id":1,"full_name":"homelab/stack"}},'
        f'"conclusion":"{conclusion}"}}'
    )


def _run_base_step(tmp_path: Path, body: str | None) -> str:
    """Run the generated base step with a fake `curl` on PATH that prints
    `body` (or fails like `curl -f` on an HTTP error when body is None)."""
    _, wf = _gen()
    step = next(s for s in wf["jobs"]["changes"]["steps"] if s.get("id") == "base")
    bindir = tmp_path / "bin"
    bindir.mkdir(parents=True)
    payload = tmp_path / "runs.json"
    if body is None:
        script = "#!/bin/sh\nexit 22\n"
    else:
        payload.write_text(body)
        script = f"#!/bin/sh\ncat '{payload}'\n"
    curl = bindir / "curl"
    curl.write_text(script)
    curl.chmod(0o755)
    out_file = tmp_path / "github_output"
    out_file.write_text("")
    proc = subprocess.run(
        ["bash", "-e", "-c", step["run"]],
        cwd=tmp_path,
        env={
            **_clean_env(),
            "PATH": f"{bindir}:{os.environ['PATH']}",
            "GITEA_TOKEN": "t",
            "API": "http://git:3000/api/v1",
            "REPO": "homelab/stack",
            "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["sha"]


def test_base_is_the_newest_successful_deploy_not_other_workflows(tmp_path):
    a, b = "a" * 40, "b" * 40
    runs = ",".join(
        [
            _gitea_run(9, "ci.yml", "c" * 40, "success"),  # newer, other workflow
            _gitea_run(8, "deploy.yml", a, "success"),
            _gitea_run(5, "deploy.yml", b, "success"),
        ]
    )
    body = f'{{"workflow_runs":[{runs}],"total_count":3}}'
    assert _run_base_step(tmp_path, body) == a


def test_no_successful_deploy_or_api_error_leaves_base_empty(tmp_path):
    only_ci = f'{{"workflow_runs":[{_gitea_run(9, "ci.yml", "c" * 40, "success")}]}}'
    assert _run_base_step(tmp_path / "1", only_ci) == ""
    assert _run_base_step(tmp_path / "2", None) == ""  # curl -f on a 401/500


def test_cancelled_push_paths_carry_into_the_next_run(history, tmp_path):
    """The cancelled-run gap: the push that touched infra/rustfs was
    cancelled, so the last success is still `base`. The next push's own
    range (direct → merge) never touched mc, but diffing from the last
    successful deploy still rebuilds it."""
    for d in ("p", "s"):
        (tmp_path / d).mkdir()
    from_push = _run_detector(
        history["shallow"], history["direct"], history["merge"], tmp_path / "p"
    )
    assert from_push["outputs"]["mc"] == "false"  # the old behaviour's miss
    from_success = _run_detector(
        history["shallow"], history["base"], history["merge"], tmp_path / "s"
    )
    assert from_success["outputs"] == {"api": "true", "mc": "true", "always": "true"}