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

Reply via email to