from testenv import Env, Nghttpx, Httpd, NghttpxQuic, NghttpxFwd
+log = logging.getLogger(__name__)
+
def pytest_report_header(config):
# Env inits its base properties only once, we can report them here
@pytest.fixture(scope='session')
def nghttpx(env, httpd) -> Generator[Union[Nghttpx,bool], None, None]:
nghttpx = NghttpxQuic(env=env)
- if nghttpx.exists() and env.have_h3():
+ if nghttpx.exists():
+ if not nghttpx.supports_h3() and env.have_h3_curl():
+ log.warning('nghttpx does not support QUIC, but curl does')
nghttpx.clear_logs()
assert nghttpx.initial_start()
yield nghttpx
if m:
earlydata[int(m.group(1))] = int(m.group(2))
continue
- m = re.match(r'\[1-1] \* SSL reusing session.*', line)
- if m:
+ if re.match(r'\[1-1] \* SSL reusing session.*', line):
reused_session = True
assert reused_session, 'session was not reused for 2nd transfer'
assert earlydata[0] == 0, f'{earlydata}'
supported = ['TLSv1', 'TLSv1.1', 'TLSv1.2', 'TLSv1.3']
elif env.curl_uses_lib('aws-lc'):
supported = ['TLSv1', 'TLSv1.1', 'TLSv1.2', 'TLSv1.3']
+ elif env.curl_uses_lib('openssl') and \
+ env.curl_lib_version_before('openssl', '3.0.0'):
+ supported = ['TLSv1', 'TLSv1.1', 'TLSv1.2', 'TLSv1.3']
else: # most SSL backends dropped support for TLSv1.0, TLSv1.1
supported = [None, None, 'TLSv1.2', 'TLSv1.3']
# test
CURL = os.path.join(TOP_PATH, 'src', 'curl')
+class NghttpxUtil:
+
+ CMD = None
+ VERSION_FULL = None
+
+ @classmethod
+ def version(cls, cmd):
+ if cmd is None:
+ return None
+ if cls.VERSION_FULL is None or cmd != cls.CMD:
+ p = subprocess.run(args=[cmd, '--version'],
+ capture_output=True, text=True)
+ if p.returncode != 0:
+ raise RuntimeError(f'{cmd} --version failed with exit code: {p.returncode}')
+ cls.CMD = cmd
+ for line in p.stdout.splitlines(keepends=False):
+ if line.startswith('nghttpx '):
+ cls.VERSION_FULL = line
+ if cls.VERSION_FULL is None:
+ raise RuntimeError(f'{cmd}: unable to determine version')
+ return cls.VERSION_FULL
+
+ @staticmethod
+ def version_with_h3(version):
+ return re.match(r'.* ngtcp2/\d+\.\d+\.\d+.*', version) is not None
+
+
class EnvConfig:
def __init__(self, pytestconfig: Optional[pytest.Config] = None,
self._nghttpx_version = None
self.nghttpx_with_h3 = False
if self.nghttpx is not None:
- p = subprocess.run(args=[self.nghttpx, '-v'],
- capture_output=True, text=True)
- if p.returncode != 0:
+ try:
+ self._nghttpx_version = NghttpxUtil.version(self.nghttpx)
+ self.nghttpx_with_h3 = NghttpxUtil.version_with_h3(self._nghttpx_version)
+ except RuntimeError:
# not a working nghttpx
+ log.exception('checking nghttpx version')
self.nghttpx = None
- else:
- self._nghttpx_version = re.sub(r'^nghttpx\s*', '', p.stdout.strip())
- self.nghttpx_with_h3 = re.match(r'.* nghttp3/.*', p.stdout.strip()) is not None
- log.debug(f'nghttpx -v: {p.stdout}')
self.caddy = self.config['caddy']['caddy']
self._caddy_version = None
Env.CONFIG.versiontuple(lversion)
return False
+ @staticmethod
+ def curl_lib_version_before(libname: str, lib_version) -> bool:
+ lversion = Env.curl_lib_version(libname)
+ if lversion != 'unknown':
+ if m := re.match(r'(\d+\.\d+\.\d+).*', lversion):
+ lversion = m.group(1)
+ return Env.CONFIG.versiontuple(lib_version) > \
+ Env.CONFIG.versiontuple(lversion)
+ return False
+
@staticmethod
def curl_os() -> str:
return Env.CONFIG.curl_props['os']
from typing import Optional, Dict
from datetime import datetime, timedelta
-from .env import Env
+from .env import Env, NghttpxUtil
from .curl import CurlClient
from .ports import alloc_ports_and_do
self._process: Optional[subprocess.Popen] = None
self._cred_name = self._def_cred_name = cred_name
self._loaded_cred_name = ''
- self._rmf(self._pid_file)
- self._rmf(self._error_log)
- self._mkpath(self._run_dir)
- self._write_config()
+ self._version = NghttpxUtil.version(self._cmd)
+
+ def supports_h3(self):
+ return NghttpxUtil.version_with_h3(self._version)
def set_cred_name(self, name: str):
self._cred_name = name
return True
def initial_start(self):
- pass
+ self._rmf(self._pid_file)
+ self._rmf(self._error_log)
+ self._mkpath(self._run_dir)
+ self._write_config()
def start(self, wait_live=True):
pass
self._https_port = env.https_port
def initial_start(self):
+ super().initial_start()
def startup(ports: Dict[str, int]) -> bool:
self._port = ports['nghttpx_https']
creds = self.env.get_credentials(self._cred_name)
assert creds # convince pytype this isn't None
self._loaded_cred_name = self._cred_name
- args = [
- self._cmd,
- f'--frontend=*,{self._port};tls',
- f'--frontend=*,{self.env.h3_port};quic',
- '--frontend-quic-early-data',
+ args = [self._cmd, f'--frontend=*,{self._port};tls']
+ if self.supports_h3():
+ args.extend([
+ f'--frontend=*,{self.env.h3_port};quic',
+ '--frontend-quic-early-data',
+ ])
+ args.extend([
f'--backend=127.0.0.1,{self.env.https_port};{self._domain};sni={self._domain};proto=h2;tls',
f'--backend=127.0.0.1,{self.env.http_port}',
'--log-level=ERROR',
'--frontend-http3-connection-window-size=10M',
'--frontend-http3-max-connection-window-size=100M',
# f'--frontend-quic-debug-log',
- ]
+ ])
ngerr = open(self._stderr, 'a')
self._process = subprocess.Popen(args=args, stderr=ngerr)
if self._process.returncode is not None:
cred_name=env.proxy_domain)
def initial_start(self):
+ super().initial_start()
def startup(ports: Dict[str, int]) -> bool:
self._port = ports['h2proxys']