Coverage for netbox_data_import/netbox_reader.py: 100%

1# SPDX-License-Identifier: Apache-2.0 

2# SPDX-FileCopyrightText: 2026 Marcin Zieba <marcinpsk@gmail.com> 

3"""Permission-scoped target-state reads for import planning.""" 

4 

5from __future__ import annotations 

6 

7 

8class PlanningTargetUnavailable(Exception): 

9 """The planning context names a target this reader cannot resolve.""" 

10 

11 

12class NetBoxReader: 

13 """Permission-scoped reads of the NetBox objects planning compares against.""" 

14 

15 def __init__(self, actor, site=None, location=None, tenant=None): 

16 self._actor = actor 

17 self._site = site 

18 self._location = location 

19 self._tenant = tenant 

20 

21 def for_target(self, *, site, location=None, tenant=None) -> NetBoxReader: 

22 """Bind the planning target without widening the actor's read scope.""" 

23 return type(self)(self._actor, site=site, location=location, tenant=tenant) 

24 

25 def for_planning_context(self, planning_context) -> NetBoxReader: 

26 """Resolve a planning context through this reader's permission scope.""" 

27 from dcim.models import Location, Site 

28 from tenancy.models import Tenant 

29 

30 if planning_context.get("site_id") is None: 

31 raise PlanningTargetUnavailable("A planning context names the site the import writes into.") 

32 site = self._required(Site, planning_context["site_id"]) 

33 location = self._optional(Location, planning_context.get("location_id")) 

34 if location is not None and location.site_id != site.pk: 

35 raise PlanningTargetUnavailable("The selected location does not belong to the selected site.") 

36 return self.for_target( 

37 site=site, 

38 location=location, 

39 tenant=self._optional(Tenant, planning_context.get("tenant_id")), 

40 ) 

41 

42 def _required(self, model, pk): 

43 """Return the object *pk* names, or refuse when it is gone or out of scope.""" 

44 found = self._scoped(model, "view").filter(pk=pk).first() 

45 if found is None: 

46 raise PlanningTargetUnavailable(f"{model._meta.verbose_name} {pk} is gone, or this actor cannot view it.") 

47 return found 

48 

49 def _optional(self, model, pk): 

50 """Return the object *pk* names, or None when the context names none.""" 

51 return None if pk is None else self._required(model, pk) 

52 

53 @property 

54 def site(self): 

55 """Return the site this import writes into, or None before a target is bound.""" 

56 return self._site 

57 

58 @property 

59 def location(self): 

60 """Return the location this import writes into, if the operator chose one.""" 

61 return self._location 

62 

63 @property 

64 def tenant(self): 

65 """Return the tenant this import writes into, if the operator chose one.""" 

66 return self._tenant 

67 

68 @classmethod 

69 def for_actor(cls, actor) -> NetBoxReader: 

70 """Return a reader scoped to *actor*, which is required.""" 

71 if actor is None: 

72 raise ValueError("A scoped NetBoxReader needs an actor. Use unrestricted() for no actor.") 

73 return cls(actor) 

74 

75 @classmethod 

76 def unrestricted(cls) -> NetBoxReader: 

77 """Return a reader that applies no object permissions.""" 

78 return cls(None) 

79 

80 @classmethod 

81 def for_optional_actor(cls, actor) -> NetBoxReader: 

82 """Return a scoped reader, or an unrestricted one for an explicit system caller.""" 

83 return cls.unrestricted() if actor is None else cls.for_actor(actor) 

84 

85 @property 

86 def actor(self): 

87 """Return the actor every read is scoped to, or None for an unrestricted reader.""" 

88 return self._actor 

89 

90 def _scoped(self, model, action: str): 

91 """Return *model*'s objects limited to what the actor may take *action* on.""" 

92 if self._actor is None: 

93 return model.objects.all() 

94 return model.objects.restrict(self._actor, action) 

95 

96 def devices(self, action: str = "view"): 

97 """Return the Devices the actor may take *action* on.""" 

98 from dcim.models import Device 

99 

100 return self._scoped(Device, action) 

101 

102 def racks(self, action: str = "view"): 

103 """Return the Racks the actor may take *action* on.""" 

104 from dcim.models import Rack 

105 

106 return self._scoped(Rack, action) 

107 

108 def interfaces(self, action: str = "view"): 

109 """Return the Interfaces the actor may take *action* on.""" 

110 from dcim.models import Interface 

111 

112 return self._scoped(Interface, action) 

113 

114 def front_ports(self, action: str = "view"): 

115 """Return the Front Ports the actor may take *action* on.""" 

116 from dcim.models import FrontPort 

117 

118 return self._scoped(FrontPort, action) 

119 

120 def rear_ports(self, action: str = "view"): 

121 """Return the Rear Ports the actor may take *action* on.""" 

122 from dcim.models import RearPort 

123 

124 return self._scoped(RearPort, action) 

125 

126 def port_mappings(self): 

127 """Return the PortMapping rows of the Devices this actor may view. 

128 

129 NetBox keeps the model private, with no manager of its own, so the parent Device carries 

130 the scope. 

131 """ 

132 from dcim.models import PortMapping 

133 

134 return PortMapping.objects.filter(device__in=self.devices()) 

135 

136 

137__all__ = ("NetBoxReader", "PlanningTargetUnavailable")