|
19 | 19 | from tests import utils |
20 | 20 | from tuf import exceptions, unittest_toolbox |
21 | 21 | from tuf.ngclient._internal.requests_fetcher import RequestsFetcher |
22 | | -from unittest.mock import patch |
| 22 | +from unittest.mock import Mock, patch |
23 | 23 |
|
24 | 24 | logger = logging.getLogger(__name__) |
25 | 25 |
|
@@ -110,19 +110,17 @@ def test_http_error(self): |
110 | 110 | self.fetcher.fetch(self.url) |
111 | 111 | self.assertEqual(cm.exception.status_code, 404) |
112 | 112 |
|
113 | | - # Read timeout error |
114 | | - def test_read_timeout(self): |
115 | | - # Reduce the read socket timeout to speed up the test |
116 | | - # while keeping the connect timeout |
117 | | - default_socket_timeout = self.fetcher.socket_timeout |
118 | | - self.fetcher.socket_timeout = (default_socket_timeout, 0.1) |
119 | | - # Launch a new "slow retrieval" server sending one byte each 40s |
120 | | - slow_server_process_handler = utils.TestServerProcess(log=logger, server='slow_retrieval_server.py') |
121 | | - self.url = f"http://{utils.TEST_HOST_ADDRESS}:{str(slow_server_process_handler.port)}/{self.rel_target_filepath}" |
| 113 | + # Response read timeout error |
| 114 | + @patch.object(requests.Session, 'get') |
| 115 | + def test_response_read_timeout(self, mock_session_get): |
| 116 | + mock_response = Mock() |
| 117 | + attr = {'raw.read.side_effect': urllib3.exceptions.ReadTimeoutError(None, None, "Read timed out.")} |
| 118 | + mock_response.configure_mock(**attr) |
| 119 | + mock_session_get.return_value = mock_response |
| 120 | + |
122 | 121 | with self.assertRaises(exceptions.SlowRetrievalError): |
123 | 122 | next(self.fetcher.fetch(self.url)) |
124 | | - |
125 | | - slow_server_process_handler.clean() |
| 123 | + mock_response.raw.read.assert_called_once() |
126 | 124 |
|
127 | 125 | # Read/connect session timeout error |
128 | 126 | @patch.object(requests.Session, 'get', side_effect=urllib3.exceptions.TimeoutError) |
|
0 commit comments