NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-07-29 11:56:25 -06:00
Code Issues Actions 8 Packages Projects Releases Wiki Activity

fix merge conflicts

Browse Source
This commit is contained in:
Arthur Hanson 2024-11-04 08:00:29 -08:00
commit 72712cc78b
58 changed files with 658 additions and 114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
docs/models/dcim/interface.md
View File

@ -120,6 +120,10 @@ The "native" (untagged) VLAN for the interface. Valid only when one of the above
The tagged VLANs which are configured to be carried by this interface. Valid only for the "tagged" 802.1Q mode above.
### Q-in-Q SVLAN
The assigned service VLAN (for Q-in-Q/802.1ad interfaces).
### Wireless Role
Indicates the configured role for wireless interfaces (access point or station).

8
docs/models/ipam/vlan.md
View File

@ -26,3 +26,11 @@ The user-defined functional [role](./role.md) assigned to the VLAN.
### VLAN Group or Site
The [VLAN group](./vlangroup.md) or [site](../dcim/site.md) to which the VLAN is assigned.
### Q-in-Q Role
For VLANs which comprise a Q-in-Q/IEEE 802.1ad topology, this field indicates whether the VLAN is treated as a service or customer VLAN.
### Q-in-Q Service VLAN
The designated parent service VLAN for a Q-in-Q customer VLAN. This may be set only for Q-in-Q custom VLANs.

4
docs/models/virtualization/vminterface.md
View File

@ -53,6 +53,10 @@ The "native" (untagged) VLAN for the interface. Valid only when one of the above
The tagged VLANs which are configured to be carried by this interface. Valid only for the "tagged" 802.1Q mode above.
### Q-in-Q SVLAN
The assigned service VLAN (for Q-in-Q/802.1ad interfaces).
### VRF
The [virtual routing and forwarding](../ipam/vrf.md) instance to which this interface is assigned.

53
docs/plugins/development/background-jobs.md
View File

@ -29,6 +29,9 @@ class MyTestJob(JobRunner):
You can schedule the background job from within your code (e.g. from a model's `save()` method or a view) by calling `MyTestJob.enqueue()`. This method passes through all arguments to `Job.enqueue()`. However, no `name` argument must be passed, as the background job name will be used instead.
!!! tip
A set of predefined intervals is available at `core.choices.JobIntervalChoices` for convenience.
### Attributes
`JobRunner` attributes are defined under a class named `Meta` within the job. These are optional, but encouraged.
@ -46,26 +49,52 @@ As described above, jobs can be scheduled for immediate execution or at any late
#### Example
```python title="models.py"
from django.db import models
from core.choices import JobIntervalChoices
from netbox.models import NetBoxModel
from .jobs import MyTestJob
class MyModel(NetBoxModel):
foo = models.CharField()
def save(self, *args, **kwargs):
MyTestJob.enqueue_once(instance=self, interval=JobIntervalChoices.INTERVAL_HOURLY)
return super().save(*args, **kwargs)
def sync(self):
MyTestJob.enqueue(instance=self)
```
### System Jobs
Some plugins may implement background jobs that are decoupled from the request/response cycle. Typical use cases would be housekeeping tasks or synchronization jobs. These can be registered as _system jobs_ using the `system_job()` decorator. The job interval must be passed as an integer (in minutes) when registering a system job. System jobs are scheduled automatically when the RQ worker (`manage.py rqworker`) is run.
#### Example
```python title="jobs.py"
from netbox.jobs import JobRunner
from core.choices import JobIntervalChoices
from netbox.jobs import JobRunner, system_job
from .models import MyModel
# Specify a predefined choice or an integer indicating
# the number of minutes between job executions
@system_job(interval=JobIntervalChoices.INTERVAL_HOURLY)
class MyHousekeepingJob(JobRunner):
class Meta:
name = "Housekeeping"
name = "My Housekeeping Job"
def run(self, *args, **kwargs):
# your logic goes here
MyModel.objects.filter(foo='bar').delete()
system_jobs = (
MyHousekeepingJob,
)
```
```python title="__init__.py"
from netbox.plugins import PluginConfig
class MyPluginConfig(PluginConfig):
def ready(self):
from .jobs import MyHousekeepingJob
MyHousekeepingJob.setup(interval=60)
```
!!! note
Ensure that any system jobs are imported on initialization. Otherwise, they won't be registered. This can be achieved by extending the PluginConfig's `ready()` method.
## Task queues

2
docs/plugins/development/data-backends.md
View File

@ -18,6 +18,6 @@ backends = [MyDataBackend]
```
!!! tip
The path to the list of search indexes can be modified by setting `data_backends` in the PluginConfig instance.
The path to the list of data backends can be modified by setting `data_backends` in the PluginConfig instance.
::: netbox.data_backends.DataBackend

14
netbox/core/choices.py
View File

@ -72,6 +72,20 @@ class JobStatusChoices(ChoiceSet):
)
class JobIntervalChoices(ChoiceSet):
INTERVAL_MINUTELY = 1
INTERVAL_HOURLY = 60
INTERVAL_DAILY = 60 * 24
INTERVAL_WEEKLY = 60 * 24 * 7
CHOICES = (
(INTERVAL_MINUTELY, _('Minutely')),
(INTERVAL_HOURLY, _('Hourly')),
(INTERVAL_DAILY, _('Daily')),
(INTERVAL_WEEKLY, _('Weekly')),
)
#
# ObjectChanges
#

11
netbox/core/management/commands/rqworker.py
View File

@ -2,6 +2,8 @@ import logging
from django_rq.management.commands.rqworker import Command as _Command
from netbox.registry import registry
DEFAULT_QUEUES = ('high', 'default', 'low')
@ -14,6 +16,15 @@ class Command(_Command):
of only the 'default' queue).
"""
def handle(self, *args, **options):
# Setup system jobs.
for job, kwargs in registry['system_jobs'].items():
try:
interval = kwargs['interval']
except KeyError:
raise TypeError("System job must specify an interval (in minutes).")
logger.debug(f"Scheduling system job {job.name} (interval={interval})")
job.enqueue_once(**kwargs)
# Run the worker with scheduler functionality
options['with_scheduler'] = True

9
netbox/dcim/api/serializers_/device_components.py
View File

@ -196,6 +196,7 @@ class InterfaceSerializer(NetBoxModelSerializer, CabledObjectSerializer, Connect
required=False,
many=True
)
qinq_svlan = VLANSerializer(nested=True, required=False, allow_null=True)
vlan_translation_policy = VLANTranslationPolicySerializer(nested=True, required=False, allow_null=True)
vrf = VRFSerializer(nested=True, required=False, allow_null=True)
l2vpn_termination = L2VPNTerminationSerializer(nested=True, read_only=True, allow_null=True)
@ -223,10 +224,10 @@ class InterfaceSerializer(NetBoxModelSerializer, CabledObjectSerializer, Connect
'id', 'url', 'display_url', 'display', 'device', 'vdcs', 'module', 'name', 'label', 'type', 'enabled',
'parent', 'bridge', 'lag', 'mtu', 'mac_address', 'speed', 'duplex', 'wwn', 'mgmt_only', 'description',
'mode', 'rf_role', 'rf_channel', 'poe_mode', 'poe_type', 'rf_channel_frequency', 'rf_channel_width',
'tx_power', 'untagged_vlan', 'tagged_vlans', 'mark_connected', 'cable', 'cable_end', 'wireless_link',
'link_peers', 'link_peers_type', 'wireless_lans', 'vrf', 'l2vpn_termination', 'connected_endpoints',
'connected_endpoints_type', 'connected_endpoints_reachable', 'tags', 'custom_fields', 'created',
'last_updated', 'count_ipaddresses', 'count_fhrp_groups', '_occupied', 'vlan_translation_policy'
'tx_power', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan', 'vlan_translation_policy', 'mark_connected',
'cable', 'cable_end', 'wireless_link', 'link_peers', 'link_peers_type', 'wireless_lans', 'vrf',
'l2vpn_termination', 'connected_endpoints', 'connected_endpoints_type', 'connected_endpoints_reachable',
'tags', 'custom_fields', 'created', 'last_updated', 'count_ipaddresses', 'count_fhrp_groups', '_occupied',
]
brief_fields = ('id', 'url', 'display', 'device', 'name', 'description', 'cable', '_occupied')

2
netbox/dcim/choices.py
View File

@ -1258,11 +1258,13 @@ class InterfaceModeChoices(ChoiceSet):
MODE_ACCESS = 'access'
MODE_TAGGED = 'tagged'
MODE_TAGGED_ALL = 'tagged-all'
MODE_Q_IN_Q = 'q-in-q'
CHOICES = (
(MODE_ACCESS, _('Access')),
(MODE_TAGGED, _('Tagged')),
(MODE_TAGGED_ALL, _('Tagged (All)')),
(MODE_Q_IN_Q, _('Q-in-Q (802.1ad)')),
)

2
netbox/dcim/constants.py
View File

@ -124,7 +124,7 @@ COMPATIBLE_TERMINATION_TYPES = {
'rearport': ['consoleport', 'consoleserverport', 'interface', 'frontport', 'rearport', 'circuittermination'],
}
# models values for ContentTypes which may be Cluster scope types
# Models which can serve to scope an object by location
LOCATION_SCOPE_TYPES = (
'region', 'sitegroup', 'site', 'location',
)

6
netbox/dcim/filtersets.py
View File

@ -1648,7 +1648,8 @@ class CommonInterfaceFilterSet(django_filters.FilterSet):
return queryset
return queryset.filter(
Q(untagged_vlan_id=value) |
Q(tagged_vlans=value)
Q(tagged_vlans=value) |
Q(qinq_svlan=value)
)
def filter_vlan(self, queryset, name, value):
@ -1657,7 +1658,8 @@ class CommonInterfaceFilterSet(django_filters.FilterSet):
return queryset
return queryset.filter(
Q(untagged_vlan_id__vid=value) |
Q(tagged_vlans__vid=value)
Q(tagged_vlans__vid=value) |
Q(qinq_svlan__vid=value)
)

2
netbox/dcim/forms/common.py
View File

@ -37,6 +37,8 @@ class InterfaceCommonForm(forms.Form):
del self.fields['vlan_group']
del self.fields['untagged_vlan']
del self.fields['tagged_vlans']
if interface_mode != InterfaceModeChoices.MODE_Q_IN_Q:
del self.fields['qinq_svlan']
def clean(self):
super().clean()

1
netbox/dcim/forms/mixins.py
View File

@ -15,6 +15,7 @@ from utilities.forms.widgets import HTMXSelect
__all__ = (
'ScopedBulkEditForm',
'ScopedForm',
'ScopedImportForm',
)

18
netbox/dcim/forms/model_forms.py
View File

@ -7,6 +7,7 @@ from dcim.choices import *
from dcim.constants import *
from dcim.models import *
from extras.models import ConfigTemplate
from ipam.choices import VLANQinQRoleChoices
from ipam.models import ASN, IPAddress, VLAN, VLANGroup, VLANTranslationPolicy, VRF
from netbox.forms import NetBoxModelForm
from tenancy.forms import TenancyForm
@ -1372,6 +1373,16 @@ class InterfaceForm(InterfaceCommonForm, ModularDeviceComponentForm):
'available_on_device': '$device',
}
)
qinq_svlan = DynamicModelMultipleChoiceField(
queryset=VLAN.objects.all(),
required=False,
label=_('Q-in-Q Service VLAN'),
query_params={
'group_id': '$vlan_group',
'available_on_device': '$device',
'qinq_role': VLANQinQRoleChoices.ROLE_SERVICE,
}
)
vrf = DynamicModelChoiceField(
queryset=VRF.objects.all(),
required=False,
@ -1396,7 +1407,10 @@ class InterfaceForm(InterfaceCommonForm, ModularDeviceComponentForm):
FieldSet('vdcs', 'mtu', 'tx_power', 'enabled', 'mgmt_only', 'mark_connected', name=_('Operation')),
FieldSet('parent', 'bridge', 'lag', name=_('Related Interfaces')),
FieldSet('poe_mode', 'poe_type', name=_('PoE')),
FieldSet('mode', 'vlan_group', 'untagged_vlan', 'tagged_vlans', 'vlan_translation_policy', name=_('802.1Q Switching')),
FieldSet(
'mode', 'vlan_group', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan', 'vlan_translation_policy',
name=_('802.1Q Switching')
),
FieldSet(
'rf_role', 'rf_channel', 'rf_channel_frequency', 'rf_channel_width', 'wireless_lan_group', 'wireless_lans',
name=_('Wireless')
@ -1409,7 +1423,7 @@ class InterfaceForm(InterfaceCommonForm, ModularDeviceComponentForm):
'device', 'module', 'vdcs', 'name', 'label', 'type', 'speed', 'duplex', 'enabled', 'parent', 'bridge', 'lag',
'mac_address', 'wwn', 'mtu', 'mgmt_only', 'mark_connected', 'description', 'poe_mode', 'poe_type', 'mode',
'rf_role', 'rf_channel', 'rf_channel_frequency', 'rf_channel_width', 'tx_power', 'wireless_lans',
'untagged_vlan', 'tagged_vlans', 'vrf', 'tags', 'vlan_translation_policy',
'untagged_vlan', 'tagged_vlans', 'qinq_svlan', 'vlan_translation_policy', 'vrf', 'tags',
]
widgets = {
'speed': NumberWithOptions(

1
netbox/dcim/graphql/types.py
View File

@ -385,6 +385,7 @@ class InterfaceType(IPAddressesMixin, ModularComponentType, CabledObjectMixin, P
wireless_link: Annotated["WirelessLinkType", strawberry.lazy('wireless.graphql.types')] | None
untagged_vlan: Annotated["VLANType", strawberry.lazy('ipam.graphql.types')] | None
vrf: Annotated["VRFType", strawberry.lazy('ipam.graphql.types')] | None
qinq_svlan: Annotated["VLANType", strawberry.lazy('ipam.graphql.types')] | None
vlan_translation_policy: Annotated["VLANTranslationPolicyType", strawberry.lazy('ipam.graphql.types')] | None
vdcs: List[Annotated["VirtualDeviceContextType", strawberry.lazy('dcim.graphql.types')]]

28
netbox/dcim/migrations/0196_qinq_svlan.py Normal file
View File

@ -0,0 +1,28 @@
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('dcim', '0195_interface_vlan_translation_policy'),
('ipam', '0075_vlan_qinq'),
]
operations = [
migrations.AddField(
model_name='interface',
name='qinq_svlan',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(class)ss_svlan', to='ipam.vlan'),
),
migrations.AlterField(
model_name='interface',
name='tagged_vlans',
field=models.ManyToManyField(blank=True, related_name='%(class)ss_as_tagged', to='ipam.vlan'),
),
migrations.AlterField(
model_name='interface',
name='untagged_vlan',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(class)ss_as_untagged', to='ipam.vlan'),
),
]

47
netbox/dcim/models/device_components.py
View File

@ -547,17 +547,48 @@ class BaseInterface(models.Model):
blank=True,
verbose_name=_('bridge interface')
)
untagged_vlan = models.ForeignKey(
to='ipam.VLAN',
on_delete=models.SET_NULL,
related_name='%(class)ss_as_untagged',
null=True,
blank=True,
verbose_name=_('untagged VLAN')
)
tagged_vlans = models.ManyToManyField(
to='ipam.VLAN',
related_name='%(class)ss_as_tagged',
blank=True,
verbose_name=_('tagged VLANs')
)
qinq_svlan = models.ForeignKey(
to='ipam.VLAN',
on_delete=models.SET_NULL,
related_name='%(class)ss_svlan',
null=True,
blank=True,
verbose_name=_('Q-in-Q SVLAN')
)
vlan_translation_policy = models.ForeignKey(
to='ipam.VLANTranslationPolicy',
on_delete=models.PROTECT,
null=True,
blank=True,
verbose_name=_('VLAN Translation Policy'),
verbose_name=_('VLAN Translation Policy')
)
class Meta:
abstract = True
def clean(self):
super().clean()
# SVLAN can be defined only for Q-in-Q interfaces
if self.qinq_svlan and self.mode != InterfaceModeChoices.MODE_Q_IN_Q:
raise ValidationError({
'qinq_svlan': _("Only Q-in-Q interfaces may specify a service VLAN.")
})
def save(self, *args, **kwargs):
# Remove untagged VLAN assignment for non-802.1Q interfaces
@ -697,20 +728,6 @@ class Interface(ModularComponentModel, BaseInterface, CabledObjectModel, PathEnd
blank=True,
verbose_name=_('wireless LANs')
)
untagged_vlan = models.ForeignKey(
to='ipam.VLAN',
on_delete=models.SET_NULL,
related_name='interfaces_as_untagged',
null=True,
blank=True,
verbose_name=_('untagged VLAN')
)
tagged_vlans = models.ManyToManyField(
to='ipam.VLAN',
related_name='interfaces_as_tagged',
blank=True,
verbose_name=_('tagged VLANs')
)
vrf = models.ForeignKey(
to='ipam.VRF',
on_delete=models.SET_NULL,

7
netbox/dcim/models/mixins.py
View File

@ -35,10 +35,9 @@ class RenderConfigMixin(models.Model):
class CachedScopeMixin(models.Model):
"""
Mixin for adding a GFK Scope to a model that can point to a Region, SiteGroup, Site, or Location. Includes
cached fields for each to allow efficient filtering. Must do any appropriate validation in the clean method
as this does not have any as validation is generally model specific.
Mixin for adding a GenericForeignKey scope to a model that can point to a Region, SiteGroup, Site, or Location.
Includes cached fields for each to allow efficient filtering. Appropriate validation must be done in the clean()
method as this does not have any as validation is generally model-specific.
"""
scope_type = models.ForeignKey(
to='contenttypes.ContentType',

16
netbox/dcim/tables/devices.py
View File

@ -585,6 +585,10 @@ class BaseInterfaceTable(NetBoxTable):
orderable=False,
verbose_name=_('Tagged VLANs')
)
qinq_svlan = tables.Column(
verbose_name=_('Q-in-Q SVLAN'),
linkify=True
)
def value_ip_addresses(self, value):
return ",".join([str(obj.address) for obj in value.all()])
@ -635,11 +639,11 @@ class InterfaceTable(ModularDeviceComponentTable, BaseInterfaceTable, PathEndpoi
model = models.Interface
fields = (
'pk', 'id', 'name', 'device', 'module_bay', 'module', 'label', 'enabled', 'type', 'mgmt_only', 'mtu',
'speed', 'speed_formatted', 'duplex', 'mode', 'mac_address', 'wwn', 'poe_mode', 'poe_type', 'rf_role', 'rf_channel',
'rf_channel_frequency', 'rf_channel_width', 'tx_power', 'description', 'mark_connected', 'cable',
'cable_color', 'wireless_link', 'wireless_lans', 'link_peer', 'connection', 'tags', 'vdcs', 'vrf', 'l2vpn',
'tunnel', 'ip_addresses', 'fhrp_groups', 'untagged_vlan', 'tagged_vlans', 'inventory_items', 'created',
'last_updated',
'speed', 'speed_formatted', 'duplex', 'mode', 'mac_address', 'wwn', 'poe_mode', 'poe_type', 'rf_role',
'rf_channel', 'rf_channel_frequency', 'rf_channel_width', 'tx_power', 'description', 'mark_connected',
'cable', 'cable_color', 'wireless_link', 'wireless_lans', 'link_peer', 'connection', 'tags', 'vdcs', 'vrf',
'l2vpn', 'tunnel', 'ip_addresses', 'fhrp_groups', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan',
'inventory_items', 'created', 'last_updated',
)
default_columns = ('pk', 'name', 'device', 'label', 'enabled', 'type', 'description')
@ -676,7 +680,7 @@ class DeviceInterfaceTable(InterfaceTable):
'mgmt_only', 'mtu', 'mode', 'mac_address', 'wwn', 'rf_role', 'rf_channel', 'rf_channel_frequency',
'rf_channel_width', 'tx_power', 'description', 'mark_connected', 'cable', 'cable_color', 'wireless_link',
'wireless_lans', 'link_peer', 'connection', 'tags', 'vdcs', 'vrf', 'l2vpn', 'tunnel', 'ip_addresses',
'fhrp_groups', 'untagged_vlan', 'tagged_vlans', 'actions',
'fhrp_groups', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan', 'actions',
)
default_columns = (
'pk', 'name', 'label', 'enabled', 'type', 'parent', 'lag', 'mtu', 'mode', 'description', 'ip_addresses',

6
netbox/dcim/tests/test_api.py
View File

@ -7,6 +7,7 @@ from dcim.choices import *
from dcim.constants import *
from dcim.models import *
from extras.models import ConfigTemplate
from ipam.choices import VLANQinQRoleChoices
from ipam.models import ASN, RIR, VLAN, VRF
from netbox.api.serializers import GenericObjectSerializer
from tenancy.models import Tenant
@ -1618,6 +1619,7 @@ class InterfaceTest(Mixins.ComponentTraceMixin, APIViewTestCases.APIViewTestCase
VLAN(name='VLAN 1', vid=1),
VLAN(name='VLAN 2', vid=2),
VLAN(name='VLAN 3', vid=3),
VLAN(name='SVLAN 1', vid=1001, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
)
VLAN.objects.bulk_create(vlans)
@ -1676,18 +1678,22 @@ class InterfaceTest(Mixins.ComponentTraceMixin, APIViewTestCases.APIViewTestCase
'vdcs': [vdcs[1].pk],
'name': 'Interface 7',
'type': InterfaceTypeChoices.TYPE_80211A,
'mode': InterfaceModeChoices.MODE_Q_IN_Q,
'tx_power': 10,
'wireless_lans': [wireless_lans[0].pk, wireless_lans[1].pk],
'rf_channel': WirelessChannelChoices.CHANNEL_5G_32,
'qinq_svlan': vlans[3].pk,
},
{
'device': device.pk,
'vdcs': [vdcs[1].pk],
'name': 'Interface 8',
'type': InterfaceTypeChoices.TYPE_80211A,
'mode': InterfaceModeChoices.MODE_Q_IN_Q,
'tx_power': 10,
'wireless_lans': [wireless_lans[0].pk, wireless_lans[1].pk],
'rf_channel': "",
'qinq_svlan': vlans[3].pk,
},
]

29
netbox/dcim/tests/test_filtersets.py
View File

@ -4,7 +4,8 @@ from circuits.models import Circuit, CircuitTermination, CircuitType, Provider
from dcim.choices import *
from dcim.filtersets import *
from dcim.models import *
from ipam.models import ASN, IPAddress, RIR, VLANTranslationPolicy, VRF
from ipam.choices import VLANQinQRoleChoices
from ipam.models import ASN, IPAddress, RIR, VLAN, VLANTranslationPolicy, VRF
from netbox.choices import ColorChoices, WeightUnitChoices
from tenancy.models import Tenant, TenantGroup
from users.models import User
@ -3520,7 +3521,7 @@ class PowerOutletTestCase(TestCase, DeviceComponentFilterSetTests, ChangeLoggedF
class InterfaceTestCase(TestCase, DeviceComponentFilterSetTests, ChangeLoggedFilterSetTests):
queryset = Interface.objects.all()
filterset = InterfaceFilterSet
ignore_fields = ('tagged_vlans', 'untagged_vlan', 'vdcs')
ignore_fields = ('tagged_vlans', 'untagged_vlan', 'qinq_svlan', 'vdcs')
@classmethod
def setUpTestData(cls):
@ -3669,6 +3670,13 @@ class InterfaceTestCase(TestCase, DeviceComponentFilterSetTests, ChangeLoggedFil
)
VirtualDeviceContext.objects.bulk_create(vdcs)
vlans = (
VLAN(name='SVLAN 1', vid=1001, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
VLAN(name='SVLAN 2', vid=1002, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
VLAN(name='SVLAN 3', vid=1003, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
)
VLAN.objects.bulk_create(vlans)
vlan_translation_policies = (
VLANTranslationPolicy(name='Policy 1'),
VLANTranslationPolicy(name='Policy 2'),
@ -3753,6 +3761,8 @@ class InterfaceTestCase(TestCase, DeviceComponentFilterSetTests, ChangeLoggedFil
duplex='full',
poe_mode=InterfacePoEModeChoices.MODE_PD,
poe_type=InterfacePoETypeChoices.TYPE_2_8023AT,
mode=InterfaceModeChoices.MODE_Q_IN_Q,
qinq_svlan=vlans[0],
vlan_translation_policy=vlan_translation_policies[1],
),
Interface(
@ -3762,7 +3772,9 @@ class InterfaceTestCase(TestCase, DeviceComponentFilterSetTests, ChangeLoggedFil
type=InterfaceTypeChoices.TYPE_OTHER,
enabled=True,
mgmt_only=True,
tx_power=40
tx_power=40,
mode=InterfaceModeChoices.MODE_Q_IN_Q,
qinq_svlan=vlans[1]
),
Interface(
device=devices[4],
@ -3771,7 +3783,9 @@ class InterfaceTestCase(TestCase, DeviceComponentFilterSetTests, ChangeLoggedFil
type=InterfaceTypeChoices.TYPE_OTHER,
enabled=False,
mgmt_only=False,
tx_power=40
tx_power=40,
mode=InterfaceModeChoices.MODE_Q_IN_Q,
qinq_svlan=vlans[2]
),
Interface(
device=devices[4],
@ -4027,6 +4041,13 @@ class InterfaceTestCase(TestCase, DeviceComponentFilterSetTests, ChangeLoggedFil
params = {'vdc_identifier': vdc.values_list('identifier', flat=True)}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
def test_vlan(self):
vlan = VLAN.objects.filter(qinq_role=VLANQinQRoleChoices.ROLE_SERVICE).first()
params = {'vlan_id': vlan.pk}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
params = {'vlan': vlan.vid}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
def test_vlan_translation_policy(self):
vlan_translation_policies = VLANTranslationPolicy.objects.all()[:2]
params = {'vlan_translation_policy_id': [vlan_translation_policies[0].pk, vlan_translation_policies[1].pk]}

8
netbox/ipam/api/serializers_/nested.py
View File

@ -6,6 +6,7 @@ from ..field_serializers import IPAddressField
__all__ = (
'NestedIPAddressSerializer',
'NestedVLANSerializer',
)
@ -16,3 +17,10 @@ class NestedIPAddressSerializer(WritableNestedSerializer):
class Meta:
model = models.IPAddress
fields = ['id', 'url', 'display_url', 'display', 'family', 'address']
class NestedVLANSerializer(WritableNestedSerializer):
class Meta:
model = models.VLAN
fields = ['id', 'url', 'display', 'vid', 'name', 'description']

7
netbox/ipam/api/serializers_/vlans.py
View File

@ -11,6 +11,7 @@ from netbox.api.serializers import NetBoxModelSerializer
from tenancy.api.serializers_.tenants import TenantSerializer
from utilities.api import get_serializer_for_model
from vpn.api.serializers_.l2vpn import L2VPNTerminationSerializer
from .nested import NestedVLANSerializer
from .roles import RoleSerializer
__all__ = (
@ -64,6 +65,8 @@ class VLANSerializer(NetBoxModelSerializer):
tenant = TenantSerializer(nested=True, required=False, allow_null=True)
status = ChoiceField(choices=VLANStatusChoices, required=False)
role = RoleSerializer(nested=True, required=False, allow_null=True)
qinq_role = ChoiceField(choices=VLANQinQRoleChoices, required=False)
qinq_svlan = NestedVLANSerializer(required=False, allow_null=True, default=None)
l2vpn_termination = L2VPNTerminationSerializer(nested=True, read_only=True, allow_null=True)
# Related object counts
@ -73,8 +76,8 @@ class VLANSerializer(NetBoxModelSerializer):
model = VLAN
fields = [
'id', 'url', 'display_url', 'display', 'site', 'group', 'vid', 'name', 'tenant', 'status', 'role',
'description', 'comments', 'l2vpn_termination', 'tags', 'custom_fields', 'created', 'last_updated',
'prefix_count',
'description', 'qinq_role', 'qinq_svlan', 'comments', 'l2vpn_termination', 'tags', 'custom_fields',
'created', 'last_updated', 'prefix_count',
]
brief_fields = ('id', 'url', 'display', 'vid', 'name', 'description')

11
netbox/ipam/choices.py
View File

@ -157,6 +157,17 @@ class VLANStatusChoices(ChoiceSet):
]
class VLANQinQRoleChoices(ChoiceSet):
ROLE_SERVICE = 's-vlan'
ROLE_CUSTOMER = 'c-vlan'
CHOICES = [
(ROLE_SERVICE, _('Service'), 'blue'),
(ROLE_CUSTOMER, _('Customer'), 'orange'),
]
#
# Services
#

11
netbox/ipam/filtersets.py
View File

@ -1041,6 +1041,17 @@ class VLANFilterSet(NetBoxModelFilterSet, TenancyFilterSet):
queryset=VirtualMachine.objects.all(),
method='get_for_virtualmachine'
)
qinq_role = django_filters.MultipleChoiceFilter(
choices=VLANQinQRoleChoices
)
qinq_svlan_id = django_filters.ModelMultipleChoiceFilter(
queryset=VLAN.objects.all(),
label=_('Q-in-Q SVLAN (ID)'),
)
qinq_svlan_vid = MultiValueNumberFilter(
field_name='qinq_svlan__vid',
label=_('Q-in-Q SVLAN number (1-4094)'),
)
l2vpn_id = django_filters.ModelMultipleChoiceFilter(
field_name='l2vpn_terminations__l2vpn',
queryset=L2VPN.objects.all(),

16
netbox/ipam/forms/bulk_edit.py
View File

@ -527,15 +527,29 @@ class VLANBulkEditForm(NetBoxModelBulkEditForm):
max_length=200,
required=False
)
qinq_role = forms.ChoiceField(
label=_('Q-in-Q role'),
choices=add_blank_choice(VLANQinQRoleChoices),
required=False
)
qinq_svlan = DynamicModelChoiceField(
label=_('Q-in-Q SVLAN'),
queryset=VLAN.objects.all(),
required=False,
query_params={
'qinq_role': VLANQinQRoleChoices.ROLE_SERVICE,
}
)
comments = CommentField()
model = VLAN
fieldsets = (
FieldSet('status', 'role', 'tenant', 'description'),
FieldSet('qinq_role', 'qinq_svlan', name=_('Q-in-Q')),
FieldSet('region', 'site_group', 'site', 'group', name=_('Site & Group')),
)
nullable_fields = (
'site', 'group', 'tenant', 'role', 'description', 'comments',
'site', 'group', 'tenant', 'role', 'description', 'qinq_role', 'qinq_svlan', 'comments',
)

18
netbox/ipam/forms/bulk_import.py
View File

@ -461,10 +461,26 @@ class VLANImportForm(NetBoxModelImportForm):
to_field_name='name',
help_text=_('Functional role')
)
qinq_role = CSVChoiceField(
label=_('Q-in-Q role'),
choices=VLANStatusChoices,
required=False,
help_text=_('Operational status')
)
qinq_svlan = CSVModelChoiceField(
label=_('Q-in-Q SVLAN'),
queryset=VLAN.objects.all(),
required=False,
to_field_name='vid',
help_text=_("Service VLAN (for Q-in-Q/802.1ad customer VLANs)")
)
class Meta:
model = VLAN
fields = ('site', 'group', 'vid', 'name', 'tenant', 'status', 'role', 'description', 'comments', 'tags')
fields = (
'site', 'group', 'vid', 'name', 'tenant', 'status', 'role', 'description', 'qinq_role', 'qinq_svlan',
'comments', 'tags',
)
class VLANTranslationPolicyImportForm(NetBoxModelImportForm):

12
netbox/ipam/forms/filtersets.py
View File

@ -506,6 +506,7 @@ class VLANFilterForm(TenancyFilterForm, NetBoxModelFilterSetForm):
FieldSet('q', 'filter_id', 'tag'),
FieldSet('region_id', 'site_group_id', 'site_id', name=_('Location')),
FieldSet('group_id', 'status', 'role_id', 'vid', 'l2vpn_id', name=_('Attributes')),
FieldSet('qinq_role', 'qinq_svlan_id', name=_('Q-in-Q/802.1ad')),
FieldSet('tenant_group_id', 'tenant_id', name=_('Tenant')),
)
selector_fields = ('filter_id', 'q', 'site_id')
@ -552,6 +553,17 @@ class VLANFilterForm(TenancyFilterForm, NetBoxModelFilterSetForm):
required=False,
label=_('VLAN ID')
)
qinq_role = forms.MultipleChoiceField(
label=_('Q-in-Q role'),
choices=VLANQinQRoleChoices,
required=False
)
qinq_svlan_id = DynamicModelMultipleChoiceField(
queryset=VLAN.objects.all(),
required=False,
null_option='None',
label=_('Q-in-Q SVLAN')
)
l2vpn_id = DynamicModelMultipleChoiceField(
queryset=L2VPN.objects.all(),
required=False,

12
netbox/ipam/forms/model_forms.py
View File

@ -683,13 +683,21 @@ class VLANForm(TenancyForm, NetBoxModelForm):
queryset=Role.objects.all(),
required=False
)
qinq_svlan = DynamicModelChoiceField(
label=_('Q-in-Q SVLAN'),
queryset=VLAN.objects.all(),
required=False,
query_params={
'qinq_role': VLANQinQRoleChoices.ROLE_SERVICE,
}
)
comments = CommentField()
class Meta:
model = VLAN
fields = [
'site', 'group', 'vid', 'name', 'status', 'role', 'tenant_group', 'tenant', 'description', 'comments',
'tags',
'site', 'group', 'vid', 'name', 'status', 'role', 'tenant_group', 'tenant', 'qinq_role', 'qinq_svlan',
'description', 'comments', 'tags',
]

6
netbox/ipam/graphql/types.py
View File

@ -236,7 +236,7 @@ class ServiceTemplateType(NetBoxObjectType):
@strawberry_django.type(
models.VLAN,
fields='__all__',
exclude=('qinq_svlan',),
filters=VLANFilter
)
class VLANType(NetBoxObjectType):
@ -252,6 +252,10 @@ class VLANType(NetBoxObjectType):
interfaces_as_tagged: List[Annotated["InterfaceType", strawberry.lazy('dcim.graphql.types')]]
vminterfaces_as_tagged: List[Annotated["VMInterfaceType", strawberry.lazy('virtualization.graphql.types')]]
@strawberry_django.field
def qinq_svlan(self) -> Annotated["VLANType", strawberry.lazy('ipam.graphql.types')] | None:
return self.qinq_svlan
@strawberry_django.type(
models.VLANGroup,

30
netbox/ipam/migrations/0075_vlan_qinq.py Normal file
View File

@ -0,0 +1,30 @@
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('ipam', '0074_vlantranslationpolicy_vlantranslationrule'),
]
operations = [
migrations.AddField(
model_name='vlan',
name='qinq_role',
field=models.CharField(blank=True, max_length=50, null=True),
),
migrations.AddField(
model_name='vlan',
name='qinq_svlan',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='qinq_cvlans', to='ipam.vlan'),
),
migrations.AddConstraint(
model_name='vlan',
constraint=models.UniqueConstraint(fields=('qinq_svlan', 'vid'), name='ipam_vlan_unique_qinq_svlan_vid'),
),
migrations.AddConstraint(
model_name='vlan',
constraint=models.UniqueConstraint(fields=('qinq_svlan', 'name'), name='ipam_vlan_unique_qinq_svlan_name'),
),
]

40
netbox/ipam/models/vlans.py
View File

@ -204,6 +204,21 @@ class VLAN(PrimaryModel):
null=True,
help_text=_("The primary function of this VLAN")
)
qinq_svlan = models.ForeignKey(
to='self',
on_delete=models.PROTECT,
related_name='qinq_cvlans',
blank=True,
null=True
)
qinq_role = models.CharField(
verbose_name=_('Q-in-Q role'),
max_length=50,
choices=VLANQinQRoleChoices,
blank=True,
null=True,
help_text=_("Customer/service VLAN designation (for Q-in-Q/IEEE 802.1ad)")
)
l2vpn_terminations = GenericRelation(
to='vpn.L2VPNTermination',
content_type_field='assigned_object_type',
@ -214,7 +229,7 @@ class VLAN(PrimaryModel):
objects = VLANQuerySet.as_manager()
clone_fields = [
'site', 'group', 'tenant', 'status', 'role', 'description',
'site', 'group', 'tenant', 'status', 'role', 'description', 'qinq_role', 'qinq_svlan',
]
class Meta:
@ -228,6 +243,14 @@ class VLAN(PrimaryModel):
fields=('group', 'name'),
name='%(app_label)s_%(class)s_unique_group_name'
),
models.UniqueConstraint(
fields=('qinq_svlan', 'vid'),
name='%(app_label)s_%(class)s_unique_qinq_svlan_vid'
),
models.UniqueConstraint(
fields=('qinq_svlan', 'name'),
name='%(app_label)s_%(class)s_unique_qinq_svlan_name'
),
)
verbose_name = _('VLAN')
verbose_name_plural = _('VLANs')
@ -255,9 +278,24 @@ class VLAN(PrimaryModel):
).format(ranges=ranges_to_string(self.group.vid_ranges), group=self.group)
})
# Only Q-in-Q customer VLANs may be assigned to a service VLAN
if self.qinq_svlan and self.qinq_role != VLANQinQRoleChoices.ROLE_CUSTOMER:
raise ValidationError({
'qinq_svlan': _("Only Q-in-Q customer VLANs maybe assigned to a service VLAN.")
})
# A Q-in-Q customer VLAN must be assigned to a service VLAN
if self.qinq_role == VLANQinQRoleChoices.ROLE_CUSTOMER and not self.qinq_svlan:
raise ValidationError({
'qinq_role': _("A Q-in-Q customer VLAN must be assigned to a service VLAN.")
})
def get_status_color(self):
return VLANStatusChoices.colors.get(self.status)
def get_qinq_role_color(self):
return VLANQinQRoleChoices.colors.get(self.qinq_role)
def get_interfaces(self):
# Return all device interfaces assigned to this VLAN
return Interface.objects.filter(

9
netbox/ipam/tables/vlans.py
View File

@ -132,6 +132,13 @@ class VLANTable(TenancyColumnsMixin, NetBoxTable):
verbose_name=_('Role'),
linkify=True
)
qinq_role = columns.ChoiceFieldColumn(
verbose_name=_('Q-in-Q role')
)
qinq_svlan = tables.Column(
verbose_name=_('Q-in-Q SVLAN'),
linkify=True
)
l2vpn = tables.Column(
accessor=tables.A('l2vpn_termination__l2vpn'),
linkify=True,
@ -154,7 +161,7 @@ class VLANTable(TenancyColumnsMixin, NetBoxTable):
model = VLAN
fields = (
'pk', 'id', 'vid', 'name', 'site', 'group', 'prefixes', 'tenant', 'tenant_group', 'status', 'role',
'description', 'comments', 'tags', 'l2vpn', 'created', 'last_updated',
'qinq_role', 'qinq_svlan', 'description', 'comments', 'tags', 'l2vpn', 'created', 'last_updated',
)
default_columns = ('pk', 'vid', 'name', 'site', 'group', 'prefixes', 'tenant', 'status', 'role', 'description')
row_attrs = {

7
netbox/ipam/tests/test_api.py
View File

@ -980,6 +980,7 @@ class VLANTest(APIViewTestCases.APIViewTestCase):
VLAN(name='VLAN 1', vid=1, group=vlan_groups[0]),
VLAN(name='VLAN 2', vid=2, group=vlan_groups[0]),
VLAN(name='VLAN 3', vid=3, group=vlan_groups[0]),
VLAN(name='SVLAN 1', vid=1001, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
)
VLAN.objects.bulk_create(vlans)
@ -999,6 +1000,12 @@ class VLANTest(APIViewTestCases.APIViewTestCase):
'name': 'VLAN 6',
'group': vlan_groups[1].pk,
},
{
'vid': 2001,
'name': 'CVLAN 1',
'qinq_role': VLANQinQRoleChoices.ROLE_CUSTOMER,
'qinq_svlan': vlans[3].pk,
},
]
def test_delete_vlan_with_prefix(self):

24
netbox/ipam/tests/test_filtersets.py
View File

@ -1630,6 +1630,7 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTests):
Site(name='Site 4', slug='site-4', region=regions[0], group=site_groups[0]),
Site(name='Site 5', slug='site-5', region=regions[1], group=site_groups[1]),
Site(name='Site 6', slug='site-6', region=regions[2], group=site_groups[2]),
Site(name='Site 7', slug='site-7'),
)
Site.objects.bulk_create(sites)
@ -1785,9 +1786,21 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTests):
# Create one globally available VLAN
VLAN(vid=1000, name='Global VLAN'),
# Create some Q-in-Q service VLANs
VLAN(vid=2001, name='SVLAN 1', site=sites[6], qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
VLAN(vid=2002, name='SVLAN 2', site=sites[6], qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
VLAN(vid=2003, name='SVLAN 3', site=sites[6], qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
)
VLAN.objects.bulk_create(vlans)
# Create Q-in-Q customer VLANs
VLAN.objects.bulk_create([
VLAN(vid=3001, name='CVLAN 1', site=sites[6], qinq_svlan=vlans[29], qinq_role=VLANQinQRoleChoices.ROLE_CUSTOMER),
VLAN(vid=3002, name='CVLAN 2', site=sites[6], qinq_svlan=vlans[30], qinq_role=VLANQinQRoleChoices.ROLE_CUSTOMER),
VLAN(vid=3003, name='CVLAN 3', site=sites[6], qinq_svlan=vlans[31], qinq_role=VLANQinQRoleChoices.ROLE_CUSTOMER),
])
# Assign VLANs to device interfaces
interfaces[0].untagged_vlan = vlans[0]
interfaces[0].tagged_vlans.add(vlans[1])
@ -1898,6 +1911,17 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTests):
params = {'vminterface_id': vminterface_id}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
def test_qinq_role(self):
params = {'qinq_role': [VLANQinQRoleChoices.ROLE_SERVICE, VLANQinQRoleChoices.ROLE_CUSTOMER]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 6)
def test_qinq_svlan(self):
vlans = VLAN.objects.filter(qinq_role=VLANQinQRoleChoices.ROLE_SERVICE)[:2]
params = {'qinq_svlan_id': [vlans[0].pk, vlans[1].pk]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
params = {'qinq_svlan_vid': [vlans[0].vid, vlans[1].vid]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
class VLANTranslationPolicyTestCase(TestCase, ChangeLoggedFilterSetTests):
queryset = VLANTranslationPolicy.objects.all()

21
netbox/ipam/tests/test_models.py
View File

@ -586,3 +586,24 @@ class TestVLANGroup(TestCase):
vlangroup.vid_ranges = string_to_ranges('2-2')
vlangroup.full_clean()
vlangroup.save()
class TestVLAN(TestCase):
@classmethod
def setUpTestData(cls):
VLAN.objects.bulk_create((
VLAN(name='VLAN 1', vid=1, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
))
def test_qinq_role(self):
svlan = VLAN.objects.filter(qinq_role=VLANQinQRoleChoices.ROLE_SERVICE).first()
vlan = VLAN(
name='VLAN X',
vid=999,
qinq_role=VLANQinQRoleChoices.ROLE_SERVICE,
qinq_svlan=svlan
)
with self.assertRaises(ValidationError):
vlan.full_clean()

21
netbox/netbox/jobs.py
View File

@ -2,6 +2,7 @@ import logging
from abc import ABC, abstractmethod
from datetime import timedelta
from django.core.exceptions import ImproperlyConfigured
from django.utils.functional import classproperty
from django_pglocks import advisory_lock
from rq.timeouts import JobTimeoutException
@ -9,12 +10,30 @@ from rq.timeouts import JobTimeoutException
from core.choices import JobStatusChoices
from core.models import Job, ObjectType
from netbox.constants import ADVISORY_LOCK_KEYS
from netbox.registry import registry
__all__ = (
'JobRunner',
'system_job',
)
def system_job(interval):
"""
Decorator for registering a `JobRunner` class as system background job.
"""
if type(interval) is not int:
raise ImproperlyConfigured("System job interval must be an integer (minutes).")
def _wrapper(cls):
registry['system_jobs'][cls] = {
'interval': interval
}
return cls
return _wrapper
class JobRunner(ABC):
"""
Background Job helper class.
@ -129,7 +148,7 @@ class JobRunner(ABC):
if job:
# If the job parameters haven't changed, don't schedule a new job and keep the current schedule. Otherwise,
# delete the existing job and schedule a new job instead.
if (schedule_at and job.scheduled == schedule_at) and (job.interval == interval):
if (not schedule_at or job.scheduled == schedule_at) and (job.interval == interval):
return job
job.delete()

1
netbox/netbox/registry.py
View File

@ -30,6 +30,7 @@ registry = Registry({
'models': collections.defaultdict(set),
'plugins': dict(),
'search': dict(),
'system_jobs': dict(),
'tables': collections.defaultdict(dict),
'views': collections.defaultdict(dict),
'widgets': dict(),

5
netbox/netbox/tests/dummy_plugin/__init__.py
View File

@ -21,5 +21,10 @@ class DummyPluginConfig(PluginConfig):
'netbox.tests.dummy_plugin.events.process_events_queue'
]
def ready(self):
super().ready()
from . import jobs # noqa: F401
config = DummyPluginConfig

9
netbox/netbox/tests/dummy_plugin/jobs.py Normal file
View File

@ -0,0 +1,9 @@
from core.choices import JobIntervalChoices
from netbox.jobs import JobRunner, system_job
@system_job(interval=JobIntervalChoices.INTERVAL_HOURLY)
class DummySystemJob(JobRunner):
def run(self, *args, **kwargs):
pass

36
netbox/netbox/tests/test_jobs.py
View File

@ -90,6 +90,15 @@ class EnqueueTest(JobRunnerTestCase):
self.assertEqual(job1, job2)
self.assertEqual(TestJobRunner.get_jobs(instance).count(), 1)
def test_enqueue_once_twice_same_no_schedule_at(self):
instance = DataSource()
schedule_at = self.get_schedule_at()
job1 = TestJobRunner.enqueue_once(instance, schedule_at=schedule_at)
job2 = TestJobRunner.enqueue_once(instance)
self.assertEqual(job1, job2)
self.assertEqual(TestJobRunner.get_jobs(instance).count(), 1)
def test_enqueue_once_twice_different_schedule_at(self):
instance = DataSource()
job1 = TestJobRunner.enqueue_once(instance, schedule_at=self.get_schedule_at())
@ -127,3 +136,30 @@ class EnqueueTest(JobRunnerTestCase):
self.assertNotEqual(job1, job2)
self.assertRaises(Job.DoesNotExist, job1.refresh_from_db)
self.assertEqual(TestJobRunner.get_jobs(instance).count(), 1)
class SystemJobTest(JobRunnerTestCase):
"""
Test that system jobs can be scheduled.
General functionality already tested by `JobRunnerTest` and `EnqueueTest`.
"""
def test_scheduling(self):
# Can job be enqueued?
job = TestJobRunner.enqueue(schedule_at=self.get_schedule_at())
self.assertIsInstance(job, Job)
self.assertEqual(TestJobRunner.get_jobs().count(), 1)
# Can job be deleted again?
job.delete()
self.assertRaises(Job.DoesNotExist, job.refresh_from_db)
self.assertEqual(TestJobRunner.get_jobs().count(), 0)
def test_enqueue_once(self):
schedule_at = self.get_schedule_at()
job1 = TestJobRunner.enqueue_once(schedule_at=schedule_at)
job2 = TestJobRunner.enqueue_once(schedule_at=schedule_at)
self.assertEqual(job1, job2)
self.assertEqual(TestJobRunner.get_jobs().count(), 1)

9
netbox/netbox/tests/test_plugins.py
View File

@ -5,8 +5,10 @@ from django.core.exceptions import ImproperlyConfigured
from django.test import Client, TestCase, override_settings
from django.urls import reverse
from core.choices import JobIntervalChoices
from netbox.tests.dummy_plugin import config as dummy_config
from netbox.tests.dummy_plugin.data_backends import DummyBackend
from netbox.tests.dummy_plugin.jobs import DummySystemJob
from netbox.plugins.navigation import PluginMenu
from netbox.plugins.utils import get_plugin_config
from netbox.graphql.schema import Query
@ -130,6 +132,13 @@ class PluginTest(TestCase):
self.assertIn('dummy', registry['data_backends'])
self.assertIs(registry['data_backends']['dummy'], DummyBackend)
def test_system_jobs(self):
"""
Check registered system jobs.
"""
self.assertIn(DummySystemJob, registry['system_jobs'])
self.assertEqual(registry['system_jobs'][DummySystemJob]['interval'], JobIntervalChoices.INTERVAL_HOURLY)
def test_queues(self):
"""
Check that plugin queues are registered with the accurate name.

31
netbox/templates/ipam/vlan.html
View File

@ -62,6 +62,22 @@
<th scope="row">{% trans "Description" %}</th>
<td>{{ object.description|placeholder }}</td>
</tr>
<tr>
<th scope="row">{% trans "Q-in-Q Role" %}</th>
<td>
{% if object.qinq_role %}
{% badge object.get_qinq_role_display bg_color=object.get_qinq_role_color %}
{% else %}
{{ ''|placeholder }}
{% endif %}
</td>
</tr>
{% if object.qinq_role == 'c-vlan' %}
<tr>
<th scope="row">{% trans "Q-in-Q SVLAN" %}</th>
<td>{{ object.qinq_svlan|linkify|placeholder }}</td>
</tr>
{% endif %}
<tr>
<th scope="row">{% trans "L2VPN" %}</th>
<td>{{ object.l2vpn_termination.l2vpn|linkify|placeholder }}</td>
@ -92,6 +108,21 @@
</h2>
{% htmx_table 'ipam:prefix_list' vlan_id=object.pk %}
</div>
{% if object.qinq_role == 's-vlan' %}
<div class="card">
<h2 class="card-header">
{% trans "Customer VLANs" %}
{% if perms.ipam.add_vlan %}
<div class="card-actions">
<a href="{% url 'ipam:vlan_add' %}?qinq_role=c-vlan&qinq_svlan={{ object.pk }}" class="btn btn-ghost-primary btn-sm">
<i class="mdi mdi-plus-thick" aria-hidden="true"></i> {% trans "Add a VLAN" %}
</a>
</div>
{% endif %}
</h2>
{% htmx_table 'ipam:vlan_list' qinq_svlan_id=object.pk %}
</div>
{% endif %}
{% plugin_full_width_page object %}
</div>
</div>

8
netbox/templates/ipam/vlan_edit.html
View File

@ -17,6 +17,14 @@
{% render_field form.tags %}
</div>
<div class="field-group my-5">
<div class="row">
<h2 class="col-9 offset-3">{% trans "Q-in-Q (802.1ad)" %}</h2>
</div>
{% render_field form.qinq_role %}
{% render_field form.qinq_svlan %}
</div>
<div class="field-group my-5">
<div class="row">
<h2 class="col-9 offset-3">{% trans "Tenancy" %}</h2>

8
netbox/templates/virtualization/cluster.html
View File

@ -39,8 +39,12 @@
</td>
</tr>
<tr>
<th scope="row">{% trans "Site" %}</th>
<td>{{ object.site|linkify|placeholder }}</td>
<th scope="row">{% trans "Scope" %}</th>
{% if object.scope %}
<td>{{ object.scope|linkify }} ({% trans object.scope_type.name %})</td>
{% else %}
<td>{{ ''|placeholder }}</td>
{% endif %}
</tr>
</table>
</div>

7
netbox/virtualization/api/serializers_/virtualmachines.py
View File

@ -89,6 +89,7 @@ class VMInterfaceSerializer(NetBoxModelSerializer):
required=False,
many=True
)
qinq_svlan = VLANSerializer(nested=True, required=False, allow_null=True)
vlan_translation_policy = VLANTranslationPolicySerializer(nested=True, required=False, allow_null=True)
vrf = VRFSerializer(nested=True, required=False, allow_null=True)
l2vpn_termination = L2VPNTerminationSerializer(nested=True, read_only=True, allow_null=True)
@ -104,9 +105,9 @@ class VMInterfaceSerializer(NetBoxModelSerializer):
model = VMInterface
fields = [
'id', 'url', 'display_url', 'display', 'virtual_machine', 'name', 'enabled', 'parent', 'bridge', 'mtu',
'mac_address', 'description', 'mode', 'untagged_vlan', 'tagged_vlans', 'vrf', 'l2vpn_termination',
'tags', 'custom_fields', 'created', 'last_updated', 'count_ipaddresses', 'count_fhrp_groups',
'vlan_translation_policy',
'mac_address', 'description', 'mode', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan',
'vlan_translation_policy', 'vrf', 'l2vpn_termination', 'tags', 'custom_fields', 'created', 'last_updated',
'count_ipaddresses', 'count_fhrp_groups',
]
brief_fields = ('id', 'url', 'display', 'virtual_machine', 'name', 'description')

20
netbox/virtualization/forms/model_forms.py
View File

@ -7,6 +7,7 @@ from dcim.forms.common import InterfaceCommonForm
from dcim.forms.mixins import ScopedForm
from dcim.models import Device, DeviceRole, Platform, Rack, Region, Site, SiteGroup
from extras.models import ConfigTemplate
from ipam.choices import VLANQinQRoleChoices
from ipam.models import IPAddress, VLAN, VLANGroup, VLANTranslationPolicy, VRF
from netbox.forms import NetBoxModelForm
from tenancy.forms import TenancyForm
@ -334,6 +335,16 @@ class VMInterfaceForm(InterfaceCommonForm, VMComponentForm):
'available_on_virtualmachine': '$virtual_machine',
}
)
qinq_svlan = DynamicModelChoiceField(
queryset=VLAN.objects.all(),
required=False,
label=_('Q-in-Q Service VLAN'),
query_params={
'group_id': '$vlan_group',
'available_on_virtualmachine': '$virtual_machine',
'qinq_role': VLANQinQRoleChoices.ROLE_SERVICE,
}
)
vrf = DynamicModelChoiceField(
queryset=VRF.objects.all(),
required=False,
@ -350,17 +361,20 @@ class VMInterfaceForm(InterfaceCommonForm, VMComponentForm):
FieldSet('vrf', 'mac_address', name=_('Addressing')),
FieldSet('mtu', 'enabled', name=_('Operation')),
FieldSet('parent', 'bridge', name=_('Related Interfaces')),
FieldSet('mode', 'vlan_group', 'untagged_vlan', 'tagged_vlans', 'vlan_translation_policy', name=_('802.1Q Switching')),
FieldSet(
'mode', 'vlan_group', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan', 'vlan_translation_policy',
name=_('802.1Q Switching')
),
)
class Meta:
model = VMInterface
fields = [
'virtual_machine', 'name', 'parent', 'bridge', 'enabled', 'mac_address', 'mtu', 'description', 'mode',
'vlan_group', 'untagged_vlan', 'tagged_vlans', 'vrf', 'tags', 'vlan_translation_policy',
'vlan_group', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan', 'vlan_translation_policy', 'vrf', 'tags',
]
labels = {
'mode': '802.1Q Mode',
'mode': _('802.1Q Mode'),
}
widgets = {
'mode': HTMXSelect(),

1
netbox/virtualization/graphql/types.py
View File

@ -107,6 +107,7 @@ class VMInterfaceType(IPAddressesMixin, ComponentType):
bridge: Annotated["VMInterfaceType", strawberry.lazy('virtualization.graphql.types')] | None
untagged_vlan: Annotated["VLANType", strawberry.lazy('ipam.graphql.types')] | None
vrf: Annotated["VRFType", strawberry.lazy('ipam.graphql.types')] | None
qinq_svlan: Annotated["VLANType", strawberry.lazy('ipam.graphql.types')] | None
vlan_translation_policy: Annotated["VLANTranslationPolicyType", strawberry.lazy('ipam.graphql.types')] | None
tagged_vlans: List[Annotated["VLANType", strawberry.lazy('ipam.graphql.types')]]

2
netbox/virtualization/migrations/0042_vminterface_vlan_translation_policy.py
View File

@ -1,5 +1,3 @@
# Generated by Django 5.0.9 on 2024-10-11 19:45
import django.db.models.deletion
from django.db import migrations, models

28
netbox/virtualization/migrations/0043_qinq_svlan.py Normal file
View File

@ -0,0 +1,28 @@
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('ipam', '0075_vlan_qinq'),
('virtualization', '0042_vminterface_vlan_translation_policy'),
]
operations = [
migrations.AddField(
model_name='vminterface',
name='qinq_svlan',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(class)ss_svlan', to='ipam.vlan'),
),
migrations.AlterField(
model_name='vminterface',
name='tagged_vlans',
field=models.ManyToManyField(blank=True, related_name='%(class)ss_as_tagged', to='ipam.vlan'),
),
migrations.AlterField(
model_name='vminterface',
name='untagged_vlan',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(class)ss_as_untagged', to='ipam.vlan'),
),
]

2
netbox/virtualization/migrations/0043_cluster_scope.py → netbox/virtualization/migrations/0044_cluster_scope.py
View File

@ -20,7 +20,7 @@ class Migration(migrations.Migration):
dependencies = [
('contenttypes', '0002_remove_content_type_name'),
('virtualization', '0042_vminterface_vlan_translation_policy'),
('virtualization', '0043_qinq_svlan'),
]
operations = [

2
netbox/virtualization/migrations/0044_clusters_cached_relations.py → netbox/virtualization/migrations/0045_clusters_cached_relations.py
View File

@ -21,7 +21,7 @@ def populate_denormalized_fields(apps, schema_editor):
class Migration(migrations.Migration):
dependencies = [
('virtualization', '0043_cluster_scope'),
('virtualization', '0044_cluster_scope'),
]
operations = [

14
netbox/virtualization/models/virtualmachines.py
View File

@ -322,20 +322,6 @@ class VMInterface(ComponentModel, BaseInterface, TrackingModelMixin):
max_length=100,
blank=True
)
untagged_vlan = models.ForeignKey(
to='ipam.VLAN',
on_delete=models.SET_NULL,
related_name='vminterfaces_as_untagged',
null=True,
blank=True,
verbose_name=_('untagged VLAN')
)
tagged_vlans = models.ManyToManyField(
to='ipam.VLAN',
related_name='vminterfaces_as_tagged',
blank=True,
verbose_name=_('tagged VLANs')
)
ip_addresses = GenericRelation(
to='ipam.IPAddress',
content_type_field='assigned_object_type',

7
netbox/virtualization/tables/virtualmachines.py
View File

@ -151,8 +151,8 @@ class VMInterfaceTable(BaseInterfaceTable):
model = VMInterface
fields = (
'pk', 'id', 'name', 'virtual_machine', 'enabled', 'mac_address', 'mtu', 'mode', 'description', 'tags',
'vrf', 'l2vpn', 'tunnel', 'ip_addresses', 'fhrp_groups', 'untagged_vlan', 'tagged_vlans', 'created',
'last_updated',
'vrf', 'l2vpn', 'tunnel', 'ip_addresses', 'fhrp_groups', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan',
'created', 'last_updated',
)
default_columns = ('pk', 'name', 'virtual_machine', 'enabled', 'description')
@ -175,7 +175,8 @@ class VirtualMachineVMInterfaceTable(VMInterfaceTable):
model = VMInterface
fields = (
'pk', 'id', 'name', 'enabled', 'parent', 'bridge', 'mac_address', 'mtu', 'mode', 'description', 'tags',
'vrf', 'l2vpn', 'tunnel', 'ip_addresses', 'fhrp_groups', 'untagged_vlan', 'tagged_vlans', 'actions',
'vrf', 'l2vpn', 'tunnel', 'ip_addresses', 'fhrp_groups', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan',
'actions',
)
default_columns = ('pk', 'name', 'enabled', 'mac_address', 'mtu', 'mode', 'description', 'ip_addresses')
row_attrs = {

8
netbox/virtualization/tests/test_api.py
View File

@ -4,6 +4,7 @@ from rest_framework import status
from dcim.choices import InterfaceModeChoices
from dcim.models import Site
from extras.models import ConfigTemplate
from ipam.choices import VLANQinQRoleChoices
from ipam.models import VLAN, VRF
from utilities.testing import APITestCase, APIViewTestCases, create_test_device, create_test_virtualmachine
from virtualization.choices import *
@ -272,6 +273,7 @@ class VMInterfaceTest(APIViewTestCases.APIViewTestCase):
VLAN(name='VLAN 1', vid=1),
VLAN(name='VLAN 2', vid=2),
VLAN(name='VLAN 3', vid=3),
VLAN(name='SVLAN 1', vid=1001, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
)
VLAN.objects.bulk_create(vlans)
@ -309,6 +311,12 @@ class VMInterfaceTest(APIViewTestCases.APIViewTestCase):
'untagged_vlan': vlans[2].pk,
'vrf': vrfs[2].pk,
},
{
'virtual_machine': virtualmachine.pk,
'name': 'Interface 7',
'mode': InterfaceModeChoices.MODE_Q_IN_Q,
'qinq_svlan': vlans[3].pk,
},
]
def test_bulk_delete_child_interfaces(self):

24
netbox/virtualization/tests/test_filtersets.py
View File

@ -1,7 +1,9 @@
from django.test import TestCase
from dcim.choices import InterfaceModeChoices
from dcim.models import Device, DeviceRole, Platform, Region, Site, SiteGroup
from ipam.models import IPAddress, VLANTranslationPolicy, VRF
from ipam.choices import VLANQinQRoleChoices
from ipam.models import IPAddress, VLAN, VLANTranslationPolicy, VRF
from tenancy.models import Tenant, TenantGroup
from utilities.testing import ChangeLoggedFilterSetTests, create_test_device
from virtualization.choices import *
@ -530,7 +532,7 @@ class VirtualMachineTestCase(TestCase, ChangeLoggedFilterSetTests):
class VMInterfaceTestCase(TestCase, ChangeLoggedFilterSetTests):
queryset = VMInterface.objects.all()
filterset = VMInterfaceFilterSet
ignore_fields = ('tagged_vlans', 'untagged_vlan',)
ignore_fields = ('tagged_vlans', 'untagged_vlan', 'qinq_svlan')
@classmethod
def setUpTestData(cls):
@ -556,6 +558,13 @@ class VMInterfaceTestCase(TestCase, ChangeLoggedFilterSetTests):
)
VRF.objects.bulk_create(vrfs)
vlans = (
VLAN(name='SVLAN 1', vid=1001, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
VLAN(name='SVLAN 2', vid=1002, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
VLAN(name='SVLAN 3', vid=1003, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
)
VLAN.objects.bulk_create(vlans)
vms = (
VirtualMachine(name='Virtual Machine 1', cluster=clusters[0]),
VirtualMachine(name='Virtual Machine 2', cluster=clusters[1]),
@ -598,7 +607,9 @@ class VMInterfaceTestCase(TestCase, ChangeLoggedFilterSetTests):
mtu=300,
mac_address='00-00-00-00-00-03',
vrf=vrfs[2],
description='foobar3'
description='foobar3',
mode=InterfaceModeChoices.MODE_Q_IN_Q,
qinq_svlan=vlans[0]
),
)
VMInterface.objects.bulk_create(interfaces)
@ -669,6 +680,13 @@ class VMInterfaceTestCase(TestCase, ChangeLoggedFilterSetTests):
params = {'description': ['foobar1', 'foobar2']}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
def test_vlan(self):
vlan = VLAN.objects.filter(qinq_role=VLANQinQRoleChoices.ROLE_SERVICE).first()
params = {'vlan_id': vlan.pk}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
params = {'vlan': vlan.vid}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
def test_vlan_translation_policy(self):
vlan_translation_policies = VLANTranslationPolicy.objects.all()[:2]
params = {'vlan_translation_policy_id': [vlan_translation_policies[0].pk, vlan_translation_policies[1].pk]}

21
netbox/wireless/forms/model_forms.py
View File

@ -1,15 +1,13 @@
from django.contrib.contenttypes.models import ContentType
from django.forms import PasswordInput
from django.utils.translation import gettext_lazy as _
from dcim.models import Device, Interface, Location, Site
from dcim.forms.mixins import ScopedForm
from ipam.models import VLAN
from netbox.forms import NetBoxModelForm, ScopedForm
from netbox.forms import NetBoxModelForm
from tenancy.forms import TenancyForm
from utilities.forms.fields import CommentField, ContentTypeChoiceField, DynamicModelChoiceField, SlugField
from utilities.forms.fields import CommentField, DynamicModelChoiceField, SlugField
from utilities.forms.rendering import FieldSet, InlineFields
from utilities.forms.widgets import HTMXSelect
from wireless.constants import WIRELESSLAN_SCOPE_TYPES
from wireless.models import *
__all__ = (
@ -50,19 +48,6 @@ class WirelessLANForm(ScopedForm, TenancyForm, NetBoxModelForm):
selector=True,
label=_('VLAN')
)
scope_type = ContentTypeChoiceField(
queryset=ContentType.objects.filter(model__in=WIRELESSLAN_SCOPE_TYPES),
widget=HTMXSelect(),
required=False,
label=_('Scope type')
)
scope = DynamicModelChoiceField(
label=_('Scope'),
queryset=Site.objects.none(), # Initial queryset
required=False,
disabled=True,
selector=True
)
comments = CommentField()
fieldsets = (

6
netbox/wireless/migrations/0011_wirelesslan__location_wirelesslan__region_and_more.py
View File

@ -1,4 +1,4 @@
# Generated by Django 5.0.9 on 2024-10-28 17:44
# Generated by Django 5.0.9 on 2024-11-04 16:00
import django.db.models.deletion
from django.db import migrations, models
@ -8,7 +8,7 @@ class Migration(migrations.Migration):
dependencies = [
('contenttypes', '0002_remove_content_type_name'),
('dcim', '0194_charfield_null_choices'),
('dcim', '0196_qinq_svlan'),
('wireless', '0010_charfield_null_choices'),
]
@ -48,7 +48,7 @@ class Migration(migrations.Migration):
),
migrations.AddField(
model_name='wirelesslan',
name='_sitegroup',
name='_site_group',
field=models.ForeignKey(
blank=True,
null=True,