tests: simplify by removing unneeded Python code

* combine separate if statements
* remove an unneeded encode() call
* remove unneeded returns
* simplify code when returning early

This fixes ruff rules SIM102, SIM114, UP012, PLR1711.
This commit is contained in:
Dan Fandrich 2026-07-25 13:49:29 -07:00
parent 4eb691e89c
commit 3901c16933
8 changed files with 41 additions and 53 deletions

View file

@ -795,7 +795,7 @@ def run_score(args, protocol):
socks_args = None
if args.socks4 and args.socks5:
raise ScoreCardError('unable to run --socks4 and --socks5 together')
elif args.socks4 or args.socks5:
if args.socks4 or args.socks5:
sockd = Dante(env=env)
if sockd:
assert sockd.initial_start()

View file

@ -701,10 +701,6 @@ class TestDownload:
if url_junk <= 1024:
r.check_exit_code(0)
r.check_response(http_status=200)
elif url_junk <= 16 * 1024:
r.check_exit_code(0)
# server replies with 414, Request URL too long
r.check_response(http_status=414)
elif url_junk <= 32 * 1024:
r.check_exit_code(0)
# server replies with 414, Request URL too long

View file

@ -71,12 +71,12 @@ class UDSFaker:
c, client_address = self._socket.accept()
try:
c.recv(16)
c.sendall("""HTTP/1.1 200 Ok
c.sendall(b"""HTTP/1.1 200 Ok
Server: UdsFaker
Content-Type: application/json
Content-Length: 19
{ "host": "faked" }""".encode())
{ "host": "faked" }""")
finally:
c.close()

View file

@ -80,9 +80,8 @@ class TestSSLUse:
count = 3
exp_resumed = 'Resumed'
xargs = ['--sessionid', '--tls-max', tls_max, f'--tlsv{tls_max}']
if env.curl_uses_lib('libressl'):
if tls_max == '1.3':
exp_resumed = 'Initial' # 1.2 works in LibreSSL, but 1.3 does not, TODO
if env.curl_uses_lib('libressl') and tls_max == '1.3':
exp_resumed = 'Initial' # 1.2 works in LibreSSL, but 1.3 does not, TODO
if env.curl_uses_lib('rustls-ffi'):
exp_resumed = 'Initial' # Rustls does not support sessions, TODO
if env.curl_uses_lib('mbedtls') and tls_max == '1.3' and \
@ -265,9 +264,9 @@ class TestSSLUse:
elif env.curl_uses_lib('schannel'): # not in CI, so untested
if ciphers12 is not None:
pytest.skip('Schannel does not support setting TLSv1.2 ciphers by name')
elif env.curl_uses_lib('mbedtls') and not env.curl_lib_version_at_least('mbedtls', '3.6.0'):
if tls_proto == 'TLSv1.3':
pytest.skip('mbedTLS < 3.6.0 does not support TLSv1.3')
elif (env.curl_uses_lib('mbedtls') and not env.curl_lib_version_at_least('mbedtls', '3.6.0')
and tls_proto == 'TLSv1.3'):
pytest.skip('mbedTLS < 3.6.0 does not support TLSv1.3')
# test
extra_args = ['--tls13-ciphers', ':'.join(ciphers13)] if ciphers13 else []
extra_args += ['--ciphers', ':'.join(ciphers12)] if ciphers12 else []
@ -317,13 +316,12 @@ class TestSSLUse:
])
httpd.reload_if_config_changed()
# curl's TLS backend supported version
if env.curl_uses_lib('gnutls') or \
env.curl_uses_lib('quiche') or \
env.curl_uses_lib('aws-lc') or \
env.curl_uses_lib('boringssl'):
curl_supported = [0x301, 0x302, 0x303, 0x304]
elif env.curl_uses_lib('openssl') and \
env.curl_lib_version_before('openssl', '3.0.0'):
if (env.curl_uses_lib('gnutls') or
env.curl_uses_lib('quiche') or
env.curl_uses_lib('aws-lc') or
env.curl_uses_lib('boringssl') or
(env.curl_uses_lib('openssl') and
env.curl_lib_version_before('openssl', '3.0.0'))):
curl_supported = [0x301, 0x302, 0x303, 0x304]
else: # most SSL backends dropped support for TLSv1.0, TLSv1.1
curl_supported = [0x303, 0x304]

View file

@ -389,20 +389,18 @@ class TestCA:
:returns: the certificate and private key PEM file paths
"""
if spec.domains and len(spec.domains):
creds = TestCA._make_server_credentials(name=spec.name, domains=spec.domains,
issuer=issuer, valid_from=valid_from,
valid_to=valid_to, key_type=key_type)
elif spec.client:
creds = TestCA._make_client_credentials(name=spec.name, issuer=issuer,
email=spec.email, valid_from=valid_from,
valid_to=valid_to, key_type=key_type)
elif spec.name:
creds = TestCA._make_ca_credentials(name=spec.name, issuer=issuer,
valid_from=valid_from, valid_to=valid_to,
key_type=key_type)
else:
raise CertError(f"unrecognized certificate specification: {spec}")
return creds
return TestCA._make_server_credentials(name=spec.name, domains=spec.domains,
issuer=issuer, valid_from=valid_from,
valid_to=valid_to, key_type=key_type)
if spec.client:
return TestCA._make_client_credentials(name=spec.name, issuer=issuer,
email=spec.email, valid_from=valid_from,
valid_to=valid_to, key_type=key_type)
if spec.name:
return TestCA._make_ca_credentials(name=spec.name, issuer=issuer,
valid_from=valid_from, valid_to=valid_to,
key_type=key_type)
raise CertError(f"unrecognized certificate specification: {spec}")
@staticmethod
def _make_x509_name(org_name: Optional[str] = None, common_name: Optional[str] = None, parent: x509.Name = None) -> x509.Name:

View file

@ -95,7 +95,6 @@ class H2o:
def _rmf(self, path):
if os.path.isfile(path):
os.remove(path)
return
def _dump_file(self, path, lines):
if os.path.isfile(path):
@ -106,7 +105,6 @@ class H2o:
def _mkpath(self, path):
if not os.path.exists(path):
os.makedirs(path)
return
def _log(self, level, msg):
getattr(log, level)(f"[{self._name}] {msg}")

View file

@ -293,15 +293,13 @@ class TestSmbServer(imp_smbserver.SMBSERVER):
# If we have a rootFid, the path is relative to that fid
path = conn_data["OpenedFiles"][root_fid]["FileName"]
log.debug(f'RootFid present {path}!')
else:
if "path" in conn_shares[tid]:
path = conn_shares[tid]["path"]
else:
raise SmbError(STATUS_ACCESS_DENIED, "Connection share had no path")
else:
raise SmbError(imp_smbserver.STATUS_SMB_BAD_TID, "TID was invalid")
return path
return path
if "path" in conn_shares[tid]:
return conn_shares[tid]["path"]
raise SmbError(STATUS_ACCESS_DENIED, "Connection share had no path")
raise SmbError(imp_smbserver.STATUS_SMB_BAD_TID, "TID was invalid")
def get_server_path(self, requested_filename):
log.debug("[SMB] Get server path '%s'", requested_filename)

View file

@ -48,15 +48,15 @@ class ClosingFileHandler(logging.StreamHandler):
if callable(setStream):
return setStream(stream)
if stream is self.stream:
result = None
else:
result = self.stream
self.acquire()
try:
self.flush()
self.stream = stream
finally:
self.release()
return None
result = self.stream
self.acquire()
try:
self.flush()
self.stream = stream
finally:
self.release()
return result