fix: use configured default_branch in generators, transforms, checks

changelog/1290.fixed.md

@@ -0,0 +1 @@
+Fixed generators, transforms and checks run without an explicit branch, which targeted the local Git branch instead of the configured `default_branch` and failed with `BranchNotFoundError` when that Git branch did not exist in Infrahub. The local Git branch is now used only when `default_branch_from_git` is enabled.

infrahub_sdk/checks.py

@@ -10,9 +10,8 @@
 import ujson
 from pydantic import BaseModel, Field
 
-from infrahub_sdk.repository import GitRepoManager
-
 from .exceptions import UninitializedError
+from .utils import get_branch
 
 if TYPE_CHECKING:
     from . import InfrahubClient
@@ -44,7 +43,6 @@ def __init__(
         params: dict | None = None,
         client: InfrahubClient | None = None,
     ) -> None:
-        self.git: GitRepoManager | None = None
         self.initializer = initializer or InfrahubCheckInitializer()
 
         self.logs: list[dict[str, Any]] = []
@@ -132,14 +130,16 @@ def log_entries(self) -> str:
 
     @property
     def branch_name(self) -> str:
-        """Return the name of the current git branch."""
+        """Return the name of the Infrahub branch this check targets."""
         if self.branch:
             return self.branch
 
-        if not self.git:
-            self.git = GitRepoManager(self.root_directory)
+        if self._client:
+            # The config resolver honours the `default_branch_from_git` flag.
+            self.branch = self._client.config.get_default_infrahub_branch(directory=self.root_directory)
+        else:
+            self.branch = get_branch(directory=self.root_directory)
 
-        self.branch = str(self.git.active_branch)
         return self.branch
 
     @abstractmethod

infrahub_sdk/config.py

@@ -2,6 +2,7 @@
 
 import ssl
 from copy import deepcopy
+from pathlib import Path
 from typing import Any
 
 from pydantic import Field, PrivateAttr, field_validator, model_validator
@@ -221,13 +222,16 @@ def validate_proxy_config(self) -> Self:
             raise ValueError("'proxy' and 'proxy_mounts' are mutually exclusive")
         return self
 
-    @property
-    def default_infrahub_branch(self) -> str:
+    def get_default_infrahub_branch(self, directory: str | Path = ".") -> str:
         branch: str | None = None
         if not self.default_branch_from_git:
             branch = self.default_branch
 
-        return get_branch(branch=branch)
+        return get_branch(branch=branch, directory=directory)
+
+    @property
+    def default_infrahub_branch(self) -> str:
+        return self.get_default_infrahub_branch()
 
     @property
     def password_authentication(self) -> bool:

infrahub_sdk/operation.py

@@ -3,8 +3,6 @@
 import pathlib
 from typing import TYPE_CHECKING
 
-from .repository import GitRepoManager
-
 if TYPE_CHECKING:
     from . import InfrahubClient
     from .node import InfrahubNode
@@ -20,26 +18,18 @@ def __init__(
         branch: str,
         root_directory: str,
     ) -> None:
-        self.branch = branch
         self.convert_query_response = convert_query_response
         self.root_directory = root_directory or str(pathlib.Path.cwd())
+        # The config resolver honours the `default_branch_from_git` flag.
+        self.branch = branch or client.config.get_default_infrahub_branch(directory=self.root_directory)
         self.infrahub_node = infrahub_node
         self._nodes: list[InfrahubNode] = []
         self._related_nodes: list[InfrahubNode] = []
-        self._init_client = client.clone(branch=self.branch_name)
-        self.git: GitRepoManager | None = None
+        self._init_client = client.clone(branch=self.branch)
 
     @property
     def branch_name(self) -> str:
-        """Return the name of the current git branch."""
-        if self.branch:
-            return self.branch
-
-        if not hasattr(self, "git") or not self.git:
-            self.git = GitRepoManager(self.root_directory)
-
-        self.branch = str(self.git.active_branch)
-
+        """Return the name of the Infrahub branch this operation targets."""
         return self.branch
 
     @property

tests/unit/sdk/checks/test_checks.py

@@ -5,7 +5,7 @@
 
 import pytest
 
-from infrahub_sdk import InfrahubClient
+from infrahub_sdk import Config, InfrahubClient
 from infrahub_sdk.checks import InfrahubCheck
 
 if TYPE_CHECKING:
@@ -72,3 +72,48 @@ def validate(self, data: dict) -> None:
     await check.run()
 
     assert check.passed is False
+
+
+async def test_branch_name_falls_back_to_configured_default_branch() -> None:
+    """Without an explicit branch, the configured default branch wins over the local Git branch."""
+
+    class IFCheck(InfrahubCheck):
+        query = "my_query"
+
+        def validate(self, data: dict) -> None: ...
+
+    client = InfrahubClient(config=Config(address="http://mock", default_branch="test"))
+
+    assert IFCheck(client=client).branch_name == "test"
+    assert IFCheck(client=client, branch="explicit").branch_name == "explicit"
+
+
+async def test_branch_name_falls_back_to_git_branch_without_a_client(monkeypatch: pytest.MonkeyPatch) -> None:
+    class IFCheck(InfrahubCheck):
+        query = "my_query"
+
+        def validate(self, data: dict) -> None: ...
+
+    monkeypatch.setattr("infrahub_sdk.checks.get_branch", lambda **_: "my-git-branch")
+    assert IFCheck().branch_name == "my-git-branch"
+
+
+async def test_git_branch_is_read_from_the_root_directory(monkeypatch: pytest.MonkeyPatch) -> None:
+    """With `default_branch_from_git`, the Git branch comes from the check's own repository."""
+
+    class IFCheck(InfrahubCheck):
+        query = "my_query"
+
+        def validate(self, data: dict) -> None: ...
+
+    # The stub encodes the directory it was asked about, so the assertion pins which repository
+    # the branch was resolved from.
+    monkeypatch.setattr(
+        "infrahub_sdk.config.get_branch",
+        lambda branch=None, directory=".": branch or f"git:{directory}",
+    )
+    client = InfrahubClient(config=Config(address="http://mock", default_branch="test", default_branch_from_git=True))
+
+    check = IFCheck(client=client, root_directory="/some/repository")
+
+    assert check.branch_name == "git:/some/repository"

tests/unit/sdk/test_operation.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+import pathlib
+from typing import TYPE_CHECKING
+
+from infrahub_sdk import Config, InfrahubClient
+from infrahub_sdk.node import InfrahubNode
+from infrahub_sdk.transforms import InfrahubTransform
+
+if TYPE_CHECKING:
+    import pytest
+
+
+class DummyTransform(InfrahubTransform):
+    query = "my_query"
+
+    def transform(self, data: dict) -> dict:
+        return data
+
+
+def _build_transform(client: InfrahubClient, branch: str = "", root_directory: str = "") -> DummyTransform:
+    return DummyTransform(client=client, infrahub_node=InfrahubNode, branch=branch, root_directory=root_directory)
+
+
+def _stub_git_branch(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Replace the Git lookup so tests never depend on the state of the local checkout.
+
+    The stub encodes the directory it was asked about, so callers can assert which repository
+    the branch was resolved from.
+    """
+
+    def fake_get_branch(branch: str | None = None, directory: str = ".") -> str:
+        return branch or f"git:{directory}"
+
+    monkeypatch.setattr("infrahub_sdk.config.get_branch", fake_get_branch)
+
+
+async def test_branch_name_uses_explicit_branch() -> None:
+    client = InfrahubClient(config=Config(address="http://mock", default_branch="test"))
+
+    transform = _build_transform(client=client, branch="explicit")
+
+    assert transform.branch_name == "explicit"
+    assert transform._init_client.default_branch == "explicit"
+
+
+async def test_branch_name_falls_back_to_configured_default_branch() -> None:
+    """Without an explicit branch, the configured default branch wins over the local Git branch."""
+    client = InfrahubClient(config=Config(address="http://mock", default_branch="test"))
+
+    transform = _build_transform(client=client)
+
+    assert transform.branch_name == "test"
+    assert transform._init_client.default_branch == "test"
+
+
+async def test_branch_name_falls_back_to_git_branch_when_opted_in(monkeypatch: pytest.MonkeyPatch) -> None:
+    _stub_git_branch(monkeypatch)
+    client = InfrahubClient(config=Config(address="http://mock", default_branch="test", default_branch_from_git=True))
+
+    transform = _build_transform(client=client)
+
+    assert transform.branch_name == f"git:{pathlib.Path.cwd()}"
+
+
+async def test_git_branch_is_read_from_the_root_directory(monkeypatch: pytest.MonkeyPatch) -> None:
+    _stub_git_branch(monkeypatch)
+    client = InfrahubClient(config=Config(address="http://mock", default_branch="test", default_branch_from_git=True))
+
+    transform = _build_transform(client=client, root_directory="/some/repository")
+
+    assert transform.branch_name == "git:/some/repository"