class Kresd(ContextDecorator):
- def __init__(
+ def __init__( # noqa: PLR0913
self, workdir, port=None, tls_port=None, ip=None, ip6=None, certname=None,
verbose=True, hints=None, forward=None, policy_test_pass=False, rr=False,
valgrind=False):
port = self.tls_port if tls else self.port
if family == socket.AF_INET:
return self.ip, port
- elif family == socket.AF_INET6:
+ if family == socket.AF_INET6:
return self.ip6, port, 0, 0
raise RuntimeError("Unsupported socket family: {}".format(family))
for line in log:
if past_startup:
partial_log += line
- else: # find real start of test log (after initial alive-pings)
- if not past_startup_msgid:
- if re.match(KRESD_LOG_STARTUP_MSGID, line) is not None:
- past_startup_msgid = True
- else:
- if re.match(KRESD_LOG_IO_CLOSE, line) is not None:
- past_startup = True
+ # find real start of test log (after initial alive-pings)
+ elif not past_startup_msgid:
+ if re.match(KRESD_LOG_STARTUP_MSGID, line) is not None:
+ past_startup_msgid = True
+ elif re.match(KRESD_LOG_IO_CLOSE, line) is not None:
+ past_startup = True
return partial_log
except OSError as exc:
if exc.errno == 98: # address already in use
return False
- else:
- raise
+ raise
return True
class TLSProxy(Proxy):
EXECUTABLE = 'tlsproxy'
- def __init__(
+ def __init__( # noqa: PLR0913
self,
local_ip: str = '127.0.0.1',
local_port: Optional[int] = None,
kresd_fwd_target_kwargs: Optional[Dict[Any, Any]] = None
) -> Kresd:
"""kresd_tls_client --(tls)--> tlsproxy --(tcp)--> kresd_fwd_target"""
- ALLOWED_IPS = {'127.0.0.1', '::1'}
- assert proxy.local_ip in ALLOWED_IPS, "only localhost IPs supported for proxy"
- assert proxy.upstream_ip in ALLOWED_IPS, "only localhost IPs are supported for proxy"
+ allowed_ips = {'127.0.0.1', '::1'}
+ assert proxy.local_ip in allowed_ips, "only localhost IPs supported for proxy"
+ assert proxy.upstream_ip in allowed_ips, "only localhost IPs are supported for proxy"
if kresd_tls_client_kwargs is None:
kresd_tls_client_kwargs = {}