Coverage for utilities/tracking.py: 77%

1from django.db.models.query_utils import DeferredAttribute 

2 

3from netbox.registry import registry 

4 

5 

6class Tracker: 

7 """ 

8 An ephemeral instance employed to record which tracked fields on an instance have been modified. 

9 """ 

10 def __init__(self): 

11 self._changed_fields = {} 

12 

13 def __contains__(self, item): 

14 return item in self._changed_fields 

15 

16 def set(self, name, value): 

17 """ 

18 Mark an attribute as having been changed and record its original value. 

19 """ 

20 self._changed_fields[name] = value 

21 

22 def get(self, name): 

23 """ 

24 Return the original value of a changed field. Raises KeyError if name is not found. 

25 """ 

26 return self._changed_fields[name] 

27 

28 def clear(self, *names): 

29 """ 

30 Clear any fields that were recorded as having been changed. 

31 """ 

32 for name in names: 32 ↛ 33line 32 didn't jump to line 33 because the loop on line 32 never started

33 self._changed_fields.pop(name, None) 

34 else: 

35 self._changed_fields = {} 

36 

37 

38class TrackingModelMixin: 

39 

40 def __init__(self, *args, **kwargs): 

41 super().__init__(*args, **kwargs) 

42 

43 # Mark the instance as initialized, to enable our custom __setattr__() 

44 self._initialized = True 

45 

46 @property 

47 def tracker(self): 

48 """ 

49 Return the Tracker instance for this instance, first creating it if necessary. 

50 """ 

51 if not hasattr(self._state, "_tracker"): 

52 self._state._tracker = Tracker() 

53 return self._state._tracker 

54 

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

56 super().save(*args, **kwargs) 

57 

58 # Clear any tracked fields now that changes have been saved 

59 update_fields = kwargs.get('update_fields', []) 

60 self.tracker.clear(*update_fields) 

61 

62 def __setattr__(self, name, value): 

63 if hasattr(self, "_initialized"): 

64 # Record any changes to a tracked field 

65 if name in registry['counter_fields'][self.__class__]: 

66 if name not in self.tracker: 66 ↛ 74line 66 didn't jump to line 74 because the condition on line 66 was always true

67 # The attribute has been created or changed 

68 if name in self.__dict__: 68 ↛ 73line 68 didn't jump to line 73 because the condition on line 68 was always true

69 old_value = getattr(self, name) 

70 if value != old_value: 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true

71 self.tracker.set(name, old_value) 

72 else: 

73 self.tracker.set(name, DeferredAttribute) 

74 elif value == self.tracker.get(name): 

75 # A previously changed attribute has been restored 

76 self.tracker.clear(name) 

77 

78 super().__setattr__(name, value)