import logging
import os
import re
+from dataclasses import dataclass
+from typing import ClassVar, Dict, List
import pytest
from testenv import CurlClient, Env, LocalClient
log = logging.getLogger(__name__)
+@dataclass(frozen=True)
class TLSDefs:
- TLS_VERSIONS = ['TLSv1', 'TLSv1.1', 'TLSv1.2', 'TLSv1.3']
- TLS_VERSION_IDS = {
+ TLS_VERSIONS: ClassVar[List[str]] = ['TLSv1', 'TLSv1.1', 'TLSv1.2', 'TLSv1.3']
+ TLS_VERSION_IDS: ClassVar[Dict[str, int]] = {
'TLSv1': 0x301,
'TLSv1.1': 0x302,
'TLSv1.2': 0x303,
'TLSv1.3': 0x304
}
- CURL_ARG_MIN_VERSION_ID = {
+ CURL_ARG_MIN_VERSION_ID: ClassVar[Dict[str, int]] = {
'none': 0x0,
'tlsv1': 0x301,
'tlsv1.0': 0x301,
'tlsv1.2': 0x303,
'tlsv1.3': 0x304,
}
- CURL_ARG_MAX_VERSION_ID = {
+ CURL_ARG_MAX_VERSION_ID: ClassVar[Dict[str, int]] = {
'none': 0x0,
'1.0': 0x301,
'1.1': 0x302,
import subprocess
import threading
import time
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
from typing import Dict
import pytest
def check_alive(self, env, port, timeout=Env.SERVER_TIMEOUT):
curl = CurlClient(env=env)
url = f'http://localhost:{port}/'
- end = datetime.now() + timedelta(seconds=timeout)
- while datetime.now() < end:
+ end = datetime.now(timezone.utc) + timedelta(seconds=timeout)
+ while datetime.now(timezone.utc) < end:
r = curl.http_download(urls=[url])
if r.exit_code == 0:
return True
import socket
import subprocess
import time
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
from json import JSONEncoder
-from typing import Dict
+from typing import ClassVar, Dict
from .curl import CurlClient
from .env import Env
class Caddy:
- PORT_SPECS = {
+ PORT_SPECS: ClassVar[Dict[str, int]] = {
'caddy': socket.SOCK_STREAM,
'caddys': socket.SOCK_STREAM,
}
def wait_dead(self, timeout: timedelta):
curl = CurlClient(env=self.env, run_dir=self._tmp_dir)
- try_until = datetime.now() + timeout
- while datetime.now() < try_until:
+ try_until = datetime.now(timezone.utc) + timeout
+ while datetime.now(timezone.utc) < try_until:
check_url = f'https://{self.env.domain1}:{self.port}/'
r = curl.http_get(url=check_url)
if r.exit_code != 0:
def wait_live(self, timeout: timedelta):
curl = CurlClient(env=self.env, run_dir=self._tmp_dir)
- try_until = datetime.now() + timeout
- while datetime.now() < try_until:
+ try_until = datetime.now(timezone.utc) + timeout
+ while datetime.now(timezone.utc) < try_until:
check_url = f'https://{self.env.domain1}:{self.port}/'
r = curl.http_get(url=check_url)
if r.exit_code == 0:
(cert.not_valid_before_utc > now)):
return None
except AttributeError: # older python
- now = datetime.now()
+ now = datetime.now(timezone.utc)
if check_valid and \
((cert.not_valid_after < now) or
(cert.not_valid_before > now)):
pubkey = pkey.public_key()
issuer_subject = issuer_subject if issuer_subject is not None else subject
- valid_from = datetime.now()
+ valid_from = datetime.now(timezone.utc)
if valid_until_delta is not None:
valid_from += valid_from_delta
- valid_until = datetime.now()
+ valid_until = datetime.now(timezone.utc)
if valid_until_delta is not None:
valid_until += valid_until_delta
import os
import shutil
import subprocess
-from datetime import datetime
+from datetime import datetime, timezone
from typing import Dict, Optional
from . import ExecResult
def run(self, args):
self._rmf(self._stdoutfile)
self._rmf(self._stderrfile)
- start = datetime.now()
+ start = datetime.now(timezone.utc)
exception = None
myargs = [self.path, self.name]
myargs.extend(args)
cerrput = ferr.readlines()
return ExecResult(args=myargs, exit_code=exitcode, exception=exception,
stdout=coutput, stderr=cerrput,
- duration=datetime.now() - start)
+ duration=datetime.now(timezone.utc) - start)
def dump_logs(self):
lines = []
from functools import cmp_to_key
from statistics import fmean, mean
from threading import Thread
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import Any, ClassVar, Dict, List, Optional, Tuple, Union
from urllib.parse import urlparse
import psutil
class RunProfile:
- STAT_KEYS = ['cpu', 'rss', 'vsz']
+ STAT_KEYS: ClassVar[List[str]] = ['cpu', 'rss', 'vsz']
@classmethod
def AverageStats(cls, profiles: List['RunProfile']):
return self._stats
def sample(self):
- elapsed = datetime.now() - self._started_at
+ elapsed = datetime.now(timezone.utc) - self._started_at
try:
if self._psu is None:
self._psu = psutil.Process(pid=self._pid)
pass
def finish(self):
- self._duration = datetime.now() - self._started_at
+ self._duration = datetime.now(timezone.utc) - self._started_at
if len(self._samples) > 0:
weights = [s['time'].total_seconds() for s in self._samples]
self._stats = {}
class CurlClient:
- ALPN_ARG = {
+ ALPN_ARG: ClassVar[Dict[str, str]] = {
'http/0.9': '--http0.9',
'http/1.0': '--http1.0',
'http/1.1': '--http1.1',
if with_tcpdump:
tcpdump = RunTcpDump(self.env, self._run_dir)
tcpdump.start()
- started_at = datetime.now()
+ started_at = datetime.now(timezone.utc)
try:
with open(self._stdoutfile, 'w') as cout, open(self._stderrfile, 'w') as cerr:
if with_profile:
p.wait(timeout=ptimeout)
break
except subprocess.TimeoutExpired as e:
- if end_at and datetime.now() >= end_at:
+ if end_at and datetime.now(timezone.utc) >= end_at:
p.kill()
raise subprocess.TimeoutExpired(cmd=args, timeout=self._timeout) from e
profile.sample()
env=self._run_env, check=False)
exitcode = p.returncode
except subprocess.TimeoutExpired:
- now = datetime.now()
+ now = datetime.now(timezone.utc)
duration = now - started_at
log.warning(f'Timeout at {now} after {duration.total_seconds()}s '
f'(configured {self._timeout}s): {args}')
exitcode = -1
exception = 'TimeoutExpired'
- ended_at = datetime.now()
+ ended_at = datetime.now(timezone.utc)
if tcpdump:
tcpdump.finish()
if perf:
import socket
import subprocess
import time
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
from typing import Dict
from . import CurlClient
timeout=timeout.total_seconds(), socks_args=[
'--socks5', f'127.0.0.1:{self._port}'
])
- try_until = datetime.now() + timeout
- while datetime.now() < try_until:
+ try_until = datetime.now(timezone.utc) + timeout
+ while datetime.now(timezone.utc) < try_until:
r = curl.http_get(url=f'http://{self.env.domain1}:{self.env.http_port}/')
if r.exit_code == 0:
return True
import socket
import subprocess
import time
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
from typing import Dict, List, Optional
from .env import Env
return self.wait_live(timeout=timedelta(seconds=Env.SERVER_TIMEOUT))
def wait_live(self, timeout: timedelta):
- try_until = datetime.now() + timeout
- while datetime.now() < try_until:
+ try_until = datetime.now(timezone.utc) + timeout
+ while datetime.now(timezone.utc) < try_until:
if os.path.exists(self._log_file):
return True
time.sleep(.1)
import socket
import subprocess
import time
-from datetime import datetime, timedelta
-from typing import Dict, Optional
+from datetime import datetime, timedelta, timezone
+from typing import ClassVar, Dict, Optional
from .curl import CurlClient
from .env import Env
running = self._process
self._process = None
os.kill(running.pid, signal.SIGQUIT)
- end_wait = datetime.now() + timedelta(seconds=5)
+ end_wait = datetime.now(timezone.utc) + timedelta(seconds=5)
exited = False
if not self.start(wait_live=False):
self._process = running
return False
- while datetime.now() < end_wait:
+ while datetime.now(timezone.utc) < end_wait:
try:
self._log("debug", f"waiting for h2o({running.pid}) to exit.")
running.wait(1)
except subprocess.TimeoutExpired:
self._log("warning", f"h2o({running.pid}), not shut down yet.")
os.kill(running.pid, signal.SIGQUIT)
- if not exited and datetime.now() >= end_wait:
+ if not exited and datetime.now(timezone.utc) >= end_wait:
self._log("error", f"h2o({running.pid}), terminate forcefully.")
os.kill(running.pid, signal.SIGKILL)
running.terminate()
log_prefix: str = "h2o",
):
curl = CurlClient(env=self.env, run_dir=self._tmp_dir)
- try_until = datetime.now() + timeout
+ try_until = datetime.now(timezone.utc) + timeout
if url is None:
url = f"https://{self._domain}:{self._port}/"
- while datetime.now() < try_until:
+ while datetime.now(timezone.utc) < try_until:
if live:
r = curl.http_get(
url=url, extra_args=["--trace", "curl.trace", "--trace-time"]
class H2oServer(H2o):
"""h2o HTTP/3 server for testing."""
- PORT_SPECS = {
+ PORT_SPECS: ClassVar[Dict[str, int]] = {
"h2o_https": socket.SOCK_STREAM,
}
log_prefix: str = "h2o",
):
curl = CurlClient(env=self.env, run_dir=self._tmp_dir)
- try_until = datetime.now() + timeout
+ try_until = datetime.now(timezone.utc) + timeout
if url is None:
url = f"https://{self.env.proxy_domain}:{self._port}/"
- while datetime.now() < try_until:
+ while datetime.now(timezone.utc) < try_until:
if live:
r = curl.http_get(
url=url, extra_args=["--trace", "curl.trace", "--trace-time"]
import subprocess
import textwrap
import time
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
from json import JSONEncoder
-from typing import Dict, List, Optional, Union
+from typing import ClassVar, Dict, List, Optional, Union
from .curl import CurlClient, ExecResult
from .env import Env, EnvError
class Httpd:
- MODULES = [
+ MODULES: ClassVar[List[str]] = [
'log_config', 'logio', 'unixd', 'version', 'watchdog',
'authn_core', 'authn_file',
'authz_user', 'authz_core', 'authz_host',
'brotli',
'mpm_event',
]
- COMMON_MODULES_DIRS = [
+ COMMON_MODULES_DIRS: ClassVar[List[str]] = [
'/usr/lib/apache2/modules', # debian
'/usr/libexec/apache2/', # macos
]
MOD_CURLTEST = None
- PORT_SPECS = {
+ PORT_SPECS: ClassVar[Dict[str, int]] = {
'http': socket.SOCK_STREAM,
'https': socket.SOCK_STREAM,
'https-tcp-only': socket.SOCK_STREAM,
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()
+ start = datetime.now(timezone.utc)
return ExecResult(args=args, exit_code=p.returncode,
stdout=p.stdout.decode().splitlines(),
stderr=p.stderr.decode().splitlines(),
- duration=datetime.now() - start)
+ duration=datetime.now(timezone.utc) - start)
def _cmd_httpd(self, cmd: str):
args = [self.env.httpd,
def wait_dead(self, timeout: timedelta):
curl = CurlClient(env=self.env, run_dir=self._tmp_dir)
- try_until = datetime.now() + timeout
- while datetime.now() < try_until:
+ try_until = datetime.now(timezone.utc) + timeout
+ while datetime.now(timezone.utc) < try_until:
r = curl.http_get(url=f'http://{self.env.domain1}:{self.ports["http"]}/')
if r.exit_code != 0:
self._maybe_running = False
def wait_live(self, timeout: timedelta):
curl = CurlClient(env=self.env, run_dir=self._tmp_dir,
timeout=timeout.total_seconds())
- try_until = datetime.now() + timeout
- while datetime.now() < try_until:
+ try_until = datetime.now(timezone.utc) + timeout
+ while datetime.now(timezone.utc) < try_until:
r = curl.http_get(url=f'http://{self.env.domain1}:{self.ports["http"]}/')
if r.exit_code == 0:
self._maybe_running = True
import subprocess
import textwrap
import time
-from datetime import datetime, timedelta
-from typing import Dict, Optional
+from datetime import datetime, timedelta, timezone
+from typing import ClassVar, Dict, Optional
from .curl import CurlClient
from .env import Env, NghttpxUtil
running = self._process
self._process = None
os.kill(running.pid, signal.SIGQUIT)
- end_wait = datetime.now() + timedelta(seconds=5)
+ end_wait = datetime.now(timezone.utc) + timedelta(seconds=5)
if not self.start(wait_live=False):
self._process = running
return False
- while datetime.now() < end_wait:
+ while datetime.now(timezone.utc) < end_wait:
try:
log.debug(f'waiting for nghttpx({running.pid}) to exit.')
running.wait(1)
except subprocess.TimeoutExpired:
log.warning(f'nghttpx({running.pid}), not shut down yet.')
os.kill(running.pid, signal.SIGQUIT)
- if running and datetime.now() >= end_wait:
+ if running and datetime.now(timezone.utc) >= end_wait:
log.error(f'nghttpx({running.pid}), terminate forcefully.')
os.kill(running.pid, signal.SIGKILL)
running.terminate()
def wait_dead(self, timeout: timedelta):
curl = CurlClient(env=self.env, run_dir=self._tmp_dir)
- try_until = datetime.now() + timeout
- while datetime.now() < try_until:
+ try_until = datetime.now(timezone.utc) + timeout
+ while datetime.now(timezone.utc) < try_until:
xargs = [
'--trace', 'curl.trace', '--trace-time',
'--connect-timeout', '1'
def wait_live(self, timeout: timedelta):
curl = CurlClient(env=self.env, run_dir=self._tmp_dir)
- try_until = datetime.now() + timeout
- while datetime.now() < try_until:
+ try_until = datetime.now(timezone.utc) + timeout
+ while datetime.now(timezone.utc) < try_until:
xargs = [
'--trace', 'curl.trace', '--trace-time',
'--connect-timeout', '1'
class NghttpxQuic(Nghttpx):
- PORT_SPECS = {
+ PORT_SPECS: ClassVar[Dict[str, int]] = {
'nghttpx_https': socket.SOCK_STREAM,
}
def wait_dead(self, timeout: timedelta):
curl = CurlClient(env=self.env, run_dir=self._tmp_dir)
- try_until = datetime.now() + timeout
- while datetime.now() < try_until:
+ try_until = datetime.now(timezone.utc) + timeout
+ while datetime.now(timezone.utc) < try_until:
check_url = f'https://{self.env.proxy_domain}:{self._port}/'
r = curl.http_get(url=check_url)
if r.exit_code != 0:
def wait_live(self, timeout: timedelta):
curl = CurlClient(env=self.env, run_dir=self._tmp_dir)
- try_until = datetime.now() + timeout
- while datetime.now() < try_until:
+ try_until = datetime.now(timezone.utc) + timeout
+ while datetime.now(timezone.utc) < try_until:
check_url = f'https://{self.env.proxy_domain}:{self._port}/'
r = curl.http_get(url=check_url, extra_args=[
'--trace', 'curl.trace', '--trace-time'
import stat
import subprocess
import time
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
from typing import Dict
from . import CurlClient
def wait_live(self, timeout: timedelta):
curl = CurlClient(env=self.env, run_dir=self._tmp_dir,
timeout=timeout.total_seconds())
- try_until = datetime.now() + timeout
- while datetime.now() < try_until:
+ try_until = datetime.now(timezone.utc) + timeout
+ while datetime.now(timezone.utc) < try_until:
r = curl.http_get(url=f'scp://{self.env.domain1}:{self._port}/{self.home_dir}/data',
extra_args=[
'--insecure',
import socket
import subprocess
import time
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
from typing import Dict, List, Tuple
from .curl import CurlClient, ExecResult
def wait_dead(self, timeout: timedelta):
curl = CurlClient(env=self.env, run_dir=self._tmp_dir)
- try_until = datetime.now() + timeout
- while datetime.now() < try_until:
+ try_until = datetime.now(timezone.utc) + timeout
+ while datetime.now(timezone.utc) < try_until:
check_url = f'{self._scheme}://{self.domain}:{self.port}/'
r = curl.ftp_get(urls=[check_url], extra_args=['-v'])
if r.exit_code != 0:
def wait_live(self, timeout: timedelta):
curl = CurlClient(env=self.env, run_dir=self._tmp_dir)
- try_until = datetime.now() + timeout
- while datetime.now() < try_until:
+ try_until = datetime.now(timezone.utc) + timeout
+ while datetime.now(timezone.utc) < try_until:
check_url = f'{self._scheme}://{self.domain}:{self.port}/'
r = curl.ftp_get(urls=[check_url], extra_args=[
'--trace', 'curl-start.trace', '--trace-time'
projects / thirdparty / curl.git / commitdiff
