romenskiy2012/curl
1
0
Fork 0
mirror of https://github.com/curl/curl.git synced 2026-07-29 03:43:09 +03:00
Code Issues Projects Releases Packages Wiki Activity

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

Browse source 
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.
This commit is contained in:
Dan Fandrich 2026-07-25 13:23:21 -07:00
parent b151a0bb90
commit e13362c20a
13 changed files with 81 additions and 78 deletions
Download patch file Download diff file Expand all files Collapse all files

11
tests/http/test_17_ssl_use.py
View file

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

6
tests/http/test_20_websockets.py
View file

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

14
tests/http/testenv/caddy.py
View file

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

6
tests/http/testenv/certs.py
View file

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

6
tests/http/testenv/client.py
View file

@ -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 = []

18
tests/http/testenv/curl.py
View file

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

6
tests/http/testenv/dante.py
View file

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

6
tests/http/testenv/dnsd.py
View file

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

20
tests/http/testenv/h2o.py
View file

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

22
tests/http/testenv/httpd.py
View file

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

28
tests/http/testenv/nghttpx.py
View file

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

6
tests/http/testenv/sshd.py
View file

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

10
tests/http/testenv/vsftpd.py
View file

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