reduce duplication in testserver implementations

testdata/generic_keys_testserver.py

@@ -1,57 +1,32 @@
 # -*- coding: utf-8 -*-
-import http.server
 import json
-import socketserver
 
+from testserver import TestServerHandler, launchServerWith
 
-class GenericKeysHandler(http.server.BaseHTTPRequestHandler):
-    def do_POST(self):
-        content_length = int(self.headers["Content-Length"])
-        raw_data = self.rfile.read(content_length)
 
-        try:
+class GenericKeysHandler(TestServerHandler):
-            # Decode received data
+    def transform_data(self, raw_data):
+        data = json.loads(raw_data)
 
-            print(raw_data)
+        data["numeric"]["0000"] = {
-            data = json.loads(raw_data)
+            "lorem": "ipsum",
+            "dolor": "sit",
+            "amet": "consectetur",
+        }
+        data["uuid"]["123e4567-e89b-12d3-a456-426614174002"] = {
+            "lorem": "ipsum",
+            "dolor": "sit",
+            "amet": "consectetur",
+        }
+        data["mixed"]["0000"] = {
+            "lorem": "ipsum",
+            "dolor": "sit",
+            "amet": "consectetur",
+        }
 
-            data["numeric"]["0000"] = {
+        # Encode the modified data
-                "lorem": "ipsum",
+        return bytes(json.dumps(data), "utf-8")
-                "dolor": "sit",
-                "amet": "consectetur",
-            }
-            data["uuid"]["123e4567-e89b-12d3-a456-426614174002"] = {
-                "lorem": "ipsum",
-                "dolor": "sit",
-                "amet": "consectetur",
-            }
-            data["mixed"]["0000"] = {
-                "lorem": "ipsum",
-                "dolor": "sit",
-                "amet": "consectetur",
-            }
-
-            # Encode the modified data
-            modified_data = bytes(json.dumps(data), "utf-8")
-
-            # Send the response
-            self.send_response(200)
-            self.send_header("Content-type", "application/json")
-            self.send_header("Content-length", len(modified_data))
-            self.end_headers()
-            self.wfile.write(modified_data)
-
-        except Exception as e:
-            print(f"Error processing request: {str(e)}")
-            self.send_response(500)
-            self.send_header("Content-type", "text/plain")
-            self.end_headers()
-            self.wfile.write(f"Error processing request: {str(e)}".encode())
 
 
 if __name__ == "__main__":
-    PORT = 8082
+    launchServerWith(GenericKeysHandler)
-
-    with socketserver.TCPServer(("", PORT), GenericKeysHandler) as httpd:
-        print(f"Serving on port {PORT}")
-        httpd.serve_forever()

testdata/msgpack_testserver.py

@@ -1,45 +1,18 @@
 # -*- coding: utf-8 -*-
-import http.server
-import socketserver
-
 import msgpack
+from testserver import TestServerHandler, launchServerWith
 
 
-class MessagePackHandler(http.server.BaseHTTPRequestHandler):
+class MessagePackHandler(TestServerHandler):
-    def do_POST(self):
+    def transform_data(self, raw_data):
-        content_length = int(self.headers["Content-Length"])
+        data = msgpack.unpackb(raw_data, raw=False)
-        raw_data = self.rfile.read(content_length)
-
-        try:
-            # Decode received MessagePack data
-
-            print(raw_data)
-            data = msgpack.unpackb(raw_data, raw=False)
 
-            # Add a new field to the data
+        # Add a new field to the data
-            data["new_field"] = "Added Field"
+        data["new_field"] = "Added Field"
 
-            # Encode the modified data as MessagePack
+        # Encode the modified data as MessagePack
-            modified_data = msgpack.packb(data, use_bin_type=True)
+        return msgpack.packb(data, use_bin_type=True)
-
-            # Send the response
-            self.send_response(200)
-            self.send_header("Content-type", "application/msgpack")
-            self.send_header("Content-length", len(modified_data))
-            self.end_headers()
-            self.wfile.write(modified_data)
-
-        except Exception as e:
-            print(f"Error processing request: {str(e)}")
-            self.send_response(500)
-            self.send_header("Content-type", "text/plain")
-            self.end_headers()
-            self.wfile.write(f"Error processing request: {str(e)}".encode())
 
 
 if __name__ == "__main__":
-    PORT = 8082
+    launchServerWith(MessagePackHandler)
-
-    with socketserver.TCPServer(("", PORT), MessagePackHandler) as httpd:
-        print(f"Serving on port {PORT}")
-        httpd.serve_forever()

testdata/testserver.py

@@ -0,0 +1,40 @@
+# -*- coding: utf-8 -*-
+import http.server
+import socketserver
+from typing import Type
+
+
+class TestServerHandler(http.server.BaseHTTPRequestHandler):
+    def do_POST(self):
+        content_length = int(self.headers["Content-Length"])
+        raw_data = self.rfile.read(content_length)
+
+        try:
+            # Decode received data
+
+            print(raw_data)
+            modified_data = self.transform_data(raw_data)
+
+            # Send the response
+            self.send_response(200)
+            self.send_header("Content-type", self.headers["Content-type"])
+            self.send_header("Content-length", len(modified_data))
+            self.end_headers()
+            self.wfile.write(modified_data)
+
+        except Exception as e:
+            print(f"Error processing request: {str(e)}")
+            self.send_response(500)
+            self.send_header("Content-type", "text/plain")
+            self.end_headers()
+            self.wfile.write(f"Error processing request: {str(e)}".encode())
+
+    def transform_data(self, raw_data):
+        raise NotImplementedError("Subclasses must implement this method")
+
+
+def launchServerWith(handler: Type[TestServerHandler]):
+    PORT = 8082
+    with socketserver.TCPServer(("", PORT), handler) as httpd:
+        print(f"Serving on port {PORT}")
+        httpd.serve_forever()