test/common.py

   1 """ Common housekeeping for tests that rely on daemons """
   2
   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
  13
  14
  15 class TestWithServers(TestCase):
  16     def setUp(self, *args: str) -> None:
  17         with closing(socket(AF_INET6, SOCK_DGRAM)) as sock:
  18             sock.bind(("", 0))
  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",
  27         }
  28         self.children = []
  29         for srvname in args:
  30             if srvname == "collector":
  31                 kwargs = {"handle_hibernate": False}
  32             else:
  33                 kwargs = {}
  34             cls = import_module("gps303." + srvname, package=".")
  35             p = Process(target=cls.runserver, args=(self.conf,), kwargs=kwargs)
  36             p.start()
  37             self.children.append((srvname, p))
  38         sleep(1)
  39
  40     def tearDown(self) -> None:
  41         sleep(1)
  42         for srvname, p in self.children:
  43             if p.pid is not None:
  44                 kill(p.pid, SIGINT)
  45             p.join()
  46             print(srvname, "terminated with return code", p.exitcode)
  47         for sfx in (".pub", ".pul"):
  48             unlink(self.tmpfilebase + sfx)