Coverage for extras/api/mixins.py: 71%

1from django.utils.translation import gettext_lazy as _ 

2from drf_spectacular.utils import OpenApiResponse, OpenApiTypes, extend_schema 

3from rest_framework.decorators import action 

4from rest_framework.renderers import JSONRenderer 

5from rest_framework.response import Response 

6from rest_framework.status import HTTP_400_BAD_REQUEST, HTTP_500_INTERNAL_SERVER_ERROR 

7 

8from extras.models import ConfigTemplate 

9from netbox.api.authentication import TokenWritePermission 

10from netbox.api.renderers import TextRenderer 

11 

12from .serializers import RenderConfigInputSerializer, RenderedConfigSerializer 

13 

14__all__ = ( 

15 'ConfigContextQuerySetMixin', 

16 'ConfigTemplateRenderMixin', 

17 'RenderConfigMixin', 

18 'SharedObjectQuerySetMixin', 

19) 

20 

21 

22class SharedObjectQuerySetMixin: 

23 """ 

24 Restrict the queryset to shared objects, or those owned by the current user, unless the user is a superuser. 

25 This mirrors the visibility enforced in the UI by extras.utils.SharedObjectViewMixin. 

26 """ 

27 def get_queryset(self): 

28 return super().get_queryset().restrict_to_shared(self.request.user) 

29 

30 

31class ConfigContextQuerySetMixin: 

32 """ 

33 Used by viewsets for config context models (Device, VirtualMachine). 

34 

35 For non-brief requests, annotates the queryset so that config context data is computed in a 

36 single query for any object whose pre-rendered cache (`_config_context_data`) has been 

37 invalidated (NULL). Objects with a warm cache are served directly from it by 

38 ConfigContextModel.get_config_context() and incur no subquery — PostgreSQL short-circuits the 

39 CASE, so the correlated aggregation runs only for the invalidated rows. This avoids the 

40 per-object fallback query that would otherwise occur when listing objects with cold caches 

41 (e.g. immediately following an upgrade or a broad invalidation). 

42 """ 

43 def get_queryset(self): 

44 queryset = super().get_queryset() 

45 # Brief responses omit config_context entirely, so the annotation would be pure overhead. 

46 if self.brief: 

47 return queryset 

48 return queryset.annotate_config_context_data(only_invalidated=True) 

49 

50 

51class ConfigTemplateRenderMixin: 

52 """ 

53 Provides a method to return a rendered ConfigTemplate as REST API data. 

54 """ 

55 def render_configtemplate(self, request, configtemplate, context): 

56 try: 

57 output = configtemplate.render(context=context) 

58 except Exception as e: 

59 detail = configtemplate.format_render_error(e) 

60 if request.accepted_renderer.format == 'txt': 

61 return Response(detail, status=HTTP_500_INTERNAL_SERVER_ERROR) 

62 return Response({'detail': detail}, status=HTTP_500_INTERNAL_SERVER_ERROR) 

63 

64 # If the client has requested "text/plain", return the raw content. 

65 if request.accepted_renderer.format == 'txt': 

66 return Response(output) 

67 

68 serializer = RenderedConfigSerializer( 

69 instance={'configtemplate': configtemplate, 'content': output}, 

70 context={'request': request}, 

71 ) 

72 return Response(serializer.data) 

73 

74 

75class RenderConfigMixin(ConfigTemplateRenderMixin): 

76 """ 

77 Provides a /render-config/ endpoint for REST API views whose model may have a ConfigTemplate assigned. 

78 """ 

79 

80 def get_permissions(self): 

81 # For render_config action, check only token write ability (not model permissions) 

82 if self.action == 'render_config': 

83 return [TokenWritePermission()] 

84 return super().get_permissions() 

85 

86 @extend_schema( 

87 request=RenderConfigInputSerializer, 

88 responses={ 

89 200: OpenApiResponse( 

90 response=RenderedConfigSerializer, 

91 description=_( 

92 "The rendered config template. When the client requests `text/plain`, the raw " 

93 "rendered content is returned in place of the JSON object." 

94 ), 

95 ), 

96 400: OpenApiResponse( 

97 response=OpenApiTypes.OBJECT, 

98 description=_("No config template could be resolved for this object."), 

99 ), 

100 500: OpenApiResponse( 

101 response=OpenApiTypes.OBJECT, 

102 description=_("An error occurred while rendering the config template."), 

103 ), 

104 }, 

105 ) 

106 @action(detail=True, methods=['post'], url_path='render-config', renderer_classes=[JSONRenderer, TextRenderer]) 

107 def render_config(self, request, pk): 

108 """ 

109 Resolve and render the preferred ConfigTemplate for this Device or Virtual Machine. 

110 """ 

111 # Override restrict() on the default queryset to enforce the render_config & view actions 

112 self.queryset = self.queryset.model.objects.restrict(request.user, 'render_config').restrict( 

113 request.user, 'view' 

114 ) 

115 instance = self.get_object() 

116 

117 object_type = instance._meta.model_name 

118 

119 # Check for an optional config_template_id override in the request data 

120 if config_template_id := request.data.get('config_template_id'): 

121 try: 

122 configtemplate = ConfigTemplate.objects.restrict(request.user, 'view').get(pk=config_template_id) 

123 except (ConfigTemplate.DoesNotExist, ValueError): 

124 return Response({ 

125 'error': _('Config template with ID {id} not found.').format(id=config_template_id) 

126 }, status=HTTP_400_BAD_REQUEST) 

127 else: 

128 configtemplate = instance.get_config_template() 

129 if not configtemplate: 129 ↛ 135line 129 didn't jump to line 135 because the condition on line 129 was always true

130 return Response({ 

131 'error': _('No config template found for this {object_type}.').format(object_type=object_type) 

132 }, status=HTTP_400_BAD_REQUEST) 

133 

134 # Compile context data 

135 context_data = instance.get_config_context() 

136 context_data.update({k: v for k, v in request.data.items() if k != 'config_template_id'}) 

137 context_data.update({object_type: instance}) 

138 

139 return self.render_configtemplate(request, configtemplate, context_data)