Coverage for core/models/change_logging.py: 75%

1from functools import cached_property 

2 

3from django.conf import settings 

4from django.contrib.contenttypes.fields import GenericForeignKey 

5from django.core.exceptions import ValidationError 

6from django.db import models 

7from django.urls import reverse 

8from django.utils.translation import gettext_lazy as _ 

9 

10from core.choices import ObjectChangeActionChoices 

11from core.querysets import ObjectChangeQuerySet 

12from netbox.models.features import ChangeLoggingMixin, has_feature 

13from utilities.data import deep_compare_dict 

14 

15__all__ = ( 

16 'ObjectChange', 

17) 

18 

19 

20class ObjectChange(models.Model): 

21 """ 

22 Record a change to an object and the user account associated with that change. A change record may optionally 

23 indicate an object related to the one being changed. For example, a change to an interface may also indicate the 

24 parent device. This will ensure changes made to component models appear in the parent model's changelog. 

25 """ 

26 time = models.DateTimeField( 

27 verbose_name=_('time'), 

28 auto_now_add=True, 

29 editable=False, 

30 db_index=True 

31 ) 

32 user = models.ForeignKey( 

33 to=settings.AUTH_USER_MODEL, 

34 on_delete=models.SET_NULL, 

35 related_name='changes', 

36 blank=True, 

37 null=True 

38 ) 

39 user_name = models.CharField( 

40 verbose_name=_('user name'), 

41 max_length=150, 

42 editable=False 

43 ) 

44 request_id = models.UUIDField( 

45 verbose_name=_('request ID'), 

46 editable=False, 

47 db_index=True 

48 ) 

49 action = models.CharField( 

50 verbose_name=_('action'), 

51 max_length=50, 

52 choices=ObjectChangeActionChoices 

53 ) 

54 changed_object_type = models.ForeignKey( 

55 to='contenttypes.ContentType', 

56 on_delete=models.PROTECT, 

57 related_name='+' 

58 ) 

59 changed_object_id = models.PositiveBigIntegerField() 

60 changed_object = GenericForeignKey( 

61 ct_field='changed_object_type', 

62 fk_field='changed_object_id' 

63 ) 

64 related_object_type = models.ForeignKey( 

65 to='contenttypes.ContentType', 

66 on_delete=models.PROTECT, 

67 related_name='+', 

68 blank=True, 

69 null=True 

70 ) 

71 related_object_id = models.PositiveBigIntegerField( 

72 blank=True, 

73 null=True 

74 ) 

75 related_object = GenericForeignKey( 

76 ct_field='related_object_type', 

77 fk_field='related_object_id' 

78 ) 

79 object_repr = models.CharField( 

80 max_length=200, 

81 editable=False 

82 ) 

83 message = models.CharField( 

84 verbose_name=_('message'), 

85 max_length=200, 

86 editable=False, 

87 blank=True 

88 ) 

89 prechange_data = models.JSONField( 

90 verbose_name=_('pre-change data'), 

91 editable=False, 

92 blank=True, 

93 null=True 

94 ) 

95 postchange_data = models.JSONField( 

96 verbose_name=_('post-change data'), 

97 editable=False, 

98 blank=True, 

99 null=True 

100 ) 

101 

102 objects = ObjectChangeQuerySet.as_manager() 

103 

104 class Meta: 

105 ordering = ['-time'] 

106 indexes = ( 

107 models.Index(fields=('changed_object_type', 'changed_object_id')), 

108 models.Index(fields=('related_object_type', 'related_object_id')), 

109 ) 

110 verbose_name = _('object change') 

111 verbose_name_plural = _('object changes') 

112 

113 def __str__(self): 

114 return '{} {} {} by {}'.format( 

115 self.changed_object_type, 

116 self.object_repr, 

117 self.get_action_display().lower(), 

118 self.user_name 

119 ) 

120 

121 def clean(self): 

122 super().clean() 

123 

124 # Validate the assigned object type 

125 if not has_feature(self.changed_object_type, 'change_logging'): 

126 raise ValidationError( 

127 _("Change logging is not supported for this object type ({type}).").format( 

128 type=self.changed_object_type 

129 ) 

130 ) 

131 

132 def save(self, *args, **kwargs): 

133 

134 # Record the user's name and the object's representation as static strings 

135 if not self.user_name: 135 ↛ 137line 135 didn't jump to line 137 because the condition on line 135 was always true

136 self.user_name = self.user.username 

137 if not self.object_repr: 137 ↛ 138line 137 didn't jump to line 138 because the condition on line 137 was never true

138 self.object_repr = str(self.changed_object) 

139 

140 return super().save(*args, **kwargs) 

141 

142 def get_absolute_url(self): 

143 return reverse('core:objectchange', args=[self.pk]) 

144 

145 def get_action_color(self): 

146 return ObjectChangeActionChoices.colors.get(self.action) 

147 

148 @cached_property 

149 def has_changes(self): 

150 return self.prechange_data != self.postchange_data 

151 

152 @cached_property 

153 def diff_exclude_fields(self): 

154 """ 

155 Return a set of attributes which should be ignored when calculating a diff 

156 between the pre- and post-change data. (For instance, it would not make 

157 sense to compare the "last updated" times as these are expected to differ.) 

158 """ 

159 model = self.changed_object_type.model_class() 

160 attrs = set() 

161 

162 # model_class() returns None when the model's app is no longer installed 

163 # (e.g. a removed plugin); there are no fields to exclude, and the 

164 # issubclass() checks below would raise TypeError on None. 

165 if model is None: 165 ↛ 166line 165 didn't jump to line 166 because the condition on line 165 was never true

166 return attrs 

167 

168 # Exclude auto-populated change tracking fields 

169 if issubclass(model, ChangeLoggingMixin): 169 ↛ 173line 169 didn't jump to line 173 because the condition on line 169 was always true

170 attrs.update({'created', 'last_updated'}) 

171 

172 # Exclude trigger-maintained ltree columns (path and the optional sort_path) 

173 from netbox.models.ltree import LtreeModel 

174 if issubclass(model, LtreeModel): 

175 attrs.update({'path', 'sort_path'}) 

176 

177 # Exclude MPTT bookkeeping columns for the deprecated MPTT-backed 

178 # NestedGroupModel still shipped for plugin compatibility. 

179 from mptt.models import MPTTModel 

180 if issubclass(model, MPTTModel): 180 ↛ 181line 180 didn't jump to line 181 because the condition on line 180 was never true

181 attrs.update({'lft', 'rght', 'tree_id', 'level'}) 

182 

183 return attrs 

184 

185 def get_clean_data(self, prefix): 

186 """ 

187 Return only the pre-/post-change attributes which are relevant for calculating a diff. 

188 """ 

189 ret = {} 

190 change_data = getattr(self, f'{prefix}_data') or {} 

191 for k, v in change_data.items(): 

192 if k not in self.diff_exclude_fields and not k.startswith('_'): 

193 ret[k] = v 

194 return ret 

195 

196 @cached_property 

197 def prechange_data_clean(self): 

198 return self.get_clean_data('prechange') 

199 

200 @cached_property 

201 def postchange_data_clean(self): 

202 return self.get_clean_data('postchange') 

203 

204 def diff(self): 

205 """ 

206 Return a dictionary of pre- and post-change values for attributes which have changed. 

207 """ 

208 prechange_data = self.prechange_data_clean 

209 postchange_data = self.postchange_data_clean 

210 

211 # Determine which attributes have changed 

212 if self.action == ObjectChangeActionChoices.ACTION_CREATE: 

213 changed_attrs = sorted(postchange_data.keys()) 

214 return { 

215 'pre': {k: prechange_data.get(k) for k in changed_attrs}, 

216 'post': {k: postchange_data.get(k) for k in changed_attrs}, 

217 } 

218 if self.action == ObjectChangeActionChoices.ACTION_DELETE: 

219 changed_attrs = sorted(prechange_data.keys()) 

220 return { 

221 'pre': {k: prechange_data.get(k) for k in changed_attrs}, 

222 'post': {k: postchange_data.get(k) for k in changed_attrs}, 

223 } 

224 diff_added, diff_removed = deep_compare_dict(prechange_data, postchange_data) 

225 return { 

226 'pre': dict(sorted(diff_removed.items())), 

227 'post': dict(sorted(diff_added.items())), 

228 }