Resolves #130: Replace GitPython with dulwich (#218)

* resolves #130: Replace GitPython with dulwich

* Remove debugging

* Add better testing for initializing git repos

* Refactor RepoManager

* Add integration test for dulwich

* Add minimum version for dulwich

* Fix linting for exception match

* Add debug for ci

* Move asserts so it fails early

* Add debugging for integration test

* Return docker compose config in CI

* Add more debugs for files

* Run command on docker host

* Get container logs

* Comment out integration test for git repo

* Remove print

Author: minitriga

changelog/130.added.md

@@ -0,0 +1 @@
+Replace GitPython with dulwich

infrahub_sdk/checks.py

@@ -8,9 +8,10 @@
 from typing import TYPE_CHECKING, Any
 
 import ujson
-from git.repo import Repo
 from pydantic import BaseModel, Field
 
+from infrahub_sdk.repository import GitRepoManager
+
 from .exceptions import UninitializedError
 
 if TYPE_CHECKING:
@@ -43,7 +44,7 @@ def __init__(
         params: dict | None = None,
         client: InfrahubClient | None = None,
     ):
-        self.git: Repo | None = None
+        self.git: GitRepoManager | None = None
         self.initializer = initializer or InfrahubCheckInitializer()
 
         self.logs: list[dict[str, Any]] = []
@@ -137,10 +138,9 @@ def branch_name(self) -> str:
             return self.branch
 
         if not self.git:
-            self.git = Repo(self.root_directory)
+            self.git = GitRepoManager(self.root_directory)
 
         self.branch = str(self.git.active_branch)
-
         return self.branch
 
     @abstractmethod

infrahub_sdk/generator.py

@@ -4,7 +4,7 @@
 from abc import abstractmethod
 from typing import TYPE_CHECKING
 
-from git.repo import Repo
+from infrahub_sdk.repository import GitRepoManager
 
 from .exceptions import UninitializedError
 
@@ -30,7 +30,7 @@ def __init__(
     ) -> None:
         self.query = query
         self.branch = branch
-        self.git: Repo | None = None
+        self.git: GitRepoManager | None = None
         self.params = params or {}
         self.root_directory = root_directory or os.getcwd()
         self.generator_instance = generator_instance
@@ -81,7 +81,7 @@ def branch_name(self) -> str:
             return self.branch
 
         if not self.git:
-            self.git = Repo(self.root_directory)
+            self.git = GitRepoManager(self.root_directory)
 
         self.branch = str(self.git.active_branch)
 

infrahub_sdk/pytest_plugin/items/base.py

@@ -6,7 +6,6 @@
 
 import pytest
 import ujson
-from git.exc import InvalidGitRepositoryError
 
 from ..exceptions import InvalidResourceConfigError
 from ..models import InfrahubInputOutputTest
@@ -28,7 +27,6 @@ def __init__(
         **kwargs: dict[str, Any],
     ):
         super().__init__(*args, **kwargs)  # type: ignore[arg-type]
-
         self.resource_name: str = resource_name
         self.resource_config: InfrahubRepositoryConfigElement = resource_config
         self.test: InfrahubTest = test
@@ -68,9 +66,6 @@ def runtest(self) -> None:
         """Run the test logic."""
 
     def repr_failure(self, excinfo: pytest.ExceptionInfo, style: str | None = None) -> str:  # noqa: ARG002
-        if isinstance(excinfo.value, InvalidGitRepositoryError):
-            return f"Invalid Git repository at {excinfo.value}"
-
         return str(excinfo.value)
 
     def reportinfo(self) -> tuple[Path | str, int | None, str]:

infrahub_sdk/repository.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from dulwich import porcelain
+from dulwich.repo import Repo
+
+
+class GitRepoManager:
+    def __init__(self, root_directory: str, branch: str = "main"):
+        self.root_directory = root_directory
+        self.branch = branch
+        self.git: Repo = self.initialize_repo()
+
+    def initialize_repo(self) -> Repo:
+        # Check if the directory already has a repository
+
+        root_path = Path(self.root_directory)
+
+        if root_path.exists() and (root_path / ".git").is_dir():
+            repo = Repo(self.root_directory)  # Open existing repo
+        else:
+            repo = Repo.init(self.root_directory, default_branch=self.branch.encode("utf-8"))
+
+        if not repo:
+            raise ValueError("Failed to initialize or open a repository.")
+
+        return repo
+
+    @property
+    def active_branch(self) -> str | None:
+        active_branch = porcelain.active_branch(self.root_directory).decode("utf-8")
+        return active_branch

infrahub_sdk/testing/repository.py

@@ -7,10 +7,11 @@
 from pathlib import Path
 from typing import TYPE_CHECKING
 
-from git.repo import Repo
+from dulwich import porcelain
 
 from infrahub_sdk.graphql import Mutation
 from infrahub_sdk.protocols import CoreGenericRepository
+from infrahub_sdk.repository import GitRepoManager
 
 if TYPE_CHECKING:
     from infrahub_sdk import InfrahubClient
@@ -37,14 +38,14 @@ class GitRepo:
 
     type: GitRepoType = GitRepoType.INTEGRATED
 
-    _repo: Repo | None = None
+    _repo: GitRepoManager | None = None
     initial_branch: str = "main"
     directories_to_ignore: list[str] = field(default_factory=list)
     remote_directory_name: str = "/remote"
     _branches: list[str] = field(default_factory=list)
 
     @property
-    def repo(self) -> Repo:
+    def repo(self) -> GitRepoManager:
         if self._repo:
             return self._repo
         raise ValueError("Repo hasn't been initialized yet")
@@ -62,12 +63,17 @@ def init(self) -> None:
             dst=self.dst_directory / self.name,
             ignore=shutil.ignore_patterns(".git"),
         )
-        self._repo = Repo.init(self.dst_directory / self.name, initial_branch=self.initial_branch)
-        for untracked in self.repo.untracked_files:
-            self.repo.index.add(untracked)
-        self.repo.index.commit("First commit")
 
-        self.repo.git.checkout(self.initial_branch)
+        self._repo = GitRepoManager(str(Path(self.dst_directory / self.name)), branch=self.initial_branch)
+
+        files = list(
+            porcelain.get_untracked_paths(self._repo.git.path, self._repo.git.path, self._repo.git.open_index())
+        )
+        files_to_add = [str(Path(self._repo.git.path) / t) for t in files]
+        if files_to_add:
+            porcelain.add(repo=self._repo.git.path, paths=files_to_add)
+            porcelain.commit(repo=self._repo.git.path, message="First commit")
+            porcelain.checkout_branch(self._repo.git, self.initial_branch.encode("utf-8"))
 
     async def add_to_infrahub(self, client: InfrahubClient, branch: str | None = None) -> dict:
         input_data = {

infrahub_sdk/transforms.py

@@ -5,7 +5,7 @@
 from abc import abstractmethod
 from typing import TYPE_CHECKING, Any
 
-from git import Repo
+from infrahub_sdk.repository import GitRepoManager
 
 from .exceptions import UninitializedError
 
@@ -27,7 +27,7 @@ def __init__(
         server_url: str = "",
         client: InfrahubClient | None = None,
     ):
-        self.git: Repo
+        self.git: GitRepoManager
 
         self.branch = branch
         self.server_url = server_url or os.environ.get("INFRAHUB_URL", "http://127.0.0.1:8000")
@@ -56,7 +56,7 @@ def branch_name(self) -> str:
             return self.branch
 
         if not hasattr(self, "git") or not self.git:
-            self.git = Repo(self.root_directory)
+            self.git = GitRepoManager(self.root_directory)
 
         self.branch = str(self.git.active_branch)
 

infrahub_sdk/utils.py

@@ -9,13 +9,14 @@
 
 import httpx
 import ujson
-from git.repo import Repo
 from graphql import (
     FieldNode,
     InlineFragmentNode,
     SelectionSetNode,
 )
 
+from infrahub_sdk.repository import GitRepoManager
+
 from .exceptions import JsonDecodeError
 
 if TYPE_CHECKING:
@@ -246,7 +247,7 @@ def get_branch(branch: str | None = None, directory: str | Path = ".") -> str:
     if branch:
         return branch
 
-    repo = Repo(directory)
+    repo = GitRepoManager(directory)
     return str(repo.active_branch)