Add tests for migration runner

Author: gmazoyer

backend/infrahub/core/branch/tasks.py

@@ -49,7 +49,6 @@
 from infrahub.workflows.catalogue import (
     BRANCH_CANCEL_PROPOSED_CHANGES,
     BRANCH_MERGE_POST_PROCESS,
-    BRANCH_MIGRATE,
     DIFF_REFRESH_ALL,
     DIFF_UPDATE,
     GIT_REPOSITORIES_CREATE_BRANCH,

backend/infrahub/core/migrations/runner.py

@@ -2,6 +2,8 @@
 
 from typing import TYPE_CHECKING, Sequence
 
+from infrahub.core import registry
+from infrahub.core.constants import GLOBAL_BRANCH_NAME
 from infrahub.core.migrations.graph import MIGRATIONS
 
 from .exceptions import MigrationFailureError
@@ -14,6 +16,9 @@
 
 class MigrationRunner:
     def __init__(self, branch: Branch) -> None:
+        if branch.name in (registry.default_branch, GLOBAL_BRANCH_NAME):
+            raise ValueError("MigrationRunner cannot be used to apply migration on default branches")
+
         self.branch = branch
         self.applicable_migrations = self._get_applicable_migrations()
 

backend/infrahub/graphql/types/branch.py

@@ -2,7 +2,7 @@
 
 from typing import TYPE_CHECKING, Any
 
-from graphene import Boolean, Field, String
+from graphene import Boolean, Field, Int, String
 
 from infrahub.core.branch import Branch
 from infrahub.core.constants import GLOBAL_BRANCH_NAME
@@ -21,6 +21,7 @@ class BranchType(InfrahubObjectType):
     origin_branch = String(required=False)
     branched_from = String(required=False)
     status = InfrahubBranchStatus(required=True)
+    graph_version = Int(required=False)
     created_at = String(required=False)
     sync_with_git = Boolean(required=False)
     is_default = Boolean(required=False)

backend/tests/unit/core/migrations/test_runner.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from infrahub.core.branch.models import Branch
+from infrahub.core.constants import GLOBAL_BRANCH_NAME
+from infrahub.core.initialization import create_branch
+from infrahub.core.migrations.runner import MigrationRunner
+
+if TYPE_CHECKING:
+    from infrahub.database import InfrahubDatabase
+
+
+def test_instantiation(default_branch: Branch) -> None:
+    assert MigrationRunner(branch=Branch(name="foo"))
+
+    with pytest.raises(ValueError):
+        MigrationRunner(branch=default_branch)
+
+    with pytest.raises(ValueError):
+        MigrationRunner(branch=Branch(name=GLOBAL_BRANCH_NAME))
+
+
+async def test_has_migrations(default_branch: Branch, db: InfrahubDatabase) -> None:
+    branch = await create_branch(db=db, branch_name="foo")
+
+    runner = MigrationRunner(branch=branch)
+    assert not runner.has_migrations()
+
+    branch.graph_version = None
+    runner = MigrationRunner(branch=branch)
+    await branch.save(db=db)
+    assert runner.has_migrations()
+
+    branch.graph_version = 40
+    runner = MigrationRunner(branch=branch)
+    await branch.save(db=db)
+    assert runner.has_migrations()
+
+
+async def test_applicable_migrations(default_branch: Branch, db: InfrahubDatabase) -> None:
+    branch = await create_branch(db=db, branch_name="foo")
+
+    runner = MigrationRunner(branch=branch)
+    assert not runner.applicable_migrations
+
+    branch.graph_version = None
+    runner = MigrationRunner(branch=branch)
+    await branch.save(db=db)
+    assert runner.applicable_migrations
+    assert [m.name for m in runner.applicable_migrations] == ["043_backfill_hfid_display_label_in_db"]
+
+    branch.graph_version = 40
+    runner = MigrationRunner(branch=branch)
+    await branch.save(db=db)
+    assert runner.applicable_migrations
+    assert [m.name for m in runner.applicable_migrations] == ["043_backfill_hfid_display_label_in_db"]