collector: do not remove clients before all sends

[loctrkd.git] / test / common.py

diff --git a/test/common.py b/test/common.py
index 5f6c753fd0ae6f5863b29d41f1e76d489b8e865e..2b45c53400cc2ab363752f4dd5a02eee9ce714e8 100644 (file)
--- a/test/common.py
+++ b/test/common.py
@@ -1,7 +1,8 @@
 """ 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 multiprocessing import Process
 from os import kill, unlink
 from importlib import import_module
 from multiprocessing import Process
 from os import kill, unlink


@@ -15,25 +16,44 @@ from socket import (
     socket,
     SocketType,
 )
     socket,
     SocketType,
 )

+from sys import exit

 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 sock:
-            sock.bind(("", 0))
-            sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
-            freeport = sock.getsockname()[1]
+    def setUp(self, *args: str, httpd: 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"] = {
         _, self.tmpfilebase = mkstemp()
         self.conf = ConfigParser()
         self.conf["collector"] = {

-            "port": str(freeport),
+            "port": str(freeports[0]),

             "publishurl": "ipc://" + self.tmpfilebase + ".pub",
             "listenurl": "ipc://" + self.tmpfilebase + ".pul",
         }
             "publishurl": "ipc://" + self.tmpfilebase + ".pub",
             "listenurl": "ipc://" + self.tmpfilebase + ".pul",
         }

+        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",
+        }
+        self.conf["wsgateway"] = {
+            "port": str(freeports[1]),
+        }

         self.children = []
         for srvname in args:
             if srvname == "collector":
         self.children = []
         for srvname in args:
             if srvname == "collector":


@@ -44,10 +64,22 @@ class TestWithServers(TestCase):
             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:

-        sleep(1)

         for srvname, p in self.children:
             if p.pid is not None:
                 kill(p.pid, SIGINT)
         for srvname, p in self.children:
             if p.pid is not None:
                 kill(p.pid, SIGINT)


@@ -57,8 +89,17 @@ class TestWithServers(TestCase):
                 0,
                 srvname + " terminated with non-zero return code",
             )
                 0,
                 srvname + " terminated with non-zero return code",
             )

-        for sfx in (".pub", ".pul"):
-            unlink(self.tmpfilebase + sfx)
+        for sfx in (
+            "",
+            ".pub",
+            ".pul",
+            ".storage.sqlite",
+            ".opencellid.sqlite",
+        ):
+            try:
+                unlink(self.tmpfilebase + sfx)
+            except OSError:
+                pass

 
 
 def send_and_drain(sock: SocketType, buf: Optional[bytes]) -> None:
 
 
 def send_and_drain(sock: SocketType, buf: Optional[bytes]) -> None: