Coverage for fastapi_restly / testing / _client.py: 90%

1import json 

2 

3import httpx 

4from fastapi.testclient import TestClient 

5 

6# httpx accepts either a string or a `httpx.URL` for request URLs. The base 

7# class' `URLTypes` alias is private, so we replicate the public union here. 

8URLTypes = httpx.URL | str 

9 

10 

11class RestlyTestClient(TestClient): 

12 """Custom TestClient that automatically checks response codes and provides clear error messages.""" 

13 

14 def assert_status(self, response: httpx.Response, expected_code: int | None = None): 

15 """Check if the response status code matches the expected code.""" 

16 __tracebackhide__ = True 

17 

18 status_code = response.status_code 

19 

20 if expected_code is not None and status_code == expected_code: 

21 return # All good 

22 

23 if expected_code is None and status_code < 400: 

24 return # Also fine 

25 

26 # Raise AssertionError with detailed error message 

27 try: 

28 response_content = response.json() 

29 except (ValueError, TypeError, json.JSONDecodeError): 

30 response_content = response.content.decode(errors="ignore") 

31 

32 content_str_raw = str(response_content) 

33 if len(content_str_raw) > 1000: 33 ↛ 34line 33 didn't jump to line 34 because the condition on line 33 was never true

34 content_str_raw = content_str_raw[:1000] + "...(truncated)" 

35 content_str = f"Response JSON: {content_str_raw}" 

36 

37 # Safe method/URL extraction 

38 try: 

39 method = response.request.method.upper() 

40 url = str(response.request.url) 

41 request_info = f"{method} {url}" 

42 except Exception: 

43 request_info = "(request info unavailable)" 

44 

45 raise AssertionError( 

46 f"Expected {request_info} to return {expected_code}, got {status_code}\n" 

47 f"{content_str}" 

48 ) 

49 

50 def get( 

51 self, url: URLTypes, *, assert_status_code: int | None = 200, **kwargs 

52 ) -> httpx.Response: 

53 """Make a GET request. Asserts the response status code matches `assert_status_code` (default: 200). 

54 Pass `assert_status_code=None` to skip the assertion.""" 

55 __tracebackhide__ = True 

56 response = super().get(url, **kwargs) 

57 self.assert_status(response, assert_status_code) 

58 return response 

59 

60 def post( 

61 self, url: URLTypes, *, assert_status_code: int | None = 201, **kwargs 

62 ) -> httpx.Response: 

63 """Make a POST request. Asserts the response status code matches `assert_status_code` (default: 201). 

64 Pass `assert_status_code=None` to skip the assertion.""" 

65 __tracebackhide__ = True 

66 response = super().post(url, **kwargs) 

67 self.assert_status(response, assert_status_code) 

68 return response 

69 

70 def put( 

71 self, url: URLTypes, *, assert_status_code: int | None = 200, **kwargs 

72 ) -> httpx.Response: 

73 """Make a PUT request. Asserts the response status code matches `assert_status_code` (default: 200). 

74 Pass `assert_status_code=None` to skip the assertion.""" 

75 __tracebackhide__ = True 

76 response = super().put(url, **kwargs) 

77 self.assert_status(response, assert_status_code) 

78 return response 

79 

80 def patch( 

81 self, url: URLTypes, *, assert_status_code: int | None = 200, **kwargs 

82 ) -> httpx.Response: 

83 """Make a PATCH request. Asserts the response status code matches `assert_status_code` (default: 200). 

84 Pass `assert_status_code=None` to skip the assertion.""" 

85 __tracebackhide__ = True 

86 response = super().patch(url, **kwargs) 

87 self.assert_status(response, assert_status_code) 

88 return response 

89 

90 def delete( 

91 self, url: URLTypes, *, assert_status_code: int | None = 204, **kwargs 

92 ) -> httpx.Response: 

93 """Make a DELETE request. Asserts the response status code matches `assert_status_code` (default: 204). 

94 Pass `assert_status_code=None` to skip the assertion.""" 

95 __tracebackhide__ = True 

96 response = super().delete(url, **kwargs) 

97 self.assert_status(response, assert_status_code) 

98 return response