Skip to content

Commit

Permalink
feat(models): Restrict cross ACI Tenant assignment
Browse files Browse the repository at this point in the history
Enhances ACIBridgeDomain and ACIVRF assignment validations and provides
clearer error messages. Validate assigned ACIBridgeDomain and ACIVRF
must belong to same ACITenant or ACITenant 'common' on the model level.

Tests are added to verify new assignment rules.
  • Loading branch information
pheus committed Jul 16, 2024
1 parent d63e1f8 commit 88a9de2
Show file tree
Hide file tree
Showing 5 changed files with 113 additions and 5 deletions.
6 changes: 3 additions & 3 deletions netbox_aci_plugin/forms/tenant_app_profiles.py
Original file line number Diff line number Diff line change
Expand Up @@ -388,9 +388,9 @@ def clean(self):
self.add_error(
"aci_bridge_domain",
_(
"A Bridge Domain can only be assigned from the same "
"ACI Tenant as the Endpoint Group or ACI Tenant"
"'common'."
"A Bridge Domain can only be assigned belonging to "
"same ACI Tenant as the Application Profile or to the "
"special ACI Tenant 'common'."
),
)

Expand Down
5 changes: 3 additions & 2 deletions netbox_aci_plugin/forms/tenant_networks.py
Original file line number Diff line number Diff line change
Expand Up @@ -752,8 +752,9 @@ def clean(self):
self.add_error(
"aci_vrf",
_(
"A VRF can only be assigned from the same ACI Tenant "
"as the Bridge Domain or ACI Tenant 'common'."
"A VRF can only be assigned belonging to the same ACI "
"Tenant as the Bridge Domain or to the special ACI "
"Tenant 'common'."
),
)

Expand Down
17 changes: 17 additions & 0 deletions netbox_aci_plugin/models/tenant_app_profiles.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
#
# SPDX-License-Identifier: GPL-3.0-or-later

from django.core.exceptions import ValidationError
from django.core.validators import MaxLengthValidator
from django.db import models
from django.urls import reverse
Expand Down Expand Up @@ -238,6 +239,22 @@ def __str__(self) -> str:
"""Return string representation of the instance."""
return self.name

def save(self, *args, **kwargs):
"""Saves the current instance to the database."""
# Ensure the assigned ACIBrideDomain belongs to either the same
# ACITenant as the ACIAppProfile or to the special ACITenant 'common'.
if (
self.aci_bridge_domain.aci_tenant
!= self.aci_app_profile.aci_tenant
and self.aci_bridge_domain.aci_tenant.name != "common"
):
raise ValidationError(
"Assigned ACIBridgeDomain have to belong to the same ACITenant"
"as the ACIAppProfile or to the special ACITenant 'common'."
)

super().save(*args, **kwargs)

@property
def aci_tenant(self) -> ACITenant:
"""Return the ACITenant instance of related ACIAppProfile."""
Expand Down
16 changes: 16 additions & 0 deletions netbox_aci_plugin/models/tenant_networks.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

from dcim.fields import MACAddressField
from django.contrib.postgres.fields import ArrayField
from django.core.exceptions import ValidationError
from django.core.validators import MaxLengthValidator
from django.db import models
from django.urls import reverse
Expand Down Expand Up @@ -480,6 +481,21 @@ def __str__(self) -> str:
else:
return self.name

def save(self, *args, **kwargs):
"""Saves the current instance to the database."""
# Ensure the assigned ACIVRF belongs to either the same ACITenant as
# the ACIBridgeDomain or to the special ACITenant 'common'.
if (
self.aci_vrf.aci_tenant != self.aci_tenant
and self.aci_vrf.aci_tenant.name != "common"
):
raise ValidationError(
"Assigned ACIVRF have to belong to the same ACITenant as the"
"ACIBridgeDomain or to the special ACITenant 'common'."
)

super().save(*args, **kwargs)

def get_absolute_url(self) -> str:
"""Return the absolute URL of the instance."""
return reverse(
Expand Down
74 changes: 74 additions & 0 deletions netbox_aci_plugin/tests/test_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -806,6 +806,40 @@ def test_invalid_aci_bridge_domain_description_length(self) -> None:
with self.assertRaises(ValidationError):
bd.full_clean()

def test_valid_aci_bridge_domain_aci_vrf_assignment_form_tenant_common(
self,
) -> None:
"""Test valid assignment of ACI VRF from ACI Tenant 'common'."""
tenant_common = ACITenant.objects.get_or_create(name="common")[0]
vrf_common = ACIVRF.objects.create(
name="common_vrf", aci_tenant=tenant_common
)
bd = ACIBridgeDomain(
name="ACIBDTest1",
aci_tenant=self.aci_tenant,
aci_vrf=vrf_common,
)
bd.full_clean()
bd.save()
self.assertEqual(bd.aci_vrf, vrf_common)

def test_invalid_aci_bridge_domain_aci_vrf_assignment_from_tenant_other(
self,
) -> None:
"""Test invalid assignment of ACI VRF from ACI Tenant 'other'."""
tenant_other = ACITenant.objects.get_or_create(name="other")[0]
vrf_other = ACIVRF.objects.create(
name="other_vrf", aci_tenant=tenant_other
)
bd = ACIBridgeDomain(
name="ACIBDTest1",
aci_tenant=self.aci_tenant,
aci_vrf=vrf_other,
)
with self.assertRaises(ValidationError):
bd.full_clean()
bd.save()

def test_constraint_unique_aci_bridge_domain_name_per_aci_tenant(
self,
) -> None:
Expand Down Expand Up @@ -1323,6 +1357,46 @@ def test_invalid_aci_endpoint_group_description_length(self) -> None:
with self.assertRaises(ValidationError):
epg.full_clean()

def test_valid_aci_endpoint_group_aci_bd_assignment_form_tenant_common(
self,
) -> None:
"""Test valid assignment of ACI BD from ACI Tenant 'common'."""
tenant_common = ACITenant.objects.get_or_create(name="common")[0]
vrf_common = ACIVRF.objects.create(
name="common_vrf", aci_tenant=tenant_common
)
bd_common = ACIBridgeDomain.objects.create(
name="common_bd", aci_tenant=tenant_common, aci_vrf=vrf_common
)
epg = ACIEndpointGroup.objects.create(
name="ACIEPGTest1",
aci_app_profile=self.aci_app_profile,
aci_bridge_domain=bd_common,
)
epg.full_clean()
epg.save()
self.assertEqual(epg.aci_bridge_domain, bd_common)

def test_invalid_aci_endpoint_group_aci_bd_assignment_from_tenant_other(
self,
) -> None:
"""Test invalid assignment of ACI BD from ACI Tenant 'other'."""
tenant_other = ACITenant.objects.get_or_create(name="other")[0]
vrf_other = ACIVRF.objects.create(
name="other_vrf", aci_tenant=tenant_other
)
bd_other = ACIBridgeDomain.objects.create(
name="other_bd", aci_tenant=tenant_other, aci_vrf=vrf_other
)
epg = ACIEndpointGroup(
name="ACIEPGTest1",
aci_app_profile=self.aci_app_profile,
aci_bridge_domain=bd_other,
)
with self.assertRaises(ValidationError):
epg.full_clean()
epg.save()

def test_constraint_unique_aci_endpoint_group_name_per_aci_app_profile(
self,
) -> None:
Expand Down

0 comments on commit 88a9de2

Please sign in to comment.