Closes #189: Introduce a mechanism to automatically register pre-action branch validators

netbox_branching/__init__.py

@@ -41,6 +41,7 @@ class AppConfig(PluginConfig):
     def ready(self):
         super().ready()
         from . import constants, events, search, signal_receivers
+        from .models import Branch
         from .utilities import DynamicSchemaDict
 
         # Validate required settings
@@ -56,13 +57,14 @@ def ready(self):
         # Register models which support branching
         register_models()
 
-        # Validate branch action validators
+        # Validate & register configured branch action validators
         for action in BRANCH_ACTIONS:
             for validator_path in get_plugin_config('netbox_branching', f'{action}_validators'):
                 try:
-                    import_string(validator_path)
+                    func = import_string(validator_path)
                 except ImportError:
                     raise ImproperlyConfigured(f"Branch {action} validator not found: {validator_path}")
+                Branch.register_preaction_check(func, action)
 
 
 config = AppConfig

netbox_branching/models/branches.py

@@ -28,7 +28,8 @@
 from netbox_branching.contextvars import active_branch
 from netbox_branching.signals import *
 from netbox_branching.utilities import (
-    ChangeSummary, activate_branch, get_branchable_object_types, get_tables_to_replicate, record_applied_change,
+    BranchActionIndicator, ChangeSummary, activate_branch, get_branchable_object_types, get_tables_to_replicate,
+    record_applied_change,
 )
 from utilities.exceptions import AbortRequest, AbortTransaction
 from .changes import ObjectChange
@@ -83,6 +84,13 @@ class Branch(JobsMixin, PrimaryModel):
         related_name='+'
     )
 
+    _preaction_validators = {
+        'sync': set(),
+        'merge': set(),
+        'revert': set(),
+        'archive': set(),
+    }
+
     class Meta:
         ordering = ('name',)
         verbose_name = _('branch')
@@ -190,6 +198,15 @@ def _generate_schema_id(length=8):
         chars = [*string.ascii_lowercase, *string.digits]
         return ''.join(random.choices(chars, k=length))
 
+    @classmethod
+    def register_preaction_check(cls, func, action):
+        """
+        Register a validator to run before a specific branch action (i.e. sync or merge).
+        """
+        if action not in BRANCH_ACTIONS:
+            raise ValueError(f"Invalid branch action: {action}")
+        cls._preaction_validators[action].add(func)
+
     def get_changes(self):
         """
         Return a queryset of all ObjectChange records created within the Branch.
@@ -265,33 +282,39 @@ def _can_do_action(self, action):
         """
         if action not in BRANCH_ACTIONS:
             raise Exception(f"Unrecognized branch action: {action}")
-        for validator_path in get_plugin_config('netbox_branching', f'{action}_validators'):
-            if not import_string(validator_path)(self):
-                return False
-        return True
 
-    @property
+        # Run any pre-action validators
+        for func in self._preaction_validators[action]:
+            if not (indicator := func(self)):
+                # Backward compatibility for pre-v0.6.0 validators
+                if type(indicator) is not BranchActionIndicator:
+                    return BranchActionIndicator(False, _(f"Validation failed for {action}: {func}"))
+                return indicator
+
+        return BranchActionIndicator(True)
+
+    @cached_property
     def can_sync(self):
         """
         Indicates whether the branch can be synced.
         """
         return self._can_do_action('sync')
 
-    @property
+    @cached_property
     def can_merge(self):
         """
         Indicates whether the branch can be merged.
         """
         return self._can_do_action('merge')
 
-    @property
+    @cached_property
     def can_revert(self):
         """
         Indicates whether the branch can be reverted.
         """
         return self._can_do_action('revert')
 
-    @property
+    @cached_property
     def can_archive(self):
         """
         Indicates whether the branch can be archived.

netbox_branching/templates/netbox_branching/branch_action.html

@@ -20,9 +20,8 @@
     {% if not action_permitted %}
       <div class="alert alert-warning">
         <i class="mdi mdi-alert-circle"></i>
-        {% blocktrans %}
-          This action is disallowed per policy, however dry runs are permitted.
-        {% endblocktrans %}
+        {{ action_permitted.message }}
+        {% blocktrans %}Only dry runs are permitted.{% endblocktrans %}
       </div>
     {% endif %}
     {% if conflicts_table.rows %}

netbox_branching/utilities.py

@@ -17,6 +17,7 @@
 from .contextvars import active_branch
 
 __all__ = (
+    'BranchActionIndicator',
     'ChangeSummary',
     'DynamicSchemaDict',
     'ListHandler',
@@ -261,3 +262,15 @@ def ActiveBranchContextManager(request):
     if branch := get_active_branch(request):
         return activate_branch(branch)
     return nullcontext()
+
+
+@dataclass
+class BranchActionIndicator:
+    """
+    An indication of whether a particular branch action is permitted. If not, an explanatory message must be provided.
+    """
+    permitted: bool
+    message: str = ''
+
+    def __bool__(self):
+        return self.permitted