Genarrative/scripts/test_gitea_cache_builder.py

#!/usr/bin/env python3
"""Targeted regression tests for the trusted Gitea Rust-cache image builder."""

from __future__ import annotations

import os
from pathlib import Path
import shutil
import shlex
import stat
import subprocess
import tempfile
import textwrap
import unittest


REPOSITORY_ROOT = Path(__file__).resolve().parent.parent
BUILDER_NAME = "build-gitea-rust-cache.sh"
FIXED_SHA = "a" * 40
MASTER_SHA = "b" * 40


class GiteaRustCacheBuilderTest(unittest.TestCase):
    def setUp(self) -> None:
        self.temporary_directory = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
        self.root = Path(self.temporary_directory.name)
        self.repo = self.root / "repo"
        scripts = self.repo / "scripts"
        scripts.mkdir(parents=True)
        shutil.copy2(REPOSITORY_ROOT / "scripts" / BUILDER_NAME, scripts / BUILDER_NAME)
        (scripts / "gitea-ci-job-image.sh").write_text("#!/usr/bin/env bash\nexit 0\n")
        (scripts / "ci-rust-cache.sh").write_text("#!/usr/bin/env bash\nexit 0\n")
        self.trace = self.root / "trace.log"
        self.bin = self.root / "bin"
        self.bin.mkdir()
        self.write_fake_tools()
        self.execution_bin = self.wsl_path(self.bin)
        self.wsl_fake_root: str | None = None
        if os.name == "nt":
            self.wsl_fake_root = f"/tmp/gitea-cache-builder-{self.root.name}"
            subprocess.run(
                [
                    "bash",
                    "-c",
                    "rm -rf {root}; mkdir -p {root}/bin; cp {source}/* {root}/bin/; chmod +x {root}/bin/*".format(
                        root=shlex.quote(self.wsl_fake_root),
                        source=shlex.quote(self.wsl_path(self.bin)),
                    ),
                ],
                check=True,
            )
            self.execution_bin = f"{self.wsl_fake_root}/bin"

    def tearDown(self) -> None:
        if self.wsl_fake_root is not None:
            subprocess.run(
                ["bash", "-c", f"rm -rf {shlex.quote(self.wsl_fake_root)}"],
                check=False,
            )
        self.temporary_directory.cleanup()

    def write_tool(self, name: str, source: str) -> None:
        target = self.bin / name
        target.write_bytes(textwrap.dedent(source).encode("utf-8"))
        target.chmod(target.stat().st_mode | stat.S_IXUSR)

    def write_fake_tools(self) -> None:
        self.write_tool(
            "git",
            """#!/usr/bin/env bash
            set -eu
            printf 'git:%s\\n' "$*" >> "$TRACE"
            if [[ "$1" == '-C' ]]; then shift 2; fi
            case "$1" in
              fetch|cat-file) exit 0 ;;
              merge-base)
                [[ "${FAKE_NON_ANCESTOR:-}" != 1 ]]
                exit
                ;;
              rev-parse)
                if [[ "$2" == 'FETCH_HEAD^{commit}' ]]; then
                  printf '%s\\n' "$FAKE_MASTER_SHA"
                else
                  printf '%s\\n' "$FAKE_FIXED_SHA"
                fi
                ;;
              archive) printf 'archived %s\\n' "$2" ;;
              *) echo "unexpected git command: $*" >&2; exit 9 ;;
            esac
            """,
        )
        self.write_tool(
            "docker",
            """#!/usr/bin/env bash
            set -eu
            # Drain the archive pipeline before logging to avoid concurrent
            # appends to the shared trace on Windows/WSL filesystems.
            if [[ "$1" == cp && "$2" == '-' ]]; then
              cat >/dev/null
            fi
            printf 'docker:%s\\n' "$*" >> "$TRACE"
            if [[ "$1" == image && "$2" == inspect ]]; then
              printf 'sha256:%064d\\n' 0
            elif [[ "$1" == run && " $* " == *' --detach '* ]]; then
              printf 'fake-container\\n'
            elif [[ "$1" == exec && " $* " == *' bash -s '* ]]; then
              cat >/dev/null
            fi
            """,
        )
        self.write_tool(
            "curl",
            """#!/usr/bin/env bash
            set -eu
            while [[ "$#" -gt 0 ]]; do
              if [[ "$1" == --output ]]; then touch "$2"; exit 0; fi
              shift
            done
            exit 9
            """,
        )
        self.write_tool("sha256sum", "#!/usr/bin/env bash\ncat >/dev/null\n")
        self.write_tool(
            "tar",
            """#!/usr/bin/env bash
            set -eu
            while [[ "$#" -gt 0 ]]; do
              if [[ "$1" == --directory ]]; then
                mkdir -p "$2/sccache-v0.18.0-x86_64-unknown-linux-musl"
                printf '#!/usr/bin/env bash\\nexit 0\\n' > "$2/sccache-v0.18.0-x86_64-unknown-linux-musl/sccache"
                chmod +x "$2/sccache-v0.18.0-x86_64-unknown-linux-musl/sccache"
                exit 0
              fi
              shift
            done
            exit 9
            """,
        )

    @staticmethod
    def wsl_path(path: Path) -> str:
        value = path.resolve().as_posix()
        if len(value) >= 3 and value[1] == ":":
            return f"/mnt/{value[0].lower()}{value[2:]}"
        return value

    def run_builder(self, *arguments: str, non_ancestor: bool = False) -> subprocess.CompletedProcess[str]:
        script_path = self.wsl_path(self.repo / "scripts" / BUILDER_NAME)
        assignments = {
            "TRACE": self.wsl_path(self.trace),
            "FAKE_FIXED_SHA": FIXED_SHA,
            "FAKE_MASTER_SHA": MASTER_SHA,
            "FAKE_NON_ANCESTOR": "1" if non_ancestor else "",
        }
        exports = "; ".join(
            f"export {name}={shlex.quote(value)}" for name, value in assignments.items()
        )
        command = (
            f"unset CI; {exports}; export PATH={shlex.quote(self.execution_bin)}:\"$PATH\"; cd /; "
            f"exec bash {shlex.quote(script_path)} "
            + " ".join(shlex.quote(argument) for argument in arguments)
        )
        return subprocess.run(
            ["bash", "-c", command],
            env=os.environ,
            text=True,
            capture_output=True,
            check=False,
        )

    def trace_text(self) -> str:
        return self.trace.read_bytes().decode("utf-8", errors="replace")

    def test_archives_requested_master_ancestor_and_uses_its_helper(self) -> None:
        result = self.run_builder("trusted-base", "candidate", FIXED_SHA)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertIn(f"snapshot_source={FIXED_SHA}", result.stdout)
        trace = self.trace_text()
        self.assertIn(f"archive {FIXED_SHA}", trace)
        self.assertIn(
            "docker:exec fake-container cp "
            "/workspace/GenarrativeAI/Genarrative/scripts/ci-rust-cache.sh "
            "/tmp/ci-rust-cache.sh",
            trace,
        )

    def test_rejects_commit_outside_fetched_master(self) -> None:
        result = self.run_builder("trusted-base", "candidate", FIXED_SHA, non_ancestor=True)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("not contained in fetched master", result.stderr)
        self.assertNotIn(" archive ", self.trace_text())

    def test_rejects_non_full_sha_before_calling_external_tools(self) -> None:
        result = self.run_builder("trusted-base", "candidate", "abc123")
        self.assertEqual(result.returncode, 2)
        self.assertIn("complete 40-character SHA", result.stderr)
        self.assertFalse(self.trace.exists())


if __name__ == "__main__":
    unittest.main()