Coverage for ipam/signals.py: 73%

1from django.db.models.signals import post_delete, post_save, pre_delete 

2from django.dispatch import receiver 

3 

4from dcim.models import Device 

5from virtualization.models import VirtualMachine 

6 

7from .models import IPAddress, Prefix 

8 

9 

10def update_parents_children(prefix, using=None): 

11 """ 

12 Update depth on prefix & containing prefixes 

13 """ 

14 parents = prefix.get_parents(include_self=True).using(using).annotate_hierarchy() 

15 for parent in parents: 

16 parent._children = parent.hierarchy_children 

17 Prefix.objects.using(using).bulk_update(parents, ['_children'], batch_size=100) 

18 

19 

20def update_children_depth(prefix, using=None): 

21 """ 

22 Update children count on prefix & contained prefixes 

23 """ 

24 children = prefix.get_children(include_self=True).using(using).annotate_hierarchy() 

25 for child in children: 

26 child._depth = child.hierarchy_depth 

27 Prefix.objects.using(using).bulk_update(children, ['_depth'], batch_size=100) 

28 

29 

30@receiver(post_save, sender=Prefix) 

31def handle_prefix_saved(instance, created, using=None, **kwargs): 

32 """ 

33 Recompute the cached hierarchy counters for the prefixes surrounding this one. 

34 """ 

35 # Prefix has changed (or new instance has been created) 

36 if created or instance.vrf_id != instance._vrf_id or instance.prefix != instance._prefix: 

37 

38 update_parents_children(instance, using) 

39 update_children_depth(instance, using) 

40 

41 # If this is not a new prefix, clean up parent/children of previous prefix 

42 if not created: 42 ↛ 43line 42 didn't jump to line 43 because the condition on line 42 was never true

43 old_prefix = Prefix(vrf_id=instance._vrf_id, prefix=instance._prefix) 

44 update_parents_children(old_prefix, using) 

45 update_children_depth(old_prefix, using) 

46 

47 

48@receiver(post_delete, sender=Prefix) 

49def handle_prefix_deleted(instance, using=None, **kwargs): 

50 

51 update_parents_children(instance, using) 

52 update_children_depth(instance, using) 

53 

54 

55@receiver(pre_delete, sender=IPAddress) 

56def clear_primary_ip(instance, **kwargs): 

57 """ 

58 When an IPAddress is deleted, trigger save() on any Devices/VirtualMachines for which it was a primary IP. 

59 """ 

60 field_name = f'primary_ip{instance.family}' 

61 if device := Device.objects.filter(**{field_name: instance}).first(): 61 ↛ 62line 61 didn't jump to line 62 because the condition on line 61 was never true

62 device.snapshot() 

63 setattr(device, field_name, None) 

64 device.save() 

65 if virtualmachine := VirtualMachine.objects.filter(**{field_name: instance}).first(): 65 ↛ 66line 65 didn't jump to line 66 because the condition on line 65 was never true

66 virtualmachine.snapshot() 

67 setattr(virtualmachine, field_name, None) 

68 virtualmachine.save() 

69 

70 

71@receiver(pre_delete, sender=IPAddress) 

72def clear_oob_ip(instance, **kwargs): 

73 """ 

74 When an IPAddress is deleted, trigger save() on any Devices for which it was a OOB IP. 

75 """ 

76 if device := Device.objects.filter(oob_ip=instance).first(): 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true

77 device.snapshot() 

78 device.oob_ip = None 

79 device.save()