Add delete branch from git

backend/infrahub/core/branch/tasks.py

@@ -52,6 +52,7 @@
     DIFF_REFRESH_ALL,
     DIFF_UPDATE,
     GIT_REPOSITORIES_CREATE_BRANCH,
+    GIT_REPOSITORIES_DELETE_BRANCH,
     IPAM_RECONCILIATION,
     TRIGGER_ARTIFACT_DEFINITION_GENERATE,
     TRIGGER_GENERATOR_DEFINITION_RUN,
@@ -411,7 +412,7 @@ async def merge_branch(branch: str, context: InfrahubContext, proposed_change_id
 
 
 @flow(name="branch-delete", flow_run_name="Delete branch {branch}")
-async def delete_branch(branch: str, context: InfrahubContext) -> None:
+async def delete_branch(branch: str, context: InfrahubContext, delete_from_git: bool = False) -> None:
     await add_tags(branches=[branch])
 
     database = await get_database()
@@ -438,6 +439,14 @@ async def delete_branch(branch: str, context: InfrahubContext) -> None:
         event_service = await get_event_service()
         await event_service.send(event=event)
 
+    should_delete_git = (config.SETTINGS.git.delete_git_branch_after_merge or delete_from_git) and obj.sync_with_git
+    if should_delete_git:
+        await get_workflow().submit_workflow(
+            workflow=GIT_REPOSITORIES_DELETE_BRANCH,
+            context=context,
+            parameters={"branch": branch},
+        )
+
 
 @flow(
     name="branch-validate",

backend/infrahub/git/base.py

@@ -536,6 +536,20 @@ def get_branches_from_local(self, include_worktree: bool = True) -> dict[str, Br
 
         return branches
 
+    def origin_has_branch(self, branch_name: str) -> bool:
+        """Return True if branch_name exists as a remote branch on origin."""
+        return branch_name in self.get_branches_from_remote()
+
+    async def delete_remote_branch(self, branch_name: str) -> None:
+        """Delete branch_name from origin and remove the local tracking ref."""
+        if not self.has_origin:
+            return
+        repo = self.get_git_repo_main()
+        repo.git.push("origin", "--delete", branch_name)
+        local_branches = self.get_branches_from_local(include_worktree=False)
+        if branch_name in local_branches:
+            repo.delete_head(branch_name, force=True)
+
     @abstractmethod
     def get_commit_value(self, branch_name: str, remote: bool = False) -> str:
         raise NotImplementedError()

backend/infrahub/git/tasks.py

@@ -160,6 +160,25 @@ async def create_branch(branch: str, branch_id: str) -> None:
         pass
 
 
+@flow(name="git-repositories-delete-branch", flow_run_name="Delete git branch '{branch}'")
+async def delete_git_branch(branch: str) -> None:
+    """Fan out branch deletion across all CoreRepository instances."""
+    client = get_client()
+    repositories: list[CoreRepository] = await client.filters(kind=CoreRepository)
+    batch = await client.create_batch()
+    for repository in repositories:
+        batch.add(
+            task=git_branch_delete,
+            client=client,
+            branch=branch,
+            repository_name=repository.name.value,
+            repository_id=repository.id,
+            repository_location=repository.location.value,
+        )
+    async for _, _ in batch.execute():
+        pass
+
+
 @flow(name="sync-git-repo-with-origin", flow_run_name="Sync git repo with origin")
 async def sync_git_repo_with_origin_and_tag_on_failure(
     client: InfrahubClient,
@@ -312,6 +331,33 @@ async def git_branch_create(
         log.debug("Sent message to all workers to fetch the latest version of the repository (RefreshGitFetch)")
 
 
+@task(  # type: ignore[arg-type]
+    name="git-branch-delete",
+    task_run_name="Delete branch '{branch}' in repository {repository_name}",
+    cache_policy=NONE,
+)
+async def git_branch_delete(
+    client: InfrahubClient,
+    branch: str,
+    repository_id: str,
+    repository_name: str,
+    repository_location: str,
+) -> None:
+    log = get_run_logger()
+    await add_branch_tag(branch_name=branch)
+    repo = await InfrahubRepository.init(
+        id=repository_id, name=repository_name, location=repository_location, client=client
+    )
+    async with lock.registry.get(name=repository_name, namespace="repository"):
+        if not repo.origin_has_branch(branch):
+            return
+
+        try:
+            await repo.delete_remote_branch(branch_name=branch)
+        except Exception as exc:
+            log.exception(f"Failed to delete Git branch '{branch}' from repository '{repository_name}' - {str(exc)}")
+
+
 @flow(name="artifact-definition-generate", flow_run_name="Generate all artifacts")
 async def generate_artifact_definition(branch: str, context: InfrahubContext) -> None:
     await add_branch_tag(branch_name=branch)

backend/infrahub/graphql/mutations/branch.py

@@ -109,6 +109,11 @@ class BranchNameInput(InputObjectType):
     name = String(required=False)
 
 
+class BranchDeleteInput(InputObjectType):
+    name = String(required=False)
+    delete_from_git = Boolean(required=False, default_value=False)
+
+
 class BranchUpdateInput(InputObjectType):
     name = String(required=True)
     description = String(required=False)
@@ -117,7 +122,7 @@ class BranchUpdateInput(InputObjectType):
 
 class BranchDelete(Mutation):
     class Arguments:
-        data = BranchNameInput(required=True)
+        data = BranchDeleteInput(required=True)
         context = ContextInput(required=False)
         wait_until_completion = Boolean(required=False)
 
@@ -129,22 +134,27 @@ async def mutate(
         cls,
         root: dict,  # noqa: ARG003
         info: GraphQLResolveInfo,
-        data: BranchNameInput,
+        data: BranchDeleteInput,
         context: ContextInput | None = None,
         wait_until_completion: bool = True,
     ) -> Self:
         graphql_context: GraphqlContext = info.context
         obj = await Branch.get_by_name(db=graphql_context.db, name=str(data.name))
         await apply_external_context(graphql_context=graphql_context, context_input=context)
 
+        parameters = {
+            "branch": obj.name,
+            "delete_from_git": bool(data.delete_from_git),
+        }
+
         if wait_until_completion:
             await graphql_context.active_service.workflow.execute_workflow(
-                workflow=BRANCH_DELETE, context=graphql_context.get_context(), parameters={"branch": obj.name}
+                workflow=BRANCH_DELETE, context=graphql_context.get_context(), parameters=parameters
             )
             return cls(ok=True)
 
         workflow = await graphql_context.active_service.workflow.submit_workflow(
-            workflow=BRANCH_DELETE, context=graphql_context.get_context(), parameters={"branch": obj.name}
+            workflow=BRANCH_DELETE, context=graphql_context.get_context(), parameters=parameters
         )
         return cls(ok=True, task={"id": str(workflow.id)})
 

backend/infrahub/workflows/catalogue.py

@@ -192,6 +192,14 @@
     tags=[WorkflowTag.DATABASE_CHANGE],
 )
 
+GIT_REPOSITORIES_DELETE_BRANCH = WorkflowDefinition(
+    name="git-repositories-delete-branch",
+    type=WorkflowType.CORE,
+    module="infrahub.git.tasks",
+    function="delete_git_branch",
+    tags=[WorkflowTag.DATABASE_CHANGE],
+)
+
 GIT_REPOSITORY_ADD = WorkflowDefinition(
     name="git-repository-add-read-write",
     type=WorkflowType.CORE,
@@ -687,6 +695,7 @@
     GIT_READ_ONLY_REPOSITORY_IMPORT_LAST_COMMIT,
     GIT_REPOSITORIES_CHECK_ARTIFACT_CREATE,
     GIT_REPOSITORIES_CREATE_BRANCH,
+    GIT_REPOSITORIES_DELETE_BRANCH,
     GIT_REPOSITORIES_DIFF_NAMES_ONLY,
     GIT_REPOSITORIES_IMPORT_OBJECTS,
     GIT_REPOSITORIES_MERGE,

backend/tests/component/git/test_delete_git_branch.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+from git import Repo
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from infrahub.git import InfrahubRepository
+
+
+async def test_has_branch_returns_true_for_existing_branch(git_repo_01: InfrahubRepository) -> None:
+    assert git_repo_01.origin_has_branch("branch01") is True
+
+
+async def test_has_branch_returns_false_for_missing_branch(git_repo_01: InfrahubRepository) -> None:
+    assert git_repo_01.origin_has_branch("nonexistent-branch-xyz") is False
+
+
+async def test_delete_remote_branch_removes_branch_from_origin(
+    git_repo_01: InfrahubRepository,
+    git_upstream_repo_01: dict[str, str | Path],
+) -> None:
+    assert git_repo_01.origin_has_branch("clean-branch") is True
+
+    await git_repo_01.delete_remote_branch(branch_name="clean-branch")
+
+    # Fetch to sync remote tracking refs, then verify it's gone
+    await git_repo_01.fetch()
+    assert git_repo_01.origin_has_branch("clean-branch") is False
+
+    # Verify branch is also gone from the upstream (origin) repo
+    upstream = Repo(git_upstream_repo_01["path"])
+    upstream_branches = [b.name for b in upstream.refs if not b.is_remote()]
+    assert "clean-branch" not in upstream_branches
+
+
+@pytest.mark.parametrize("branch_name", ["branch01", "branch02"])
+async def test_has_branch_true_for_all_remote_branches(git_repo_01: InfrahubRepository, branch_name: str) -> None:
+    assert git_repo_01.origin_has_branch(branch_name) is True

backend/tests/functional/branch/test_delete_git_branch.py

@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from unittest.mock import ANY, AsyncMock, patch
+
+import pytest
+from infrahub_sdk.graphql import Mutation
+
+from infrahub import config
+from infrahub.services.adapters.workflow.local import WorkflowLocalExecution
+from infrahub.workflows.catalogue import GIT_REPOSITORIES_DELETE_BRANCH
+from tests.helpers.schema import CAR_SCHEMA, load_schema
+from tests.helpers.test_app import TestInfrahubApp
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from infrahub_sdk import InfrahubClient
+
+    from infrahub.database import InfrahubDatabase
+    from tests.adapters.message_bus import BusSimulator
+
+
+class TestDeleteBranchGitWorkflow(TestInfrahubApp):
+    @pytest.fixture(scope="class")
+    async def initial_dataset(
+        self,
+        db: InfrahubDatabase,
+        initialize_registry: None,
+        git_repos_source_dir_module_scope: Path,
+        client: InfrahubClient,
+        bus_simulator: BusSimulator,
+        prefect_test_fixture: None,
+    ) -> None:
+        await load_schema(db, schema=CAR_SCHEMA)
+
+    async def test_git_deletion_triggered_when_config_enabled_and_sync_with_git(
+        self, initial_dataset: None, client: InfrahubClient, delete_git_branch_after_merge_reset_config: None
+    ) -> None:
+        config.SETTINGS.git.delete_git_branch_after_merge = True
+        branch = await client.branch.create(branch_name="git_del_cfg_enabled", sync_with_git=True)
+
+        with patch.object(WorkflowLocalExecution, "submit_workflow", new_callable=AsyncMock) as mock_submit:
+            query = Mutation(
+                mutation="BranchDelete",
+                input_data={"data": {"name": branch.name}},
+                query={"ok": None},
+            )
+            await client.execute_graphql(query=query.render())
+
+            mock_submit.assert_any_call(
+                workflow=GIT_REPOSITORIES_DELETE_BRANCH,
+                context=ANY,
+                parameters={"branch": branch.name},
+            )
+
+    async def test_git_deletion_not_triggered_when_config_disabled(
+        self, initial_dataset: None, client: InfrahubClient, delete_git_branch_after_merge_reset_config: None
+    ) -> None:
+        config.SETTINGS.git.delete_git_branch_after_merge = False
+        branch = await client.branch.create(branch_name="git_del_cfg_disabled", sync_with_git=True)
+
+        with patch.object(WorkflowLocalExecution, "submit_workflow", new_callable=AsyncMock) as mock_submit:
+            query = Mutation(
+                mutation="BranchDelete",
+                input_data={"data": {"name": branch.name}},
+                query={"ok": None},
+            )
+            await client.execute_graphql(query=query.render())
+
+            git_del_calls = [
+                c for c in mock_submit.call_args_list if c.kwargs.get("workflow") == GIT_REPOSITORIES_DELETE_BRANCH
+            ]
+            assert not git_del_calls
+
+    async def test_git_deletion_not_triggered_when_branch_not_sync_with_git(
+        self, initial_dataset: None, client: InfrahubClient, delete_git_branch_after_merge_reset_config: None
+    ) -> None:
+        config.SETTINGS.git.delete_git_branch_after_merge = True
+        branch = await client.branch.create(branch_name="git_del_no_sync", sync_with_git=False)
+
+        with patch.object(WorkflowLocalExecution, "submit_workflow", new_callable=AsyncMock) as mock_submit:
+            query = Mutation(
+                mutation="BranchDelete",
+                input_data={"data": {"name": branch.name}},
+                query={"ok": None},
+            )
+            await client.execute_graphql(query=query.render())
+
+            git_del_calls = [
+                c for c in mock_submit.call_args_list if c.kwargs.get("workflow") == GIT_REPOSITORIES_DELETE_BRANCH
+            ]
+            assert not git_del_calls
+
+    async def test_git_deletion_triggered_when_delete_from_git_true_and_config_disabled(
+        self, initial_dataset: None, client: InfrahubClient, delete_git_branch_after_merge_reset_config: None
+    ) -> None:
+        config.SETTINGS.git.delete_git_branch_after_merge = False
+        branch = await client.branch.create(branch_name="git_del_explicit_true", sync_with_git=True)
+
+        with patch.object(WorkflowLocalExecution, "submit_workflow", new_callable=AsyncMock) as mock_submit:
+            query = Mutation(
+                mutation="BranchDelete",
+                input_data={"data": {"name": branch.name, "delete_from_git": True}},
+                query={"ok": None},
+            )
+            await client.execute_graphql(query=query.render())
+
+            mock_submit.assert_any_call(
+                workflow=GIT_REPOSITORIES_DELETE_BRANCH,
+                context=ANY,
+                parameters={"branch": branch.name},
+            )
+
+    async def test_git_deletion_not_triggered_when_delete_from_git_false_and_config_disabled(
+        self, initial_dataset: None, client: InfrahubClient, delete_git_branch_after_merge_reset_config: None
+    ) -> None:
+        config.SETTINGS.git.delete_git_branch_after_merge = False
+        branch = await client.branch.create(branch_name="git_del_explicit_false", sync_with_git=True)
+
+        with patch.object(WorkflowLocalExecution, "submit_workflow", new_callable=AsyncMock) as mock_submit:
+            query = Mutation(
+                mutation="BranchDelete",
+                input_data={"data": {"name": branch.name, "delete_from_git": False}},
+                query={"ok": None},
+            )
+            await client.execute_graphql(query=query.render())
+
+            git_del_calls = [
+                c for c in mock_submit.call_args_list if c.kwargs.get("workflow") == GIT_REPOSITORIES_DELETE_BRANCH
+            ]
+            assert not git_del_calls

backend/tests/functional/conftest.py

@@ -17,3 +17,10 @@ def delete_branch_after_merge_reset_config() -> Generator[None]:
     original = config.SETTINGS.main.delete_branch_after_merge
     yield
     config.SETTINGS.main.delete_branch_after_merge = original
+
+
+@pytest.fixture
+def delete_git_branch_after_merge_reset_config() -> Generator[None]:
+    original = config.SETTINGS.git.delete_git_branch_after_merge
+    yield
+    config.SETTINGS.git.delete_git_branch_after_merge = original