Function create_worker

benchmarks/dirty_streaming.py:124–144

Create a test worker for benchmarks.

()

def create_worker():
    """Create a test worker for benchmarks."""
    cfg = Config()
    cfg.set("dirty_timeout", 300)
    log = MockLog()

    with mock.patch('gunicorn.dirty.worker.WorkerTmp'):
        worker = DirtyWorker(
            age=1,
            ppid=os.getpid(),
            app_paths=["benchmark:App"],
            cfg=cfg,
            log=log,
            socket_path="/tmp/benchmark.sock"
        )

    worker.apps = {}
    worker._executor = None
    worker.tmp = mock.Mock()

    return worker


def create_arbiter():