computer-playbook/tests/unit/python/cli/administration/roles/test_source.py

from __future__ import annotations

import re
import subprocess
import tempfile
import unittest
from pathlib import Path
from unittest import mock

from cli.administration.roles import source as source_module
from cli.administration.roles.declaration import Source
from cli.administration.roles.source import check, stage
from utils.cache.files import read_text
from utils.roles.custom import CustomRoleError
from utils.roles.mapping import ROLE_FILE_TASKS_MAIN

from ._fixtures import ROLE, ROLE_PATH, git, role_repo, roles_dir, write


class _SourceFixture:
    """Mixin with a core roles/ and a role repository.

    Combined with ``unittest.TestCase`` on each concrete class so the
    test-discovery lint sees the ``TestCase`` base directly.
    """

    def setUp(self) -> None:
        super().setUp()
        tmp = tempfile.TemporaryDirectory()
        self.addCleanup(tmp.cleanup)
        self.root = Path(tmp.name)
        self.roles = roles_dir(self.root)
        self.repo = role_repo(self.root)

    def check(self, source: Source, placements: tuple[Path, ...] = ()) -> None:
        check(ROLE, source, list(placements), self.roles)


class TestCheckVersion(_SourceFixture, unittest.TestCase):
    def test_tag_is_accepted(self):
        self.check(Source(str(self.repo), "v1.0.0", ROLE_PATH))

    def test_full_commit_sha_is_accepted(self):
        sha = git(self.repo, "rev-parse", "v1.0.0").strip()
        self.check(Source(str(self.repo), sha, ROLE_PATH))

    def test_full_commit_sha_needs_no_ref_lookup(self):
        self.check(Source("https://example.invalid/widgets.git", "a" * 40, ROLE_PATH))

    def test_branch_is_refused(self):
        with self.assertRaisesRegex(
            CustomRoleError, f"{ROLE}: source.version 'feature' names a branch"
        ):
            self.check(Source(str(self.repo), "feature", ROLE_PATH))

    def test_abbreviated_sha_is_refused(self):
        sha = git(self.repo, "rev-parse", "--short", "v1.0.0").strip()
        with self.assertRaisesRegex(
            CustomRoleError, "neither a tag nor a full commit SHA"
        ):
            self.check(Source(str(self.repo), sha, ROLE_PATH))

    def test_unknown_tag_is_refused(self):
        with self.assertRaisesRegex(CustomRoleError, "'v9.9.9' is neither a tag"):
            self.check(Source(str(self.repo), "v9.9.9", ROLE_PATH))


class TestCheckLocalSource(_SourceFixture, unittest.TestCase):
    def test_missing_directory_is_refused(self):
        with self.assertRaisesRegex(CustomRoleError, "is not a directory"):
            self.check(Source(str(self.root / "absent"), None, ROLE_PATH))

    def test_own_placement_directory_is_refused_naming_the_path(self):
        (self.roles / ROLE).mkdir()
        placement = re.escape(str((self.roles / ROLE).resolve()))
        with self.assertRaisesRegex(
            CustomRoleError, f"{placement} lies in the placement directory"
        ):
            self.check(Source(str(self.root), None, f"roles/{ROLE}"))

    def test_another_placement_is_refused(self):
        other = self.roles / "web-app-cst-other"
        other.mkdir()
        with self.assertRaisesRegex(CustomRoleError, "never deletes its own source"):
            self.check(
                Source(str(self.root), None, f"roles/{other.name}/{ROLE}"), (other,)
            )

    def test_placement_reached_through_a_symlink_is_refused(self):
        (self.roles / ROLE).symlink_to(self.repo / ROLE_PATH)
        with self.assertRaisesRegex(CustomRoleError, "never deletes its own source"):
            self.check(Source(str(self.repo), None, ROLE_PATH))

    def test_source_through_a_symlinked_checkout_is_refused(self):
        alias = self.root / "alias"
        alias.symlink_to(self.root)
        with self.assertRaisesRegex(CustomRoleError, "never deletes its own source"):
            self.check(Source(str(alias), None, f"roles/{ROLE}"))

    def test_checkout_outside_roles_is_accepted(self):
        self.check(Source(str(self.repo), None, ROLE_PATH))


class TestStage(_SourceFixture, unittest.TestCase):
    def test_tag_stages_only_the_role_path_at_that_tag(self):
        target = self.root / "staged"
        origin = stage(ROLE, Source(str(self.repo), "v1.0.0", ROLE_PATH), target)
        sha = git(self.repo, "rev-parse", "v1.0.0^{commit}").strip()
        self.assertEqual("release: v1\n", read_text(str(target / ROLE_FILE_TASKS_MAIN)))
        self.assertFalse((target / "README.md").exists())
        self.assertIn(f"(commit {sha})", origin)

    def test_full_commit_sha_is_staged(self):
        sha = git(self.repo, "rev-parse", "v1.0.0^{commit}").strip()
        target = self.root / "staged"
        stage(ROLE, Source(str(self.repo), sha, ROLE_PATH), target)
        self.assertEqual("release: v1\n", read_text(str(target / ROLE_FILE_TASKS_MAIN)))

    def test_local_directory_without_version_is_its_working_tree(self):
        write(self.repo / ROLE_PATH / ROLE_FILE_TASKS_MAIN, "release: uncommitted\n")
        (self.repo / ROLE_PATH / "__pycache__").mkdir()
        target = self.root / "staged"
        origin = stage(ROLE, Source(str(self.repo), None, ROLE_PATH), target)
        self.assertEqual(
            "release: uncommitted\n", read_text(str(target / ROLE_FILE_TASKS_MAIN))
        )
        self.assertFalse((target / "__pycache__").exists())
        self.assertIn("(working tree)", origin)

    def test_path_that_is_a_file_is_refused(self):
        with self.assertRaisesRegex(CustomRoleError, "source.path is not a directory"):
            stage(ROLE, Source(str(self.repo), "v1.0.0", "README.md"), self.root / "s")

    def test_missing_path_is_refused(self):
        with self.assertRaisesRegex(CustomRoleError, "source.path is not a directory"):
            stage(
                ROLE,
                Source(str(self.repo), "v1.0.0", f"absent/{ROLE}"),
                self.root / "s",
            )

    def test_failed_fetch_names_the_group(self):
        with self.assertRaisesRegex(
            CustomRoleError, f"{ROLE}: git fetch of v1.0.0 failed"
        ):
            stage(
                ROLE,
                Source(str(self.root / "absent"), "v1.0.0", ROLE_PATH),
                self.root / "staged",
            )


class TestGitNeverPrompts(unittest.TestCase):
    def test_git_runs_detached_without_a_prompt_or_askpass(self):
        failed = subprocess.CompletedProcess(
            ["git"], 128, stdout=b"", stderr=b"fatal: could not read Username"
        )
        source = Source("git@example.invalid:acme/widgets.git", "v1.0.0", ROLE_PATH)
        with (
            tempfile.TemporaryDirectory() as tmp,
            mock.patch.dict(
                "os.environ",
                {
                    "GIT_ASKPASS": "/bin/askpass",
                    "SSH_ASKPASS": "/bin/askpass",
                    "SSH_AUTH_SOCK": "/run/agent.sock",
                },
            ),
            mock.patch.object(
                source_module.subprocess, "run", return_value=failed
            ) as run,
            self.assertRaisesRegex(CustomRoleError, "could not read Username"),
        ):
            stage(ROLE, source, Path(tmp) / "staged")
        kwargs = run.call_args.kwargs
        self.assertEqual("0", kwargs["env"]["GIT_TERMINAL_PROMPT"])
        self.assertEqual("never", kwargs["env"]["SSH_ASKPASS_REQUIRE"])
        self.assertNotIn("GIT_ASKPASS", kwargs["env"])
        self.assertNotIn("SSH_ASKPASS", kwargs["env"])
        self.assertEqual("/run/agent.sock", kwargs["env"]["SSH_AUTH_SOCK"])
        self.assertTrue(kwargs["start_new_session"])
        self.assertEqual(subprocess.DEVNULL, kwargs["stdin"])


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