Coverage for /home/runner/work/netbox-InterfaceNameRules-plugin/netbox-InterfaceNameRules-plugin/netbox-InterfaceNameRules-plugin/netbox_interface_name_rules/jobs.py: 93%

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright (C) 2025 Marcin Zieba <marcinpsk@gmail.com> 

3"""Background jobs for bulk rule application.""" 

4 

5from netbox.jobs import JobRunner 

6 

7 

8class ApplyRuleJob(JobRunner): 

9 """Apply an InterfaceNameRule retroactively to all matching installed modules.""" 

10 

11 class Meta: 

12 name = "Apply Interface Name Rule" 

13 

14 def run(self, *args, **kwargs): 

15 """Retrieve the rule by pk from kwargs and apply it to all matching interfaces.""" 

16 from .engine import apply_rule_to_existing 

17 from .models import InterfaceNameRule 

18 

19 rule_id = kwargs.get("rule_id") 

20 if not rule_id: 

21 self.logger.warning("ApplyRuleJob called without rule_id; skipping.") 

22 return 

23 

24 try: 

25 rule = InterfaceNameRule.objects.get(pk=rule_id) 

26 except InterfaceNameRule.DoesNotExist: 

27 self.logger.warning("InterfaceNameRule with pk=%s does not exist; skipping.", rule_id) 

28 return 

29 

30 conflicts = [] 

31 try: 

32 count = apply_rule_to_existing(rule, conflicts=conflicts) 

33 except Exception as exc: 

34 self.logger.exception("Failed to apply rule '%s': %s", rule_id, exc) 

35 raise 

36 

37 self.logger.info("Renamed %d interface(s) using rule '%s'", count, rule) 

38 if conflicts: 38 ↛ 39line 38 didn't jump to line 39 because the condition on line 38 was never true

39 self.logger.warning("%d interface(s) skipped — the plugin log names each one.", len(conflicts)) 

40 

41 

42class ConvertFlatFamiliesJob(JobRunner): 

43 """Convert the flat breakout families a rule's modules still carry to the channelized topology.""" 

44 

45 class Meta: 

46 name = "Convert Flat Interface Families" 

47 

48 def run(self, *args, **kwargs): 

49 """Convert every convertible flat family of the rule identified by rule_id in kwargs.""" 

50 from .engine import convert_flat_families 

51 from .models import InterfaceNameRule 

52 

53 rule_id = kwargs.get("rule_id") 

54 if not rule_id: 

55 self.logger.warning("ConvertFlatFamiliesJob called without rule_id; skipping.") 

56 return 

57 

58 try: 

59 rule = InterfaceNameRule.objects.get(pk=rule_id) 

60 except InterfaceNameRule.DoesNotExist: 

61 self.logger.warning("InterfaceNameRule with pk=%s does not exist; skipping.", rule_id) 

62 return 

63 

64 conflicts = [] 

65 try: 

66 count = convert_flat_families(rule, conflicts=conflicts) 

67 except Exception as exc: 

68 self.logger.exception("Failed to convert families for rule '%s': %s", rule_id, exc) 

69 raise 

70 

71 self.logger.info("Converted %d interface family(ies) using rule '%s'", count, rule) 

72 if conflicts: 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true

73 self.logger.warning("%d family(ies) skipped — the plugin log names each one.", len(conflicts))