From 3a3992badeb121b435d8f6aedc090dff215c4281 Mon Sep 17 00:00:00 2001 From: Dariusz Duda Date: Thu, 5 Dec 2024 16:51:26 -0500 Subject: [PATCH] feat: extend craft_application.git module (#576) Signed-off-by: Dariusz Duda Co-authored-by: Tiago Nobrega Co-authored-by: Michael DuBelko --- craft_application/git/__init__.py | 27 +- craft_application/git/_consts.py | 8 + craft_application/git/_git_repo.py | 247 ++++++++++++++- craft_application/git/_models.py | 14 + docs/reference/changelog.rst | 30 ++ tests/integration/conftest.py | 6 + tests/integration/git/__init__.py | 0 tests/integration/git/test_git.py | 97 ++++++ tests/unit/conftest.py | 28 +- tests/unit/git/test_git.py | 487 ++++++++++++++++++++++++++++- 10 files changed, 927 insertions(+), 17 deletions(-) create mode 100644 tests/integration/git/__init__.py create mode 100644 tests/integration/git/test_git.py diff --git a/craft_application/git/__init__.py b/craft_application/git/__init__.py index 6b37afd5..888406fb 100644 --- a/craft_application/git/__init__.py +++ b/craft_application/git/__init__.py @@ -14,18 +14,39 @@ """Git repository utilities.""" -from ._consts import COMMIT_SHORT_SHA_LEN +from ._consts import ( + NO_PUSH_URL, + COMMIT_SHA_LEN, + COMMIT_SHORT_SHA_LEN, + CRAFTGIT_BINARY_NAME, + GIT_FALLBACK_BINARY_NAME, +) from ._errors import GitError -from ._models import GitType, short_commit_sha -from ._git_repo import GitRepo, get_git_repo_type, is_repo, parse_describe +from ._models import GitType, Commit, short_commit_sha + +from ._git_repo import ( + GitRepo, + get_git_repo_type, + is_repo, + is_commit, + is_short_commit, + parse_describe, +) __all__ = [ "GitError", "GitRepo", "GitType", + "Commit", "get_git_repo_type", "is_repo", "parse_describe", + "is_commit", + "is_short_commit", "short_commit_sha", + "NO_PUSH_URL", + "COMMIT_SHA_LEN", "COMMIT_SHORT_SHA_LEN", + "CRAFTGIT_BINARY_NAME", + "GIT_FALLBACK_BINARY_NAME", ] diff --git a/craft_application/git/_consts.py b/craft_application/git/_consts.py index 1c45290c..d84a3e59 100644 --- a/craft_application/git/_consts.py +++ b/craft_application/git/_consts.py @@ -16,4 +16,12 @@ from typing import Final +NO_PUSH_URL: Final[str] = "no_push" + +COMMIT_SHA_LEN: Final[int] = 40 + COMMIT_SHORT_SHA_LEN: Final[int] = 7 + +CRAFTGIT_BINARY_NAME: Final[str] = "craft.git" + +GIT_FALLBACK_BINARY_NAME: Final[str] = "git" diff --git a/craft_application/git/_git_repo.py b/craft_application/git/_git_repo.py index c87dd870..0f0bf583 100644 --- a/craft_application/git/_git_repo.py +++ b/craft_application/git/_git_repo.py @@ -18,10 +18,14 @@ import logging import os +import re +import shutil import subprocess import time +from functools import lru_cache from pathlib import Path from shlex import quote +from typing import Final, cast from craft_parts.utils import os_utils from typing_extensions import Self @@ -45,11 +49,25 @@ else: del os.environ["SSL_CERT_DIR"] +from ._consts import CRAFTGIT_BINARY_NAME, GIT_FALLBACK_BINARY_NAME, NO_PUSH_URL from ._errors import GitError -from ._models import GitType +from ._models import Commit, GitType, short_commit_sha logger = logging.getLogger(__name__) +COMMIT_REGEX: Final[re.Pattern[str]] = re.compile("[0-9a-f]{40}") +SHORT_COMMIT_REGEX: Final[re.Pattern[str]] = re.compile("[0-9a-f]{7}") + + +def is_commit(ref: str) -> bool: + """Check if given commit is a valid git commit sha.""" + return bool(COMMIT_REGEX.fullmatch(ref)) + + +def is_short_commit(ref: str) -> bool: + """Check if given short commit is a valid git commit sha.""" + return bool(SHORT_COMMIT_REGEX.fullmatch(ref)) + def is_repo(path: Path) -> bool: """Check if a directory is a git repo. @@ -196,6 +214,68 @@ def commit(self, message: str = "auto commit") -> str: f"in {str(self.path)!r}." ) from error + def get_last_commit(self) -> Commit: + """Get the last Commit on the current head.""" + try: + last_commit = self._repo[self._repo.head.target] + except pygit2.GitError as error: + raise GitError("could not retrieve last commit") from error + else: + commit_message = cast( + str, + last_commit.message, # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType] + ) + return Commit( + sha=str(last_commit.id), + message=commit_message, + ) + + def get_last_commit_on_branch_or_tag( + self, + branch_or_tag: str, + *, + remote: str | None = None, + fetch: bool = False, + ) -> Commit: + """Find last commit corresponding to given branch or tag.""" + if fetch and remote is not None: + self.fetch(remote=remote, tags=True) + rev_list_output = [ + self.get_git_command(), + "rev-list", + "-n", + "1", + branch_or_tag, + ] + try: + rev_parse_output = subprocess.check_output( + rev_list_output, + text=True, + stderr=subprocess.STDOUT, + ) + except subprocess.CalledProcessError as error: + error_details = ( + f"cannot find ref: {branch_or_tag!r}.\nCommand output:\n{error.stdout}" + ) + raise GitError(error_details) from error + + commit_sha = rev_parse_output.strip() + try: + commit_obj = self._repo.get(commit_sha) + except (pygit2.GitError, ValueError) as error: + raise GitError( + f"cannot find commit: {short_commit_sha(commit_sha)!r}" + ) from error + else: + commit_message = cast( + str, + commit_obj.message, # pyright: ignore[reportOptionalMemberAccess,reportAttributeAccessIssue,reportUnknownMemberType] + ) + return Commit( + sha=commit_sha, + message=commit_message, + ) + def is_clean(self) -> bool: """Check if the repo is clean. @@ -286,6 +366,76 @@ def rename_remote( f"cannot rename '{remote_name}' to '{new_remote_name}'" ) from error + def get_remote_url(self, remote_name: str) -> str: + """Get URL associated with the given remote. + + Equivalent of git remote get-url + + + :param remote_name: the remote repository name + + :raises GitError: if remote does not exist + """ + if not self.remote_exists(remote_name=remote_name): + raise GitError(f"cannot get URL for non-existing remote '{remote_name}'") + return cast(str, self._repo.remotes[remote_name].url) + + def set_remote_url(self, remote_name: str, remote_url: str) -> None: + """Set new URL for the existing remote. + + Equivalent of git remote set-url + + + :param remote_name: the remote repository name + :param remote_url: URL to be associated with the given remote + + :raises GitError: if remote does not exist + """ + if not self.remote_exists(remote_name=remote_name): + raise GitError(f"cannot set URL for non-existing remote: {remote_name!r}") + self._repo.remotes.set_url(remote_name, remote_url) + + def get_remote_push_url(self, remote_name: str) -> str: + """Get push-only URL associated with the given remote. + + Equivalent of git remote get-url --push + + :param remote_name: the remote repository name + + :raises GitError: if remote does not exist + """ + if not self.remote_exists(remote_name=remote_name): + raise GitError( + f"cannot get push URL for non-existing remote: {remote_name!r}" + ) + return cast(str, self._repo.remotes[remote_name].push_url) + + def set_remote_push_url(self, remote_name: str, remote_push_url: str) -> None: + """Set new push-only URL for the existing remote. + + Equivalent of git remote set-url --push + + + :param remote_name: the remote repository name + :param remote_url: push URL to be associated with the given remote + + :raises GitError: if remote does not exist + """ + if not self.remote_exists(remote_name=remote_name): + raise GitError( + f"cannot set push URL for non-existing remote: {remote_name!r}" + ) + self._repo.remotes.set_push_url(remote_name, remote_push_url) + + def set_no_push(self, remote_name: str) -> None: + """Disable pushing to the selected remote. + + :param remote_name: the remote repository name + + :raises GitError: if remote does not exist + """ + self.set_remote_push_url(remote_name, NO_PUSH_URL) + def push_url( # noqa: PLR0912 (too-many-branches) self, remote_url: str, @@ -321,7 +471,14 @@ def push_url( # noqa: PLR0912 (too-many-branches) # Force push in case this repository already exists. The repository is always # going to exist solely for remote builds, so the only potential issue here is a # race condition with multiple remote builds on the same machine. - cmd: list[str] = ["git", "push", "--force", remote_url, refspec, "--progress"] + cmd: list[str] = [ + self.get_git_command(), + "push", + "--force", + remote_url, + refspec, + "--progress", + ] if push_tags: cmd.append("--tags") @@ -382,6 +539,72 @@ def push_url( # noqa: PLR0912 (too-many-branches) f"for the git repository in {str(self.path)!r}." ) + def fetch( + self, + *, + remote: str, + tags: bool = False, + ref: str | None = None, + depth: int | None = None, + ) -> None: + """Fetch the contents of the given remote. + + :param remote: The name of the remote. + :param tags: Whether to fetch tags. + :param ref: Optional reference to the specific object to fetch. + :param depth: Maximum number of commits to fetch (all by default). + """ + fetch_command = [self.get_git_command(), "fetch"] + + if not self.remote_exists(remote): + raise GitError(f"cannot fetch undefined remote: {remote!r}") + + if tags: + fetch_command.append("--tags") + if depth is not None: + fetch_command.extend(["--depth", f"{depth}"]) + + fetch_command.append(remote) + if ref is not None: + fetch_command.append(ref) + + try: + os_utils.process_run(fetch_command, logger.debug) + except FileNotFoundError as error: + raise GitError("git command not found in the system") from error + except subprocess.CalledProcessError as error: + raise GitError(f"cannot fetch remote: {remote!r}") from error + + def remote_contains( + self, + *, + remote: str, + commit_sha: str, + ) -> bool: + """Check if the given commit is pushed to the remote repository.""" + logger.debug( + "Checking if %r was pushed to %r", short_commit_sha(commit_sha), remote + ) + checking_command = [ + self.get_git_command(), + "branch", + "--remotes", + "--contains", + commit_sha, + ] + try: + remotes_that_has_given_commit = subprocess.check_output( + checking_command, + text=True, + ) + except subprocess.CalledProcessError as error: + raise GitError("incorrect commit provided, cannot check") from error + else: + for line in remotes_that_has_given_commit.splitlines(): + if line.strip().startswith(f"{remote}/"): + return True + return False + def describe( self, *, @@ -461,7 +684,7 @@ def clone_repository( raise GitError("Cannot clone to existing repository") logger.debug("Cloning %s to %s", url, path) - clone_cmd = ["git", "clone"] + clone_cmd = [cls.get_git_command(), "clone"] if checkout_branch is not None: logger.debug("Checking out to branch: %s", checkout_branch) clone_cmd.extend(["--branch", quote(checkout_branch)]) @@ -482,3 +705,21 @@ def clone_repository( f"cannot clone repository: {url} to {str(path)!r}" ) from error return cls(path) + + @classmethod + @lru_cache(maxsize=1) + def get_git_command(cls) -> str: + """Get name of the git executable that may be used in subprocesses. + + Fallback to the previous behavior in case of non-snap / local installation or + if snap does not provide expected binary. + """ + craftgit_binary = CRAFTGIT_BINARY_NAME + if shutil.which(craftgit_binary): + return craftgit_binary + logger.warning( + "Cannot find craftgit binary: %r. Is it a part of snap package?", + craftgit_binary, + ) + logger.warning("Falling back to: %r", GIT_FALLBACK_BINARY_NAME) + return GIT_FALLBACK_BINARY_NAME diff --git a/craft_application/git/_models.py b/craft_application/git/_models.py index 8af52271..be39a7b8 100644 --- a/craft_application/git/_models.py +++ b/craft_application/git/_models.py @@ -14,6 +14,7 @@ """Git repository models.""" +from dataclasses import dataclass from enum import Enum from ._consts import COMMIT_SHORT_SHA_LEN @@ -30,3 +31,16 @@ class GitType(Enum): INVALID = 0 NORMAL = 1 SHALLOW = 2 + + +@dataclass +class Commit: + """Model representing a commit.""" + + sha: str + message: str + + @property + def short_sha(self) -> str: + """Get short commit sha.""" + return short_commit_sha(self.sha) diff --git a/docs/reference/changelog.rst b/docs/reference/changelog.rst index 17021b5c..2227f2ba 100644 --- a/docs/reference/changelog.rst +++ b/docs/reference/changelog.rst @@ -4,6 +4,36 @@ Changelog ********* +4.6.0 (YYYY-MMM-DD) +------------------- + +Git +=== + +- Extend the ``craft_application.git`` module with the following APIs: + + - Add ``is_commit(ref)`` and ``is_short_commit(ref)`` helpers for checking if + a given ref is a valid commit hash. + - Add a ``Commit`` model to represent the result of ``get_last_commit()``. + +- Extend the ``GitRepo`` class with additional methods: + + - Add ``set_remote_url()`` and ``set_remote_push_url()`` methods and their + getter counterparts. + - Add ``set_no_push()`` method, which explicitly disables ``push`` for + specific remotes. + - Add ``get_last_commit()`` method, which retrieves the last commit hash and + message. + - Add ``get_last_commit_on_branch_or_tag()`` method, which retrieves the last + commit associated with a given ref. + - Add ``fetch()`` method, which retrieves remote objects. + +- Use ``craft.git`` for Git-related operations run with ``subprocess`` in + ``GitRepo``. + + +.. For a complete list of commits, check out the `4.6.0`_ release on GitHub. + 4.5.0 (2024-Nov-28) ------------------- diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index dfe7a9d7..5ea262c2 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -14,6 +14,7 @@ # You should have received a copy of the GNU Lesser General Public License # along with this program. If not, see . """Configuration for craft-application integration tests.""" + import os import pathlib import sys @@ -92,3 +93,8 @@ def pretend_jammy(mocker) -> None: """Pretend we're running on jammy. Used for tests that use destructive mode.""" fake_host = bases.BaseName(name="ubuntu", version="22.04") mocker.patch("craft_application.util.get_host_base", return_value=fake_host) + + +@pytest.fixture +def hello_repository_lp_url() -> str: + return "https://git.launchpad.net/ubuntu/+source/hello" diff --git a/tests/integration/git/__init__.py b/tests/integration/git/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/integration/git/test_git.py b/tests/integration/git/test_git.py new file mode 100644 index 00000000..937667a2 --- /dev/null +++ b/tests/integration/git/test_git.py @@ -0,0 +1,97 @@ +# This file is part of starcraft. +# +# Copyright 2024 Canonical Ltd. +# +# This program is free software: you can redistribute it and/or modify it +# under the terms of the GNU Lesser General Public License version 3, as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranties of MERCHANTABILITY, +# SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# General Public License for more details. +# +# You should have received a copy of the GNU General Public License along +# with this program. If not, see . +"""Git module integration tests.""" + +import pathlib + +import pytest +from craft_application.git import NO_PUSH_URL, Commit, GitError, GitRepo + + +def test_fetching_hello_repository( + empty_repository: pathlib.Path, + hello_repository_lp_url: str, +) -> None: + """Check if it is possible to fetch existing remote.""" + git_repo = GitRepo(empty_repository) + test_remote = "test-remote" + ref = "ubuntu/noble" + git_repo.add_remote(test_remote, hello_repository_lp_url) + git_repo.fetch(remote=test_remote, ref=ref, depth=1) + last_commit_on_fetched_ref = git_repo.get_last_commit_on_branch_or_tag( + remote=test_remote, branch_or_tag=f"{test_remote}/{ref}" + ) + assert isinstance( + last_commit_on_fetched_ref, Commit + ), "There should be a commit after fetching" + + +def test_fetching_remote_that_does_not_exist( + empty_repository: pathlib.Path, +) -> None: + """Check if it is possible to fetch existing remote.""" + git_repo = GitRepo(empty_repository) + test_remote = "test-remote" + ref = "ubuntu/noble" + git_repo.add_remote(test_remote, "git+ssh://non-existing-remote.localhost") + with pytest.raises(GitError) as git_error: + git_repo.fetch(remote=test_remote, ref=ref, depth=1) + assert git_error.value.details == f"cannot fetch remote: {test_remote!r}" + + +def test_set_url(empty_repository: pathlib.Path, hello_repository_lp_url: str) -> None: + """Check if remote URL can be set using API.""" + new_remote_url = "https://non-existing-remote-url.localhost" + git_repo = GitRepo(empty_repository) + test_remote = "test-remote" + git_repo.add_remote(test_remote, hello_repository_lp_url) + assert git_repo.get_remote_url(remote_name=test_remote) == hello_repository_lp_url + assert git_repo.get_remote_push_url(remote_name=test_remote) is None + + git_repo.set_remote_url(test_remote, new_remote_url) + assert git_repo.get_remote_url(remote_name=test_remote) == new_remote_url + assert git_repo.get_remote_push_url(remote_name=test_remote) is None + + +def test_set_push_url( + empty_repository: pathlib.Path, hello_repository_lp_url: str +) -> None: + """Check if remote push URL can be set using API.""" + new_remote_push_url = "https://non-existing-remote-push-url.localhost" + git_repo = GitRepo(empty_repository) + test_remote = "test-remote" + git_repo.add_remote(test_remote, hello_repository_lp_url) + assert git_repo.get_remote_url(remote_name=test_remote) == hello_repository_lp_url + assert git_repo.get_remote_push_url(remote_name=test_remote) is None + + git_repo.set_remote_push_url(test_remote, new_remote_push_url) + assert git_repo.get_remote_url(remote_name=test_remote) == hello_repository_lp_url + assert git_repo.get_remote_push_url(remote_name=test_remote) == new_remote_push_url + + +def test_set_no_push( + empty_repository: pathlib.Path, hello_repository_lp_url: str +) -> None: + """Check if remote push URL can be set using API.""" + git_repo = GitRepo(empty_repository) + test_remote = "test-remote" + git_repo.add_remote(test_remote, hello_repository_lp_url) + assert git_repo.get_remote_url(remote_name=test_remote) == hello_repository_lp_url + assert git_repo.get_remote_push_url(remote_name=test_remote) is None + + git_repo.set_no_push(test_remote) + assert git_repo.get_remote_url(remote_name=test_remote) == hello_repository_lp_url + assert git_repo.get_remote_push_url(remote_name=test_remote) == NO_PUSH_URL diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index 7b9b5c98..91bf03bc 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -14,12 +14,14 @@ # You should have received a copy of the GNU Lesser General Public License along # with this program. If not, see . """Configuration for craft-application unit tests.""" + from __future__ import annotations from unittest import mock import pytest -from craft_application import services, util +import pytest_mock +from craft_application import git, services, util @pytest.fixture(params=["amd64", "arm64", "riscv64"]) @@ -53,3 +55,27 @@ def mock_services(app_metadata, fake_project, fake_package_service_class): factory.fetch = mock.Mock(spec=services.FetchService) factory.init = mock.Mock(spec=services.InitService) return factory + + +@pytest.fixture +def clear_git_binary_name_cache() -> None: + from craft_application.git import GitRepo + + GitRepo.get_git_command.cache_clear() + + +@pytest.fixture( + params=[ + pytest.param(True, id="craftgit_available"), + pytest.param(False, id="fallback_to_git"), + ], +) +def expected_git_command( + request: pytest.FixtureRequest, + mocker: pytest_mock.MockerFixture, + clear_git_binary_name_cache: None, # noqa: ARG001 - Unused function argument +) -> str: + craftgit_exists = request.param + which_res = f"/some/path/to/{git.CRAFTGIT_BINARY_NAME}" if craftgit_exists else None + mocker.patch("shutil.which", return_value=which_res) + return git.CRAFTGIT_BINARY_NAME if craftgit_exists else git.GIT_FALLBACK_BINARY_NAME diff --git a/tests/unit/git/test_git.py b/tests/unit/git/test_git.py index cec290c7..89256d2f 100644 --- a/tests/unit/git/test_git.py +++ b/tests/unit/git/test_git.py @@ -24,12 +24,21 @@ import pygit2 import pygit2.enums import pytest +import pytest_mock +import pytest_subprocess from craft_application.git import ( + COMMIT_SHA_LEN, + COMMIT_SHORT_SHA_LEN, + CRAFTGIT_BINARY_NAME, + GIT_FALLBACK_BINARY_NAME, + NO_PUSH_URL, GitError, GitRepo, GitType, get_git_repo_type, + is_commit, is_repo, + is_short_commit, parse_describe, short_commit_sha, ) @@ -631,7 +640,7 @@ def test_clone_repository_wraps_called_process_error( patched_cloning_process.side_effect = subprocess.CalledProcessError( returncode=1, cmd="git clone" ) - fake_repo_url = "fake-repository-url.local" + fake_repo_url = "fake-repository-url.localhost" with pytest.raises(GitError) as raised: GitRepo.clone_repository(url=fake_repo_url, path=empty_working_directory) assert raised.value.details == ( @@ -645,7 +654,7 @@ def test_clone_repository_wraps_file_not_found_error( ): """Test if error is raised if git is not found.""" patched_cloning_process.side_effect = FileNotFoundError - fake_repo_url = "fake-repository-url.local" + fake_repo_url = "fake-repository-url.localhost" fake_branch = "some-fake-branch" with pytest.raises(GitError) as raised: GitRepo.clone_repository( @@ -692,7 +701,7 @@ def test_clone_repository_appends_correct_parameters_to_clone_command( # it is not a repo before clone is triggered, but will be after fake pygit2.clone_repository is called mocker.patch("craft_application.git._git_repo.is_repo", side_effect=[False, True]) mocked_init = mocker.patch.object(GitRepo, "_init_repo") - fake_repo_url = "fake-repository-url.local" + fake_repo_url = "fake-repository-url.localhost" from craft_application.git._git_repo import logger as git_repo_logger _ = GitRepo.clone_repository( @@ -719,7 +728,7 @@ def test_clone_repository_returns_git_repo_on_succcess_clone(mocker, empty_repos # it is not a repo before clone is triggered, but will be after fake pygit2.clone_repository is called mocker.patch("craft_application.git._git_repo.is_repo", side_effect=[False, True]) mocked_init = mocker.patch.object(GitRepo, "_init_repo") - fake_repo_url = "fake-repository-url.local" + fake_repo_url = "fake-repository-url.localhost" fake_branch = "some-fake-branch" repo = GitRepo.clone_repository( @@ -737,7 +746,7 @@ def test_clone_repository_raises_in_existing_repo(mocker, empty_working_director mocker.patch("craft_application.git._git_repo.is_repo", return_value=True) with pytest.raises(GitError) as exc: - GitRepo.clone_repository(url="some-url.local", path=empty_working_directory) + GitRepo.clone_repository(url="some-url.localhost", path=empty_working_directory) assert exc.value.details == "Cannot clone to existing repository" @@ -777,8 +786,10 @@ def test_check_git_repo_add_remote(mocker, empty_working_directory): repo = GitRepo(empty_working_directory) new_remote_name = "new-remote" mocked_fn = mocker.patch.object(repo._repo.remotes, "create") - repo.add_remote(new_remote_name, "https://git.fake-remote-url.local") - mocked_fn.assert_called_with(new_remote_name, "https://git.fake-remote-url.local") + repo.add_remote(new_remote_name, "https://git.fake-remote-url.localhost") + mocked_fn.assert_called_with( + new_remote_name, "https://git.fake-remote-url.localhost" + ) def test_check_git_repo_add_remote_value_error_is_wrapped( @@ -788,10 +799,10 @@ def test_check_git_repo_add_remote_value_error_is_wrapped( repo = GitRepo(empty_working_directory) new_remote_name = "new-remote" mocker.patch.object(repo._repo.remotes, "create", side_effect=[True, ValueError]) - repo.add_remote(new_remote_name, "https://git.fake-remote-url.local") + repo.add_remote(new_remote_name, "https://git.fake-remote-url.localhost") with pytest.raises(GitError) as ge: - repo.add_remote(new_remote_name, "https://git.fake-remote-url.local") + repo.add_remote(new_remote_name, "https://git.fake-remote-url.localhost") assert ge.value.details == f"remote '{new_remote_name}' already exists." @@ -804,7 +815,7 @@ def test_check_git_repo_add_remote_pygit_error_is_wrapped( mocker.patch.object(repo._repo.remotes, "create", side_effect=pygit2.GitError) with pytest.raises(GitError) as ge: - repo.add_remote(new_remote_name, "https://git.fake-remote-url.local") + repo.add_remote(new_remote_name, "https://git.fake-remote-url.localhost") expected_err_msg = ( "could not add remote to a git " f"repository in {str(empty_working_directory)!r}." @@ -865,6 +876,154 @@ def test_check_git_repo_rename_remote_pygit_error_is_wrapped( assert ge.value.details == expected_err_msg +class _FakeRemote: + def __init__( + self, + fake_remote_url: str, + fake_remote_push_url: str | None = None, + ) -> None: + self._fake_remote_url = fake_remote_url + self._fake_remote_push_url = fake_remote_push_url + + @property + def url(self) -> str: + return self._fake_remote_url + + @property + def push_url(self) -> str | None: + return self._fake_remote_push_url + + def set_url(self, new_url: str) -> None: + self._fake_remote_url = new_url + + def set_push_url(self, new_push_url: str) -> None: + self._fake_remote_push_url = new_push_url + + +def test_git_repo_get_remote_url(mocker, empty_working_directory): + repo = GitRepo(empty_working_directory) + test_fake_existing_remote = "test-remote" + fake_remote_url = "https://test-remote-url.localhost" + + mocked_remotes = mocker.patch.object(repo._repo, "remotes") + mocked_remotes.__getitem__ = lambda _s, _i: _FakeRemote(fake_remote_url) + + assert repo.get_remote_url(test_fake_existing_remote) == fake_remote_url + + +def test_git_repo_get_remote_push_url_default(mocker, empty_working_directory): + repo = GitRepo(empty_working_directory) + test_fake_existing_remote = "test-remote" + fake_remote_url = "https://test-remote-url.localhost" + + mocked_remotes = mocker.patch.object(repo._repo, "remotes") + mocked_remotes.__getitem__ = lambda _s, _i: _FakeRemote(fake_remote_url) + + assert repo.get_remote_push_url(test_fake_existing_remote) is None + + +def test_check_git_repo_get_remote_push_url_if_set(mocker, empty_working_directory): + repo = GitRepo(empty_working_directory) + test_fake_existing_remote = "test-remote" + fake_remote_url = "https://test-remote-url.localhost" + + mocked_remotes = mocker.patch.object(repo._repo, "remotes") + fake_remote = _FakeRemote(fake_remote_url) + custom_push_url = "https://custom-push-url.localhost" + fake_remote.set_push_url(custom_push_url) + + mocked_remotes.__getitem__ = lambda _s, _i: fake_remote + + assert repo.get_remote_push_url(test_fake_existing_remote) == custom_push_url + + +def test_git_repo_get_remote_push_url_fails_if_remote_does_not_exist( + mocker, empty_working_directory +): + """Check if GitError is raised if remote does not exist.""" + repo = GitRepo(empty_working_directory) + test_fake_existing_remote = "test-remote" + mocked_remotes = mocker.patch.object(repo._repo, "remotes") + mocked_remotes.__getitem__.side_effect = KeyError + + with pytest.raises(GitError): + repo.get_remote_push_url(test_fake_existing_remote) + + +def test_git_repo_get_remote_url_fails_if_remote_does_not_exist( + mocker, empty_working_directory +): + """Check if GitError is raised if remote does not exist.""" + repo = GitRepo(empty_working_directory) + test_fake_existing_remote = "test-remote" + mocked_remotes = mocker.patch.object(repo._repo, "remotes") + mocked_remotes.__getitem__.side_effect = KeyError + + with pytest.raises(GitError): + repo.get_remote_url(test_fake_existing_remote) + + +def test_git_repo_set_remote_url(mocker, empty_working_directory): + repo = GitRepo(empty_working_directory) + test_remote = "test-remote" + updated_remote_url = "https://new-remote-url.localhost" + + mocked_remotes = mocker.patch.object(repo._repo, "remotes") + repo.set_remote_url(test_remote, updated_remote_url) + + mocked_remotes.set_url.assert_called_with(test_remote, updated_remote_url) + + +def test_git_repo_set_remote_url_non_existing_remote(empty_repository: Path): + repo = GitRepo(empty_repository) + non_existing_remote = "non-existing-remote" + updated_remote_url = "https://new-remote-url.localhost" + + with pytest.raises(GitError) as git_error: + repo.set_remote_url(non_existing_remote, updated_remote_url) + + assert ( + git_error.value.details + == f"cannot set URL for non-existing remote: {non_existing_remote!r}" + ) + + +def test_git_repo_set_remote_push_url(mocker, empty_working_directory): + repo = GitRepo(empty_working_directory) + test_remote = "test-remote" + updated_remote_url = "https://new-remote-push-url.localhost" + + mocked_remotes = mocker.patch.object(repo._repo, "remotes") + repo.set_remote_push_url(test_remote, updated_remote_url) + + mocked_remotes.set_push_url.assert_called_with(test_remote, updated_remote_url) + + +def test_git_repo_set_remote_push_url_non_existing_remote(empty_repository: Path): + repo = GitRepo(empty_repository) + non_existing_remote = "non-existing-remote" + updated_remote_url = "https://new-remote-push-url.localhost" + + with pytest.raises(GitError) as git_error: + repo.set_remote_push_url(non_existing_remote, updated_remote_url) + + assert ( + git_error.value.details + == f"cannot set push URL for non-existing remote: {non_existing_remote!r}" + ) + + +def test_check_git_repo_set_no_push(mocker, empty_working_directory): + """Check if url is returned correctly using the get_url API.""" + repo = GitRepo(empty_working_directory) + test_remote = "test-remote" + + mocked_remotes = mocker.patch.object(repo._repo, "remotes") + repo.set_no_push(test_remote) + + mocked_remotes.set_push_url.assert_called_with(test_remote, NO_PUSH_URL) + + def test_check_git_repo_for_remote_build_shallow(empty_working_directory): """Check if directory is a shallow cloned repo.""" root_path = Path(empty_working_directory) @@ -905,6 +1064,182 @@ def test_check_git_repo_for_remote_build_shallow(empty_working_directory): check_git_repo_for_remote_build(git_shallow_path) +def test_retriving_last_commit(empty_repository: Path) -> None: + git_repo = GitRepo(empty_repository) + (git_repo.path / "1").write_text("1") + git_repo.add_all() + test_commit_message = "test: add commit message" + git_repo.commit(test_commit_message) + commit = git_repo.get_last_commit() + assert commit is not None, "Commit should be created and retrieved" + assert len(commit.sha) == COMMIT_SHA_LEN, "Commit hash should have full length" + assert is_commit(commit.sha), "Returned value should be a valid commit" + assert ( + commit.sha[:COMMIT_SHORT_SHA_LEN] == commit.short_sha + ), "Commit should have proper short version" + + +def test_last_commit_on_empty_repository(empty_repository: Path) -> None: + """Test if last_commit errors out with correct error.""" + git_repo = GitRepo(empty_repository) + with pytest.raises(GitError) as git_error: + git_repo.get_last_commit() + assert git_error.value.details == "could not retrieve last commit" + + +@pytest.mark.parametrize( + "tags", [True, False], ids=lambda x: "tags" if x else "no_tags" +) +@pytest.mark.parametrize( + "depth", [None, 1], ids=lambda x: "with_{x}_depth" if x else "no_depth" +) +@pytest.mark.parametrize( + "ref", [None, "aaaaaaa"], ids=lambda x: f"with_ref_{x}" if x else "no_ref_specified" +) +def test_fetching_remote( + empty_repository: Path, + fake_process: pytest_subprocess.FakeProcess, + expected_git_command: str, + depth: int | None, + ref: str | None, + *, + tags: bool, +) -> None: + git_repo = GitRepo(empty_repository) + remote = "test-remote" + git_repo.add_remote(remote, "https://non-existing-repo.localhost") + cmd = [expected_git_command, "fetch"] + if tags: + cmd.append("--tags") + if depth is not None: + cmd.extend(["--depth", str(depth)]) + cmd.append(remote) + if ref is not None: + cmd.append(ref) + fake_process.register(cmd) + git_repo.fetch(remote=remote, tags=tags, depth=depth, ref=ref) + + +def test_fetching_remote_fails( + empty_repository: Path, + fake_process: pytest_subprocess.FakeProcess, + expected_git_command: str, +) -> None: + git_repo = GitRepo(empty_repository) + remote = "test-remote" + git_repo.add_remote(remote, "https://non-existing-repo.localhost") + cmd = [expected_git_command, "fetch"] + cmd.append(remote) + fake_process.register(cmd, returncode=1) + with pytest.raises(GitError) as git_error: + git_repo.fetch(remote=remote) + assert git_error.value.details == f"cannot fetch remote: {remote!r}" + + +def test_fetching_fails_if_git_not_available( + empty_repository: Path, + mocker: pytest_mock.MockerFixture, +) -> None: + git_repo = GitRepo(empty_repository) + remote = "test-remote" + git_repo.add_remote(remote, "https://non-existing-repo.localhost") + patched_process_run = mocker.patch("craft_parts.utils.os_utils.process_run") + patched_process_run.side_effect = FileNotFoundError + with pytest.raises(GitError) as git_error: + git_repo.fetch(remote=remote) + assert git_error.value.details == "git command not found in the system" + + +def test_fetching_undefined_remote(empty_repository: Path) -> None: + git_repo = GitRepo(empty_repository) + remote = "test-non-existing-remote" + with pytest.raises(GitError) as git_error: + git_repo.fetch(remote=remote) + assert git_error.value.details == f"cannot fetch undefined remote: {remote!r}" + + +@pytest.mark.parametrize( + ("commit_str", "is_valid"), + [ + ("test", False), + ("fake-commit", False), + ("1234", False), + ("xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx", False), + ("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", True), + ("9d51ca832224e9a31e1898dd11b2764374402f09", True), + ("9d51ca832224e9a31e1898dd11b2764374402f09a", False), + ("9d51ca8", False), + ("aaaaaaa", False), + ], +) +def test_is_commit(commit_str: str, *, is_valid: bool) -> None: + """Check function that checks if something is a valid sha.""" + assert is_commit(commit_str) is is_valid + + +@pytest.mark.parametrize( + ("commit_str", "is_valid"), + [ + ("test", False), + ("fake-commit", False), + ("1234", False), + ("xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx", False), + ("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", False), + ("9d51ca832224e9a31e1898dd11b2764374402f09", False), + ("9d51ca832224e9a31e1898dd11b2764374402f09a", False), + ("9d51ca8", True), + ("aaaaaaa", True), + ], +) +def test_is_short_commit(commit_str: str, *, is_valid: bool) -> None: + """Check function that checks if something is a valid sha.""" + assert is_short_commit(commit_str) is is_valid + + +@pytest.mark.parametrize( + ("remote", "command_output", "response"), + [ + ("some-remote", "some-remote-postfix/branch", False), + ("some-remote", "some-remote/main", True), + ("some-remote", "other-remote/main", False), + ("some-remote", "", False), + ], +) +def test_remote_contains( + empty_repository: Path, + fake_process: pytest_subprocess.FakeProcess, + remote: str, + command_output: str, + expected_git_command: str, + *, + response: bool, +) -> None: + fake_process.register( + [expected_git_command, "branch", "--remotes", "--contains", "fake-commit-sha"], + stdout=command_output, + ) + git_repo = GitRepo(empty_repository) + assert ( + git_repo.remote_contains(remote=remote, commit_sha="fake-commit-sha") + is response + ) + + +def test_remote_contains_fails_if_subprocess_fails( + empty_repository: Path, + fake_process: pytest_subprocess.FakeProcess, + expected_git_command: str, +) -> None: + fake_process.register( + [expected_git_command, "branch", "--remotes", "--contains", "fake-commit-sha"], + returncode=1, + ) + git_repo = GitRepo(empty_repository) + with pytest.raises(GitError) as git_error: + git_repo.remote_contains(remote="fake-remote", commit_sha="fake-commit-sha") + assert git_error.value.details == "incorrect commit provided, cannot check" + + @pytest.mark.parametrize( "always_use_long_format", [True, False, None], @@ -985,3 +1320,135 @@ def test_describing_fallback_to_commit_for_unannotated_tags( repo = GitRepo(repository_with_unannotated_tag.repository_path) describe_result = repo.describe(show_commit_oid_as_fallback=True) assert describe_result == repository_with_unannotated_tag.short_commit + + +@pytest.mark.usefixtures("clear_git_binary_name_cache") +@pytest.mark.parametrize( + ("craftgit_exists"), + [ + pytest.param(True, id="craftgit_available"), + pytest.param(False, id="fallback_to_git"), + ], +) +def test_craftgit_is_used_for_git_operations( + mocker: pytest_mock.MockerFixture, + *, + craftgit_exists: bool, +) -> None: + which_res = f"/some/path/to/{CRAFTGIT_BINARY_NAME}" if craftgit_exists else None + which_mock = mocker.patch("shutil.which", return_value=which_res) + + expected_binary = ( + CRAFTGIT_BINARY_NAME if craftgit_exists else GIT_FALLBACK_BINARY_NAME + ) + assert GitRepo.get_git_command() == expected_binary + + which_mock.assert_called_once_with(CRAFTGIT_BINARY_NAME) + + +@pytest.mark.usefixtures("clear_git_binary_name_cache") +@pytest.mark.parametrize( + ("craftgit_exists"), + [ + pytest.param(True, id="craftgit_available"), + pytest.param(False, id="fallback_to_git"), + ], +) +def test_get_git_command_result_is_cached( + mocker: pytest_mock.MockerFixture, + *, + craftgit_exists: bool, +) -> None: + which_res = f"/some/path/to/{CRAFTGIT_BINARY_NAME}" if craftgit_exists else None + which_mock = mocker.patch("shutil.which", return_value=which_res) + + expected_binary = ( + CRAFTGIT_BINARY_NAME if craftgit_exists else GIT_FALLBACK_BINARY_NAME + ) + for _ in range(3): + assert GitRepo.get_git_command() == expected_binary + + which_mock.assert_called_once_with(CRAFTGIT_BINARY_NAME) + + +def test_last_commit_on_branch_or_tag_fails_if_commit_not_found( + empty_repository: Path, + fake_process: pytest_subprocess.FakeProcess, + expected_git_command: str, +) -> None: + git_repo = GitRepo(empty_repository) + branch_or_tag = "test" + commit = "non-existent-commit" + fake_process.register( + [expected_git_command, "rev-list", "-n", "1", branch_or_tag], + stdout=commit, + ) + with pytest.raises(GitError) as git_error: + git_repo.get_last_commit_on_branch_or_tag(branch_or_tag) + assert ( + git_error.value.details == f"cannot find commit: {short_commit_sha(commit)!r}" + ) + + +def test_last_commit_on_branch_or_tag_fails_if_ref_not_found( + empty_repository: Path, + fake_process: pytest_subprocess.FakeProcess, + expected_git_command: str, +) -> None: + git_repo = GitRepo(empty_repository) + branch_or_tag = "test" + fake_process.register( + [expected_git_command, "rev-list", "-n", "1", branch_or_tag], + returncode=1, + stderr=f"fatal: ambiguous argument {branch_or_tag!r}", + ) + with pytest.raises(GitError) as git_error: + git_repo.get_last_commit_on_branch_or_tag(branch_or_tag) + err_details = cast(str, git_error.value.details) + assert err_details.startswith(f"cannot find ref: {branch_or_tag!r}") + assert "fatal" in err_details + + +def test_get_last_commit_on_branch_or_tag( + repository_with_commit: RepositoryDefinition, + fake_process: pytest_subprocess.FakeProcess, + expected_git_command: str, +) -> None: + git_repo = GitRepo(repository_with_commit.repository_path) + branch_or_tag = "test" + last_commit = git_repo.get_last_commit() + fake_process.register( + [expected_git_command, "rev-list", "-n", "1", branch_or_tag], + stdout=last_commit.sha, + ) + last_commit_on_branch = git_repo.get_last_commit_on_branch_or_tag(branch_or_tag) + assert last_commit_on_branch == last_commit + + +@pytest.mark.parametrize("fetch", [True, False]) +def test_last_commit_on_branch_with_fetching_remote( + repository_with_commit: RepositoryDefinition, + fake_process: pytest_subprocess.FakeProcess, + expected_git_command: str, + mocker: pytest_mock.MockerFixture, + *, + fetch: bool, +) -> None: + git_repo = GitRepo(repository_with_commit.repository_path) + fetch_mock = mocker.patch.object(git_repo, "fetch") + test_remote = "remote-repo" + + branch_or_tag = "test" + last_commit = git_repo.get_last_commit() + fake_process.register( + [expected_git_command, "rev-list", "-n", "1", branch_or_tag], + stdout=last_commit.sha, + ) + + git_repo.get_last_commit_on_branch_or_tag( + branch_or_tag, remote=test_remote, fetch=fetch + ) + if fetch: + fetch_mock.assert_called_with(remote=test_remote, tags=True) + else: + fetch_mock.assert_not_called()