Coverage for vpn/models/l2vpn.py: 53%

1from django.contrib.contenttypes.fields import GenericForeignKey 

2from django.core.exceptions import ValidationError 

3from django.db import models 

4from django.utils.functional import cached_property 

5from django.utils.translation import gettext_lazy as _ 

6 

7from core.models import ObjectType 

8from netbox.models import NetBoxModel, PrimaryModel 

9from netbox.models.features import ContactsMixin 

10from vpn.choices import L2VPNStatusChoices, L2VPNTypeChoices 

11 

12__all__ = ( 

13 'L2VPN', 

14 'L2VPNTermination', 

15) 

16 

17 

18class L2VPN(ContactsMixin, PrimaryModel): 

19 name = models.CharField( 

20 verbose_name=_('name'), 

21 max_length=100, 

22 unique=True, 

23 db_collation="natural_sort" 

24 ) 

25 slug = models.SlugField( 

26 verbose_name=_('slug'), 

27 max_length=100, 

28 unique=True 

29 ) 

30 type = models.CharField( 

31 verbose_name=_('type'), 

32 max_length=50, 

33 choices=L2VPNTypeChoices 

34 ) 

35 status = models.CharField( 

36 verbose_name=_('status'), 

37 max_length=50, 

38 choices=L2VPNStatusChoices, 

39 default=L2VPNStatusChoices.STATUS_ACTIVE, 

40 ) 

41 identifier = models.BigIntegerField( 

42 verbose_name=_('identifier'), 

43 null=True, 

44 blank=True 

45 ) 

46 import_targets = models.ManyToManyField( 

47 to='ipam.RouteTarget', 

48 related_name='importing_l2vpns', 

49 blank=True, 

50 ) 

51 export_targets = models.ManyToManyField( 

52 to='ipam.RouteTarget', 

53 related_name='exporting_l2vpns', 

54 blank=True 

55 ) 

56 tenant = models.ForeignKey( 

57 to='tenancy.Tenant', 

58 on_delete=models.PROTECT, 

59 related_name='l2vpns', 

60 blank=True, 

61 null=True 

62 ) 

63 

64 clone_fields = ('type', 'status') 

65 

66 class Meta: 

67 ordering = ('name', 'identifier') 

68 verbose_name = _('L2VPN') 

69 verbose_name_plural = _('L2VPNs') 

70 

71 def __str__(self): 

72 if self.identifier: 

73 return f'{self.name} ({self.identifier})' 

74 return f'{self.name}' 

75 

76 def get_status_color(self): 

77 return L2VPNStatusChoices.colors.get(self.status) 

78 

79 @cached_property 

80 def can_add_termination(self): 

81 if self.type in L2VPNTypeChoices.P2P and self.terminations.count() >= 2: 

82 return False 

83 return True 

84 

85 

86class L2VPNTermination(NetBoxModel): 

87 l2vpn = models.ForeignKey( 

88 to='vpn.L2VPN', 

89 on_delete=models.CASCADE, 

90 related_name='terminations' 

91 ) 

92 assigned_object_type = models.ForeignKey( 

93 to='contenttypes.ContentType', 

94 on_delete=models.PROTECT, 

95 related_name='+' 

96 ) 

97 assigned_object_id = models.PositiveBigIntegerField() 

98 assigned_object = GenericForeignKey( 

99 ct_field='assigned_object_type', 

100 fk_field='assigned_object_id' 

101 ) 

102 

103 clone_fields = ('l2vpn',) 

104 prerequisite_models = ( 

105 'vpn.L2VPN', 

106 ) 

107 

108 class Meta: 

109 ordering = ('l2vpn',) 

110 constraints = ( 

111 models.UniqueConstraint( 

112 fields=('assigned_object_type', 'assigned_object_id'), 

113 name='vpn_l2vpntermination_assigned_object' 

114 ), 

115 ) 

116 verbose_name = _('L2VPN termination') 

117 verbose_name_plural = _('L2VPN terminations') 

118 

119 def __str__(self): 

120 if self.pk is not None: 

121 return f'{self.assigned_object} <> {self.l2vpn}' 

122 return super().__str__() 

123 

124 def clean(self): 

125 # Only check is assigned_object is set. Required otherwise we have an Integrity Error thrown. 

126 if self.assigned_object: 

127 obj_id = self.assigned_object.pk 

128 obj_type = ObjectType.objects.get_for_model(self.assigned_object) 

129 terminations = L2VPNTermination.objects.filter(assigned_object_id=obj_id, assigned_object_type=obj_type) 

130 if terminations.exclude(pk=self.pk).exists(): 

131 raise ValidationError( 

132 _('L2VPN Termination already assigned ({assigned_object})').format( 

133 assigned_object=self.assigned_object 

134 ) 

135 ) 

136 

137 # Only check if L2VPN is set and is of type P2P 

138 if hasattr(self, 'l2vpn') and self.l2vpn.type in L2VPNTypeChoices.P2P: 

139 terminations_count = L2VPNTermination.objects.filter(l2vpn=self.l2vpn).exclude(pk=self.pk).count() 

140 if terminations_count >= 2: 

141 l2vpn_type = self.l2vpn.get_type_display() 

142 raise ValidationError( 

143 _( 

144 '{l2vpn_type} L2VPNs cannot have more than two terminations; found {terminations_count} ' 

145 'already defined.' 

146 ).format(l2vpn_type=l2vpn_type, terminations_count=terminations_count)) 

147 

148 @property 

149 def assigned_object_parent(self): 

150 obj_type = ObjectType.objects.get_for_model(self.assigned_object) 

151 if obj_type.model == 'vminterface': 

152 return self.assigned_object.virtual_machine 

153 if obj_type.model == 'interface': 

154 return self.assigned_object.device 

155 if obj_type.model == 'vminterface': 

156 return self.assigned_object.virtual_machine 

157 return None 

158 

159 @property 

160 def assigned_object_site(self): 

161 return self.assigned_object_parent.site