Coverage for ipam/api/serializers_/services.py: 52%

63 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-10-10 18:35 +0000

1from django.contrib.contenttypes.models import ContentType 

2from django.core.exceptions import ValidationError as DjangoValidationError 

3from django.utils.translation import gettext as _ 

4from rest_framework import serializers 

5 

6from ipam.choices import * 

7from ipam.constants import SERVICE_ASSIGNMENT_MODELS, SERVICE_PORT_MAX, SERVICE_PORT_MIN 

8from ipam.models import IPAddress, Service, ServiceTemplate 

9from ipam.utils import legacy_protocol_and_ports 

10from ipam.validators import validate_port_mappings 

11from netbox.api.fields import ChoiceField, ContentTypeField, SerializedPKRelatedField 

12from netbox.api.gfk_fields import GFKSerializerField 

13from netbox.api.serializers import PrimaryModelSerializer 

14 

15from .ip import IPAddressSerializer 

16 

17__all__ = ( 

18 'ServiceSerializer', 

19 'ServiceTemplateSerializer', 

20) 

21 

22 

23class PortMappingsField(serializers.ListField): 

24 """ 

25 A service's port mappings as a flat list of ``protocol/port`` strings (e.g. ``["tcp/80", "udp/53"]``), 

26 matching how they are stored. Each entry is validated (and normalized) on write. 

27 """ 

28 child = serializers.CharField() 

29 

30 def to_internal_value(self, data): 

31 mappings = super().to_internal_value(data) 

32 try: 

33 return validate_port_mappings(mappings) 

34 except DjangoValidationError as exc: 

35 raise serializers.ValidationError(exc.messages) 

36 

37 

38class PortMappingsSerializerMixin(serializers.Serializer): 

39 """ 

40 Shared port-mapping handling for the Service and ServiceTemplate serializers, including backward 

41 compatibility for the legacy single-protocol ``protocol``/``ports`` representation. 

42 

43 Read: alongside the ``port_mappings`` list, a service that uses a single protocol also reports the 

44 legacy ``protocol`` and ``ports`` fields; a multi-protocol service reports ``null`` for both (it 

45 cannot be expressed in the old single-protocol format). 

46 

47 Write: either format is accepted. When the legacy ``protocol``/``ports`` pair is supplied (and 

48 ``port_mappings`` is not), it is translated into ``port_mappings``. Supplying both together is 

49 accepted only when the legacy fields agree with what ``port_mappings`` implies (e.g. a full-object 

50 round-trip that echoes back the read representation); a genuine conflict is rejected as ambiguous. 

51 

52 Subclassing ``serializers.Serializer`` (rather than a plain mixin) lets DRF's metaclass collect the 

53 fields declared here into the inheriting serializers. 

54 """ 

55 port_mappings = PortMappingsField(required=False) 

56 

57 # Legacy single-protocol fields, retained for backward compatibility. They are read straight off the 

58 # model's protocol/ports properties (which share these field names), so DRF sources them directly — 

59 # matching the {"value", "label"} shape every other choice field uses. default=None applies only on 

60 # write, where validate() consumes them. 

61 # TODO: Remove protocol/ports in v5.0 along with the legacy handling in validate(). 

62 protocol = ChoiceField( 

63 choices=ServiceProtocolChoices, 

64 required=False, 

65 allow_null=True, 

66 default=None, 

67 help_text=_("Deprecated; use port_mappings. Reported only for single-protocol services."), 

68 ) 

69 ports = serializers.ListField( 

70 child=serializers.IntegerField(min_value=SERVICE_PORT_MIN, max_value=SERVICE_PORT_MAX), 

71 required=False, 

72 allow_null=True, 

73 default=None, 

74 help_text=_("Deprecated; use port_mappings. Reported only for single-protocol services."), 

75 ) 

76 

77 def validate(self, data): 

78 # Consume the legacy fields and translate them into port_mappings *before* calling super(), 

79 # which instantiates the model (via full_clean()) and would choke on these now-nonexistent kwargs. 

80 legacy_protocol = data.pop('protocol', None) 

81 legacy_ports = data.pop('ports', None) 

82 # protocol/ports carry default=None, so an omitted field arrives as None; an explicitly-supplied 

83 # value (including a falsy ports=[]) is a legacy write and must be handled — checking `is not None` 

84 # rather than truthiness so an intentional empty list isn't silently dropped. 

85 if legacy_protocol is not None or legacy_ports is not None: 

86 # `port_mappings` and `protocol`/`ports` are mutually exclusive as *representations*, but a 

87 # full-object round-trip (GET then PUT/PATCH) legitimately resubmits port_mappings alongside 

88 # the legacy protocol/ports the read emitted. Only reject a genuine *conflict*: when the legacy 

89 # fields agree with what port_mappings already implies they're merely redundant, so accept the 

90 # request and let port_mappings win. 

91 if 'port_mappings' in data: 

92 expected_protocol, expected_ports = legacy_protocol_and_ports(data['port_mappings']) 

93 protocol_agrees = legacy_protocol is None or legacy_protocol == expected_protocol 

94 ports_agree = legacy_ports is None or sorted(legacy_ports) == (expected_ports or []) 

95 if not (protocol_agrees and ports_agree): 

96 raise serializers.ValidationError(_( 

97 "Specify either 'port_mappings' or the deprecated 'protocol'/'ports' fields, not both." 

98 )) 

99 return super().validate(data) 

100 # The old API accepted an empty ports list (the ArrayField had no minimum length); the new 

101 # model requires at least one mapping. Report that directly — both fields may have been 

102 # supplied, so the "both are required" message below would be misleading. 

103 if legacy_ports == []: 

104 raise serializers.ValidationError( 

105 {'ports': _("At least one port mapping is required.")} 

106 ) 

107 # The legacy API let either field be updated on its own (e.g. a PATCH that adjusts only the 

108 # port list). Preserve that by backfilling the omitted field from the instance's current 

109 # single-protocol representation. 

110 if not (legacy_protocol and legacy_ports): 

111 legacy_protocol = legacy_protocol or (self.instance.protocol if self.instance else None) 

112 if legacy_ports is None: 

113 legacy_ports = self.instance.ports if self.instance else None 

114 # If the pair still can't be resolved — a create, or an existing multi-protocol service that 

115 # has no single-protocol form — the request can't be expressed in the legacy format. 

116 if not (legacy_protocol and legacy_ports): 

117 raise serializers.ValidationError(_( 

118 "Both 'protocol' and 'ports' are required when writing via the deprecated legacy " 

119 "format; use port_mappings instead." 

120 )) 

121 try: 

122 data['port_mappings'] = validate_port_mappings( 

123 [f'{legacy_protocol}/{port}' for port in legacy_ports] 

124 ) 

125 except DjangoValidationError as exc: 

126 raise serializers.ValidationError({'ports': exc.messages}) 

127 

128 return super().validate(data) 

129 

130 

131class ServiceTemplateSerializer(PortMappingsSerializerMixin, PrimaryModelSerializer): 

132 

133 class Meta: 

134 model = ServiceTemplate 

135 fields = [ 

136 'id', 'url', 'display_url', 'display', 'name', 'port_mappings', 'protocol', 'ports', 'description', 

137 'owner', 'comments', 'tags', 'custom_fields', 'created', 'last_updated', 

138 ] 

139 brief_fields = ('id', 'url', 'display', 'name', 'port_mappings', 'description') 

140 

141 

142class ServiceSerializer(PortMappingsSerializerMixin, PrimaryModelSerializer): 

143 ipaddresses = SerializedPKRelatedField( 

144 queryset=IPAddress.objects.all(), 

145 serializer=IPAddressSerializer, 

146 nested=True, 

147 required=False, 

148 many=True 

149 ) 

150 parent_object_type = ContentTypeField( 

151 queryset=ContentType.objects.filter(SERVICE_ASSIGNMENT_MODELS) 

152 ) 

153 parent = GFKSerializerField(read_only=True) 

154 

155 class Meta: 

156 model = Service 

157 fields = [ 

158 'id', 'url', 'display_url', 'display', 'parent_object_type', 'parent_object_id', 'parent', 'name', 

159 'port_mappings', 'protocol', 'ports', 'ipaddresses', 'description', 'owner', 'comments', 'tags', 

160 'custom_fields', 'created', 'last_updated', 

161 ] 

162 brief_fields = ('id', 'url', 'display', 'name', 'port_mappings', 'description')