Merge pull request #140 from nautobot/patch-fix_138

Add Check for Relationship - Fix 138

development/docker-compose.base.yml

@@ -19,8 +19,9 @@ services:
     depends_on:
       - "postgres"
       - "redis"
-    <<: *nautobot-build
-    <<: *nautobot-base
+    <<:
+      - *nautobot-build
+      - *nautobot-base
   worker:
     entrypoint:
       - "sh"

development/docker-compose.requirements.yml

@@ -2,7 +2,7 @@
 version: "3.8"
 services:
   postgres:
-    image: "postgres:14-alpine"
+    image: "postgres:13-alpine"
     env_file:
       - "development.env"
       - "creds.env"

nautobot_ssot_infoblox/tests/test_client.py

@@ -406,7 +406,6 @@ def test_get_authoritative_zone_fail(self):
         self.assertEqual(err.value.response.status_code, 404)
 
     def test_create_ptr_record_success(self):
-
         mock_uri = "record:ptr"
         mock_fqdn = "test-device.test-site"
         mock_ip_address = "10.1.1.1"
@@ -434,7 +433,6 @@ def test_create_ptr_record_failure(self):
         self.assertEqual(err.value.response.status_code, 404)
 
     def test_create_a_record_success(self):
-
         mock_uri = "record:a"
         mock_fqdn = "test-device.test-site"
         mock_ip_address = "10.1.1.1"
@@ -461,7 +459,6 @@ def test_create_a_record_failure(self):
         self.assertEqual(err.value.response.status_code, 404)
 
     def test_create_host_record_success(self):
-
         mock_uri = "record:host"
         mock_fqdn = "test-device.test-site"
         mock_ip_address = "10.1.1.1"

nautobot_ssot_infoblox/tests/test_utils_nautobot.py

@@ -0,0 +1,57 @@
+"""Test utility methods for Nautobot."""
+from django.contrib.contenttypes.models import ContentType
+from nautobot.extras.models import Relationship, RelationshipAssociation, Status
+from nautobot.ipam.models import Prefix, VLAN, VLANGroup
+from nautobot.utilities.testing import TransactionTestCase
+from nautobot_ssot_infoblox.utils.nautobot import build_vlan_map_from_relations, get_prefix_vlans
+
+
+class TestNautobotUtils(TransactionTestCase):
+    """Test Nautobot Utility methods."""
+
+    def setUp(self):
+        """Configure common objects for tests."""
+        super().setUp()
+        self.status_active = Status.objects.get(name="Active")
+        self.test_pf = Prefix.objects.get_or_create(prefix="192.168.1.0/24")[0]
+        self.vlan_group = VLANGroup.objects.create(name="Test")
+        self.test_vlan1 = VLAN.objects.create(name="Test1", vid=1, status=self.status_active, group=self.vlan_group)
+        self.test_vlan1.validated_save()
+        self.test_vlan2 = VLAN.objects.create(name="Test2", vid=2, status=self.status_active, group=self.vlan_group)
+        self.test_vlan2.validated_save()
+
+    def test_build_vlan_map_from_relations(self):
+        """Validate functionality of the build_vlan_map_from_relations() function."""
+        test_list = [self.test_vlan1, self.test_vlan2]
+        actual = build_vlan_map_from_relations(vlans=test_list)
+        expected = {1: {"vid": 1, "name": "Test1", "group": "Test"}, 2: {"vid": 2, "name": "Test2", "group": "Test"}}
+        self.assertEqual(actual, expected)
+
+    def test_get_prefix_vlans_success(self):
+        """Validate functionality of the get_prefix_vlans() function success."""
+        pf_vlan_rel = Relationship.objects.get(slug="prefix_to_vlan")
+        rel_assoc1 = RelationshipAssociation.objects.create(
+            relationship_id=pf_vlan_rel.id,
+            source_type=ContentType.objects.get_for_model(Prefix),
+            source_id=self.test_pf.id,
+            destination_type=ContentType.objects.get_for_model(VLAN),
+            destination_id=self.test_vlan1.id,
+        )
+        rel_assoc1.validated_save()
+        rel_assoc2 = RelationshipAssociation.objects.create(
+            relationship_id=pf_vlan_rel.id,
+            source_type=ContentType.objects.get_for_model(Prefix),
+            source_id=self.test_pf.id,
+            destination_type=ContentType.objects.get_for_model(VLAN),
+            destination_id=self.test_vlan2.id,
+        )
+        rel_assoc2.validated_save()
+        expected = [self.test_vlan1, self.test_vlan2]
+        actual = get_prefix_vlans(self.test_pf)
+        self.assertEqual(actual, expected)
+
+    def test_get_prefix_vlans_failure(self):
+        """Validate functionality of the get_prefix_vlans() function failure where Prefix has no RelationshipAssocations to VLANs."""
+        expected = []
+        actual = get_prefix_vlans(self.test_pf)
+        self.assertEqual(actual, expected)

nautobot_ssot_infoblox/utils/nautobot.py

@@ -26,6 +26,10 @@ def get_prefix_vlans(prefix: Prefix) -> list:
     Returns:
         list: List of VLAN objects with RelationshipAssociation to passed Prefix.
     """
+    vlan_list = []
     pf_relations = prefix.get_relationships()
     pf_vlan_relationship = Relationship.objects.get(name="Prefix -> VLAN")
-    return [x.destination for x in pf_relations["source"][pf_vlan_relationship]]
+    if pf_vlan_relationship in pf_relations["source"]:
+        vlan_list = [x.destination for x in pf_relations["source"][pf_vlan_relationship]]
+        vlan_list.sort(key=lambda x: x.vid)
+    return vlan_list