Coverage for backend/core/client.py: 83%

1from rest_framework.test import APIClient 

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

3import json 

4 

5from CoreRoot import settings 

6from authentication.user.models import User 

7 

8class FlowsheetClient(APIClient): 

9 """ 

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

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

12 Used in tests. 

13 """ 

14 

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

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

17 self.flowsheet = flowsheet 

18 additional_headers = { 

19 settings.REMOTE_USER_HEADER: user.username, 

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

21 } 

22 self.credentials(**additional_headers) 

23 

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

25 url_parts = list(urlparse(path)) 

26 query = parse_qs(url_parts[4]) 

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

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

29 return urlunparse(url_parts) 

30 

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

32 path = self._add_flowsheet_query_param(path) 

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

34 

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

36 path = self._add_flowsheet_query_param(path) 

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

38 data = json.dumps(data) 

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

40 

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

42 path = self._add_flowsheet_query_param(path) 

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

44 data = json.dumps(data) 

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

46 

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

48 path = self._add_flowsheet_query_param(path) 

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

50 data = json.dumps(data) 

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

52 

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

54 path = self._add_flowsheet_query_param(path) 

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

56 data = json.dumps(data) 

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