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 — target name already in use on the device.", len(conflicts))