""" Common housekeeping for tests that rely on daemons """
from configparser import ConfigParser, SectionProxy
-from contextlib import closing
+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, stderr
from tempfile import mkstemp
from time import sleep
from typing import Optional
from unittest import TestCase
+NUMPORTS = 3
+
class TestWithServers(TestCase):
- def setUp(self, *args: str) -> None:
- with closing(socket(AF_INET6, SOCK_DGRAM)) as sock1, closing(
- socket(AF_INET6, SOCK_DGRAM)
- ) as sock2:
- freeports = []
- for sock in sock1, sock2:
- sock.bind(("", 0))
- sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
- freeports.append(sock.getsockname()[1])
+ def setUp(
+ self, *args: str, httpd: bool = False, verbose: bool = False
+ ) -> None:
+ freeports = []
+ with ExitStack() as stack:
+ for _ in range(NUMPORTS):
+ sk = stack.enter_context(closing(socket(AF_INET6, SOCK_DGRAM)))
+ sk.bind(("", 0))
+ sk.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
+ freeports.append(sk.getsockname()[1])
_, self.tmpfilebase = mkstemp()
self.conf = ConfigParser()
self.conf["collector"] = {
"port": str(freeports[0]),
"publishurl": "ipc://" + self.tmpfilebase + ".pub",
"listenurl": "ipc://" + self.tmpfilebase + ".pul",
+ "protocols": "zx303proto",
}
self.conf["storage"] = {
"dbfn": self.tmpfilebase + ".storage.sqlite",
}
self.conf["opencellid"] = {
"dbfn": self.tmpfilebase + ".opencellid.sqlite",
+ "downloadurl": f"http://localhost:{freeports[2]}/test/262.csv.gz",
}
self.conf["lookaside"] = {
"backend": "opencellid",
kwargs = {"handle_hibernate": False}
else:
kwargs = {}
- cls = import_module("gps303." + srvname, package=".")
+ cls = import_module("loctrkd." + 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))
+ if httpd:
+ server = HTTPServer(("", freeports[2]), SimpleHTTPRequestHandler)
+
+ def run(server: HTTPServer) -> None:
+ try:
+ server.serve_forever()
+ except KeyboardInterrupt:
+ # TODO: this still leaves unclosed socket in the server
+ server.shutdown()
+
+ p = Process(target=run, args=(server,))
+ p.start()
+ self.children.append(("httpd", p))
sleep(1)
def tearDown(self) -> None:
self.assertEqual(
p.exitcode,
0,
- srvname + " terminated with non-zero return code",
+ f"{srvname} terminated with return code {p.exitcode}",
)
for sfx in (
"",