who_need_help/test/scripts/test_release_artifact_root_test.py

import gzip
import hashlib
import os
import shutil
import subprocess
import tempfile
import textwrap
import unittest
from pathlib import Path


ROOT = Path(__file__).resolve().parents[2]


class TestReleaseArtifactRootTest(unittest.TestCase):
    def setUp(self):
        self.tempdir = tempfile.TemporaryDirectory()
        self.base = Path(self.tempdir.name)
        self.project = self.base / "project"
        self.scripts = self.project / "scripts"
        self.scripts.mkdir(parents=True)
        script = self.scripts / "prepare-test-images.sh"
        shutil.copy2(ROOT / "scripts" / script.name, script)
        script.chmod(0o755)

        self.run_command(["git", "init", "--quiet"], cwd=self.project)
        self.run_command(
            ["git", "config", "user.email", "test-release@example.invalid"],
            cwd=self.project,
        )
        self.run_command(
            ["git", "config", "user.name", "Test release"], cwd=self.project
        )
        self.run_command(["git", "add", "scripts"], cwd=self.project)
        self.run_command(
            ["git", "commit", "--quiet", "-m", "test fixture"], cwd=self.project
        )
        self.commit = self.run_command(
            ["git", "rev-parse", "HEAD"], cwd=self.project
        ).stdout.strip()
        self.artifact_root = self.base / "test-artifacts"

        self.fake_bin = self.base / "bin"
        self.fake_bin.mkdir()
        for name in ("docker", "jq"):
            command = self.fake_bin / name
            command.write_text("#!/bin/sh\nexit 97\n", encoding="utf-8")
            command.chmod(0o755)

        self.test_env = self.base / "test.env"
        self.test_env.write_text(
            textwrap.dedent(
                """\
                DEPLOYMENT_ENV=test
                DATABASE_MODE=container
                APP_TOPOLOGY=compact
                APP_IMAGE=replace-me
                SOCKET_PROXY_IMAGE=replace-me
                POSTGIS_IMAGE=replace-me
                """
            ),
            encoding="utf-8",
        )
        self.test_env.chmod(0o600)

    def tearDown(self):
        self.tempdir.cleanup()

    def run_command(self, command, *, cwd=None, env=None, check=True):
        return subprocess.run(
            command,
            cwd=cwd,
            env=env,
            capture_output=True,
            text=True,
            check=check,
        )

    def environment(self):
        env = os.environ.copy()
        env["WNH_TEST_RELEASE_ARTIFACT_ROOT"] = str(self.artifact_root)
        env["PATH"] = f"{self.fake_bin}:{env['PATH']}"
        return env

    def write_existing_artifact(self):
        release_dir = self.artifact_root / self.commit
        release_dir.mkdir(parents=True)
        archive = (
            release_dir
            / f"who_need_help-{self.commit}-test-images-linux-amd64.tar.gz"
        )
        with archive.open("wb") as output:
            with gzip.GzipFile(fileobj=output, mode="wb", mtime=0) as compressed:
                compressed.write(b"verified test image archive fixture")
        archive.chmod(0o600)
        digest = hashlib.sha256(archive.read_bytes()).hexdigest()
        checksum = Path(f"{archive}.sha256")
        checksum.write_text(f"{digest}  {archive.name}\n", encoding="utf-8")
        checksum.chmod(0o600)

        short_commit = self.commit[:12]
        manifest = release_dir / f"who_need_help-{self.commit}-test-images.manifest"
        manifest.write_text(
            textwrap.dedent(
                f"""\
                format=1
                deployment=test
                commit={self.commit}
                platform=linux/amd64
                topology=compact
                image_count=2
                image=who-need-help:test-{short_commit}|sha256:{'a' * 64}
                image=who-need-help:postgis-test-{short_commit}|sha256:{'b' * 64}
                """
            ),
            encoding="utf-8",
        )
        manifest.chmod(0o600)
        return archive, manifest

    def test_existing_test_archive_is_verified_without_building(self):
        archive, manifest = self.write_existing_artifact()

        result = self.run_command(
            [str(self.scripts / "prepare-test-images.sh"), str(self.test_env)],
            cwd=self.project,
            env=self.environment(),
        )

        self.assertIn("verifying all metadata", result.stdout)
        self.assertIn(str(archive), result.stdout)

        manifest.write_text(
            manifest.read_text(encoding="utf-8").replace(
                "deployment=test", "deployment=production"
            ),
            encoding="utf-8",
        )
        rejected = self.run_command(
            [str(self.scripts / "prepare-test-images.sh"), str(self.test_env)],
            cwd=self.project,
            env=self.environment(),
            check=False,
        )
        self.assertEqual(rejected.returncode, 2)
        self.assertIn("wrong deployment identity", rejected.stderr)

    def test_relative_artifact_root_is_rejected(self):
        env = self.environment()
        env["WNH_TEST_RELEASE_ARTIFACT_ROOT"] = "relative/test-releases"
        result = self.run_command(
            [str(self.scripts / "prepare-test-images.sh"), str(self.test_env)],
            cwd=self.project,
            env=env,
            check=False,
        )

        self.assertEqual(result.returncode, 2)
        self.assertIn("must be an absolute path", result.stderr)

    def test_untracked_build_input_is_rejected(self):
        (self.project / "untracked-build-input.txt").write_text(
            "must not enter an immutable release\n", encoding="utf-8"
        )

        result = self.run_command(
            [str(self.scripts / "prepare-test-images.sh"), str(self.test_env)],
            cwd=self.project,
            env=self.environment(),
            check=False,
        )

        self.assertEqual(result.returncode, 2)
        self.assertIn("dirty checkout", result.stderr)


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