feat(ipam): Add bulk creation support for prefixes

Implement bulk prefix creation using network patterns
(e.g., 10.[0-2].0/2). Refactor bulk creation views to support reusable
context and templates. Rename IPAddressBulkCreateForm to
IPNetworkBulkCreateForm for IPv4/IPv6 support.
This commit is contained in:
Martin Hauser
2026-03-12 21:45:14 +01:00
parent 02165a28a0
commit 577fcff1be
14 changed files with 258 additions and 96 deletions

View File

@@ -1,14 +1,17 @@
from django import forms
from django.utils.translation import gettext_lazy as _
from utilities.forms.fields import ExpandableIPAddressField
from utilities.forms.fields import ExpandableIPNetworkField
__all__ = (
'IPAddressBulkCreateForm',
'IPNetworkBulkCreateForm',
)
class IPAddressBulkCreateForm(forms.Form):
pattern = ExpandableIPAddressField(
class IPNetworkBulkCreateForm(forms.Form):
"""
Pattern form for bulk-creating IP-based objects (addresses, prefixes).
"""
pattern = ExpandableIPNetworkField(
label=_('Address pattern')
)

View File

@@ -37,6 +37,7 @@ __all__ = (
'IPAddressBulkAddForm',
'IPAddressForm',
'IPRangeForm',
'PrefixBulkAddForm',
'PrefixForm',
'RIRForm',
'RoleForm',
@@ -249,6 +250,32 @@ class PrefixForm(TenancyForm, ScopedForm, PrimaryModelForm):
self.fields['vlan'].widget.attrs.pop('data-dynamic-params', None)
class PrefixBulkAddForm(TenancyForm, NetBoxModelForm):
vrf = DynamicModelChoiceField(
queryset=VRF.objects.all(),
required=False,
label=_('VRF')
)
role = DynamicModelChoiceField(
label=_('Role'),
queryset=Role.objects.all(),
required=False,
quick_add=True
)
fieldsets = (
FieldSet('status', 'role', 'vrf', 'is_pool', 'mark_utilized', 'description', 'tags', name=_('Prefix')),
FieldSet('tenant_group', 'tenant', name=_('Tenancy')),
)
class Meta:
model = Prefix
fields = [
'prefix', 'vrf', 'status', 'role', 'is_pool', 'mark_utilized', 'description', 'tenant_group', 'tenant',
'tags',
]
class IPRangeForm(TenancyForm, PrimaryModelForm):
vrf = DynamicModelChoiceField(
queryset=VRF.objects.all(),
@@ -472,6 +499,11 @@ class IPAddressBulkAddForm(TenancyForm, NetBoxModelForm):
label=_('VRF')
)
fieldsets = (
FieldSet('status', 'role', 'vrf', 'dns_name', 'description', 'tags', name=_('IP Address')),
FieldSet('tenant_group', 'tenant', name=_('Tenancy')),
)
class Meta:
model = IPAddress
fields = [

View File

@@ -467,6 +467,66 @@ class PrefixTestCase(ViewTestCases.PrimaryObjectViewTestCase):
'description': 'New description',
}
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_bulk_add_ipv4_prefixes(self):
"""Test bulk creating IPv4 prefixes using a pattern."""
obj_perm = ObjectPermission(name='Test permission', actions=['add'])
obj_perm.save()
obj_perm.users.add(self.user)
obj_perm.object_types.add(ObjectType.objects.get_for_model(Prefix))
initial_count = Prefix.objects.count()
url = reverse('ipam:prefix_bulk_add')
data = {
'pattern': '10.0.[0-2].0/24',
'status': PrefixStatusChoices.STATUS_ACTIVE,
}
response = self.client.post(url, data)
self.assertHttpStatus(response, 302)
self.assertEqual(Prefix.objects.count(), initial_count + 3)
for i in range(3):
self.assertTrue(Prefix.objects.filter(prefix=IPNetwork(f'10.0.{i}.0/24')).exists())
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_bulk_add_ipv6_prefixes(self):
"""Test bulk creating IPv6 prefixes using a pattern."""
obj_perm = ObjectPermission(name='Test permission', actions=['add'])
obj_perm.save()
obj_perm.users.add(self.user)
obj_perm.object_types.add(ObjectType.objects.get_for_model(Prefix))
initial_count = Prefix.objects.count()
url = reverse('ipam:prefix_bulk_add')
data = {
'pattern': 'fd00:db8:[0-3]::/48',
'status': PrefixStatusChoices.STATUS_ACTIVE,
}
response = self.client.post(url, data)
self.assertHttpStatus(response, 302)
self.assertEqual(Prefix.objects.count(), initial_count + 4)
for i in range(4):
self.assertTrue(Prefix.objects.filter(prefix=IPNetwork(f'fd00:db8:{i}::/48')).exists())
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_bulk_add_ipv6_prefixes_uppercase_hex(self):
"""Test bulk creating IPv6 prefixes using uppercase hex in the pattern."""
obj_perm = ObjectPermission(name='Test permission', actions=['add'])
obj_perm.save()
obj_perm.users.add(self.user)
obj_perm.object_types.add(ObjectType.objects.get_for_model(Prefix))
initial_count = Prefix.objects.count()
url = reverse('ipam:prefix_bulk_add')
data = {
'pattern': 'fd00:0:0:[48-4F]00::/56',
'status': PrefixStatusChoices.STATUS_ACTIVE,
}
response = self.client.post(url, data)
self.assertHttpStatus(response, 302)
self.assertEqual(Prefix.objects.count(), initial_count + 8)
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_prefix_prefixes(self):
prefixes = (

View File

@@ -714,6 +714,7 @@ class PrefixIPAddressesView(generic.ObjectChildrenView):
class PrefixEditView(generic.ObjectEditView):
queryset = Prefix.objects.all()
form = forms.PrefixForm
template_name = 'ipam/prefix_edit.html'
@register_model_view(Prefix, 'delete')
@@ -721,6 +722,15 @@ class PrefixDeleteView(generic.ObjectDeleteView):
queryset = Prefix.objects.all()
@register_model_view(Prefix, 'bulk_add', path='bulk-add', detail=False)
class PrefixBulkCreateView(generic.BulkCreateView):
queryset = Prefix.objects.all()
form = forms.IPNetworkBulkCreateForm
model_form = forms.PrefixBulkAddForm
pattern_target = 'prefix'
template_name = 'ipam/prefix_bulk_add.html'
@register_model_view(Prefix, 'bulk_import', path='import', detail=False)
class PrefixBulkImportView(generic.BulkImportView):
queryset = Prefix.objects.all()
@@ -979,7 +989,7 @@ class IPAddressDeleteView(generic.ObjectDeleteView):
@register_model_view(IPAddress, 'bulk_add', path='bulk-add', detail=False)
class IPAddressBulkCreateView(generic.BulkCreateView):
queryset = IPAddress.objects.all()
form = forms.IPAddressBulkCreateForm
form = forms.IPNetworkBulkCreateForm
model_form = forms.IPAddressBulkAddForm
pattern_target = 'address'
template_name = 'ipam/ipaddress_bulk_add.html'

View File

@@ -254,6 +254,17 @@ class BulkCreateView(GetReturnURLMixin, BaseMultiObjectView):
return new_objects
def _get_context(self, request, form, model_form):
model = self.queryset.model
return {
'obj_type': model._meta.verbose_name,
'form': form,
'model_form': model_form,
'return_url': self.get_return_url(request),
'add_url': get_action_url(model, 'add'),
**self.get_extra_context(request),
}
#
# Request handlers
#
@@ -268,13 +279,7 @@ class BulkCreateView(GetReturnURLMixin, BaseMultiObjectView):
form = self.form()
model_form = self.model_form(initial=initial)
return render(request, self.template_name, {
'obj_type': self.model_form._meta.model._meta.verbose_name,
'form': form,
'model_form': model_form,
'return_url': self.get_return_url(request),
**self.get_extra_context(request),
})
return render(request, self.template_name, self._get_context(request, form, model_form))
def post(self, request):
logger = logging.getLogger('netbox.views.BulkCreateView')
@@ -313,13 +318,7 @@ class BulkCreateView(GetReturnURLMixin, BaseMultiObjectView):
else:
logger.debug("Form validation failed")
return render(request, self.template_name, {
'form': form,
'model_form': model_form,
'obj_type': model._meta.verbose_name,
'return_url': self.get_return_url(request),
**self.get_extra_context(request),
})
return render(request, self.template_name, self._get_context(request, form, model_form))
class BulkImportView(GetReturnURLMixin, BaseMultiObjectView):

View File

@@ -0,0 +1,49 @@
{% extends 'generic/object_edit.html' %}
{% load helpers %}
{% load form_helpers %}
{% load i18n %}
{% block title %}{% blocktrans trimmed with object_type=obj_type %}Bulk Add {{ object_type }}s{% endblocktrans %}{% endblock %}
{% block tabs %}
<ul class="nav nav-tabs">
<li class="nav-item">
<a href="{{ add_url }}{% querystring request %}" class="nav-link">
{% trans "Create" %}
</a>
</li>
<li class="nav-item">
<a class="nav-link active">
{% trans "Bulk Create" %}
</a>
</li>
</ul>
{% endblock %}
{% block form %}
<div class="field-group my-5">
<div class="row">
<h2 class="col-9 offset-3">{% trans "Pattern" %}</h2>
</div>
{% render_field form.pattern %}
</div>
{% if model_form.fieldsets %}
{% for fieldset in model_form.fieldsets %}
{% render_fieldset model_form fieldset %}
{% endfor %}
{% else %}
<div class="field-group my-5">
{% render_form model_form %}
</div>
{% endif %}
{% if model_form.custom_fields %}
<div class="field-group my-5">
<div class="row">
<h2 class="col-9 offset-3">{% trans "Custom Fields" %}</h2>
</div>
{% render_custom_fields model_form %}
</div>
{% endif %}
{% endblock %}

View File

@@ -0,0 +1,17 @@
{% load helpers %}
{% load i18n %}
<ul class="nav nav-tabs">
<li class="nav-item">
<a href="{% url 'ipam:prefix_add' %}{% querystring request %}" class="nav-link {% if active_tab == 'add' %}active{% endif %}">
{% if object.pk %}{% trans "Edit" %}{% else %}{% trans "Create" %}{% endif %}
</a>
</li>
{% if not object.pk %}
<li class="nav-item">
<a href="{% url 'ipam:prefix_bulk_add' %}{% querystring request %}" class="nav-link {% if active_tab == 'bulk_add' %}active{% endif %}">
{% trans "Bulk Create" %}
</a>
</li>
{% endif %}
</ul>

View File

@@ -1,40 +1,5 @@
{% extends 'generic/object_edit.html' %}
{% load static %}
{% load form_helpers %}
{% load i18n %}
{% block title %}{% trans "Bulk Add IP Addresses" %}{% endblock %}
{% extends 'generic/bulk_add.html' %}
{% block tabs %}
{% include 'ipam/inc/ipaddress_edit_header.html' with active_tab='bulk_add' %}
{% endblock %}
{% block form %}
<div class="field-group my-5">
<div class="row">
<h2 class="col-9 offset-3">{% trans "IP Addresses" %}</h2>
</div>
{% render_field form.pattern %}
{% render_field model_form.status %}
{% render_field model_form.role %}
{% render_field model_form.vrf %}
{% render_field model_form.description %}
{% render_field model_form.tags %}
</div>
<div class="field-group my-5">
<div class="row">
<h2 class="col-9 offset-3">{% trans "Tenancy" %}</h2>
</div>
{% render_field model_form.tenant_group %}
{% render_field model_form.tenant %}
</div>
{% if model_form.custom_fields %}
<div class="field-group my-5">
<div class="row">
<h2 class="col-9 offset-3">{% trans "Custom Fields" %}</h2>
</div>
{% render_custom_fields model_form %}
</div>
{% endif %}
{% endblock %}

View File

@@ -0,0 +1,5 @@
{% extends 'generic/bulk_add.html' %}
{% block tabs %}
{% include 'ipam/inc/prefix_edit_header.html' with active_tab='bulk_add' %}
{% endblock %}

View File

@@ -0,0 +1,5 @@
{% extends 'generic/object_edit.html' %}
{% block tabs %}
{% include 'ipam/inc/prefix_edit_header.html' with active_tab='add' %}
{% endblock %}

View File

@@ -4,7 +4,7 @@ ALPHANUMERIC_EXPANSION_PATTERN = r'\[((?:[a-zA-Z0-9]+[?:,-])+[a-zA-Z0-9]+)\]'
# IP address expansion patterns
IP4_EXPANSION_PATTERN = r'\[((?:[0-9]{1,3}[?:,-])+[0-9]{1,3})\]'
IP6_EXPANSION_PATTERN = r'\[((?:[0-9a-f]{1,4}[?:,-])+[0-9a-f]{1,4})\]'
IP6_EXPANSION_PATTERN = r'\[((?:[0-9a-fA-F]{1,4}[?:,-])+[0-9a-fA-F]{1,4})\]'
# Boolean widget choices
BOOLEAN_WITH_BLANK_CHOICES = (

View File

@@ -1,13 +1,14 @@
import re
import netaddr
from django import forms
from django.utils.translation import gettext_lazy as _
from utilities.forms.constants import *
from utilities.forms.utils import expand_alphanumeric_pattern, expand_ipaddress_pattern
from utilities.forms.utils import expand_alphanumeric_pattern, expand_ipnetwork_pattern
__all__ = (
'ExpandableIPAddressField',
'ExpandableIPNetworkField',
'ExpandableNameField',
)
@@ -34,22 +35,38 @@ class ExpandableNameField(forms.CharField):
return [value]
class ExpandableIPAddressField(forms.CharField):
class ExpandableIPNetworkField(forms.CharField):
"""
A field which allows for expansion of IP address ranges
Example: '192.0.2.[1-254]/24' => ['192.0.2.1/24', '192.0.2.2/24', '192.0.2.3/24' ... '192.0.2.254/24']
A CharField that expands numeric range patterns in IPv4/IPv6 CIDR notation into multiple entries.
Examples:
'192.0.2.[1-254]/32' => ['192.0.2.1/32', '192.0.2.2/32', ...]
'10.[0-3,10-13].0.0/16' => ['10.0.0.0/16', '10.1.0.0/16', ..., '10.10.0.0/16', ...]
'2001:db8:[0-f]::/64' => ['2001:db8:0::/64', '2001:db8:1::/64', ...]
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
if not self.help_text:
self.help_text = _('Specify a numeric range to create multiple IPs.<br />'
'Example: <code>192.0.2.[1,5,100-254]/24</code>')
self.help_text = _(
'Use bracket notation to specify numeric ranges for bulk creation (CIDR required).<br />'
'Examples: <code>192.0.2.[1-10]/32</code>, <code>10.[0-3,10-13].0.0/16</code>, '
'<code>2001:db8:[a-f]::/64</code>'
)
def to_python(self, value):
# Hackish address family detection but it's all we have to work with
if '.' in value and re.search(IP4_EXPANSION_PATTERN, value):
return list(expand_ipaddress_pattern(value, 4))
if ':' in value and re.search(IP6_EXPANSION_PATTERN, value):
return list(expand_ipaddress_pattern(value, 6))
if not value:
return [value]
# Replace expansion brackets with a neutral value to get a parseable IP/CIDR
stripped = re.sub(r'\[[^\]]+\]', '0', value)
try:
family = netaddr.IPNetwork(stripped).version
except (netaddr.AddrFormatError, ValueError):
return [value]
if family == 4 and re.search(IP4_EXPANSION_PATTERN, value):
return list(expand_ipnetwork_pattern(value, 4))
if family == 6 and re.search(IP6_EXPANSION_PATTERN, value):
return list(expand_ipnetwork_pattern(value.lower(), 6))
return [value]

View File

@@ -12,7 +12,7 @@ from .constants import *
__all__ = (
'add_blank_choice',
'expand_alphanumeric_pattern',
'expand_ipaddress_pattern',
'expand_ipnetwork_pattern',
'form_from_model',
'get_field_value',
'get_selected_values',
@@ -106,9 +106,9 @@ def expand_alphanumeric_pattern(string):
yield "{}{}{}".format(lead, i, remnant)
def expand_ipaddress_pattern(string, family):
def expand_ipnetwork_pattern(string, family):
"""
Expand an IP address pattern into a list of strings. Examples:
Expand an IP network pattern into a list of strings. Examples:
'192.0.2.[1,2,100-250]/24' => ['192.0.2.1/24', '192.0.2.2/24', '192.0.2.100/24' ... '192.0.2.250/24']
'2001:db8:0:[0,fd-ff]::/64' => ['2001:db8:0:0::/64', '2001:db8:0:fd::/64', ... '2001:db8:0:ff::/64']
"""
@@ -124,7 +124,7 @@ def expand_ipaddress_pattern(string, family):
parsed_range = parse_numeric_range(pattern, base)
for i in parsed_range:
if re.search(regex, remnant):
for string in expand_ipaddress_pattern(remnant, family):
for string in expand_ipnetwork_pattern(remnant, family):
yield ''.join([lead, format(i, 'x' if family == 6 else 'd'), string])
else:
yield ''.join([lead, format(i, 'x' if family == 6 else 'd'), remnant])

View File

@@ -6,13 +6,13 @@ from netbox.choices import ImportFormatChoices
from utilities.forms.bulk_import import BulkImportForm
from utilities.forms.fields.csv import CSVSelectWidget
from utilities.forms.forms import BulkRenameForm
from utilities.forms.utils import expand_alphanumeric_pattern, expand_ipaddress_pattern, get_field_value
from utilities.forms.utils import expand_alphanumeric_pattern, expand_ipnetwork_pattern, get_field_value
from utilities.forms.widgets.select import AvailableOptions, SelectedOptions
class ExpandIPAddress(TestCase):
class ExpandIPNetwork(TestCase):
"""
Validate the operation of expand_ipaddress_pattern().
Validate the operation of expand_ipnetwork_pattern().
"""
def test_ipv4_range(self):
input = '1.2.3.[9-10]/32'
@@ -21,7 +21,7 @@ class ExpandIPAddress(TestCase):
'1.2.3.10/32',
])
self.assertEqual(sorted(expand_ipaddress_pattern(input, 4)), output)
self.assertEqual(sorted(expand_ipnetwork_pattern(input, 4)), output)
def test_ipv4_set(self):
input = '1.2.3.[4,44]/32'
@@ -30,7 +30,7 @@ class ExpandIPAddress(TestCase):
'1.2.3.44/32',
])
self.assertEqual(sorted(expand_ipaddress_pattern(input, 4)), output)
self.assertEqual(sorted(expand_ipnetwork_pattern(input, 4)), output)
def test_ipv4_multiple_ranges(self):
input = '1.[9-10].3.[9-11]/32'
@@ -43,7 +43,7 @@ class ExpandIPAddress(TestCase):
'1.10.3.11/32',
])
self.assertEqual(sorted(expand_ipaddress_pattern(input, 4)), output)
self.assertEqual(sorted(expand_ipnetwork_pattern(input, 4)), output)
def test_ipv4_multiple_sets(self):
input = '1.[2,22].3.[4,44]/32'
@@ -54,7 +54,7 @@ class ExpandIPAddress(TestCase):
'1.22.3.44/32',
])
self.assertEqual(sorted(expand_ipaddress_pattern(input, 4)), output)
self.assertEqual(sorted(expand_ipnetwork_pattern(input, 4)), output)
def test_ipv4_set_and_range(self):
input = '1.[2,22].3.[9-11]/32'
@@ -67,7 +67,7 @@ class ExpandIPAddress(TestCase):
'1.22.3.11/32',
])
self.assertEqual(sorted(expand_ipaddress_pattern(input, 4)), output)
self.assertEqual(sorted(expand_ipnetwork_pattern(input, 4)), output)
def test_ipv6_range(self):
input = 'fec::abcd:[9-b]/64'
@@ -77,7 +77,7 @@ class ExpandIPAddress(TestCase):
'fec::abcd:b/64',
])
self.assertEqual(sorted(expand_ipaddress_pattern(input, 6)), output)
self.assertEqual(sorted(expand_ipnetwork_pattern(input, 6)), output)
def test_ipv6_range_multichar_field(self):
input = 'fec::abcd:[f-11]/64'
@@ -87,7 +87,7 @@ class ExpandIPAddress(TestCase):
'fec::abcd:11/64',
])
self.assertEqual(sorted(expand_ipaddress_pattern(input, 6)), output)
self.assertEqual(sorted(expand_ipnetwork_pattern(input, 6)), output)
def test_ipv6_set(self):
input = 'fec::abcd:[9,ab]/64'
@@ -96,7 +96,7 @@ class ExpandIPAddress(TestCase):
'fec::abcd:ab/64',
])
self.assertEqual(sorted(expand_ipaddress_pattern(input, 6)), output)
self.assertEqual(sorted(expand_ipnetwork_pattern(input, 6)), output)
def test_ipv6_multiple_ranges(self):
input = 'fec::[1-2]bcd:[9-b]/64'
@@ -109,7 +109,7 @@ class ExpandIPAddress(TestCase):
'fec::2bcd:b/64',
])
self.assertEqual(sorted(expand_ipaddress_pattern(input, 6)), output)
self.assertEqual(sorted(expand_ipnetwork_pattern(input, 6)), output)
def test_ipv6_multiple_sets(self):
input = 'fec::[a,f]bcd:[9,ab]/64'
@@ -120,7 +120,7 @@ class ExpandIPAddress(TestCase):
'fec::fbcd:ab/64',
])
self.assertEqual(sorted(expand_ipaddress_pattern(input, 6)), output)
self.assertEqual(sorted(expand_ipnetwork_pattern(input, 6)), output)
def test_ipv6_set_and_range(self):
input = 'fec::[dead,beaf]:[9-b]/64'
@@ -133,41 +133,41 @@ class ExpandIPAddress(TestCase):
'fec::beaf:b/64',
])
self.assertEqual(sorted(expand_ipaddress_pattern(input, 6)), output)
self.assertEqual(sorted(expand_ipnetwork_pattern(input, 6)), output)
def test_invalid_address_family(self):
with self.assertRaisesRegex(Exception, 'Invalid IP address family: 5'):
sorted(expand_ipaddress_pattern(None, 5))
sorted(expand_ipnetwork_pattern(None, 5))
def test_invalid_non_pattern(self):
with self.assertRaises(ValueError):
sorted(expand_ipaddress_pattern('1.2.3.4/32', 4))
sorted(expand_ipnetwork_pattern('1.2.3.4/32', 4))
def test_invalid_range(self):
with self.assertRaises(ValueError):
sorted(expand_ipaddress_pattern('1.2.3.[4-]/32', 4))
sorted(expand_ipnetwork_pattern('1.2.3.[4-]/32', 4))
with self.assertRaises(ValueError):
sorted(expand_ipaddress_pattern('1.2.3.[-4]/32', 4))
sorted(expand_ipnetwork_pattern('1.2.3.[-4]/32', 4))
with self.assertRaises(ValueError):
sorted(expand_ipaddress_pattern('1.2.3.[4--5]/32', 4))
sorted(expand_ipnetwork_pattern('1.2.3.[4--5]/32', 4))
def test_invalid_range_bounds(self):
self.assertEqual(sorted(expand_ipaddress_pattern('1.2.3.[4-3]/32', 6)), [])
self.assertEqual(sorted(expand_ipnetwork_pattern('1.2.3.[4-3]/32', 6)), [])
def test_invalid_set(self):
with self.assertRaises(ValueError):
sorted(expand_ipaddress_pattern('1.2.3.[4]/32', 4))
sorted(expand_ipnetwork_pattern('1.2.3.[4]/32', 4))
with self.assertRaises(ValueError):
sorted(expand_ipaddress_pattern('1.2.3.[4,]/32', 4))
sorted(expand_ipnetwork_pattern('1.2.3.[4,]/32', 4))
with self.assertRaises(ValueError):
sorted(expand_ipaddress_pattern('1.2.3.[,4]/32', 4))
sorted(expand_ipnetwork_pattern('1.2.3.[,4]/32', 4))
with self.assertRaises(ValueError):
sorted(expand_ipaddress_pattern('1.2.3.[4,,5]/32', 4))
sorted(expand_ipnetwork_pattern('1.2.3.[4,,5]/32', 4))
class ExpandAlphanumeric(TestCase):