add restart tests

tests/apps/restart.py

@@ -0,0 +1,17 @@
+import json
+import os
+
+
+def pid(environ, protocol):
+    protocol('200 OK', [('content-type', 'text/plain; charset=utf-8')])
+    return [
+        json.dumps(
+            {
+                'pid': os.getpid(),
+            }
+        ).encode('utf8')
+    ]
+
+
+def app(environ, protocol):
+    return {'/pid': pid}[environ['PATH_INFO']](environ, protocol)

tests/conftest.py

@@ -10,19 +10,21 @@
 from granian import Granian
 
 
-def _serve(**kwargs):
-    server = Granian(f'tests.apps.{kwargs["interface"]}:app', **kwargs)
+def _serve(app, **kwargs):
+    server = Granian(f'tests.apps.{app}:app', **kwargs)
     server.serve()
 
 
 @asynccontextmanager
-async def _server(interface, port, threading_mode, tls=False):
+async def _server(interface, app, port, threading_mode, tls=False, extra_args=None):
     certs_path = Path.cwd() / 'tests' / 'fixtures' / 'tls'
     kwargs = {
         'interface': interface,
         'port': port,
         'threading_mode': threading_mode,
     }
+    if extra_args:
+        kwargs.update(extra_args)
     if tls:
         if tls == 'private':
             kwargs['ssl_cert'] = certs_path / 'pcert.pem'
@@ -34,9 +36,8 @@ async def _server(interface, port, threading_mode, tls=False):
 
     succeeded, spawn_failures = False, 0
     while spawn_failures < 3:
-        proc = mp.get_context('spawn').Process(target=_serve, kwargs=kwargs)
+        proc = mp.get_context('spawn').Process(target=_serve, args=(app,), kwargs=kwargs)
         proc.start()
-
         conn_failures = 0
         while conn_failures < 3:
             try:
@@ -74,24 +75,29 @@ def server_port():
 
 @pytest.fixture(scope='function')
 def asgi_server(server_port):
-    return partial(_server, 'asgi', server_port)
+    return partial(_server, 'asgi', 'asgi', server_port)
 
 
 @pytest.fixture(scope='function')
 def rsgi_server(server_port):
-    return partial(_server, 'rsgi', server_port)
+    return partial(_server, 'rsgi', 'rsgi', server_port)
 
 
 @pytest.fixture(scope='function')
 def wsgi_server(server_port):
-    return partial(_server, 'wsgi', server_port)
+    return partial(_server, 'wsgi', 'wsgi', server_port)
 
 
 @pytest.fixture(scope='function')
 def server(server_port, request):
-    return partial(_server, request.param, server_port)
+    return partial(_server, request.param, request.param, server_port)
 
 
 @pytest.fixture(scope='function')
 def server_tls(server_port, request):
-    return partial(_server, request.param, server_port, tls=True)
+    return partial(_server, request.param, request.param, server_port, tls=True)
+
+
+@pytest.fixture(scope='function')
+def server_app(server_port):
+    return partial(_server, port=server_port)

tests/test_restart.py

@@ -0,0 +1,84 @@
+import os
+import platform
+import signal
+import tempfile
+import time
+from pathlib import Path
+
+import httpx
+import pytest
+
+
+def _wait_for_new_pid(port: int, old_pids):
+    for retry in range(1, 5):
+        res = httpx.get(f'http://localhost:{port}/pid')
+        assert res.status_code == 200
+        new_pid = res.json()['pid']
+        if new_pid not in old_pids:
+            assert True, 'Worker successfully restarted'
+            return new_pid
+        print(f'Worker not restarted, sleeping for {retry} seconds.')
+        time.sleep(retry)
+
+    return None
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(platform.system() == 'Windows', reason='SIGHUP not available on Windows')
+@pytest.mark.parametrize('threading_mode', ['runtime', 'workers'])
+async def test_app_worker_restart(server_app, threading_mode):
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        pid_file_path = Path(tmp_dir, 'server.pid')
+        async with server_app(
+            interface='wsgi', app='restart', threading_mode=threading_mode, extra_args={'pid_file': pid_file_path}
+        ) as port:
+            with pid_file_path.open('r') as pid_fd:
+                server_pid = int(pid_fd.read().strip())
+
+            res = httpx.get(f'http://localhost:{port}/pid')
+            assert res.status_code == 200
+            worker_pid = res.json()['pid']
+
+            os.kill(server_pid, signal.SIGHUP)
+
+            assert _wait_for_new_pid(port, [worker_pid]) is not None
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(platform.system() == 'Windows', reason='SIGHUP/SIGSTOP not available on Windows')
+@pytest.mark.parametrize('threading_mode', ['runtime', 'workers'])
+async def test_app_workers_kill_timeout(server_app, threading_mode):
+    workers_kill_timeout = 2
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        pid_file_path = Path(tmp_dir, 'server.pid')
+        async with server_app(
+            interface='wsgi',
+            app='restart',
+            threading_mode=threading_mode,
+            extra_args={'workers_kill_timeout': workers_kill_timeout, 'pid_file': pid_file_path},
+        ) as port:
+            with pid_file_path.open('r') as pid_fd:
+                server_pid = int(pid_fd.read().strip())
+
+            res = httpx.get(f'http://localhost:{port}/pid')
+            assert res.status_code == 200
+            worker_pid = res.json()['pid']
+
+            # suspend the worker process to simulate that it hangs
+            os.kill(worker_pid, signal.SIGSTOP)
+
+            # restart
+            os.kill(server_pid, signal.SIGHUP)
+            worker_pid_after_one_restart = _wait_for_new_pid(port, [worker_pid])
+            assert worker_pid_after_one_restart is not None
+
+            # wait until the worker_pid is gone
+            time.sleep(workers_kill_timeout + 0.01)
+
+            # suspend the new worker process to simulate that it hangs
+            os.kill(worker_pid_after_one_restart, signal.SIGSTOP)
+
+            # restart a 2nd time
+            os.kill(server_pid, signal.SIGHUP)
+
+            assert _wait_for_new_pid(port, [worker_pid, worker_pid_after_one_restart]) is not None