#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
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
# 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
#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
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='')
print(f' {s:<{statw}}', end='')
if 'errors' in cell:
errors.extend(cell['errors'])
- print('')
+ print()
if len(errors):
print(f'Errors: {errors}')
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"])
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',
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
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
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"
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
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
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
])
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:
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']
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]}'
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
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,
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
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=[
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'
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
'-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}'
'-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}'
])
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}'
])
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}'
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
['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 {
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', [
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
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
}
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):
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
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)
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'
])
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'
])
])
# 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):
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
['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)
['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)
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
['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)
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'
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
# ***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
# SPDX-License-Identifier: curl
#
###########################################################################
-# ruff: noqa: F401, E402
+# ruff: noqa: F401
import pytest
pytest.register_assert_rewrite("testenv.env", "testenv.curl", "testenv.caddy",
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
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
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
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
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
return self._duration
@property
- def stats(self) -> Optional[Dict[str,Any]]:
+ def stats(self) -> Optional[Dict[str, Any]]:
return self._stats
def sample(self):
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):
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}'
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:
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:
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([
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])
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):
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],
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):
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
'-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
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}/')
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
'--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
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
# ***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
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:
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
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:
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
# ***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
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
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
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}")
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
'--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
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
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
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
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
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
#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#
# Project ___| | | | _ \| |
# / __| | | | |_) | |
#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#
# Project ___| | | | _ \| |
# / __| | | | |_) | |
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,
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)
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",
#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
#
# Project ___| | | | _ \| |
# / __| | | | |_) | |
projects / thirdparty / curl.git / commitdiff
