fprintf(stderr,
" -A number abort transfer after `number` response bytes\n"
" -F number fail writing response after `number` response bytes\n"
+ " -M number max concurrent connections to a host\n"
" -P number pause transfer after `number` response bytes\n"
" -r <host>:<port>:<addr> resolve information\n"
+ " -T number max concurrent connections total\n"
" -V http_version (http/1.1, h2, h3) http version to use\n"
);
}
# 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'])
- def test_01_06_timings(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_01_06_timings(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
curl = CurlClient(env=env)
# simple https: HEAD
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
@pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
- def test_01_07_head(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_01_07_head(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
curl = CurlClient(env=env)
# download 1 file
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_02_01_download_1(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_02_01_download_1(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
curl = CurlClient(env=env)
# download 2 files
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_02_02_download_2(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_02_02_download_2(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
curl = CurlClient(env=env)
# download 100 files sequentially
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_02_03_download_sequential(self, env: Env,
- httpd, nghttpx, repeat, proto):
+ def test_02_03_download_sequential(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 10
# download 100 files parallel
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_02_04_download_parallel(self, env: Env,
- httpd, nghttpx, repeat, proto):
+ def test_02_04_download_parallel(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 10
# download 500 files sequential
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_02_05_download_many_sequential(self, env: Env,
- httpd, nghttpx, repeat, proto):
+ def test_02_05_download_many_sequential(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# download 500 files parallel
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_02_06_download_many_parallel(self, env: Env,
- httpd, nghttpx, repeat, proto):
+ def test_02_06_download_many_parallel(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 200
# download files parallel, check connection reuse/multiplex
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_02_07_download_reuse(self, env: Env,
- httpd, nghttpx, repeat, proto):
+ def test_02_07_download_reuse(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 200
# download files parallel with http/1.1, check connection not reused
@pytest.mark.parametrize("proto", ['http/1.1'])
- def test_02_07b_download_reuse(self, env: Env,
- httpd, nghttpx, repeat, proto):
+ def test_02_07b_download_reuse(self, env: Env, httpd, nghttpx, proto):
count = 6
curl = CurlClient(env=env)
urln = f'https://{env.authority_for(env.domain1, proto)}/data.json?[0-{count-1}]'
assert r.total_connects == count, "http/1.1 should use this many connections"
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_02_08_1MB_serial(self, env: Env,
- httpd, nghttpx, repeat, proto):
+ def test_02_08_1MB_serial(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 5
r.check_response(count=count, http_status=200)
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_02_09_1MB_parallel(self, env: Env,
- httpd, nghttpx, repeat, proto):
+ def test_02_09_1MB_parallel(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 5
@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'])
- def test_02_10_10MB_serial(self, env: Env,
- httpd, nghttpx, repeat, proto):
+ def test_02_10_10MB_serial(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 3
@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'])
- def test_02_11_10MB_parallel(self, env: Env,
- httpd, nghttpx, repeat, proto):
+ def test_02_11_10MB_parallel(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
r.check_response(count=count, http_status=200)
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_02_12_head_serial_https(self, env: Env,
- httpd, nghttpx, repeat, proto):
+ def test_02_12_head_serial_https(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 5
r.check_response(count=count, http_status=200)
@pytest.mark.parametrize("proto", ['h2'])
- def test_02_13_head_serial_h2c(self, env: Env,
- httpd, nghttpx, repeat, proto):
+ def test_02_13_head_serial_h2c(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 5
r.check_response(count=count, http_status=200)
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_02_14_not_found(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_02_14_not_found(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
remote_ip='127.0.0.1')
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_02_15_fail_not_found(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_02_15_fail_not_found(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
remote_ip='127.0.0.1')
@pytest.mark.skipif(condition=Env().slow_network, reason="not suitable for slow network tests")
- def test_02_20_h2_small_frames(self, env: Env, httpd, repeat):
+ def test_02_20_h2_small_frames(self, env: Env, httpd):
# 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
# 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'])
- def test_02_21_lib_serial(self, env: Env, httpd, nghttpx, proto, pause_offset, repeat):
+ def test_02_21_lib_serial(self, env: Env, httpd, nghttpx, proto, pause_offset):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 2
# 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'])
- def test_02_22_lib_parallel_resume(self, env: Env, httpd, nghttpx, proto, pause_offset, repeat):
+ def test_02_22_lib_parallel_resume(self, env: Env, httpd, nghttpx, proto, pause_offset):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 2
# download, several at a time, pause and abort paused
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_02_23a_lib_abort_paused(self, env: Env, httpd, nghttpx, proto, repeat):
+ def test_02_23a_lib_abort_paused(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_ossl_quic():
# download, several at a time, abort after n bytes
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_02_23b_lib_abort_offset(self, env: Env, httpd, nghttpx, proto, repeat):
+ def test_02_23b_lib_abort_offset(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_ossl_quic():
# download, several at a time, abort after n bytes
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_02_23c_lib_fail_offset(self, env: Env, httpd, nghttpx, proto, repeat):
+ def test_02_23c_lib_fail_offset(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_ossl_quic():
# speed limited download
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_02_24_speed_limit(self, env: Env, httpd, nghttpx, proto, repeat):
+ def test_02_24_speed_limit(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 1
# make extreme parallel h2 upgrades, check invalid conn reuse
# before protocol switch has happened
- def test_02_25_h2_upgrade_x(self, env: Env, httpd, repeat):
+ def test_02_25_h2_upgrade_x(self, env: Env, httpd):
url = f'http://localhost:{env.http_port}/data-100k'
client = LocalClient(name='h2-upgrade-extreme', env=env, timeout=15)
if not client.exists():
# 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'])
- def test_02_26_session_shared_reuse(self, env: Env, proto, httpd, nghttpx, repeat):
+ def test_02_26_session_shared_reuse(self, env: Env, proto, httpd, nghttpx):
url = f'https://{env.authority_for(env.domain1, proto)}/data-100k'
client = LocalClient(name='tls-session-reuse', env=env)
if not client.exists():
# test on paused transfers, based on issue #11982
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_02_27a_paused_no_cl(self, env: Env, httpd, nghttpx, proto, repeat):
+ def test_02_27a_paused_no_cl(self, env: Env, httpd, nghttpx, proto):
url = f'https://{env.authority_for(env.domain1, proto)}' \
'/curltest/tweak/?&chunks=6&chunk_size=8000'
client = LocalClient(env=env, name='h2-pausing')
# test on paused transfers, based on issue #11982
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_02_27b_paused_no_cl(self, env: Env, httpd, nghttpx, proto, repeat):
+ def test_02_27b_paused_no_cl(self, env: Env, httpd, nghttpx, proto):
url = f'https://{env.authority_for(env.domain1, proto)}' \
'/curltest/tweak/?error=502'
client = LocalClient(env=env, name='h2-pausing')
# test on paused transfers, based on issue #11982
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_02_27c_paused_no_cl(self, env: Env, httpd, nghttpx, proto, repeat):
+ def test_02_27c_paused_no_cl(self, env: Env, httpd, nghttpx, proto):
url = f'https://{env.authority_for(env.domain1, proto)}' \
'/curltest/tweak/?status=200&chunks=1&chunk_size=100'
client = LocalClient(env=env, name='h2-pausing')
r.check_exit_code(0)
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_02_28_get_compressed(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_02_28_get_compressed(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 1
# 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'])
- def test_02_29_h2_lib_serial(self, env: Env, httpd, nghttpx, proto, pause_offset, repeat):
+ def test_02_29_h2_lib_serial(self, env: Env, httpd, nghttpx, proto, pause_offset):
count = 2
docname = 'data-10m'
url = f'https://localhost:{env.https_port}/{docname}'
httpd.reload()
# download files sequentially with delay, reload server for GOAWAY
- def test_03_01_h2_goaway(self, env: Env, httpd, nghttpx, repeat):
+ def test_03_01_h2_goaway(self, env: Env, httpd, nghttpx):
proto = 'h2'
count = 3
self.r = None
# download files sequentially with delay, reload server for GOAWAY
@pytest.mark.skipif(condition=not Env.have_h3(), reason="h3 not supported")
- def test_03_02_h3_goaway(self, env: Env, httpd, nghttpx, repeat):
+ def test_03_02_h3_goaway(self, env: Env, httpd, nghttpx):
proto = 'h3'
if proto == 'h3' and env.curl_uses_lib('msh3'):
pytest.skip("msh3 stalls here")
log.debug(f'request {idx} connected')
# download files sequentially with delay, reload server for GOAWAY
- def test_03_03_h1_goaway(self, env: Env, httpd, nghttpx, repeat):
+ def test_03_03_h1_goaway(self, env: Env, httpd, nghttpx):
proto = 'http/1.1'
count = 3
self.r = None
# download 1 file, check that delayed response works in general
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_04_01_download_1(self, env: Env, httpd, nghttpx, repeat,
- proto):
+ def test_04_01_download_1(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 1
# prepend 100 file requests to warm up connection processing limits
# (Apache2 increases # of parallel processed requests after successes)
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_04_02_100_100_10(self, env: Env,
- httpd, nghttpx, repeat, proto):
+ def test_04_02_100_100_10(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 50
# prepend 100 file requests to warm up connection processing limits
# (Apache2 increases # of parallel processed requests after successes)
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_04_03_1000_10_1(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_04_03_1000_10_1(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 50
# prepend 100 file requests to warm up connection processing limits
# (Apache2 increases # of parallel processed requests after successes)
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_04_04_1000_10_1(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_04_04_1000_10_1(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 50
# download 1 file, check that we get CURLE_PARTIAL_FILE
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_05_01_partial_1(self, env: Env, httpd, nghttpx, repeat,
- proto):
+ def test_05_01_partial_1(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# download files, check that we get CURLE_PARTIAL_FILE for all
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_05_02_partial_20(self, env: Env, httpd, nghttpx, repeat,
- proto):
+ def test_05_02_partial_20(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
assert len(invalid_stats) == 0, f'failed: {invalid_stats}'
# access a resource that, on h2, RST the stream with HTTP_1_1_REQUIRED
- def test_05_03_required(self, env: Env, httpd, nghttpx, repeat):
+ def test_05_03_required(self, env: Env, httpd, nghttpx):
curl = CurlClient(env=env)
proto = 'http/1.1'
urln = f'https://{env.authority_for(env.domain1, proto)}/curltest/1_1'
# - h2 to work since it will signal the end of the response before
# and not see the "unclean" close either
@pytest.mark.parametrize("proto", ['http/1.0', 'http/1.1', 'h2'])
- def test_05_04_unclean_tls_shutdown(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_05_04_unclean_tls_shutdown(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 10 if proto == 'h2' else 1
# download using only HTTP/3 on working server
@pytest.mark.skipif(condition=not Env.have_h3(), reason="missing HTTP/3 support")
- def test_06_01_h3_only(self, env: Env, httpd, nghttpx, repeat):
+ def test_06_01_h3_only(self, env: Env, httpd, nghttpx):
curl = CurlClient(env=env)
urln = f'https://{env.authority_for(env.domain1, "h3")}/data.json'
r = curl.http_download(urls=[urln], extra_args=['--http3-only'])
# download using only HTTP/3 on missing server
@pytest.mark.skipif(condition=not Env.have_h3(), reason="missing HTTP/3 support")
- def test_06_02_h3_only(self, env: Env, httpd, nghttpx, repeat):
+ def test_06_02_h3_only(self, env: Env, httpd, nghttpx):
nghttpx.stop_if_running()
curl = CurlClient(env=env)
urln = f'https://{env.authority_for(env.domain1, "h3")}/data.json'
# download using HTTP/3 on missing server with fallback on h2
@pytest.mark.skipif(condition=not Env.have_h3(), reason="missing HTTP/3 support")
- def test_06_03_h3_fallback_h2(self, env: Env, httpd, nghttpx, repeat):
+ def test_06_03_h3_fallback_h2(self, env: Env, httpd, nghttpx):
nghttpx.stop_if_running()
curl = CurlClient(env=env)
urln = f'https://{env.authority_for(env.domain1, "h3")}/data.json'
# download using HTTP/3 on missing server with fallback on http/1.1
@pytest.mark.skipif(condition=not Env.have_h3(), reason="missing HTTP/3 support")
- def test_06_04_h3_fallback_h1(self, env: Env, httpd, nghttpx, repeat):
+ def test_06_04_h3_fallback_h1(self, env: Env, httpd, nghttpx):
nghttpx.stop_if_running()
curl = CurlClient(env=env)
urln = f'https://{env.authority_for(env.domain2, "h3")}/data.json'
assert r.stats[0]['http_version'] == '1.1'
# make a successful https: transfer and observer the timer stats
- def test_06_10_stats_success(self, env: Env, httpd, nghttpx, repeat):
+ def test_06_10_stats_success(self, env: Env, httpd, nghttpx):
curl = CurlClient(env=env)
urln = f'https://{env.authority_for(env.domain1, "h2")}/data.json'
r = curl.http_download(urls=[urln])
assert r.stats[0]['time_appconnect'] > 0.0
# make https: to a hostname that tcp connects, but will not verify
- def test_06_11_stats_fail_verify(self, env: Env, httpd, nghttpx, repeat):
+ def test_06_11_stats_fail_verify(self, env: Env, httpd, nghttpx):
curl = CurlClient(env=env)
urln = f'https://not-valid.com:{env.https_port}/data.json'
r = curl.http_download(urls=[urln], extra_args=[
assert r.stats[0]['time_appconnect'] == 0 # but not SSL verified
# make https: to an invalid address
- def test_06_12_stats_fail_tcp(self, env: Env, httpd, nghttpx, repeat):
+ def test_06_12_stats_fail_tcp(self, env: Env, httpd, nghttpx):
curl = CurlClient(env=env)
urln = 'https://not-valid.com:1/data.json'
r = curl.http_download(urls=[urln], extra_args=[
# upload small data, check that this is what was echoed
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_01_upload_1_small(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_01_upload_1_small(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# upload large data, check that this is what was echoed
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_02_upload_1_large(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_02_upload_1_large(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# upload data sequentially, check that they were echoed
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_10_upload_sequential(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_10_upload_sequential(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# upload data parallel, check that they were echoed
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_07_11_upload_parallel(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_11_upload_parallel(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# upload large data sequentially, check that this is what was echoed
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_12_upload_seq_large(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_12_upload_seq_large(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# upload very large data sequentially, check that this is what was echoed
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_13_upload_seq_large(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_13_upload_seq_large(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# upload data parallel, check that they were echoed
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_07_20_upload_parallel(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_20_upload_parallel(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# upload large data parallel, check that this is what was echoed
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_07_21_upload_parallel_large(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_21_upload_parallel_large(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# upload large data parallel to a URL that denies uploads
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_07_22_upload_parallel_fail(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_22_upload_parallel_fail(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# PUT 100k
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_30_put_100k(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_30_put_100k(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# PUT 10m
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_31_put_10m(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_31_put_10m(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# issue #10591
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_32_issue_10591(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_32_issue_10591(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# issue #11157, upload that is 404'ed by server, needs to terminate
# correctly and not time out on sending
- def test_07_33_issue_11157a(self, env: Env, httpd, nghttpx, repeat):
+ def test_07_33_issue_11157a(self, env: Env, httpd, nghttpx):
proto = 'h2'
fdata = os.path.join(env.gen_dir, 'data-10m')
# send a POST to our PUT handler which will send immediately a 404 back
r.check_stats(1, 404)
# issue #11157, send upload that is slowly read in
- def test_07_33_issue_11157b(self, env: Env, httpd, nghttpx, repeat):
+ def test_07_33_issue_11157b(self, env: Env, httpd, nghttpx):
proto = 'h2'
fdata = os.path.join(env.gen_dir, 'data-10m')
# tell our test PUT handler to read the upload more slowly, so
assert r.exit_code == 0, r.dump_logs()
r.check_stats(1, 200)
- def test_07_34_issue_11194(self, env: Env, httpd, nghttpx, repeat):
+ def test_07_34_issue_11194(self, env: Env, httpd, nghttpx):
proto = 'h2'
# tell our test PUT handler to read the upload more slowly, so
# that the send buffering and transfer loop needs to wait
r.check_stats(1, 200)
# upload large data on a h1 to h2 upgrade
- def test_07_35_h1_h2_upgrade_upload(self, env: Env, httpd, nghttpx, repeat):
+ def test_07_35_h1_h2_upgrade_upload(self, env: Env, httpd, nghttpx):
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'])
- def test_07_36_upload_30x(self, env: Env, httpd, nghttpx, repeat, redir, proto):
+ def test_07_36_upload_30x(self, env: Env, httpd, nghttpx, redir, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# upload to a 307 response
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_37_upload_307(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_37_upload_307(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# POST form data, yet another code path in transfer
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_38_form_small(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_38_form_small(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# POST data urlencoded, small enough to be sent with request headers
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_39_post_urlenc_small(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_39_post_urlenc_small(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# POST data urlencoded, large enough to be sent separate from request headers
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_40_post_urlenc_large(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_40_post_urlenc_large(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# 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'])
- def test_07_41_post_urlenc_small(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_41_post_urlenc_small(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# upload data, pause, let connection die with an incomplete response
# issues #11769 #13260
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_42a_upload_disconnect(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_42a_upload_disconnect(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# upload data, pause, let connection die without any response at all
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_42b_upload_disconnect(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_42b_upload_disconnect(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# upload data, pause, let connection die after 100 continue
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_42c_upload_disconnect(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_42c_upload_disconnect(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
r.check_exit_code(exp_code) # GOT_NOTHING
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_43_upload_denied(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_43_upload_denied(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# speed limited on put handler
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_50_put_speed_limit(self, env: Env, httpd, nghttpx, proto, repeat):
+ def test_07_50_put_speed_limit(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 1
# speed limited on echo handler
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_07_51_echo_speed_limit(self, env: Env, httpd, nghttpx, proto, repeat):
+ def test_07_51_echo_speed_limit(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 1
# upload larger data, triggering "Expect: 100-continue" code paths
@pytest.mark.parametrize("proto", ['http/1.1'])
- def test_07_60_upload_exp100(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_60_upload_exp100(self, env: Env, httpd, nghttpx, proto):
fdata = os.path.join(env.gen_dir, 'data-1m+')
read_delay = 1
curl = CurlClient(env=env)
# upload larger data, triggering "Expect: 100-continue" code paths
@pytest.mark.parametrize("proto", ['http/1.1'])
- def test_07_61_upload_exp100_timeout(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_07_61_upload_exp100_timeout(self, env: Env, httpd, nghttpx, proto):
fdata = os.path.join(env.gen_dir, 'data-1m+')
read_delay = 2
curl = CurlClient(env=env)
# download 1 file
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_08_01_download_1(self, env: Env, caddy: Caddy, repeat, proto):
+ def test_08_01_download_1(self, env: Env, caddy: Caddy, proto):
if proto == 'h3' and not env.have_h3_curl():
pytest.skip("h3 not supported in curl")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# download 1MB files sequentially
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_08_02_download_1mb_sequential(self, env: Env, caddy: Caddy,
- repeat, proto):
+ def test_08_02_download_1mb_sequential(self, env: Env, caddy: Caddy, proto):
if proto == 'h3' and not env.have_h3_curl():
pytest.skip("h3 not supported in curl")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# download 1MB files parallel
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_08_03_download_1mb_parallel(self, env: Env, caddy: Caddy,
- repeat, proto):
+ def test_08_03_download_1mb_parallel(self, env: Env, caddy: Caddy, proto):
if proto == 'h3' and not env.have_h3_curl():
pytest.skip("h3 not supported in curl")
if proto == 'h3' and env.curl_uses_lib('msh3'):
@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'])
- def test_08_04a_download_10mb_sequential(self, env: Env, caddy: Caddy,
- repeat, proto):
+ def test_08_04a_download_10mb_sequential(self, env: Env, caddy: Caddy, proto):
if proto == 'h3' and not env.have_h3_curl():
pytest.skip("h3 not supported in curl")
if proto == 'h3' and env.curl_uses_lib('msh3'):
@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'])
- def test_08_04b_download_10mb_sequential(self, env: Env, caddy: Caddy,
- repeat, proto):
+ def test_08_04b_download_10mb_sequential(self, env: Env, caddy: Caddy, proto):
if proto == 'h3' and not env.have_h3_curl():
pytest.skip("h3 not supported in curl")
if proto == 'h3' and env.curl_uses_lib('msh3'):
@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.skipif(condition=Env().ci_run, reason="not suitable for CI runs")
- def test_08_05_download_1mb_parallel(self, env: Env, caddy: Caddy,
- repeat, proto):
+ def test_08_05_download_1mb_parallel(self, env: Env, caddy: Caddy, proto):
if proto == 'h3' and not env.have_h3_curl():
pytest.skip("h3 not supported in curl")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# post data parallel, check that they were echoed
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_08_06_post_parallel(self, env: Env, httpd, caddy, repeat, proto):
+ def test_08_06_post_parallel(self, env: Env, httpd, caddy, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
# put large file, check that they length were echoed
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_08_07_put_large(self, env: Env, httpd, caddy, repeat, proto):
+ def test_08_07_put_large(self, env: Env, httpd, caddy, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('msh3'):
httpd.reload()
# download a file that triggers a "103 Early Hints" response
- def test_09_01_h2_early_hints(self, env: Env, httpd, repeat):
+ def test_09_01_h2_early_hints(self, env: Env, httpd):
curl = CurlClient(env=env)
url = f'https://{env.domain1}:{env.https_port}/push/data1'
r = curl.http_download(urls=[url], alpn_proto='h2', with_stats=False,
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]}'
- def test_09_02_h2_push(self, env: Env, httpd, repeat):
+ def test_09_02_h2_push(self, env: Env, httpd):
# use localhost as we do not have resolve support in local client
url = f'https://localhost:{env.https_port}/push/data1'
client = LocalClient(name='h2-serverpush', env=env)
return None
# download via http: proxy (no tunnel)
- def test_10_01_proxy_http(self, env: Env, httpd, repeat):
+ def test_10_01_proxy_http(self, env: Env, httpd):
curl = CurlClient(env=env)
url = f'http://localhost:{env.http_port}/data.json'
r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
@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'])
- def test_10_02_proxys_down(self, env: Env, httpd, proto, repeat):
+ def test_10_02_proxys_down(self, env: Env, httpd, proto):
if proto == '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_nghttpx(),
reason="no nghttpx available")
def test_10_02_proxys_up(self, env: Env, httpd, nghttpx, proto,
- fname, fcount, repeat):
+ fname, fcount):
if proto == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
count = fcount
assert respdata == indata
# download http: via http: proxytunnel
- def test_10_03_proxytunnel_http(self, env: Env, httpd, repeat):
+ def test_10_03_proxytunnel_http(self, env: Env, httpd):
curl = CurlClient(env=env)
url = f'http://localhost:{env.http_port}/data.json'
xargs = curl.get_proxy_args(proxys=False, tunnel=True)
@pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'),
reason='curl lacks HTTPS-proxy support')
@pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
- def test_10_04_proxy_https(self, env: Env, httpd, nghttpx_fwd, repeat):
+ def test_10_04_proxy_https(self, env: Env, httpd, nghttpx_fwd):
curl = CurlClient(env=env)
url = f'http://localhost:{env.http_port}/data.json'
xargs = curl.get_proxy_args(tunnel=True)
# download https: with proto via http: proxytunnel
@pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
@pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
- def test_10_05_proxytunnel_http(self, env: Env, httpd, proto, repeat):
+ def test_10_05_proxytunnel_http(self, env: Env, httpd, proto):
curl = CurlClient(env=env)
url = f'https://localhost:{env.https_port}/data.json'
xargs = curl.get_proxy_args(proxys=False, tunnel=True)
@pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
@pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
@pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
- def test_10_06_proxytunnel_https(self, env: Env, httpd, nghttpx_fwd, proto, tunnel, repeat):
+ def test_10_06_proxytunnel_https(self, env: Env, httpd, nghttpx_fwd, proto, tunnel):
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_nghttpx(), reason="no nghttpx available")
def test_10_07_pts_down_small(self, env: Env, httpd, nghttpx_fwd, proto,
- tunnel, fname, fcount, repeat):
+ tunnel, fname, fcount):
if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
count = fcount
])
@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, repeat):
+ tunnel, fname, fcount):
if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
count = fcount
@pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
@pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
@pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
- def test_10_09_reuse_ser(self, env: Env, httpd, nghttpx_fwd, tunnel, repeat):
+ def test_10_09_reuse_ser(self, env: Env, httpd, nghttpx_fwd, tunnel):
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.parametrize("tunnel", ['http/1.1', 'h2'])
@pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
- def test_10_10_reuse_proxy(self, env: Env, httpd, nghttpx_fwd, tunnel, repeat):
+ 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.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
@pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
@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")
- def test_10_11_noreuse_proxy_https(self, env: Env, httpd, nghttpx_fwd, tunnel, repeat):
+ def test_10_11_noreuse_proxy_https(self, env: Env, httpd, nghttpx_fwd, tunnel):
# different --proxy-tls13-ciphers, no reuse of connection for https:
curl = CurlClient(env=env)
if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
@pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
@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")
- def test_10_12_noreuse_proxy_http(self, env: Env, httpd, nghttpx_fwd, tunnel, repeat):
+ 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.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
@pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
@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")
- def test_10_13_noreuse_https(self, env: Env, httpd, nghttpx_fwd, tunnel, repeat):
+ 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.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
reason='curl lacks HTTPS-proxy support')
@pytest.mark.skipif(condition=Env.curl_uses_lib('bearssl'), reason="ip address cert verification not supported")
@pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
- def test_10_14_proxys_ip_addr(self, env: Env, httpd, proto, repeat):
+ def test_10_14_proxys_ip_addr(self, env: Env, httpd, proto):
if proto == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
curl = CurlClient(env=env)
faker.stop()
# download http: via Unix socket
- def test_11_01_unix_connect_http(self, env: Env, httpd, uds_faker, repeat):
+ def test_11_01_unix_connect_http(self, env: Env, httpd, uds_faker):
curl = CurlClient(env=env)
url = f'http://{env.domain1}:{env.http_port}/data.json'
r = curl.http_download(urls=[url], with_stats=True,
# download https: via Unix socket
@pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
- def test_11_02_unix_connect_http(self, env: Env, httpd, uds_faker, repeat):
+ def test_11_02_unix_connect_http(self, env: Env, httpd, uds_faker):
curl = CurlClient(env=env)
url = f'https://{env.domain1}:{env.https_port}/data.json'
r = curl.http_download(urls=[url], with_stats=True,
# download HTTP/3 via Unix socket
@pytest.mark.skipif(condition=not Env.have_h3(), reason='h3 not supported')
- def test_11_03_unix_connect_quic(self, env: Env, httpd, uds_faker, repeat):
+ def test_11_03_unix_connect_quic(self, env: Env, httpd, uds_faker):
curl = CurlClient(env=env)
url = f'https://{env.domain1}:{env.https_port}/data.json'
r = curl.http_download(urls=[url], with_stats=True,
# check if HTTP/1.1 handles 'Connection: close' correctly
@pytest.mark.parametrize("proto", ['http/1.1'])
- def test_12_01_h1_conn_close(self, env: Env,
- httpd, nghttpx, repeat, proto):
+ def test_12_01_h1_conn_close(self, env: Env, httpd, nghttpx, proto):
httpd.clear_extra_configs()
httpd.set_extra_config('base', [
'MaxKeepAliveRequests 1',
@pytest.mark.skipif(condition=Env.httpd_is_at_least('2.5.0'),
reason="httpd 2.5+ handles KeepAlives different")
@pytest.mark.parametrize("proto", ['http/1.1'])
- def test_12_02_h1_conn_timeout(self, env: Env,
- httpd, nghttpx, repeat, proto):
+ def test_12_02_h1_conn_timeout(self, env: Env, httpd, nghttpx, proto):
httpd.clear_extra_configs()
httpd.set_extra_config('base', [
'KeepAliveTimeout 1',
return None
# download via http: proxy (no tunnel), no auth
- def test_13_01_proxy_no_auth(self, env: Env, httpd, repeat):
+ def test_13_01_proxy_no_auth(self, env: Env, httpd):
curl = CurlClient(env=env)
url = f'http://localhost:{env.http_port}/data.json'
r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
r.check_response(count=1, http_status=407)
# download via http: proxy (no tunnel), auth
- def test_13_02_proxy_auth(self, env: Env, httpd, repeat):
+ def test_13_02_proxy_auth(self, env: Env, httpd):
curl = CurlClient(env=env)
url = f'http://localhost:{env.http_port}/data.json'
xargs = curl.get_proxy_args(proxys=False)
@pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'),
reason='curl lacks HTTPS-proxy support')
@pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
- def test_13_03_proxys_no_auth(self, env: Env, httpd, nghttpx_fwd, repeat):
+ def test_13_03_proxys_no_auth(self, env: Env, httpd, nghttpx_fwd):
curl = CurlClient(env=env)
url = f'http://localhost:{env.http_port}/data.json'
xargs = curl.get_proxy_args(proxys=True)
@pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'),
reason='curl lacks HTTPS-proxy support')
@pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
- def test_13_04_proxys_auth(self, env: Env, httpd, nghttpx_fwd, repeat):
+ def test_13_04_proxys_auth(self, env: Env, httpd, nghttpx_fwd):
curl = CurlClient(env=env)
url = f'http://localhost:{env.http_port}/data.json'
xargs = curl.get_proxy_args(proxys=True)
extra_args=xargs)
r.check_response(count=1, http_status=200)
- def test_13_05_tunnel_http_no_auth(self, env: Env, httpd, repeat):
+ def test_13_05_tunnel_http_no_auth(self, env: Env, httpd):
curl = CurlClient(env=env)
url = f'http://localhost:{env.http_port}/data.json'
xargs = curl.get_proxy_args(proxys=False, tunnel=True)
# expect "COULD_NOT_CONNECT"
r.check_response(exitcode=56, http_status=None)
- def test_13_06_tunnel_http_auth(self, env: Env, httpd, repeat):
+ def test_13_06_tunnel_http_auth(self, env: Env, httpd):
curl = CurlClient(env=env)
url = f'http://localhost:{env.http_port}/data.json'
xargs = curl.get_proxy_args(proxys=False, tunnel=True)
reason='curl lacks HTTPS-proxy support')
@pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
@pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
- def test_13_07_tunnels_no_auth(self, env: Env, httpd, proto, tunnel, repeat):
+ def test_13_07_tunnels_no_auth(self, env: Env, httpd, proto, tunnel):
if tunnel == '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.parametrize("proto", ['http/1.1', 'h2'])
@pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
- def test_13_08_tunnels_auth(self, env: Env, httpd, proto, tunnel, repeat):
+ def test_13_08_tunnels_auth(self, env: Env, httpd, proto, tunnel):
if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
pytest.skip('only supported with nghttp2')
curl = CurlClient(env=env)
# download 1 file, not authenticated
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_14_01_digest_get_noauth(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_14_01_digest_get_noauth(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
curl = CurlClient(env=env)
# download 1 file, authenticated
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_14_02_digest_get_auth(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_14_02_digest_get_auth(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
curl = CurlClient(env=env)
# PUT data, authenticated
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_14_03_digest_put_auth(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_14_03_digest_put_auth(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
data='0123456789'
# PUT data, digest auth large pw
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_14_04_digest_large_pw(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_14_04_digest_large_pw(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
data='0123456789'
# PUT data, basic auth large pw
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_14_05_basic_large_pw(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_14_05_basic_large_pw(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and not env.curl_uses_lib('ngtcp2'):
# PUT data, basic auth with very large pw
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_14_06_basic_very_large_pw(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_14_06_basic_very_large_pw(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if proto == 'h3' and env.curl_uses_lib('quiche'):
# download plain file
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_16_01_info_download(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_16_01_info_download(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 2
# download plain file with a 302 redirect
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_16_02_info_302_download(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_16_02_info_302_download(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 2
self.check_stat(idx, s, r, dl_size=30, ul_size=0)
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_16_03_info_upload(self, env: Env, httpd, nghttpx, proto, repeat):
+ def test_16_03_info_upload(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 2
# download plain file via http: ('time_appconnect' is 0)
@pytest.mark.parametrize("proto", ['http/1.1'])
- def test_16_04_info_http_download(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_16_04_info_http_download(self, env: Env, httpd, nghttpx, proto):
count = 2
curl = CurlClient(env=env)
url = f'http://{env.domain1}:{env.http_port}/data.json?[0-{count-1}]'
if 'httpd' not in request.node._fixtureinfo.argnames:
httpd.reload_if_config_changed()
- def test_17_01_sslinfo_plain(self, env: Env, nghttpx, repeat):
+ def test_17_01_sslinfo_plain(self, env: Env, nghttpx):
proto = 'http/1.1'
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/curltest/sslinfo'
# download 1 file
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
- def test_18_01_delete(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_18_01_delete(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
count = 1
# - HEADER frame with 204 and eos=0
# - 10ms later DATA frame length=0 and eos=1
# should be accepted
- def test_18_02_delete_h2_special(self, env: Env, httpd, nghttpx, repeat):
+ def test_18_02_delete_h2_special(self, env: Env, httpd, nghttpx):
proto = 'h2'
count = 1
curl = CurlClient(env=env)
# check with `tcpdump` that we see curl TCP RST packets
@pytest.mark.skipif(condition=not Env.tcpdump(), reason="tcpdump not available")
@pytest.mark.parametrize("proto", ['http/1.1'])
- def test_19_01_check_tcp_rst(self, env: Env, httpd, repeat, proto):
+ def test_19_01_check_tcp_rst(self, env: Env, httpd, proto):
if env.ci_run:
pytest.skip("seems not to work in CI")
curl = CurlClient(env=env)
# 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'])
- def test_19_02_check_shutdown(self, env: Env, httpd, repeat, proto):
+ def test_19_02_check_shutdown(self, env: Env, httpd, proto):
if not env.curl_is_debug():
pytest.skip('only works for curl debug builds')
curl = CurlClient(env=env, run_env={
# run downloads where the server closes the connection after each request
@pytest.mark.parametrize("proto", ['http/1.1'])
- def test_19_03_shutdown_by_server(self, env: Env, httpd, repeat, proto):
+ def test_19_03_shutdown_by_server(self, env: Env, httpd, proto):
if not env.curl_is_debug():
pytest.skip('only works for curl debug builds')
count = 10
# run downloads with CURLOPT_FORBID_REUSE set, meaning *we* close
# the connection after each request
@pytest.mark.parametrize("proto", ['http/1.1'])
- def test_19_04_shutdown_by_curl(self, env: Env, httpd, proto, repeat):
+ def test_19_04_shutdown_by_curl(self, env: Env, httpd, proto):
if not env.curl_is_debug():
pytest.skip('only works for curl debug builds')
count = 10
# run event-based downloads with CURLOPT_FORBID_REUSE set, meaning *we* close
# the connection after each request
@pytest.mark.parametrize("proto", ['http/1.1'])
- def test_19_05_event_shutdown_by_server(self, env: Env, httpd, proto, repeat):
+ def test_19_05_event_shutdown_by_server(self, env: Env, httpd, proto):
if not env.curl_is_debug():
pytest.skip('only works for curl debug builds')
count = 10
# check graceful shutdown on multiplexed http
@pytest.mark.parametrize("proto", ['h2', 'h3'])
- def test_19_06_check_shutdown(self, env: Env, httpd, nghttpx, repeat, proto):
+ def test_19_06_check_shutdown(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.have_h3():
pytest.skip("h3 not supported")
if not env.curl_is_debug():
yield
p.terminate()
- def test_20_01_basic(self, env: Env, ws_echo, repeat):
+ def test_20_01_basic(self, env: Env, ws_echo):
curl = CurlClient(env=env)
url = f'http://localhost:{env.ws_port}/'
r = curl.http_download(urls=[url])
r.check_response(http_status=426)
- def test_20_02_pingpong_small(self, env: Env, ws_echo, repeat):
+ def test_20_02_pingpong_small(self, env: Env, ws_echo):
payload = 125 * "x"
client = LocalClient(env=env, name='ws-pingpong')
if not client.exists():
r.check_exit_code(0)
# the python websocket server does not like 'large' control frames
- def test_20_03_pingpong_too_large(self, env: Env, ws_echo, repeat):
+ def test_20_03_pingpong_too_large(self, env: Env, ws_echo):
payload = 127 * "x"
client = LocalClient(env=env, name='ws-pingpong')
if not client.exists():
r = client.run(args=[url, payload])
r.check_exit_code(56)
- def test_20_04_data_small(self, env: Env, ws_echo, repeat):
+ def test_20_04_data_small(self, env: Env, ws_echo):
client = LocalClient(env=env, name='ws-data')
if not client.exists():
pytest.skip(f'example client not built: {client.name}')
r = client.run(args=['-m', str(0), '-M', str(10), url])
r.check_exit_code(0)
- def test_20_05_data_med(self, env: Env, ws_echo, repeat):
+ def test_20_05_data_med(self, env: Env, ws_echo):
client = LocalClient(env=env, name='ws-data')
if not client.exists():
pytest.skip(f'example client not built: {client.name}')
r = client.run(args=['-m', str(120), '-M', str(130), url])
r.check_exit_code(0)
- def test_20_06_data_large(self, env: Env, ws_echo, repeat):
+ def test_20_06_data_large(self, env: Env, ws_echo):
client = LocalClient(env=env, name='ws-data')
if not client.exists():
pytest.skip(f'example client not built: {client.name}')
r = client.run(args=['-m', str(65535 - 5), '-M', str(65535 + 5), url])
r.check_exit_code(0)
- def test_20_07_data_large_small_recv(self, env: Env, ws_echo, repeat):
+ def test_20_07_data_large_small_recv(self, env: Env, ws_echo):
run_env = os.environ.copy()
run_env['CURL_WS_CHUNK_SIZE'] = '1024'
client = LocalClient(env=env, name='ws-data', run_env=run_env)
# Send large frames and simulate send blocking on 8192 bytes chunks
# Simlates error reported in #15865
- def test_20_08_data_very_large(self, env: Env, ws_echo, repeat):
+ def test_20_08_data_very_large(self, env: Env, ws_echo):
run_env = os.environ.copy()
run_env['CURL_WS_CHUNK_EAGAIN'] = '8192'
client = LocalClient(env=env, name='ws-data', run_env=run_env)
env.make_data_file(indir=env.gen_dir, fname="upload-100k", fsize=100*1024)
env.make_data_file(indir=env.gen_dir, fname="upload-1m", fsize=1024*1024)
- def test_30_01_list_dir(self, env: Env, vsftpd: VsFTPD, repeat):
+ def test_30_01_list_dir(self, env: Env, vsftpd: VsFTPD):
curl = CurlClient(env=env)
url = f'ftp://{env.ftp_domain}:{vsftpd.port}/'
r = curl.ftp_get(urls=[url], with_stats=True)
@pytest.mark.parametrize("docname", [
'data-1k', 'data-1m', 'data-10m'
])
- def test_30_02_download_1(self, env: Env, vsftpd: VsFTPD, docname, repeat):
+ def test_30_02_download_1(self, env: Env, vsftpd: VsFTPD, docname):
curl = CurlClient(env=env)
srcfile = os.path.join(vsftpd.docs_dir, f'{docname}')
count = 1
@pytest.mark.parametrize("docname", [
'data-1k', 'data-1m', 'data-10m'
])
- def test_30_03_download_10_serial(self, env: Env, vsftpd: VsFTPD, docname, repeat):
+ def test_30_03_download_10_serial(self, env: Env, vsftpd: VsFTPD, docname):
curl = CurlClient(env=env)
srcfile = os.path.join(vsftpd.docs_dir, f'{docname}')
count = 10
@pytest.mark.parametrize("docname", [
'data-1k', 'data-1m', 'data-10m'
])
- def test_30_04_download_10_parallel(self, env: Env, vsftpd: VsFTPD, docname, repeat):
+ def test_30_04_download_10_parallel(self, env: Env, vsftpd: VsFTPD, docname):
curl = CurlClient(env=env)
srcfile = os.path.join(vsftpd.docs_dir, f'{docname}')
count = 10
@pytest.mark.parametrize("docname", [
'upload-1k', 'upload-100k', 'upload-1m'
])
- def test_30_05_upload_1(self, env: Env, vsftpd: VsFTPD, docname, repeat):
+ def test_30_05_upload_1(self, env: Env, vsftpd: VsFTPD, docname):
curl = CurlClient(env=env)
srcfile = os.path.join(env.gen_dir, docname)
dstfile = os.path.join(vsftpd.docs_dir, docname)
# check with `tcpdump` if curl causes any TCP RST packets
@pytest.mark.skipif(condition=not Env.tcpdump(), reason="tcpdump not available")
- def test_30_06_shutdownh_download(self, env: Env, vsftpd: VsFTPD, repeat):
+ def test_30_06_shutdownh_download(self, env: Env, vsftpd: VsFTPD):
docname = 'data-1k'
curl = CurlClient(env=env)
count = 1
# check with `tcpdump` if curl causes any TCP RST packets
@pytest.mark.skipif(condition=not Env.tcpdump(), reason="tcpdump not available")
- def test_30_07_shutdownh_upload(self, env: Env, vsftpd: VsFTPD, repeat):
+ def test_30_07_shutdownh_upload(self, env: Env, vsftpd: VsFTPD):
docname = 'upload-1k'
curl = CurlClient(env=env)
srcfile = os.path.join(env.gen_dir, docname)
env.make_data_file(indir=env.gen_dir, fname="upload-100k", fsize=100*1024)
env.make_data_file(indir=env.gen_dir, fname="upload-1m", fsize=1024*1024)
- def test_31_01_list_dir(self, env: Env, vsftpds: VsFTPD, repeat):
+ def test_31_01_list_dir(self, env: Env, vsftpds: VsFTPD):
curl = CurlClient(env=env)
url = f'ftp://{env.ftp_domain}:{vsftpds.port}/'
r = curl.ftp_ssl_get(urls=[url], with_stats=True)
@pytest.mark.parametrize("docname", [
'data-1k', 'data-1m', 'data-10m'
])
- def test_31_02_download_1(self, env: Env, vsftpds: VsFTPD, docname, repeat):
+ def test_31_02_download_1(self, env: Env, vsftpds: VsFTPD, docname):
curl = CurlClient(env=env)
srcfile = os.path.join(vsftpds.docs_dir, f'{docname}')
count = 1
@pytest.mark.parametrize("docname", [
'data-1k', 'data-1m', 'data-10m'
])
- def test_31_03_download_10_serial(self, env: Env, vsftpds: VsFTPD, docname, repeat):
+ def test_31_03_download_10_serial(self, env: Env, vsftpds: VsFTPD, docname):
curl = CurlClient(env=env)
srcfile = os.path.join(vsftpds.docs_dir, f'{docname}')
count = 10
@pytest.mark.parametrize("docname", [
'data-1k', 'data-1m', 'data-10m'
])
- def test_31_04_download_10_parallel(self, env: Env, vsftpds: VsFTPD, docname, repeat):
+ def test_31_04_download_10_parallel(self, env: Env, vsftpds: VsFTPD, docname):
curl = CurlClient(env=env)
srcfile = os.path.join(vsftpds.docs_dir, f'{docname}')
count = 10
@pytest.mark.parametrize("docname", [
'upload-1k', 'upload-100k', 'upload-1m'
])
- def test_31_05_upload_1(self, env: Env, vsftpds: VsFTPD, docname, repeat):
+ def test_31_05_upload_1(self, env: Env, vsftpds: VsFTPD, docname):
curl = CurlClient(env=env)
srcfile = os.path.join(env.gen_dir, docname)
dstfile = os.path.join(vsftpds.docs_dir, docname)
# check with `tcpdump` if curl causes any TCP RST packets
@pytest.mark.skipif(condition=not Env.tcpdump(), reason="tcpdump not available")
- def test_31_06_shutdownh_download(self, env: Env, vsftpds: VsFTPD, repeat):
+ def test_31_06_shutdownh_download(self, env: Env, vsftpds: VsFTPD):
docname = 'data-1k'
curl = CurlClient(env=env)
count = 1
# check with `tcpdump` if curl causes any TCP RST packets
@pytest.mark.skipif(condition=not Env.tcpdump(), reason="tcpdump not available")
- def test_31_07_shutdownh_upload(self, env: Env, vsftpds: VsFTPD, repeat):
+ def test_31_07_shutdownh_upload(self, env: Env, vsftpds: VsFTPD):
docname = 'upload-1k'
curl = CurlClient(env=env)
srcfile = os.path.join(env.gen_dir, docname)
projects / thirdparty / curl.git / commitdiff
