diff --git a/changelog/1290.fixed.md b/changelog/1290.fixed.md new file mode 100644 index 000000000..57fda4e1d --- /dev/null +++ b/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. diff --git a/infrahub_sdk/checks.py b/infrahub_sdk/checks.py index 36ce1199a..507136aed 100644 --- a/infrahub_sdk/checks.py +++ b/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 diff --git a/infrahub_sdk/config.py b/infrahub_sdk/config.py index f81fc911e..6af56e8d2 100644 --- a/infrahub_sdk/config.py +++ b/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: diff --git a/infrahub_sdk/operation.py b/infrahub_sdk/operation.py index 8ecd0173d..b864d8201 100644 --- a/infrahub_sdk/operation.py +++ b/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 diff --git a/tests/unit/sdk/checks/test_checks.py b/tests/unit/sdk/checks/test_checks.py index 6433b8b42..27b0da556 100644 --- a/tests/unit/sdk/checks/test_checks.py +++ b/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" diff --git a/tests/unit/sdk/test_operation.py b/tests/unit/sdk/test_operation.py new file mode 100644 index 000000000..278e7fc1d --- /dev/null +++ b/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"