diff --git a/backend/infrahub/git/__init__.py b/backend/infrahub/git/__init__.py index baad34c96b..0faeb87f24 100644 --- a/backend/infrahub/git/__init__.py +++ b/backend/infrahub/git/__init__.py @@ -1,7 +1,5 @@ +from infrahub.git.directory import initialize_repositories_directory from infrahub.git.repository import ( - BRANCHES_DIRECTORY_NAME, - COMMITS_DIRECTORY_NAME, - TEMPORARY_DIRECTORY_NAME, ArtifactGenerateResult, CheckDefinitionInformation, GraphQLQueryInformation, @@ -9,15 +7,10 @@ InfrahubRepository, RepoFileInformation, TransformPythonInformation, - Worktree, extract_repo_file_information, - initialize_repositories_directory, ) __all__ = [ - "BRANCHES_DIRECTORY_NAME", - "COMMITS_DIRECTORY_NAME", - "TEMPORARY_DIRECTORY_NAME", "ArtifactGenerateResult", "InfrahubReadOnlyRepository", "InfrahubRepository", @@ -25,7 +18,6 @@ "CheckDefinitionInformation", "RepoFileInformation", "GraphQLQueryInformation", - "Worktree", "extract_repo_file_information", "initialize_repositories_directory", ] diff --git a/backend/infrahub/git/constants.py b/backend/infrahub/git/constants.py new file mode 100644 index 0000000000..a14317080e --- /dev/null +++ b/backend/infrahub/git/constants.py @@ -0,0 +1,3 @@ +COMMITS_DIRECTORY_NAME = "commits" +BRANCHES_DIRECTORY_NAME = "branches" +TEMPORARY_DIRECTORY_NAME = "temp" diff --git a/backend/infrahub/git/directory.py b/backend/infrahub/git/directory.py new file mode 100644 index 0000000000..090ea57270 --- /dev/null +++ b/backend/infrahub/git/directory.py @@ -0,0 +1,33 @@ +import os + +from infrahub import config +from infrahub.log import get_logger + +log = get_logger() + + +def get_repositories_directory() -> str: + """Return the absolute path to the main directory used for the repositories.""" + repos_dir = config.SETTINGS.git.repositories_directory + if not os.path.isabs(repos_dir): + current_dir = os.getcwd() + repos_dir = os.path.join(current_dir, config.SETTINGS.git.repositories_directory) + + return str(repos_dir) + + +def initialize_repositories_directory() -> bool: + """Check if the main repositories_directory already exist, if not create it. + + Return + True if the directory has been created, + False if the directory was already present. + """ + repos_dir = get_repositories_directory() + if not os.path.isdir(repos_dir): + os.makedirs(repos_dir) + log.debug(f"Initialized the repositories_directory at {repos_dir}") + return True + + log.debug(f"Repositories_directory already present at {repos_dir}") + return False diff --git a/backend/infrahub/git/repository.py b/backend/infrahub/git/repository.py index b76b7483af..860cff2ff4 100644 --- a/backend/infrahub/git/repository.py +++ b/backend/infrahub/git/repository.py @@ -11,12 +11,12 @@ from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union from uuid import UUID -import git import jinja2 import ujson import yaml from git import Repo from git.exc import BadName, GitCommandError, InvalidGitRepositoryError +from git.refs.remote import RemoteReference from infrahub_sdk import ( GraphQLError, InfrahubClient, @@ -44,13 +44,15 @@ from infrahub.exceptions import ( CheckError, CommitNotFoundError, - Error, FileOutOfRepositoryError, InitializationError, RepositoryError, RepositoryFileNotFoundError, TransformError, ) +from infrahub.git.constants import BRANCHES_DIRECTORY_NAME, COMMITS_DIRECTORY_NAME, TEMPORARY_DIRECTORY_NAME +from infrahub.git.directory import initialize_repositories_directory +from infrahub.git.worktree import Worktree from infrahub.log import get_logger from infrahub.services import InfrahubServices @@ -61,41 +63,10 @@ from infrahub_sdk.transforms import InfrahubTransform from infrahub.message_bus import messages -# pylint: disable=too-few-public-methods,too-many-lines +# pylint: disable=too-many-lines log = get_logger("infrahub.git") -COMMITS_DIRECTORY_NAME = "commits" -BRANCHES_DIRECTORY_NAME = "branches" -TEMPORARY_DIRECTORY_NAME = "temp" - - -def get_repositories_directory() -> str: - """Return the absolute path to the main directory used for the repositories.""" - repos_dir = config.SETTINGS.git.repositories_directory - if not os.path.isabs(repos_dir): - current_dir = os.getcwd() - repos_dir = os.path.join(current_dir, config.SETTINGS.git.repositories_directory) - - return str(repos_dir) - - -def initialize_repositories_directory() -> bool: - """Check if the main repositories_directory already exist, if not create it. - - Return - True if the directory has been created, - False if the directory was already present. - """ - repos_dir = get_repositories_directory() - if not os.path.isdir(repos_dir): - os.makedirs(repos_dir) - log.debug(f"Initialized the repositories_directory at {repos_dir}") - return True - - log.debug(f"Repositories_directory already present at {repos_dir}") - return False - class GraphQLQueryInformation(BaseModel): name: str @@ -238,49 +209,6 @@ def extract_repo_file_information( ) -class Worktree(BaseModel): - identifier: str - directory: str - commit: str - branch: Optional[str] = None - - @classmethod - def init(cls, text: str) -> Worktree: - lines = text.split("\n") - - full_directory = lines[0].replace("worktree ", "") - - # Extract the identifier from the directory name - # We first need to substract the main repository_directory to get the relative path. - repo_directory = get_repositories_directory() - relative_directory = full_directory.replace(repo_directory, "") - relative_paths = relative_directory.split("/") - - identifier = None - if len(relative_paths) == 3: - # this is the main worktree for the main branch - identifier = relative_paths[2] - - elif len(relative_paths) == 4 and relative_paths[2] == COMMITS_DIRECTORY_NAME: - # this is the either a commit or a branch worktree - identifier = relative_paths[3] - elif len(relative_paths) == 4 and relative_paths[2] == BRANCHES_DIRECTORY_NAME and lines[2] != "detached": - identifier = lines[2].replace("branch refs/heads/", "") - else: - raise Error("Unexpected path for a worktree.") - - item = cls( - identifier=identifier, - directory=full_directory, - commit=lines[1].replace("HEAD ", ""), - ) - - if lines[2] != "detached": - item.branch = lines[2].replace("branch refs/heads/", "") - - return item - - class BranchInGraph(BaseModel): id: str name: str @@ -588,7 +516,7 @@ def get_branches_from_remote(self) -> Dict[str, BranchInRemote]: branches = {} for remote_branch in git_repo.remotes.origin.refs: - if not isinstance(remote_branch, git.refs.remote.RemoteReference): # pylint: disable=no-member + if not isinstance(remote_branch, RemoteReference): continue short_name = remote_branch.name.replace("origin/", "") diff --git a/backend/infrahub/git/worktree.py b/backend/infrahub/git/worktree.py new file mode 100644 index 0000000000..3e726727c6 --- /dev/null +++ b/backend/infrahub/git/worktree.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +from typing import Optional + +from pydantic import BaseModel + +from infrahub.exceptions import Error +from infrahub.git.constants import BRANCHES_DIRECTORY_NAME, COMMITS_DIRECTORY_NAME +from infrahub.git.directory import get_repositories_directory + + +class Worktree(BaseModel): + identifier: str + directory: str + commit: str + branch: Optional[str] = None + + @classmethod + def init(cls, text: str) -> Worktree: + lines = text.split("\n") + + full_directory = lines[0].replace("worktree ", "") + + # Extract the identifier from the directory name + # We first need to substract the main repository_directory to get the relative path. + repo_directory = get_repositories_directory() + relative_directory = full_directory.replace(repo_directory, "") + relative_paths = relative_directory.split("/") + + identifier = None + if len(relative_paths) == 3: + # this is the main worktree for the main branch + identifier = relative_paths[2] + + elif len(relative_paths) == 4 and relative_paths[2] == COMMITS_DIRECTORY_NAME: + # this is the either a commit or a branch worktree + identifier = relative_paths[3] + elif len(relative_paths) == 4 and relative_paths[2] == BRANCHES_DIRECTORY_NAME and lines[2] != "detached": + identifier = lines[2].replace("branch refs/heads/", "") + else: + raise Error("Unexpected path for a worktree.") + + item = cls( + identifier=identifier, + directory=full_directory, + commit=lines[1].replace("HEAD ", ""), + ) + + if lines[2] != "detached": + item.branch = lines[2].replace("branch refs/heads/", "") + + return item diff --git a/backend/tests/unit/git/test_git_repository.py b/backend/tests/unit/git/test_git_repository.py index 300653ff37..222b17dcfc 100644 --- a/backend/tests/unit/git/test_git_repository.py +++ b/backend/tests/unit/git/test_git_repository.py @@ -15,17 +15,15 @@ TransformError, ) from infrahub.git import ( - BRANCHES_DIRECTORY_NAME, - COMMITS_DIRECTORY_NAME, - TEMPORARY_DIRECTORY_NAME, ArtifactGenerateResult, CheckDefinitionInformation, GraphQLQueryInformation, InfrahubRepository, RepoFileInformation, - Worktree, extract_repo_file_information, ) +from infrahub.git.constants import BRANCHES_DIRECTORY_NAME, COMMITS_DIRECTORY_NAME, TEMPORARY_DIRECTORY_NAME +from infrahub.git.worktree import Worktree from infrahub.utils import find_first_file_in_directory from tests.helpers.test_client import dummy_async_request