Start and stop a gunicorn server.
(app_module)
| 80 | |
| 81 | @pytest.fixture |
| 82 | def gunicorn_server(app_module): |
| 83 | """Start and stop a gunicorn server.""" |
| 84 | app_dir, app_name = app_module |
| 85 | port = find_free_port() |
| 86 | |
| 87 | # Start gunicorn |
| 88 | cmd = [ |
| 89 | sys.executable, '-m', 'gunicorn', |
| 90 | '--bind', f'127.0.0.1:{port}', |
| 91 | '--workers', '2', |
| 92 | '--worker-class', 'sync', |
| 93 | '--access-logfile', '-', |
| 94 | '--error-logfile', '-', |
| 95 | '--log-level', 'info', |
| 96 | '--timeout', '30', |
| 97 | '--graceful-timeout', '30', |
| 98 | app_name |
| 99 | ] |
| 100 | |
| 101 | # Use setsid to create new process group for proper signal handling |
| 102 | proc = subprocess.Popen( |
| 103 | cmd, |
| 104 | cwd=app_dir, |
| 105 | stdout=subprocess.PIPE, |
| 106 | stderr=subprocess.PIPE, |
| 107 | env={**os.environ, 'PYTHONPATH': app_dir}, |
| 108 | preexec_fn=os.setsid |
| 109 | ) |
| 110 | |
| 111 | # Wait for server to start |
| 112 | if not wait_for_server('127.0.0.1', port): |
| 113 | proc.terminate() |
| 114 | proc.wait() |
| 115 | stdout, stderr = proc.communicate() |
| 116 | pytest.fail(f"Gunicorn failed to start:\nstdout: {stdout.decode()}\nstderr: {stderr.decode()}") |
| 117 | |
| 118 | yield proc, port |
| 119 | |
| 120 | # Cleanup - use process group kill for better cleanup |
| 121 | if proc.poll() is None: |
| 122 | try: |
| 123 | os.killpg(os.getpgid(proc.pid), signal.SIGTERM) |
| 124 | except (ProcessLookupError, OSError): |
| 125 | pass |
| 126 | try: |
| 127 | proc.wait(timeout=5) |
| 128 | except subprocess.TimeoutExpired: |
| 129 | try: |
| 130 | os.killpg(os.getpgid(proc.pid), signal.SIGKILL) |
| 131 | except (ProcessLookupError, OSError): |
| 132 | pass |
| 133 | proc.wait() |
| 134 | |
| 135 | |
| 136 | class TestSignalHandlingIntegration: |
nothing calls this directly
no test coverage detected