Misc cleanup

netbox/netbox/staging.py

@@ -65,17 +65,25 @@ class checkout:
         # Process queued changes
         self.process_queue()
 
+    #
+    # Queuing
+    #
+
+    @staticmethod
+    def get_key_for_instance(instance):
+        return ContentType.objects.get_for_model(instance), instance.pk
+
     def process_queue(self):
         """
         Create Change instances for all actions stored in the queue.
         """
-        changes = []
         if not self.queue:
             logger.debug(f"No queued changes; aborting")
             return
         logger.debug(f"Processing {len(self.queue)} queued changes")
 
         # Iterate through the in-memory queue, creating Change instances
+        changes = []
         for key, change in self.queue.items():
             logger.debug(f'  {key}: {change}')
             object_type, pk = key
@@ -95,10 +103,6 @@ class checkout:
         # Save all Change instances to the database
         Change.objects.bulk_create(changes)
 
-    @staticmethod
-    def get_key_for_instance(instance):
-        return ContentType.objects.get_for_model(instance), instance.pk
-
     #
     # Signal handlers
     #
@@ -112,11 +116,11 @@ class checkout:
 
         if created:
             # Creating a new object
-            logger.debug(f"[{self.branch}] Staging creation of {object_type} {instance}")
+            logger.debug(f"[{self.branch}] Staging creation of {object_type} {instance} (PK: {instance.pk})")
             self.queue[key] = (ChangeActionChoices.ACTION_CREATE, instance)
         elif key in self.queue:
             # Object has already been created/updated at least once
-            logger.debug(f"[{self.branch}] Updating staged value for {object_type} {instance}")
+            logger.debug(f"[{self.branch}] Updating staged value for {object_type} {instance} (PK: {instance.pk})")
             self.queue[key] = (self.queue[key][0], instance)
         else:
             # Modifying an existing object
@@ -128,11 +132,13 @@ class checkout:
         Hooks to the pre_delete signal when a branch is active to queue delete actions.
         """
         key = self.get_key_for_instance(instance)
-        if key in self.queue and self.queue[key][0] == 'create':
+        object_type = instance._meta.verbose_name
+
+        if key in self.queue and self.queue[key][0] == ChangeActionChoices.ACTION_CREATE:
             # Cancel the creation of a new object
-            logger.debug(f"[{self.branch}] Removing staged deletion of {instance} (PK: {instance.pk})")
+            logger.debug(f"[{self.branch}] Removing staged creation of {object_type} {instance} (PK: {instance.pk})")
             del self.queue[key]
         else:
             # Delete an existing object
-            logger.debug(f"[{self.branch}] Staging deletion of {instance} (PK: {instance.pk})")
+            logger.debug(f"[{self.branch}] Staging deletion of {object_type} {instance} (PK: {instance.pk})")
             self.queue[key] = (ChangeActionChoices.ACTION_DELETE, instance)

netbox/netbox/tests/test_staging.py

@@ -1,6 +1,7 @@
 from django.test import TransactionTestCase
 
 from circuits.models import Provider, Circuit, CircuitType
+from extras.choices import ChangeActionChoices
 from extras.models import Branch, Change, Tag
 from netbox.staging import checkout
 from utilities.testing import create_tags
@@ -9,7 +10,7 @@ from utilities.testing import create_tags
 class StagingTestCase(TransactionTestCase):
 
     def setUp(self):
-        create_tags('Alpha', 'Bravo', 'Charlie', 'Delta', 'Echo', 'Foxtrot')
+        create_tags('Alpha', 'Bravo', 'Charlie')
 
         providers = (
             Provider(name='Provider A', slug='provider-a'),
@@ -34,7 +35,7 @@ class StagingTestCase(TransactionTestCase):
 
     def test_object_creation(self):
         branch = Branch.objects.create(name='Branch 1')
-        tags = Tag.objects.all()[:3]
+        tags = Tag.objects.all()
 
         with checkout(branch):
             provider = Provider.objects.create(name='Provider D', slug='provider-d')
@@ -68,8 +69,7 @@ class StagingTestCase(TransactionTestCase):
 
     def test_object_modification(self):
         branch = Branch.objects.create(name='Branch 1')
-        tags = Tag.objects.all()[:3]
+        tags = Tag.objects.all()
-        self.assertEqual(len(tags), 3)
 
         with checkout(branch):
             provider = Provider.objects.get(name='Provider A')
@@ -174,7 +174,7 @@ class StagingTestCase(TransactionTestCase):
         # Check that a create Change was recorded
         self.assertEqual(Change.objects.count(), 1)
         change = Change.objects.first()
-        self.assertEqual(change.action, 'create')
+        self.assertEqual(change.action, ChangeActionChoices.ACTION_CREATE)
         self.assertEqual(change.data['name'], provider.name)
 
         with checkout(branch):
@@ -187,7 +187,7 @@ class StagingTestCase(TransactionTestCase):
         # Check that a second Change object has been created for the object
         self.assertEqual(Change.objects.count(), 2)
         change = Change.objects.last()
-        self.assertEqual(change.action, 'update')
+        self.assertEqual(change.action, ChangeActionChoices.ACTION_UPDATE)
         self.assertEqual(change.data['name'], provider.name)
         self.assertEqual(change.data['comments'], provider.comments)
 
@@ -200,5 +200,5 @@ class StagingTestCase(TransactionTestCase):
         # Check that a third Change has recorded the object's deletion
         self.assertEqual(Change.objects.count(), 3)
         change = Change.objects.last()
-        self.assertEqual(change.action, 'delete')
+        self.assertEqual(change.action, ChangeActionChoices.ACTION_DELETE)
         self.assertIsNone(change.data)