Coverage for netbox_interface_name_rules / branching.py: 100.00%

1# SPDX-License-Identifier: Apache-2.0 

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

3"""The one module that uses netbox-branching. It imports netbox-branching only in a function that needs it.""" 

4 

5import functools 

6from contextvars import ContextVar 

7 

8from django.apps import apps 

9from django.core.exceptions import ImproperlyConfigured 

10from packaging.version import Version 

11 

12APP_LABEL = "netbox_branching" 

13SUPPORTED_RELEASE = (1, 2) 

14SUPPORTED_SERIES = ".".join(map(str, SUPPORTED_RELEASE)) + ".x" 

15# The methods of netbox-branching's Branch that replay logged changes. 

16REPLAYING_METHODS = ("merge", "revert", "sync") 

17# The attribute that marks a method this module wrapped. 

18REPLAY_MARK = "marks_a_replay" 

19 

20_replaying = ContextVar("netbox_interface_name_rules_replay", default=False) 

21 

22 

23def check_version(version: str) -> None: 

24 """Raise ``ImproperlyConfigured`` unless *version* is a release of the ``SUPPORTED_SERIES``.""" 

25 if Version(version).release[:2] != SUPPORTED_RELEASE: 

26 raise ImproperlyConfigured( 

27 f"Interface Name Rules supports netbox-branching {SUPPORTED_SERIES} only. The installed version is {version}." 

28 ) 

29 

30 

31def installed_version() -> str: 

32 """Return the version of the installed netbox-branching.""" 

33 return apps.get_app_config(APP_LABEL).version 

34 

35 

36def ready() -> None: 

37 """When netbox-branching is installed, check its version and mark each of its replays in the context that runs it.""" 

38 if not apps.is_installed(APP_LABEL): 

39 return 

40 check_version(installed_version()) 

41 from netbox_branching.models import Branch 

42 

43 for name in REPLAYING_METHODS: 

44 method = getattr(Branch, name) 

45 if not getattr(method, REPLAY_MARK, False): 

46 setattr(Branch, name, _marking_a_replay(method)) 

47 

48 

49def _marking_a_replay(method): 

50 """Return *method*, wrapped so that the context that runs it is in a replay until it returns or raises.""" 

51 

52 @functools.wraps(method) 

53 def replay(*args, **kwargs): 

54 token = _replaying.set(True) 

55 try: 

56 return method(*args, **kwargs) 

57 finally: 

58 _replaying.reset(token) 

59 

60 setattr(replay, REPLAY_MARK, True) 

61 return replay 

62 

63 

64def replay_in_progress() -> bool: 

65 """Return whether a netbox-branching merge, revert or sync runs in this context.""" 

66 return _replaying.get() 

67 

68 

69def branch_identity() -> str | None: 

70 """Return the schema ID of the active branch, or None on main and when netbox-branching is not installed.""" 

71 if not apps.is_installed(APP_LABEL): 

72 return None 

73 from netbox_branching.contextvars import active_branch 

74 

75 branch = active_branch.get() 

76 return None if branch is None else branch.schema_id 

77 

78 

79def activate_on(request, schema_id: str | None) -> None: 

80 """Put netbox-branching's branch cookie for *schema_id* on *request*; without either, the request stays on main.""" 

81 if schema_id is None or not apps.is_installed(APP_LABEL): 

82 return 

83 from netbox_branching.constants import COOKIE_NAME 

84 

85 request.COOKIES[COOKIE_NAME] = schema_id