rectifier: add PUB zmq socket

[loctrkd.git] / test / common.py

diff --git a/test/common.py b/test/common.py
index 38c59d588e72c87e1683cb2c4a3dc9d3853ae392..284ca4d10c8509606dd6879f026c07f211104f83 100644 (file)
--- a/test/common.py
+++ b/test/common.py
@@ -1,8 +1,10 @@
 """ Common housekeeping for tests that rely on daemons """
 
 from configparser import ConfigParser, SectionProxy
 """ 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 importlib import import_module

+from logging import DEBUG, StreamHandler

 from multiprocessing import Process
 from os import kill, unlink
 from signal import SIGINT
 from multiprocessing import Process
 from os import kill, unlink
 from signal import SIGINT


@@ -15,24 +17,31 @@ from socket import (
     socket,
     SocketType,
 )
     socket,
     SocketType,
 )

+from sys import exit, stderr

 from tempfile import mkstemp
 from time import sleep
 from typing import Optional
 from unittest import TestCase
 
 from tempfile import mkstemp
 from time import sleep
 from typing import Optional
 from unittest import TestCase
 

+NUMPORTS = 3
+

 
 class TestWithServers(TestCase):
 
 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.tmpfilebase = mkstemp()
         self.conf = ConfigParser()

+        self.conf["common"] = {
+            "protocols": "zx303proto",
+        }

         self.conf["collector"] = {
             "port": str(freeports[0]),
             "publishurl": "ipc://" + self.tmpfilebase + ".pub",
         self.conf["collector"] = {
             "port": str(freeports[0]),
             "publishurl": "ipc://" + self.tmpfilebase + ".pub",


@@ -43,9 +52,11 @@ class TestWithServers(TestCase):
         }
         self.conf["opencellid"] = {
             "dbfn": self.tmpfilebase + ".opencellid.sqlite",
         }
         self.conf["opencellid"] = {
             "dbfn": self.tmpfilebase + ".opencellid.sqlite",

+            "downloadurl": f"http://localhost:{freeports[2]}/test/262.csv.gz",

         }
         }

-        self.conf["lookaside"] = {
-            "backend": "opencellid",
+        self.conf["rectifier"] = {
+            "lookaside": "opencellid",
+            "publishurl": "ipc://" + self.tmpfilebase + ".rect.pub",

         }
         self.conf["wsgateway"] = {
             "port": str(freeports[1]),
         }
         self.conf["wsgateway"] = {
             "port": str(freeports[1]),


@@ -56,10 +67,26 @@ class TestWithServers(TestCase):
                 kwargs = {"handle_hibernate": False}
             else:
                 kwargs = {}
                 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))
             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:
         sleep(1)
 
     def tearDown(self) -> None:


@@ -70,11 +97,12 @@ class TestWithServers(TestCase):
             self.assertEqual(
                 p.exitcode,
                 0,
             self.assertEqual(
                 p.exitcode,
                 0,

-                srvname + " terminated with non-zero return code",
+                f"{srvname} terminated with return code {p.exitcode}",

             )
         for sfx in (
             "",
             ".pub",
             )
         for sfx in (
             "",
             ".pub",

+            ".rect.pub",

             ".pul",
             ".storage.sqlite",
             ".opencellid.sqlite",
             ".pul",
             ".storage.sqlite",
             ".opencellid.sqlite",