validate_and_send(rh, Request(f'https://127.0.0.1:{self.https_port}/headers'))
@pytest.mark.parametrize('handler', ['Urllib'], indirect=True)
- def test_httplib_validation_errors(self, handler):
- with handler() as rh:
-
- # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1256
- with pytest.raises(RequestError, match='method can\'t contain control characters') as exc_info:
- validate_and_send(rh, Request('http://127.0.0.1', method='GET\n'))
- assert not isinstance(exc_info.value, TransportError)
-
- # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1265
- with pytest.raises(RequestError, match='URL can\'t contain control characters') as exc_info:
- validate_and_send(rh, Request('http://127.0.0. 1', method='GET\n'))
- assert not isinstance(exc_info.value, TransportError)
+ @pytest.mark.parametrize('req,match,version_check', [
+ # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1256
+ # bpo-39603: Check implemented in 3.7.9+, 3.8.5+
+ (
+ 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)
+ ),
+ # 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)
+ ),
+ # 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),
+ ])
+ def test_httplib_validation_errors(self, handler, req, match, version_check):
+ if version_check and version_check(sys.version_info):
+ pytest.skip(f'Python {sys.version} version does not have the required validation for this test.')
- # https://github.com/python/cpython/blob/987b712b4aeeece336eed24fcc87a950a756c3e2/Lib/http/client.py#L1288C31-L1288C50
- with pytest.raises(RequestError, match='Invalid header name') as exc_info:
- validate_and_send(rh, Request('http://127.0.0.1', headers={'foo\n': 'bar'}))
+ with handler() as rh:
+ with pytest.raises(RequestError, match=match) as exc_info:
+ validate_and_send(rh, req)
assert not isinstance(exc_info.value, TransportError)
-def run_validation(handler, fail, req, **handler_kwargs):
+def run_validation(handler, error, req, **handler_kwargs):
with handler(**handler_kwargs) as rh:
- if fail:
- with pytest.raises(UnsupportedRequest):
+ if error:
+ with pytest.raises(error):
rh.validate(req)
else:
rh.validate(req)
_SUPPORTED_PROXY_SCHEMES = None
_SUPPORTED_URL_SCHEMES = None
+ def _check_extensions(self, extensions):
+ extensions.clear()
+
class HTTPSupportedRH(ValidationRH):
_SUPPORTED_URL_SCHEMES = ('http',)
('https', False, {}),
('data', False, {}),
('ftp', False, {}),
- ('file', True, {}),
+ ('file', UnsupportedRequest, {}),
('file', False, {'enable_file_urls': True}),
]),
(NoCheckRH, [('http', False, {})]),
- (ValidationRH, [('http', True, {})])
+ (ValidationRH, [('http', UnsupportedRequest, {})])
]
PROXY_SCHEME_TESTS = [
# scheme, expected to fail
('Urllib', [
('http', False),
- ('https', True),
+ ('https', UnsupportedRequest),
('socks4', False),
('socks4a', False),
('socks5', False),
('socks5h', False),
- ('socks', True),
+ ('socks', UnsupportedRequest),
]),
(NoCheckRH, [('http', False)]),
- (HTTPSupportedRH, [('http', True)]),
+ (HTTPSupportedRH, [('http', UnsupportedRequest)]),
]
PROXY_KEY_TESTS = [
('unrelated', False),
]),
(NoCheckRH, [('all', False)]),
- (HTTPSupportedRH, [('all', True)]),
- (HTTPSupportedRH, [('no', True)]),
+ (HTTPSupportedRH, [('all', UnsupportedRequest)]),
+ (HTTPSupportedRH, [('no', UnsupportedRequest)]),
+ ]
+
+ EXTENSION_TESTS = [
+ ('Urllib', [
+ ({'cookiejar': 'notacookiejar'}, AssertionError),
+ ({'cookiejar': CookieJar()}, False),
+ ({'timeout': 1}, False),
+ ({'timeout': 'notatimeout'}, AssertionError),
+ ({'unsupported': 'value'}, UnsupportedRequest),
+ ]),
+ (NoCheckRH, [
+ ({'cookiejar': 'notacookiejar'}, False),
+ ({'somerandom': 'test'}, False), # but any extension is allowed through
+ ]),
]
@pytest.mark.parametrize('handler,scheme,fail,handler_kwargs', [
@pytest.mark.parametrize('proxy_url', ['//example.com', 'example.com', '127.0.0.1'])
@pytest.mark.parametrize('handler', ['Urllib'], indirect=True)
def test_missing_proxy_scheme(self, handler, proxy_url):
- run_validation(handler, True, Request('http://', proxies={'http': 'example.com'}))
+ run_validation(handler, UnsupportedRequest, Request('http://', proxies={'http': 'example.com'}))
- @pytest.mark.parametrize('handler', ['Urllib'], indirect=True)
- def test_cookiejar_extension(self, handler):
- run_validation(handler, True, Request('http://', extensions={'cookiejar': 'notacookiejar'}))
-
- @pytest.mark.parametrize('handler', ['Urllib'], indirect=True)
- def test_timeout_extension(self, handler):
- run_validation(handler, True, Request('http://', extensions={'timeout': 'notavalidtimeout'}))
+ @pytest.mark.parametrize('handler,extensions,fail', [
+ (handler_tests[0], extensions, fail)
+ for handler_tests in EXTENSION_TESTS
+ for extensions, fail in handler_tests[1]
+ ], indirect=['handler'])
+ def test_extension(self, handler, extensions, fail):
+ run_validation(
+ handler, fail, Request('http://', extensions=extensions))
def test_invalid_request_type(self):
rh = self.ValidationRH(logger=FakeLogger())
urllib_req = urllib.request.Request('http://foo.bar', data=b'test', method='PUT', headers={'X-Test': '1'})
urllib_req.add_unredirected_header('Cookie', 'bob=bob')
urllib_req.timeout = 2
-
- req = ydl.urlopen(urllib_req).request
- assert req.url == urllib_req.get_full_url()
- assert req.data == urllib_req.data
- assert req.method == urllib_req.get_method()
- assert 'X-Test' in req.headers
- assert 'Cookie' in req.headers
- assert req.extensions.get('timeout') == 2
+ with warnings.catch_warnings():
+ warnings.simplefilter('ignore', category=DeprecationWarning)
+ req = ydl.urlopen(urllib_req).request
+ assert req.url == urllib_req.get_full_url()
+ assert req.data == urllib_req.data
+ assert req.method == urllib_req.get_method()
+ assert 'X-Test' in req.headers
+ assert 'Cookie' in req.headers
+ assert req.extensions.get('timeout') == 2
with pytest.raises(AssertionError):
ydl.urlopen(None)
'debug_printtraffic': True,
'compat_opts': ['no-certifi'],
'nocheckcertificate': True,
- 'legacy_server_connect': True,
+ 'legacyserverconnect': True,
}) as ydl:
rh = self.build_handler(ydl)
assert rh.headers.get('test') == 'testtest'
req.data = b'test3'
assert req.headers.get('Content-Type') == 'application/x-www-form-urlencoded'
+ def test_update_req(self):
+ req = Request('http://example.com')
+ assert req.data is None
+ assert req.method == 'GET'
+ assert 'Content-Type' not in req.headers
+ # Test that zero-byte payloads will be sent
+ req.update(data=b'')
+ assert req.data == b''
+ assert req.method == 'POST'
+ assert req.headers.get('Content-Type') == 'application/x-www-form-urlencoded'
+
def test_proxies(self):
req = Request(url='http://example.com', proxies={'http': 'http://127.0.0.1:8080'})
assert req.proxies == {'http': 'http://127.0.0.1:8080'}
def test_compat(self):
res = Response(io.BytesIO(b''), url='test://', status=404, headers={'test': 'test'})
- assert res.code == res.getcode() == res.status
- assert res.geturl() == res.url
- assert res.info() is res.headers
- assert res.getheader('test') == res.get_header('test')
+ with warnings.catch_warnings():
+ warnings.simplefilter('ignore', category=DeprecationWarning)
+ assert res.code == res.getcode() == res.status
+ assert res.geturl() == res.url
+ assert res.info() is res.headers
+ assert res.getheader('test') == res.get_header('test')
projects / yt-dlp.git / blobdiff