Closes #20285: Support for multi-protocol application services
This commit is contained in:
parent
b620be0f46
commit
891e40ffb4
|
|
@ -23,13 +23,15 @@ The parent object to which the application service is assigned. This must be one
|
|||
|
||||
A service or protocol name.
|
||||
|
||||
### Protocol
|
||||
### Port Assignments
|
||||
|
||||
The wire protocol on which the service runs. Choices include UDP, TCP, and SCTP.
|
||||
!!! note "Changed in NetBox v4.7"
|
||||
|
||||
### Ports
|
||||
Previously, a service defined a single `protocol` (UDP, TCP, or SCTP) shared by all of its `ports`. A service now defines a list of port assignments, where each assignment pairs an individual protocol with a port number. This allows a single service to combine multiple protocols — for example, a DNS service listening on both TCP/53 and UDP/53.
|
||||
|
||||
One or more numeric ports to which the service is bound. Multiple ports can be expressed using commas and/or hyphens. For example, `80,8001-8003` specifies ports 80, 8001, 8002, and 8003.
|
||||
Each port assignment comprises a wire protocol (UDP, TCP, or SCTP) and a numeric port. In the UI, selecting multiple protocols alongside one or more ports creates an assignment for every protocol/port combination.
|
||||
|
||||
The deprecated `protocol` and `ports` fields remain available in the REST and GraphQL APIs for backward compatibility. On read, `protocol` returns the single protocol shared by all assignments, or `null` when a service mixes protocols; `ports` returns the flattened list of port numbers. On write, supplying `protocol` and `ports` is translated into the equivalent port assignments (unless `port_assignments` is provided directly).
|
||||
|
||||
### IP Addresses
|
||||
|
||||
|
|
|
|||
|
|
@ -12,10 +12,12 @@ Application service templates can be used to instantiate [application services](
|
|||
|
||||
A service or protocol name.
|
||||
|
||||
### Protocol
|
||||
### Port Assignments
|
||||
|
||||
The wire protocol on which the service runs. Choices include UDP, TCP, and SCTP.
|
||||
!!! note "Changed in NetBox v4.7"
|
||||
|
||||
### Ports
|
||||
Previously, a service template defined a single `protocol` (UDP, TCP, or SCTP) shared by all of its `ports`. A template now defines a list of port assignments, where each assignment pairs an individual protocol with a port number. This allows a single template to combine multiple protocols — for example, both TCP/53 and UDP/53.
|
||||
|
||||
One or more numeric ports to which the service is bound. Multiple ports can be expressed using commas and/or hyphens. For example, `80,8001-8003` specifies ports 80, 8001, 8002, and 8003.
|
||||
Each port assignment comprises a wire protocol (UDP, TCP, or SCTP) and a numeric port. In the UI, selecting multiple protocols alongside one or more ports creates an assignment for every protocol/port combination.
|
||||
|
||||
The deprecated `protocol` and `ports` fields remain available in the REST and GraphQL APIs for backward compatibility, as described for [application services](./service.md).
|
||||
|
|
|
|||
|
|
@ -1,7 +1,8 @@
|
|||
from django.contrib.contenttypes.models import ContentType
|
||||
from rest_framework import serializers
|
||||
|
||||
from ipam.choices import *
|
||||
from ipam.constants import SERVICE_ASSIGNMENT_MODELS
|
||||
from ipam.constants import SERVICE_ASSIGNMENT_MODELS, SERVICE_PORT_MAX, SERVICE_PORT_MIN
|
||||
from ipam.models import IPAddress, Service, ServiceTemplate
|
||||
from netbox.api.fields import ChoiceField, ContentTypeField, SerializedPKRelatedField
|
||||
from netbox.api.gfk_fields import GFKSerializerField
|
||||
|
|
@ -10,25 +11,60 @@ from netbox.api.serializers import PrimaryModelSerializer
|
|||
from .ip import IPAddressSerializer
|
||||
|
||||
__all__ = (
|
||||
'ServicePortSerializer',
|
||||
'ServiceSerializer',
|
||||
'ServiceTemplateSerializer',
|
||||
)
|
||||
|
||||
|
||||
class ServiceTemplateSerializer(PrimaryModelSerializer):
|
||||
protocol = ChoiceField(choices=ServiceProtocolChoices, required=False)
|
||||
class ServicePortSerializer(serializers.Serializer):
|
||||
"""
|
||||
A single protocol/port assignment on a service.
|
||||
"""
|
||||
protocol = ChoiceField(choices=ServiceProtocolChoices)
|
||||
port = serializers.IntegerField(min_value=SERVICE_PORT_MIN, max_value=SERVICE_PORT_MAX)
|
||||
|
||||
|
||||
class ServiceSerializerBase(PrimaryModelSerializer):
|
||||
port_assignments = ServicePortSerializer(many=True, required=False)
|
||||
|
||||
# Deprecated fields, retained for backward compatibility. On read they are derived from
|
||||
# port_assignments (protocol is null when the service mixes protocols). On write they are
|
||||
# translated into port_assignments, unless port_assignments is provided explicitly.
|
||||
protocol = ChoiceField(choices=ServiceProtocolChoices, required=False, allow_null=True)
|
||||
ports = serializers.ListField(
|
||||
child=serializers.IntegerField(min_value=SERVICE_PORT_MIN, max_value=SERVICE_PORT_MAX),
|
||||
required=False
|
||||
)
|
||||
|
||||
def validate(self, data):
|
||||
# Translate the deprecated protocol/ports fields into port_assignments before model
|
||||
# validation runs (the model has no protocol/ports attributes to assign).
|
||||
if 'port_assignments' in data:
|
||||
data.pop('protocol', None)
|
||||
data.pop('ports', None)
|
||||
else:
|
||||
protocol = data.pop('protocol', None)
|
||||
ports = data.pop('ports', None)
|
||||
if protocol is not None and ports is not None:
|
||||
data['port_assignments'] = [
|
||||
{'protocol': protocol, 'port': port} for port in ports
|
||||
]
|
||||
return super().validate(data)
|
||||
|
||||
|
||||
class ServiceTemplateSerializer(ServiceSerializerBase):
|
||||
|
||||
class Meta:
|
||||
model = ServiceTemplate
|
||||
fields = [
|
||||
'id', 'url', 'display_url', 'display', 'name', 'protocol', 'ports', 'description', 'owner', 'comments',
|
||||
'tags', 'custom_fields', 'created', 'last_updated',
|
||||
'id', 'url', 'display_url', 'display', 'name', 'port_assignments', 'protocol', 'ports', 'description',
|
||||
'owner', 'comments', 'tags', 'custom_fields', 'created', 'last_updated',
|
||||
]
|
||||
brief_fields = ('id', 'url', 'display', 'name', 'protocol', 'ports', 'description')
|
||||
brief_fields = ('id', 'url', 'display', 'name', 'port_assignments', 'protocol', 'ports', 'description')
|
||||
|
||||
|
||||
class ServiceSerializer(PrimaryModelSerializer):
|
||||
protocol = ChoiceField(choices=ServiceProtocolChoices, required=False)
|
||||
class ServiceSerializer(ServiceSerializerBase):
|
||||
ipaddresses = SerializedPKRelatedField(
|
||||
queryset=IPAddress.objects.all(),
|
||||
serializer=IPAddressSerializer,
|
||||
|
|
@ -45,7 +81,7 @@ class ServiceSerializer(PrimaryModelSerializer):
|
|||
model = Service
|
||||
fields = [
|
||||
'id', 'url', 'display_url', 'display', 'parent_object_type', 'parent_object_id', 'parent', 'name',
|
||||
'protocol', 'ports', 'ipaddresses', 'description', 'owner', 'comments', 'tags', 'custom_fields',
|
||||
'created', 'last_updated',
|
||||
'port_assignments', 'protocol', 'ports', 'ipaddresses', 'description', 'owner', 'comments', 'tags',
|
||||
'custom_fields', 'created', 'last_updated',
|
||||
]
|
||||
brief_fields = ('id', 'url', 'display', 'name', 'protocol', 'ports', 'description')
|
||||
brief_fields = ('id', 'url', 'display', 'name', 'port_assignments', 'protocol', 'ports', 'description')
|
||||
|
|
|
|||
|
|
@ -22,7 +22,6 @@ from utilities.filters import (
|
|||
MultiValueCharFilter,
|
||||
MultiValueContentTypeFilter,
|
||||
MultiValueNumberFilter,
|
||||
NumericArrayFilter,
|
||||
TreeNodeMultipleChoiceFilter,
|
||||
)
|
||||
from utilities.filtersets import register_filterset
|
||||
|
|
@ -1206,16 +1205,43 @@ class VLANTranslationRuleFilterSet(NetBoxModelFilterSet):
|
|||
return queryset.filter(qs_filter)
|
||||
|
||||
|
||||
@register_filterset
|
||||
class ServiceTemplateFilterSet(PrimaryModelFilterSet):
|
||||
port = NumericArrayFilter(
|
||||
field_name='ports',
|
||||
lookup_expr='contains'
|
||||
class ServicePortFilterMixin(django_filters.FilterSet):
|
||||
"""
|
||||
Shared protocol/port filters operating on the port_assignments JSON field.
|
||||
"""
|
||||
protocol = django_filters.MultipleChoiceFilter(
|
||||
choices=ServiceProtocolChoices,
|
||||
method='filter_protocol',
|
||||
label=_('Protocol'),
|
||||
)
|
||||
port = MultiValueNumberFilter(
|
||||
method='filter_port',
|
||||
label=_('Port'),
|
||||
)
|
||||
|
||||
def filter_protocol(self, queryset, name, value):
|
||||
if not value:
|
||||
return queryset
|
||||
query = Q()
|
||||
for protocol in value:
|
||||
query |= Q(port_assignments__contains=[{'protocol': protocol}])
|
||||
return queryset.filter(query)
|
||||
|
||||
def filter_port(self, queryset, name, value):
|
||||
if not value:
|
||||
return queryset
|
||||
query = Q()
|
||||
for port in value:
|
||||
query |= Q(port_assignments__contains=[{'port': int(port)}])
|
||||
return queryset.filter(query)
|
||||
|
||||
|
||||
@register_filterset
|
||||
class ServiceTemplateFilterSet(ServicePortFilterMixin, PrimaryModelFilterSet):
|
||||
|
||||
class Meta:
|
||||
model = ServiceTemplate
|
||||
fields = ('id', 'name', 'protocol', 'description')
|
||||
fields = ('id', 'name', 'description')
|
||||
|
||||
def search(self, queryset, name, value):
|
||||
if not value.strip():
|
||||
|
|
@ -1228,7 +1254,7 @@ class ServiceTemplateFilterSet(PrimaryModelFilterSet):
|
|||
|
||||
|
||||
@register_filterset
|
||||
class ServiceFilterSet(ContactModelFilterSet, PrimaryModelFilterSet):
|
||||
class ServiceFilterSet(ServicePortFilterMixin, ContactModelFilterSet, PrimaryModelFilterSet):
|
||||
parent_object_type = MultiValueContentTypeFilter()
|
||||
device = MultiValueCharFilter(
|
||||
method='filter_device',
|
||||
|
|
@ -1271,14 +1297,9 @@ class ServiceFilterSet(ContactModelFilterSet, PrimaryModelFilterSet):
|
|||
to_field_name='address',
|
||||
label=_('IP address'),
|
||||
)
|
||||
port = NumericArrayFilter(
|
||||
field_name='ports',
|
||||
lookup_expr='contains'
|
||||
)
|
||||
|
||||
class Meta:
|
||||
model = Service
|
||||
fields = ('id', 'name', 'protocol', 'description', 'parent_object_type', 'parent_object_id')
|
||||
fields = ('id', 'name', 'description', 'parent_object_type', 'parent_object_id')
|
||||
|
||||
def search(self, queryset, name, value):
|
||||
if not value.strip():
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ from utilities.forms.fields import (
|
|||
CSVContentTypeField,
|
||||
CSVModelChoiceField,
|
||||
CSVModelMultipleChoiceField,
|
||||
NumericArrayField,
|
||||
NumericRangeArrayField,
|
||||
SlugField,
|
||||
)
|
||||
|
|
@ -586,19 +587,41 @@ class VLANTranslationRuleImportForm(NetBoxModelImportForm):
|
|||
fields = ('policy', 'local_vid', 'remote_vid')
|
||||
|
||||
|
||||
class ServiceTemplateImportForm(PrimaryModelImportForm):
|
||||
class ServicePortImportMixin:
|
||||
"""
|
||||
Compose a service's port_assignments from the deprecated protocol and ports import columns.
|
||||
"""
|
||||
def clean(self):
|
||||
super().clean()
|
||||
protocol = self.cleaned_data.get('protocol')
|
||||
ports = self.cleaned_data.get('ports')
|
||||
if protocol and ports:
|
||||
# Buffer the values for recomposition into port_assignments during model validation
|
||||
self.instance.protocol = protocol
|
||||
self.instance.ports = ports
|
||||
|
||||
|
||||
class ServiceTemplateImportForm(ServicePortImportMixin, PrimaryModelImportForm):
|
||||
protocol = CSVChoiceField(
|
||||
label=_('Protocol'),
|
||||
choices=ServiceProtocolChoices,
|
||||
help_text=_('IP protocol')
|
||||
)
|
||||
ports = NumericArrayField(
|
||||
label=_('Ports'),
|
||||
base_field=forms.IntegerField(
|
||||
min_value=SERVICE_PORT_MIN,
|
||||
max_value=SERVICE_PORT_MAX
|
||||
),
|
||||
help_text=_('Comma-separated list of one or more port numbers')
|
||||
)
|
||||
|
||||
class Meta:
|
||||
model = ServiceTemplate
|
||||
fields = ('name', 'protocol', 'ports', 'description', 'owner', 'comments', 'tags')
|
||||
|
||||
|
||||
class ServiceImportForm(PrimaryModelImportForm):
|
||||
class ServiceImportForm(ServicePortImportMixin, PrimaryModelImportForm):
|
||||
parent_object_type = CSVContentTypeField(
|
||||
queryset=ContentType.objects.filter(SERVICE_ASSIGNMENT_MODELS),
|
||||
required=True,
|
||||
|
|
@ -620,6 +643,14 @@ class ServiceImportForm(PrimaryModelImportForm):
|
|||
choices=ServiceProtocolChoices,
|
||||
help_text=_('IP protocol')
|
||||
)
|
||||
ports = NumericArrayField(
|
||||
label=_('Ports'),
|
||||
base_field=forms.IntegerField(
|
||||
min_value=SERVICE_PORT_MIN,
|
||||
max_value=SERVICE_PORT_MAX
|
||||
),
|
||||
help_text=_('Comma-separated list of one or more port numbers')
|
||||
)
|
||||
ipaddresses = CSVModelMultipleChoiceField(
|
||||
queryset=IPAddress.objects.all(),
|
||||
required=False,
|
||||
|
|
|
|||
|
|
@ -643,9 +643,9 @@ class ServiceTemplateFilterForm(PrimaryModelFilterSetForm):
|
|||
FieldSet('protocol', 'port', name=_('Attributes')),
|
||||
FieldSet('owner_group_id', 'owner_id', name=_('Ownership')),
|
||||
)
|
||||
protocol = forms.ChoiceField(
|
||||
protocol = forms.MultipleChoiceField(
|
||||
label=_('Protocol'),
|
||||
choices=add_blank_choice(ServiceProtocolChoices),
|
||||
choices=ServiceProtocolChoices,
|
||||
required=False
|
||||
)
|
||||
port = forms.IntegerField(
|
||||
|
|
|
|||
|
|
@ -800,10 +800,37 @@ class VLANTranslationRuleForm(NetBoxModelForm):
|
|||
]
|
||||
|
||||
|
||||
class ServiceTemplateForm(PrimaryModelForm):
|
||||
protocol = ChoiceField(
|
||||
label=_('Protocol'),
|
||||
class ServicePortAssignmentsMixin:
|
||||
"""
|
||||
Shared handling for composing a service's port_assignments from a set of protocols and ports. Each
|
||||
selected protocol is paired with every port (the cartesian product).
|
||||
"""
|
||||
def _init_port_assignments(self):
|
||||
# Populate the protocols/ports selections from an existing instance
|
||||
if self.instance and self.instance.pk:
|
||||
self.initial.setdefault(
|
||||
'protocols', sorted({a['protocol'] for a in self.instance.port_assignments})
|
||||
)
|
||||
self.initial.setdefault(
|
||||
'ports', sorted({a['port'] for a in self.instance.port_assignments})
|
||||
)
|
||||
|
||||
def _apply_port_assignments(self):
|
||||
protocols = self.cleaned_data.get('protocols')
|
||||
ports = self.cleaned_data.get('ports')
|
||||
if protocols and ports:
|
||||
self.instance.port_assignments = [
|
||||
{'protocol': protocol, 'port': port}
|
||||
for port in ports
|
||||
for protocol in protocols
|
||||
]
|
||||
|
||||
|
||||
class ServiceTemplateForm(ServicePortAssignmentsMixin, PrimaryModelForm):
|
||||
protocols = forms.MultipleChoiceField(
|
||||
choices=ServiceProtocolChoices,
|
||||
label=_('Protocols'),
|
||||
help_text=_("Each selected protocol is paired with every port below to form the port assignments.")
|
||||
)
|
||||
ports = NumericArrayField(
|
||||
label=_('Ports'),
|
||||
|
|
@ -815,19 +842,23 @@ class ServiceTemplateForm(PrimaryModelForm):
|
|||
)
|
||||
|
||||
fieldsets = (
|
||||
FieldSet('name', 'protocol', 'ports', 'description', 'tags', name=_('Application Service Template')),
|
||||
FieldSet('name', 'protocols', 'ports', 'description', 'tags', name=_('Application Service Template')),
|
||||
)
|
||||
|
||||
class Meta:
|
||||
model = ServiceTemplate
|
||||
fields = ('name', 'protocol', 'ports', 'description', 'owner', 'comments', 'tags')
|
||||
fields = ('name', 'description', 'owner', 'comments', 'tags')
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self._init_port_assignments()
|
||||
|
||||
def clean(self):
|
||||
super().clean()
|
||||
self._apply_port_assignments()
|
||||
|
||||
|
||||
class ServiceForm(GenericObjectFormMixin, PrimaryModelForm):
|
||||
protocol = ChoiceField(
|
||||
label=_('Protocol'),
|
||||
choices=ServiceProtocolChoices,
|
||||
)
|
||||
class ServiceForm(ServicePortAssignmentsMixin, GenericObjectFormMixin, PrimaryModelForm):
|
||||
parent = GenericObjectChoiceField(
|
||||
label=_('Parent'),
|
||||
content_type_queryset=ContentType.objects.filter(SERVICE_ASSIGNMENT_MODELS),
|
||||
|
|
@ -835,6 +866,11 @@ class ServiceForm(GenericObjectFormMixin, PrimaryModelForm):
|
|||
selector=True,
|
||||
hx_target_id='service',
|
||||
)
|
||||
protocols = forms.MultipleChoiceField(
|
||||
choices=ServiceProtocolChoices,
|
||||
label=_('Protocols'),
|
||||
help_text=_("Each selected protocol is paired with every port below to form the port assignments.")
|
||||
)
|
||||
ports = NumericArrayField(
|
||||
label=_('Ports'),
|
||||
base_field=forms.IntegerField(
|
||||
|
|
@ -852,7 +888,7 @@ class ServiceForm(GenericObjectFormMixin, PrimaryModelForm):
|
|||
fieldsets = (
|
||||
FieldSet(
|
||||
'parent', 'name',
|
||||
InlineFields('protocol', 'ports', label=_('Port(s)')),
|
||||
InlineFields('protocols', 'ports', label=_('Port(s)')),
|
||||
'ipaddresses', 'description', 'tags', name=_('Application Service'),
|
||||
html_id='service',
|
||||
),
|
||||
|
|
@ -861,11 +897,12 @@ class ServiceForm(GenericObjectFormMixin, PrimaryModelForm):
|
|||
class Meta:
|
||||
model = Service
|
||||
fields = [
|
||||
'name', 'protocol', 'ports', 'ipaddresses', 'description', 'owner', 'comments', 'tags',
|
||||
'name', 'ipaddresses', 'description', 'owner', 'comments', 'tags',
|
||||
]
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self._init_port_assignments()
|
||||
|
||||
# Filter the IP address selector to those belonging to the selected parent. The object subwidget is
|
||||
# named "parent_object_id", so the dynamic param references "$parent_object_id".
|
||||
|
|
@ -877,6 +914,10 @@ class ServiceForm(GenericObjectFormMixin, PrimaryModelForm):
|
|||
elif parent_model is FHRPGroup:
|
||||
self.fields['ipaddresses'].widget.add_query_params({'fhrpgroup_id': '$parent_object_id'})
|
||||
|
||||
def clean(self):
|
||||
super().clean()
|
||||
self._apply_port_assignments()
|
||||
|
||||
|
||||
class ServiceCreateForm(ServiceForm):
|
||||
service_template = DynamicModelChoiceField(
|
||||
|
|
@ -890,7 +931,7 @@ class ServiceCreateForm(ServiceForm):
|
|||
'parent',
|
||||
TabbedGroups(
|
||||
FieldSet('service_template', name=_('From Template')),
|
||||
FieldSet('name', 'protocol', 'ports', name=_('Custom')),
|
||||
FieldSet('name', 'protocols', 'ports', name=_('Custom')),
|
||||
),
|
||||
'ipaddresses', 'description', 'tags', name=_('Application Service'),
|
||||
html_id='service',
|
||||
|
|
@ -899,7 +940,7 @@ class ServiceCreateForm(ServiceForm):
|
|||
|
||||
class Meta(ServiceForm.Meta):
|
||||
fields = [
|
||||
'service_template', 'name', 'protocol', 'ports', 'ipaddresses', 'description',
|
||||
'service_template', 'name', 'ipaddresses', 'description',
|
||||
'comments', 'tags',
|
||||
]
|
||||
|
||||
|
|
@ -907,21 +948,21 @@ class ServiceCreateForm(ServiceForm):
|
|||
super().__init__(*args, **kwargs)
|
||||
|
||||
# Fields which may be populated from a ServiceTemplate are not required
|
||||
for field in ('name', 'protocol', 'ports'):
|
||||
for field in ('name', 'protocols', 'ports'):
|
||||
self.fields[field].required = False
|
||||
self.fields[field].widget.is_required = False
|
||||
|
||||
def clean(self):
|
||||
super().clean()
|
||||
if self.cleaned_data['service_template']:
|
||||
if self.cleaned_data.get('service_template'):
|
||||
# Create a new Service from the specified template
|
||||
service_template = self.cleaned_data['service_template']
|
||||
self.cleaned_data['name'] = service_template.name
|
||||
self.cleaned_data['protocol'] = service_template.protocol
|
||||
self.cleaned_data['ports'] = service_template.ports
|
||||
if not self.cleaned_data['description']:
|
||||
self.instance.port_assignments = service_template.port_assignments
|
||||
if not self.cleaned_data.get('description'):
|
||||
self.cleaned_data['description'] = service_template.description
|
||||
elif not all(self.cleaned_data[f] for f in ('name', 'protocol', 'ports')):
|
||||
elif not (self.cleaned_data.get('name') and self.cleaned_data.get('protocols') and
|
||||
self.cleaned_data.get('ports')):
|
||||
raise forms.ValidationError(
|
||||
_("Must specify name, protocol, and port(s) if not using an application service template.")
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,13 +1,10 @@
|
|||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING, Annotated
|
||||
|
||||
import strawberry
|
||||
import strawberry_django
|
||||
from strawberry_django import BaseFilterLookup
|
||||
from django.db.models import Q
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from netbox.graphql.filter_lookups import IntegerLookup
|
||||
|
||||
from .enums import *
|
||||
|
||||
__all__ = (
|
||||
|
|
@ -15,11 +12,26 @@ __all__ = (
|
|||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class ServiceFilterMixin:
|
||||
protocol: BaseFilterLookup[Annotated['ServiceProtocolEnum', strawberry.lazy('ipam.graphql.enums')]] | None = (
|
||||
strawberry_django.filter_field()
|
||||
)
|
||||
ports: Annotated['IntegerLookup', strawberry.lazy('netbox.graphql.filter_lookups')] | None = (
|
||||
strawberry_django.filter_field()
|
||||
)
|
||||
|
||||
@strawberry_django.filter_field()
|
||||
def protocol(
|
||||
self,
|
||||
value: list[Annotated['ServiceProtocolEnum', strawberry.lazy('ipam.graphql.enums')]],
|
||||
prefix,
|
||||
) -> Q:
|
||||
if not value:
|
||||
return Q()
|
||||
q = Q()
|
||||
for protocol in value:
|
||||
q |= Q(**{f'{prefix}port_assignments__contains': [{'protocol': protocol.value}]})
|
||||
return q
|
||||
|
||||
@strawberry_django.filter_field()
|
||||
def port(self, value: list[int], prefix) -> Q:
|
||||
if not value:
|
||||
return Q()
|
||||
q = Q()
|
||||
for port in value:
|
||||
q |= Q(**{f'{prefix}port_assignments__contains': [{'port': port}]})
|
||||
return q
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ from ipam import models
|
|||
from netbox.graphql.scalars import BigInt
|
||||
from netbox.graphql.types import BaseObjectType, NetBoxObjectType, OrganizationalObjectType, PrimaryObjectType
|
||||
|
||||
from .enums import ServiceProtocolEnum
|
||||
from .filters import *
|
||||
from .mixins import IPAddressesMixin
|
||||
|
||||
|
|
@ -249,9 +250,18 @@ class RouteTargetType(PrimaryObjectType):
|
|||
pagination=True
|
||||
)
|
||||
class ServiceType(ContactsMixin, PrimaryObjectType):
|
||||
ports: list[int]
|
||||
# port_assignments (a JSONField) is auto-exposed as a JSON scalar
|
||||
ipaddresses: list[Annotated['IPAddressType', strawberry.lazy('ipam.graphql.types')]]
|
||||
|
||||
# Deprecated backward-compatibility fields, derived from port_assignments
|
||||
@strawberry.field
|
||||
def ports(self) -> list[int]:
|
||||
return self.ports
|
||||
|
||||
@strawberry.field
|
||||
def protocol(self) -> ServiceProtocolEnum | None:
|
||||
return ServiceProtocolEnum(self.protocol) if self.protocol else None
|
||||
|
||||
@strawberry_django.field(prefetch_related='parent')
|
||||
def parent(self) -> Annotated[
|
||||
Annotated['DeviceType', strawberry.lazy('dcim.graphql.types')]
|
||||
|
|
@ -269,7 +279,16 @@ class ServiceType(ContactsMixin, PrimaryObjectType):
|
|||
pagination=True
|
||||
)
|
||||
class ServiceTemplateType(PrimaryObjectType):
|
||||
ports: list[int]
|
||||
# port_assignments (a JSONField) is auto-exposed as a JSON scalar
|
||||
|
||||
# Deprecated backward-compatibility fields, derived from port_assignments
|
||||
@strawberry.field
|
||||
def ports(self) -> list[int]:
|
||||
return self.ports
|
||||
|
||||
@strawberry.field
|
||||
def protocol(self) -> ServiceProtocolEnum | None:
|
||||
return ServiceProtocolEnum(self.protocol) if self.protocol else None
|
||||
|
||||
|
||||
@strawberry_django.type(
|
||||
|
|
|
|||
|
|
@ -0,0 +1,87 @@
|
|||
# Generated by Django 6.0.6 on 2026-07-01 21:08
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
def populate_port_assignments(apps, schema_editor):
|
||||
"""
|
||||
Populate port_assignments from the legacy protocol and ports fields.
|
||||
"""
|
||||
for model_name in ('Service', 'ServiceTemplate'):
|
||||
model = apps.get_model('ipam', model_name)
|
||||
for obj in model.objects.all().iterator():
|
||||
obj.port_assignments = [
|
||||
{'protocol': obj.protocol, 'port': port}
|
||||
for port in (obj.ports or [])
|
||||
]
|
||||
obj.save(update_fields=['port_assignments'])
|
||||
|
||||
|
||||
def restore_protocol_ports(apps, schema_editor):
|
||||
"""
|
||||
Reverse: reconstruct the legacy protocol and ports fields from port_assignments. This is lossy
|
||||
for services which mix protocols (only the first assignment's protocol is retained).
|
||||
"""
|
||||
for model_name in ('Service', 'ServiceTemplate'):
|
||||
model = apps.get_model('ipam', model_name)
|
||||
for obj in model.objects.all().iterator():
|
||||
assignments = obj.port_assignments or []
|
||||
obj.protocol = assignments[0]['protocol'] if assignments else ''
|
||||
obj.ports = sorted({assignment['port'] for assignment in assignments})
|
||||
obj.save(update_fields=['protocol', 'ports'])
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('contenttypes', '0002_remove_content_type_name'),
|
||||
('extras', '0140_imageattachment_image_size'),
|
||||
('ipam', '0093_denormalization_triggers'),
|
||||
('users', '0016_default_ordering_indexes'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
# Add the new field to both models
|
||||
migrations.AddField(
|
||||
model_name='service',
|
||||
name='port_assignments',
|
||||
field=models.JSONField(blank=True, default=list),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='servicetemplate',
|
||||
name='port_assignments',
|
||||
field=models.JSONField(blank=True, default=list),
|
||||
),
|
||||
# Copy existing protocol/ports data into port_assignments
|
||||
migrations.RunPython(populate_port_assignments, restore_protocol_ports),
|
||||
# Remove the legacy protocol/ports fields and their index
|
||||
migrations.RemoveIndex(
|
||||
model_name='service',
|
||||
name='ipam_servic_protoco_e2901d_idx',
|
||||
),
|
||||
migrations.RemoveField(
|
||||
model_name='service',
|
||||
name='protocol',
|
||||
),
|
||||
migrations.RemoveField(
|
||||
model_name='service',
|
||||
name='ports',
|
||||
),
|
||||
migrations.RemoveField(
|
||||
model_name='servicetemplate',
|
||||
name='protocol',
|
||||
),
|
||||
migrations.RemoveField(
|
||||
model_name='servicetemplate',
|
||||
name='ports',
|
||||
),
|
||||
# Update ordering and index to reflect the removal of protocol
|
||||
migrations.AddIndex(
|
||||
model_name='service',
|
||||
index=models.Index(fields=['_ports_lowest', 'id'], name='ipam_servic__ports__10a2cc_idx'),
|
||||
),
|
||||
migrations.AlterModelOptions(
|
||||
name='service',
|
||||
options={'ordering': ('_ports_lowest', 'id')},
|
||||
),
|
||||
]
|
||||
|
|
@ -1,6 +1,5 @@
|
|||
from django.contrib.contenttypes.fields import GenericForeignKey
|
||||
from django.contrib.postgres.fields import ArrayField
|
||||
from django.core.validators import MaxValueValidator, MinValueValidator
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.db import models
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
|
||||
|
|
@ -17,19 +16,11 @@ __all__ = (
|
|||
|
||||
|
||||
class ServiceBase(models.Model):
|
||||
protocol = models.CharField(
|
||||
verbose_name=_('protocol'),
|
||||
max_length=50,
|
||||
choices=ServiceProtocolChoices
|
||||
)
|
||||
ports = ArrayField(
|
||||
base_field=models.PositiveIntegerField(
|
||||
validators=[
|
||||
MinValueValidator(SERVICE_PORT_MIN),
|
||||
MaxValueValidator(SERVICE_PORT_MAX)
|
||||
]
|
||||
),
|
||||
verbose_name=_('port numbers')
|
||||
port_assignments = models.JSONField(
|
||||
verbose_name=_('port assignments'),
|
||||
default=list,
|
||||
blank=True,
|
||||
help_text=_('A list of protocol/port assignments, e.g. [{"protocol": "tcp", "port": 53}]')
|
||||
)
|
||||
_ports_lowest = models.PositiveIntegerField(
|
||||
null=True,
|
||||
|
|
@ -40,19 +31,115 @@ class ServiceBase(models.Model):
|
|||
abstract = True
|
||||
|
||||
def save(self, *args, **kwargs):
|
||||
# Compose port_assignments from any deprecated protocol/ports values assigned directly
|
||||
self._recompose_port_assignments()
|
||||
# On saving find the smallest port and save for default ordering
|
||||
self._ports_lowest = min(self.ports) if self.ports else None
|
||||
self._ports_lowest = min(
|
||||
(assignment['port'] for assignment in self.port_assignments), default=None
|
||||
)
|
||||
update_fields = kwargs.get('update_fields')
|
||||
if update_fields is not None and '_ports_lowest' not in update_fields:
|
||||
kwargs['update_fields'] = list(update_fields) + ['_ports_lowest']
|
||||
super().save(*args, **kwargs)
|
||||
|
||||
def __str__(self):
|
||||
return f'{self.name} ({self.get_protocol_display()}/{self.port_list})'
|
||||
return f'{self.name} ({self.port_list})'
|
||||
|
||||
def clean(self):
|
||||
super().clean()
|
||||
|
||||
# Compose port_assignments from any deprecated protocol/ports values assigned directly
|
||||
self._recompose_port_assignments()
|
||||
|
||||
if not self.port_assignments:
|
||||
raise ValidationError({
|
||||
'port_assignments': _("At least one protocol/port assignment must be defined.")
|
||||
})
|
||||
|
||||
valid_protocols = ServiceProtocolChoices.values()
|
||||
for assignment in self.port_assignments:
|
||||
if not isinstance(assignment, dict) or set(assignment) != {'protocol', 'port'}:
|
||||
raise ValidationError({
|
||||
'port_assignments': _("Each assignment must define exactly a protocol and a port.")
|
||||
})
|
||||
if assignment['protocol'] not in valid_protocols:
|
||||
raise ValidationError({
|
||||
'port_assignments': _("Invalid protocol: {protocol}").format(protocol=assignment['protocol'])
|
||||
})
|
||||
port = assignment['port']
|
||||
if not isinstance(port, int) or not SERVICE_PORT_MIN <= port <= SERVICE_PORT_MAX:
|
||||
raise ValidationError({
|
||||
'port_assignments': _("Invalid port number: {port}").format(port=port)
|
||||
})
|
||||
|
||||
@property
|
||||
def protocol(self):
|
||||
"""
|
||||
Deprecated backward-compatibility accessor. Returns the single protocol shared by all port
|
||||
assignments, or None if the service mixes protocols (or has no assignments).
|
||||
"""
|
||||
protocols = {assignment['protocol'] for assignment in self.port_assignments}
|
||||
return protocols.pop() if len(protocols) == 1 else None
|
||||
|
||||
@protocol.setter
|
||||
def protocol(self, value):
|
||||
# Deprecated: buffer the value for recomposition into port_assignments (see save()/clean())
|
||||
self._legacy_protocol = value
|
||||
|
||||
@property
|
||||
def ports(self):
|
||||
"""
|
||||
Deprecated backward-compatibility accessor. Returns a sorted list of all assigned port numbers.
|
||||
"""
|
||||
return sorted({assignment['port'] for assignment in self.port_assignments})
|
||||
|
||||
@ports.setter
|
||||
def ports(self, value):
|
||||
# Deprecated: buffer the value for recomposition into port_assignments (see save()/clean())
|
||||
self._legacy_ports = list(value) if value else []
|
||||
|
||||
def _recompose_port_assignments(self):
|
||||
"""
|
||||
If deprecated protocol and/or ports values were assigned directly (e.g. via bulk edit),
|
||||
rebuild port_assignments as the cartesian product of the effective protocols and ports.
|
||||
Missing values fall back to those already present in port_assignments.
|
||||
"""
|
||||
has_protocol = hasattr(self, '_legacy_protocol')
|
||||
has_ports = hasattr(self, '_legacy_ports')
|
||||
if not (has_protocol or has_ports):
|
||||
return
|
||||
|
||||
if has_protocol and self._legacy_protocol:
|
||||
protocols = [self._legacy_protocol]
|
||||
else:
|
||||
protocols = sorted({assignment['protocol'] for assignment in self.port_assignments})
|
||||
if has_ports:
|
||||
ports = self._legacy_ports
|
||||
else:
|
||||
ports = sorted({assignment['port'] for assignment in self.port_assignments})
|
||||
|
||||
self.port_assignments = [
|
||||
{'protocol': protocol, 'port': port}
|
||||
for port in ports
|
||||
for protocol in protocols
|
||||
]
|
||||
|
||||
if has_protocol:
|
||||
del self._legacy_protocol
|
||||
if has_ports:
|
||||
del self._legacy_ports
|
||||
|
||||
@property
|
||||
def port_list(self):
|
||||
return array_to_string(self.ports)
|
||||
# Group ports by protocol for compact display, e.g. "TCP/80, 443; UDP/53"
|
||||
protocol_labels = dict(ServiceProtocolChoices)
|
||||
grouped = {}
|
||||
for assignment in self.port_assignments:
|
||||
grouped.setdefault(assignment['protocol'], []).append(assignment['port'])
|
||||
return '; '.join(
|
||||
f'{protocol_labels.get(protocol, protocol)}/{array_to_string(sorted(ports))}'
|
||||
for protocol, ports in grouped.items()
|
||||
)
|
||||
|
||||
|
||||
class ServiceTemplate(ServiceBase, PrimaryModel):
|
||||
|
|
@ -99,14 +186,14 @@ class Service(ContactsMixin, ServiceBase, PrimaryModel):
|
|||
)
|
||||
|
||||
clone_fields = (
|
||||
'protocol', 'ports', 'description', 'parent', 'ipaddresses',
|
||||
'port_assignments', 'description', 'parent', 'ipaddresses',
|
||||
)
|
||||
|
||||
class Meta:
|
||||
indexes = (
|
||||
models.Index(fields=('protocol', '_ports_lowest', 'id')), # Default ordering
|
||||
models.Index(fields=('_ports_lowest', 'id')), # Default ordering
|
||||
models.Index(fields=('parent_object_type', 'parent_object_id')),
|
||||
)
|
||||
ordering = ('protocol', '_ports_lowest', 'id')
|
||||
ordering = ('_ports_lowest', 'id')
|
||||
verbose_name = _('application service')
|
||||
verbose_name_plural = _('application services')
|
||||
|
|
|
|||
|
|
@ -16,10 +16,16 @@ class ServiceTemplateTable(PrimaryModelTable):
|
|||
verbose_name=_('Name'),
|
||||
linkify=True
|
||||
)
|
||||
# Deprecated: derived from port_assignments (empty when the service mixes protocols)
|
||||
protocol = tables.Column(
|
||||
verbose_name=_('Protocol'),
|
||||
accessor=tables.A('protocol'),
|
||||
orderable=False,
|
||||
)
|
||||
ports = tables.Column(
|
||||
verbose_name=_('Ports'),
|
||||
accessor=tables.A('port_list'),
|
||||
order_by=tables.A('ports'),
|
||||
order_by=tables.A('_ports_lowest'),
|
||||
)
|
||||
tags = columns.TagColumn(
|
||||
url_name='ipam:servicetemplate_list'
|
||||
|
|
@ -30,7 +36,7 @@ class ServiceTemplateTable(PrimaryModelTable):
|
|||
fields = (
|
||||
'pk', 'id', 'name', 'protocol', 'ports', 'description', 'comments', 'tags', 'created', 'last_updated',
|
||||
)
|
||||
default_columns = ('pk', 'name', 'protocol', 'ports', 'description')
|
||||
default_columns = ('pk', 'name', 'ports', 'description')
|
||||
|
||||
|
||||
class ServiceTable(ContactsColumnMixin, PrimaryModelTable):
|
||||
|
|
@ -43,10 +49,16 @@ class ServiceTable(ContactsColumnMixin, PrimaryModelTable):
|
|||
linkify=True,
|
||||
order_by=('device', 'virtual_machine')
|
||||
)
|
||||
# Deprecated: derived from port_assignments (empty when the service mixes protocols)
|
||||
protocol = tables.Column(
|
||||
verbose_name=_('Protocol'),
|
||||
accessor=tables.A('protocol'),
|
||||
orderable=False,
|
||||
)
|
||||
ports = tables.Column(
|
||||
verbose_name=_('Ports'),
|
||||
accessor=tables.A('port_list'),
|
||||
order_by=tables.A('ports'),
|
||||
order_by=tables.A('_ports_lowest'),
|
||||
)
|
||||
tags = columns.TagColumn(
|
||||
url_name='ipam:service_list'
|
||||
|
|
@ -58,4 +70,4 @@ class ServiceTable(ContactsColumnMixin, PrimaryModelTable):
|
|||
'pk', 'id', 'name', 'parent', 'protocol', 'ports', 'ipaddresses', 'description', 'contacts', 'comments',
|
||||
'tags', 'created', 'last_updated',
|
||||
)
|
||||
default_columns = ('pk', 'name', 'parent', 'protocol', 'ports', 'description')
|
||||
default_columns = ('pk', 'name', 'parent', 'ports', 'description')
|
||||
|
|
|
|||
|
|
@ -1418,7 +1418,7 @@ class VLANTranslationRuleTestCase(APIViewTestCases.APIViewTestCase):
|
|||
|
||||
class ServiceTemplateTestCase(APIViewTestCases.APIViewTestCase):
|
||||
model = ServiceTemplate
|
||||
brief_fields = ['description', 'display', 'id', 'name', 'ports', 'protocol', 'url']
|
||||
brief_fields = ['description', 'display', 'id', 'name', 'port_assignments', 'ports', 'protocol', 'url']
|
||||
bulk_update_data = {
|
||||
'description': 'New description',
|
||||
}
|
||||
|
|
@ -1427,9 +1427,18 @@ class ServiceTemplateTestCase(APIViewTestCases.APIViewTestCase):
|
|||
@classmethod
|
||||
def setUpTestData(cls):
|
||||
service_templates = (
|
||||
ServiceTemplate(name='Service Template 1', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[1, 2]),
|
||||
ServiceTemplate(name='Service Template 2', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[3, 4]),
|
||||
ServiceTemplate(name='Service Template 3', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[5, 6]),
|
||||
ServiceTemplate(name='Service Template 1', port_assignments=[
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 1},
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 2},
|
||||
]),
|
||||
ServiceTemplate(name='Service Template 2', port_assignments=[
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 3},
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 4},
|
||||
]),
|
||||
ServiceTemplate(name='Service Template 3', port_assignments=[
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 5},
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 6},
|
||||
]),
|
||||
)
|
||||
ServiceTemplate.objects.bulk_create(service_templates)
|
||||
|
||||
|
|
@ -1454,7 +1463,7 @@ class ServiceTemplateTestCase(APIViewTestCases.APIViewTestCase):
|
|||
|
||||
class ServiceTestCase(APIViewTestCases.APIViewTestCase):
|
||||
model = Service
|
||||
brief_fields = ['description', 'display', 'id', 'name', 'ports', 'protocol', 'url']
|
||||
brief_fields = ['description', 'display', 'id', 'name', 'port_assignments', 'ports', 'protocol', 'url']
|
||||
bulk_update_data = {
|
||||
'description': 'New description',
|
||||
}
|
||||
|
|
@ -1474,9 +1483,15 @@ class ServiceTestCase(APIViewTestCases.APIViewTestCase):
|
|||
Device.objects.bulk_create(devices)
|
||||
|
||||
services = (
|
||||
Service(parent=devices[0], name='Service 1', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[1]),
|
||||
Service(parent=devices[0], name='Service 2', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[2]),
|
||||
Service(parent=devices[0], name='Service 3', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[3]),
|
||||
Service(parent=devices[0], name='Service 1', port_assignments=[
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 1},
|
||||
]),
|
||||
Service(parent=devices[0], name='Service 2', port_assignments=[
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 2},
|
||||
]),
|
||||
Service(parent=devices[0], name='Service 3', port_assignments=[
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 3},
|
||||
]),
|
||||
)
|
||||
Service.objects.bulk_create(services)
|
||||
|
||||
|
|
@ -1504,6 +1519,40 @@ class ServiceTestCase(APIViewTestCases.APIViewTestCase):
|
|||
},
|
||||
]
|
||||
|
||||
def test_create_with_port_assignments(self):
|
||||
"""
|
||||
A service may be created via the new port_assignments field, mixing protocols on a port.
|
||||
"""
|
||||
self.add_permissions('ipam.add_service')
|
||||
device = Device.objects.first()
|
||||
data = {
|
||||
'parent_object_id': device.pk,
|
||||
'parent_object_type': 'dcim.device',
|
||||
'name': 'DNS',
|
||||
'port_assignments': [
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 53},
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_UDP, 'port': 53},
|
||||
],
|
||||
}
|
||||
response = self.client.post(self._get_list_url(), data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
||||
service = Service.objects.get(pk=response.data['id'])
|
||||
self.assertEqual(len(service.port_assignments), 2)
|
||||
# Deprecated protocol accessor is null for a mixed-protocol service; ports is flattened
|
||||
self.assertIsNone(response.data['protocol'])
|
||||
self.assertEqual(response.data['ports'], [53])
|
||||
|
||||
def test_backward_compatible_protocol_read(self):
|
||||
"""
|
||||
A uniform-protocol service exposes the deprecated protocol field as a single value.
|
||||
"""
|
||||
self.add_permissions('ipam.view_service')
|
||||
service = Service.objects.get(name='Service 1')
|
||||
url = self._get_detail_url(service)
|
||||
response = self.client.get(url, **self.header)
|
||||
self.assertEqual(response.data['protocol']['value'], ServiceProtocolChoices.PROTOCOL_TCP)
|
||||
self.assertEqual(response.data['ports'], [1])
|
||||
|
||||
|
||||
class NestedObjectPermissionAPITest(APITestCase):
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -2358,43 +2358,37 @@ class VLANTranslationRuleTestCase(TestCase, ChangeLoggedFilterSetTests):
|
|||
class ServiceTemplateTestCase(TestCase, ChangeLoggedFilterSetTests):
|
||||
queryset = ServiceTemplate.objects.all()
|
||||
filterset = ServiceTemplateFilterSet
|
||||
ignore_fields = ('ports',)
|
||||
ignore_fields = ('port_assignments',)
|
||||
|
||||
@classmethod
|
||||
def setUpTestData(cls):
|
||||
service_templates = (
|
||||
ServiceTemplate(
|
||||
name='Service Template 1',
|
||||
protocol=ServiceProtocolChoices.PROTOCOL_TCP,
|
||||
ports=[1001],
|
||||
port_assignments=[{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 1001}],
|
||||
description='foobar1'
|
||||
),
|
||||
ServiceTemplate(
|
||||
name='Service Template 2',
|
||||
protocol=ServiceProtocolChoices.PROTOCOL_TCP,
|
||||
ports=[1002],
|
||||
port_assignments=[{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 1002}],
|
||||
description='foobar2'
|
||||
),
|
||||
ServiceTemplate(
|
||||
name='Service Template 3',
|
||||
protocol=ServiceProtocolChoices.PROTOCOL_UDP,
|
||||
ports=[1003],
|
||||
port_assignments=[{'protocol': ServiceProtocolChoices.PROTOCOL_UDP, 'port': 1003}],
|
||||
description='foobar3'
|
||||
),
|
||||
ServiceTemplate(
|
||||
name='Service Template 4',
|
||||
protocol=ServiceProtocolChoices.PROTOCOL_TCP,
|
||||
ports=[2001]
|
||||
port_assignments=[{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 2001}],
|
||||
),
|
||||
ServiceTemplate(
|
||||
name='Service Template 5',
|
||||
protocol=ServiceProtocolChoices.PROTOCOL_TCP,
|
||||
ports=[2002]
|
||||
port_assignments=[{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 2002}],
|
||||
),
|
||||
ServiceTemplate(
|
||||
name='Service Template 6',
|
||||
protocol=ServiceProtocolChoices.PROTOCOL_UDP,
|
||||
ports=[2003]
|
||||
port_assignments=[{'protocol': ServiceProtocolChoices.PROTOCOL_UDP, 'port': 2003}],
|
||||
),
|
||||
)
|
||||
ServiceTemplate.objects.bulk_create(service_templates)
|
||||
|
|
@ -2408,11 +2402,11 @@ class ServiceTemplateTestCase(TestCase, ChangeLoggedFilterSetTests):
|
|||
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
|
||||
|
||||
def test_protocol(self):
|
||||
params = {'protocol': ServiceProtocolChoices.PROTOCOL_TCP}
|
||||
params = {'protocol': [ServiceProtocolChoices.PROTOCOL_TCP]}
|
||||
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
|
||||
|
||||
def test_port(self):
|
||||
params = {'port': '1001'}
|
||||
params = {'port': [1001]}
|
||||
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
|
||||
|
||||
def test_description(self):
|
||||
|
|
@ -2423,7 +2417,7 @@ class ServiceTemplateTestCase(TestCase, ChangeLoggedFilterSetTests):
|
|||
class ServiceTestCase(TestCase, ChangeLoggedFilterSetTests):
|
||||
queryset = Service.objects.all()
|
||||
filterset = ServiceFilterSet
|
||||
ignore_fields = ('ports',)
|
||||
ignore_fields = ('port_assignments',)
|
||||
|
||||
@classmethod
|
||||
def setUpTestData(cls):
|
||||
|
|
@ -2472,46 +2466,39 @@ class ServiceTestCase(TestCase, ChangeLoggedFilterSetTests):
|
|||
Service(
|
||||
parent=devices[0],
|
||||
name='Service 1',
|
||||
protocol=ServiceProtocolChoices.PROTOCOL_TCP,
|
||||
ports=[1001],
|
||||
port_assignments=[{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 1001}],
|
||||
description='foobar1',
|
||||
),
|
||||
Service(
|
||||
parent=devices[1],
|
||||
name='Service 2',
|
||||
protocol=ServiceProtocolChoices.PROTOCOL_TCP,
|
||||
ports=[1002],
|
||||
port_assignments=[{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 1002}],
|
||||
description='foobar2',
|
||||
),
|
||||
Service(
|
||||
parent=devices[2],
|
||||
name='Service 3',
|
||||
protocol=ServiceProtocolChoices.PROTOCOL_UDP,
|
||||
ports=[1003]
|
||||
port_assignments=[{'protocol': ServiceProtocolChoices.PROTOCOL_UDP, 'port': 1003}],
|
||||
),
|
||||
Service(
|
||||
parent=virtual_machines[0],
|
||||
name='Service 4',
|
||||
protocol=ServiceProtocolChoices.PROTOCOL_TCP,
|
||||
ports=[2001],
|
||||
port_assignments=[{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 2001}],
|
||||
),
|
||||
Service(
|
||||
parent=virtual_machines[1],
|
||||
name='Service 5',
|
||||
protocol=ServiceProtocolChoices.PROTOCOL_TCP,
|
||||
ports=[2002],
|
||||
port_assignments=[{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 2002}],
|
||||
),
|
||||
Service(
|
||||
parent=virtual_machines[2],
|
||||
name='Service 6',
|
||||
protocol=ServiceProtocolChoices.PROTOCOL_UDP,
|
||||
ports=[2003],
|
||||
port_assignments=[{'protocol': ServiceProtocolChoices.PROTOCOL_UDP, 'port': 2003}],
|
||||
),
|
||||
Service(
|
||||
parent=fhrp_group,
|
||||
name='Service 7',
|
||||
protocol=ServiceProtocolChoices.PROTOCOL_UDP,
|
||||
ports=[2004],
|
||||
port_assignments=[{'protocol': ServiceProtocolChoices.PROTOCOL_UDP, 'port': 2004}],
|
||||
),
|
||||
)
|
||||
Service.objects.bulk_create(services)
|
||||
|
|
@ -2528,7 +2515,7 @@ class ServiceTestCase(TestCase, ChangeLoggedFilterSetTests):
|
|||
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
|
||||
|
||||
def test_protocol(self):
|
||||
params = {'protocol': ServiceProtocolChoices.PROTOCOL_TCP}
|
||||
params = {'protocol': [ServiceProtocolChoices.PROTOCOL_TCP]}
|
||||
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
|
||||
|
||||
def test_description(self):
|
||||
|
|
@ -2536,7 +2523,7 @@ class ServiceTestCase(TestCase, ChangeLoggedFilterSetTests):
|
|||
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
|
||||
|
||||
def test_port(self):
|
||||
params = {'port': '1001'}
|
||||
params = {'port': [1001]}
|
||||
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
|
||||
|
||||
def test_device(self):
|
||||
|
|
|
|||
|
|
@ -1894,6 +1894,55 @@ class ServiceTemplateTestCase(TestCase):
|
|||
)
|
||||
self.assertRaises(ValidationError, template.full_clean)
|
||||
|
||||
def test_servicetemplate_port_assignments(self):
|
||||
"""
|
||||
Test a template defined directly via port_assignments, mixing protocols on the same port.
|
||||
"""
|
||||
template = ServiceTemplate(
|
||||
name='Template 4',
|
||||
port_assignments=[
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 53},
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_UDP, 'port': 53},
|
||||
],
|
||||
)
|
||||
template.full_clean()
|
||||
template.save()
|
||||
self.assertEqual(template._ports_lowest, 53)
|
||||
# Deprecated accessors: protocol is None when mixed, ports is the flattened set
|
||||
self.assertIsNone(template.protocol)
|
||||
self.assertEqual(template.ports, [53])
|
||||
|
||||
def test_servicetemplate_uniform_protocol_accessor(self):
|
||||
"""
|
||||
The deprecated protocol accessor returns the single protocol when assignments are uniform.
|
||||
"""
|
||||
template = ServiceTemplate(
|
||||
name='Template 5',
|
||||
port_assignments=[
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 80},
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 443},
|
||||
],
|
||||
)
|
||||
template.full_clean()
|
||||
self.assertEqual(template.protocol, ServiceProtocolChoices.PROTOCOL_TCP)
|
||||
self.assertEqual(template.ports, [80, 443])
|
||||
|
||||
def test_servicetemplate_invalid_port_assignment(self):
|
||||
"""
|
||||
Invalid protocol/port values are rejected.
|
||||
"""
|
||||
template = ServiceTemplate(
|
||||
name='Template 6',
|
||||
port_assignments=[{'protocol': 'bogus', 'port': 80}],
|
||||
)
|
||||
self.assertRaises(ValidationError, template.full_clean)
|
||||
|
||||
template = ServiceTemplate(
|
||||
name='Template 7',
|
||||
port_assignments=[{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 99999}],
|
||||
)
|
||||
self.assertRaises(ValidationError, template.full_clean)
|
||||
|
||||
|
||||
class ServiceTestCase(TestCase):
|
||||
|
||||
|
|
|
|||
|
|
@ -1845,13 +1845,18 @@ class VLANTranslationRuleTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|||
|
||||
class ServiceTemplateTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
||||
model = ServiceTemplate
|
||||
# ports is a deprecated property derived from port_assignments; exclude it from instance comparison
|
||||
validation_excluded_fields = ('ports',)
|
||||
|
||||
@classmethod
|
||||
def setUpTestData(cls):
|
||||
service_templates = (
|
||||
ServiceTemplate(name='Service Template 1', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[101]),
|
||||
ServiceTemplate(name='Service Template 2', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[102]),
|
||||
ServiceTemplate(name='Service Template 3', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[103]),
|
||||
ServiceTemplate(name='Service Template 1', port_assignments=[
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 101}]),
|
||||
ServiceTemplate(name='Service Template 2', port_assignments=[
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 102}]),
|
||||
ServiceTemplate(name='Service Template 3', port_assignments=[
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 103}]),
|
||||
)
|
||||
ServiceTemplate.objects.bulk_create(service_templates)
|
||||
|
||||
|
|
@ -1859,7 +1864,7 @@ class ServiceTemplateTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|||
|
||||
cls.form_data = {
|
||||
'name': 'Service Template X',
|
||||
'protocol': ServiceProtocolChoices.PROTOCOL_UDP,
|
||||
'protocols': [ServiceProtocolChoices.PROTOCOL_UDP],
|
||||
'ports': '104,105',
|
||||
'description': 'A new service template',
|
||||
'tags': [t.pk for t in tags],
|
||||
|
|
@ -1880,16 +1885,16 @@ class ServiceTemplateTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|||
)
|
||||
|
||||
cls.bulk_edit_data = {
|
||||
'protocol': ServiceProtocolChoices.PROTOCOL_UDP,
|
||||
'ports': '106,107',
|
||||
'description': 'New description',
|
||||
'protocol': ServiceProtocolChoices.PROTOCOL_UDP,
|
||||
}
|
||||
|
||||
|
||||
class ServiceTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
||||
model = Service
|
||||
# TODO, related to #9816, cannot validate GFK
|
||||
validation_excluded_fields = ('device',)
|
||||
# TODO, related to #9816, cannot validate GFK. ports is a deprecated property derived from
|
||||
# port_assignments; exclude it from instance comparison.
|
||||
validation_excluded_fields = ('device', 'ports')
|
||||
|
||||
@classmethod
|
||||
def setUpTestData(cls):
|
||||
|
|
@ -1905,9 +1910,12 @@ class ServiceTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|||
)
|
||||
|
||||
services = (
|
||||
Service(parent=device, name='Service 1', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[101]),
|
||||
Service(parent=device, name='Service 2', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[102]),
|
||||
Service(parent=device, name='Service 3', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[103]),
|
||||
Service(parent=device, name='Service 1', port_assignments=[
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 101}]),
|
||||
Service(parent=device, name='Service 2', port_assignments=[
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 102}]),
|
||||
Service(parent=device, name='Service 3', port_assignments=[
|
||||
{'protocol': ServiceProtocolChoices.PROTOCOL_TCP, 'port': 103}]),
|
||||
)
|
||||
Service.objects.bulk_create(services)
|
||||
|
||||
|
|
@ -1924,7 +1932,7 @@ class ServiceTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|||
'parent_content_type': ContentType.objects.get_for_model(Device).pk,
|
||||
'parent_object_id': device.pk,
|
||||
'name': 'Service X',
|
||||
'protocol': ServiceProtocolChoices.PROTOCOL_TCP,
|
||||
'protocols': [ServiceProtocolChoices.PROTOCOL_TCP],
|
||||
'ports': '104,105',
|
||||
'ipaddresses': [],
|
||||
'description': 'A new service',
|
||||
|
|
@ -1947,9 +1955,8 @@ class ServiceTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|||
)
|
||||
|
||||
cls.bulk_edit_data = {
|
||||
'protocol': ServiceProtocolChoices.PROTOCOL_UDP,
|
||||
'ports': '106,107',
|
||||
'description': 'New description',
|
||||
'protocol': ServiceProtocolChoices.PROTOCOL_UDP,
|
||||
}
|
||||
|
||||
def test_unassigned_ip_addresses(self):
|
||||
|
|
|
|||
|
|
@ -236,16 +236,14 @@ class VLANCustomerVLANsPanel(panels.ObjectsTablePanel):
|
|||
|
||||
class ServiceTemplatePanel(panels.ObjectAttributesPanel):
|
||||
name = attrs.TextAttr('name')
|
||||
protocol = attrs.ChoiceAttr('protocol')
|
||||
ports = attrs.TextAttr('port_list', label=_('Ports'))
|
||||
ports = attrs.TextAttr('port_list', label=_('Port assignments'))
|
||||
description = attrs.TextAttr('description')
|
||||
|
||||
|
||||
class ServicePanel(panels.ObjectAttributesPanel):
|
||||
name = attrs.TextAttr('name')
|
||||
parent = attrs.RelatedObjectAttr('parent', linkify=True)
|
||||
protocol = attrs.ChoiceAttr('protocol')
|
||||
ports = attrs.TextAttr('port_list', label=_('Ports'))
|
||||
ports = attrs.TextAttr('port_list', label=_('Port assignments'))
|
||||
ip_addresses = attrs.TemplatedAttr(
|
||||
'ipaddresses',
|
||||
template_name='ipam/service/attrs/ip_addresses.html',
|
||||
|
|
|
|||
Loading…
Reference in New Issue