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

1import json 

2 

3import httpx 

4from fastapi.testclient import TestClient 

5 

6 

7class RestlyTestClient(TestClient): 

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

9 

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

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

12 __tracebackhide__ = True 

13 

14 status_code = response.status_code 

15 

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

17 return # All good 

18 

19 if expected_code is None and status_code < 400: 

20 return # Also fine 

21 

22 # Raise AssertionError with detailed error message 

23 try: 

24 response_content = response.json() 

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

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

27 

28 content_str_raw = str(response_content) 

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

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

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

32 

33 # Safe method/URL extraction 

34 try: 

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

36 url = str(response.request.url) 

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

38 except Exception: 

39 request_info = "(request info unavailable)" 

40 

41 raise AssertionError( 

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

43 f"{content_str}" 

44 ) 

45 

46 def get(self, url: str, *, assert_status_code: int | None = 200, **kwargs) -> httpx.Response: 

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

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

49 __tracebackhide__ = True 

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

51 self.assert_status(response, assert_status_code) 

52 return response 

53 

54 def post(self, url: str, *, assert_status_code: int | None = 201, **kwargs) -> httpx.Response: 

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

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

57 __tracebackhide__ = True 

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

59 self.assert_status(response, assert_status_code) 

60 return response 

61 

62 def put(self, url: str, *, assert_status_code: int | None = 200, **kwargs) -> httpx.Response: 

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

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

65 __tracebackhide__ = True 

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

67 self.assert_status(response, assert_status_code) 

68 return response 

69 

70 def patch(self, url: str, *, assert_status_code: int | None = 200, **kwargs) -> httpx.Response: 

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

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

73 __tracebackhide__ = True 

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

75 self.assert_status(response, assert_status_code) 

76 return response 

77 

78 def delete(self, url: str, *, assert_status_code: int | None = 204, **kwargs) -> httpx.Response: 

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

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

81 __tracebackhide__ = True 

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

83 self.assert_status(response, assert_status_code) 

84 return response