Coverage for tests / test_error_handling.py: 100%

1# SPDX-FileCopyrightText: 2025-2026 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5"""Unit tests for error handling edge cases using mocks.""" 

6 

7from unittest.mock import MagicMock, patch 

8 

9import pycurl 

10import pytest 

11 

12from sparqlite import EndpointError, SPARQLClient 

13 

14 

15class TestHTTPErrorHandling: 

16 """Tests for HTTP error handling.""" 

17 

18 def test_server_error_500_triggers_retry(self): 

19 """Test that 500 errors trigger retry logic.""" 

20 mock_curl = MagicMock() 

21 mock_curl.getinfo.return_value = 500 

22 

23 call_count = 0 

24 

25 def mock_perform(): 

26 nonlocal call_count 

27 call_count += 1 

28 

29 mock_curl.perform = mock_perform 

30 

31 with patch("pycurl.Curl", return_value=mock_curl): 

32 client = SPARQLClient("http://example.org/sparql", max_retries=2) 

33 client._curl = mock_curl 

34 

35 with pytest.raises(EndpointError) as exc_info: 

36 client._request("SELECT * WHERE { ?s ?p ?o }", "application/json") 

37 

38 assert "500" in str(exc_info.value) 

39 assert exc_info.value.status_code == 500 

40 assert call_count == 3 

41 

42 client.close() 

43 

44 def test_http_403_forbidden(self): 

45 """Test that 403 errors raise EndpointError without retry.""" 

46 mock_curl = MagicMock() 

47 mock_curl.getinfo.return_value = 403 

48 

49 buffer_content = b"Forbidden" 

50 

51 def mock_setopt(opt, val): 

52 if opt == pycurl.WRITEDATA: 

53 val.write(buffer_content) 

54 

55 mock_curl.setopt = mock_setopt 

56 

57 with patch("pycurl.Curl", return_value=mock_curl): 

58 client = SPARQLClient("http://example.org/sparql") 

59 client._curl = mock_curl 

60 

61 with pytest.raises(EndpointError) as exc_info: 

62 client._request("SELECT * WHERE { ?s ?p ?o }", "application/json") 

63 

64 assert exc_info.value.status_code == 403 

65 assert "403" in str(exc_info.value) 

66 

67 client.close() 

68 

69 

70class TestPycurlErrorHandling: 

71 """Tests for pycurl error handling.""" 

72 

73 def test_timeout_error(self): 

74 """Test that timeout errors are handled.""" 

75 mock_curl = MagicMock() 

76 mock_curl.perform.side_effect = pycurl.error( 

77 pycurl.E_OPERATION_TIMEDOUT, "Operation timed out" 

78 ) 

79 

80 with patch("pycurl.Curl", return_value=mock_curl): 

81 client = SPARQLClient("http://example.org/sparql", max_retries=0) 

82 client._curl = mock_curl 

83 

84 with pytest.raises(EndpointError) as exc_info: 

85 client._request("SELECT * WHERE { ?s ?p ?o }", "application/json") 

86 

87 assert "Timeout" in str(exc_info.value) 

88 

89 client.close() 

90 

91 def test_resolve_host_error(self): 

92 """Test that host resolution errors are handled.""" 

93 mock_curl = MagicMock() 

94 mock_curl.perform.side_effect = pycurl.error( 

95 pycurl.E_COULDNT_RESOLVE_HOST, "Could not resolve host" 

96 ) 

97 

98 with patch("pycurl.Curl", return_value=mock_curl): 

99 client = SPARQLClient("http://nonexistent.invalid/sparql", max_retries=0) 

100 client._curl = mock_curl 

101 

102 with pytest.raises(EndpointError) as exc_info: 

103 client._request("SELECT * WHERE { ?s ?p ?o }", "application/json") 

104 

105 assert "Connection error" in str(exc_info.value) 

106 

107 client.close() 

108 

109 def test_generic_pycurl_error(self): 

110 """Test that generic pycurl errors are handled.""" 

111 mock_curl = MagicMock() 

112 mock_curl.perform.side_effect = pycurl.error( 

113 pycurl.E_SSL_CONNECT_ERROR, "SSL connection error" 

114 ) 

115 

116 with patch("pycurl.Curl", return_value=mock_curl): 

117 client = SPARQLClient("https://example.org/sparql", max_retries=0) 

118 client._curl = mock_curl 

119 

120 with pytest.raises(EndpointError) as exc_info: 

121 client._request("SELECT * WHERE { ?s ?p ?o }", "application/json") 

122 

123 assert "Request error" in str(exc_info.value) 

124 

125 client.close() 

126 

127 

128class TestTimeoutConfiguration: 

129 """Tests for timeout configuration.""" 

130 

131 def test_timeout_parameter_stored(self): 

132 """Test that timeout parameter is stored correctly.""" 

133 with patch("pycurl.Curl") as mock_curl_class: 

134 mock_curl = MagicMock() 

135 mock_curl_class.return_value = mock_curl 

136 

137 client = SPARQLClient("http://example.org/sparql", timeout=30.0) 

138 assert client.timeout == 30.0 

139 client.close() 

140 

141 def test_timeout_default_is_none(self): 

142 """Test that timeout defaults to None.""" 

143 with patch("pycurl.Curl") as mock_curl_class: 

144 mock_curl = MagicMock() 

145 mock_curl_class.return_value = mock_curl 

146 

147 client = SPARQLClient("http://example.org/sparql") 

148 assert client.timeout is None 

149 client.close() 

150 

151 def test_timeout_ms_set_when_timeout_provided(self): 

152 """Test that TIMEOUT_MS is set when timeout is provided.""" 

153 mock_curl = MagicMock() 

154 mock_curl.getinfo.return_value = 200 

155 

156 setopt_calls = [] 

157 

158 def track_setopt(opt, val): 

159 setopt_calls.append((opt, val)) 

160 

161 mock_curl.setopt = track_setopt 

162 

163 with patch("pycurl.Curl", return_value=mock_curl): 

164 client = SPARQLClient("http://example.org/sparql", timeout=5.5) 

165 client._curl = mock_curl 

166 client._request("SELECT * WHERE { ?s ?p ?o }", "application/json") 

167 client.close() 

168 

169 timeout_calls = [(opt, val) for opt, val in setopt_calls if opt == pycurl.TIMEOUT_MS] 

170 assert timeout_calls == [(pycurl.TIMEOUT_MS, 5500)] 

171 

172 def test_timeout_ms_not_set_when_timeout_none(self): 

173 """Test that TIMEOUT_MS is not set when timeout is None.""" 

174 mock_curl = MagicMock() 

175 mock_curl.getinfo.return_value = 200 

176 

177 setopt_calls = [] 

178 

179 def track_setopt(opt, val): 

180 setopt_calls.append((opt, val)) 

181 

182 mock_curl.setopt = track_setopt 

183 

184 with patch("pycurl.Curl", return_value=mock_curl): 

185 client = SPARQLClient("http://example.org/sparql") 

186 client._curl = mock_curl 

187 client._request("SELECT * WHERE { ?s ?p ?o }", "application/json") 

188 client.close() 

189 

190 timeout_calls = [opt for opt, _ in setopt_calls if opt == pycurl.TIMEOUT_MS] 

191 assert timeout_calls == [] 

192 

193 

194class TestHTTPMethod: 

195 

196 def test_post_method_uses_postfields(self): 

197 mock_curl = MagicMock() 

198 mock_curl.getinfo.return_value = 200 

199 setopt_calls: list[tuple[int, object]] = [] 

200 

201 def track_setopt(opt: int, val: object) -> None: 

202 setopt_calls.append((opt, val)) 

203 

204 mock_curl.setopt = track_setopt 

205 

206 with patch("pycurl.Curl", return_value=mock_curl): 

207 client = SPARQLClient("http://example.org/sparql") 

208 client._curl = mock_curl 

209 client._request("SELECT ?s WHERE { ?s ?p ?o }", "application/json", method="POST") 

210 client.close() 

211 

212 assert any(opt == pycurl.POSTFIELDS for opt, _ in setopt_calls) 

213 assert not any(opt == pycurl.HTTPGET for opt, _ in setopt_calls) 

214 

215 def test_request_on_closed_client_raises(self): 

216 with patch("pycurl.Curl") as mock_curl_class: 

217 mock_curl_class.return_value = MagicMock() 

218 client = SPARQLClient("http://example.org/sparql") 

219 client.close() 

220 

221 with pytest.raises(EndpointError): 

222 client._request("SELECT ?s WHERE { ?s ?p ?o }", "application/json")