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"))
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
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')
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:
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)
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:
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:
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'
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'
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
import shutil
import socket
import subprocess
+import textwrap
import time
from datetime import datetime, timedelta
from json import JSONEncoder
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()
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()
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:
import signal
import socket
import subprocess
+import textwrap
import time
from datetime import datetime, timedelta
from typing import Dict, Optional
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):
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:
"""
def __init__(self, server):
- super(ShutdownHandler, self).__init__()
+ super().__init__()
self.server = server
self.shutdown_event = threading.Event()
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
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:
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)
projects / thirdparty / curl.git / commitdiff
