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

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 if data is not None: 

36 data = data.copy() 

37 data["flowsheet"] = self.flowsheet 

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

39 

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

41 path = self._add_flowsheet_query_param(path) 

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

43 data = json.dumps(data) 

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

45 

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

47 path = self._add_flowsheet_query_param(path) 

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

49 data = json.dumps(data) 

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

51 

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

53 path = self._add_flowsheet_query_param(path) 

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

55 data = json.dumps(data) 

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

57 

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

59 path = self._add_flowsheet_query_param(path) 

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

61 data = json.dumps(data) 

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