# simple https: GET, h2 wanted and got
@pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_01_03_h2_get(self, env: Env, httpd):
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
curl = CurlClient(env=env)
url = f'https://{env.domain1}:{env.https_port}/data.json'
r = curl.http_get(url=url, extra_args=['--http2'])
# simple https: GET, h2 unsupported, fallback to h1
@pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_01_04_h2_unsupported(self, env: Env, httpd):
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
curl = CurlClient(env=env)
url = f'https://{env.domain2}:{env.https_port}/data.json'
r = curl.http_get(url=url, extra_args=['--http2'])
# simple download, check connect/handshake timings
@pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_01_06_timings(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/data.json'
r = curl.http_download(urls=[url], alpn_proto=proto, with_stats=True)
assert r.stats[0]['time_appconnect'] > 0, f'{r.stats[0]}'
# simple https: HEAD
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
@pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
def test_01_07_head(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/data.json'
r = curl.http_download(urls=[url], with_stats=True, with_headers=True,
assert r.stats[0]['size_download'] == 0, f'{r.stats[0]}'
# http: GET for HTTP/2, see Upgrade:, 101 switch
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_01_08_h2_upgrade(self, env: Env, httpd):
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
curl = CurlClient(env=env)
url = f'http://{env.domain1}:{env.http_port}/data.json'
r = curl.http_get(url=url, extra_args=['--http2'])
assert r.json['server'] == env.domain1
# http: GET for HTTP/2 with prior knowledge
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_01_09_h2_prior_knowledge(self, env: Env, httpd):
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
curl = CurlClient(env=env)
url = f'http://{env.domain1}:{env.http_port}/data.json'
r = curl.http_get(url=url, extra_args=['--http2-prior-knowledge'])
assert r.json['server'] == env.domain1
# http: strip TE header in HTTP/2 requests
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_01_10_te_strip(self, env: Env, httpd):
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, "h2")}/data.json'
r = curl.http_get(url=url, extra_args=['--http2', '-H', 'TE: gzip'])
# send 48KB+ sized response headers to check we handle that correctly
# larger than 64KB headers expose a bug in Apache HTTP/2 that is not
# RSTing the stream correctly when its internal limits are exceeded.
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_01_11_large_resp_headers(self, env: Env, httpd, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}' \
f'/curltest/tweak?x-hd={48 * 1024}'
# http: response headers larger than what curl buffers for
@pytest.mark.skipif(condition=not Env.httpd_is_at_least('2.4.64'),
reason='httpd must be at least 2.4.64')
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
def test_01_12_xlarge_resp_headers(self, env: Env, httpd, configures_httpd, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
httpd.set_extra_config('base', [
f'H2MaxHeaderBlockLen {130 * 1024}',
])
# http: 1 response header larger than what curl buffers for
@pytest.mark.skipif(condition=not Env.httpd_is_at_least('2.4.64'),
reason='httpd must be at least 2.4.64')
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
def test_01_13_megalarge_resp_headers(self, env: Env, httpd, configures_httpd, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
httpd.set_extra_config('base', [
'LogLevel http2:trace2',
f'H2MaxHeaderBlockLen {130 * 1024}',
# nghttp2 error -905: Too many CONTINUATION frames following a HEADER frame
@pytest.mark.skipif(condition=not Env.httpd_is_at_least('2.4.64'),
reason='httpd must be at least 2.4.64')
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
def test_01_14_gigalarge_resp_headers(self, env: Env, httpd, configures_httpd, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
httpd.set_extra_config('base', [
'LogLevel http2:trace2',
f'H2MaxHeaderBlockLen {1024 * 1024}',
# http: one response header > 256 KB
@pytest.mark.skipif(condition=not Env.httpd_is_at_least('2.4.64'),
reason='httpd must be at least 2.4.64')
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
def test_01_15_gigalarge_resp_headers(self, env: Env, httpd, configures_httpd, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
httpd.set_extra_config('base', [
'LogLevel http2:trace2',
f'H2MaxHeaderBlockLen {1024 * 1024}',
r.check_exit_code(100) # CURLE_TOO_LARGE
# http: invalid request headers, GET, issue #16998
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_01_16_inv_req_get(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/curltest/echo'
r = curl.http_get(url=url, alpn_proto=proto, extra_args=[
pytest.param('gzip ;q=0.2;x="y,x", trailers', 'trailers', id='gzip+q+x+trailers'),
pytest.param('gzip ;x="trailers", chunks', None, id='gzip+x+chunks'),
])
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_01_17_TE(self, env: Env, httpd, te_in, te_out):
proto = 'h2'
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/curltest/echo'
r = curl.http_download(urls=[url], alpn_proto=proto, with_stats=True,
assert 'request-te' not in r.responses[0]['header'], f'{r.responses[0]}'
# check that an existing https: connection is not reused for http:
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_01_18_tls_reuse(self, env: Env, httpd):
proto = 'h2'
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
curl = CurlClient(env=env)
url1 = f'https://{env.authority_for(env.domain1, proto)}/data.json'
url2 = f'http://{env.authority_for(env.domain1, proto)}/data.json'
assert r.total_connects == 2, f'{r.dump_logs()}'
# check that an existing http: connection is not reused for https:
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_01_19_plain_reuse(self, env: Env, httpd):
proto = 'h2'
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
curl = CurlClient(env=env)
url1 = f'http://{env.domain1}:{env.http_port}/data.json'
url2 = f'https://{env.domain1}:{env.http_port}/data.json'
env.make_data_gzipbomb(indir=indir, fname="bomb-100m.txt", fsize=100*1024*1024)
# download 1 file
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_01_download_1(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/data.json'
r = curl.http_download(urls=[url], alpn_proto=proto)
r.check_response(http_status=200)
# download 2 files
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_02_download_2(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/data.json?[0-1]'
r = curl.http_download(urls=[url], alpn_proto=proto)
r.check_response(http_status=200, count=2)
# download 100 files sequentially
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_03_download_sequential(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if (proto == 'http/1.1' or proto == 'h2') and env.curl_uses_lib('mbedtls') and \
sys.platform.startswith('darwin') and env.ci_run:
pytest.skip('mbedtls 3.6.3 fails this test on macOS CI runners')
r.check_response(http_status=200, count=count, connect_count=1)
# download 100 files parallel
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_04_download_parallel(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if proto == 'h2' and env.curl_uses_lib('mbedtls') and \
sys.platform.startswith('darwin') and env.ci_run:
pytest.skip('mbedtls 3.6.3 fails this test on macOS CI runners')
assert r.total_connects == 1, r.dump_logs()
# download 500 files sequential
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_05_download_many_sequential(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if proto == 'h2' and env.curl_uses_lib('mbedtls') and \
sys.platform.startswith('darwin') and env.ci_run:
pytest.skip('mbedtls 3.6.3 fails this test on macOS CI runners')
assert r.total_connects == 1, r.dump_logs()
# download 500 files parallel
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_02_06_download_many_parallel(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if proto == 'h2' and env.curl_uses_lib('mbedtls') and \
sys.platform.startswith('darwin') and env.ci_run:
pytest.skip('mbedtls 3.6.3 fails this test on macOS CI runners')
r.check_response(http_status=200, count=count, connect_count=1)
# download files parallel, check connection reuse/multiplex
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_02_07_download_reuse(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 200
curl = CurlClient(env=env)
urln = f'https://{env.authority_for(env.domain1, proto)}/data.json?[0-{count-1}]'
# http/1.1 should have used count connections
assert r.total_connects == count, "http/1.1 should use this many connections"
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_08_1MB_serial(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 5
urln = f'https://{env.authority_for(env.domain1, proto)}/data-1m?[0-{count-1}]'
curl = CurlClient(env=env)
r = curl.http_download(urls=[urln], alpn_proto=proto)
r.check_response(count=count, http_status=200)
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_02_09_1MB_parallel(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 5
urln = f'https://{env.authority_for(env.domain1, proto)}/data-1m?[0-{count-1}]'
curl = CurlClient(env=env)
@pytest.mark.skipif(condition=Env().slow_network, reason="not suitable for slow network tests")
@pytest.mark.skipif(condition=Env().ci_run, reason="not suitable for CI runs")
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_10_10MB_serial(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 3
urln = f'https://{env.authority_for(env.domain1, proto)}/data-10m?[0-{count-1}]'
curl = CurlClient(env=env)
@pytest.mark.skipif(condition=Env().slow_network, reason="not suitable for slow network tests")
@pytest.mark.skipif(condition=Env().ci_run, reason="not suitable for CI runs")
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_02_11_10MB_parallel(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 3
urln = f'https://{env.authority_for(env.domain1, proto)}/data-10m?[0-{count-1}]'
curl = CurlClient(env=env)
])
r.check_response(count=count, http_status=200)
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_02_12_head_serial_https(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 5
urln = f'https://{env.authority_for(env.domain1, proto)}/data-10m?[0-{count-1}]'
curl = CurlClient(env=env)
r.check_response(count=count, http_status=200)
@pytest.mark.parametrize("proto", ['h2'])
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_02_13_head_serial_h2c(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 5
urln = f'http://{env.domain1}:{env.http_port}/data-10m?[0-{count-1}]'
curl = CurlClient(env=env)
])
r.check_response(count=count, http_status=200)
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_02_14_not_found(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 5
urln = f'https://{env.authority_for(env.domain1, proto)}/not-found?[0-{count-1}]'
curl = CurlClient(env=env)
remote_port=env.port_for(alpn_proto=proto),
remote_ip='127.0.0.1')
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_02_15_fail_not_found(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 5
urln = f'https://{env.authority_for(env.domain1, proto)}/not-found?[0-{count-1}]'
curl = CurlClient(env=env)
remote_ip='127.0.0.1')
@pytest.mark.skipif(condition=Env().slow_network, reason="not suitable for slow network tests")
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_02_20_h2_small_frames(self, env: Env, httpd, configures_httpd):
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
# Test case to reproduce content corruption as observed in
# https://github.com/curl/curl/issues/10525
# To reliably reproduce, we need an Apache httpd that supports
# debug-override stream window size to reproduce #16955
@pytest.mark.parametrize("pause_offset", [0, 10*1024, 100*1023, 640000])
@pytest.mark.parametrize("swin_max", [0, 10*1024])
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_02_21_h2_lib_serial(self, env: Env, httpd, pause_offset, swin_max):
proto = 'h2'
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
count = 2
docname = 'data-10m'
url = f'https://localhost:{env.https_port}/{docname}'
# download via lib client, several at a time, pause/resume
@pytest.mark.parametrize("pause_offset", [100*1023])
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_22_lib_parallel_resume(self, env: Env, httpd, nghttpx, proto, pause_offset):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 2
max_parallel = 5
docname = 'data-10m'
self.check_downloads(client, srcfile, count)
# download, several at a time, pause and abort paused
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_23a_lib_abort_paused(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_ossl_quic():
pytest.skip('OpenSSL QUIC fails here')
if proto == 'h3' and env.ci_run and env.curl_uses_lib('quiche'):
pytest.skip("fails in CI, but works locally for unknown reasons")
count = 10
max_parallel = 5
- if proto in ['h2', 'h3']:
+ if proto in Env.http_mplx_protos():
pause_offset = 64 * 1024
else:
pause_offset = 12 * 1024
self.check_downloads(client, srcfile, count, complete=False)
# download, several at a time, abort after n bytes
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_23b_lib_abort_offset(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_ossl_quic():
pytest.skip('OpenSSL QUIC fails here')
if proto == 'h3' and env.ci_run and env.curl_uses_lib('quiche'):
pytest.skip("fails in CI, but works locally for unknown reasons")
count = 10
max_parallel = 5
- if proto in ['h2', 'h3']:
+ if proto in Env.http_mplx_protos():
abort_offset = 64 * 1024
else:
abort_offset = 12 * 1024
self.check_downloads(client, srcfile, count, complete=False)
# download, several at a time, abort after n bytes
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_23c_lib_fail_offset(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_ossl_quic():
pytest.skip('OpenSSL QUIC fails here')
if proto == 'h3' and env.ci_run and env.curl_uses_lib('quiche'):
pytest.skip("fails in CI, but works locally for unknown reasons")
count = 10
max_parallel = 5
- if proto in ['h2', 'h3']:
+ if proto in Env.http_mplx_protos():
fail_offset = 64 * 1024
else:
fail_offset = 12 * 1024
self.check_downloads(client, srcfile, count, complete=False)
# speed limited download
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_24_speed_limit(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 1
url = f'https://{env.authority_for(env.domain1, proto)}/data-1m'
curl = CurlClient(env=env)
# Special client that tests TLS session reuse in parallel transfers
# TODO: just uses a single connection for h2/h3. Not sure how to prevent that
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_26_session_shared_reuse(self, env: Env, proto, httpd, nghttpx):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
url = f'https://{env.authority_for(env.domain1, proto)}/data-100k'
client = LocalClient(name='cli_tls_session_reuse', env=env)
if not client.exists():
r.check_exit_code(0)
# test on paused transfers, based on issue #11982
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_27a_paused_no_cl(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
url = f'https://{env.authority_for(env.domain1, proto)}' \
'/curltest/tweak/?&chunks=6&chunk_size=8000'
client = LocalClient(env=env, name='cli_h2_pausing')
r.check_exit_code(0)
# test on paused transfers, based on issue #11982
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_27b_paused_no_cl(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
url = f'https://{env.authority_for(env.domain1, proto)}' \
'/curltest/tweak/?error=502'
client = LocalClient(env=env, name='cli_h2_pausing')
r.check_exit_code(0)
# test on paused transfers, based on issue #11982
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_27c_paused_no_cl(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
url = f'https://{env.authority_for(env.domain1, proto)}' \
'/curltest/tweak/?status=200&chunks=1&chunk_size=100'
client = LocalClient(env=env, name='cli_h2_pausing')
r = client.run(args=['-V', proto, url])
r.check_exit_code(0)
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_28_get_compressed(self, env: Env, httpd, nghttpx, proto):
if not env.have_compressed_curl():
pytest.skip("--compressed not supported")
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 1
urln = f'https://{env.authority_for(env.domain1brotli, proto)}/data-100k?[0-{count-1}]'
curl = CurlClient(env=env)
# download via lib client, 1 at a time, pause/resume at different offsets
@pytest.mark.parametrize("pause_offset", [0, 10*1024, 100*1023, 640000])
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_29_h2_lib_serial(self, env: Env, httpd, nghttpx, proto, pause_offset):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 2
docname = 'data-10m'
url = f'https://localhost:{env.https_port}/{docname}'
self.check_downloads(client, srcfile, count)
# download parallel with prior knowledge
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_02_30_parallel_prior_knowledge(self, env: Env, httpd):
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
count = 3
curl = CurlClient(env=env)
urln = f'http://{env.domain1}:{env.http_port}/data.json?[0-{count-1}]'
assert r.total_connects == 1, r.dump_logs()
# download parallel with h2 "Upgrade:"
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_02_31_parallel_upgrade(self, env: Env, httpd, nghttpx):
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
count = 3
curl = CurlClient(env=env)
urln = f'http://{env.domain1}:{env.http_port}/data.json?[0-{count-1}]'
@pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx")
@pytest.mark.skipif(condition=not Env.curl_is_debug(), reason="needs curl debug")
@pytest.mark.skipif(condition=not Env.curl_is_verbose(), reason="needs curl verbose strings")
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_32_earlydata(self, env: Env, httpd, nghttpx, proto):
if not env.curl_can_early_data():
pytest.skip('TLS earlydata not implemented')
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and \
- (not env.have_h3() or not env.curl_can_h3_early_data()):
- pytest.skip("h3 not supported")
+ if proto == 'h3' and not env.curl_can_h3_early_data():
+ pytest.skip("h3 early data not supported")
if proto != 'h3' and sys.platform.startswith('darwin') and env.ci_run:
pytest.skip('failing on macOS CI runners')
if proto == 'h3' and sys.platform.startswith('darwin') and env.curl_uses_lib('wolfssl'):
elif proto == 'h3':
assert earlydata[1] == 109, f'{earlydata}'
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
@pytest.mark.parametrize("max_host_conns", [0, 1, 5])
def test_02_33_max_host_conns(self, env: Env, httpd, nghttpx, proto, max_host_conns):
if not env.curl_is_debug():
pytest.skip('only works for curl debug builds')
if not env.curl_is_verbose():
pytest.skip('only works for curl with verbose strings')
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 50
max_parallel = 50
docname = 'data-10k'
assert n <= max_host_conns
assert matched_lines > 0
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
@pytest.mark.parametrize("max_total_conns", [0, 1, 5])
def test_02_34_max_total_conns(self, env: Env, httpd, nghttpx, proto, max_total_conns):
if not env.curl_is_debug():
pytest.skip('only works for curl debug builds')
if not env.curl_is_verbose():
pytest.skip('only works for curl with verbose strings')
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 50
max_parallel = 50
docname = 'data-10k'
# * h2/h3: server continues sending what the stream window allows and
# since the one connection involved unpaused transfers, data continues
# to be received, requiring buffering.
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_02_35_pause_bomb(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 2
pause_offset = 1024 * 1024
docname = 'bomb-100m.txt.var'
r.check_exit_code(0)
# download with looong urls
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
@pytest.mark.parametrize("url_junk", [1024, 16*1024, 32*1024, 64*1024, 80*1024, 96*1024])
def test_02_36_looong_urls(self, env: Env, httpd, nghttpx, proto, url_junk):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('quiche'):
pytest.skip("quiche fails from 16k onwards")
curl = CurlClient(env=env)
class TestGoAway:
# download files sequentially with delay, reload server for GOAWAY
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_03_01_h2_goaway(self, env: Env, httpd, nghttpx):
proto = 'h2'
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
count = 3
self.r = None
class TestStuttered:
# download 1 file, check that delayed response works in general
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_04_01_download_1(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 1
curl = CurlClient(env=env)
urln = f'https://{env.authority_for(env.domain1, proto)}' \
# download 50 files in 100 chunks a 100 bytes with 10ms delay between
# prepend 100 file requests to warm up connection processing limits
# (Apache2 increases # of parallel processed requests after successes)
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_04_02_100_100_10(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 50
warmups = 100
curl = CurlClient(env=env)
# download 50 files in 1000 chunks a 10 bytes with 1ms delay between
# prepend 100 file requests to warm up connection processing limits
# (Apache2 increases # of parallel processed requests after successes)
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_04_03_1000_10_1(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 50
warmups = 100
curl = CurlClient(env=env)
# download 50 files in 10000 chunks a 1 byte with 10us delay between
# prepend 100 file requests to warm up connection processing limits
# (Apache2 increases # of parallel processed requests after successes)
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_04_04_1000_10_1(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 50
warmups = 100
curl = CurlClient(env=env)
class TestErrors:
# download 1 file, check that we get CURLE_PARTIAL_FILE
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_05_01_partial_1(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 1
curl = CurlClient(env=env)
urln = f'https://{env.authority_for(env.domain1, proto)}' \
assert len(invalid_stats) == 0, f'failed: {invalid_stats}'
# download files, check that we get CURLE_PARTIAL_FILE for all
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_05_02_partial_20(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_ossl_quic():
pytest.skip("openssl-quic is flaky in yielding proper error codes")
count = 20
assert len(invalid_stats) == 0, f'failed: {invalid_stats}'
# access a resource that, on h2, RST the stream with HTTP_1_1_REQUIRED
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_05_03_required(self, env: Env, httpd, nghttpx):
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
curl = CurlClient(env=env)
proto = 'http/1.1'
urln = f'https://{env.authority_for(env.domain1, proto)}/curltest/1_1'
env.make_data_file(indir=env.gen_dir, fname="data-10m", fsize=10*1024*1024)
# upload small data, check that this is what was echoed
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_01_upload_1_small(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
data = '0123456789'
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/curltest/echo?id=[0-0]'
assert respdata == [data]
# upload large data, check that this is what was echoed
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_02_upload_1_large(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
fdata = os.path.join(env.gen_dir, 'data-100k')
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/curltest/echo?id=[0-0]'
assert respdata == indata
# upload data sequentially, check that they were echoed
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_10_upload_sequential(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 20
data = '0123456789'
curl = CurlClient(env=env)
assert respdata == [data]
# upload data parallel, check that they were echoed
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_07_11_upload_parallel(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
# limit since we use a separate connection in h1
count = 20
data = '0123456789'
assert respdata == [data]
# upload large data sequentially, check that this is what was echoed
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_12_upload_seq_large(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
fdata = os.path.join(env.gen_dir, 'data-100k')
count = 10
curl = CurlClient(env=env)
assert respdata == indata
# upload very large data sequentially, check that this is what was echoed
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_13_upload_seq_large(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
fdata = os.path.join(env.gen_dir, 'data-10m')
count = 2
curl = CurlClient(env=env)
assert respdata == indata
# upload from stdin, issue #14870
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
@pytest.mark.parametrize("indata", [
'', '1', '123\n456andsomething\n\n'
])
def test_07_14_upload_stdin(self, env: Env, httpd, nghttpx, proto, indata):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 1
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/curltest/put?id=[0-{count-1}]'
respdata = open(curl.response_file(i)).readlines()
assert respdata == [f'{len(indata)}']
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_15_hx_put(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 2
upload_size = 128*1024
url = f'https://localhost:{env.https_port}/curltest/put'
r.check_exit_code(0)
self.check_downloads(client, r, [f"{upload_size}"], count)
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_16_hx_put_reuse(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 2
upload_size = 128*1024
url = f'https://localhost:{env.https_port}/curltest/put'
r.check_exit_code(0)
self.check_downloads(client, r, [f"{upload_size}"], count)
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_17_hx_post_reuse(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 2
upload_size = 128*1024
url = f'https://localhost:{env.https_port}/curltest/echo'
self.check_downloads(client, r, ["x" * upload_size], count)
# upload data parallel, check that they were echoed
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_07_20_upload_parallel(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
# limit since we use a separate connection in h1
count = 10
data = '0123456789'
assert respdata == [data]
# upload large data parallel, check that this is what was echoed
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_07_21_upload_parallel_large(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
fdata = os.path.join(env.gen_dir, 'data-100k')
# limit since we use a separate connection in h1
count = 10
# (We used to do this for 20 parallel transfers, but the triggered
# stream resets make nghttpx drop the connection after several, which
# then gives a non-deterministic number of completely failed transfers)
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_07_22_upload_fail(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
fdata = os.path.join(env.gen_dir, 'data-10m')
count = 1
curl = CurlClient(env=env)
r.check_stats(count=count, exitcode=[18, 55, 56, 92, 95])
# PUT 100k
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_30_put_100k(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
fdata = os.path.join(env.gen_dir, 'data-100k')
count = 1
curl = CurlClient(env=env)
assert respdata == exp_data
# PUT 10m
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_31_put_10m(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
fdata = os.path.join(env.gen_dir, 'data-10m')
count = 1
curl = CurlClient(env=env)
assert respdata == exp_data
# issue #10591
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_32_issue_10591(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
fdata = os.path.join(env.gen_dir, 'data-10m')
count = 1
curl = CurlClient(env=env)
# issue #11157, upload that is 404'ed by server, needs to terminate
# correctly and not time out on sending
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_07_33_issue_11157a(self, env: Env, httpd, nghttpx):
proto = 'h2'
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
fdata = os.path.join(env.gen_dir, 'data-10m')
# send a POST to our PUT handler which will send immediately a 404 back
url = f'https://{env.authority_for(env.domain1, proto)}/curltest/put'
r.check_stats(1, 404)
# issue #11157, send upload that is slowly read in
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_07_33_issue_11157b(self, env: Env, httpd, nghttpx):
proto = 'h2'
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
fdata = os.path.join(env.gen_dir, 'data-10m')
# tell our test PUT handler to read the upload more slowly, so
# that the send buffering and transfer loop needs to wait
assert r.exit_code == 0, r.dump_logs()
r.check_stats(1, 200)
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_07_34_issue_11194(self, env: Env, httpd, nghttpx):
proto = 'h2'
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
# tell our test PUT handler to read the upload more slowly, so
# that the send buffering and transfer loop needs to wait
fdata = os.path.join(env.gen_dir, 'data-100k')
r.check_stats(1, 200)
# upload large data on a h1 to h2 upgrade
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_07_35_h1_h2_upgrade_upload(self, env: Env, httpd, nghttpx):
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
fdata = os.path.join(env.gen_dir, 'data-100k')
curl = CurlClient(env=env)
url = f'http://{env.domain1}:{env.http_port}/curltest/echo?id=[0-0]'
# upload to a 301,302,303 response
@pytest.mark.parametrize("redir", ['301', '302', '303'])
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_36_upload_30x(self, env: Env, httpd, nghttpx, redir, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_ossl_quic():
pytest.skip("OpenSSL's own QUIC is flaky here")
data = '0123456789' * 10
assert respdata == [] # was transformed to a GET
# upload to a 307 response
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_37_upload_307(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_ossl_quic():
pytest.skip("OpenSSL's own QUIC is flaky here")
data = '0123456789' * 10
assert respdata == [data] # was POST again
# POST form data, yet another code path in transfer
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_38_form_small(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/curltest/echo?id=[0-0]'
r = curl.http_form(urls=[url], alpn_proto=proto, form={
r.check_stats(count=1, http_status=200, exitcode=0)
# POST data urlencoded, small enough to be sent with request headers
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_39_post_urlenc_small(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
fdata = os.path.join(env.gen_dir, 'data-63k')
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/curltest/echo?id=[0-0]'
assert respdata == indata
# POST data urlencoded, large enough to be sent separate from request headers
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_40_post_urlenc_large(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
fdata = os.path.join(env.gen_dir, 'data-64k')
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/curltest/echo?id=[0-0]'
# than our default upload buffer length (64KB).
# Unfixed, this will fail when run with CURL_DBG_SOCK_WBLOCK=80 most
# of the time
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_41_post_urlenc_small(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('quiche'):
pytest.skip("quiche has CWND issues with large requests")
fdata = os.path.join(env.gen_dir, 'data-63k')
# upload data, pause, let connection die with an incomplete response
# issues #11769 #13260
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_42a_upload_disconnect(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
client = LocalClient(name='cli_upload_pausing', env=env, timeout=60)
if not client.exists():
pytest.skip(f'example client not built: {client.name}')
r.check_exit_code(18) # will fail as it should
# upload data, pause, let connection die without any response at all
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_42b_upload_disconnect(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
client = LocalClient(name='cli_upload_pausing', env=env, timeout=60)
if not client.exists():
pytest.skip(f'example client not built: {client.name}')
r.check_exit_code(exp_code) # GOT_NOTHING
# upload data, pause, let connection die after 100 continue
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_42c_upload_disconnect(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
client = LocalClient(name='cli_upload_pausing', env=env, timeout=60)
if not client.exists():
pytest.skip(f'example client not built: {client.name}')
exp_code = 0 # we get a 500 from the server
r.check_exit_code(exp_code) # GOT_NOTHING
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_43_upload_denied(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_ossl_quic():
pytest.skip("openssl-quic is flaky in filed PUTs")
fdata = os.path.join(env.gen_dir, 'data-10m')
extra_args=['--trace-config', 'all'])
r.check_stats(count=count, http_status=413, exitcode=0)
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
@pytest.mark.parametrize("httpcode", [301, 302, 307, 308])
def test_07_44_put_redir(self, env: Env, httpd, nghttpx, proto, httpcode):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 1
upload_size = 128*1024
url = f'https://localhost:{env.https_port}/curltest/put-redir-{httpcode}'
assert httpcodes[0] == httpcode, f'{r}'
# speed limited on put handler
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_50_put_speed_limit(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 1
fdata = os.path.join(env.gen_dir, 'data-100k')
up_len = 100 * 1024
assert (speed_limit * 0.5) <= up_speed <= (speed_limit * 1.5), f'{r.stats[0]}'
# speed limited on echo handler
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_07_51_echo_speed_limit(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 1
fdata = os.path.join(env.gen_dir, 'data-100k')
speed_limit = 50 * 1024
env.make_data_file(indir=env.gen_dir, fname="data-10m", fsize=10*1024*1024)
# download 1 file
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_08_01_download_1(self, env: Env, caddy: Caddy, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3_curl():
- pytest.skip("h3 not supported in curl")
curl = CurlClient(env=env)
url = f'https://{env.domain1}:{caddy.port}/data.json'
r = curl.http_download(urls=[url], alpn_proto=proto)
r.check_response(count=1, http_status=200)
# download 1MB files sequentially
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_08_02_download_1mb_sequential(self, env: Env, caddy: Caddy, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3_curl():
- pytest.skip("h3 not supported in curl")
count = 50
curl = CurlClient(env=env)
urln = f'https://{env.domain1}:{caddy.port}/data1.data?[0-{count-1}]'
r.check_response(count=count, http_status=200, connect_count=1)
# download 1MB files parallel
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_08_03_download_1mb_parallel(self, env: Env, caddy: Caddy, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3_curl():
- pytest.skip("h3 not supported in curl")
count = 20
curl = CurlClient(env=env)
urln = f'https://{env.domain1}:{caddy.port}/data1.data?[0-{count-1}]'
# download 5MB files sequentially
@pytest.mark.skipif(condition=Env().slow_network, reason="not suitable for slow network tests")
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_08_04a_download_10mb_sequential(self, env: Env, caddy: Caddy, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3_curl():
- pytest.skip("h3 not supported in curl")
count = 40
curl = CurlClient(env=env)
urln = f'https://{env.domain1}:{caddy.port}/data5.data?[0-{count-1}]'
# download 10MB files sequentially
@pytest.mark.skipif(condition=Env().slow_network, reason="not suitable for slow network tests")
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_08_04b_download_10mb_sequential(self, env: Env, caddy: Caddy, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3_curl():
- pytest.skip("h3 not supported in curl")
count = 20
curl = CurlClient(env=env)
urln = f'https://{env.domain1}:{caddy.port}/data10.data?[0-{count-1}]'
# download 10MB files parallel
@pytest.mark.skipif(condition=Env().slow_network, reason="not suitable for slow network tests")
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_08_05_download_1mb_parallel(self, env: Env, caddy: Caddy, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3_curl():
- pytest.skip("h3 not supported in curl")
if proto == 'http/1.1' and env.curl_uses_lib('mbedtls'):
pytest.skip("mbedtls 3.6.0 fails on 50 connections with: "
"ssl_handshake returned: (-0x7F00) SSL - Memory allocation failed")
assert r.total_connects == 1, r.dump_logs()
# post data parallel, check that they were echoed
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_08_06_post_parallel(self, env: Env, httpd, caddy, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
# limit since we use a separate connection in h1
count = 20
data = '0123456789'
assert respdata == [data]
# put large file, check that they length were echoed
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_08_07_put_large(self, env: Env, httpd, caddy, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
# limit since we use a separate connection in h1<
count = 1
fdata = os.path.join(env.gen_dir, 'data-10m')
respdata = open(curl.response_file(i)).readlines()
assert respdata == exp_data
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_08_08_earlydata(self, env: Env, httpd, caddy, proto):
if not env.curl_can_early_data():
pytest.skip('TLS earlydata not implemented')
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and \
- (not env.have_h3() or not env.curl_can_h3_early_data()):
- pytest.skip("h3 not supported")
+ if proto == 'h3' and not env.curl_can_h3_early_data():
+ pytest.skip("h3 early data not supported")
count = 2
docname = 'data10k.data'
url = f'https://{env.domain1}:{caddy.port}/{docname}'
httpd.reload_if_config_changed()
# download a file that triggers a "103 Early Hints" response
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_09_01_h2_early_hints(self, env: Env, httpd, configures_httpd):
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
self.httpd_configure(env, httpd)
curl = CurlClient(env=env)
url = f'https://{env.domain1}:{env.https_port}/push/data1'
assert 'link' in r.responses[0]['header'], f'{r.responses[0]}'
assert r.responses[0]['header']['link'] == '</push/data2>; rel=preload', f'{r.responses[0]}'
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_09_02_h2_push(self, env: Env, httpd, configures_httpd):
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
self.httpd_configure(env, httpd)
# use localhost as we do not have resolve support in local client
url = f'https://localhost:{env.https_port}/push/data1'
# download via https: proxy (no tunnel)
@pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'),
reason='curl lacks HTTPS-proxy support')
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
def test_10_02_proxys_down(self, env: Env, httpd, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
if proto == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
curl = CurlClient(env=env)
# upload via https: with proto (no tunnel)
@pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
@pytest.mark.parametrize("fname, fcount", [
['data.json', 5],
['data-100k', 5],
reason="no nghttpx available")
def test_10_02_proxys_up(self, env: Env, httpd, nghttpx, proto,
fname, fcount):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
if proto == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
count = fcount
r.check_response(count=1, http_status=200)
# download https: with proto via http: proxytunnel
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
@pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
def test_10_05_proxytunnel_http(self, env: Env, httpd, nghttpx_fwd, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
curl = CurlClient(env=env)
url = f'https://localhost:{env.https_port}/data.json'
xargs = curl.get_proxy_args(proxys=False, tunnel=True)
# download https: with proto via https: proxytunnel
@pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'),
reason='curl lacks HTTPS-proxy support')
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
- @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
+ @pytest.mark.parametrize("tunnel", Env.http_h1_h2_protos())
@pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
@pytest.mark.skipif(condition=not Env.curl_is_debug(), reason="needs curl debug")
@pytest.mark.skipif(condition=not Env.curl_is_verbose(), reason="needs curl verbose strings")
def test_10_06_proxytunnel_https(self, env: Env, httpd, nghttpx_fwd, proto, tunnel):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
curl = CurlClient(env=env)
@pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
@pytest.mark.skipif(condition=not Env.curl_is_debug(), reason="needs curl debug")
@pytest.mark.skipif(condition=not Env.curl_is_verbose(), reason="needs curl verbose strings")
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
- @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
+ @pytest.mark.parametrize("tunnel", Env.http_h1_h2_protos())
@pytest.mark.parametrize("fname, fcount", [
['data.json', 100],
['data-100k', 20],
@pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
def test_10_07_pts_down_small(self, env: Env, httpd, nghttpx_fwd, proto,
tunnel, fname, fcount):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
if env.curl_uses_lib('mbedtls') and \
# upload many https: with proto via https: proxytunnel
@pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
- @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
+ @pytest.mark.parametrize("tunnel", Env.http_h1_h2_protos())
@pytest.mark.parametrize("fname, fcount", [
['data.json', 50],
['data-100k', 20],
@pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
def test_10_08_upload_seq_large(self, env: Env, httpd, nghttpx, proto,
tunnel, fname, fcount):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
if env.curl_uses_lib('mbedtls') and \
assert respdata == indata, f'response {i} differs'
@pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
- @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("tunnel", Env.http_h1_h2_protos())
@pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
@pytest.mark.skipif(condition=not Env.curl_is_debug(), reason="needs curl debug")
@pytest.mark.skipif(condition=not Env.curl_is_verbose(), reason="needs curl verbose strings")
def test_10_09_reuse_server(self, env: Env, httpd, nghttpx_fwd, tunnel):
- if tunnel == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
curl = CurlClient(env=env)
assert r.total_connects == 2
@pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
- @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("tunnel", Env.http_h1_h2_protos())
@pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
@pytest.mark.skipif(condition=not Env.curl_is_debug(), reason="needs curl debug")
@pytest.mark.skipif(condition=not Env.curl_is_verbose(), reason="needs curl verbose strings")
def test_10_10_reuse_proxy(self, env: Env, httpd, nghttpx_fwd, tunnel):
# url twice via https: proxy separated with '--next', will reuse
- if tunnel == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
if env.curl_uses_lib('mbedtls') and \
assert r2.total_connects == 1
@pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
- @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("tunnel", Env.http_h1_h2_protos())
@pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
@pytest.mark.skipif(condition=not Env.curl_uses_lib('openssl'), reason="tls13-ciphers not supported")
@pytest.mark.skipif(condition=not Env.curl_is_debug(), reason="needs curl debug")
@pytest.mark.skipif(condition=not Env.curl_is_verbose(), reason="needs curl verbose strings")
def test_10_11_noreuse_proxy_https(self, env: Env, httpd, nghttpx_fwd, tunnel):
# different --proxy-tls13-ciphers, no reuse of connection for https:
- if tunnel == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
curl = CurlClient(env=env)
if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
assert r2.total_connects == 2
@pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
- @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("tunnel", Env.http_h1_h2_protos())
@pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
@pytest.mark.skipif(condition=not Env.curl_uses_lib('openssl'), reason="tls13-ciphers not supported")
@pytest.mark.skipif(condition=not Env.curl_is_debug(), reason="needs curl debug")
@pytest.mark.skipif(condition=not Env.curl_is_verbose(), reason="needs curl verbose strings")
def test_10_12_noreuse_proxy_http(self, env: Env, httpd, nghttpx_fwd, tunnel):
# different --proxy-tls13-ciphers, no reuse of connection for http:
- if tunnel == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
curl = CurlClient(env=env)
assert r2.total_connects == 2
@pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
- @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("tunnel", Env.http_h1_h2_protos())
@pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
@pytest.mark.skipif(condition=not Env.curl_uses_lib('openssl'), reason="tls13-ciphers not supported")
@pytest.mark.skipif(condition=not Env.curl_is_debug(), reason="needs curl debug")
@pytest.mark.skipif(condition=not Env.curl_is_verbose(), reason="needs curl verbose strings")
def test_10_13_noreuse_https(self, env: Env, httpd, nghttpx_fwd, tunnel):
# different --tls13-ciphers on https: same proxy config
- if tunnel == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
curl = CurlClient(env=env)
# download via https: proxy (no tunnel) using IP address
@pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'),
reason='curl lacks HTTPS-proxy support')
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
def test_10_14_proxys_ip_addr(self, env: Env, httpd, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
if proto == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
curl = CurlClient(env=env)
reason='curl lacks HTTPS-proxy support')
@pytest.mark.skipif(condition=not Env.curl_is_debug(), reason="needs curl debug")
@pytest.mark.skipif(condition=not Env.curl_is_verbose(), reason="needs curl verbose strings")
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
- @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
+ @pytest.mark.parametrize("tunnel", Env.http_h1_h2_protos())
def test_13_07_tunnels_no_auth(self, env: Env, httpd, configures_httpd, nghttpx_fwd, proto, tunnel):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
self.httpd_configure(env, httpd)
if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
reason='curl lacks HTTPS-proxy support')
@pytest.mark.skipif(condition=not Env.curl_is_debug(), reason="needs curl debug")
@pytest.mark.skipif(condition=not Env.curl_is_verbose(), reason="needs curl verbose strings")
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
- @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
+ @pytest.mark.parametrize("tunnel", Env.http_h1_h2_protos())
def test_13_08_tunnels_auth(self, env: Env, httpd, configures_httpd, nghttpx_fwd, proto, tunnel):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
self.httpd_configure(env, httpd)
if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
env.make_data_file(indir=env.gen_dir, fname="data-10m", fsize=10*1024*1024)
# download 1 file, not authenticated
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_14_01_digest_get_noauth(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/restricted/digest/data.json'
r = curl.http_download(urls=[url], alpn_proto=proto)
r.check_response(http_status=401)
# download 1 file, authenticated
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_14_02_digest_get_auth(self, env: Env, httpd, nghttpx, proto):
if not env.curl_has_feature('digest'):
pytest.skip("curl built without digest")
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/restricted/digest/data.json'
r = curl.http_download(urls=[url], alpn_proto=proto, extra_args=[
r.check_response(http_status=200)
# PUT data, authenticated
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_14_03_digest_put_auth(self, env: Env, httpd, nghttpx, proto):
if not env.curl_has_feature('digest'):
pytest.skip("curl built without digest")
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_ossl_quic():
pytest.skip("openssl-quic is flaky in retrying POST")
data='0123456789'
r.check_response(http_status=200)
# PUT data, digest auth large pw
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_14_04_digest_large_pw(self, env: Env, httpd, nghttpx, proto):
if not env.curl_has_feature('digest'):
pytest.skip("curl built without digest")
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
data='0123456789'
password = 'x' * 65535
curl = CurlClient(env=env)
r.check_response(http_status=401)
# PUT data, basic auth large pw
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_14_05_basic_large_pw(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if proto == 'h3' and not env.curl_uses_lib('ngtcp2'):
# See <https://github.com/cloudflare/quiche/issues/1573>
pytest.skip("quiche/openssl-quic have problems with large requests")
r.check_response(http_status=431)
# PUT data, basic auth with very large pw
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_14_06_basic_very_large_pw(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('quiche'):
# See <https://github.com/cloudflare/quiche/issues/1573>
pytest.skip("quiche has problems with large requests")
env.make_data_file(indir=env.gen_dir, fname="data-100k", fsize=100*1024)
# download plain file
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_16_01_info_download(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 2
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/data.json?[0-{count-1}]'
self.check_stat(idx, s, r, dl_size=30, ul_size=0)
# download plain file with a 302 redirect
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_16_02_info_302_download(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 2
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/data.json.302?[0-{count-1}]'
for idx, s in enumerate(r.stats):
self.check_stat(idx, s, r, dl_size=30, ul_size=0)
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_16_03_info_upload(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 2
fdata = os.path.join(env.gen_dir, 'data-100k')
fsize = 100 * 1024
assert djson['SSL_SESSION_RESUMED'] == exp_resumed, f'{i}: {djson}\n{r.dump_logs()}'
# use host name with trailing dot, verify handshake
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_17_03_trailing_dot(self, env: Env, proto, httpd, nghttpx):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
curl = CurlClient(env=env)
domain = f'{env.domain1}.'
url = f'https://{env.authority_for(domain, proto)}/curltest/sslinfo'
assert r.json['SSL_TLS_SNI'] == env.domain1, f'{r.json}'
# use host name with double trailing dot, verify handshake
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_17_04_double_dot(self, env: Env, proto, httpd, nghttpx):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
curl = CurlClient(env=env)
domain = f'{env.domain1}..'
url = f'https://{env.authority_for(domain, proto)}/curltest/sslinfo'
assert r.exit_code in [7, 35, 60], f'{r}'
# use ip address for connect
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_17_05_good_ip_addr(self, env: Env, proto, httpd, nghttpx):
if env.curl_uses_lib('mbedtls'):
pytest.skip("mbedTLS does use IP addresses in SNI")
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
curl = CurlClient(env=env)
domain = '127.0.0.1'
url = f'https://{env.authority_for(domain, proto)}/curltest/sslinfo'
assert 'SSL_TLS_SNI' not in r.json, f'{r.json}'
# use IP address that is not in cert
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_17_05_bad_ip_addr(self, env: Env, proto,
httpd, configures_httpd,
nghttpx, configures_nghttpx):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
httpd.set_domain1_cred_name('domain1-no-ip')
httpd.reload_if_config_changed()
if proto == 'h3':
assert r.exit_code == 60, f'{r}'
# use IP address that is in cert as DNS name (not really legal)
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_17_05_very_bad_ip_addr(self, env: Env, proto,
httpd, configures_httpd,
nghttpx, configures_nghttpx):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if env.curl_uses_lib('mbedtls'):
pytest.skip("mbedtls falsely verifies a DNS: altname as IP address")
if env.curl_uses_lib('wolfssl') and \
assert r.exit_code == 60, f'{r}'
# use localhost for connect
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_17_06_localhost(self, env: Env, proto, httpd, nghttpx):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
curl = CurlClient(env=env)
domain = 'localhost'
url = f'https://{env.authority_for(domain, proto)}/curltest/sslinfo'
else:
assert r.exit_code != 0, r.dump_logs()
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_17_08_cert_status(self, env: Env, proto, httpd, nghttpx):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if not env.curl_uses_lib('openssl') and \
not env.curl_uses_lib('gnutls') and \
not env.curl_uses_lib('quictls'):
assert reused_session, f'{r}\n{r.dump_logs()}'
# use host name server has no certificate for
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_17_11_wrong_host(self, env: Env, proto, httpd, nghttpx):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
curl = CurlClient(env=env)
domain = f'insecure.{env.tld}'
url = f'https://{domain}:{env.port_for(proto)}/curltest/sslinfo'
assert r.exit_code == 60, f'{r}'
# use host name server has no cert for with --insecure
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_17_12_insecure(self, env: Env, proto, httpd, nghttpx):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
curl = CurlClient(env=env)
domain = f'insecure.{env.tld}'
url = f'https://{domain}:{env.port_for(proto)}/curltest/sslinfo'
assert r.json, f'{r}'
# connect to an expired certificate
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
def test_17_14_expired_cert(self, env: Env, proto, httpd):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
curl = CurlClient(env=env)
url = f'https://{env.expired_domain}:{env.port_for(proto)}/'
r = curl.http_get(url=url, alpn_proto=proto)
else:
assert r.exit_code != 0, r.dump_logs()
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_17_19_wrong_pin(self, env: Env, proto, httpd):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
+ @pytest.mark.parametrize("proto", Env.http_protos())
+ def test_17_19_wrong_pin(self, env: Env, proto, httpd, nghttpx):
if env.curl_uses_lib('rustls-ffi'):
pytest.skip('TLS backend ignores --pinnedpubkey')
curl = CurlClient(env=env)
# expect NOT_IMPLEMENTED or CURLE_SSL_PINNEDPUBKEYNOTMATCH
assert r.exit_code in [2, 90], f'{r.dump_logs()}'
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_17_20_correct_pin(self, env: Env, proto, httpd):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
+ @pytest.mark.parametrize("proto", Env.http_protos())
+ def test_17_20_correct_pin(self, env: Env, proto, httpd, nghttpx):
curl = CurlClient(env=env)
creds = env.get_credentials(env.domain1)
assert creds
env.make_data_file(indir=indir, fname="data-1m", fsize=1024*1024)
# download 1 file
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_18_01_delete(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
count = 1
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/curltest/tweak?id=[0-{count-1}]'
# - HEADER frame with 204 and eos=0
# - 10ms later DATA frame length=0 and eos=1
# should be accepted
+ @pytest.mark.skipif(condition=not Env.have_h2_curl(), reason="curl without h2")
def test_18_02_delete_h2_special(self, env: Env, httpd, nghttpx):
proto = 'h2'
- if not env.have_h2_curl():
- pytest.skip("h2 not supported")
count = 1
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/curltest/tweak?id=[0-{count-1}]'\
# check with `tcpdump` that we do NOT see TCP RST when CURL_GRACEFUL_SHUTDOWN set
@pytest.mark.skipif(condition=not Env.tcpdump(), reason="tcpdump not available")
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
def test_19_02_check_shutdown(self, env: Env, httpd, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
if not env.curl_is_debug():
pytest.skip('only works for curl debug builds')
run_env = os.environ.copy()
assert len(removes) == count, f'{removes}'
# check graceful shutdown on multiplexed http
- @pytest.mark.parametrize("proto", ['h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_mplx_protos())
def test_19_06_check_shutdown(self, env: Env, httpd, nghttpx, proto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
if not env.curl_is_debug():
pytest.skip('only works for curl debug builds')
if not env.curl_is_verbose():
r.check_response(http_status=200)
@pytest.mark.parametrize("sproto", ['socks4', 'socks5'])
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+ @pytest.mark.parametrize("proto", Env.http_protos())
def test_40_02_socks_https(self, env: Env, sproto, proto, danted: Dante, httpd):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
- if proto == 'h3' and not env.have_h3():
- pytest.skip("h3 not supported")
curl = CurlClient(env=env, socks_args=[
f'--{sproto}', f'127.0.0.1:{danted.port}'
])
r.check_response(http_status=200)
@pytest.mark.parametrize("sproto", ['socks4', 'socks5'])
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
def test_40_03_dl_serial(self, env: Env, httpd, danted, proto, sproto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
count = 3
urln = f'https://{env.authority_for(env.domain1, proto)}/data-10m?[0-{count-1}]'
curl = CurlClient(env=env, socks_args=[
r.check_response(count=count, http_status=200)
@pytest.mark.parametrize("sproto", ['socks4', 'socks5'])
- @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
+ @pytest.mark.parametrize("proto", Env.http_h1_h2_protos())
def test_40_04_ul_serial(self, env: Env, httpd, danted, proto, sproto):
- if proto == 'h2' and not env.have_h2_curl():
- pytest.skip("h2 not supported")
fdata = os.path.join(env.gen_dir, 'data-10m')
count = 2
curl = CurlClient(env=env, socks_args=[
return Env.curl_can_early_data() and \
Env.curl_uses_lib('ngtcp2')
+ @staticmethod
+ def http_protos() -> List[str]:
+ # http protocols we can test
+ if Env.have_h2_curl():
+ if Env.have_h3():
+ return ['http/1.1', 'h2', 'h3']
+ else:
+ return ['http/1.1', 'h2']
+ else:
+ return ['http/1.1']
+
+ @staticmethod
+ def http_h1_h2_protos() -> List[str]:
+ # http 1+2 protocols we can test
+ if Env.have_h2_curl():
+ return ['http/1.1', 'h2']
+ else:
+ return ['http/1.1']
+
+ @staticmethod
+ def http_mplx_protos() -> List[str]:
+ # http multiplexing protocols we can test
+ if Env.have_h2_curl():
+ if Env.have_h3():
+ return ['h2', 'h3']
+ else:
+ return ['h2']
+ else:
+ return []
+
@staticmethod
def have_h3() -> bool:
return Env.have_h3_curl() and Env.have_h3_server()
def issue_certs(self):
if self._ca is None:
- ca_dir = os.path.join(self.CONFIG.gen_root, 'ca')
+ # ca_dir = os.path.join(self.CONFIG.gen_root, 'ca')
+ ca_dir = os.path.join(self.gen_dir, 'ca')
os.makedirs(ca_dir, exist_ok=True)
lock_file = os.path.join(ca_dir, 'ca.lock')
with FileLock(lock_file):
projects / thirdparty / curl.git / commitdiff
