Coverage for netbox_interface_name_rules / jobs.py: 97.40%

1# SPDX-License-Identifier: Apache-2.0 

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

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

4 

5import contextvars 

6import functools 

7import logging 

8from abc import abstractmethod 

9from contextlib import ExitStack 

10 

11from django.http import HttpRequest 

12from netbox.jobs import JobRunner 

13from netbox.registry import registry 

14 

15from . import branching 

16from .transactions import write_alias, write_scope 

17 

18 

19def rule_job_kwargs(rule_id): 

20 """Return the kwargs of a job on the rule *rule_id* in the branch of the current request, derived on the server.""" 

21 return {"rule_id": rule_id, "branch_schema_id": branching.branch_identity(), "expected_alias": write_alias()} 

22 

23 

24def _run_under_request_processors(request, body): 

25 """Call *body* inside every registered request processor; unlike NetBox, a processor that fails to enter raises.""" 

26 with ExitStack() as stack: 

27 for request_processor in registry["request_processors"]: 

28 stack.enter_context(request_processor(request)) 

29 return body() 

30 

31 

32def run_as_job_user(job, body, *, branch_schema_id): 

33 """Call *body*, which takes no arguments, as a request of the user who enqueued *job*. 

34 

35 NetBox writes a change log only while a request is current, and a worker has none. The request 

36 carries the job ID, so the change log lists the changes of one job under one request ID. The 

37 request is in the branch *branch_schema_id*, or on main for None. 

38 """ 

39 if job.user is None: 

40 raise ValueError(f"Job {job.pk} has no user, and the change log must name the user of each change.") 

41 # A Django request has each attribute that NetBox copies when an event rule acts on a change. 

42 request = HttpRequest() 

43 request.method = "POST" 

44 request.user = job.user 

45 request.id = job.job_id 

46 branching.activate_on(request, branch_schema_id) 

47 # The copy discards what the processors set; before NetBox 4.7, event_tracking keeps it when the body raises. 

48 return contextvars.copy_context().run(_run_under_request_processors, request, body) 

49 

50 

51class RuleJobRunner(JobRunner): 

52 """A background job over the rule named by rule_id, run as a request of the user who enqueued it. 

53 

54 Its kwargs are those of ``rule_job_kwargs``. A job enqueued without them fails. 

55 """ 

56 

57 def __init__(self, job): 

58 super().__init__(job) 

59 # NetBox 4.4 added this logger, which also writes the job log; NetBox 4.3 has none. 

60 if not hasattr(self, "logger"): 60 ↛ 61line 60 didn't jump to line 61 because the condition on line 60 was never true

61 self.logger = logging.getLogger(f"netbox.jobs.{type(self).__name__}") 

62 

63 def run(self, *, rule_id, branch_schema_id, expected_alias): 

64 """Run the job on the rule *rule_id* in the branch it was enqueued from; a missing rule is a warning.""" 

65 run_as_job_user( 

66 self.job, 

67 functools.partial(self._run_on_rule_id, rule_id, expected_alias), 

68 branch_schema_id=branch_schema_id, 

69 ) 

70 

71 def _run_on_rule_id(self, rule_id, expected_alias): 

72 from .models import InterfaceNameRule 

73 

74 # A branch that is not ready does not activate, so the scope raises before a read on main. 

75 with write_scope(expected_alias=expected_alias): 

76 rule = InterfaceNameRule.objects.filter(pk=rule_id).first() 

77 if rule is None: 

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

79 return 

80 try: 

81 self.run_on_rule(rule) 

82 except Exception: 

83 self.logger.exception("%s failed on rule '%s'", self.name, rule_id) 

84 raise 

85 

86 @abstractmethod 

87 def run_on_rule(self, rule): 

88 """Run the job on *rule* and log what it did.""" 

89 

90 

91class ApplyRuleJob(RuleJobRunner): 

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

93 

94 class Meta: 

95 name = "Apply Interface Name Rule" 

96 

97 def run_on_rule(self, rule): 

98 """Apply *rule* to all matching interfaces.""" 

99 from .engine import apply_rule_to_existing 

100 

101 outcome = apply_rule_to_existing(rule) 

102 self.logger.info("Renamed %d interface(s) using rule '%s'", outcome.changed_count, rule) 

103 if outcome.skipped_members: 

104 self.logger.warning("%d interface(s) skipped. The plugin log names each one.", len(outcome.skipped_members)) 

105 

106 

107class ConvertFlatFamiliesJob(RuleJobRunner): 

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

109 

110 class Meta: 

111 name = "Convert Flat Interface Families" 

112 

113 def run_on_rule(self, rule): 

114 """Convert every convertible flat family of *rule*.""" 

115 from .engine import convert_flat_families 

116 

117 outcome = convert_flat_families(rule) 

118 self.logger.info("Converted %d interface family(ies) using rule '%s'", len(outcome.changed_families), rule) 

119 if outcome.blocked_families: 

120 self.logger.warning("%d family(ies) skipped. The plugin log names each one.", len(outcome.blocked_families))