X-Git-Url: https://jfr.im/git/yt-dlp.git/blobdiff_plain/db50f19d76c6870a5a13d0cab9287d684fd7449a..add96eb9f84cfffe85682bf2fb85135746994ee8:/test/test_networking.py diff --git a/test/test_networking.py b/test/test_networking.py index d127cbb94..af3ece3b4 100644 --- a/test/test_networking.py +++ b/test/test_networking.py @@ -375,10 +375,10 @@ def test_raise_http_error(self, handler): with handler() as rh: for bad_status in (400, 500, 599, 302): with pytest.raises(HTTPError): - validate_and_send(rh, Request('http://127.0.0.1:%d/gen_%d' % (self.http_port, bad_status))) + validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/gen_{bad_status}')) # Should not raise an error - validate_and_send(rh, Request('http://127.0.0.1:%d/gen_200' % self.http_port)).close() + validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/gen_200')).close() def test_response_url(self, handler): with handler() as rh: @@ -472,7 +472,7 @@ def test_redirect_loop(self, handler): def test_incompleteread(self, handler): with handler(timeout=2) as rh: with pytest.raises(IncompleteRead, match='13 bytes read, 234221 more expected'): - validate_and_send(rh, Request('http://127.0.0.1:%d/incompleteread' % self.http_port)).read() + validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/incompleteread')).read() def test_cookies(self, handler): cookiejar = YoutubeDLCookieJar() @@ -740,7 +740,7 @@ class TestRequestHandlerMisc: @pytest.mark.parametrize('handler,logger_name', [ ('Requests', 'urllib3'), ('Websockets', 'websockets.client'), - ('Websockets', 'websockets.server') + ('Websockets', 'websockets.server'), ], indirect=['handler']) def test_remove_logging_handler(self, handler, logger_name): # Ensure any logging handlers, which may contain a YoutubeDL instance, @@ -794,7 +794,7 @@ def test_verify_cert_error_text(self, handler): with handler() as rh: with pytest.raises( CertificateVerifyError, - match=r'\[SSL: CERTIFICATE_VERIFY_FAILED\] certificate verify failed: self.signed certificate' + match=r'\[SSL: CERTIFICATE_VERIFY_FAILED\] certificate verify failed: self.signed certificate', ): validate_and_send(rh, Request(f'https://127.0.0.1:{self.https_port}/headers')) @@ -804,14 +804,14 @@ def test_verify_cert_error_text(self, handler): ( Request('http://127.0.0.1', method='GET\n'), 'method can\'t contain control characters', - lambda v: v < (3, 7, 9) or (3, 8, 0) <= v < (3, 8, 5) + lambda v: v < (3, 7, 9) or (3, 8, 0) <= v < (3, 8, 5), ), # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1265 # bpo-38576: Check implemented in 3.7.8+, 3.8.3+ ( Request('http://127.0.0. 1', method='GET'), 'URL can\'t contain control characters', - lambda v: v < (3, 7, 8) or (3, 8, 0) <= v < (3, 8, 3) + lambda v: v < (3, 7, 8) or (3, 8, 0) <= v < (3, 8, 3), ), # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1288C31-L1288C50 (Request('http://127.0.0.1', headers={'foo\n': 'bar'}), 'Invalid header name', None), @@ -840,7 +840,7 @@ class TestRequestsRequestHandler(TestRequestHandlerBase): (lambda: requests.exceptions.InvalidHeader(), RequestError), # catch-all: https://github.com/psf/requests/blob/main/src/requests/adapters.py#L535 (lambda: urllib3.exceptions.HTTPError(), TransportError), - (lambda: requests.exceptions.RequestException(), RequestError) + (lambda: requests.exceptions.RequestException(), RequestError), # (lambda: requests.exceptions.TooManyRedirects(), HTTPError) - Needs a response object ]) def test_request_error_mapping(self, handler, monkeypatch, raised, expected): @@ -868,12 +868,12 @@ def request(self, *args, **kwargs): ( lambda: urllib3.exceptions.ProtocolError('error', http.client.IncompleteRead(partial=b'abc', expected=4)), IncompleteRead, - '3 bytes read, 4 more expected' + '3 bytes read, 4 more expected', ), ( lambda: urllib3.exceptions.ProtocolError('error', urllib3.exceptions.IncompleteRead(partial=3, expected=5)), IncompleteRead, - '3 bytes read, 5 more expected' + '3 bytes read, 5 more expected', ), ]) def test_response_error_mapping(self, handler, monkeypatch, raised, expected, match): @@ -1125,7 +1125,7 @@ class HTTPSupportedRH(ValidationRH): ('https', False, {}), ]), (NoCheckRH, [('http', False, {})]), - (ValidationRH, [('http', UnsupportedRequest, {})]) + (ValidationRH, [('http', UnsupportedRequest, {})]), ] PROXY_SCHEME_TESTS = [ @@ -1219,7 +1219,7 @@ class HTTPSupportedRH(ValidationRH): ({'impersonate': ImpersonateTarget('chrome', None, None, None)}, False), ({'impersonate': ImpersonateTarget(None, None, None, None)}, False), ({'impersonate': ImpersonateTarget()}, False), - ({'impersonate': 'chrome'}, AssertionError) + ({'impersonate': 'chrome'}, AssertionError), ]), (NoCheckRH, 'http', [ ({'cookiejar': 'notacookiejar'}, False), @@ -1235,7 +1235,7 @@ class HTTPSupportedRH(ValidationRH): ('Urllib', False, 'http'), ('Requests', False, 'http'), ('CurlCFFI', False, 'http'), - ('Websockets', False, 'ws') + ('Websockets', False, 'ws'), ], indirect=['handler']) def test_no_proxy(self, handler, fail, scheme): run_validation(handler, fail, Request(f'{scheme}://', proxies={'no': '127.0.0.1,github.com'})) @@ -1246,7 +1246,7 @@ def test_no_proxy(self, handler, fail, scheme): (HTTPSupportedRH, 'http'), ('Requests', 'http'), ('CurlCFFI', 'http'), - ('Websockets', 'ws') + ('Websockets', 'ws'), ], indirect=['handler']) def test_empty_proxy(self, handler, scheme): run_validation(handler, False, Request(f'{scheme}://', proxies={scheme: None})) @@ -1258,7 +1258,7 @@ def test_empty_proxy(self, handler, scheme): (HTTPSupportedRH, 'http'), ('Requests', 'http'), ('CurlCFFI', 'http'), - ('Websockets', 'ws') + ('Websockets', 'ws'), ], indirect=['handler']) def test_invalid_proxy_url(self, handler, scheme, proxy_url): run_validation(handler, UnsupportedRequest, Request(f'{scheme}://', proxies={scheme: proxy_url})) @@ -1474,7 +1474,7 @@ def test_compat_opener(self): @pytest.mark.parametrize('proxy,expected', [ ('http://127.0.0.1:8080', {'all': 'http://127.0.0.1:8080'}), ('', {'all': '__noproxy__'}), - (None, {'http': 'http://127.0.0.1:8081', 'https': 'http://127.0.0.1:8081'}) # env, set https + (None, {'http': 'http://127.0.0.1:8081', 'https': 'http://127.0.0.1:8081'}), # env, set https ]) def test_proxy(self, proxy, expected, monkeypatch): monkeypatch.setenv('HTTP_PROXY', 'http://127.0.0.1:8081') @@ -1546,7 +1546,7 @@ def _send(self, request: Request): with FakeImpersonationRHYDL() as ydl: with pytest.raises( RequestError, - match=r'Impersonate target "test" is not available' + match=r'Impersonate target "test" is not available', ): ydl.urlopen(Request('http://', extensions={'impersonate': ImpersonateTarget('test', None, None, None)})) @@ -1558,7 +1558,7 @@ def _send(self, request: Request): pass _SUPPORTED_URL_SCHEMES = ('http',) - _SUPPORTED_IMPERSONATE_TARGET_MAP = {ImpersonateTarget('abc',): 'test'} + _SUPPORTED_IMPERSONATE_TARGET_MAP = {ImpersonateTarget('abc'): 'test'} _SUPPORTED_PROXY_SCHEMES = None super().__init__(*args, **kwargs) @@ -1567,14 +1567,14 @@ def _send(self, request: Request): with FakeHTTPRHYDL() as ydl: with pytest.raises( RequestError, - match=r'Impersonate target "test" is not available' + match=r'Impersonate target "test" is not available', ): ydl.urlopen(Request('http://', extensions={'impersonate': ImpersonateTarget('test', None, None, None)})) def test_raise_impersonate_error(self): with pytest.raises( YoutubeDLError, - match=r'Impersonate target "test" is not available' + match=r'Impersonate target "test" is not available', ): FakeYDL({'impersonate': ImpersonateTarget('test', None, None, None)}) @@ -1592,7 +1592,7 @@ def _send(self, request: Request): monkeypatch.setattr(FakeYDL, 'build_request_director', lambda cls, handlers, preferences=None: brh(cls, handlers=[IRH])) with FakeYDL({ - 'impersonate': ImpersonateTarget('abc', None, None, None) + 'impersonate': ImpersonateTarget('abc', None, None, None), }) as ydl: rh = self.build_handler(ydl, IRH) assert rh.impersonate == ImpersonateTarget('abc', None, None, None) @@ -1604,7 +1604,7 @@ class TestRH(ImpersonateRequestHandler): def _send(self, request: Request): pass _SUPPORTED_URL_SCHEMES = ('http',) - _SUPPORTED_IMPERSONATE_TARGET_MAP = {ImpersonateTarget(target_client,): 'test'} + _SUPPORTED_IMPERSONATE_TARGET_MAP = {ImpersonateTarget(target_client): 'test'} RH_KEY = target_client RH_NAME = target_client handlers.append(TestRH) @@ -1614,7 +1614,7 @@ def _send(self, request: Request): assert set(ydl._get_available_impersonate_targets()) == { (ImpersonateTarget('xyz'), 'xyz'), (ImpersonateTarget('abc'), 'abc'), - (ImpersonateTarget('asd'), 'asd') + (ImpersonateTarget('asd'), 'asd'), } assert ydl._impersonate_target_available(ImpersonateTarget('abc')) assert ydl._impersonate_target_available(ImpersonateTarget()) @@ -1837,7 +1837,7 @@ def test_copy(self): extensions={'cookiejar': CookieJar()}, headers={'Accept-Encoding': 'br'}, proxies={'http': 'http://127.0.0.1'}, - data=[b'123'] + data=[b'123'], ) req_copy = req.copy() assert req_copy is not req @@ -1863,7 +1863,7 @@ class AnotherRequest(Request): assert isinstance(req.copy(), AnotherRequest) def test_url(self): - req = Request(url='https://фtest.example.com/ some spaceв?ä=c',) + req = Request(url='https://фtest.example.com/ some spaceв?ä=c') assert req.url == 'https://xn--test-z6d.example.com/%20some%20space%D0%B2?%C3%A4=c' assert Request(url='//example.com').url == 'http://example.com' @@ -1878,7 +1878,7 @@ class TestResponse: ('custom', 200, 'custom'), (None, 404, 'Not Found'), # fallback status ('', 403, 'Forbidden'), - (None, 999, None) + (None, 999, None), ]) def test_reason(self, reason, status, expected): res = Response(io.BytesIO(b''), url='test://', headers={}, status=status, reason=reason) @@ -1933,7 +1933,7 @@ def test_target_from_str(self, target_str, expected): @pytest.mark.parametrize('target_str', [ '-120', ':-12.0', '-12:-12', '-:-', - '::', 'a-c-d:', 'a-c-d:e-f-g', 'a:b:' + '::', 'a-c-d:', 'a-c-d:e-f-g', 'a:b:', ]) def test_target_from_invalid_str(self, target_str): with pytest.raises(ValueError): @@ -1949,7 +1949,7 @@ def test_target_from_invalid_str(self, target_str): (ImpersonateTarget('abc', '120', 'xyz', None), 'abc-120:xyz'), (ImpersonateTarget('abc', None, 'xyz'), 'abc:xyz'), (ImpersonateTarget(None, None, 'xyz', '6.5'), ':xyz-6.5'), - (ImpersonateTarget('abc', ), 'abc'), + (ImpersonateTarget('abc'), 'abc'), (ImpersonateTarget(None, None, None, None), ''), ]) def test_str(self, target, expected):