Add cycle identification

Use a depth-first search to identify any cycles, to be able to report
them as such to the user.
This commit is contained in:
Marko Hauptvogel 2025-08-22 09:53:52 +02:00
parent c158bfa87a
commit 96cdddd3f8
2 changed files with 84 additions and 12 deletions

View File

@ -1149,15 +1149,24 @@ model: TEST-5000
slug: test-5000 slug: test-5000
u_height: 1 u_height: 1
interfaces: interfaces:
- name: Interface 1 - name: Cycle Interface 1
type: 1000base-t type: 1000base-t
bridge: Interface 2 bridge: Cycle Interface 2
- name: Interface 2 - name: Cycle Interface 2
type: 1000base-t type: 1000base-t
bridge: Interface 3 bridge: Cycle Interface 3
- name: Interface 3 - name: Cycle Interface 3
type: 1000base-t type: 1000base-t
bridge: Interface 1 bridge: Cycle Interface 1
- name: Unrelated Interface 1
type: 1000base-t
- name: Unrelated Interface 2
type: 1000base-t
bridge: Cycle Interface 1
- name: Unrelated Interface 3
type: 1000base-t
bridge: Cycle Interface 3
""" """
# Add all required permissions to the test user # Add all required permissions to the test user
@ -1183,8 +1192,51 @@ interfaces:
response = self.client.post(reverse('dcim:devicetype_bulk_import'), data=form_data, follow=True) response = self.client.post(reverse('dcim:devicetype_bulk_import'), data=form_data, follow=True)
self.assertHttpStatus(response, 200) self.assertHttpStatus(response, 200)
self.assertContains(response, "interfaces: Dependency cycle detected in subset " self.assertContains(response, "interfaces: Dependency cycle [Cycle Interface 1, Cycle Interface 2, "
"[Interface 1, Interface 2, Interface 3]") "Cycle Interface 3, Cycle Interface 1] detected")
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_import_interfacebridge_invalid(self):
IMPORT_DATA = """
manufacturer: Manufacturer 1
model: TEST-6000
slug: test-6000
u_height: 1
interfaces:
- name: Interface 1
type: 1000base-t
- name: Interface 2
type: 1000base-t
bridge: Non-existent Bridge
- name: Interface 3
type: 1000base-t
"""
# Add all required permissions to the test user
self.add_permissions(
'dcim.view_devicetype',
'dcim.add_devicetype',
'dcim.add_consoleporttemplate',
'dcim.add_consoleserverporttemplate',
'dcim.add_powerporttemplate',
'dcim.add_poweroutlettemplate',
'dcim.add_interfacetemplate',
'dcim.add_frontporttemplate',
'dcim.add_rearporttemplate',
'dcim.add_modulebaytemplate',
'dcim.add_devicebaytemplate',
'dcim.add_inventoryitemtemplate',
)
form_data = {
'data': IMPORT_DATA,
'format': 'yaml'
}
response = self.client.post(reverse('dcim:devicetype_bulk_import'), data=form_data, follow=True)
self.assertHttpStatus(response, 200)
self.assertContains(response, "interfaces[1] bridge: Select a valid choice. "
"That choice is not one of the available choices.")
def test_export_objects(self): def test_export_objects(self):
url = reverse('dcim:devicetype_list') url = reverse('dcim:devicetype_list')

View File

@ -1339,10 +1339,30 @@ class DeviceTypeImportView(generic.BulkImportView):
unsatisfied_requirements = list(ifname for ifname, deps in requires.items() if deps) unsatisfied_requirements = list(ifname for ifname, deps in requires.items() if deps)
if unsatisfied_requirements: if unsatisfied_requirements:
def find_cycle(visited_interfaces):
"""Recursive depth-first search to identify cycles."""
for ifname in list(required_by.get(visited_interfaces[-1], list())):
if ifname in visited_interfaces:
# found a cycle and its start
start_index = visited_interfaces.index(ifname)
visited_interfaces.append(ifname)
return list(reversed(visited_interfaces[start_index:]))
result = find_cycle(visited_interfaces + [ifname])
if result:
return result
return None
# Check if there is a cycle
for ifname in unsatisfied_requirements:
cycle = find_cycle([ifname])
if cycle:
# stop at the first one, finding all while avoiding duplicates would be hard
raise ValidationError( raise ValidationError(
_("Dependency cycle detected in subset [%(interfaces)s]"), _("Dependency cycle [%(interfaces)s] detected"),
params={"interfaces": ", ".join(unsatisfied_requirements)}, params={"interfaces": ", ".join(cycle)},
) )
# no cycle, so the unsatisfied requirements must be due to requirements on non-existent interfaces,
# which will cause a validation error later when checking the individual interface objects.
# apply the topological sorting to the actual list # apply the topological sorting to the actual list
def get_sort_key(interface): def get_sort_key(interface):