#!/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()