from contextlib import closing, ExitStack
from http.server import HTTPServer, SimpleHTTPRequestHandler
from importlib import import_module
+from logging import DEBUG, StreamHandler
from multiprocessing import Process
from os import kill, unlink
from signal import SIGINT
socket,
SocketType,
)
-from sys import exit
+from sys import exit, stderr
from tempfile import mkstemp
from time import sleep
from typing import Optional
class TestWithServers(TestCase):
- def setUp(self, *args: str, httpd: bool = False) -> None:
+ def setUp(
+ self, *args: str, httpd: bool = False, verbose: bool = False
+ ) -> None:
freeports = []
with ExitStack() as stack:
for _ in range(NUMPORTS):
"port": str(freeports[0]),
"publishurl": "ipc://" + self.tmpfilebase + ".pub",
"listenurl": "ipc://" + self.tmpfilebase + ".pul",
+ "protocols": "zx303proto",
}
self.conf["storage"] = {
"dbfn": self.tmpfilebase + ".storage.sqlite",
else:
kwargs = {}
cls = import_module("gps303." + srvname, package=".")
+ if verbose:
+ cls.log.addHandler(StreamHandler(stderr))
+ cls.log.setLevel(DEBUG)
p = Process(target=cls.runserver, args=(self.conf,), kwargs=kwargs)
p.start()
self.children.append((srvname, p))