|
|
@@ -2,13 +2,14 @@ from django.test import override_settings
|
|
|
from django.urls import reverse
|
|
|
|
|
|
from dcim.models import *
|
|
|
+from utilities.counters import connect_counters
|
|
|
from utilities.testing.base import TestCase
|
|
|
from utilities.testing.utils import create_test_device
|
|
|
|
|
|
|
|
|
class CountersTest(TestCase):
|
|
|
"""
|
|
|
- Validate the operation of dict_to_filter_params().
|
|
|
+ Validate the operation of the CounterCacheField (tracking counters).
|
|
|
"""
|
|
|
@classmethod
|
|
|
def setUpTestData(cls):
|
|
|
@@ -24,7 +25,7 @@ class CountersTest(TestCase):
|
|
|
|
|
|
def test_interface_count_creation(self):
|
|
|
"""
|
|
|
- When a tracked object (Interface) is added the tracking counter should be updated.
|
|
|
+ When a tracked object (Interface) is added, the tracking counter should be updated.
|
|
|
"""
|
|
|
device1, device2 = Device.objects.all()
|
|
|
self.assertEqual(device1.interface_count, 2)
|
|
|
@@ -51,7 +52,7 @@ class CountersTest(TestCase):
|
|
|
|
|
|
def test_interface_count_deletion(self):
|
|
|
"""
|
|
|
- When a tracked object (Interface) is deleted the tracking counter should be updated.
|
|
|
+ When a tracked object (Interface) is deleted, the tracking counter should be updated.
|
|
|
"""
|
|
|
device1, device2 = Device.objects.all()
|
|
|
self.assertEqual(device1.interface_count, 2)
|
|
|
@@ -66,7 +67,7 @@ class CountersTest(TestCase):
|
|
|
|
|
|
def test_interface_count_move(self):
|
|
|
"""
|
|
|
- When a tracked object (Interface) is moved the tracking counter should be updated.
|
|
|
+ When a tracked object (Interface) is moved, the tracking counter should be updated.
|
|
|
"""
|
|
|
device1, device2 = Device.objects.all()
|
|
|
self.assertEqual(device1.interface_count, 2)
|
|
|
@@ -102,3 +103,35 @@ class CountersTest(TestCase):
|
|
|
self.client.post(reverse("dcim:inventoryitem_bulk_delete"), data)
|
|
|
device1.refresh_from_db()
|
|
|
self.assertEqual(device1.inventory_item_count, 0)
|
|
|
+
|
|
|
+ def test_signal_connections_are_idempotent_per_sender(self):
|
|
|
+ """
|
|
|
+ Calling connect_counters() again must not register duplicate receivers.
|
|
|
+ Creating a device after repeated "connect_counters" should still yield +1.
|
|
|
+ """
|
|
|
+ connect_counters(DeviceType, VirtualChassis)
|
|
|
+ vc, _ = VirtualChassis.objects.get_or_create(name='Virtual Chassis 1')
|
|
|
+ device1, device2 = Device.objects.all()
|
|
|
+ self.assertEqual(device1.device_type.device_count, 2)
|
|
|
+ self.assertEqual(vc.member_count, 0)
|
|
|
+
|
|
|
+ # Call again (should be a no-op for sender registrations)
|
|
|
+ connect_counters(DeviceType, VirtualChassis)
|
|
|
+
|
|
|
+ # Create one new device
|
|
|
+ device3 = create_test_device('Device 3')
|
|
|
+ device3.virtual_chassis = vc
|
|
|
+ device3.save()
|
|
|
+
|
|
|
+ # Ensure counter incremented correctly
|
|
|
+ device1.refresh_from_db()
|
|
|
+ vc.refresh_from_db()
|
|
|
+ self.assertEqual(device1.device_type.device_count, 3, 'device_count should increment exactly once')
|
|
|
+ self.assertEqual(vc.member_count, 1, 'member_count should increment exactly once')
|
|
|
+
|
|
|
+ # Clean up and ensure counter decremented correctly
|
|
|
+ device3.delete()
|
|
|
+ device1.refresh_from_db()
|
|
|
+ vc.refresh_from_db()
|
|
|
+ self.assertEqual(device1.device_type.device_count, 2, 'device_count should decrement exactly once')
|
|
|
+ self.assertEqual(vc.member_count, 0, 'member_count should decrement exactly once')
|