pull/14067/merge
Simon Sawicki 2 days ago committed by GitHub
commit 0b633558c1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194

@ -19,12 +19,14 @@ def parse_args():
'test', help='an extractor test, test path, or one of "core" or "download"', nargs='*')
parser.add_argument(
'-k', help='run a test matching EXPRESSION. Same as "pytest -k"', metavar='EXPRESSION')
parser.add_argument(
'--no-segfaults', action='store_true', help='skip tests known to cause segfaults in CI')
parser.add_argument(
'--pytest-args', help='arguments to passthrough to pytest')
return parser.parse_args()
def run_tests(*tests, pattern=None, ci=False):
def run_tests(*tests, pattern=None, ci=False, allow_segfaults=True):
# XXX: hatch uses `tests` if no arguments are passed
run_core = 'core' in tests or 'tests' in tests or (not pattern and not tests)
run_download = 'download' in tests
@ -35,15 +37,25 @@ def run_tests(*tests, pattern=None, ci=False):
arguments.append('--color=yes')
if pattern:
arguments.extend(['-k', pattern])
positional = []
markers = []
if not allow_segfaults:
markers.append('not segfaults')
if run_core:
arguments.extend(['-m', 'not download'])
markers.append('not download')
elif run_download:
arguments.extend(['-m', 'download'])
markers.append('download')
else:
arguments.extend(
positional = [
test if '/' in test
else f'test/test_download.py::TestDownload::test_{fix_test_name(test)}'
for test in tests)
for test in tests
]
if markers:
arguments.extend(['-m', ' and '.join(markers)])
arguments.extend(positional)
print(f'Running {arguments}', flush=True)
try:
@ -72,6 +84,8 @@ if __name__ == '__main__':
args = parse_args()
os.chdir(Path(__file__).parent.parent)
sys.exit(run_tests(*args.test, pattern=args.k, ci=bool(os.getenv('CI'))))
sys.exit(run_tests(
*args.test, pattern=args.k, ci=bool(os.getenv('CI')),
allow_segfaults=not args.no_segfaults))
except KeyboardInterrupt:
pass

@ -53,6 +53,9 @@ def skip_handlers_if(request, handler):
def pytest_configure(config):
config.addinivalue_line(
'markers', 'segfaults: mark a test as potentially segfaulting in CI',
)
config.addinivalue_line(
'markers', 'skip_handler(handler): skip test for the given handler',
)

@ -245,8 +245,11 @@ def ctx(request):
return CTX_MAP[request.param]()
@pytest.mark.parametrize(
'handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
@pytest.mark.parametrize('handler', [
'Urllib',
'Requests',
pytest.param('CurlCFFI', marks=pytest.mark.segfaults),
], indirect=True)
@pytest.mark.parametrize('ctx', ['http'], indirect=True) # pure http proxy can only support http
class TestHTTPProxy:
def test_http_no_auth(self, handler, ctx):
@ -313,7 +316,7 @@ class TestHTTPProxy:
@pytest.mark.parametrize(
'handler,ctx', [
('Requests', 'https'),
('CurlCFFI', 'https'),
pytest.param('CurlCFFI', 'https', marks=pytest.mark.segfaults),
], indirect=True)
class TestHTTPConnectProxy:
def test_http_connect_no_auth(self, handler, ctx):

@ -310,7 +310,11 @@ class TestRequestHandlerBase:
cls.https_server_thread.start()
@pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
@pytest.mark.parametrize('handler', [
'Urllib',
'Requests',
pytest.param('CurlCFFI', marks=pytest.mark.segfaults),
], indirect=True)
class TestHTTPRequestHandler(TestRequestHandlerBase):
def test_verify_cert(self, handler):
@ -737,7 +741,11 @@ class TestHTTPRequestHandler(TestRequestHandlerBase):
assert res.read() == b'<video src="/vid.mp4" /></html>'
@pytest.mark.parametrize('handler', ['Urllib', 'Requests', 'CurlCFFI'], indirect=True)
@pytest.mark.parametrize('handler', [
'Urllib',
'Requests',
pytest.param('CurlCFFI', marks=pytest.mark.segfaults),
], indirect=True)
class TestClientCertificate:
@classmethod
def setup_class(cls):
@ -789,7 +797,7 @@ class TestClientCertificate:
})
@pytest.mark.parametrize('handler', ['CurlCFFI'], indirect=True)
@pytest.mark.parametrize('handler', [pytest.param('CurlCFFI', marks=pytest.mark.segfaults)], indirect=True)
class TestHTTPImpersonateRequestHandler(TestRequestHandlerBase):
def test_supported_impersonate_targets(self, handler):
with handler(headers=std_headers) as rh:
@ -824,8 +832,8 @@ 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'),
pytest.param('Websockets', 'websockets.client', marks=pytest.mark.segfaults),
pytest.param('Websockets', 'websockets.server', marks=pytest.mark.segfaults),
], indirect=['handler'])
def test_remove_logging_handler(self, handler, logger_name):
# Ensure any logging handlers, which may contain a YoutubeDL instance,
@ -1013,7 +1021,7 @@ class TestRequestsRequestHandler(TestRequestHandlerBase):
assert called
@pytest.mark.parametrize('handler', ['CurlCFFI'], indirect=True)
@pytest.mark.parametrize('handler', [pytest.param('CurlCFFI', marks=pytest.mark.segfaults)], indirect=True)
class TestCurlCFFIRequestHandler(TestRequestHandlerBase):
@pytest.mark.parametrize('params,extensions', [
@ -1352,8 +1360,8 @@ class TestRequestHandlerValidation:
@pytest.mark.parametrize('handler,fail,scheme', [
('Urllib', False, 'http'),
('Requests', False, 'http'),
('CurlCFFI', False, 'http'),
('Websockets', False, 'ws'),
pytest.param('CurlCFFI', False, 'http', marks=pytest.mark.segfaults),
pytest.param('Websockets', False, 'ws', marks=pytest.mark.segfaults),
], 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'}))
@ -1363,8 +1371,8 @@ class TestRequestHandlerValidation:
('Urllib', 'http'),
(HTTPSupportedRH, 'http'),
('Requests', 'http'),
('CurlCFFI', 'http'),
('Websockets', 'ws'),
pytest.param('CurlCFFI', 'http', marks=pytest.mark.segfaults),
pytest.param('Websockets', 'ws', marks=pytest.mark.segfaults),
], indirect=['handler'])
def test_empty_proxy(self, handler, scheme):
run_validation(handler, False, Request(f'{scheme}://', proxies={scheme: None}))
@ -1375,42 +1383,54 @@ class TestRequestHandlerValidation:
('Urllib', 'http'),
(HTTPSupportedRH, 'http'),
('Requests', 'http'),
('CurlCFFI', 'http'),
('Websockets', 'ws'),
pytest.param('CurlCFFI', 'http', marks=pytest.mark.segfaults),
pytest.param('Websockets', 'ws', marks=pytest.mark.segfaults),
], 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]
pytest.param(
handler, scheme, fail, handler_kwargs,
marks=[pytest.mark.segfaults] if handler in ['CurlCFFI', 'Websockets'] else [],
)
for handler, handler_tests in URL_SCHEME_TESTS
for scheme, fail, handler_kwargs in handler_tests
], 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,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, proxy_scheme, fail in handler_tests[2]
pytest.param(
handler, scheme, proxy_key, proxy_scheme, fail,
marks=[pytest.mark.segfaults] if handler in ['CurlCFFI', 'Websockets'] else [],
)
for handler, scheme, handler_tests in PROXY_KEY_TESTS
for proxy_key, proxy_scheme, fail in handler_tests
], indirect=['handler'])
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)
for handler_tests in PROXY_SCHEME_TESTS
for scheme, fail in handler_tests[2]
pytest.param(
handler, scheme, proxy_scheme, fail,
marks=[pytest.mark.segfaults] if handler in ['CurlCFFI', 'Websockets'] else [],
)
for handler, scheme, handler_tests in PROXY_SCHEME_TESTS
for proxy_scheme, fail in handler_tests
], indirect=['handler'])
def test_proxy_scheme(self, handler, req_scheme, 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,scheme,extensions,fail', [
(handler_tests[0], handler_tests[1], extensions, fail)
for handler_tests in EXTENSION_TESTS
for extensions, fail in handler_tests[2]
pytest.param(
handler, scheme, extensions, fail,
marks=[pytest.mark.segfaults] if handler in ['CurlCFFI', 'Websockets'] else [],
)
for handler, scheme, handler_tests in EXTENSION_TESTS
for extensions, fail in handler_tests
], indirect=['handler'])
def test_extension(self, handler, scheme, extensions, fail):
run_validation(

@ -292,8 +292,8 @@ def ctx(request):
'handler,ctx', [
('Urllib', 'http'),
('Requests', 'http'),
('Websockets', 'ws'),
('CurlCFFI', 'http'),
pytest.param('Websockets', 'ws', marks=pytest.mark.segfaults),
pytest.param('CurlCFFI', 'http', marks=pytest.mark.segfaults),
], indirect=True)
class TestSocks4Proxy:
def test_socks4_no_auth(self, handler, ctx):
@ -367,8 +367,8 @@ class TestSocks4Proxy:
'handler,ctx', [
('Urllib', 'http'),
('Requests', 'http'),
('Websockets', 'ws'),
('CurlCFFI', 'http'),
pytest.param('Websockets', 'ws', marks=pytest.mark.segfaults),
pytest.param('CurlCFFI', 'http', marks=pytest.mark.segfaults),
], indirect=True)
class TestSocks5Proxy:

@ -130,7 +130,7 @@ def ws_validate_and_send(rh, req):
@pytest.mark.skipif(not websockets, reason='websockets must be installed to test websocket request handlers')
@pytest.mark.parametrize('handler', ['Websockets'], indirect=True)
@pytest.mark.parametrize('handler', [pytest.param('Websockets', marks=pytest.mark.segfaults)], indirect=True)
class TestWebsSocketRequestHandlerConformance:
@classmethod
def setup_class(cls):
@ -439,7 +439,7 @@ def create_fake_ws_connection(raised):
return FakeWsConnection()
@pytest.mark.parametrize('handler', ['Websockets'], indirect=True)
@pytest.mark.parametrize('handler', [pytest.param('Websockets', marks=pytest.mark.segfaults)], indirect=True)
class TestWebsocketsRequestHandler:
@pytest.mark.parametrize('raised,expected', [
# https://websockets.readthedocs.io/en/stable/reference/exceptions.html

Loading…
Cancel
Save