Coverage for netbox_data_import/api/views.py: 96.69%

1# SPDX-License-Identifier: Apache-2.0 

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

3"""DRF viewsets for the data-import plugin API.""" 

4 

5from django.http import Http404 

6from netbox.api.viewsets import NetBoxModelViewSet 

7from rest_framework import mixins, permissions, viewsets 

8from rest_framework.exceptions import PermissionDenied, ValidationError 

9from rest_framework.permissions import DjangoModelPermissions 

10 

11from ..filters import ImportProfileFilterSet, InferenceBackendFilterSet 

12from ..field_keys import SELECT_TERMINATION_TASK, parse_termination_field_key 

13from ..models import ( 

14 locked_profile_policy, 

15 ImportProfile, 

16 CableClassMapping, 

17 CableImportSource, 

18 ColumnMapping, 

19 ClassRoleMapping, 

20 DeviceTypeMapping, 

21 IgnoredDevice, 

22 ColumnTransformRule, 

23 SourceResolution, 

24 ImportExecution, 

25 InferenceBackend, 

26 PortClassMapping, 

27 ResolutionProposal, 

28) 

29from ..object_permissions import ( 

30 ObjectPermissionDenied, 

31 delete_permission_scoped_objects, 

32 save_permission_scoped_object, 

33) 

34from .serializers import ( 

35 ImportProfileSerializer, 

36 CableClassMappingSerializer, 

37 CableImportSourceSerializer, 

38 ColumnMappingSerializer, 

39 ClassRoleMappingSerializer, 

40 DeviceTypeMappingSerializer, 

41 IgnoredDeviceSerializer, 

42 ColumnTransformRuleSerializer, 

43 SourceResolutionSerializer, 

44 ImportExecutionSerializer, 

45 InferenceBackendSerializer, 

46 PortClassMappingSerializer, 

47 ResolutionProposalHistorySerializer, 

48 ResolutionProposalSerializer, 

49) 

50 

51 

52class DjangoModelPermissionsWithView(DjangoModelPermissions): 

53 """Extends DjangoModelPermissions to require view_* permission for GET requests. 

54 

55 The stock DjangoModelPermissions does not map GET to any model permission, 

56 so list/retrieve endpoints are accessible to any authenticated user. This 

57 subclass closes that gap. 

58 """ 

59 

60 perms_map = { 

61 **DjangoModelPermissions.perms_map, 

62 "GET": ["%(app_label)s.view_%(model_name)s"], 

63 "HEAD": ["%(app_label)s.view_%(model_name)s"], 

64 "OPTIONS": [], 

65 } 

66 

67 

68class ImportProfileViewSet(NetBoxModelViewSet): 

69 """CRUD viewset for ImportProfile (NetBoxModel).""" 

70 

71 queryset = ImportProfile.objects.prefetch_related( 

72 "tags", "column_mappings", "class_role_mappings", "device_type_mappings" 

73 ) 

74 serializer_class = ImportProfileSerializer 

75 filterset_class = ImportProfileFilterSet 

76 

77 

78class _ProfileScopedQuerySetMixin(viewsets.GenericViewSet): 

79 """Restrict profile-owned rows and validate their optional profile filter.""" 

80 

81 def get_queryset(self): 

82 """Return viewable rows, filtered by a valid profile ID when supplied.""" 

83 qs = super().get_queryset().restrict(self.request.user, "view") 

84 profile_id = self.request.query_params.get("profile_id") 

85 if profile_id is not None: 

86 try: 

87 profile_id = int(profile_id) 

88 except (TypeError, ValueError) as exc: 

89 raise ValidationError({"profile_id": "Enter a whole number."}) from exc 

90 qs = qs.filter(profile_id=profile_id) 

91 return qs 

92 

93 

94class _PluginModelViewSet(_ProfileScopedQuerySetMixin, viewsets.ModelViewSet): 

95 """Base class for plain-model viewsets in this plugin.""" 

96 

97 permission_classes = [permissions.IsAuthenticated, DjangoModelPermissionsWithView] 

98 

99 def perform_create(self, serializer): 

100 """Create one policy row inside its profile and object-permission scope.""" 

101 model = serializer.Meta.model 

102 profile_id = serializer.validated_data["profile"].pk 

103 try: 

104 with locked_profile_policy(profile_id): 

105 serializer.run_validation(serializer.initial_data) 

106 values = serializer.model_cleaned_values() 

107 profile = values.pop("profile") 

108 result = save_permission_scoped_object( 

109 self.request.user, 

110 model, 

111 {"pk": None, "profile": profile}, 

112 values, 

113 on_existing="reject", 

114 ) 

115 except ImportProfile.DoesNotExist: 

116 raise Http404 from None 

117 except ObjectPermissionDenied as exc: 

118 raise PermissionDenied from exc 

119 serializer.instance = result.instance 

120 

121 def perform_update(self, serializer): 

122 """Update one policy row under every affected profile and object scope.""" 

123 model = serializer.Meta.model 

124 row_pk = serializer.instance.pk 

125 stored_profile_id = model.objects.filter(pk=row_pk).values_list("profile_id", flat=True).first() 

126 if stored_profile_id is None: 

127 raise Http404 

128 requested_profile = serializer.validated_data.get("profile") 

129 requested_profile_id = requested_profile.pk if requested_profile is not None else stored_profile_id 

130 try: 

131 with locked_profile_policy(stored_profile_id, requested_profile_id): 

132 serializer.instance = model.objects.get(pk=row_pk, profile_id=stored_profile_id) 

133 serializer.run_validation(serializer.initial_data) 

134 result = save_permission_scoped_object( 

135 self.request.user, 

136 model, 

137 {"pk": row_pk, "profile_id": stored_profile_id}, 

138 serializer.model_cleaned_values(), 

139 ) 

140 except (model.DoesNotExist, ImportProfile.DoesNotExist): 

141 raise Http404 from None 

142 except ObjectPermissionDenied as exc: 

143 raise PermissionDenied from exc 

144 serializer.instance = result.instance 

145 

146 def perform_destroy(self, instance): 

147 """Delete one policy row under its current profile and object scope.""" 

148 model = type(instance) 

149 profile_id = model.objects.filter(pk=instance.pk).values_list("profile_id", flat=True).first() 

150 if profile_id is None: 

151 raise Http404 

152 try: 

153 with locked_profile_policy(profile_id): 

154 deleted = delete_permission_scoped_objects( 

155 self.request.user, 

156 model.objects.filter(pk=instance.pk, profile_id=profile_id), 

157 ) 

158 if deleted != 1: 

159 raise Http404 

160 except ImportProfile.DoesNotExist: 

161 raise Http404 from None 

162 except ObjectPermissionDenied as exc: 

163 raise PermissionDenied from exc 

164 

165 

166class ColumnMappingViewSet(_PluginModelViewSet): 

167 """CRUD viewset for ColumnMapping.""" 

168 

169 queryset = ColumnMapping.objects.select_related("profile") 

170 serializer_class = ColumnMappingSerializer 

171 

172 

173class CableClassMappingViewSet(_PluginModelViewSet): 

174 """CRUD viewset for CableClassMapping.""" 

175 

176 queryset = CableClassMapping.objects.select_related("profile") 

177 serializer_class = CableClassMappingSerializer 

178 

179 

180class PortClassMappingViewSet(_PluginModelViewSet): 

181 """CRUD viewset for PortClassMapping.""" 

182 

183 queryset = PortClassMapping.objects.select_related("profile") 

184 serializer_class = PortClassMappingSerializer 

185 

186 

187class ClassRoleMappingViewSet(_PluginModelViewSet): 

188 """CRUD viewset for ClassRoleMapping.""" 

189 

190 queryset = ClassRoleMapping.objects.select_related("profile", "rack_type") 

191 serializer_class = ClassRoleMappingSerializer 

192 

193 

194class DeviceTypeMappingViewSet(_PluginModelViewSet): 

195 """CRUD viewset for DeviceTypeMapping.""" 

196 

197 queryset = DeviceTypeMapping.objects.select_related("profile") 

198 serializer_class = DeviceTypeMappingSerializer 

199 

200 

201class IgnoredDeviceViewSet(_PluginModelViewSet): 

202 """CRUD viewset for IgnoredDevice.""" 

203 

204 queryset = IgnoredDevice.objects.select_related("profile") 

205 serializer_class = IgnoredDeviceSerializer 

206 

207 

208class ColumnTransformRuleViewSet(_PluginModelViewSet): 

209 """CRUD viewset for ColumnTransformRule.""" 

210 

211 queryset = ColumnTransformRule.objects.select_related("profile") 

212 serializer_class = ColumnTransformRuleSerializer 

213 

214 

215class SourceResolutionViewSet(_PluginModelViewSet): 

216 """CRUD viewset for SourceResolution (rerere).""" 

217 

218 queryset = SourceResolution.objects.select_related("profile") 

219 serializer_class = SourceResolutionSerializer 

220 

221 

222class ImportExecutionViewSet(_ProfileScopedQuerySetMixin, viewsets.ReadOnlyModelViewSet): 

223 """Read-only viewset for the Import Execution audit history.""" 

224 

225 queryset = ImportExecution.objects.select_related("profile") 

226 serializer_class = ImportExecutionSerializer 

227 permission_classes = [permissions.IsAuthenticated, DjangoModelPermissionsWithView] 

228 

229 

230class CableImportSourceViewSet(_ProfileScopedQuerySetMixin, viewsets.ReadOnlyModelViewSet): 

231 """Read-only viewset for per-Cable import provenance.""" 

232 

233 queryset = CableImportSource.objects.select_related("cable", "profile") 

234 serializer_class = CableImportSourceSerializer 

235 permission_classes = [permissions.IsAuthenticated, DjangoModelPermissionsWithView] 

236 

237 def get_queryset(self): 

238 """Apply the profile scope and an optional Cable ID filter.""" 

239 profile_scope = ImportProfile.objects.restrict(self.request.user, "view") 

240 qs = super().get_queryset().filter(profile__in=profile_scope) 

241 cable_id = self.request.query_params.get("cable_id") 

242 if cable_id is not None: 

243 try: 

244 cable_id = int(cable_id) 

245 except (TypeError, ValueError) as exc: 

246 raise ValidationError({"cable_id": "Enter a whole number."}) from exc 

247 qs = qs.filter(cable_id=cable_id) 

248 return qs 

249 

250 

251class ResolutionProposalViewSet(_ProfileScopedQuerySetMixin, viewsets.ReadOnlyModelViewSet): 

252 """Read-only viewset for Resolution Proposal history.""" 

253 

254 queryset = ResolutionProposal.objects.select_related("profile") 

255 serializer_class = ResolutionProposalSerializer 

256 permission_classes = [permissions.IsAuthenticated, DjangoModelPermissionsWithView] 

257 

258 def get_queryset(self): 

259 """Return proposals only for profiles that the actor can view.""" 

260 profile_scope = ImportProfile.objects.restrict(self.request.user, "view") 

261 return super().get_queryset().filter(profile__in=profile_scope) 

262 

263 

264class ResolutionProposalHistoryViewSet(mixins.ListModelMixin, viewsets.GenericViewSet): 

265 """Read one field's complete paginated history with workspace permissions.""" 

266 

267 queryset = ResolutionProposal.objects.none() 

268 serializer_class = ResolutionProposalHistorySerializer 

269 permission_classes = [permissions.IsAuthenticated] 

270 

271 def get_queryset(self): 

272 """Return one profile field's history when the actor can view that profile.""" 

273 profile_values = self.request.query_params.getlist("profile_id") 

274 field_values = self.request.query_params.getlist("field_key") 

275 if len(profile_values) != 1 or not profile_values[0]: 

276 raise ValidationError({"profile_id": "Provide one whole-number profile ID."}) 

277 if len(field_values) != 1 or not field_values[0]: 

278 raise ValidationError({"field_key": "Provide one canonical termination field key."}) 

279 try: 

280 profile_id = int(profile_values[0]) 

281 except (TypeError, ValueError) as exc: 

282 raise ValidationError({"profile_id": "Provide one whole-number profile ID."}) from exc 

283 try: 

284 parse_termination_field_key(field_values[0]) 

285 except ValueError as exc: 

286 raise ValidationError({"field_key": "Provide one canonical termination field key."}) from exc 

287 if not ImportProfile.objects.restrict(self.request.user, "view").filter(pk=profile_id).exists(): 

288 raise Http404 

289 return ( 

290 ResolutionProposal.objects.filter( 

291 profile_id=profile_id, 

292 task_type=SELECT_TERMINATION_TASK, 

293 field_key=field_values[0], 

294 ) 

295 .select_related("profile") 

296 .order_by("-created", "-pk") 

297 ) 

298 

299 

300class InferenceBackendViewSet(NetBoxModelViewSet): 

301 """CRUD viewset whose configuration fields derive from the Inference Backend form.""" 

302 

303 queryset = InferenceBackend.objects.prefetch_related("tags") 

304 serializer_class = InferenceBackendSerializer 

305 filterset_class = InferenceBackendFilterSet