Coverage for dcim/api/serializers_/base.py: 40%

1from drf_spectacular.types import OpenApiTypes 

2from drf_spectacular.utils import extend_schema_field 

3from rest_framework import serializers 

4 

5from dcim.models import FrontPort, FrontPortTemplate, PortMapping, PortTemplateMapping, RearPort, RearPortTemplate 

6from dcim.utils import reconcile_port_mappings 

7from utilities.api import get_serializer_for_model 

8 

9__all__ = ( 

10 'ConnectedEndpointsSerializer', 

11 'PortSerializer', 

12) 

13 

14 

15class ConnectedEndpointsSerializer(serializers.ModelSerializer): 

16 """ 

17 Legacy serializer for pre-v3.3 connections 

18 """ 

19 connected_endpoints_type = serializers.SerializerMethodField(read_only=True, allow_null=True) 

20 connected_endpoints = serializers.SerializerMethodField(read_only=True) 

21 connected_endpoints_reachable = serializers.SerializerMethodField(read_only=True) 

22 

23 @extend_schema_field(OpenApiTypes.STR) 

24 def get_connected_endpoints_type(self, obj): 

25 if endpoints := obj.connected_endpoints: 25 ↛ 26line 25 didn't jump to line 26 because the condition on line 25 was never true

26 return f'{endpoints[0]._meta.app_label}.{endpoints[0]._meta.model_name}' 

27 return None 

28 

29 @extend_schema_field(serializers.ListField(allow_null=True)) 

30 def get_connected_endpoints(self, obj): 

31 """ 

32 Return the appropriate serializer for the type of connected object. 

33 """ 

34 if endpoints := obj.connected_endpoints: 34 ↛ 35line 34 didn't jump to line 35 because the condition on line 34 was never true

35 serializer = get_serializer_for_model(endpoints[0]) 

36 context = {'request': self.context['request']} 

37 return serializer(endpoints, nested=True, many=True, context=context).data 

38 return None 

39 

40 @extend_schema_field(serializers.BooleanField) 

41 def get_connected_endpoints_reachable(self, obj): 

42 """ 

43 Return whether the connected endpoints are reachable via a complete, active cable path. 

44 """ 

45 # Use the public `path` accessor rather than dereferencing `_path` 

46 # directly. `path` already handles the stale in-memory relation case 

47 # that can occur while CablePath rows are rebuilt during cable edits. 

48 if path := obj.path: 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true

49 return path.is_complete and path.is_active 

50 return False 

51 

52 

53class PortSerializer(serializers.ModelSerializer): 

54 """ 

55 Base serializer for front & rear port and port templates. 

56 """ 

57 @property 

58 def _mapper(self): 

59 """ 

60 Return the model and ForeignKey field name used to track port mappings for this model. 

61 """ 

62 if self.Meta.model is FrontPort: 

63 return PortMapping, 'front_port' 

64 if self.Meta.model is RearPort: 

65 return PortMapping, 'rear_port' 

66 if self.Meta.model is FrontPortTemplate: 

67 return PortTemplateMapping, 'front_port' 

68 if self.Meta.model is RearPortTemplate: 

69 return PortTemplateMapping, 'rear_port' 

70 raise ValueError(f"Could not determine mapping details for {self.__class__}") 

71 

72 def _reconcile_mappings(self, instance, mappings): 

73 mapping_model, fk_name = self._mapper 

74 other_field = 'rear_port' if fk_name == 'front_port' else 'front_port' 

75 

76 # Normalize the opposite-port FK from a model instance to its id so the mappings can be 

77 # reconciled by value. 

78 desired = [] 

79 for attrs in mappings: 

80 attrs = dict(attrs) 

81 if other_field in attrs: 

82 attrs[f'{other_field}_id'] = attrs.pop(other_field).pk 

83 desired.append(attrs) 

84 

85 reconcile_port_mappings(mapping_model, parent_field=fk_name, parent=instance, desired=desired) 

86 

87 def create(self, validated_data): 

88 mappings = validated_data.pop('mappings', []) 

89 instance = super().create(validated_data) 

90 self._reconcile_mappings(instance, mappings) 

91 

92 return instance 

93 

94 def update(self, instance, validated_data): 

95 mappings = validated_data.pop('mappings', None) 

96 instance = super().update(instance, validated_data) 

97 

98 # Only reconcile when the client supplied rear_ports; a PATCH that omits it leaves the 

99 # existing mappings untouched. 

100 if mappings is not None: 

101 self._reconcile_mappings(instance, mappings) 

102 

103 return instance