awoooi/ops/runner/test_select_latest_inclusive_cd_carrier.py

from __future__ import annotations

import importlib.util
import subprocess
import sys
from pathlib import Path

import pytest


ROOT = Path(__file__).resolve().parents[2]
SCRIPT = ROOT / "ops/runner/select-latest-inclusive-cd-carrier.py"


def _load_module():
    spec = importlib.util.spec_from_file_location("select_latest_inclusive_cd_carrier", SCRIPT)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


def _git(cwd: Path, *args: str) -> str:
    return subprocess.run(
        ["git", *args],
        cwd=cwd,
        check=True,
        capture_output=True,
        text=True,
    ).stdout.strip()


@pytest.fixture
def linear_remote(tmp_path: Path) -> tuple[Path, Path, str, str]:
    source = tmp_path / "source"
    remote = tmp_path / "remote.git"
    checkout = tmp_path / "checkout"
    source.mkdir()
    _git(source, "init", "-b", "main")
    _git(source, "config", "user.name", "Carrier Test")
    _git(source, "config", "user.email", "carrier@example.invalid")
    (source / "value.txt").write_text("one\n", encoding="utf-8")
    _git(source, "add", "value.txt")
    _git(source, "commit", "-m", "first")
    first = _git(source, "rev-parse", "HEAD")
    _git(tmp_path, "clone", "--bare", str(source), str(remote))
    _git(tmp_path, "clone", "--depth=1", remote.as_uri(), str(checkout))

    (source / "value.txt").write_text("two\n", encoding="utf-8")
    _git(source, "commit", "-am", "second")
    second = _git(source, "rev-parse", "HEAD")
    _git(source, "push", str(remote), "main")
    return checkout, remote, first, second


def test_latest_source_is_selected(linear_remote: tuple[Path, Path, str, str]) -> None:
    module = _load_module()
    checkout, remote, _, second = linear_remote

    decision = module.select_latest_inclusive_carrier(
        source_sha=second,
        remote_url=str(remote),
        cwd=checkout,
    )

    assert decision.selected is True
    assert decision.status == "selected_latest_main_carrier"
    assert decision.latest_sha == second


def test_older_source_is_coalesced_only_when_latest_includes_it(
    linear_remote: tuple[Path, Path, str, str],
) -> None:
    module = _load_module()
    checkout, remote, first, second = linear_remote

    decision = module.select_latest_inclusive_carrier(
        source_sha=first,
        remote_url=str(remote),
        cwd=checkout,
    )

    assert decision.selected is False
    assert decision.status == "superseded_by_latest_inclusive_carrier"
    assert decision.source_sha == first
    assert decision.latest_sha == second


@pytest.mark.parametrize(
    "marker_subject",
    [
        "chore(cd): deploy abcdef0 [skip ci]",
        "chore(cd): cancel-stale-cd",
        "chore(cd): refresh receipt [metadata-only]",
    ],
)
def test_excluded_main_tip_does_not_displace_deployable_carrier(
    linear_remote: tuple[Path, Path, str, str],
    tmp_path: Path,
    marker_subject: str,
) -> None:
    module = _load_module()
    checkout, remote, _, second = linear_remote
    marker_work = tmp_path / "marker-work"
    _git(tmp_path, "clone", str(remote), str(marker_work))
    _git(marker_work, "config", "user.name", "Carrier Test")
    _git(marker_work, "config", "user.email", "carrier@example.invalid")
    _git(marker_work, "commit", "--allow-empty", "-m", marker_subject)
    marker_sha = _git(marker_work, "rev-parse", "HEAD")
    _git(marker_work, "push", "origin", "main")

    decision = module.select_latest_inclusive_carrier(
        source_sha=second,
        remote_url=str(remote),
        cwd=checkout,
    )

    assert decision.selected is True
    assert decision.latest_sha == second
    assert decision.main_tip_sha == marker_sha


@pytest.mark.parametrize(
    "marker_body",
    [
        "release receipt [skip ci]",
        "queue receipt cancel-stale-cd",
        "runtime receipt [metadata-only]",
    ],
)
def test_excluded_marker_in_commit_body_does_not_displace_deployable_carrier(
    linear_remote: tuple[Path, Path, str, str],
    tmp_path: Path,
    marker_body: str,
) -> None:
    module = _load_module()
    checkout, remote, _, second = linear_remote
    marker_work = tmp_path / "marker-body-work"
    _git(tmp_path, "clone", str(remote), str(marker_work))
    _git(marker_work, "config", "user.name", "Carrier Test")
    _git(marker_work, "config", "user.email", "carrier@example.invalid")
    _git(
        marker_work,
        "commit",
        "--allow-empty",
        "-m",
        "receipt marker",
        "-m",
        marker_body,
    )
    marker_sha = _git(marker_work, "rev-parse", "HEAD")
    _git(marker_work, "push", "origin", "main")

    decision = module.select_latest_inclusive_carrier(
        source_sha=second,
        remote_url=str(remote),
        cwd=checkout,
    )

    assert decision.selected is True
    assert decision.latest_sha == second
    assert decision.main_tip_sha == marker_sha


def test_unrelated_latest_main_fails_closed(tmp_path: Path) -> None:
    module = _load_module()
    checkout = tmp_path / "checkout"
    unrelated = tmp_path / "unrelated"
    remote = tmp_path / "remote.git"
    for repo, value in ((checkout, "source"), (unrelated, "unrelated")):
        repo.mkdir()
        _git(repo, "init", "-b", "main")
        _git(repo, "config", "user.name", "Carrier Test")
        _git(repo, "config", "user.email", "carrier@example.invalid")
        (repo / "value.txt").write_text(f"{value}\n", encoding="utf-8")
        _git(repo, "add", "value.txt")
        _git(repo, "commit", "-m", value)
    source_sha = _git(checkout, "rev-parse", "HEAD")
    _git(tmp_path, "clone", "--bare", str(unrelated), str(remote))

    with pytest.raises(RuntimeError, match="latest_main_does_not_include_source"):
        module.select_latest_inclusive_carrier(
            source_sha=source_sha,
            remote_url=str(remote),
            cwd=checkout,
        )


def test_deep_linear_history_is_proven_after_bounded_deepen(tmp_path: Path) -> None:
    module = _load_module()
    source = tmp_path / "source"
    remote = tmp_path / "remote.git"
    checkout = tmp_path / "checkout"
    source.mkdir()
    _git(source, "init", "-b", "main")
    _git(source, "config", "user.name", "Carrier Test")
    _git(source, "config", "user.email", "carrier@example.invalid")
    _git(source, "commit", "--allow-empty", "-m", "first")
    first = _git(source, "rev-parse", "HEAD")
    _git(tmp_path, "clone", "--bare", str(source), str(remote))
    _git(tmp_path, "clone", "--depth=1", remote.as_uri(), str(checkout))
    for index in range(270):
        _git(source, "commit", "--allow-empty", "-m", f"linear-{index}")
    latest = _git(source, "rev-parse", "HEAD")
    _git(source, "push", str(remote), "main")

    decision = module.select_latest_inclusive_carrier(
        source_sha=first,
        remote_url=str(remote),
        cwd=checkout,
    )

    assert decision.selected is False
    assert decision.latest_sha == latest


def test_invalid_source_sha_fails_closed(tmp_path: Path) -> None:
    module = _load_module()

    with pytest.raises(RuntimeError, match="source_sha_invalid"):
        module.select_latest_inclusive_carrier(
            source_sha="not-a-sha",
            remote_url="unused",
            cwd=tmp_path,
        )