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

Mark Python mutable class variables with ClassVar, to denote that the
danger this can cause has been considered.  Since any change made to
these in any object affects all other objects, this can cause locality
errors. However, as used in the test suite, they are are never modified
and so they are annotated as being intended.

Always set a timezone in datetime objects, as mixing naive and
timezone-aware object can cause errors.

These fix ruff rules DTZ005, RUF012.

tests/http/test_17_ssl_use.py

@@ -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_VERSIONS: ClassVar[List[str]] = ['TLSv1', 'TLSv1.1', 'TLSv1.2', 'TLSv1.3']
-    TLS_VERSION_IDS = {
+    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

@@ -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)
+        end = datetime.now(timezone.utc) + timedelta(seconds=timeout)
-        while datetime.now() < end:
+        while datetime.now(timezone.utc) < end:
             r = curl.http_download(urls=[url])
             if r.exit_code == 0:
                 return True

tests/http/testenv/caddy.py

@@ -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
+        try_until = datetime.now(timezone.utc) + timeout
-        while datetime.now() < try_until:
+        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
+        try_until = datetime.now(timezone.utc) + timeout
-        while datetime.now() < try_until:
+        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

@@ -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

@@ -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

@@ -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

@@ -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
+        try_until = datetime.now(timezone.utc) + timeout
-        while datetime.now() < try_until:
+        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

@@ -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
+        try_until = datetime.now(timezone.utc) + timeout
-        while datetime.now() < try_until:
+        while datetime.now(timezone.utc) < try_until:
             if os.path.exists(self._log_file):
                 return True
             time.sleep(.1)

tests/http/testenv/h2o.py

@@ -28,8 +28,8 @@ import signal
 import socket
 import subprocess
 import time
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
-from typing import Dict, Optional
+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

@@ -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
+        try_until = datetime.now(timezone.utc) + timeout
-        while datetime.now() < try_until:
+        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
+        try_until = datetime.now(timezone.utc) + timeout
-        while datetime.now() < try_until:
+        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

@@ -29,8 +29,8 @@ import socket
 import subprocess
 import textwrap
 import time
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
-from typing import Dict, Optional
+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
+        try_until = datetime.now(timezone.utc) + timeout
-        while datetime.now() < try_until:
+        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
+        try_until = datetime.now(timezone.utc) + timeout
-        while datetime.now() < try_until:
+        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
+        try_until = datetime.now(timezone.utc) + timeout
-        while datetime.now() < try_until:
+        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
+        try_until = datetime.now(timezone.utc) + timeout
-        while datetime.now() < try_until:
+        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

@@ -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
+        try_until = datetime.now(timezone.utc) + timeout
-        while datetime.now() < try_until:
+        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

@@ -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
+        try_until = datetime.now(timezone.utc) + timeout
-        while datetime.now() < try_until:
+        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
+        try_until = datetime.now(timezone.utc) + timeout
-        while datetime.now() < try_until:
+        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'