test: Add a unit test for non default handlers

* Test will check if registered non default handlers are called

* Cast port number to int in simplehttpserver.py script

Signed-off-by: Mahendra Paipuri <[email protected]>

tests/child_scripts/signalprinter.py

@@ -3,17 +3,29 @@
 """
 import asyncio
 import sys
+import signal
 from functools import partial
 
 from simpervisor.atexitasync import add_handler
 
 
+def _non_default_handle(sig, frame):
+    # Print the received signum and then exit
+    print(f"non default handler received {sig}", flush=True)
+    sys.exit(0)
+
+
 def _handle_sigterm(number, received_signum):
     # Print the received signum so we know our handler was called
     print(f"handler {number} received", int(received_signum), flush=True)
 
 
 handlercount = int(sys.argv[1])
+# Add non default handler if arg true is passed
+if len(sys.argv) == 3:
+    if bool(sys.argv[2]):
+        signal.signal(signal.SIGINT, _non_default_handle)
+        signal.signal(signal.SIGTERM, _non_default_handle)
 for i in range(handlercount):
     add_handler(partial(_handle_sigterm, i))
 

tests/child_scripts/simplehttpserver.py

@@ -23,4 +23,4 @@ async def hello(request):
 
 app = web.Application()
 app.add_routes(routes)
-web.run_app(app, port=PORT)
+web.run_app(app, port=int(PORT))

tests/test_atexitasync.py

@@ -49,3 +49,48 @@ def test_atexitasync(signum, handlercount):
     # The code should exit cleanly
     retcode = proc.wait()
     assert retcode == 0
+
+
+@pytest.mark.parametrize(
+    "signum, handlercount",
+    [
+        (signal.SIGTERM, 1),
+        (signal.SIGINT, 1),
+        (signal.SIGTERM, 5),
+        (signal.SIGINT, 5),
+    ],
+)
+@pytest.mark.skipif(
+    sys.platform == "win32",
+    reason="Testing signals on Windows doesn't seem to be possible",
+)
+def test_atexitasync_with_non_default_handlers(signum, handlercount):
+    """
+    Test signal handlers receive signals properly and handler existing default handlers
+    correctly
+    """
+    signalprinter_file = os.path.join(
+        os.path.dirname(os.path.abspath(__file__)), "child_scripts", "signalprinter.py"
+    )
+    proc = subprocess.Popen(
+        [sys.executable, signalprinter_file, str(handlercount), "true"],
+        stdout=subprocess.PIPE,
+        text=True,
+    )
+
+    # Give the process time to register signal handlers
+    time.sleep(1)
+    proc.send_signal(signum)
+
+    # Make sure the signal is handled by our handler in the code
+    stdout, stderr = proc.communicate()
+    expected_output = (
+        "\n".join([f"handler {i} received {signum}" for i in range(handlercount)])
+        + f"\nnon default handler received {signum}\n"
+    )
+
+    assert stdout == expected_output
+
+    # The code should exit cleanly
+    retcode = proc.wait()
+    assert retcode == 0