Coverage for cli / tests / test_request.py: 100%

1""" 

2Tests for `_request` - HTTP helper with retries via `stamina` (mocked with `respx`). 

3""" 

4 

5import json 

6from unittest.mock import patch 

7 

8import httpx 

9import pytest 

10import respx 

11 

12from covered.cli import _request 

13 

14 

15async def test_request_returns_response_on_first_success(respx_mock: respx.MockRouter): 

16 """ 

17 Single attempt, no retry; response object is returned. 

18 """ 

19 route = respx_mock.get("https://example.com").mock( 

20 return_value=httpx.Response(200, text="ok") 

21 ) 

22 

23 resp = await _request("GET", "https://example.com") 

24 

25 assert resp.status_code == 200 

26 assert resp.text == "ok" 

27 assert route.call_count == 1 

28 

29 

30async def test_request_retries_on_transport_error_then_succeeds( 

31 respx_mock: respx.MockRouter, 

32): 

33 """ 

34 First call raises `httpx.TransportError`; retry succeeds and returns the response. 

35 """ 

36 route = respx_mock.get("https://example.com").mock( 

37 side_effect=[httpx.ConnectError("boom"), httpx.Response(200)] 

38 ) 

39 

40 resp = await _request("GET", "https://example.com") 

41 

42 assert resp.status_code == 200 

43 assert route.call_count == 2 

44 

45 

46async def test_request_retries_on_5xx_then_succeeds(respx_mock: respx.MockRouter): 

47 """ 

48 First response is 503; retry returns 200 and that response is returned. 

49 """ 

50 route = respx_mock.get("https://example.com").mock( 

51 side_effect=[httpx.Response(503), httpx.Response(200)] 

52 ) 

53 

54 resp = await _request("GET", "https://example.com") 

55 

56 assert resp.status_code == 200 

57 assert route.call_count == 2 

58 

59 

60async def test_request_does_not_retry_on_4xx(respx_mock: respx.MockRouter): 

61 """ 

62 A 404 response is returned without retry - only 5xx status triggers `raise_for_status`. 

63 """ 

64 route = respx_mock.get("https://example.com").mock(return_value=httpx.Response(404)) 

65 

66 resp = await _request("GET", "https://example.com") 

67 

68 assert resp.status_code == 404 

69 assert route.call_count == 1 

70 

71 

72async def test_request_gives_up_after_three_attempts(respx_mock: respx.MockRouter): 

73 """ 

74 Three consecutive failures raise. 

75 """ 

76 route = respx_mock.get("https://example.com").mock(return_value=httpx.Response(500)) 

77 

78 with pytest.raises(httpx.HTTPStatusError): 

79 await _request("GET", "https://example.com") 

80 

81 assert route.call_count == 3 

82 

83 

84async def test_request_passes_parameters(respx_mock: respx.MockRouter): 

85 """ 

86 Parameters like `headers` and `json` are forwarded to the underlying httpx request. 

87 """ 

88 route = respx_mock.post("https://example.com/api").mock( 

89 return_value=httpx.Response(200) 

90 ) 

91 

92 headers = {"X-Custom": "value", "Authorization": "Bearer abc"} 

93 body = {"key": "value", "num": 42} 

94 

95 await _request("POST", "https://example.com/api", headers=headers, json=body) 

96 

97 request = route.calls.last.request 

98 assert request.headers["X-Custom"] == "value" 

99 assert request.headers["Authorization"] == "Bearer abc" 

100 assert json.loads(request.content) == body 

101 

102 

103async def test_request_uses_configured_timeout(respx_mock: respx.MockRouter): 

104 """ 

105 The custom `timeout` value is propagated to the underlying `AsyncClient`. 

106 """ 

107 respx_mock.get("https://example.com").mock(return_value=httpx.Response(200)) 

108 

109 with patch.object(httpx, "AsyncClient", wraps=httpx.AsyncClient) as spy: 

110 await _request("GET", "https://example.com", timeout=5) 

111 

112 spy.assert_called_once_with(timeout=5)