diff --git a/src/tmt_web/utils/git_handler.py b/src/tmt_web/utils/git_handler.py index 3de82e8..b67e174 100644 --- a/src/tmt_web/utils/git_handler.py +++ b/src/tmt_web/utils/git_handler.py @@ -6,6 +6,7 @@ It uses tmt's Git utilities for robust clone operations with retry logic. """ +import re from shutil import rmtree from tmt import Logger @@ -50,7 +51,7 @@ def clear_tmp_dir(logger: Logger) -> None: raise GeneralError(f"Failed to clear repository clone directory '{path}'") from err -def clone_repository(url: str, logger: Logger, ref: str | None = None) -> Path: +def clone_repository(url: str, logger: Logger) -> Path: """ Clone a Git repository to a unique path. @@ -71,15 +72,6 @@ def clone_repository(url: str, logger: Logger, ref: str | None = None) -> Path: # Clone with retry logic git_clone(url=url, destination=destination, logger=logger) - # If ref provided, checkout after clone - if ref: - common = Common(logger=logger) - try: - common.run(Command("git", "checkout", ref), cwd=destination) - except RunError as err: - logger.fail(f"Failed to checkout ref '{ref}'") - raise AttributeError(f"Failed to checkout ref '{ref}': {err}") from err - return destination @@ -92,17 +84,94 @@ def get_git_repository(url: str, logger: Logger, ref: str | None = None) -> Path :param ref: Optional ref to checkout :return: Path to the cloned repository :raises: GitUrlError if URL is invalid - :raises: GeneralError if clone fails + :raises: GeneralError if cloning, fetching, or updating a branch fails :raises: AttributeError if ref doesn't exist """ destination = get_unique_clone_path(url) if not destination.exists(): - clone_repository(url, logger, ref) - elif ref: - common = Common(logger=logger) + clone_repository(url, logger) + + common = Common(logger=logger) + + # Fetch remote refs + _fetch_remote(common, destination, logger) + + # If no ref is specified, the default branch is used + ref = ref or _get_default_branch(common, destination, logger) + + try: + common.run(Command("git", "checkout", ref), cwd=destination) + except RunError as err: + logger.fail(f"Failed to checkout ref '{ref}'") + raise AttributeError(f"Failed to checkout ref '{ref}'") from err + + # If the ref is a branch, ensure it's up to date + if _is_branch(common, destination, ref): + _update_branch(common, destination, ref, logger) + + return destination + + +def _get_default_branch(common: Common, repo_path: Path, logger: Logger) -> str: + """Determine the default branch of a Git repository using a remote HEAD.""" + try: + output = common.run( + Command("git", "symbolic-ref", "refs/remotes/origin/HEAD"), cwd=repo_path + ) + if output.stdout: + match = re.search(r"refs/remotes/origin/(.*)", output.stdout.strip()) + if match: + return match.group(1) + + logger.fail(f"Failed to determine default branch for repository '{repo_path}'") + raise GeneralError(f"Failed to determine default branch for repository '{repo_path}'") + + except RunError as err: + logger.fail(f"Failed to determine default branch for repository '{repo_path}'") + raise GeneralError( + f"Failed to determine default branch for repository '{repo_path}'" + ) from err + + +def _fetch_remote(common: Common, repo_path: Path, logger: Logger) -> None: + """Fetch updates from the remote repository.""" + try: + common.run(Command("git", "fetch"), cwd=repo_path) + except RunError as err: + logger.fail(f"Failed to fetch remote for repository '{repo_path}'") + raise GeneralError(f"Failed to fetch remote for repository '{repo_path}'") from err + + +def _update_branch(common: Common, repo_path: Path, branch: str, logger: Logger) -> None: + """Ensure the specified branch is up to date with its remote counterpart.""" + try: + common.run(Command("git", "show-branch", f"origin/{branch}"), cwd=repo_path) + except RunError as err: + logger.fail(f"Branch '{branch}' does not exist in repository '{repo_path}'") + raise GeneralError(f"Branch {branch}' does not exist in repository '{repo_path}'") from err + try: + # Check if the branch is already up to date + common.run(Command("git", "diff", "--quiet", branch, f"origin/{branch}"), cwd=repo_path) + return + except RunError: + # Branch is not up to date, proceed with update try: - common.run(Command("git", "checkout", ref), cwd=destination) + common.run(Command("git", "reset", "--hard", f"origin/{branch}"), cwd=repo_path) except RunError as err: - logger.fail(f"Failed to checkout ref '{ref}'") - raise AttributeError(f"Failed to checkout ref '{ref}': {err}") from err - return destination + logger.fail(f"Failed to update branch '{branch}' for repository '{repo_path}'") + raise GeneralError( + f"Failed to update branch '{branch}' for repository '{repo_path}'" + ) from err + + +def _is_branch(common: Common, repo_path: Path, ref: str) -> bool: + """ + Check if the given ref is a branch in the Git repository. + + :return: True if the ref is a branch, False otherwise. + """ + try: + common.run(Command("git", "show-ref", "-q", "--verify", f"refs/heads/{ref}"), cwd=repo_path) + return True + except RunError: + return False diff --git a/tests/unit/test_git_handler.py b/tests/unit/test_git_handler.py index e2b08fa..99ecebc 100644 --- a/tests/unit/test_git_handler.py +++ b/tests/unit/test_git_handler.py @@ -3,7 +3,7 @@ import pytest import tmt -from tmt.utils import Command, GeneralError, GitUrlError, RunError +from tmt.utils import GeneralError, GitUrlError, RunError from tmt_web import settings from tmt_web.utils import git_handler @@ -121,8 +121,12 @@ def test_get_git_repository_existing_checkout_error(self, mocker, logger): assert path.exists() # Mock checkout to fail - cmd = Command("git", "checkout", "invalid-branch") - mocker.patch("tmt.utils.Command.run", side_effect=RunError("Command failed", cmd, 1)) + def side_effect(cmd, *args, **kwargs): + if cmd._command == ["git", "checkout", "invalid-branch"]: + raise RunError("Command failed", cmd, 1) + return mocker.DEFAULT + + mocker.patch("tmt.utils.Command.run", side_effect=side_effect, autospec=True) # Try to get same repo with invalid ref with pytest.raises(AttributeError, match="Failed to checkout ref"):