|
3 | 3 | # standard library |
4 | 4 | import time |
5 | 5 | from json import JSONDecodeError |
| 6 | +from requests.models import Response |
6 | 7 | from unittest.mock import MagicMock, patch |
7 | 8 |
|
8 | 9 | # first party |
@@ -41,6 +42,8 @@ def localSetUp(self): |
41 | 42 | # use the local instance of the Epidata API |
42 | 43 | Epidata.BASE_URL = 'http://delphi_web_epidata/epidata' |
43 | 44 | Epidata.auth = ('epidata', 'key') |
| 45 | + Epidata.debug = False |
| 46 | + Epidata.sandbox = False |
44 | 47 |
|
45 | 48 | # use the local instance of the epidata database |
46 | 49 | secrets.db.host = 'delphi_database_epidata' |
@@ -221,6 +224,82 @@ def test_retry_request(self, get): |
221 | 224 | {'result': 0, 'message': 'error: Expecting value: line 1 column 1 (char 0)'} |
222 | 225 | ) |
223 | 226 |
|
| 227 | + @patch('requests.post') |
| 228 | + @patch('requests.get') |
| 229 | + def test_debug(self, get, post): |
| 230 | + """Test that in debug mode request params are correctly logged.""" |
| 231 | + class MockResponse: |
| 232 | + def __init__(self, content, status_code): |
| 233 | + self.content = content |
| 234 | + self.status_code = status_code |
| 235 | + def raise_for_status(self): pass |
| 236 | + |
| 237 | + Epidata.debug = True |
| 238 | + |
| 239 | + try: |
| 240 | + with self.subTest(name='test multiple GET'): |
| 241 | + with self.assertLogs('delphi_epidata_client', level='INFO') as logs: |
| 242 | + get.reset_mock() |
| 243 | + get.return_value = MockResponse(b'{"key": "value"}', 200) |
| 244 | + Epidata._request_with_retry("test_endpoint1", params={"key1": "value1"}) |
| 245 | + Epidata._request_with_retry("test_endpoint2", params={"key2": "value2"}) |
| 246 | + |
| 247 | + output = logs.output |
| 248 | + self.assertEqual(len(output), 4) # [request, response, request, response] |
| 249 | + self.assertIn("Sending GET request", output[0]) |
| 250 | + self.assertIn("\"url\": \"http://delphi_web_epidata/epidata/test_endpoint1/\"", output[0]) |
| 251 | + self.assertIn("\"params\": {\"key1\": \"value1\"}", output[0]) |
| 252 | + self.assertIn("Received response", output[1]) |
| 253 | + self.assertIn("\"status_code\": 200", output[1]) |
| 254 | + self.assertIn("\"len\": 16", output[1]) |
| 255 | + self.assertIn("Sending GET request", output[2]) |
| 256 | + self.assertIn("\"url\": \"http://delphi_web_epidata/epidata/test_endpoint2/\"", output[2]) |
| 257 | + self.assertIn("\"params\": {\"key2\": \"value2\"}", output[2]) |
| 258 | + self.assertIn("Received response", output[3]) |
| 259 | + self.assertIn("\"status_code\": 200", output[3]) |
| 260 | + self.assertIn("\"len\": 16", output[3]) |
| 261 | + |
| 262 | + with self.subTest(name='test GET and POST'): |
| 263 | + with self.assertLogs('delphi_epidata_client', level='INFO') as logs: |
| 264 | + get.reset_mock() |
| 265 | + get.return_value = MockResponse(b'{"key": "value"}', 414) |
| 266 | + post.reset_mock() |
| 267 | + post.return_value = MockResponse(b'{"key": "value"}', 200) |
| 268 | + Epidata._request_with_retry("test_endpoint3", params={"key3": "value3"}) |
| 269 | + |
| 270 | + output = logs.output |
| 271 | + self.assertEqual(len(output), 3) # [request, response, request, response] |
| 272 | + self.assertIn("Sending GET request", output[0]) |
| 273 | + self.assertIn("\"url\": \"http://delphi_web_epidata/epidata/test_endpoint3/\"", output[0]) |
| 274 | + self.assertIn("\"params\": {\"key3\": \"value3\"}", output[0]) |
| 275 | + self.assertIn("Received 414 response, retrying as POST request", output[1]) |
| 276 | + self.assertIn("\"url\": \"http://delphi_web_epidata/epidata/test_endpoint3/\"", output[1]) |
| 277 | + self.assertIn("\"params\": {\"key3\": \"value3\"}", output[1]) |
| 278 | + self.assertIn("Received response", output[2]) |
| 279 | + self.assertIn("\"status_code\": 200", output[2]) |
| 280 | + self.assertIn("\"len\": 16", output[2]) |
| 281 | + finally: # make sure this global is always reset |
| 282 | + Epidata.debug = False |
| 283 | + |
| 284 | + @patch('requests.post') |
| 285 | + @patch('requests.get') |
| 286 | + def test_sandbox(self, get, post): |
| 287 | + """Test that in debug + sandbox mode request params are correctly logged, but no queries are sent.""" |
| 288 | + Epidata.debug = True |
| 289 | + Epidata.sandbox = True |
| 290 | + try: |
| 291 | + with self.assertLogs('delphi_epidata_client', level='INFO') as logs: |
| 292 | + Epidata.covidcast('src', 'sig', 'day', 'county', 20200414, '01234') |
| 293 | + output = logs.output |
| 294 | + self.assertEqual(len(output), 1) |
| 295 | + self.assertIn("Sending GET request", output[0]) |
| 296 | + self.assertIn("\"url\": \"http://delphi_web_epidata/epidata/covidcast/\"", output[0]) |
| 297 | + get.assert_not_called() |
| 298 | + post.assert_not_called() |
| 299 | + finally: # make sure these globals are always reset |
| 300 | + Epidata.debug = False |
| 301 | + Epidata.sandbox = False |
| 302 | + |
224 | 303 | def test_geo_value(self): |
225 | 304 | """test different variants of geo types: single, *, multi.""" |
226 | 305 |
|
|
0 commit comments