Coverage for core/models/config.py: 47%

1from django.core.cache import cache 

2from django.db import models, router, transaction 

3from django.urls import reverse 

4from django.utils.translation import gettext 

5from django.utils.translation import gettext_lazy as _ 

6 

7from utilities.querysets import RestrictedQuerySet, chunked_update 

8 

9__all__ = ( 

10 'ConfigRevision', 

11) 

12 

13 

14class ConfigRevision(models.Model): 

15 """ 

16 An atomic revision of NetBox's configuration. 

17 """ 

18 active = models.BooleanField( 

19 default=False 

20 ) 

21 created = models.DateTimeField( 

22 verbose_name=_('created'), 

23 auto_now_add=True 

24 ) 

25 comment = models.CharField( 

26 verbose_name=_('comment'), 

27 max_length=200, 

28 blank=True 

29 ) 

30 data = models.JSONField( 

31 blank=True, 

32 null=True, 

33 verbose_name=_('configuration data') 

34 ) 

35 

36 objects = RestrictedQuerySet.as_manager() 

37 

38 class Meta: 

39 ordering = ['-created'] 

40 indexes = ( 

41 models.Index(fields=('-created',)), # Default ordering 

42 ) 

43 verbose_name = _('config revision') 

44 verbose_name_plural = _('config revisions') 

45 constraints = [ 

46 models.UniqueConstraint( 

47 fields=('active',), 

48 condition=models.Q(active=True), 

49 name='unique_active_config_revision', 

50 ) 

51 ] 

52 

53 def __str__(self): 

54 if not self.pk: 

55 return gettext('Default configuration') 

56 if self.is_active: 

57 return gettext('Current configuration') 

58 return gettext('Config revision #{id}').format(id=self.pk) 

59 

60 def __getattr__(self, item): 

61 if self.data and item in self.data: 

62 return self.data[item] 

63 return super().__getattribute__(item) 

64 

65 def get_absolute_url(self): 

66 if not self.pk: 

67 return reverse('core:config') # Default config view 

68 return reverse('core:configrevision', args=[self.pk]) 

69 

70 def activate(self, update_db=True): 

71 """ 

72 Cache the configuration data. 

73 

74 Parameters: 

75 update_db: Mark the ConfigRevision as active in the database (default: True) 

76 """ 

77 cache.set('config', self.data, None) 

78 cache.set('config_version', self.pk, None) 

79 

80 if update_db: 

81 # Set all instances of ConfigRevision to false and set this instance to true. Wrap both 

82 # statements in a transaction so the "exactly one active revision" invariant is preserved 

83 # even when the deactivation is chunked into multiple statements. Resolve the write alias 

84 # once and pin the transaction and both querysets to it, so the transaction genuinely 

85 # covers the (potentially router-directed) writes performed by chunked_update(). 

86 using = router.db_for_write(ConfigRevision) 

87 with transaction.atomic(using=using): 

88 chunked_update(ConfigRevision.objects.using(using).all(), active=False) 

89 ConfigRevision.objects.using(using).filter(pk=self.pk).update(active=True) 

90 

91 activate.alters_data = True 

92 

93 @property 

94 def is_active(self): 

95 return self.active