Coverage for backend/django/core/viewset.py: 93%

1from rest_framework import viewsets 

2from rest_framework.permissions import SAFE_METHODS 

3 

4STANDARD_WRITE_ACTIONS = {"create", "update", "partial_update", "destroy"} 

5 

6 

7class HistoricalRevisionWriteGuardMixin: 

8 """Reject revision-scoped writes before a plain DRF viewset action runs.""" 

9 

10 def initial(self, request, *args, **kwargs): 

11 from core.validation import reject_historical_revision_write 

12 

13 reject_historical_revision_write(request) 

14 return super().initial(request, *args, **kwargs) 

15 

16 

17class ContextMixin: 

18 """ 

19 A mixin that provides the flowsheet context to all views to enforce access control. 

20 """ 

21 def initial(self, request, *args, **kwargs): 

22 response = super().initial(request, *args, **kwargs) 

23 

24 from core.validation import reject_historical_revision_write 

25 

26 reject_historical_revision_write(request) 

27 action = getattr(self, "action", None) 

28 if action is not None: 28 ↛ 33line 28 didn't jump to line 33 because the condition on line 28 was always true

29 view_method = getattr(self, action, None) 

30 if hasattr(view_method, 'ignore_access_control') and view_method.ignore_access_control: 

31 return response 

32 

33 if hasattr(request, "query_params"): 33 ↛ 37line 33 didn't jump to line 37 because the condition on line 33 was always true

34 flowsheet_id = request.query_params.get("flowsheet") 

35 revision_state_id = request.query_params.get("revision") 

36 else: 

37 flowsheet_id = request.GET.get("flowsheet") 

38 revision_state_id = request.GET.get("revision") 

39 

40 from core.validation import sanitize_flowsheet_id, flowsheet_context 

41 

42 flowsheet_id = sanitize_flowsheet_id(flowsheet_id) 

43 action = getattr(self, "action", None) 

44 write_intent = request.method not in SAFE_METHODS 

45 enforce_write_access = write_intent and action not in STANDARD_WRITE_ACTIONS 

46 self.flowsheet_context = flowsheet_context( 

47 flowsheet_id, 

48 request.user, 

49 revision_state_id=revision_state_id, 

50 write_intent=write_intent, 

51 enforce_write_access=enforce_write_access, 

52 ) 

53 # start the context 

54 try: 

55 self.flowsheet_token = self.flowsheet_context.__enter__() 

56 except Exception: 

57 del self.flowsheet_context 

58 raise 

59 

60 return response 

61 

62 def finalize_response(self, request, response, *args, **kwargs): 

63 # Clean up context when the response is finalized 

64 if hasattr(self, 'flowsheet_context'): 

65 self.flowsheet_context.__exit__(None, None, None) 

66 

67 return super().finalize_response(request, response, *args, **kwargs) 

68 

69 def handle_exception(self, exc): 

70 # Clean up context in case of exception 

71 if hasattr(self, 'flowsheet_context'): 

72 self.flowsheet_context.__exit__(type(exc), exc, exc.__traceback__) 

73 

74 return super().handle_exception(exc) 

75 

76class ReadOnlyModelViewSet(ContextMixin, viewsets.ReadOnlyModelViewSet): 

77 """ 

78 Use this model viewset for read-only operations instead of the default one to enforce access control. 

79 """ 

80 pass 

81 

82class ModelViewSet(ContextMixin, viewsets.ModelViewSet): 

83 """ 

84 Use this model viewset instead of the default one to enforce access control. 

85 """ 

86 pass