1 """ Common housekeeping for tests that rely on daemons """
3 from configparser import ConfigParser, SectionProxy
4 from contextlib import closing
5 from importlib import import_module
6 from multiprocessing import Process
7 from os import kill, unlink
8 from signal import SIGINT
9 from socket import AF_INET6, SOCK_DGRAM, SOL_SOCKET, SO_REUSEADDR, socket
10 from tempfile import mkstemp
11 from time import sleep
12 from unittest import TestCase
15 class TestWithServers(TestCase):
16 def setUp(self, *args: str) -> None:
17 with closing(socket(AF_INET6, SOCK_DGRAM)) as sock:
19 sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
20 freeport = sock.getsockname()[1]
21 _, self.tmpfilebase = mkstemp()
22 self.conf = ConfigParser()
23 self.conf["collector"] = {
24 "port": str(freeport),
25 "publishurl": "ipc://" + self.tmpfilebase + ".pub",
26 "listenurl": "ipc://" + self.tmpfilebase + ".pul",
30 if srvname == "collector":
31 kwargs = {"handle_hibernate": False}
34 cls = import_module("gps303." + srvname, package=".")
35 p = Process(target=cls.runserver, args=(self.conf,), kwargs=kwargs)
37 self.children.append((srvname, p))
40 def tearDown(self) -> None:
42 for srvname, p in self.children:
46 print(srvname, "terminated with return code", p.exitcode)
47 for sfx in (".pub", ".pul"):
48 unlink(self.tmpfilebase + sfx)