From 88f204a0bd67fa753f04ff6bb09e90d9579794bd Mon Sep 17 00:00:00 2001 From: The MMGen Project Date: Mon, 17 Mar 2025 10:16:18 +0000 Subject: [PATCH] cmdtest.py: new WSGI http server framework --- MANIFEST.in | 1 + test/cmdtest_d/httpd/__init__.py | 61 ++++++++++++++++++++++++++++++++ 2 files changed, 62 insertions(+) create mode 100755 test/cmdtest_d/httpd/__init__.py diff --git a/MANIFEST.in b/MANIFEST.in index dd014752..b4b67aac 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -13,6 +13,7 @@ include nix/* include test/*.py include test/*/*.py +include test/*/*/*.py include test/ref/* include test/ref/*/* include test/ref/*/*/*/* diff --git a/test/cmdtest_d/httpd/__init__.py b/test/cmdtest_d/httpd/__init__.py new file mode 100755 index 00000000..32743ee4 --- /dev/null +++ b/test/cmdtest_d/httpd/__init__.py @@ -0,0 +1,61 @@ +#!/usr/bin/env python3 +# +# MMGen Wallet, a terminal-based cryptocurrency wallet +# Copyright (C)2013-2025 The MMGen Project +# Licensed under the GNU General Public License, Version 3: +# https://www.gnu.org/licenses +# Public project repositories: +# https://github.com/mmgen/mmgen-wallet +# https://gitlab.com/mmgen/mmgen-wallet + +""" +test.cmdtest_d.httpd: WSGI http server +""" + +from wsgiref.simple_server import make_server, WSGIRequestHandler + +from mmgen.util import msg +from mmgen.util2 import port_in_use + +class SilentRequestHandler(WSGIRequestHandler): + + def log_request(self, code='-', size='-'): + return None + +class HTTPD: + + def start(self): + + if port_in_use(self.port): + msg(f'\nPort {self.port} in use. Assuming {self.name} is running') + return True + + self.httpd = make_server( + 'localhost', + self.port, + self.application, + handler_class = SilentRequestHandler) + + import threading + t = threading.Thread(target=self.httpd.serve_forever, name=f'{self.name} thread') + t.daemon = True + t.start() + + def stop(self): + self.httpd.server_close() + + def application(self, environ, start_response): + + method = environ['REQUEST_METHOD'] + + response_body = self.make_response_body(method, environ) + + status = '200 OK' + response_headers = [ + ('Content-Type', self.content_type), + ('Content-Length', str(len(response_body))) + ] + + start_response(status, response_headers) + + return [response_body]