Changeset: 61f71d59ed8d for MonetDB URL: https://dev.monetdb.org/hg/MonetDB/rev/61f71d59ed8d Modified Files: clients/mapilib/Tests/tlssecurity.py testing/tlstester.py Branch: Aug2024 Log Message:
Move TLSTesterClient from tlssecurity,py to testing/tlstester.py So we can use it from other tests as well diffs (129 lines): diff --git a/clients/mapilib/Tests/tlssecurity.py b/clients/mapilib/Tests/tlssecurity.py --- a/clients/mapilib/Tests/tlssecurity.py +++ b/clients/mapilib/Tests/tlssecurity.py @@ -14,9 +14,8 @@ import os import re import subprocess import sys -import urllib.request -from MonetDBtesting import tlstester +from MonetDBtesting.tlstester import TLSTesterClient level = logging.WARNING # if sys.platform == 'win32': @@ -32,50 +31,6 @@ assert os.path.isdir(tgtdir) scratchdir = os.path.join(tgtdir, "scratch") logging.debug(f"scratchdir={scratchdir}") -class TLSTesterClient: - """Connect to TLSTester to figure out port numbers and download certificates""" - def __init__(self, scratchdir, base_port=None, host='localhost'): - if not base_port: - base_port = os.environ['TST_TLSTESTERPORT'] - self.url = f'http://{host}:{base_port}/' - self.scratch = scratchdir - try: - os.mkdir(scratchdir) - except FileExistsError: - pass - self.filenames = dict() - self.contents = dict() - self.portmap = dict() - for line in self.fetch('').splitlines(): - name, port = str(line, 'ascii').split(':', 1) - self.portmap[name] = int(port) - logging.debug(f'port {name} = {port}') - - def get_port(self, name): - return self.portmap[name] - - def fetch(self, name): - cached = self.contents.get(name) - if cached is not None: - return cached - url = self.url + name - logging.debug(f'fetch {url}') - with urllib.request.urlopen(url) as response: - content = response.read() - self.contents[name] = content - return content - - def download(self, name): - cached = self.filenames.get(name) - if cached: - return cached - content = self.fetch(name) - path = os.path.join(self.scratch, name) - with open(path, 'wb') as f: - f.write(content) - self.filenames[name] = path - return path - tlstester = TLSTesterClient(scratchdir) diff --git a/testing/tlstester.py b/testing/tlstester.py --- a/testing/tlstester.py +++ b/testing/tlstester.py @@ -27,6 +27,7 @@ import tempfile from threading import Thread import threading from typing import Any, Callable, Dict, List, Optional, Tuple, Union +import urllib.request # Our TLS implementation never uses anything less than TLSv1.3. assert ssl.HAS_TLSv1_3 @@ -98,6 +99,52 @@ argparser.add_argument( ) + +class TLSTesterClient: + """Connect to TLSTester to figure out port numbers and download certificates""" + def __init__(self, scratchdir, base_port=None, host='localhost'): + if not base_port: + base_port = os.environ['TST_TLSTESTERPORT'] + self.url = f'http://{host}:{base_port}/' + self.scratch = scratchdir + try: + os.mkdir(scratchdir) + except FileExistsError: + pass + self.filenames = dict() + self.contents = dict() + self.portmap = dict() + for line in self.fetch('').splitlines(): + name, port = str(line, 'ascii').split(':', 1) + self.portmap[name] = int(port) + logging.debug(f'port {name} = {port}') + + def get_port(self, name): + return self.portmap[name] + + def fetch(self, name): + cached = self.contents.get(name) + if cached is not None: + return cached + url = self.url + name + logging.debug(f'fetch {url}') + with urllib.request.urlopen(url) as response: + content = response.read() + self.contents[name] = content + return content + + def download(self, name): + cached = self.filenames.get(name) + if cached: + return cached + content = self.fetch(name) + path = os.path.join(self.scratch, name) + with open(path, 'wb') as f: + f.write(content) + self.filenames[name] = path + return path + + class Certs: hostnames: str _files: Dict[str, bytes] _______________________________________________ checkin-list mailing list -- checkin-list@monetdb.org To unsubscribe send an email to checkin-list-le...@monetdb.org