← Commit list

e13362c20a

Dan Fandrich
Dan Fandrich
2026-07-28
SUCCESS Duration: 187s

Subject

tests: address mutable class vars and naive datetime in Python code

Binary

No binary (build failed)

Architecture & Compiler Flags

Architecture: x86_64-linux-gnu Compiler: gcc (Ubuntu 11.4.0-1ubuntu1~22.04.3) 11.4.0

—

Build Log

(log yok)

Commit Diff GitHub →

tests/http/test_17_ssl_use.py +7 -4tests/http/test_20_websockets.py +3 -3tests/http/testenv/caddy.py +7 -7tests/http/testenv/certs.py +3 -3tests/http/testenv/client.py +3 -3tests/http/testenv/curl.py +9 -9tests/http/testenv/dante.py +3 -3tests/http/testenv/dnsd.py +3 -3tests/http/testenv/h2o.py +10 -10tests/http/testenv/httpd.py +11 -11tests/http/testenv/nghttpx.py +14 -14tests/http/testenv/sshd.py +3 -3tests/http/testenv/vsftpd.py +5 -5
tests/http/test_17_ssl_use.py+7 -4
@@ -26,6 +26,8 @@ import json
import logging
import os
import re
+from dataclasses import dataclass
+from typing import ClassVar, Dict, List
import pytest
from testenv import CurlClient, Env, LocalClient
@@ -33,15 +35,16 @@ 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,
@@ -49,7 +52,7 @@ class TLSDefs:
'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,
tests/http/test_20_websockets.py+3 -3
@@ -32,7 +32,7 @@ import socket
import subprocess
import threading
import time
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
from typing import Dict
import pytest
@@ -59,8 +59,8 @@ class WsServer:
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
tests/http/testenv/caddy.py+7 -7
@@ -27,9 +27,9 @@ import os
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
@@ -40,7 +40,7 @@ log = logging.getLogger(__name__)
class Caddy:
- PORT_SPECS = {
+ PORT_SPECS: ClassVar[Dict[str, int]] = {
'caddy': socket.SOCK_STREAM,
'caddys': socket.SOCK_STREAM,
}
@@ -137,8 +137,8 @@ class Caddy:
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:
@@ -150,8 +150,8 @@ class Caddy:
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:
tests/http/testenv/certs.py+3 -3
@@ -350,7 +350,7 @@ class CertStore:
(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)):
@@ -422,10 +422,10 @@ class TestCA:
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
tests/http/testenv/client.py+3 -3
@@ -26,7 +26,7 @@ import logging
import os
import shutil
import subprocess
-from datetime import datetime
+from datetime import datetime, timezone
from typing import Dict, Optional
from . import ExecResult
@@ -81,7 +81,7 @@ class LocalClient:
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)
@@ -107,7 +107,7 @@ class LocalClient:
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 = []
tests/http/testenv/curl.py+9 -9
@@ -35,7 +35,7 @@ from datetime import datetime, timedelta, timezone
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
@@ -47,7 +47,7 @@ log = logging.getLogger(__name__)
class RunProfile:
- STAT_KEYS = ['cpu', 'rss', 'vsz']
+ STAT_KEYS: ClassVar[List[str]] = ['cpu', 'rss', 'vsz']
@classmethod
def AverageStats(cls, profiles: List['RunProfile']):
@@ -76,7 +76,7 @@ class 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)
@@ -92,7 +92,7 @@ class RunProfile:
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 = {}
@@ -605,7 +605,7 @@ class ExecResult:
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',
@@ -1028,7 +1028,7 @@ class CurlClient:
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:
@@ -1051,7 +1051,7 @@ class CurlClient:
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()
@@ -1067,13 +1067,13 @@ class CurlClient:
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:
tests/http/testenv/dante.py+3 -3
@@ -27,7 +27,7 @@ import os
import socket
import subprocess
import time
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
from typing import Dict
from . import CurlClient
@@ -137,8 +137,8 @@ class Dante:
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
tests/http/testenv/dnsd.py+3 -3
@@ -27,7 +27,7 @@ import os
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
@@ -133,8 +133,8 @@ class Dnsd:
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)
tests/http/testenv/h2o.py+10 -10
@@ -28,8 +28,8 @@ import signal
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
@@ -181,12 +181,12 @@ class H2o:
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)
@@ -199,7 +199,7 @@ class H2o:
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()
@@ -215,10 +215,10 @@ class H2o:
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"]
@@ -245,7 +245,7 @@ class H2o:
class H2oServer(H2o):
"""h2o HTTP/3 server for testing."""
- PORT_SPECS = {
+ PORT_SPECS: ClassVar[Dict[str, int]] = {
"h2o_https": socket.SOCK_STREAM,
}
@@ -422,10 +422,10 @@ error-log: {self._error_log}
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"]
tests/http/testenv/httpd.py+11 -11
@@ -31,9 +31,9 @@ import socket
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
@@ -44,7 +44,7 @@ log = logging.getLogger(__name__)
class Httpd:
- MODULES = [
+ MODULES: ClassVar[List[str]] = [
'log_config', 'logio', 'unixd', 'version', 'watchdog',
'authn_core', 'authn_file',
'authz_user', 'authz_core', 'authz_host',
@@ -55,14 +55,14 @@ class Httpd:
'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,
@@ -141,11 +141,11 @@ class Httpd:
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,
@@ -221,8 +221,8 @@ class 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
@@ -234,8 +234,8 @@ class Httpd:
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
tests/http/testenv/nghttpx.py+14 -14
@@ -29,8 +29,8 @@ import socket
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
@@ -135,11 +135,11 @@ class Nghttpx:
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)
@@ -149,7 +149,7 @@ class Nghttpx:
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()
@@ -159,8 +159,8 @@ class Nghttpx:
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'
@@ -178,8 +178,8 @@ class Nghttpx:
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'
@@ -212,7 +212,7 @@ class Nghttpx:
class NghttpxQuic(Nghttpx):
- PORT_SPECS = {
+ PORT_SPECS: ClassVar[Dict[str, int]] = {
'nghttpx_https': socket.SOCK_STREAM,
}
@@ -327,8 +327,8 @@ class NghttpxFwd(Nghttpx):
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:
@@ -340,8 +340,8 @@ class NghttpxFwd(Nghttpx):
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'
tests/http/testenv/sshd.py+3 -3
@@ -28,7 +28,7 @@ import socket
import stat
import subprocess
import time
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
from typing import Dict
from . import CurlClient
@@ -242,8 +242,8 @@ class Sshd:
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',
tests/http/testenv/vsftpd.py+5 -5
@@ -28,7 +28,7 @@ import re
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
@@ -152,8 +152,8 @@ class VsFTPD:
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:
@@ -165,8 +165,8 @@ class VsFTPD:
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'