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: use simpler constructions in Python code

Browse source 
* call super() without arguments
* mark an unused variable as such
* simplify by using dict getter for default values
* use writelines() when possible
* use dedent to simplify some text formatting
* avoid items() on dict in a loop when unnecessary
* replace most Python format() calls with f-strings
* use capture_output in subprocess.run

This fixes ruff rules FLY002, FURB122, PERF102, RUF059, SIM401, UP008,
UP022, UP030.
This commit is contained in:
Dan Fandrich 2026-07-25 14:05:01 -07:00
parent 3901c16933
commit b151a0bb90
12 changed files with 34 additions and 41 deletions
Download patch file Download diff file Expand all files Collapse all files

2
tests/dictserver.py
View file

@ -96,7 +96,7 @@ class DictHandler(socketserver.BaseRequestHandler):
response_data = "No matches"
# Send back a failure to find.
response = "552 {0}\n".format(response_data)
response = f"552 {response_data}\n"
log.debug("[DICT] Responding with %r", response)
self.request.sendall(response.encode("utf-8"))

2
tests/http/test_11_unix.py
View file

@ -68,7 +68,7 @@ class UDSFaker:
def _process(self):
while self._done is False:
try:
c, client_address = self._socket.accept()
c, _client_address = self._socket.accept()
try:
c.recv(16)
c.sendall(b"""HTTP/1.1 200 Ok

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

@ -47,7 +47,7 @@ class Caddy:
def __init__(self, env: Env):
self.env = env
self._caddy = os.environ['CADDY'] if 'CADDY' in os.environ else env.caddy
self._caddy = os.environ.get('CADDY', env.caddy)
self._caddy_dir = os.path.join(env.gen_dir, 'caddy')
self._docs_dir = os.path.join(self._caddy_dir, 'docs')
self._conf_file = os.path.join(self._caddy_dir, 'Caddyfile')

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

@ -280,8 +280,7 @@ class CertStore:
with open(cert_file, "wb") as fd:
fd.write(creds.cert_pem)
if chain:
for c in chain:
fd.write(c.cert_pem)
fd.writelines(c.cert_pem for c in chain)
if pkey_file is None:
fd.write(creds.pkey_pem)
if pkey_file is not None:
@ -290,8 +289,7 @@ class CertStore:
with open(comb_file, "wb") as fd:
fd.write(creds.cert_pem)
if chain:
for c in chain:
fd.write(c.cert_pem)
fd.writelines(c.cert_pem for c in chain)
fd.write(creds.pkey_pem)
creds.set_files(cert_file, pkey_file, comb_file)
self._add_credentials(name, creds)
@ -306,8 +304,7 @@ class CertStore:
chain = chain[:-1]
chain_file = os.path.join(self._store_dir, f'{name}-{infix}.pem')
with open(chain_file, "wb") as fd:
for c in chain:
fd.write(c.cert_pem)
fd.writelines(c.cert_pem for c in chain)
def _add_credentials(self, name: str, creds: Credentials):
if name not in self._creds_by_name:
@ -315,14 +312,14 @@ class CertStore:
self._creds_by_name[name].append(creds)
def get_credentials_for_name(self, name) -> List[Credentials]:
return self._creds_by_name[name] if name in self._creds_by_name else []
return self._creds_by_name.get(name, [])
def get_cert_file(self, name: str, key_type=None) -> str:
key_infix = ".{0}".format(key_type) if key_type is not None else ""
key_infix = f".{key_type}" if key_type is not None else ""
return os.path.join(self._store_dir, f'{name}{key_infix}.cert.pem')
def get_pkey_file(self, name: str, key_type=None) -> str:
key_infix = ".{0}".format(key_type) if key_type is not None else ""
key_infix = f".{key_type}" if key_type is not None else ""
return os.path.join(self._store_dir, f'{name}{key_infix}.pkey.pem')
def get_combined_file(self, name: str, key_type=None) -> str:

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

@ -45,7 +45,7 @@ class LocalClient:
self.env = env
self._run_env = run_env
self._timeout = timeout if timeout else env.test_timeout
self._curl = os.environ['CURL'] if 'CURL' in os.environ else env.curl
self._curl = os.environ.get('CURL', env.curl)
self._run_dir = run_dir if run_dir else os.path.join(env.gen_dir, name)
self._stdoutfile = f'{self._run_dir}/stdout'
self._stderrfile = f'{self._run_dir}/stderr'

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

@ -627,7 +627,7 @@ class CurlClient:
socks_args: Optional[List[str]] = None):
self.env = env
self._timeout = timeout if timeout else env.test_timeout
self._curl = os.environ['CURL'] if 'CURL' in os.environ else env.curl
self._curl = os.environ.get('CURL', env.curl)
self._run_dir = run_dir if run_dir else os.path.join(env.gen_dir, 'curl')
self._stdoutfile = f'{self._run_dir}/curl.stdout'
self._stderrfile = f'{self._run_dir}/curl.stderr'

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

@ -914,8 +914,7 @@ class Env:
s = round((line_length / 10) + 1) * s10
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")
fd.writelines(f"{i:09d}-{s}\n" for i in range(int(fsize / line_length)))
remain = int(fsize % line_length)
if remain != 0:
i = int(fsize / line_length) + 1

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

@ -29,6 +29,7 @@ import os
import shutil
import socket
import subprocess
import textwrap
import time
from datetime import datetime, timedelta
from json import JSONEncoder
@ -137,8 +138,7 @@ class Httpd:
env['APACHE_RUN_USER'] = os.environ['USER']
env['APACHE_LOCK_DIR'] = self._lock_dir
env['APACHE_CONFDIR'] = self._apache_dir
p = subprocess.run(args, stderr=subprocess.PIPE, stdout=subprocess.PIPE,
cwd=self.env.gen_dir,
p = subprocess.run(args, capture_output=True, cwd=self.env.gen_dir,
input=intext.encode() if intext else None,
env=env, check=True)
start = datetime.now()
@ -170,7 +170,7 @@ class Httpd:
def start(self):
# assure ports are allocated
for key, _ in Httpd.PORT_SPECS.items():
for key in Httpd.PORT_SPECS:
assert self.ports[key] is not None
if self._maybe_running:
self.stop()
@ -479,14 +479,13 @@ class Httpd:
fd.write("\n".join(conf))
with open(os.path.join(self._conf_dir, 'mime.types'), 'w') as fd:
fd.write("\n".join([
'text/plain txt',
'text/html html',
'application/json json',
'application/x-gzip gzip',
'application/x-gzip gz',
''
]))
fd.write(textwrap.dedent("""\
text/plain txt
text/html html
application/json json
application/x-gzip gzip
application/x-gzip gz
"""))
def _get_proxy_conf(self):
if self._proxy_auth_basic:

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

@ -27,6 +27,7 @@ import os
import signal
import socket
import subprocess
import textwrap
import time
from datetime import datetime, timedelta
from typing import Dict, Optional
@ -203,10 +204,10 @@ class Nghttpx:
def _write_config(self):
with open(self._conf_file, 'w') as fd:
fd.write('# nghttpx test config')
fd.write("\n".join([
'# do we need something here?'
]))
fd.write(textwrap.dedent("""\
# nghttpx test config
# do we need something here?
"""))
class NghttpxQuic(Nghttpx):

4
tests/negtelnetserver.py
View file

@ -310,9 +310,7 @@ def setup_logging(options):
root_logger = logging.getLogger()
add_stdout = False
formatter = logging.Formatter("%(asctime)s %(levelname)-5.5s "
"[{ident}] %(message)s"
.format(ident=IDENT))
formatter = logging.Formatter(f"%(asctime)s %(levelname)-5.5s [{IDENT}] %(message)s")
# Write out to a logfile
if options.logfile:

4
tests/smbserver.py
View file

@ -64,7 +64,7 @@ class ShutdownHandler(threading.Thread):
"""
def __init__(self, server):
super(ShutdownHandler, self).__init__()
super().__init__()
self.server = server
self.shutdown_event = threading.Event()
@ -351,7 +351,7 @@ class TestSmbServer(imp_smbserver.SMBSERVER):
class SmbError(Exception):
def __init__(self, error_code, error_message):
super(SmbError, self).__init__(error_message)
super().__init__(error_message)
self.error_code = error_code

9
tests/util.py
View file

@ -33,18 +33,18 @@ REPLY_DATA = re.compile("<reply>[ \t\n\r]*<data[^<]*>(.*?)</data>", re.MULTILINE
class ClosingFileHandler(logging.StreamHandler):
def __init__(self, filename):
super(ClosingFileHandler, self).__init__()
super().__init__()
self.filename = os.path.abspath(filename)
self.setStream(None)
def emit(self, record):
with open(self.filename, "a") as fp:
self.setStream(fp)
super(ClosingFileHandler, self).emit(record)
super().emit(record)
self.setStream(None)
def setStream(self, stream):
setStream = getattr(super(ClosingFileHandler, self), 'setStream', None)
setStream = getattr(super(), 'setStream', None)
if callable(setStream):
return setStream(stream)
if stream is self.stream:
@ -66,8 +66,7 @@ class TestData:
def get_test_data(self, test_number):
# Create the test filename
filename = os.path.join(self.data_folder,
"test{0}".format(test_number))
filename = os.path.join(self.data_folder, f"test{test_number}")
log.debug("Parsing file %s", filename)