Coverage for backend/django/core/client.py: 84%

1from rest_framework.test import APIClient 

2from urllib.parse import urlencode, urlparse, parse_qs, urlunparse 

3import json 

4 

5from CoreRoot import settings 

6from authentication.token_helpers import build_human_user_access_token 

7from authentication.user.models import User 

8 

9class FlowsheetClient(APIClient): 

10 """ 

11 A custom APIClient that automatically includes the flowsheet ID in each request 

12 as a query parameter, and sets the appropriate user headers for authentication.  

13 Used in tests. 

14 """ 

15 

16 def __init__(self, *args, flowsheet: int, user: User, **kwargs): 

17 super().__init__(*args, **kwargs) 

18 self.flowsheet = flowsheet 

19 additional_headers = { 

20 settings.REMOTE_USER_HEADER: user.username, 

21 settings.REMOTE_USER_EMAIL_HEADER: user.email if user.email else '', 

22 settings.REMOTE_USER_ACCESS_TOKEN_HEADER: build_human_user_access_token(), 

23 } 

24 self.credentials(**additional_headers) 

25 

26 def _add_flowsheet_query_param(self, path: str) -> str: 

27 url_parts = list(urlparse(path)) 

28 query = parse_qs(url_parts[4]) 

29 query['flowsheet'] = [self.flowsheet] 

30 url_parts[4] = urlencode(query, doseq=True) 

31 return urlunparse(url_parts) 

32 

33 def get(self, path, data=None, **extra): 

34 path = self._add_flowsheet_query_param(path) 

35 return super().get(path, data=data, **extra) 

36 

37 def post(self, path, data=None, content_type="application/json", **extra): 

38 path = self._add_flowsheet_query_param(path) 

39 if content_type == "application/json" and isinstance(data, dict): 

40 data = json.dumps(data) 

41 return super().post(path, data=data, content_type=content_type, **extra) 

42 

43 def patch(self, path, data=None, content_type="application/json", **extra): 

44 path = self._add_flowsheet_query_param(path) 

45 if content_type == "application/json" and isinstance(data, dict): 

46 data = json.dumps(data) 

47 return super().patch(path, data=data, content_type=content_type, **extra) 

48 

49 def put(self, path, data=None, content_type="application/json", **extra): 

50 path = self._add_flowsheet_query_param(path) 

51 if content_type == "application/json" and isinstance(data, dict): 

52 data = json.dumps(data) 

53 return super().put(path, data=data, content_type=content_type, **extra) 

54 

55 def delete(self, path, data=None, content_type="application/json", **extra): 

56 path = self._add_flowsheet_query_param(path) 

57 if content_type == "application/json" and isinstance(data, dict): 57 ↛ 58line 57 didn't jump to line 58 because the condition on line 57 was never true

58 data = json.dumps(data) 

59 return super().delete(path, data=data, content_type=content_type, **extra)