Coverage for netbox_data_import/netbox_reader.py: 100.00%
84 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-30 09:47 +0000
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."""
5from __future__ import annotations
7from .catalog import import_location_required
8from .field_keys import CABLE_END_KINDS
9from .public_refusal import PublicRefusal
12class PlanningTargetUnavailable(PublicRefusal):
13 """The planning context names a target this reader cannot resolve."""
16class NetBoxReader:
17 """Permission-scoped reads of the NetBox objects planning compares against."""
19 def __init__(self, actor, site=None, location=None, tenant=None, *, location_unavailable=False):
20 self._actor = actor
21 self._site = site
22 self._location = location
23 self._tenant = tenant
24 self._location_unavailable = location_unavailable
26 def for_target(self, *, site, location=None, tenant=None) -> NetBoxReader:
27 """Bind the planning target without widening the actor's read scope."""
28 return type(self)(self._actor, site=site, location=location, tenant=tenant)
30 def for_planning_context(self, planning_context, *, output_kinds: frozenset[str]) -> NetBoxReader:
31 """Resolve a planning context for *output_kinds*, inside this reader's permission scope."""
32 from dcim.models import Location, Site
33 from tenancy.models import Tenant
35 if planning_context.get("site_id") is None:
36 raise PlanningTargetUnavailable("A planning context names the site the import writes into.")
37 site = self._required(Site, planning_context["site_id"])
38 location_id = planning_context.get("location_id")
39 location = None
40 if location_id is not None:
41 location = self._scoped(Location, "view").filter(pk=location_id, site=site).first()
42 location_unavailable = location_id is not None and location is None
43 # Section 6.1: a Location that only ranks candidates is evidence, not a target.
44 if location_unavailable and import_location_required(output_kinds):
45 self._required(Location, location_id)
46 raise PlanningTargetUnavailable("The selected location does not belong to the selected site.")
47 return type(self)(
48 self._actor,
49 site=site,
50 location=location,
51 tenant=self._optional(Tenant, planning_context.get("tenant_id")),
52 location_unavailable=location_unavailable,
53 )
55 def _required(self, model, pk):
56 """Return the object *pk* names, or refuse when it is gone or out of scope."""
57 found = self._scoped(model, "view").filter(pk=pk).first()
58 if found is None:
59 raise PlanningTargetUnavailable(f"{model._meta.verbose_name} {pk} is gone, or this actor cannot view it.")
60 return found
62 def _optional(self, model, pk):
63 """Return the object *pk* names, or None when the context names none."""
64 return None if pk is None else self._required(model, pk)
66 @property
67 def site(self):
68 """Return the site this import writes into, or None before a target is bound."""
69 return self._site
71 @property
72 def location(self):
73 """Return the location this import writes into, if the operator chose one."""
74 return self._location
76 @property
77 def location_unavailable(self) -> bool:
78 """Return whether the chosen import Location is gone, hidden, or outside the site."""
79 return self._location_unavailable
81 @property
82 def tenant(self):
83 """Return the tenant this import writes into, if the operator chose one."""
84 return self._tenant
86 @classmethod
87 def for_actor(cls, actor) -> NetBoxReader:
88 """Return a reader scoped to *actor*, which is required."""
89 if actor is None:
90 raise ValueError("A scoped NetBoxReader needs an actor. Use unrestricted() for no actor.")
91 return cls(actor)
93 @classmethod
94 def unrestricted(cls) -> NetBoxReader:
95 """Return a reader that applies no object permissions."""
96 return cls(None)
98 @classmethod
99 def for_optional_actor(cls, actor) -> NetBoxReader:
100 """Return a scoped reader, or an unrestricted one for an explicit system caller."""
101 return cls.unrestricted() if actor is None else cls.for_actor(actor)
103 @property
104 def actor(self):
105 """Return the actor every read is scoped to, or None for an unrestricted reader."""
106 return self._actor
108 def _scoped(self, model, action: str):
109 """Return *model*'s objects limited to what the actor may take *action* on."""
110 if self._actor is None:
111 return model.objects.all()
112 return model.objects.restrict(self._actor, action)
114 def devices(self, action: str = "view"):
115 """Return the Devices the actor may take *action* on."""
116 from dcim.models import Device
118 return self._scoped(Device, action)
120 def racks(self, action: str = "view"):
121 """Return the Racks the actor may take *action* on."""
122 from dcim.models import Rack
124 return self._scoped(Rack, action)
126 def locations(self, action: str = "view"):
127 """Return the Locations the actor may take *action* on."""
128 from dcim.models import Location
130 return self._scoped(Location, action)
132 def terminations(self, model_label: str, action: str = "view"):
133 """Return the rows of one Cable End Kind, named ``app_label.model``, the actor may take *action* on."""
134 from django.apps import apps
136 if model_label not in CABLE_END_KINDS:
137 raise ValueError(f"'{model_label}' is not a Cable End Kind.")
138 return self._scoped(apps.get_model(model_label), action)
140 def port_mappings(self):
141 """Return the PortMapping rows of the Devices this actor may view.
143 NetBox keeps the model private, with no manager of its own, so the parent Device carries
144 the scope.
145 """
146 from dcim.models import PortMapping
148 return PortMapping.objects.filter(device__in=self.devices())
151__all__ = ("NetBoxReader", "PlanningTargetUnavailable")