Migrate GitRepositoryAdd to prefect

backend/infrahub/git/models.py

@@ -35,6 +35,18 @@ class RequestArtifactGenerate(BaseModel):
     variables: dict = Field(..., description="Input variables when generating the artifact")
 
 
+class GitRepositoryAdd(BaseModel):
+    """Clone and sync an external repository after creation."""
+
+    location: str = Field(..., description="The external URL of the repository")
+    repository_id: str = Field(..., description="The unique ID of the Repository")
+    repository_name: str = Field(..., description="The name of the repository")
+    created_by: Optional[str] = Field(default=None, description="The user ID of the user that created the repository")
+    default_branch_name: Optional[str] = Field(None, description="Default branch for this repository")
+    infrahub_branch_name: str = Field(..., description="Infrahub branch on which to sync the remote repository")
+    internal_status: str = Field(..., description="Administrative status of the repository")
+
+
 class GitRepositoryPullReadOnly(BaseModel):
     """Update a read-only repository to the latest commit for its ref"""
 

backend/infrahub/git/tasks.py

@@ -13,6 +13,7 @@
 from ..workflows.catalogue import REQUEST_ARTIFACT_DEFINITION_GENERATE, REQUEST_ARTIFACT_GENERATE
 from ..workflows.utils import add_branch_tag
 from .models import (
+    GitRepositoryAdd,
     GitRepositoryMerge,
     GitRepositoryPullReadOnly,
     RequestArtifactDefinitionGenerate,
@@ -23,6 +24,32 @@
 log = get_logger()
 
 
+@flow(name="git-repository-add-read-write")
+async def add_git_repository(model: GitRepositoryAdd) -> None:
+    service = services.service
+    async with service.git_report(
+        related_node=model.repository_id,
+        title=f"Initial import of the repository in branch: {model.infrahub_branch_name}",
+        created_by=model.created_by,
+    ) as git_report:
+        async with lock.registry.get(name=model.repository_name, namespace="repository"):
+            repo = await InfrahubRepository.new(
+                id=model.repository_id,
+                name=model.repository_name,
+                location=model.location,
+                client=service.client,
+                task_report=git_report,
+                infrahub_branch_name=model.infrahub_branch_name,
+                internal_status=model.internal_status,
+                default_branch_name=model.default_branch_name,
+            )
+            await repo.import_objects_from_files(
+                infrahub_branch_name=model.infrahub_branch_name, git_branch_name=model.default_branch_name
+            )
+            if model.internal_status == RepositoryInternalStatus.ACTIVE.value:
+                await repo.sync()
+
+
 @flow(name="git_repositories_create_branch")
 async def create_branch(branch: str, branch_id: str) -> None:
     """Request to the creation of git branches in available repositories."""

backend/infrahub/graphql/mutations/repository.py

@@ -10,12 +10,12 @@
 from infrahub.core.protocols import CoreGenericRepository, CoreReadOnlyRepository, CoreRepository
 from infrahub.core.schema import NodeSchema
 from infrahub.exceptions import ValidationError
-from infrahub.git.models import GitRepositoryPullReadOnly
+from infrahub.git.models import GitRepositoryAdd, GitRepositoryPullReadOnly
 from infrahub.graphql.types.common import IdentifierInput
 from infrahub.log import get_logger
 from infrahub.message_bus import messages
 from infrahub.message_bus.messages.git_repository_connectivity import GitRepositoryConnectivityResponse
-from infrahub.workflows.catalogue import GIT_REPOSITORIES_PULL_READ_ONLY
+from infrahub.workflows.catalogue import GIT_REPOSITORIES_PULL_READ_ONLY, GIT_REPOSITORY_ADD
 
 from .main import InfrahubMutationMixin, InfrahubMutationOptions
 
@@ -99,9 +99,12 @@ async def mutate_create(
                 internal_status=obj.internal_status.value,
                 created_by=authenticated_user,
             )
+            if context.service:
+                await context.service.send(message=message)
+
         else:
             obj = cast(CoreRepository, obj)
-            message = messages.GitRepositoryAdd(
+            git_repo_add_model = GitRepositoryAdd(
                 repository_id=obj.id,
                 repository_name=obj.name.value,
                 location=obj.location.value,
@@ -111,8 +114,10 @@ async def mutate_create(
                 created_by=authenticated_user,
             )
 
-        if context.service:
-            await context.service.send(message=message)
+            if context.service:
+                await context.service.workflow.submit_workflow(
+                    workflow=GIT_REPOSITORY_ADD, parameters={"model": git_repo_add_model}
+                )
 
         # TODO Validate that the creation of the repository went as expected
 

backend/infrahub/message_bus/messages/__init__.py

@@ -15,7 +15,6 @@
 from .finalize_validator_execution import FinalizeValidatorExecution
 from .git_diff_namesonly import GitDiffNamesOnly, GitDiffNamesOnlyResponse
 from .git_file_get import GitFileGet, GitFileGetResponse
-from .git_repository_add import GitRepositoryAdd
 from .git_repository_connectivity import GitRepositoryConnectivity
 from .git_repository_importobjects import GitRepositoryImportObjects
 from .git_repository_read_only_add import GitRepositoryAddReadOnly
@@ -54,7 +53,6 @@
     "finalize.validator.execution": FinalizeValidatorExecution,
     "git.diff.names_only": GitDiffNamesOnly,
     "git.file.get": GitFileGet,
-    "git.repository.add": GitRepositoryAdd,
     "git.repository.connectivity": GitRepositoryConnectivity,
     "git.repository.add_read_only": GitRepositoryAddReadOnly,
     "git.repository.import_objects": GitRepositoryImportObjects,

backend/infrahub/message_bus/operations/__init__.py

@@ -34,7 +34,6 @@
     "finalize.validator.execution": finalize.validator.execution,
     "git.diff.names_only": git.diff.names_only,
     "git.file.get": git.file.get,
-    "git.repository.add": git.repository.add,
     "git.repository.add_read_only": git.repository.add_read_only,
     "git.repository.connectivity": git.repository.connectivity,
     "git.repository.import_objects": git.repository.import_objects,

backend/infrahub/message_bus/operations/git/repository.py

@@ -15,37 +15,6 @@
 log = get_logger()
 
 
-@flow(name="git-repository-add-read-write")
-async def add(message: messages.GitRepositoryAdd, service: InfrahubServices) -> None:
-    log.info(
-        "Cloning and importing repository",
-        repository=message.repository_name,
-        location=message.location,
-        internal_status=message.internal_status,
-    )
-    async with service.git_report(
-        related_node=message.repository_id,
-        title=f"Initial import of the repository in branch: {message.infrahub_branch_name}",
-        created_by=message.created_by,
-    ) as git_report:
-        async with lock.registry.get(name=message.repository_name, namespace="repository"):
-            repo = await InfrahubRepository.new(
-                id=message.repository_id,
-                name=message.repository_name,
-                location=message.location,
-                client=service.client,
-                task_report=git_report,
-                infrahub_branch_name=message.infrahub_branch_name,
-                internal_status=message.internal_status,
-                default_branch_name=message.default_branch_name,
-            )
-            await repo.import_objects_from_files(
-                infrahub_branch_name=message.infrahub_branch_name, git_branch_name=message.default_branch_name
-            )
-            if message.internal_status == RepositoryInternalStatus.ACTIVE.value:
-                await repo.sync()
-
-
 @flow(name="git-repository-add-read-only")
 async def add_read_only(message: messages.GitRepositoryAddReadOnly, service: InfrahubServices) -> None:
     log.info(

backend/infrahub/workflows/catalogue.py

@@ -138,6 +138,15 @@
     tags=[WorkflowTag.DATABASE_CHANGE],
 )
 
+GIT_REPOSITORY_ADD = WorkflowDefinition(
+    name="git-repository-add-read-write",
+    type=WorkflowType.INTERNAL,
+    module="infrahub.git.tasks",
+    function="add_git_repository",
+    branch_support=BranchSupportType.AWARE,
+    tags=[WorkflowTag.DATABASE_CHANGE],
+)
+
 GIT_REPOSITORIES_PULL_READ_ONLY = WorkflowDefinition(
     name="git-repository-pull-read-only",
     type=WorkflowType.INTERNAL,
@@ -222,4 +231,5 @@
     BRANCH_CANCEL_PROPOSED_CHANGES,
     REQUEST_GENERATOR_DEFINITION_RUN,
     UPDATE_GRAPHQL_QUERY_GROUP,
+    GIT_REPOSITORY_ADD,
 ]

backend/tests/helpers/file_repo.py

@@ -15,6 +15,12 @@
 class FileRepo:
     name: str
     sources_directory: Path
+
+    # Some tests make a prior copy of fixtures/repos/car-dealership folder in a temp folder,
+    # in which case we need to use that temp folder instead of fixture dir. This could probably be removed
+    # when https://github.com/opsmill/infrahub/issues/4296 is fixed.
+    local_repo_base_path: Path = get_fixtures_dir() / "repos"
+
     _repo: Optional[Repo] = None
     _initial_branch: Optional[str] = None
     _branches: list[str] = field(default_factory=list)
@@ -48,7 +54,7 @@ def _apply_pull_requests(self, repo_base: Path) -> None:
             self.repo.git.commit("-m", pull_request)
 
     def __post_init__(self) -> None:
-        repo_base = Path(get_fixtures_dir(), "repos", self.name)
+        repo_base = Path(self.local_repo_base_path, self.name)
         initial_directory = self._initial_directory(repo_base=repo_base)
         shutil.copytree(repo_base / initial_directory, self.sources_directory / self.name)
         self._repo = Repo.init(self.sources_directory / self.name, initial_branch=self._initial_branch)

backend/tests/integration/proposed_change/test_proposed_change_conflict.py

@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+import shutil
+import tempfile
+from pathlib import Path
 from typing import TYPE_CHECKING
 
 import pytest
@@ -10,21 +13,37 @@
 from infrahub.core.manager import NodeManager
 from infrahub.core.node import Node
 from infrahub.services.adapters.cache.redis import RedisCache
+from infrahub.utils import get_fixtures_dir
 from tests.constants import TestKind
 from tests.helpers.file_repo import FileRepo
 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 TestProposedChangePipelineConflict(TestInfrahubApp):
+    @pytest.fixture(scope="class")
+    def car_dealership_copy(self):
+        """
+        Copies car-dealership local repository to a temporary folder, with a new name.
+        This is needed for this test as using car-dealership folder leads to issues most probably
+        related to https://github.com/opsmill/infrahub/issues/4296 as some other tests use this same repository.
+        """
+
+        source_folder = Path(get_fixtures_dir(), "repos", "car-dealership")
+        new_folder_name = "car-dealership-copy"
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = Path(temp_dir)
+            destination_folder = temp_path / new_folder_name
+            shutil.copytree(source_folder, destination_folder)
+            yield temp_path, new_folder_name
+
     @pytest.fixture(scope="class")
     async def initial_dataset(
         self,
@@ -33,6 +52,7 @@ async def initial_dataset(
         git_repos_source_dir_module_scope: Path,
         client: InfrahubClient,
         bus_simulator: BusSimulator,
+        car_dealership_copy: tuple[Path, str],
     ) -> str:
         await load_schema(db, schema=CAR_SCHEMA)
         john = await Node.init(schema=TestKind.PERSON, db=db)
@@ -57,10 +77,11 @@ async def initial_dataset(
         await jesko.save(db=db)
 
         bus_simulator.service.cache = RedisCache()
-        FileRepo(name="car-dealership", sources_directory=git_repos_source_dir_module_scope)
+        repo_path, repo_name = car_dealership_copy
+        FileRepo(name=repo_name, local_repo_base_path=repo_path, sources_directory=git_repos_source_dir_module_scope)
         client_repository = await client.create(
             kind=InfrahubKind.REPOSITORY,
-            data={"name": "car-dealership", "location": f"{git_repos_source_dir_module_scope}/car-dealership"},
+            data={"name": "dealership-car", "location": f"{git_repos_source_dir_module_scope}/{repo_name}"},
         )
         await client_repository.save()
         return client_repository.id

backend/tests/unit/git/test_git_rpc.py

@@ -10,9 +10,9 @@
 from infrahub.core.constants import InfrahubKind, RepositoryInternalStatus
 from infrahub.exceptions import RepositoryError
 from infrahub.git import InfrahubRepository
-from infrahub.git.models import GitRepositoryMerge, GitRepositoryPullReadOnly
+from infrahub.git.models import GitRepositoryAdd, GitRepositoryMerge, GitRepositoryPullReadOnly
 from infrahub.git.repository import InfrahubReadOnlyRepository
-from infrahub.git.tasks import pull_read_only
+from infrahub.git.tasks import add_git_repository, pull_read_only
 from infrahub.lock import InfrahubLockRegistry
 from infrahub.message_bus import Meta, messages
 from infrahub.message_bus.operations import git
@@ -53,28 +53,22 @@ def setup_method(self):
         self.default_branch_name = "default-branch"
         self.client = AsyncMock(spec=InfrahubClient)
         self.git_report = AsyncContextManagerMock()
+        self.original_services = services.service
+        services.service = InfrahubServices(client=self.client)
+        services.service.git_report = self.git_report
 
-        self.services = InfrahubServices(client=self.client)
-        self.services.git_report = self.git_report
-        lock_patcher = patch("infrahub.message_bus.operations.git.repository.lock")
-        self.mock_infra_lock = lock_patcher.start()
-        self.mock_infra_lock.registry = AsyncMock(spec=InfrahubLockRegistry)
-        repo_class_patcher = patch(
-            "infrahub.message_bus.operations.git.repository.InfrahubRepository", spec=InfrahubRepository
-        )
-        self.mock_repo_class = repo_class_patcher.start()
         self.mock_repo = AsyncMock(spec=InfrahubRepository)
         self.mock_repo.default_branch = self.default_branch_name
         self.mock_repo.infrahub_branch_name = self.default_branch_name
         self.mock_repo.internal_status = "active"
-        self.mock_repo_class.new.return_value = self.mock_repo
 
     def teardown_method(self):
         patch.stopall()
+        services.service = self.original_services
 
     async def test_git_rpc_create_successful(self, git_upstream_repo_01: dict[str, str]):
         repo_id = str(UUIDT())
-        message = messages.GitRepositoryAdd(
+        model = GitRepositoryAdd(
             repository_id=repo_id,
             repository_name=git_upstream_repo_01["name"],
             location=git_upstream_repo_01["path"],
@@ -83,25 +77,33 @@ async def test_git_rpc_create_successful(self, git_upstream_repo_01: dict[str, s
             internal_status="active",
         )
 
-        await git.repository.add(message=message, service=self.services)
-
-        self.mock_infra_lock.registry.get.assert_called_once_with(
-            name=git_upstream_repo_01["name"], namespace="repository"
-        )
-        self.mock_repo_class.new.assert_awaited_once_with(
-            id=repo_id,
-            name=git_upstream_repo_01["name"],
-            location=git_upstream_repo_01["path"],
-            client=self.client,
-            task_report=self.git_report,
-            infrahub_branch_name=self.default_branch_name,
-            internal_status="active",
-            default_branch_name=self.default_branch_name,
-        )
-        self.mock_repo.import_objects_from_files.assert_awaited_once_with(
-            infrahub_branch_name=self.default_branch_name, git_branch_name=self.default_branch_name
-        )
-        self.mock_repo.sync.assert_awaited_once_with()
+        with (
+            patch("infrahub.git.tasks.lock") as mock_infra_lock,
+            patch("infrahub.git.tasks.InfrahubRepository", spec=InfrahubRepository) as mock_repo_class,
+        ):
+            mock_infra_lock.registry = AsyncMock(spec=InfrahubLockRegistry)
+            mock_repo_class.new.return_value = self.mock_repo
+
+            await add_git_repository(model=model)
+
+            mock_infra_lock.registry.get.assert_called_once_with(
+                name=git_upstream_repo_01["name"], namespace="repository"
+            )
+
+            mock_repo_class.new.assert_awaited_once_with(
+                id=repo_id,
+                name=git_upstream_repo_01["name"],
+                location=git_upstream_repo_01["path"],
+                client=self.client,
+                task_report=self.git_report,
+                infrahub_branch_name=self.default_branch_name,
+                internal_status="active",
+                default_branch_name=self.default_branch_name,
+            )
+            self.mock_repo.import_objects_from_files.assert_awaited_once_with(
+                infrahub_branch_name=self.default_branch_name, git_branch_name=self.default_branch_name
+            )
+            self.mock_repo.sync.assert_awaited_once_with()
 
 
 async def test_git_rpc_merge(