Add test servers

scripts/https-bff.py

@@ -0,0 +1,61 @@
+from http.server import HTTPServer, SimpleHTTPRequestHandler
+import logging
+import os
+from pathlib import Path
+import ssl
+import sys
+
+LOGGER = logging.getLogger(__package__)
+
+
+class SecureHTTPRequestHandler(SimpleHTTPRequestHandler):
+    """Überschreibt die Methode, um das Basisverzeichnis anzupassen."""
+
+    def translate_path(self, path):
+        """Basisverzeichnis für Dateien."""
+        base_directory = "../local"
+
+        # Anhängen von ".json", wenn der Path keine Dateiendung enthält
+        if not Path(path).suffix:  # Kein "." in der Datei
+            path += ".json"
+
+        # Normalisierung des Pfads
+        path = super().translate_path(path)
+
+        # Ersetzen des Basisverzeichnisses durch das angegebene
+        relative_path = os.path.relpath(path, Path.cwd())
+        LOGGER.debug("request %s", Path(base_directory) / relative_path)
+        return Path(base_directory) / relative_path
+
+
+def set_logger():
+    """Set Logger properties."""
+
+    fmt = "%(asctime)s.%(msecs)03d %(levelname)s (%(threadName)s) [%(name)s] %(message)s"
+    LOGGER.setLevel(logging.DEBUG)
+
+    handler = logging.StreamHandler(sys.stdout)
+    handler.setLevel(logging.DEBUG)
+    formatter = logging.Formatter(fmt)
+    handler.setFormatter(formatter)
+    LOGGER.addHandler(handler)
+
+
+def start_server(host="127.0.0.1", port=8002, certfile="../local/selfsigned.crt", keyfile="../local/selfsigned.key"):
+    """HTTPS-Server mit dem spezifizierten Handler."""
+    httpd = HTTPServer((host, port), SecureHTTPRequestHandler)
+
+    # SSL-Kontext erstellen
+    ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    ssl_context.load_cert_chain(certfile=certfile, keyfile=keyfile)
+
+    # Socket mit SSL-Kontext umschließen
+    httpd.socket = ssl_context.wrap_socket(httpd.socket, server_side=True)
+
+    LOGGER.debug("HTTPS-Server gestartet auf https://%s:%s", host, port)
+    LOGGER.debug("Daten werden aus dem Verzeichnis ../local geladen.")
+    httpd.serve_forever()
+
+
+if __name__ == "__main__":
+    start_server()

scripts/https-ws-case-429.py

@@ -2,49 +2,31 @@
 
 from __future__ import annotations
 
+from http.server import BaseHTTPRequestHandler, HTTPServer
 import logging
 import ssl
 import sys
-from http.server import BaseHTTPRequestHandler, HTTPServer
 from urllib.parse import urlparse
 
 HTTP_SERVER_IP = "0.0.0.0"
 HTTP_SERVER_PORT = 8001
 LOGGER = logging.getLogger(__package__)
 
-THROW_429 = False
-
 
 class MBAPI2020SimulatorServer(BaseHTTPRequestHandler):
     """Simple HTTP Server to simulate the MBAPI2020 API."""
 
-    throw_429 = THROW_429
-
     def do_GET(self):
         """Answer get requests."""
 
         parsed = urlparse(self.path)
-        if parsed.path == "toggle429":
-            self.throw_429 = not self.throw_429
-            self.send_response(200)
-            self.send_header("Content-type", "text/plain")
+        if parsed.path == "/v2/ws":
+            self.send_response(429)
+            self.send_header("Content-type", "")
             self.end_headers()
-            self.wfile.write(bytes(f"ok throw 429: {self.throw_429}", "utf-8"))
+            self.wfile.write("".encode("utf-8"))
             return
 
-        if parsed.path == "/v2/ws":
-            if self.throw_429:
-                self.send_response(429)
-                self.send_header("Content-type", "")
-                self.end_headers()
-                self.wfile.write("".encode("utf-8"))
-                return
-            else:
-                self.send_response(200)
-                self.send_header("Content-type", "")
-                self.end_headers()
-                self.wfile.write("ok".encode("utf-8"))
-
     def do_POST(self):
         """Answer post requests."""
         self.do_GET()

scripts/ws-simulator.py

@@ -0,0 +1,66 @@
+import asyncio
+from http import HTTPStatus
+import ssl
+
+from websockets.asyncio.server import serve
+
+file1_path = "../local/asv"
+file2_path = "../local/vep"
+
+throw_429 = True
+
+
+def toggle(connection, request):
+    global throw_429
+    if request.path == "/toggle":
+        throw_429 = not throw_429
+        if throw_429:
+            return connection.respond(HTTPStatus.OK, f"OK {throw_429}\n")
+        else:
+            return connection.respond(HTTPStatus.IM_A_TEAPOT, f"\n")
+
+    if throw_429:
+        return connection.respond(HTTPStatus.TOO_MANY_REQUESTS, "\n")
+
+
+async def send_files(websocket):
+    try:
+        async for message in websocket:
+            print(message)
+            return
+
+        # Datei 1 lesen
+        with open(file1_path, "rb") as file1:
+            file1_content = file1.read()
+
+        # Datei 2 lesen
+        with open(file2_path, "rb") as file2:
+            file2_content = file2.read()
+
+        # Nachricht an den Client senden
+        await websocket.send(file1_content)
+        print("file 1 sent")
+        asyncio.sleep(2)
+        await websocket.send(file2_content)
+        print("file 2 sent")
+
+    except FileNotFoundError as e:
+        error_message = f"Fehler: Datei nicht gefunden - {e.filename}"
+        print(error_message)
+        await websocket.send(error_message)
+
+    except Exception as e:
+        error_message = f"Ein Fehler ist aufgetreten: {e}"
+        print(error_message)
+        await websocket.send(error_message)
+
+
+async def main():
+    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    context.load_cert_chain(certfile="../local/selfsigned.crt", keyfile="../local/selfsigned.key")
+    context.check_hostname = False
+    async with serve(send_files, "localhost", 8001, process_request=toggle, ssl=context):
+        await asyncio.get_running_loop().create_future()  # run forever
+
+
+asyncio.run(main())