import pytest
+from yt_dlp.networking.common import Features, DEFAULT_TIMEOUT
+
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import gzip
from email.message import Message
from http.cookiejar import CookieJar
-from test.conftest import validate_and_send
-from test.helper import FakeYDL, http_server_port, verify_address_availability
+from test.helper import (
+ FakeYDL,
+ http_server_port,
+ validate_and_send,
+ verify_address_availability,
+)
from yt_dlp.cookies import YoutubeDLCookieJar
from yt_dlp.dependencies import brotli, curl_cffi, requests, urllib3
from yt_dlp.networking import (
TEST_DIR = os.path.dirname(os.path.abspath(__file__))
-def _build_proxy_handler(name):
- class HTTPTestRequestHandler(http.server.BaseHTTPRequestHandler):
- proxy_name = name
-
- def log_message(self, format, *args):
- pass
-
- def do_GET(self):
- self.send_response(200)
- self.send_header('Content-Type', 'text/plain; charset=utf-8')
- self.end_headers()
- self.wfile.write(f'{self.proxy_name}: {self.path}'.encode())
- return HTTPTestRequestHandler
-
-
class HTTPTestRequestHandler(http.server.BaseHTTPRequestHandler):
protocol_version = 'HTTP/1.1'
default_request_version = 'HTTP/1.1'
cls.https_server_thread.start()
+@pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
class TestHTTPRequestHandler(TestRequestHandlerBase):
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
+
def test_verify_cert(self, handler):
with handler() as rh:
with pytest.raises(CertificateVerifyError):
assert r.status == 200
r.close()
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_ssl_error(self, handler):
# HTTPS server with too old TLS version
# XXX: is there a better way to test this than to create a new server?
validate_and_send(rh, Request(f'https://127.0.0.1:{https_port}/headers'))
assert not issubclass(exc_info.type, CertificateVerifyError)
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_percent_encode(self, handler):
with handler() as rh:
# Unicode characters should be encoded with uppercase percent-encoding
assert res.status == 200
res.close()
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
@pytest.mark.parametrize('path', [
'/a/b/./../../headers',
'/redirect_dotsegments',
assert res.url == f'http://127.0.0.1:{self.http_port}/headers'
res.close()
- # Not supported by CurlCFFI (non-standard)
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
+ @pytest.mark.skip_handler('CurlCFFI', 'not supported by curl-cffi (non-standard)')
def test_unicode_path_redirection(self, handler):
with handler() as rh:
r = validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/302-non-ascii-redirect'))
assert r.url == f'http://127.0.0.1:{self.http_port}/%E4%B8%AD%E6%96%87.html'
r.close()
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
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()
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_response_url(self, handler):
with handler() as rh:
# Response url should be that of the last url in redirect chain
res2.close()
# Covers some basic cases we expect some level of consistency between request handlers for
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
@pytest.mark.parametrize('redirect_status,method,expected', [
# A 303 must either use GET or HEAD for subsequent request
(303, 'POST', ('', 'GET', False)),
assert expected[1] == res.headers.get('method')
assert expected[2] == ('content-length' in headers.decode().lower())
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_request_cookie_header(self, handler):
# We should accept a Cookie header being passed as in normal headers and handle it appropriately.
with handler() as rh:
assert b'cookie: test=ytdlp' not in data.lower()
assert b'cookie: test=test3' in data.lower()
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_redirect_loop(self, handler):
with handler() as rh:
with pytest.raises(HTTPError, match='redirect loop'):
validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/redirect_loop'))
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
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()
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_cookies(self, handler):
cookiejar = YoutubeDLCookieJar()
cookiejar.set_cookie(http.cookiejar.Cookie(
rh, Request(f'http://127.0.0.1:{self.http_port}/headers', extensions={'cookiejar': cookiejar})).read()
assert b'cookie: test=ytdlp' in data.lower()
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_headers(self, handler):
with handler(headers=HTTPHeaderDict({'test1': 'test', 'test2': 'test2'})) as rh:
assert b'test2: test2' not in data
assert b'test3: test3' in data
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_read_timeout(self, handler):
with handler() as rh:
# Default timeout is 20 seconds, so this should go through
validate_and_send(
rh, Request(f'http://127.0.0.1:{self.http_port}/timeout_1', extensions={'timeout': 4}))
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_connect_timeout(self, handler):
# nothing should be listening on this port
connect_timeout_url = 'http://10.255.255.255'
- with handler(timeout=0.01) as rh:
+ with handler(timeout=0.01) as rh, pytest.raises(TransportError):
now = time.time()
- with pytest.raises(TransportError):
- validate_and_send(
- rh, Request(connect_timeout_url))
- assert 0.01 <= time.time() - now < 20
+ validate_and_send(rh, Request(connect_timeout_url))
+ assert time.time() - now < DEFAULT_TIMEOUT
- with handler() as rh:
- with pytest.raises(TransportError):
- # Per request timeout, should override handler timeout
- now = time.time()
- validate_and_send(
- rh, Request(connect_timeout_url, extensions={'timeout': 0.01}))
- assert 0.01 <= time.time() - now < 20
+ # Per request timeout, should override handler timeout
+ request = Request(connect_timeout_url, extensions={'timeout': 0.01})
+ with handler() as rh, pytest.raises(TransportError):
+ now = time.time()
+ validate_and_send(rh, request)
+ assert time.time() - now < DEFAULT_TIMEOUT
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_source_address(self, handler):
source_address = f'127.0.0.{random.randint(5, 255)}'
# on some systems these loopback addresses we need for testing may not be available
assert source_address == data
# Not supported by CurlCFFI
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
+ @pytest.mark.skip_handler('CurlCFFI', 'not supported by curl-cffi')
def test_gzip_trailing_garbage(self, handler):
with handler() as rh:
data = validate_and_send(rh, Request(f'http://localhost:{self.http_port}/trailing_garbage')).read().decode()
assert data == '<html><video src="/vid.mp4" /></html>'
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
+ @pytest.mark.skip_handler('CurlCFFI', 'not applicable to curl-cffi')
@pytest.mark.skipif(not brotli, reason='brotli support is not installed')
def test_brotli(self, handler):
with handler() as rh:
assert res.headers.get('Content-Encoding') == 'br'
assert res.read() == b'<html><video src="/vid.mp4" /></html>'
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_deflate(self, handler):
with handler() as rh:
res = validate_and_send(
assert res.headers.get('Content-Encoding') == 'deflate'
assert res.read() == b'<html><video src="/vid.mp4" /></html>'
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_gzip(self, handler):
with handler() as rh:
res = validate_and_send(
assert res.headers.get('Content-Encoding') == 'gzip'
assert res.read() == b'<html><video src="/vid.mp4" /></html>'
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_multiple_encodings(self, handler):
with handler() as rh:
for pair in ('gzip,deflate', 'deflate, gzip', 'gzip, gzip', 'deflate, deflate'):
assert res.headers.get('Content-Encoding') == pair
assert res.read() == b'<html><video src="/vid.mp4" /></html>'
- # Not supported by curl_cffi
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
+ @pytest.mark.skip_handler('CurlCFFI', 'not supported by curl-cffi')
def test_unsupported_encoding(self, handler):
with handler() as rh:
res = validate_and_send(
assert res.headers.get('Content-Encoding') == 'unsupported'
assert res.read() == b'raw'
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_read(self, handler):
with handler() as rh:
res = validate_and_send(
assert res.read().decode().endswith('\n\n')
assert res.read() == b''
+ def test_request_disable_proxy(self, handler):
+ for proxy_proto in handler._SUPPORTED_PROXY_SCHEMES or ['http']:
+ # Given the handler is configured with a proxy
+ with handler(proxies={'http': f'{proxy_proto}://10.255.255.255'}, timeout=5) as rh:
+ # When a proxy is explicitly set to None for the request
+ res = validate_and_send(
+ rh, Request(f'http://127.0.0.1:{self.http_port}/headers', proxies={'http': None}))
+ # Then no proxy should be used
+ res.close()
+ assert res.status == 200
-class TestHTTPProxy(TestRequestHandlerBase):
- # Note: this only tests http urls over non-CONNECT proxy
- @classmethod
- def setup_class(cls):
- super().setup_class()
- # HTTP Proxy server
- cls.proxy = http.server.ThreadingHTTPServer(
- ('127.0.0.1', 0), _build_proxy_handler('normal'))
- cls.proxy_port = http_server_port(cls.proxy)
- cls.proxy_thread = threading.Thread(target=cls.proxy.serve_forever)
- cls.proxy_thread.daemon = True
- cls.proxy_thread.start()
-
- # Geo proxy server
- cls.geo_proxy = http.server.ThreadingHTTPServer(
- ('127.0.0.1', 0), _build_proxy_handler('geo'))
- cls.geo_port = http_server_port(cls.geo_proxy)
- cls.geo_proxy_thread = threading.Thread(target=cls.geo_proxy.serve_forever)
- cls.geo_proxy_thread.daemon = True
- cls.geo_proxy_thread.start()
-
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
- def test_http_proxy(self, handler):
- http_proxy = f'http://127.0.0.1:{self.proxy_port}'
- geo_proxy = f'http://127.0.0.1:{self.geo_port}'
-
- # Test global http proxy
- # Test per request http proxy
- # Test per request http proxy disables proxy
- url = 'http://foo.com/bar'
-
- # Global HTTP proxy
- with handler(proxies={'http': http_proxy}) as rh:
- res = validate_and_send(rh, Request(url)).read().decode()
- assert res == f'normal: {url}'
-
- # Per request proxy overrides global
- res = validate_and_send(rh, Request(url, proxies={'http': geo_proxy})).read().decode()
- assert res == f'geo: {url}'
-
- # and setting to None disables all proxies for that request
- real_url = f'http://127.0.0.1:{self.http_port}/headers'
- res = validate_and_send(
- rh, Request(real_url, proxies={'http': None})).read().decode()
- assert res != f'normal: {real_url}'
- assert 'Accept' in res
-
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
+ @pytest.mark.skip_handlers_if(
+ lambda _, handler: Features.NO_PROXY not in handler._SUPPORTED_FEATURES, 'handler does not support NO_PROXY')
def test_noproxy(self, handler):
- with handler(proxies={'proxy': f'http://127.0.0.1:{self.proxy_port}'}) as rh:
- # NO_PROXY
- for no_proxy in (f'127.0.0.1:{self.http_port}', '127.0.0.1', 'localhost'):
- nop_response = validate_and_send(
- rh, Request(f'http://127.0.0.1:{self.http_port}/headers', proxies={'no': no_proxy})).read().decode(
- 'utf-8')
- assert 'Accept' in nop_response
-
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
+ for proxy_proto in handler._SUPPORTED_PROXY_SCHEMES or ['http']:
+ # Given the handler is configured with a proxy
+ with handler(proxies={'http': f'{proxy_proto}://10.255.255.255'}, timeout=5) as rh:
+ for no_proxy in (f'127.0.0.1:{self.http_port}', '127.0.0.1', 'localhost'):
+ # When request no proxy includes the request url host
+ nop_response = validate_and_send(
+ rh, Request(f'http://127.0.0.1:{self.http_port}/headers', proxies={'no': no_proxy}))
+ # Then the proxy should not be used
+ assert nop_response.status == 200
+ nop_response.close()
+
+ @pytest.mark.skip_handlers_if(
+ lambda _, handler: Features.ALL_PROXY not in handler._SUPPORTED_FEATURES, 'handler does not support ALL_PROXY')
def test_allproxy(self, handler):
- url = 'http://foo.com/bar'
- with handler() as rh:
- response = validate_and_send(rh, Request(url, proxies={'all': f'http://127.0.0.1:{self.proxy_port}'})).read().decode(
- 'utf-8')
- assert response == f'normal: {url}'
+ # This is a bit of a hacky test, but it should be enough to check whether the handler is using the proxy.
+ # 0.1s might not be enough of a timeout if proxy is not used in all cases, but should still get failures.
+ with handler(proxies={'all': 'http://10.255.255.255'}, timeout=0.1) as rh:
+ with pytest.raises(TransportError):
+ validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/headers')).close()
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
- def test_http_proxy_with_idn(self, handler):
- with handler(proxies={
- 'http': f'http://127.0.0.1:{self.proxy_port}',
- }) as rh:
- url = 'http://中文.tw/'
- response = rh.send(Request(url)).read().decode()
- # b'xn--fiq228c' is '中文'.encode('idna')
- assert response == 'normal: http://xn--fiq228c.tw/'
+ with handler(timeout=0.1) as rh:
+ with pytest.raises(TransportError):
+ validate_and_send(
+ rh, Request(
+ f'http://127.0.0.1:{self.http_port}/headers', proxies={'all': 'http://10.255.255.255'})).close()
+@pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
class TestClientCertificate:
@classmethod
def setup_class(cls):
) as rh:
validate_and_send(rh, Request(f'https://127.0.0.1:{self.port}/video.html')).read().decode()
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_certificate_combined_nopass(self, handler):
self._run_test(handler, client_cert={
'client_certificate': os.path.join(self.certdir, 'clientwithkey.crt'),
})
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_certificate_nocombined_nopass(self, handler):
self._run_test(handler, client_cert={
'client_certificate': os.path.join(self.certdir, 'client.crt'),
'client_certificate_key': os.path.join(self.certdir, 'client.key'),
})
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_certificate_combined_pass(self, handler):
self._run_test(handler, client_cert={
'client_certificate': os.path.join(self.certdir, 'clientwithencryptedkey.crt'),
'client_certificate_password': 'foobar',
})
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
def test_certificate_nocombined_pass(self, handler):
self._run_test(handler, client_cert={
'client_certificate': os.path.join(self.certdir, 'client.crt'),
assert res.status == 200
assert std_headers['user-agent'].lower() not in res.read().decode().lower()
+ def test_response_extensions(self, handler):
+ with handler() as rh:
+ for target in rh.supported_targets:
+ request = Request(
+ f'http://127.0.0.1:{self.http_port}/gen_200', extensions={'impersonate': target})
+ res = validate_and_send(rh, request)
+ assert res.extensions['impersonate'] == rh._get_request_target(request)
+
+ def test_http_error_response_extensions(self, handler):
+ with handler() as rh:
+ for target in rh.supported_targets:
+ request = Request(
+ f'http://127.0.0.1:{self.http_port}/gen_404', extensions={'impersonate': target})
+ try:
+ validate_and_send(rh, request)
+ except HTTPError as e:
+ res = e.response
+ assert res.extensions['impersonate'] == rh._get_request_target(request)
+
class TestRequestHandlerMisc:
"""Misc generic tests for request handlers, not related to request or validation testing"""
@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,
assert len(logging_handlers) == before_count
+@pytest.mark.parametrize('handler', ['Urllib'], indirect=True)
class TestUrllibRequestHandler(TestRequestHandlerBase):
- @pytest.mark.parametrize('handler', ['Urllib'], indirect=True)
def test_file_urls(self, handler):
# See https://github.com/ytdl-org/youtube-dl/issues/8227
tf = tempfile.NamedTemporaryFile(delete=False)
os.unlink(tf.name)
- @pytest.mark.parametrize('handler', ['Urllib'], indirect=True)
def test_http_error_returns_content(self, handler):
# urllib HTTPError will try close the underlying response if reference to the HTTPError object is lost
def get_response():
assert get_response().read() == b'<html></html>'
- @pytest.mark.parametrize('handler', ['Urllib'], indirect=True)
def test_verify_cert_error_text(self, handler):
# Check the output of the error message
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'))
- @pytest.mark.parametrize('handler', ['Urllib'], indirect=True)
@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)
+ 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),
(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):
(
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):
('https', False, {}),
]),
(NoCheckRH, [('http', False, {})]),
- (ValidationRH, [('http', UnsupportedRequest, {})])
+ (ValidationRH, [('http', UnsupportedRequest, {})]),
]
PROXY_SCHEME_TESTS = [
- # scheme, expected to fail
+ # proxy scheme, expected to fail
('Urllib', 'http', [
('http', False),
('https', UnsupportedRequest),
('socks5', False),
('socks5h', False),
]),
+ ('Websockets', 'ws', [
+ ('http', UnsupportedRequest),
+ ('https', UnsupportedRequest),
+ ('socks4', False),
+ ('socks4a', False),
+ ('socks5', False),
+ ('socks5h', False),
+ ]),
(NoCheckRH, 'http', [('http', False)]),
(HTTPSupportedRH, 'http', [('http', UnsupportedRequest)]),
- ('Websockets', 'ws', [('http', UnsupportedRequest)]),
(NoCheckRH, 'http', [('http', False)]),
(HTTPSupportedRH, 'http', [('http', UnsupportedRequest)]),
]
PROXY_KEY_TESTS = [
- # key, expected to fail
- ('Urllib', [
- ('all', False),
- ('unrelated', False),
+ # proxy key, proxy scheme, expected to fail
+ ('Urllib', 'http', [
+ ('all', 'http', False),
+ ('unrelated', 'http', False),
]),
- ('Requests', [
- ('all', False),
- ('unrelated', False),
+ ('Requests', 'http', [
+ ('all', 'http', False),
+ ('unrelated', 'http', False),
]),
- ('CurlCFFI', [
- ('all', False),
- ('unrelated', False),
+ ('CurlCFFI', 'http', [
+ ('all', 'http', False),
+ ('unrelated', 'http', False),
+ ]),
+ ('Websockets', 'ws', [
+ ('all', 'socks5', False),
+ ('unrelated', 'socks5', False),
]),
- (NoCheckRH, [('all', False)]),
- (HTTPSupportedRH, [('all', UnsupportedRequest)]),
- (HTTPSupportedRH, [('no', UnsupportedRequest)]),
+ (NoCheckRH, 'http', [('all', 'http', False)]),
+ (HTTPSupportedRH, 'http', [('all', 'http', UnsupportedRequest)]),
+ (HTTPSupportedRH, 'http', [('no', 'http', UnsupportedRequest)]),
]
EXTENSION_TESTS = [
({'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),
]),
]
+ @pytest.mark.parametrize('handler,fail,scheme', [
+ ('Urllib', False, 'http'),
+ ('Requests', False, 'http'),
+ ('CurlCFFI', False, 'http'),
+ ('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'}))
+ run_validation(handler, fail, Request(f'{scheme}://'), proxies={'no': '127.0.0.1,github.com'})
+
+ @pytest.mark.parametrize('handler,scheme', [
+ ('Urllib', 'http'),
+ (HTTPSupportedRH, 'http'),
+ ('Requests', 'http'),
+ ('CurlCFFI', 'http'),
+ ('Websockets', 'ws'),
+ ], indirect=['handler'])
+ def test_empty_proxy(self, handler, scheme):
+ run_validation(handler, False, Request(f'{scheme}://', proxies={scheme: None}))
+ run_validation(handler, False, Request(f'{scheme}://'), proxies={scheme: None})
+
+ @pytest.mark.parametrize('proxy_url', ['//example.com', 'example.com', '127.0.0.1', '/a/b/c'])
+ @pytest.mark.parametrize('handler,scheme', [
+ ('Urllib', 'http'),
+ (HTTPSupportedRH, 'http'),
+ ('Requests', 'http'),
+ ('CurlCFFI', 'http'),
+ ('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}))
+
@pytest.mark.parametrize('handler,scheme,fail,handler_kwargs', [
(handler_tests[0], scheme, fail, handler_kwargs)
for handler_tests in URL_SCHEME_TESTS
for scheme, fail, handler_kwargs in handler_tests[1]
-
], indirect=['handler'])
def test_url_scheme(self, handler, scheme, fail, handler_kwargs):
run_validation(handler, fail, Request(f'{scheme}://'), **(handler_kwargs or {}))
- @pytest.mark.parametrize('handler,fail', [('Urllib', False), ('Requests', False), ('CurlCFFI', False)], indirect=['handler'])
- def test_no_proxy(self, handler, fail):
- run_validation(handler, fail, Request('http://', proxies={'no': '127.0.0.1,github.com'}))
- run_validation(handler, fail, Request('http://'), proxies={'no': '127.0.0.1,github.com'})
-
- @pytest.mark.parametrize('handler,proxy_key,fail', [
- (handler_tests[0], proxy_key, fail)
+ @pytest.mark.parametrize('handler,scheme,proxy_key,proxy_scheme,fail', [
+ (handler_tests[0], handler_tests[1], proxy_key, proxy_scheme, fail)
for handler_tests in PROXY_KEY_TESTS
- for proxy_key, fail in handler_tests[1]
+ for proxy_key, proxy_scheme, fail in handler_tests[2]
], indirect=['handler'])
- def test_proxy_key(self, handler, proxy_key, fail):
- run_validation(handler, fail, Request('http://', proxies={proxy_key: 'http://example.com'}))
- run_validation(handler, fail, Request('http://'), proxies={proxy_key: 'http://example.com'})
+ def test_proxy_key(self, handler, scheme, proxy_key, proxy_scheme, fail):
+ run_validation(handler, fail, Request(f'{scheme}://', proxies={proxy_key: f'{proxy_scheme}://example.com'}))
+ run_validation(handler, fail, Request(f'{scheme}://'), proxies={proxy_key: f'{proxy_scheme}://example.com'})
@pytest.mark.parametrize('handler,req_scheme,scheme,fail', [
(handler_tests[0], handler_tests[1], scheme, fail)
run_validation(handler, fail, Request(f'{req_scheme}://', proxies={req_scheme: f'{scheme}://example.com'}))
run_validation(handler, fail, Request(f'{req_scheme}://'), proxies={req_scheme: f'{scheme}://example.com'})
- @pytest.mark.parametrize('handler', ['Urllib', HTTPSupportedRH, 'Requests', 'CurlCFFI'], indirect=True)
- def test_empty_proxy(self, handler):
- run_validation(handler, False, Request('http://', proxies={'http': None}))
- run_validation(handler, False, Request('http://'), proxies={'http': None})
-
- @pytest.mark.parametrize('proxy_url', ['//example.com', 'example.com', '127.0.0.1', '/a/b/c'])
- @pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
- def test_invalid_proxy_url(self, handler, proxy_url):
- run_validation(handler, UnsupportedRequest, Request('http://', proxies={'http': proxy_url}))
-
@pytest.mark.parametrize('handler,scheme,extensions,fail', [
(handler_tests[0], handler_tests[1], extensions, fail)
for handler_tests in EXTENSION_TESTS
@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')
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)}))
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)
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)})
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)
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)
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())
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
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'
('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)
@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):
(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):
projects / yt-dlp.git / blobdiff