Merge pull request #4806 from opsmill/lgu-migrate-git-repository-merge

Migrate GitRepositoryMerge to prefect

Author: dgarros

backend/infrahub/core/merge.py

@@ -10,8 +10,9 @@
 from infrahub.core.schema import GenericSchema, NodeSchema
 from infrahub.core.timestamp import Timestamp
 from infrahub.exceptions import ValidationError
-from infrahub.message_bus import messages
 
+from ..git.models import GitRepositoryMerge
+from ..workflows.catalogue import GIT_REPOSITORIES_MERGE
 from .diff.branch_differ import BranchDiffer
 
 if TYPE_CHECKING:
@@ -261,7 +262,6 @@ async def merge_repositories(self) -> None:
         repos_in_main = {repo.id: repo for repo in repos_in_main_list}
 
         repos_in_branch_list = await NodeManager.query(schema=CoreRepository, db=self.db, branch=self.source_branch)
-        events = []
         for repo in repos_in_branch_list:
             # Check if the repo, exist in main, if not ignore this repo
             if repo.id not in repos_in_main:
@@ -271,16 +271,14 @@ async def merge_repositories(self) -> None:
                 continue
 
             if self.source_branch.sync_with_git or repo.internal_status.value == RepositoryInternalStatus.STAGING.value:
-                events.append(
-                    messages.GitRepositoryMerge(
-                        repository_id=repo.id,
-                        repository_name=repo.name.value,
-                        internal_status=repo.internal_status.value,
-                        source_branch=self.source_branch.name,
-                        destination_branch=registry.default_branch,
-                        default_branch=repo.default_branch.value,
-                    )
+                model = GitRepositoryMerge(
+                    repository_id=repo.id,
+                    repository_name=repo.name.value,
+                    internal_status=repo.internal_status.value,
+                    source_branch=self.source_branch.name,
+                    destination_branch=registry.default_branch,
+                    default_branch=repo.default_branch.value,
+                )
+                await self.service.workflow.submit_workflow(
+                    workflow=GIT_REPOSITORIES_MERGE, parameters={"model": model}
                 )
-
-        for event in events:
-            await self.service.send(message=event)

backend/infrahub/git/models.py

@@ -44,3 +44,14 @@ class GitRepositoryPullReadOnly(BaseModel):
     ref: Optional[str] = Field(None, description="Ref to track on the external repository")
     commit: Optional[str] = Field(None, description="Specific commit to pull")
     infrahub_branch_name: str = Field(..., description="Infrahub branch on which to sync the remote repository")
+
+
+class GitRepositoryMerge(BaseModel):
+    """Merge one branch into another."""
+
+    repository_id: str = Field(..., description="The unique ID of the Repository")
+    repository_name: str = Field(..., description="The name of the repository")
+    internal_status: str = Field(..., description="Administrative status of the repository")
+    source_branch: str = Field(..., description="The source branch")
+    destination_branch: str = Field(..., description="The source branch")
+    default_branch: str = Field(..., description="The default branch in Git")

backend/infrahub/git/tasks.py

@@ -12,7 +12,12 @@
 from ..tasks.artifact import define_artifact
 from ..workflows.catalogue import REQUEST_ARTIFACT_DEFINITION_GENERATE, REQUEST_ARTIFACT_GENERATE
 from ..workflows.utils import add_branch_tag
-from .models import GitRepositoryPullReadOnly, RequestArtifactDefinitionGenerate, RequestArtifactGenerate
+from .models import (
+    GitRepositoryMerge,
+    GitRepositoryPullReadOnly,
+    RequestArtifactDefinitionGenerate,
+    RequestArtifactGenerate,
+)
 from .repository import InfrahubReadOnlyRepository, InfrahubRepository, get_initialized_repo
 
 log = get_logger()
@@ -275,3 +280,39 @@ async def pull_read_only(model: GitRepositoryPullReadOnly) -> None:
 
             await repo.import_objects_from_files(infrahub_branch_name=model.infrahub_branch_name, commit=model.commit)
             await repo.sync_from_remote(commit=model.commit)
+
+
+@flow(name="git-repository-merge")
+async def merge_git_repository(model: GitRepositoryMerge) -> None:
+    service = services.service
+    log.info(
+        "Merging repository branch",
+        repository_name=model.repository_name,
+        repository_id=model.repository_id,
+        source_branch=model.source_branch,
+        destination_branch=model.destination_branch,
+    )
+
+    repo = await InfrahubRepository.init(
+        id=model.repository_id,
+        name=model.repository_name,
+        client=service.client,
+        default_branch_name=model.default_branch,
+    )
+
+    if model.internal_status == RepositoryInternalStatus.STAGING.value:
+        repo_source = await service.client.get(
+            kind=InfrahubKind.GENERICREPOSITORY, id=model.repository_id, branch=model.source_branch
+        )
+        repo_main = await service.client.get(kind=InfrahubKind.GENERICREPOSITORY, id=model.repository_id)
+        repo_main.internal_status.value = RepositoryInternalStatus.ACTIVE.value
+        repo_main.sync_status.value = repo_source.sync_status.value
+
+        commit = repo.get_commit_value(branch_name=repo.default_branch, remote=False)
+        repo_main.commit.value = commit
+
+        await repo_main.save()
+
+    else:
+        async with lock.registry.get(name=model.repository_name, namespace="repository"):
+            await repo.merge(source_branch=model.source_branch, dest_branch=model.destination_branch)

backend/infrahub/message_bus/messages/__init__.py

@@ -18,7 +18,6 @@
 from .git_repository_add import GitRepositoryAdd
 from .git_repository_connectivity import GitRepositoryConnectivity
 from .git_repository_importobjects import GitRepositoryImportObjects
-from .git_repository_merge import GitRepositoryMerge
 from .git_repository_read_only_add import GitRepositoryAddReadOnly
 from .proposed_change.request_proposedchange_dataintegrity import RequestProposedChangeDataIntegrity
 from .proposed_change.request_proposedchange_refreshartifacts import RequestProposedChangeRefreshArtifacts
@@ -59,7 +58,6 @@
     "git.file.get": GitFileGet,
     "git.repository.add": GitRepositoryAdd,
     "git.repository.connectivity": GitRepositoryConnectivity,
-    "git.repository.merge": GitRepositoryMerge,
     "git.repository.add_read_only": GitRepositoryAddReadOnly,
     "git.repository.import_objects": GitRepositoryImportObjects,
     "schema.migration.path": SchemaMigrationPath,

backend/infrahub/message_bus/messages/git_repository_merge.py

@@ -38,7 +38,6 @@
     "git.repository.add_read_only": git.repository.add_read_only,
     "git.repository.connectivity": git.repository.connectivity,
     "git.repository.import_objects": git.repository.import_objects,
-    "git.repository.merge": git.repository.merge,
     "refresh.registry.branches": refresh.registry.branches,
     "refresh.registry.rebased_branch": refresh.registry.rebased_branch,
     "refresh.webhook.configuration": refresh.webhook.configuration,

backend/infrahub/message_bus/operations/__init__.py

@@ -1,7 +1,7 @@
 from prefect import flow
 
 from infrahub import lock
-from infrahub.core.constants import InfrahubKind, RepositoryInternalStatus
+from infrahub.core.constants import RepositoryInternalStatus
 from infrahub.exceptions import RepositoryError
 from infrahub.git.repository import InfrahubReadOnlyRepository, InfrahubRepository, get_initialized_repo
 from infrahub.log import get_logger
@@ -102,38 +102,3 @@ async def import_objects(message: messages.GitRepositoryImportObjects, service:
         )
         repo.task_report = git_report
         await repo.import_objects_from_files(infrahub_branch_name=message.infrahub_branch_name, commit=message.commit)
-
-
-@flow(name="git-repository-merge")
-async def merge(message: messages.GitRepositoryMerge, service: InfrahubServices) -> None:
-    log.info(
-        "Merging repository branch",
-        repository_name=message.repository_name,
-        repository_id=message.repository_id,
-        source_branch=message.source_branch,
-        destination_branch=message.destination_branch,
-    )
-
-    repo = await InfrahubRepository.init(
-        id=message.repository_id,
-        name=message.repository_name,
-        client=service.client,
-        default_branch_name=message.default_branch,
-    )
-
-    if message.internal_status == RepositoryInternalStatus.STAGING.value:
-        repo_source = await service.client.get(
-            kind=InfrahubKind.GENERICREPOSITORY, id=message.repository_id, branch=message.source_branch
-        )
-        repo_main = await service.client.get(kind=InfrahubKind.GENERICREPOSITORY, id=message.repository_id)
-        repo_main.internal_status.value = RepositoryInternalStatus.ACTIVE.value
-        repo_main.sync_status.value = repo_source.sync_status.value
-
-        commit = repo.get_commit_value(branch_name=repo.default_branch, remote=False)
-        repo_main.commit.value = commit
-
-        await repo_main.save()
-
-    else:
-        async with lock.registry.get(name=message.repository_name, namespace="repository"):
-            await repo.merge(source_branch=message.source_branch, dest_branch=message.destination_branch)

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

@@ -135,6 +135,13 @@
     type=WorkflowType.INTERNAL,
     module="infrahub.git.tasks",
     function="pull_read_only",
+)
+
+GIT_REPOSITORIES_MERGE = WorkflowDefinition(
+    name="git-repository-merge",
+    type=WorkflowType.INTERNAL,
+    module="infrahub.git.tasks",
+    function="merge_git_repository",
     branch_support=BranchSupportType.AWARE,
     tags=[WorkflowTag.DATABASE_CHANGE],
 )
@@ -194,6 +201,7 @@
     REQUEST_DIFF_UPDATE,
     REQUEST_DIFF_REFRESH,
     GIT_REPOSITORIES_PULL_READ_ONLY,
+    GIT_REPOSITORIES_MERGE,
     TRIGGER_GENERATOR_DEFINITION_RUN,
     BRANCH_CANCEL_PROPOSED_CHANGES,
 ]

backend/infrahub/workflows/catalogue.py

@@ -10,14 +10,15 @@
 from infrahub.core.constants import InfrahubKind, RepositoryInternalStatus
 from infrahub.exceptions import RepositoryError
 from infrahub.git import InfrahubRepository
-from infrahub.git.models import GitRepositoryPullReadOnly
+from infrahub.git.models import GitRepositoryMerge, GitRepositoryPullReadOnly
 from infrahub.git.repository import InfrahubReadOnlyRepository
 from infrahub.git.tasks import pull_read_only
 from infrahub.lock import InfrahubLockRegistry
 from infrahub.message_bus import Meta, messages
 from infrahub.message_bus.operations import git
 from infrahub.services import InfrahubServices, services
 from infrahub.services.adapters.workflow.local import WorkflowLocalExecution
+from infrahub.workflows.catalogue import GIT_REPOSITORIES_MERGE
 from tests.helpers.test_client import dummy_async_request
 
 # pylint: disable=redefined-outer-name
@@ -116,7 +117,7 @@ async def test_git_rpc_merge(
 
     commit_main_before = repo.get_commit_value(branch_name="main")
 
-    message = messages.GitRepositoryMerge(
+    model = GitRepositoryMerge(
         repository_id=str(UUIDT()),
         repository_name=repo.name,
         source_branch="branch01",
@@ -127,9 +128,17 @@ async def test_git_rpc_merge(
 
     client_config = Config(requester=dummy_async_request)
     bus_simulator = helper.get_message_bus_simulator()
-    service = InfrahubServices(client=InfrahubClient(config=client_config), message_bus=bus_simulator)
+    service = InfrahubServices(
+        client=InfrahubClient(config=client_config), message_bus=bus_simulator, workflow=WorkflowLocalExecution()
+    )
     bus_simulator.service = service
-    await service.send(message=message)
+
+    original_services = services.service
+    services.service = service
+    await service.workflow.submit_workflow(workflow=GIT_REPOSITORIES_MERGE, parameters={"model": model})
+
+    # Restore original services to not impact other tests. This global variable might/should be redesigned at some point.
+    service.service = original_services
 
     commit_main_after = repo.get_commit_value(branch_name="main")
 

backend/tests/unit/git/test_git_rpc.py

@@ -6,14 +6,15 @@
 from infrahub.exceptions import RepositoryFileNotFoundError, TransformError
 from infrahub.git import InfrahubRepository
 from infrahub.services import InfrahubServices, services
+from infrahub.services.adapters.workflow.local import WorkflowLocalExecution
 from infrahub.transformations.models import TransformJinjaTemplateData, TransformPythonData
 from infrahub.transformations.tasks import transform_python, transform_render_jinja2_template
 
 
 @pytest.fixture
 def init_service():
     original = services.service
-    service = InfrahubServices(client=InfrahubClient())
+    service = InfrahubServices(client=InfrahubClient(), workflow=WorkflowLocalExecution())
     services.service = service
     yield service
     services.service = original