romenskiy2012/curl
1
0
Fork 0
mirror of https://github.com/curl/curl.git synced 2026-07-29 00:03:12 +03:00
Code Issues Projects Releases Packages Wiki Activity

tests: change whitespace and comments in Python test code

Browse source 
* remove an unneeded ruff warning disable
* remove coding: utf-8 from Python code; PEP 3120 makes UTF-8 the
  default encoding
* remove unusable shebang lines from Python code
* remove empty print strings
* disable warnings when file objects are stored; these instances can't
  be handled with context managers
* use more consistent whitespace in Python code, fixing flake8 warnings
* set the executable bit on scorecard.py, making it easier to run

These fix ruff rules EXE001, FURB105, UP009, SIM115.
This commit is contained in:
Dan Fandrich 2026-07-25 13:35:51 -07:00
parent c39193a589
commit 4eb691e89c
51 changed files with 105 additions and 199 deletions
Download patch file Download diff file Expand all files Collapse all files

1
tests/dictserver.py
View file

@ -1,5 +1,4 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

9
tests/ech_combos.py
View file

@ -1,5 +1,4 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -45,12 +44,12 @@ def CombinationRepetitionUtil(chosen, arr, badarr, index,
if chosen[j] in badarr:
res = 0
j = j - 1
print("cli_test $turl 1", res, end = " ")
print("cli_test $turl 1", res, end=" ")
# print combination but eliminating any runs of
# two identical params
for j in range(r):
if j != 0 and chosen[j] != chosen[j-1]:
print(chosen[j], end = " ")
print(chosen[j], end=" ")
print()
return
@ -89,8 +88,8 @@ def CombinationRepetition(arr, badarr, n, r):
# Driver code
badarr = [ '--ech grease', '--ech false', '--ech ecl:$badecl', '--ech pn:$badpn' ]
goodarr = [ '--ech hard', '--ech true', '--ech ecl:$goodecl', '--ech pn:$goodpn' ]
badarr = ['--ech grease', '--ech false', '--ech ecl:$badecl', '--ech pn:$badpn']
goodarr = ['--ech hard', '--ech true', '--ech ecl:$goodecl', '--ech pn:$goodpn']
arr = badarr + goodarr
r = 8
n = len(arr) - 1

23
tests/http/scorecard.py Normal file → Executable file
View file

@ -1,5 +1,4 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -211,7 +210,7 @@ class Card:
print(f' {col:>{colw[idx]}} {"[cpu/rss]":<{statw}}', end='')
else:
print(f' {col:>{colw[idx]}}', end='')
print('')
print()
for row in rows:
for idx, cell in enumerate(row):
print(f' {cell["sval"]:>{colw[idx]}}', end='')
@ -224,7 +223,7 @@ class Card:
print(f' {s:<{statw}}', end='')
if 'errors' in cell:
errors.extend(cell['errors'])
print('')
print()
if len(errors):
print(f'Errors: {errors}')
@ -647,13 +646,15 @@ class ScoreRunner:
rows = []
mparallel = meta['request_parallels']
cols.extend([f'{mp} max' for mp in mparallel])
row = [{
'val': fsize,
'sval': Card.fmt_size(fsize)
},{
'val': count,
'sval': f'{count}',
}]
row = [
{
'val': fsize,
'sval': Card.fmt_size(fsize)
}, {
'val': count,
'sval': f'{count}',
}
]
self.info('requests, max parallel...')
row.extend([self.do_requests(url=url, count=count,
max_parallel=mp, nsamples=meta["samples"])
@ -992,7 +993,7 @@ def main():
parser.add_argument("--remote", action='store', type=str,
default=None, help="score against the remote server at <ip>:<port>")
parser.add_argument("--flame", action='store_true',
default = False, help="produce a flame graph on curl")
default=False, help="produce a flame graph on curl")
parser.add_argument("--limit-rate", action='store', type=str,
default=None, help="use curl's --limit-rate")
parser.add_argument("--http-plain", action='store_true',

2
tests/http/test_01_basic.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

10
tests/http/test_02_download.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -141,8 +139,8 @@ class TestDownload:
urln = f'https://{env.authority_for(env.domain1, proto)}/data.json?[0-{count-1}]'
r = curl.http_download(urls=[urln], alpn_proto=proto,
with_stats=True, extra_args=[
'--parallel', '--parallel-max', '200'
])
'--parallel', '--parallel-max', '200'
])
r.check_response(http_status=200, count=count)
# should have used at most 2 connections only (test servers allow 100 req/conn)
# it may be 1 on slow systems where request are answered faster than
@ -157,8 +155,8 @@ class TestDownload:
urln = f'https://{env.authority_for(env.domain1, proto)}/data.json?[0-{count-1}]'
r = curl.http_download(urls=[urln], alpn_proto=proto,
with_stats=True, extra_args=[
'--parallel'
])
'--parallel'
])
r.check_response(count=count, http_status=200)
# http/1.1 should have used count connections
assert r.total_connects == count, "http/1.1 should use this many connections"

2
tests/http/test_03_goaway.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

2
tests/http/test_04_stuttered.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

2
tests/http/test_05_errors.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

4
tests/http/test_06_eyeballs.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -129,7 +127,7 @@ class TestEyeballs:
assert r.stats[0]['time_connect'] == 0 # no one connected
# check that we indeed started attempts on all 3 addresses
tcp_attempts = [line for line in r.trace_lines
if re.match(r'.*Trying \[100::[123]]:443', line)]
if re.match(r'.*Trying \[100::[123]]:443', line)]
assert len(tcp_attempts) == 3, f'fond: {"".join(tcp_attempts)}\n{r.dump_logs()}'
# if the 0100::/64 really goes into the void, we should see 2 HAPPY_EYEBALLS
# timeouts being set here

12
tests/http/test_07_upload.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -549,7 +547,7 @@ class TestUpload:
])
r.check_exit_code(0)
results = [int(m.group(1)) for line in r.trace_lines
if (m := re.match(r'.* FINISHED, result=(\d+), response=(\d+)', line))]
if (m := re.match(r'.* FINISHED, result=(\d+), response=(\d+)', line))]
httpcodes = [int(m.group(2)) for line in r.trace_lines
if (m := re.match(r'.* FINISHED, result=(\d+), response=(\d+)', line))]
if httpcode == 308:
@ -568,8 +566,8 @@ class TestUpload:
url = f'https://{env.authority_for(env.domain1, proto)}/curltest/put?id=[0-0]'
r = curl.http_put(urls=[url], fdata=fdata, alpn_proto=proto,
with_headers=True, extra_args=[
'--limit-rate', f'{speed_limit}'
])
'--limit-rate', f'{speed_limit}'
])
r.check_response(count=count, http_status=200)
assert r.responses[0]['header']['received-length'] == f'{up_len}', f'{r.responses[0]}'
up_speed = r.stats[0]['speed_upload']
@ -585,8 +583,8 @@ class TestUpload:
url = f'https://{env.authority_for(env.domain1, proto)}/curltest/echo?id=[0-0]'
r = curl.http_upload(urls=[url], data=f'@{fdata}', alpn_proto=proto,
with_headers=True, extra_args=[
'--limit-rate', f'{speed_limit}'
])
'--limit-rate', f'{speed_limit}'
])
r.check_response(count=count, http_status=200)
up_speed = r.stats[0]['speed_upload']
assert up_speed <= (speed_limit * 1.1), f'{r.stats[0]}'

2
tests/http/test_08_caddy.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

2
tests/http/test_09_push.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

2
tests/http/test_10_proxy.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

2
tests/http/test_11_unix.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

2
tests/http/test_12_reuse.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

6
tests/http/test_13_proxy_auth.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -175,9 +173,9 @@ class TestProxyAuth:
url2 = f'http://localhost:{env.http_port}/data.json?2'
url3 = f'http://localhost:{env.http_port}/data.json?3'
xargs1 = curl.get_proxy_args(proxys=False, tunnel=True)
xargs1.extend(['--proxy-user', 'proxy:proxy']) # good auth
xargs1.extend(['--proxy-user', 'proxy:proxy']) # good auth
xargs2 = curl.get_proxy_args(proxys=False, tunnel=True)
xargs2.extend(['--proxy-user', 'ungood:ungood']) # bad auth
xargs2.extend(['--proxy-user', 'ungood:ungood']) # bad auth
xargs3 = curl.get_proxy_args(proxys=False, tunnel=True)
# no auth
r = curl.http_download(urls=[url1, url2, url3], alpn_proto='http/1.1', with_stats=True,

6
tests/http/test_14_auth.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -64,7 +62,7 @@ class TestAuth:
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")
data='0123456789'
data = '0123456789'
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/restricted/digest/data.json'
r = curl.http_upload(urls=[url], data=data, alpn_proto=proto, extra_args=[
@ -77,7 +75,7 @@ class TestAuth:
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")
data='0123456789'
data = '0123456789'
password = 'x' * 65535
curl = CurlClient(env=env)
url = f'https://{env.authority_for(env.domain1, proto)}/restricted/digest/data.json'

10
tests/http/test_15_tracing.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -55,7 +53,7 @@ class TestTracing:
'-v', '--trace-config', 'ids'
])
r.check_response(http_status=200)
for line in r.trace_lines:
for line in r.trace_lines:
m = re.match(r'^\[0-[0x]] .+', line)
if m is None:
assert False, f'no match: {line}'
@ -68,7 +66,7 @@ class TestTracing:
'-v', '--trace-config', 'ids,time'
])
r.check_response(http_status=200)
for line in r.trace_lines:
for line in r.trace_lines:
m = re.match(r'^([0-9:.]+) \[0-[0x]] .+', line)
if m is None:
assert False, f'no match: {line}'
@ -84,7 +82,7 @@ class TestTracing:
])
r.check_response(http_status=200)
found_tcp = False
for line in r.trace_lines:
for line in r.trace_lines:
m = re.match(r'^([0-9:.]+) \[0-[0x]] .+', line)
if m is None:
assert False, f'no match: {line}'
@ -102,7 +100,7 @@ class TestTracing:
])
r.check_response(http_status=200)
found_tcp = False
for line in r.trace_lines:
for line in r.trace_lines:
m = re.match(r'^\[0-[0x]] .+', line)
if m is None:
assert False, f'no match: {line}'

2
tests/http/test_16_info.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

6
tests/http/test_17_ssl_use.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -222,7 +220,7 @@ class TestSSLUse:
['AES256ish', ['ECDHE-ECDSA-AES256-GCM-SHA384', 'ECDHE-RSA-AES256-GCM-SHA384'], False],
['CHACHA20ish', ['ECDHE-ECDSA-CHACHA20-POLY1305', 'ECDHE-RSA-CHACHA20-POLY1305'], True],
['AES256ish+CHACHA20ish', ['ECDHE-ECDSA-AES256-GCM-SHA384', 'ECDHE-RSA-AES256-GCM-SHA384',
'ECDHE-ECDSA-CHACHA20-POLY1305', 'ECDHE-RSA-CHACHA20-POLY1305'], True],
'ECDHE-ECDSA-CHACHA20-POLY1305', 'ECDHE-RSA-CHACHA20-POLY1305'], True],
]
ret = []
for tls_id, tls_proto in {
@ -521,7 +519,7 @@ class TestSSLUse:
pytest.param("-MAC-ALL:+AEAD", "TLSv1.3", ['TLS_CHACHA20_POLY1305_SHA256'], True, id='TLSv1.3-MAC-only-AEAD'),
pytest.param("-GROUP-ALL:+GROUP-X25519", "TLSv1.3", ['TLS_CHACHA20_POLY1305_SHA256'], True, id='TLSv1.3-group-only-X25519'),
pytest.param("-GROUP-ALL:+GROUP-SECP192R1", "", [], False, id='group-only-SECP192R1'),
])
])
def test_17_18_gnutls_priority(self, env: Env, httpd, configures_httpd, priority, tls_proto, ciphers, success):
# to test setting cipher suites, the AES 256 ciphers are disabled in the test server
httpd.set_extra_config('base', [

2
tests/http/test_18_methods.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

2
tests/http/test_19_shutdown.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

6
tests/http/test_20_websockets.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -95,7 +93,7 @@ class WsServer:
self.wsproc = None
return False
self.cerr = open(self.err_file, 'w')
self.cerr = open(self.err_file, 'w') # noqa: SIM115
port_spec = {
self.name: socket.SOCK_STREAM
}
@ -248,7 +246,7 @@ class TestWebsockets:
r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
extra_args=xargs)
# The CONNECT through the proxy fails as it does not allow it
r.check_exit_code(7) # CURLE_COULDNT_CONNECT
r.check_exit_code(7) # CURLE_COULDNT_CONNECT
assert r.stats[0]['http_connect'] == 403, f'{r}'
def test_20_11_crazy_pings(self, env: Env):

16
tests/http/test_21_resolve.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -70,7 +68,7 @@ class TestResolve:
run_env = os.environ.copy()
run_env['CURL_DBG_RESOLV_FAIL_DELAY'] = f'{delay_ms}'
curl = CurlClient(env=env, run_env=run_env, force_resolv=False)
urls = [ f'https://test-{i}.http.curl.invalid/' for i in range(count)]
urls = [f'https://test-{i}.http.curl.invalid/' for i in range(count)]
r = curl.http_download(urls=urls, with_stats=True)
r.check_exit_code(6)
r.check_stats(count=count, http_status=0, exitcode=6)
@ -82,7 +80,7 @@ class TestResolve:
run_env = os.environ.copy()
run_env['CURL_DBG_RESOLV_FAIL_DELAY'] = f'{delay_ms}'
curl = CurlClient(env=env, run_env=run_env, force_resolv=False)
urls = [ f'https://test-{i}.http.curl.invalid/' for i in range(count)]
urls = [f'https://test-{i}.http.curl.invalid/' for i in range(count)]
r = curl.http_download(urls=urls, with_stats=True, extra_args=[
'--parallel'
])
@ -118,7 +116,7 @@ class TestResolve:
run_env['CURL_DBG_RESOLV_FAIL_DELAY'] = f'{delay_ms}'
run_env['CURL_DBG_RESOLV_MAX_THREADS'] = '1'
curl = CurlClient(env=env, run_env=run_env, force_resolv=False)
urls = [ f'https://test-{i}.http.curl.invalid/' for i in range(count)]
urls = [f'https://test-{i}.http.curl.invalid/' for i in range(count)]
r = curl.http_download(urls=urls, with_stats=True, extra_args=[
'--parallel', '-6'
])
@ -301,11 +299,13 @@ class TestResolve:
])
# should fail with CURLE_OPERATION_TIMEOUT or COULDNT_CONNECT
assert r.exit_code in (7, 28), f'{r.dump_logs()}'
af_unspec_resolves = [line for line in r.trace_lines if
re.match(r'.* \[DNS] re-queueing query .+ for AF_UNSPEC resolve', line)]
af_unspec_resolves = [
line for line in r.trace_lines
if re.match(r'.* \[DNS] re-queueing query .+ for AF_UNSPEC resolve', line)
]
assert len(af_unspec_resolves) == 1, f'{r.dump_logs()}'
aaaa_resolves = [line for line in r.trace_lines if
re.match(r'.* \* IPv6: fe80::1', line)]
re.match(r'.* \* IPv6: fe80::1', line)]
assert len(aaaa_resolves) == 1, f'{r.dump_logs()}'
def _clean_files(self, files):

2
tests/http/test_22_httpsrr.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

2
tests/http/test_30_vsftpd.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

6
tests/http/test_31_vsftpds.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -109,7 +107,7 @@ class TestVsFTPD:
['data-1k', 10, True],
['data-1m', 5, True],
['data-1m', 5, False],
['data-10m', 2,True]
['data-10m', 2, True]
])
def test_31_03_download_10_serial(self, env: Env, vsftpds: VsFTPD, docname, count, secure):
curl = CurlClient(env=env)
@ -128,7 +126,7 @@ class TestVsFTPD:
['data-1k', 10, True],
['data-1m', 5, True],
['data-1m', 5, False],
['data-10m', 2,True]
['data-10m', 2, True]
])
def test_31_04_download_10_parallel(self, env: Env, vsftpds: VsFTPD, docname, count, secure):
curl = CurlClient(env=env)

6
tests/http/test_32_ftps_vsftpd.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -109,7 +107,7 @@ class TestFtpsVsFTPD:
['data-1k', 10, True],
['data-1m', 5, True],
['data-1m', 5, False],
['data-10m', 2,True]
['data-10m', 2, True]
])
def test_32_03_download_10_serial(self, env: Env, vsftpds: VsFTPD, docname, count, secure):
curl = CurlClient(env=env)
@ -131,7 +129,7 @@ class TestFtpsVsFTPD:
docname = 'data-1k'
count = 2
url1 = f'ftps://{env.ftp_domain}:{vsftpds.port}/{docname}'
url2 = f'ftp://{env.ftp_domain}:{vsftpds.port}/{docname}'
url2 = f'ftp://{env.ftp_domain}:{vsftpds.port}/{docname}'
r = curl.ftp_get(urls=[url1, url2], with_stats=True)
r.check_stats(count=count, http_status=226)
assert r.total_connects == count + 1, 'should reuse the control conn'

2
tests/http/test_40_socks.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

2
tests/http/test_50_scp.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

2
tests/http/test_51_sftp.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

2
tests/http/test_60_h3_proxy.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# ***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

4
tests/http/testenv/__init__.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -23,7 +21,7 @@
# SPDX-License-Identifier: curl
#
###########################################################################
# ruff: noqa: F401, E402
# ruff: noqa: F401
import pytest
pytest.register_assert_rewrite("testenv.env", "testenv.curl", "testenv.caddy",

4
tests/http/testenv/caddy.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -113,7 +111,7 @@ class Caddy:
args = [
self._caddy, 'run'
]
self._error_fd = open(self._error_log, 'a')
self._error_fd = open(self._error_log, 'a') # noqa: SIM115
self._process = subprocess.Popen(args=args, cwd=self._caddy_dir, stderr=self._error_fd)
if self._process.returncode is not None:
return False

2
tests/http/testenv/certs.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

4
tests/http/testenv/client.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -41,7 +39,7 @@ class LocalClient:
def __init__(self, name: str, env: Env, run_dir: Optional[str] = None,
timeout: Optional[float] = None,
run_env: Optional[Dict[str,str]] = None):
run_env: Optional[Dict[str, str]] = None):
self.name = name
self.path = os.path.join(env.build_dir, 'tests/libtest/libtests')
self.env = env

61
tests/http/testenv/curl.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -74,7 +72,7 @@ class RunProfile:
return self._duration
@property
def stats(self) -> Optional[Dict[str,Any]]:
def stats(self) -> Optional[Dict[str, Any]]:
return self._stats
def sample(self):
@ -134,9 +132,8 @@ class PerfProfile:
self._proc.terminate()
self._rc = self._proc.returncode
with open(self._file, 'w') as cout:
subprocess.run([
'sudo', 'perf', 'script'
], stdout=cout, cwd=self._run_dir, shell=False, check=True)
subprocess.run(['sudo', 'perf', 'script'],
stdout=cout, cwd=self._run_dir, shell=False, check=True)
@property
def file(self):
@ -496,14 +493,14 @@ class ExecResult:
for idx, x in enumerate(self.stats):
assert 'remote_port' in x, f'remote_port missing\n{self.dump_stat(x)}'
assert x['remote_port'] == remote_port, \
f'status #{idx} remote_port: expected {remote_port}, '\
f'got {x["remote_port"]}\n{self.dump_stat(x)}'
f'status #{idx} remote_port: expected {remote_port}, '\
f'got {x["remote_port"]}\n{self.dump_stat(x)}'
if remote_ip is not None:
for idx, x in enumerate(self.stats):
assert 'remote_ip' in x, f'remote_ip missing\n{self.dump_stat(x)}'
assert x['remote_ip'] == remote_ip, \
f'status #{idx} remote_ip: expected {remote_ip}, '\
f'got {x["remote_ip"]}\n{self.dump_stat(x)}'
f'status #{idx} remote_ip: expected {remote_ip}, '\
f'got {x["remote_ip"]}\n{self.dump_stat(x)}'
def check_stat_positive(self, s, idx, key):
assert key in s, f'stat #{idx} "{key}" missing: {s}'
@ -602,8 +599,8 @@ class ExecResult:
return ''.join(lines)
def xfer_trace_for(self, xfer_id) -> List[str]:
pat = re.compile(f'^[^[]* \\[{xfer_id}-.*$')
return [line for line in self._stderr if pat.match(line)]
pat = re.compile(f'^[^[]* \\[{xfer_id}-.*$')
return [line for line in self._stderr if pat.match(line)]
class CurlClient:
@ -853,11 +850,11 @@ class CurlClient:
with_headers=with_headers)
def ftp_get(self, urls: List[str],
with_stats: bool = True,
with_profile: bool = False,
with_tcpdump: bool = False,
no_save: bool = False,
extra_args: Optional[List[str]] = None):
with_stats: bool = True,
with_profile: bool = False,
with_tcpdump: bool = False,
no_save: bool = False,
extra_args: Optional[List[str]] = None):
if extra_args is None:
extra_args = []
if no_save:
@ -882,11 +879,11 @@ class CurlClient:
with_tcpdump=with_tcpdump)
def ftp_ssl_get(self, urls: List[str],
with_stats: bool = True,
with_profile: bool = False,
with_tcpdump: bool = False,
no_save: bool = False,
extra_args: Optional[List[str]] = None):
with_stats: bool = True,
with_profile: bool = False,
with_tcpdump: bool = False,
no_save: bool = False,
extra_args: Optional[List[str]] = None):
if extra_args is None:
extra_args = []
extra_args.extend([
@ -1164,7 +1161,7 @@ class CurlClient:
elif insecure:
args.append('--insecure')
elif active_options and ("--cacert" in active_options or
"--capath" in active_options):
"--capath" in active_options):
pass
elif u.hostname:
args.extend(["--cacert", self.env.ca.cert_file])
@ -1246,9 +1243,8 @@ class CurlClient:
stacks_collapsed = f'{perf.file}.collapsed'
log.info(f'collapsing stacks into {stacks_collapsed}')
with open(stacks_collapsed, 'w') as cout, open(file_err, 'w') as cerr:
subprocess.run([
fg_collapse, perf.file
], stdout=cout, stderr=cerr, cwd=self._run_dir, shell=False, check=True)
subprocess.run([fg_collapse, perf.file],
stdout=cout, stderr=cerr, cwd=self._run_dir, shell=False, check=True)
return stacks_collapsed
def _dtrace_collapse(self, dtrace: DTraceProfile, file_err):
@ -1260,9 +1256,8 @@ class CurlClient:
stacks_collapsed = f'{dtrace.file}.collapsed'
log.info(f'collapsing stacks into {stacks_collapsed}')
with open(stacks_collapsed, 'w') as cout, open(file_err, 'a') as cerr:
subprocess.run([
fg_collapse, dtrace.file
], stdout=cout, stderr=cerr, cwd=self._run_dir, shell=False, check=True)
subprocess.run([fg_collapse, dtrace.file],
stdout=cout, stderr=cerr, cwd=self._run_dir, shell=False, check=True)
return stacks_collapsed
def _generate_flame(self, curl_args: List[str],
@ -1293,11 +1288,9 @@ class CurlClient:
title = cmdline
subtitle = ''
with open(file_svg, 'w') as cout, open(file_err, 'a') as cerr:
subprocess.run([
fg_gen_flame, '--colors', 'green',
'--title', title, '--subtitle', subtitle,
stacks_collapsed
], stdout=cout, stderr=cerr, cwd=self._run_dir, shell=False, check=True)
subprocess.run([fg_gen_flame, '--colors', 'green', '--title', title, '--subtitle',
subtitle, stacks_collapsed],
stdout=cout, stderr=cerr, cwd=self._run_dir, shell=False, check=True)
def mk_altsvc_file(self, name, src_alpn, src_host, src_port,
dest_alpn, dest_host, dest_port):

8
tests/http/testenv/dante.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -128,7 +126,7 @@ class Dante:
'-p', f'{self._pid_file}',
'-d', '0',
]
self._error_fd = open(self._error_log, 'a')
self._error_fd = open(self._error_log, 'a') # noqa: SIM115
self._process = subprocess.Popen(args=args, stderr=self._error_fd)
if self._process.returncode is not None:
return False
@ -137,8 +135,8 @@ class Dante:
def wait_live(self, timeout: timedelta):
curl = CurlClient(env=self.env, run_dir=self._tmp_dir,
timeout=timeout.total_seconds(), socks_args=[
'--socks5', f'127.0.0.1:{self._port}'
])
'--socks5', f'127.0.0.1:{self._port}'
])
try_until = datetime.now() + timeout
while datetime.now() < try_until:
r = curl.http_get(url=f'http://{self.env.domain1}:{self.env.http_port}/')

4
tests/http/testenv/dnsd.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -128,7 +126,7 @@ class Dnsd:
'--logfile', f'{self._log_file}',
'--pidfile', f'{self._pid_file}',
]
self._error_fd = open(self._error_log, 'a')
self._error_fd = open(self._error_log, 'a') # noqa: SIM115
self._process = subprocess.Popen(args=args, stderr=self._error_fd)
if self._process.returncode is not None:
return False

12
tests/http/testenv/env.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# ***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -513,8 +511,8 @@ class Env:
def curl_version_at_least(min_version) -> bool:
version = Env.curl_version()
return Env.CONFIG.versiontuple(min_version) <= Env.CONFIG.versiontuple(
version
)
version
)
@staticmethod
def curl_features_string() -> str:
@ -537,7 +535,7 @@ class Env:
prefix = f"{libname.lower()}/"
for lversion in Env.CONFIG.curl_props["lib_versions"]:
if lversion.startswith(prefix):
return lversion[len(prefix) :]
return lversion[len(prefix):]
return "unknown"
@staticmethod
@ -914,14 +912,14 @@ class Env:
fpath = os.path.join(indir, fname)
s10 = "0123456789"
s = round((line_length / 10) + 1) * s10
s = s[0 : line_length - 11]
s = s[0:line_length - 11]
with open(fpath, "w") as fd:
for i in range(int(fsize / line_length)):
fd.write(f"{i:09d}-{s}\n")
remain = int(fsize % line_length)
if remain != 0:
i = int(fsize / line_length) + 1
fd.write(f"{i:09d}-{s}"[0 : remain - 1] + "\n")
fd.write(f"{i:09d}-{s}"[0:remain - 1] + "\n")
return fpath
def make_data_gzipbomb(self, indir: str, fname: str, fsize: int) -> str:

4
tests/http/testenv/h2o.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# ***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -133,7 +131,7 @@ class H2o:
self._loaded_cred_name = self._cred_name
self.write_config()
args = [self._cmd, "-c", self._conf_file]
self._error_fd = open(self._stderr, "a")
self._error_fd = open(self._stderr, "a") # noqa: SIM115
self._process = subprocess.Popen(args=args, stderr=self._error_fd)
if self._process.returncode is not None:
return False

7
tests/http/testenv/httpd.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -583,9 +581,8 @@ class Httpd:
if not os.path.exists(out_source) or \
os.stat(in_source).st_mtime > os.stat(out_source).st_mtime:
shutil.copy(in_source, out_source)
p = subprocess.run([
self.env.apxs, '-c', out_source
], capture_output=True, cwd=out_dir, check=False)
p = subprocess.run([self.env.apxs, '-c', out_source],
capture_output=True, cwd=out_dir, check=False)
rv = p.returncode
if rv != 0:
log.error(f"compiling mod_curltest failed: {p.stderr}")

6
tests/http/testenv/nghttpx.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -270,7 +268,7 @@ class NghttpxQuic(Nghttpx):
'--frontend-http3-max-connection-window-size=100M',
# f'--frontend-quic-debug-log',
])
self._error_fd = open(self._stderr, 'a')
self._error_fd = open(self._stderr, 'a') # noqa: SIM115
self._process = subprocess.Popen(args=args, stderr=self._error_fd)
if self._process.returncode is not None:
return False
@ -320,7 +318,7 @@ class NghttpxFwd(Nghttpx):
creds.pkey_file,
creds.cert_file,
]
self._error_fd = open(self._stderr, 'a')
self._error_fd = open(self._stderr, 'a') # noqa: SIM115
self._process = subprocess.Popen(args=args, stderr=self._error_fd)
if self._process.returncode is not None:
return False

2
tests/http/testenv/ports.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

4
tests/http/testenv/sshd.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -235,7 +233,7 @@ class Sshd:
run_env = os.environ.copy()
# does not have any effect, sadly
# run_env['HOME'] = f'{self._home_dir}'
self._error_fd = open(self._sshd_log, 'a')
self._error_fd = open(self._sshd_log, 'a') # noqa: SIM115
self._process = subprocess.Popen(args=args, stderr=self._error_fd, env=run_env)
if self._process.returncode is not None:
return False

4
tests/http/testenv/vsftpd.py
View file

@ -1,5 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
@ -146,7 +144,7 @@ class VsFTPD:
self._cmd,
f'{self._conf_file}',
]
self._error_fd = open(self._error_log, 'a')
self._error_fd = open(self._error_log, 'a') # noqa: SIM115
self._process = subprocess.Popen(args=args, stderr=self._error_fd)
if self._process.returncode is not None:
return False

1
tests/http/testenv/ws_4frames_server.py
View file

@ -1,5 +1,4 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

1
tests/http/testenv/ws_echo_server.py
View file

@ -1,5 +1,4 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |

1
tests/negtelnetserver.py
View file

@ -1,5 +1,4 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# Project ___| | | | _ \| |
# / __| | | | |_) | |

17
tests/smbserver.py
View file

@ -1,5 +1,4 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# Project ___| | | | _ \| |
# / __| | | | |_) | |
@ -153,8 +152,9 @@ def smbserver(options):
class TestSmbServer(imp_smbserver.SMBSERVER):
"""
Test server for SMB which subclasses the impacket SMBSERVER and provides
test functionality.
Test server for SMB.
It subclasses the impacket SMBSERVER and provides test functionality.
"""
def __init__(self,
@ -176,9 +176,10 @@ class TestSmbServer(imp_smbserver.SMBSERVER):
def create_and_x(self, conn_id, smb_server, smb_command, recv_packet):
"""
Our version of smbComNtCreateAndX looks for special test files and
fools the rest of the framework into opening them as if they were
normal files.
Our version of smbComNtCreateAndX.
It looks for special test files and fools the rest of the framework
into opening them as if they were normal files.
"""
conn_data = smb_server.getConnectionData(conn_id)
@ -372,9 +373,9 @@ def get_options():
parser = argparse.ArgumentParser()
parser.add_argument("--port", action="store", default=9017,
type=int, help="port to listen on")
type=int, help="port to listen on")
parser.add_argument("--host", action="store", default="127.0.0.1",
help="host to listen on")
help="host to listen on")
parser.add_argument("--verbose", action="store", type=int, default=0,
help="verbose output")
parser.add_argument("--pidfile", action="store",

1
tests/util.py
View file

@ -1,5 +1,4 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# Project ___| | | | _ \| |
# / __| | | | |_) | |