(self, urls, handlers, retry=True)
| 260 | self.assertEqual(request.get_header('User-agent'),'Test-Agent') |
| 261 | |
| 262 | def _test_urls(self, urls, handlers, retry=True): |
| 263 | import time |
| 264 | import logging |
| 265 | debug = logging.getLogger("test_urllib2").debug |
| 266 | |
| 267 | urlopen = urllib.request.build_opener(*handlers).open |
| 268 | if retry: |
| 269 | urlopen = _wrap_with_retry_thrice(urlopen, urllib.error.URLError) |
| 270 | |
| 271 | for url in urls: |
| 272 | with self.subTest(url=url): |
| 273 | if isinstance(url, tuple): |
| 274 | url, req, expected_err = url |
| 275 | else: |
| 276 | req = expected_err = None |
| 277 | |
| 278 | if expected_err: |
| 279 | context = self.assertRaises(expected_err) |
| 280 | else: |
| 281 | context = contextlib.nullcontext() |
| 282 | |
| 283 | with socket_helper.transient_internet(url): |
| 284 | f = None |
| 285 | with context: |
| 286 | f = urlopen(url, req, support.INTERNET_TIMEOUT) |
| 287 | if f is not None: |
| 288 | try: |
| 289 | with time_out, \ |
| 290 | socket_peer_reset, \ |
| 291 | ioerror_peer_reset: |
| 292 | buf = f.read() |
| 293 | debug("read %d bytes" % len(buf)) |
| 294 | except TimeoutError: |
| 295 | print("<timeout: %s>" % url, file=sys.stderr) |
| 296 | f.close() |
| 297 | time.sleep(0.1) |
| 298 | |
| 299 | def _extra_handlers(self): |
| 300 | handlers = [] |
no test coverage detected