Merge pull request #3445 from benoitc/async-workers-security-upgrade

Require secure versions of async worker dependencies

examples/frameworks/tornadoapp.py

@@ -7,23 +7,27 @@
 #   $ gunicorn -k tornado tornadoapp:app
 #
 
+import asyncio
 import tornado.ioloop
 import tornado.web
-from tornado import gen
+
 
 class MainHandler(tornado.web.RequestHandler):
-    @gen.coroutine
+    async def get(self):
-    def get(self):
         # Your asynchronous code here
-        yield gen.sleep(1)  # Example of an asynchronous operation
+        await asyncio.sleep(1)  # Example of an asynchronous operation
         self.write("Hello, World!")
 
+
 def make_app():
     return tornado.web.Application([
         (r"/", MainHandler),
     ])
 
-if __name__ == "__main__":
+
 app = make_app()
+
+
+if __name__ == "__main__":
     app.listen(8888)
     tornado.ioloop.IOLoop.current().start()

gunicorn/config.py

@@ -678,9 +678,9 @@ class WorkerClass(Setting):
         A string referring to one of the following bundled classes:
 
         * ``sync``
-        * ``eventlet`` - Requires eventlet >= 0.24.1 (or install it via
+        * ``eventlet`` - Requires eventlet >= 0.40.3 (or install it via
           ``pip install gunicorn[eventlet]``)
-        * ``gevent``   - Requires gevent >= 1.4 (or install it via
+        * ``gevent``   - Requires gevent >= 23.9.0 (or install it via
           ``pip install gunicorn[gevent]``)
         * ``tornado``  - Requires tornado >= 0.2 (or install it via
           ``pip install gunicorn[tornado]``)

gunicorn/workers/geventlet.py

@@ -2,25 +2,34 @@
 # This file is part of gunicorn released under the MIT license.
 # See the NOTICE for more information.
 
-from functools import partial
+# NOTE: eventlet import and monkey_patch() must happen before any other imports
-import sys
+# to ensure all standard library modules are properly patched.
-
 try:
     import eventlet
 except ImportError:
-    raise RuntimeError("eventlet worker requires eventlet 0.24.1 or higher")
+    raise RuntimeError("eventlet worker requires eventlet 0.40.3 or higher")
 else:
     from packaging.version import parse as parse_version
-    if parse_version(eventlet.__version__) < parse_version('0.24.1'):
+    if parse_version(eventlet.__version__) < parse_version('0.40.3'):
-        raise RuntimeError("eventlet worker requires eventlet 0.24.1 or higher")
+        raise RuntimeError("eventlet worker requires eventlet 0.40.3 or higher")
 
-from eventlet import hubs, greenthread
+# Perform monkey patching early, before importing other modules.
-from eventlet.greenio import GreenSocket
+# This ensures that all subsequent imports get the patched versions.
-import eventlet.wsgi
+# NOTE: hubs.use_hub() must NOT be called here - it creates OS resources
-import greenlet
+# (like kqueue on macOS) that don't survive fork. It must be called in
+# each worker process after fork, in the patch() method.
+eventlet.monkey_patch()
 
-from gunicorn.workers.base_async import AsyncWorker
+from functools import partial  # noqa: E402
-from gunicorn.sock import ssl_wrap_socket
+import sys  # noqa: E402
+
+from eventlet import hubs, greenthread  # noqa: E402
+from eventlet.greenio import GreenSocket  # noqa: E402
+import eventlet.wsgi  # noqa: E402
+import greenlet  # noqa: E402
+
+from gunicorn.workers.base_async import AsyncWorker  # noqa: E402
+from gunicorn.sock import ssl_wrap_socket  # noqa: E402
 
 # ALREADY_HANDLED is removed in 0.30.3+ now it's `WSGI_LOCAL.already_handled: bool`
 # https://github.com/eventlet/eventlet/pull/544
@@ -124,8 +133,11 @@ def patch_sendfile():
 class EventletWorker(AsyncWorker):
 
     def patch(self):
+        # NOTE: eventlet.monkey_patch() is called at module import time to
+        # ensure all imports are properly patched. However, hubs.use_hub()
+        # must be called here (after fork) because it creates OS resources
+        # like kqueue that don't survive fork.
         hubs.use_hub()
-        eventlet.monkey_patch()
         patch_sendfile()
 
     def is_already_handled(self, respiter):

gunicorn/workers/ggevent.py

@@ -11,11 +11,11 @@ import time
 try:
     import gevent
 except ImportError:
-    raise RuntimeError("gevent worker requires gevent 1.4 or higher")
+    raise RuntimeError("gevent worker requires gevent 23.9.0 or higher")
 else:
     from packaging.version import parse as parse_version
-    if parse_version(gevent.__version__) < parse_version('1.4'):
+    if parse_version(gevent.__version__) < parse_version('23.9.0'):
-        raise RuntimeError("gevent worker requires gevent 1.4 or higher")
+        raise RuntimeError("gevent worker requires gevent 23.9.0 or higher")
 
 from gevent.pool import Pool
 from gevent.server import StreamServer
@@ -89,10 +89,7 @@ class GeventWorker(AsyncWorker):
         try:
             # Stop accepting requests
             for server in servers:
-                if hasattr(server, 'close'):  # gevent 1.0
                 server.close()
-                if hasattr(server, 'kill'):  # gevent < 1.0
-                    server.kill()
 
             # Handle current requests until graceful_timeout
             ts = time.time()

gunicorn/workers/gtornado.py

@@ -18,15 +18,6 @@ from gunicorn import __version__ as gversion
 from gunicorn.sock import ssl_context
 
 
-# Tornado 5.0 updated its IOLoop, and the `io_loop` arguments to many
-# Tornado functions have been removed in Tornado 5.0. Also, they no
-# longer store PeriodCallbacks in ioloop._callbacks. Instead we store
-# them on our side, and use stop() on them when stopping the worker.
-# See https://www.tornadoweb.org/en/stable/releases/v5.0.0.html#backwards-compatibility-notes
-# for more details.
-TORNADO5 = tornado.version_info >= (5, 0, 0)
-
-
 class TornadoWorker(Worker):
 
     @classmethod
@@ -69,13 +60,9 @@ class TornadoWorker(Worker):
                         pass
                 self.server_alive = False
             else:
-                if TORNADO5:
                 for callback in self.callbacks:
                     callback.stop()
                 self.ioloop.stop()
-                else:
-                    if not self.ioloop._callbacks:
-                        self.ioloop.stop()
 
     def init_process(self):
         # IOLoop cannot survive a fork or be shared across processes
@@ -90,75 +77,36 @@ class TornadoWorker(Worker):
         self.alive = True
         self.server_alive = False
 
-        if TORNADO5:
         self.callbacks = []
         self.callbacks.append(PeriodicCallback(self.watchdog, 1000))
         self.callbacks.append(PeriodicCallback(self.heartbeat, 1000))
         for callback in self.callbacks:
             callback.start()
-        else:
-            PeriodicCallback(self.watchdog, 1000, io_loop=self.ioloop).start()
-            PeriodicCallback(self.heartbeat, 1000, io_loop=self.ioloop).start()
 
         # Assume the app is a WSGI callable if its not an
-        # instance of tornado.web.Application or is an
+        # instance of tornado.web.Application or WSGIContainer
-        # instance of tornado.wsgi.WSGIApplication
         app = self.wsgi
-
+        if not isinstance(app, WSGIContainer) and \
-        if tornado.version_info[0] < 6:
-            if not isinstance(app, tornado.web.Application) or \
-                    isinstance(app, tornado.wsgi.WSGIApplication):
-                app = WSGIContainer(app)
-        elif not isinstance(app, WSGIContainer) and \
                 not isinstance(app, tornado.web.Application):
             app = WSGIContainer(app)
 
-        # Monkey-patching HTTPConnection.finish to count the
-        # number of requests being handled by Tornado. This
-        # will help gunicorn shutdown the worker if max_requests
-        # is exceeded.
-        httpserver = sys.modules["tornado.httpserver"]
-        if hasattr(httpserver, 'HTTPConnection'):
-            old_connection_finish = httpserver.HTTPConnection.finish
-
-            def finish(other):
-                self.handle_request()
-                old_connection_finish(other)
-            httpserver.HTTPConnection.finish = finish
-            sys.modules["tornado.httpserver"] = httpserver
-
-            server_class = tornado.httpserver.HTTPServer
-        else:
-
         class _HTTPServer(tornado.httpserver.HTTPServer):
 
             def on_close(instance, server_conn):
                 self.handle_request()
                 super().on_close(server_conn)
 
-            server_class = _HTTPServer
-
         if self.cfg.is_ssl:
-            if TORNADO5:
+            server = _HTTPServer(app, ssl_options=ssl_context(self.cfg))
-                server = server_class(app, ssl_options=ssl_context(self.cfg))
         else:
-                server = server_class(app, io_loop=self.ioloop,
+            server = _HTTPServer(app)
-                                      ssl_options=ssl_context(self.cfg))
-        else:
-            if TORNADO5:
-                server = server_class(app)
-            else:
-                server = server_class(app, io_loop=self.ioloop)
 
         self.server = server
         self.server_alive = True
 
         for s in self.sockets:
             s.setblocking(0)
-            if hasattr(server, "add_socket"):  # tornado > 2.0
             server.add_socket(s)
-            elif hasattr(server, "_sockets"):  # tornado 2.0
-                server._sockets[s.fileno()] = s
 
         server.no_keep_alive = self.cfg.keepalive <= 0
         server.start(num_processes=1)

pyproject.toml

@@ -47,14 +47,14 @@ Documentation = "https://docs.gunicorn.org"
 Changelog = "https://docs.gunicorn.org/en/stable/news.html"
 
 [project.optional-dependencies]
-gevent = ["gevent>=1.4.0"]
+gevent = ["gevent>=23.9.0"]
-eventlet = ["eventlet>=0.24.1,!=0.36.0"]
+eventlet = ["eventlet>=0.40.3"]
-tornado = ["tornado>=0.2"]
+tornado = ["tornado>=6.5.0"]
 gthread = []
 setproctitle = ["setproctitle"]
 testing = [
-    "gevent",
+    "gevent>=23.9.0",
-    "eventlet",
+    "eventlet>=0.40.3",
     "coverage",
     "pytest",
     "pytest-cov",

tests/test_gtornado.py

@@ -0,0 +1,511 @@
+#
+# This file is part of gunicorn released under the MIT license.
+# See the NOTICE for more information.
+
+"""Tests for the tornado worker."""
+
+import os
+from unittest import mock
+
+import pytest
+
+tornado = pytest.importorskip("tornado")
+
+from gunicorn.config import Config
+from gunicorn.workers import gtornado
+
+
+class FakeSocket:
+    """Mock socket for testing."""
+
+    def __init__(self, data=b''):
+        self.data = data
+        self.closed = False
+        self.blocking = True
+        self._fileno = id(self) % 65536
+
+    def fileno(self):
+        return self._fileno
+
+    def setblocking(self, blocking):
+        self.blocking = blocking
+
+    def recv(self, size):
+        result = self.data[:size]
+        self.data = self.data[size:]
+        return result
+
+    def send(self, data):
+        return len(data)
+
+    def close(self):
+        self.closed = True
+
+    def getsockname(self):
+        return ('127.0.0.1', 8000)
+
+    def getpeername(self):
+        return ('127.0.0.1', 12345)
+
+
+class TestTornadoWorkerInit:
+    """Tests for TornadoWorker initialization."""
+
+    def create_worker(self, cfg=None):
+        """Create a worker instance for testing."""
+        if cfg is None:
+            cfg = Config()
+        cfg.set('workers', 1)
+        cfg.set('max_requests', 0)
+
+        worker = gtornado.TornadoWorker(
+            age=1,
+            ppid=os.getpid(),
+            sockets=[],
+            app=mock.Mock(),
+            timeout=30,
+            cfg=cfg,
+            log=mock.Mock(),
+        )
+        return worker
+
+    def test_worker_init(self):
+        """Test worker initialization."""
+        worker = self.create_worker()
+        assert worker.nr == 0
+
+    def test_init_process_clears_ioloop(self):
+        """Test that init_process clears the current IOLoop."""
+        worker = self.create_worker()
+        worker.tmp = mock.Mock()
+        worker.log = mock.Mock()
+
+        with mock.patch.object(gtornado.IOLoop, 'clear_current') as mock_clear:
+            with mock.patch.object(gtornado.Worker, 'init_process'):
+                worker.init_process()
+            mock_clear.assert_called_once()
+
+
+class TestRequestCounting:
+    """Tests for request counting and max_requests behavior."""
+
+    def create_worker(self, cfg=None):
+        """Create a worker instance for testing."""
+        if cfg is None:
+            cfg = Config()
+        cfg.set('workers', 1)
+
+        worker = gtornado.TornadoWorker(
+            age=1,
+            ppid=os.getpid(),
+            sockets=[],
+            app=mock.Mock(),
+            timeout=30,
+            cfg=cfg,
+            log=mock.Mock(),
+        )
+        return worker
+
+    def test_handle_request_increments_counter(self):
+        """Test that handle_request increments the request counter."""
+        worker = self.create_worker()
+        worker.nr = 0
+        worker.max_requests = 100
+        worker.alive = True
+
+        worker.handle_request()
+
+        assert worker.nr == 1
+        assert worker.alive is True
+
+    def test_max_requests_triggers_shutdown(self):
+        """Test that reaching max_requests triggers shutdown."""
+        cfg = Config()
+        cfg.set('max_requests', 5)
+        worker = self.create_worker(cfg)
+        worker.nr = 4
+        worker.alive = True
+        worker.max_requests = 5
+
+        worker.handle_request()
+
+        assert worker.nr == 5
+        assert worker.alive is False
+
+
+class TestSignalHandling:
+    """Tests for signal handling in tornado worker."""
+
+    def create_worker(self):
+        """Create a worker for testing."""
+        cfg = Config()
+        cfg.set('workers', 1)
+
+        worker = gtornado.TornadoWorker(
+            age=1,
+            ppid=os.getpid(),
+            sockets=[],
+            app=mock.Mock(),
+            timeout=30,
+            cfg=cfg,
+            log=mock.Mock(),
+        )
+        return worker
+
+    def test_handle_exit_sets_alive_false(self):
+        """Test that handle_exit sets alive=False through parent."""
+        worker = self.create_worker()
+        worker.alive = True
+
+        # The parent's handle_exit is what sets alive=False
+        worker.handle_exit(None, None)
+
+        assert worker.alive is False
+
+    def test_handle_exit_only_once(self):
+        """Test that handle_exit only triggers once when alive."""
+        worker = self.create_worker()
+        worker.alive = True
+
+        # First call should set alive=False
+        worker.handle_exit(None, None)
+        assert worker.alive is False
+
+        # Second call should do nothing (alive is already False)
+        # Track that super().handle_exit is not called again
+        with mock.patch.object(gtornado.Worker, 'handle_exit') as mock_exit:
+            worker.handle_exit(None, None)
+            mock_exit.assert_not_called()
+
+
+class TestWatchdog:
+    """Tests for watchdog functionality."""
+
+    def create_worker(self):
+        """Create a worker for testing."""
+        cfg = Config()
+        cfg.set('workers', 1)
+
+        worker = gtornado.TornadoWorker(
+            age=1,
+            ppid=os.getpid(),
+            sockets=[],
+            app=mock.Mock(),
+            timeout=30,
+            cfg=cfg,
+            log=mock.Mock(),
+        )
+        return worker
+
+    def test_watchdog_notifies_when_alive(self):
+        """Test that watchdog calls notify when alive."""
+        worker = self.create_worker()
+        worker.alive = True
+        worker.ppid = os.getppid()
+        worker.tmp = mock.Mock()
+
+        worker.watchdog()
+
+        worker.tmp.notify.assert_called_once()
+
+    def test_watchdog_detects_parent_death(self):
+        """Test that watchdog detects parent death."""
+        worker = self.create_worker()
+        worker.alive = True
+        worker.ppid = 99999999  # Invalid ppid
+        worker.tmp = mock.Mock()
+
+        worker.watchdog()
+
+        assert worker.alive is False
+
+
+class TestHeartbeat:
+    """Tests for heartbeat functionality."""
+
+    def create_worker(self):
+        """Create a worker for testing."""
+        cfg = Config()
+        cfg.set('workers', 1)
+
+        worker = gtornado.TornadoWorker(
+            age=1,
+            ppid=os.getpid(),
+            sockets=[],
+            app=mock.Mock(),
+            timeout=30,
+            cfg=cfg,
+            log=mock.Mock(),
+        )
+        return worker
+
+    def test_heartbeat_stops_server_when_not_alive(self):
+        """Test that heartbeat stops the server when not alive."""
+        worker = self.create_worker()
+        worker.alive = False
+        worker.server_alive = True
+        worker.server = mock.Mock()
+
+        worker.heartbeat()
+
+        worker.server.stop.assert_called_once()
+        assert worker.server_alive is False
+
+    def test_heartbeat_stops_ioloop_after_server(self):
+        """Test that heartbeat stops IOLoop after server is stopped."""
+        worker = self.create_worker()
+        worker.alive = False
+        worker.server_alive = False
+        worker.callbacks = [mock.Mock(), mock.Mock()]
+        worker.ioloop = mock.Mock()
+
+        worker.heartbeat()
+
+        for callback in worker.callbacks:
+            callback.stop.assert_called_once()
+        worker.ioloop.stop.assert_called_once()
+
+
+class TestAppWrapping:
+    """Tests for app wrapping logic."""
+
+    def create_worker(self):
+        """Create a worker for testing."""
+        cfg = Config()
+        cfg.set('workers', 1)
+
+        worker = gtornado.TornadoWorker(
+            age=1,
+            ppid=os.getpid(),
+            sockets=[],
+            app=mock.Mock(),
+            timeout=30,
+            cfg=cfg,
+            log=mock.Mock(),
+        )
+        return worker
+
+    def test_wsgi_callable_wrapped_in_container(self):
+        """Test that a plain WSGI callable gets wrapped in WSGIContainer."""
+        from tornado.wsgi import WSGIContainer
+
+        def wsgi_app(environ, start_response):
+            pass
+
+        # Test that WSGIContainer is used for plain WSGI apps
+        app = wsgi_app
+        if not isinstance(app, WSGIContainer) and \
+                not isinstance(app, tornado.web.Application):
+            app = WSGIContainer(app)
+
+        assert isinstance(app, WSGIContainer)
+
+    def test_tornado_application_not_wrapped(self):
+        """Test that tornado.web.Application is not wrapped."""
+        from tornado.wsgi import WSGIContainer
+
+        tornado_app = tornado.web.Application([])
+
+        # Test the wrapping logic
+        app = tornado_app
+        if not isinstance(app, WSGIContainer) and \
+                not isinstance(app, tornado.web.Application):
+            app = WSGIContainer(app)
+
+        # Should NOT be wrapped
+        assert isinstance(app, tornado.web.Application)
+        assert not isinstance(app, WSGIContainer)
+
+
+class TestSetup:
+    """Tests for the setup class method."""
+
+    def test_setup_patches_request_handler(self):
+        """Test that setup patches RequestHandler.clear."""
+        # Save original
+        original_clear = tornado.web.RequestHandler.clear
+
+        try:
+            gtornado.TornadoWorker.setup()
+
+            # Create a mock handler to test the patched clear method
+            mock_handler = mock.Mock()
+            mock_handler._headers = {"Server": "TornadoServer/1.0"}
+
+            # Call the patched clear
+            new_clear = tornado.web.RequestHandler.clear
+            assert new_clear is not original_clear
+
+        finally:
+            # Restore original
+            tornado.web.RequestHandler.clear = original_clear
+
+
+class TestRunMethod:
+    """Tests for the run method."""
+
+    def create_worker(self):
+        """Create a worker for testing."""
+        cfg = Config()
+        cfg.set('workers', 1)
+        cfg.set('keepalive', 2)
+
+        worker = gtornado.TornadoWorker(
+            age=1,
+            ppid=os.getpid(),
+            sockets=[],
+            app=mock.Mock(),
+            timeout=30,
+            cfg=cfg,
+            log=mock.Mock(),
+        )
+        return worker
+
+    def test_run_sets_up_callbacks(self):
+        """Test that run sets up periodic callbacks."""
+        worker = self.create_worker()
+        worker.wsgi = tornado.web.Application([])
+        worker.sockets = []
+
+        mock_ioloop = mock.Mock()
+        mock_callback = mock.Mock()
+
+        with mock.patch.object(gtornado.IOLoop, 'instance', return_value=mock_ioloop):
+            with mock.patch.object(gtornado, 'PeriodicCallback', return_value=mock_callback) as mock_pc:
+                # Start the run method but stop it immediately
+                mock_ioloop.start.side_effect = lambda: None
+
+                worker.run()
+
+                # Should create two callbacks (watchdog and heartbeat)
+                assert mock_pc.call_count == 2
+                assert mock_callback.start.call_count == 2
+
+    def test_run_creates_http_server(self):
+        """Test that run creates an HTTP server."""
+        worker = self.create_worker()
+        worker.wsgi = tornado.web.Application([])
+        worker.sockets = []
+
+        mock_ioloop = mock.Mock()
+        mock_ioloop.start.side_effect = lambda: None
+
+        with mock.patch.object(gtornado.IOLoop, 'instance', return_value=mock_ioloop):
+            with mock.patch.object(gtornado, 'PeriodicCallback', return_value=mock.Mock()):
+                worker.run()
+
+                assert worker.server is not None
+                assert worker.server_alive is True
+
+    def test_run_adds_sockets_to_server(self):
+        """Test that run adds sockets to the server."""
+        worker = self.create_worker()
+        worker.wsgi = tornado.web.Application([])
+
+        mock_socket = FakeSocket()
+        worker.sockets = [mock_socket]
+
+        mock_ioloop = mock.Mock()
+        mock_ioloop.start.side_effect = lambda: None
+
+        with mock.patch.object(gtornado.IOLoop, 'instance', return_value=mock_ioloop):
+            with mock.patch.object(gtornado, 'PeriodicCallback', return_value=mock.Mock()):
+                with mock.patch.object(tornado.httpserver.HTTPServer, 'add_socket'):
+                    worker.run()
+
+                    # Socket should be set to non-blocking (setblocking(0))
+                    assert not mock_socket.blocking
+
+
+class TestSSLSupport:
+    """Tests for SSL support."""
+
+    def create_worker(self):
+        """Create a worker for testing."""
+        cfg = Config()
+        cfg.set('workers', 1)
+        cfg.set('keepalive', 2)
+
+        worker = gtornado.TornadoWorker(
+            age=1,
+            ppid=os.getpid(),
+            sockets=[],
+            app=mock.Mock(),
+            timeout=30,
+            cfg=cfg,
+            log=mock.Mock(),
+        )
+        return worker
+
+    def test_ssl_server_creation(self):
+        """Test that SSL server is created when is_ssl is True."""
+        worker = self.create_worker()
+        worker.wsgi = tornado.web.Application([])
+        worker.sockets = []
+
+        mock_ioloop = mock.Mock()
+        mock_ioloop.start.side_effect = lambda: None
+
+        mock_ssl_context = mock.Mock()
+
+        # Mock cfg.is_ssl property to return True
+        with mock.patch.object(type(worker.cfg), 'is_ssl', new_callable=mock.PropertyMock, return_value=True):
+            with mock.patch.object(gtornado.IOLoop, 'instance', return_value=mock_ioloop):
+                with mock.patch.object(gtornado, 'PeriodicCallback', return_value=mock.Mock()):
+                    with mock.patch.object(gtornado, 'ssl_context', return_value=mock_ssl_context):
+                        worker.run()
+
+                        # Server should be created with ssl_options
+                        assert worker.server is not None
+
+
+class TestKeepAlive:
+    """Tests for keep-alive configuration."""
+
+    def create_worker(self):
+        """Create a worker for testing."""
+        cfg = Config()
+        cfg.set('workers', 1)
+
+        worker = gtornado.TornadoWorker(
+            age=1,
+            ppid=os.getpid(),
+            sockets=[],
+            app=mock.Mock(),
+            timeout=30,
+            cfg=cfg,
+            log=mock.Mock(),
+        )
+        return worker
+
+    def test_keep_alive_enabled(self):
+        """Test that keep-alive is enabled when keepalive > 0."""
+        worker = self.create_worker()
+        worker.wsgi = tornado.web.Application([])
+        worker.cfg.set('keepalive', 2)
+        worker.sockets = []
+
+        mock_ioloop = mock.Mock()
+        mock_ioloop.start.side_effect = lambda: None
+
+        with mock.patch.object(gtornado.IOLoop, 'instance', return_value=mock_ioloop):
+            with mock.patch.object(gtornado, 'PeriodicCallback', return_value=mock.Mock()):
+                worker.run()
+
+                assert worker.server.no_keep_alive is False
+
+    def test_keep_alive_disabled(self):
+        """Test that keep-alive is disabled when keepalive <= 0."""
+        worker = self.create_worker()
+        worker.wsgi = tornado.web.Application([])
+        worker.cfg.set('keepalive', 0)
+        worker.sockets = []
+
+        mock_ioloop = mock.Mock()
+        mock_ioloop.start.side_effect = lambda: None
+
+        with mock.patch.object(gtornado.IOLoop, 'instance', return_value=mock_ioloop):
+            with mock.patch.object(gtornado, 'PeriodicCallback', return_value=mock.Mock()):
+                worker.run()
+
+                assert worker.server.no_keep_alive is True

tests/workers/test_geventlet.py

@@ -4,9 +4,11 @@
 
 import pytest
 import sys
+from unittest import mock
+
 
 def test_import():
-
+    """Test that the eventlet worker module can be imported."""
     try:
         import eventlet
     except AttributeError:
@@ -14,3 +16,401 @@ def test_import():
             pytest.skip("Ignoring eventlet failures on Python 3.12")
         raise
     __import__('gunicorn.workers.geventlet')
+
+
+class TestVersionRequirement:
+    """Tests for eventlet version requirement checks."""
+
+    def test_import_error_message(self):
+        """Test that ImportError gives correct version message."""
+        with mock.patch.dict('sys.modules', {'eventlet': None}):
+            # Clear cached module if present
+            sys.modules.pop('gunicorn.workers.geventlet', None)
+            with pytest.raises(RuntimeError, match="eventlet 0.40.3"):
+                import importlib
+                import gunicorn.workers.geventlet
+                importlib.reload(gunicorn.workers.geventlet)
+
+    def test_version_check_requires_0_40_3(self):
+        """Test that version check requires eventlet 0.40.3 or higher."""
+        try:
+            import eventlet
+        except (ImportError, AttributeError):
+            pytest.skip("eventlet not available")
+
+        from packaging.version import parse as parse_version
+        min_version = parse_version('0.40.3')
+        current_version = parse_version(eventlet.__version__)
+
+        # If we got this far, the import succeeded, meaning version is sufficient
+        assert current_version >= min_version
+
+
+@pytest.fixture
+def eventlet_worker():
+    """Fixture to create an EventletWorker instance for testing."""
+    try:
+        import eventlet
+    except (ImportError, AttributeError):
+        pytest.skip("eventlet not available")
+
+    from gunicorn.workers.geventlet import EventletWorker
+
+    # Create a minimal mock config
+    cfg = mock.MagicMock()
+    cfg.keepalive = 2
+    cfg.graceful_timeout = 30
+    cfg.is_ssl = False
+    cfg.worker_connections = 1000
+
+    # Create worker with mocked dependencies
+    worker = EventletWorker.__new__(EventletWorker)
+    worker.cfg = cfg
+    worker.alive = True
+    worker.sockets = []
+    worker.log = mock.MagicMock()
+
+    return worker
+
+
+class TestEventletWorker:
+    """Tests for EventletWorker class."""
+
+    def test_worker_class_exists(self):
+        """Test that EventletWorker class is properly defined."""
+        try:
+            import eventlet
+        except (ImportError, AttributeError):
+            pytest.skip("eventlet not available")
+
+        from gunicorn.workers.geventlet import EventletWorker
+        from gunicorn.workers.base_async import AsyncWorker
+
+        assert issubclass(EventletWorker, AsyncWorker)
+
+    def test_patch_method_calls_use_hub(self, eventlet_worker):
+        """Test that patch() calls hubs.use_hub().
+
+        hubs.use_hub() must be called in patch() (after fork) because it creates
+        OS resources like kqueue that don't survive fork.
+        """
+        from eventlet import hubs
+
+        with mock.patch.object(hubs, 'use_hub') as mock_use_hub:
+            with mock.patch('gunicorn.workers.geventlet.patch_sendfile'):
+                eventlet_worker.patch()
+
+        mock_use_hub.assert_called_once()
+
+    def test_patch_method_calls_patch_sendfile(self, eventlet_worker):
+        """Test that patch() calls patch_sendfile()."""
+        from eventlet import hubs
+
+        with mock.patch.object(hubs, 'use_hub'):
+            with mock.patch('gunicorn.workers.geventlet.patch_sendfile') as mock_sf:
+                eventlet_worker.patch()
+
+        mock_sf.assert_called_once()
+
+    def test_monkey_patch_called_at_import_time(self):
+        """Test that monkey_patch is called at module import time.
+
+        Note: hubs.use_hub() and eventlet.monkey_patch() are called at module
+        import time (not in patch()) to ensure all imports are properly patched.
+        This test verifies the module was patched by checking eventlet state.
+        """
+        try:
+            import eventlet
+        except (ImportError, AttributeError):
+            pytest.skip("eventlet not available")
+
+        # Verify eventlet has been patched by checking that socket is patched
+        import socket
+        from eventlet.greenio import GreenSocket
+
+        # After monkey patching, socket.socket should be GreenSocket
+        assert socket.socket is GreenSocket
+
+    def test_timeout_ctx_returns_eventlet_timeout(self, eventlet_worker):
+        """Test that timeout_ctx() returns an eventlet.Timeout."""
+        import eventlet
+
+        timeout = eventlet_worker.timeout_ctx()
+        assert isinstance(timeout, eventlet.Timeout)
+
+    def test_timeout_ctx_uses_keepalive_config(self, eventlet_worker):
+        """Test that timeout_ctx() uses cfg.keepalive value."""
+        import eventlet
+
+        eventlet_worker.cfg.keepalive = 5
+        with mock.patch.object(eventlet, 'Timeout') as mock_timeout:
+            eventlet_worker.timeout_ctx()
+
+        mock_timeout.assert_called_once_with(5, False)
+
+    def test_timeout_ctx_with_no_keepalive(self, eventlet_worker):
+        """Test that timeout_ctx() handles no keepalive (None or 0)."""
+        import eventlet
+
+        eventlet_worker.cfg.keepalive = 0
+        with mock.patch.object(eventlet, 'Timeout') as mock_timeout:
+            eventlet_worker.timeout_ctx()
+
+        mock_timeout.assert_called_once_with(None, False)
+
+    def test_handle_quit_spawns_greenthread(self, eventlet_worker):
+        """Test that handle_quit() spawns a greenthread."""
+        import eventlet
+
+        with mock.patch.object(eventlet, 'spawn') as mock_spawn:
+            eventlet_worker.handle_quit(None, None)
+
+        mock_spawn.assert_called_once()
+
+    def test_handle_usr1_spawns_greenthread(self, eventlet_worker):
+        """Test that handle_usr1() spawns a greenthread."""
+        import eventlet
+
+        with mock.patch.object(eventlet, 'spawn') as mock_spawn:
+            eventlet_worker.handle_usr1(None, None)
+
+        mock_spawn.assert_called_once()
+
+    def test_handle_wraps_ssl_when_configured(self, eventlet_worker):
+        """Test that handle() wraps socket with SSL when is_ssl is True."""
+        from gunicorn.workers import geventlet
+
+        eventlet_worker.cfg.is_ssl = True
+        mock_client = mock.MagicMock()
+        mock_listener = mock.MagicMock()
+
+        with mock.patch.object(geventlet, 'ssl_wrap_socket') as mock_ssl:
+            mock_ssl.return_value = mock_client
+            with mock.patch('gunicorn.workers.base_async.AsyncWorker.handle'):
+                eventlet_worker.handle(mock_listener, mock_client, ('127.0.0.1', 8000))
+
+        mock_ssl.assert_called_once_with(mock_client, eventlet_worker.cfg)
+
+    def test_handle_no_ssl_when_not_configured(self, eventlet_worker):
+        """Test that handle() does not wrap SSL when is_ssl is False."""
+        from gunicorn.workers import geventlet
+
+        eventlet_worker.cfg.is_ssl = False
+        mock_client = mock.MagicMock()
+        mock_listener = mock.MagicMock()
+
+        with mock.patch.object(geventlet, 'ssl_wrap_socket') as mock_ssl:
+            with mock.patch('gunicorn.workers.base_async.AsyncWorker.handle'):
+                eventlet_worker.handle(mock_listener, mock_client, ('127.0.0.1', 8000))
+
+        mock_ssl.assert_not_called()
+
+
+class TestAlreadyHandled:
+    """Tests for is_already_handled() method."""
+
+    def test_is_already_handled_new_style(self, eventlet_worker):
+        """Test is_already_handled with eventlet >= 0.30.3 (WSGI_LOCAL)."""
+        from gunicorn.workers import geventlet
+
+        # Mock the new-style WSGI_LOCAL.already_handled
+        mock_wsgi_local = mock.MagicMock()
+        mock_wsgi_local.already_handled = True
+
+        with mock.patch.object(geventlet, 'EVENTLET_WSGI_LOCAL', mock_wsgi_local):
+            with pytest.raises(StopIteration):
+                eventlet_worker.is_already_handled(mock.MagicMock())
+
+    def test_is_already_handled_old_style(self, eventlet_worker):
+        """Test is_already_handled with eventlet < 0.30.3 (ALREADY_HANDLED)."""
+        from gunicorn.workers import geventlet
+
+        sentinel = object()
+
+        with mock.patch.object(geventlet, 'EVENTLET_WSGI_LOCAL', None):
+            with mock.patch.object(geventlet, 'EVENTLET_ALREADY_HANDLED', sentinel):
+                with pytest.raises(StopIteration):
+                    eventlet_worker.is_already_handled(sentinel)
+
+    def test_is_already_handled_returns_parent_result(self, eventlet_worker):
+        """Test is_already_handled falls through to parent when not handled."""
+        from gunicorn.workers import geventlet
+
+        with mock.patch.object(geventlet, 'EVENTLET_WSGI_LOCAL', None):
+            with mock.patch.object(geventlet, 'EVENTLET_ALREADY_HANDLED', None):
+                with mock.patch('gunicorn.workers.base_async.AsyncWorker.is_already_handled') as mock_parent:
+                    mock_parent.return_value = False
+                    result = eventlet_worker.is_already_handled(mock.MagicMock())
+
+        assert result is False
+        mock_parent.assert_called_once()
+
+
+class TestPatchSendfile:
+    """Tests for patch_sendfile() function."""
+
+    def test_patch_sendfile_adds_method_when_missing(self):
+        """Test that patch_sendfile adds sendfile to GreenSocket if missing."""
+        try:
+            import eventlet
+        except (ImportError, AttributeError):
+            pytest.skip("eventlet not available")
+
+        from gunicorn.workers.geventlet import patch_sendfile, _eventlet_socket_sendfile
+        from eventlet.greenio import GreenSocket
+
+        # Remove sendfile if it exists
+        original = getattr(GreenSocket, 'sendfile', None)
+        if hasattr(GreenSocket, 'sendfile'):
+            delattr(GreenSocket, 'sendfile')
+
+        try:
+            patch_sendfile()
+            assert hasattr(GreenSocket, 'sendfile')
+            assert GreenSocket.sendfile == _eventlet_socket_sendfile
+        finally:
+            # Restore original state
+            if original is not None:
+                GreenSocket.sendfile = original
+            elif hasattr(GreenSocket, 'sendfile'):
+                delattr(GreenSocket, 'sendfile')
+
+    def test_patch_sendfile_preserves_existing_method(self):
+        """Test that patch_sendfile does not override existing sendfile."""
+        try:
+            import eventlet
+        except (ImportError, AttributeError):
+            pytest.skip("eventlet not available")
+
+        from gunicorn.workers.geventlet import patch_sendfile
+        from eventlet.greenio import GreenSocket
+
+        # If sendfile exists, it should be preserved
+        if hasattr(GreenSocket, 'sendfile'):
+            original = GreenSocket.sendfile
+            patch_sendfile()
+            assert GreenSocket.sendfile == original
+
+
+class TestEventletSocketSendfile:
+    """Tests for _eventlet_socket_sendfile() function."""
+
+    def test_sendfile_raises_on_non_blocking(self):
+        """Test that sendfile raises ValueError for non-blocking sockets."""
+        try:
+            import eventlet
+        except (ImportError, AttributeError):
+            pytest.skip("eventlet not available")
+
+        from gunicorn.workers.geventlet import _eventlet_socket_sendfile
+
+        mock_socket = mock.MagicMock()
+        mock_socket.gettimeout.return_value = 0
+
+        with pytest.raises(ValueError, match="non-blocking"):
+            _eventlet_socket_sendfile(mock_socket, mock.MagicMock())
+
+    def test_sendfile_seeks_to_offset(self):
+        """Test that sendfile seeks to offset if provided."""
+        try:
+            import eventlet
+        except (ImportError, AttributeError):
+            pytest.skip("eventlet not available")
+
+        from gunicorn.workers.geventlet import _eventlet_socket_sendfile
+
+        mock_socket = mock.MagicMock()
+        mock_socket.gettimeout.return_value = 1
+        mock_file = mock.MagicMock()
+        mock_file.read.return_value = b''
+
+        _eventlet_socket_sendfile(mock_socket, mock_file, offset=100)
+
+        mock_file.seek.assert_any_call(100)
+
+    def test_sendfile_returns_total_sent(self):
+        """Test that sendfile returns the total bytes sent."""
+        try:
+            import eventlet
+        except (ImportError, AttributeError):
+            pytest.skip("eventlet not available")
+
+        from gunicorn.workers.geventlet import _eventlet_socket_sendfile
+
+        mock_socket = mock.MagicMock()
+        mock_socket.gettimeout.return_value = 1
+        mock_socket.send.return_value = 10
+
+        mock_file = mock.MagicMock()
+        mock_file.read.side_effect = [b'x' * 10, b'']
+
+        result = _eventlet_socket_sendfile(mock_socket, mock_file)
+
+        assert result == 10
+
+
+class TestEventletServe:
+    """Tests for _eventlet_serve() function."""
+
+    def test_serve_creates_green_pool(self):
+        """Test that _eventlet_serve creates a GreenPool."""
+        try:
+            import eventlet
+        except (ImportError, AttributeError):
+            pytest.skip("eventlet not available")
+
+        from gunicorn.workers.geventlet import _eventlet_serve
+
+        mock_sock = mock.MagicMock()
+        mock_sock.accept.side_effect = eventlet.StopServe()
+
+        with mock.patch.object(eventlet.greenpool, 'GreenPool') as mock_pool:
+            mock_pool_instance = mock.MagicMock()
+            mock_pool.return_value = mock_pool_instance
+            mock_pool_instance.waitall.return_value = None
+
+            _eventlet_serve(mock_sock, mock.MagicMock(), 100)
+
+        mock_pool.assert_called_once_with(100)
+
+
+class TestEventletStop:
+    """Tests for _eventlet_stop() function."""
+
+    def test_stop_waits_for_client(self):
+        """Test that _eventlet_stop waits for the client greenlet."""
+        try:
+            import eventlet
+        except (ImportError, AttributeError):
+            pytest.skip("eventlet not available")
+
+        from gunicorn.workers.geventlet import _eventlet_stop
+
+        mock_client = mock.MagicMock()
+        mock_server = mock.MagicMock()
+        mock_conn = mock.MagicMock()
+
+        _eventlet_stop(mock_client, mock_server, mock_conn)
+
+        mock_client.wait.assert_called_once()
+        mock_conn.close.assert_called_once()
+
+    def test_stop_closes_connection_on_greenlet_exit(self):
+        """Test that connection is closed even on GreenletExit."""
+        try:
+            import eventlet
+            import greenlet
+        except (ImportError, AttributeError):
+            pytest.skip("eventlet not available")
+
+        from gunicorn.workers.geventlet import _eventlet_stop
+
+        mock_client = mock.MagicMock()
+        mock_client.wait.side_effect = greenlet.GreenletExit()
+        mock_server = mock.MagicMock()
+        mock_conn = mock.MagicMock()
+
+        # Should not raise
+        _eventlet_stop(mock_client, mock_server, mock_conn)
+
+        mock_conn.close.assert_called_once()

tests/workers/test_ggevent.py

@@ -2,5 +2,215 @@
 # This file is part of gunicorn released under the MIT license.
 # See the NOTICE for more information.
 
+from unittest import mock
+
+import pytest
+
+try:
+    import gevent
+    HAS_GEVENT = True
+except ImportError:
+    HAS_GEVENT = False
+
+pytestmark = pytest.mark.skipif(not HAS_GEVENT, reason="gevent not installed")
+
+
 def test_import():
     __import__('gunicorn.workers.ggevent')
+
+
+def test_version_requirement():
+    """Test that gevent 23.9.0+ is required."""
+    from gunicorn.workers import ggevent
+    from packaging.version import parse as parse_version
+    assert parse_version(gevent.__version__) >= parse_version('23.9.0')
+
+
+class TestGeventWorkerInit:
+    """Test GeventWorker initialization."""
+
+    def test_worker_has_no_server_class(self):
+        """Test that GeventWorker has no server_class by default."""
+        from gunicorn.workers.ggevent import GeventWorker
+        assert GeventWorker.server_class is None
+
+    def test_worker_has_no_wsgi_handler(self):
+        """Test that GeventWorker has no wsgi_handler by default."""
+        from gunicorn.workers.ggevent import GeventWorker
+        assert GeventWorker.wsgi_handler is None
+
+    def test_init_process_patches_and_reinits(self):
+        """Test that init_process calls patch and reinits the hub."""
+        from gunicorn.workers.ggevent import GeventWorker
+
+        worker = mock.Mock(spec=GeventWorker)
+        worker.sockets = []
+
+        with mock.patch('gunicorn.workers.ggevent.hub') as mock_hub, \
+             mock.patch.object(GeventWorker.__bases__[0], 'init_process'):
+            GeventWorker.init_process(worker)
+
+            # Verify patch was called
+            worker.patch.assert_called_once()
+            mock_hub.reinit.assert_called_once()
+
+
+class TestGeventWorkerRun:
+    """Test GeventWorker run method."""
+
+    def test_run_creates_stream_servers(self):
+        """Test that run creates StreamServer instances for each socket."""
+        from gunicorn.workers.ggevent import GeventWorker
+
+        worker = mock.Mock(spec=GeventWorker)
+        worker.sockets = [mock.Mock()]
+        worker.cfg = mock.Mock(is_ssl=False, workers=1, graceful_timeout=30)
+        worker.server_class = None
+        worker.worker_connections = 1000
+
+        # Make alive return True once, then False to exit the loop
+        worker.alive = False
+
+        with mock.patch('gunicorn.workers.ggevent.Pool') as mock_pool, \
+             mock.patch('gunicorn.workers.ggevent.StreamServer') as mock_server_cls, \
+             mock.patch('gunicorn.workers.ggevent.gevent') as mock_gevent:
+
+            mock_server = mock.Mock()
+            mock_server.pool = mock.Mock()
+            mock_server.pool.free_count.return_value = mock_server.pool.size
+            mock_server_cls.return_value = mock_server
+
+            GeventWorker.run(worker)
+
+            mock_server_cls.assert_called_once()
+            mock_server.start.assert_called_once()
+            mock_server.close.assert_called_once()
+
+    def test_run_with_ssl(self):
+        """Test that run configures SSL context when is_ssl is True."""
+        from gunicorn.workers.ggevent import GeventWorker
+
+        worker = mock.Mock(spec=GeventWorker)
+        worker.sockets = [mock.Mock()]
+        worker.cfg = mock.Mock(is_ssl=True, workers=1, graceful_timeout=30)
+        worker.server_class = None
+        worker.worker_connections = 1000
+        worker.alive = False
+
+        with mock.patch('gunicorn.workers.ggevent.Pool'), \
+             mock.patch('gunicorn.workers.ggevent.StreamServer') as mock_server_cls, \
+             mock.patch('gunicorn.workers.ggevent.gevent'), \
+             mock.patch('gunicorn.workers.ggevent.ssl_context') as mock_ssl_ctx:
+
+            mock_server = mock.Mock()
+            mock_server.pool = mock.Mock()
+            mock_server.pool.free_count.return_value = mock_server.pool.size
+            mock_server_cls.return_value = mock_server
+            mock_ssl_ctx.return_value = mock.Mock()
+
+            GeventWorker.run(worker)
+
+            mock_ssl_ctx.assert_called_once_with(worker.cfg)
+            # Verify ssl_context was passed to StreamServer
+            call_kwargs = mock_server_cls.call_args[1]
+            assert 'ssl_context' in call_kwargs
+
+
+class TestSignalHandling:
+    """Test signal handling in GeventWorker."""
+
+    def test_handle_quit_spawns_greenlet(self):
+        """Test that handle_quit spawns a greenlet instead of blocking."""
+        from gunicorn.workers.ggevent import GeventWorker
+
+        worker = mock.Mock(spec=GeventWorker)
+
+        with mock.patch('gunicorn.workers.ggevent.gevent') as mock_gevent:
+            GeventWorker.handle_quit(worker, mock.Mock(), mock.Mock())
+            mock_gevent.spawn.assert_called_once()
+
+    def test_handle_usr1_spawns_greenlet(self):
+        """Test that handle_usr1 spawns a greenlet instead of blocking."""
+        from gunicorn.workers.ggevent import GeventWorker
+
+        worker = mock.Mock(spec=GeventWorker)
+
+        with mock.patch('gunicorn.workers.ggevent.gevent') as mock_gevent:
+            GeventWorker.handle_usr1(worker, mock.Mock(), mock.Mock())
+            mock_gevent.spawn.assert_called_once()
+
+    def test_notify_exits_on_parent_change(self):
+        """Test that notify exits when parent PID changes."""
+        from gunicorn.workers.ggevent import GeventWorker
+
+        worker = mock.Mock(spec=GeventWorker)
+        worker.ppid = 1234
+        worker.log = mock.Mock()
+
+        with mock.patch('gunicorn.workers.ggevent.os') as mock_os, \
+             mock.patch.object(GeventWorker.__bases__[0], 'notify'):
+            mock_os.getppid.return_value = 5678  # Different PID
+
+            with pytest.raises(SystemExit):
+                GeventWorker.notify(worker)
+
+
+class TestPyWSGIWorker:
+    """Test PyWSGI-based worker classes."""
+
+    def test_pywsgi_worker_has_server_class(self):
+        """Test that GeventPyWSGIWorker has proper server_class."""
+        from gunicorn.workers.ggevent import GeventPyWSGIWorker, PyWSGIServer
+        assert GeventPyWSGIWorker.server_class is PyWSGIServer
+
+    def test_pywsgi_worker_has_handler(self):
+        """Test that GeventPyWSGIWorker has proper wsgi_handler."""
+        from gunicorn.workers.ggevent import GeventPyWSGIWorker, PyWSGIHandler
+        assert GeventPyWSGIWorker.wsgi_handler is PyWSGIHandler
+
+    def test_pywsgi_handler_get_environ(self):
+        """Test that PyWSGIHandler adds gunicorn-specific environ keys."""
+        from gunicorn.workers.ggevent import PyWSGIHandler
+
+        handler = mock.Mock(spec=PyWSGIHandler)
+        handler.socket = mock.Mock()
+        handler.path = '/test/path'
+
+        # Mock the parent get_environ
+        with mock.patch.object(PyWSGIHandler.__bases__[0], 'get_environ', return_value={}):
+            env = PyWSGIHandler.get_environ(handler)
+            assert env['gunicorn.sock'] == handler.socket
+            assert env['RAW_URI'] == '/test/path'
+
+
+class TestGeventResponse:
+    """Test GeventResponse helper class."""
+
+    def test_response_attributes(self):
+        """Test GeventResponse stores status, headers, and sent."""
+        from gunicorn.workers.ggevent import GeventResponse
+
+        resp = GeventResponse('200 OK', {'Content-Type': 'text/html'}, 1024)
+        assert resp.status == '200 OK'
+        assert resp.headers == {'Content-Type': 'text/html'}
+        assert resp.sent == 1024
+
+
+class TestTimeoutContext:
+    """Test timeout context manager."""
+
+    def test_timeout_ctx_uses_keepalive(self):
+        """Test that timeout_ctx uses cfg.keepalive."""
+        from gunicorn.workers.ggevent import GeventWorker
+
+        worker = mock.Mock(spec=GeventWorker)
+        worker.cfg = mock.Mock(keepalive=30)
+
+        with mock.patch('gunicorn.workers.ggevent.gevent') as mock_gevent:
+            mock_timeout = mock.Mock()
+            mock_gevent.Timeout.return_value = mock_timeout
+
+            result = GeventWorker.timeout_ctx(worker)
+
+            mock_gevent.Timeout.assert_called_once_with(30, False)
+            assert result == mock_timeout