test: add fuzzer for beesure protocol

[loctrkd.git] / test / common.py

diff --git a/test/common.py b/test/common.py
index 5f6c753fd0ae6f5863b29d41f1e76d489b8e865e..a31ff094d9d806ebfa5b1572d49587a2a20c7bd8 100644 (file)
--- a/test/common.py
+++ b/test/common.py
@@ -1,53 +1,103 @@
 """ 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 socket import (
 from multiprocessing import Process
 from os import kill, unlink
 from signal import SIGINT
 from socket import (

+    AF_INET,

     AF_INET6,
     AF_INET6,

+    getaddrinfo,

     MSG_DONTWAIT,
     SOCK_DGRAM,
     MSG_DONTWAIT,
     SOCK_DGRAM,

+    SOCK_STREAM,

     SOL_SOCKET,
     SO_REUSEADDR,
     socket,
     SocketType,
 )
     SOL_SOCKET,
     SO_REUSEADDR,
     socket,
     SocketType,
 )

+from sys import exit, stderr
+from random import Random

 from tempfile import mkstemp
 from time import sleep
 from tempfile import mkstemp
 from time import sleep

-from typing import Optional
+from typing import Any, Optional

 from unittest import TestCase
 
 from unittest import TestCase
 

+from loctrkd.common import init_protocols
+
+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, 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,beesure",
+        }

         self.conf["collector"] = {
         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",
+            "events": "yes",
+        }
+        self.conf["opencellid"] = {
+            "dbfn": self.tmpfilebase + ".opencellid.sqlite",
+            "downloadurl": f"http://localhost:{freeports[2]}/test/262.csv.gz",
+        }
+        self.conf["rectifier"] = {
+            "lookaside": "opencellid",
+            "publishurl": "ipc://" + self.tmpfilebase + ".rect.pub",
+        }
+        self.conf["wsgateway"] = {
+            "port": str(freeports[1]),
+        }
+        init_protocols(self.conf)

         self.children = []
         for srvname in args:
             if srvname == "collector":
                 kwargs = {"handle_hibernate": False}
             else:
                 kwargs = {}
         self.children = []
         for srvname in args:
             if srvname == "collector":
                 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:

-        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)


@@ -55,10 +105,42 @@ 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", ".pul"):
-            unlink(self.tmpfilebase + sfx)
+        for sfx in (
+            "",
+            ".pub",
+            ".rect.pub",
+            ".pul",
+            ".storage.sqlite",
+            ".opencellid.sqlite",
+        ):
+            try:
+                unlink(self.tmpfilebase + sfx)
+            except OSError:
+                pass
+
+
+class Fuzz(TestWithServers):
+    def setUp(self, *args: str, **kwargs: Any) -> None:
+        super().setUp("collector")
+        self.rnd = Random()
+        for fam, typ, pro, cnm, skadr in getaddrinfo(
+            "127.0.0.1",
+            self.conf.getint("collector", "port"),
+            family=AF_INET,
+            type=SOCK_STREAM,
+        ):
+            break  # Just take the first element
+        self.sock = socket(AF_INET, SOCK_STREAM)
+        self.sock.connect(skadr)
+
+    def tearDown(self) -> None:
+        sleep(1)  # give collector some time
+        send_and_drain(self.sock, None)
+        self.sock.close()
+        sleep(1)  # Let the server close their side
+        super().tearDown()

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