Fixes #260: Ignore duplicate SQL indexes when provisioning a branch

netbox_branching/constants.py

@@ -35,3 +35,12 @@
     'netbox_branching.*',
     'netbox_changes.*',
 )
+
+# Indexes to ignore as they are removed in a NetBox v4.3 migration, but might be present
+# in earlier NetBox releases.
+# TODO: Remove in v0.6.0
+SKIP_INDEXES = (
+    'dcim_cabletermination_termination_type_id_termination_id_idx',     # Removed in dcim.0207_remove_redundant_indexes
+    'vpn_l2vpntermination_assigned_object_type_id_assigned_objec_idx',  # Removed in vpn.0009_remove_redundant_indexes
+    'vpn_tunneltermination_termination_type_id_termination_id_idx',     # Removed in vpn.0009_remove_redundant_indexes
+)

netbox_branching/models/branches.py

@@ -24,7 +24,7 @@
 from netbox.models.features import JobsMixin
 from netbox.plugins import get_plugin_config
 from netbox_branching.choices import BranchEventTypeChoices, BranchStatusChoices
-from netbox_branching.constants import MAIN_SCHEMA
+from netbox_branching.constants import MAIN_SCHEMA, SKIP_INDEXES
 from netbox_branching.contextvars import active_branch
 from netbox_branching.signals import *
 from netbox_branching.utilities import (
@@ -571,6 +571,11 @@ def provision(self, user):
                     f"SELECT tablename, indexname, indexdef FROM pg_indexes WHERE schemaname = '{schema}'"
                 )
                 for index in get_sql_results(cursor):
+                    # Skip duplicate indexes
+                    # TODO: Remove in v0.6.0
+                    if index.indexname in SKIP_INDEXES:
+                        continue
+
                     # Find the matching index in main based on its table & definition
                     definition = index.indexdef.split(' USING ', maxsplit=1)[1]
                     cursor.execute(
@@ -579,10 +584,9 @@ def provision(self, user):
                     )
                     if result := cursor.fetchone():
                         # Rename the branch schema index (if needed)
-                        original_name = index.indexname
                         new_name = result[0]
-                        if new_name != original_name:
-                            sql = f"ALTER INDEX {schema}.{original_name} RENAME TO {new_name}"
+                        if new_name != index.indexname:
+                            sql = f"ALTER INDEX {schema}.{index.indexname} RENAME TO {new_name}"
                             try:
                                 cursor.execute(sql)
                                 logger.debug(sql)

netbox_branching/tests/test_branches.py

@@ -7,7 +7,7 @@
 from django.utils import timezone
 
 from netbox_branching.choices import BranchStatusChoices
-from netbox_branching.constants import MAIN_SCHEMA
+from netbox_branching.constants import MAIN_SCHEMA, SKIP_INDEXES
 from netbox_branching.models import Branch
 from netbox_branching.utilities import get_tables_to_replicate
 from .utils import fetchall, fetchone
@@ -44,7 +44,7 @@ def test_create_branch(self):
 
             # Check that all indexes were renamed to match the main schema
             cursor.execute(
-                "SELECT idx_a.schemaname AS schema_a, idx_a.tablename AS table_a, idx_a.indexname AS index_in_a "
+                "SELECT idx_a.schemaname, idx_a.tablename, idx_a.indexname "
                 "FROM pg_indexes idx_a "
                 "WHERE idx_a.schemaname=%s "
                 "AND NOT EXISTS ("
@@ -53,7 +53,12 @@ def test_create_branch(self):
                 ") ORDER BY idx_a.indexname",
                 [branch.schema_name, MAIN_SCHEMA]
             )
-            self.assertListEqual(fetchall(cursor), [], "Found indexes with unique names in branch schema.")
+            # Omit skipped indexes
+            # TODO: Remove in v0.6.0
+            found_indexes = [
+                idx for idx in fetchall(cursor) if idx.indexname not in SKIP_INDEXES
+            ]
+            self.assertListEqual(found_indexes, [], "Found indexes with unique names in branch schema.")
 
             # Check that object counts match the main schema for each table
             for table_name in tables_to_replicate: