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 -- [email protected]
To unsubscribe send an email to [email protected]
