# coding=utf-8 """ @Time :2026/6/24 @Author :tao.chen Regression tests for gunicorn.conf.py. The MCP session-affinity constraint is documented in gunicorn.conf.py: the mcp library stores each session in a per-process dict, so a multi-worker gunicorn deploy returns "Session not found" ~50% of the time. The fix in this config is an `on_starting` hook that warns loudly when workers > 1. These tests assert that: 1. The hook exists and is wired up. 2. workers == 1 is silent (no false alarms in the default single-worker mode people use after reading the comment). 3. workers > 1 emits a WARNING that mentions the symptom and the fix. If anyone ever "cleans up" the hook, these tests fail — that's the point. """ import importlib.util import sys from pathlib import Path import pytest REPO_ROOT = Path(__file__).resolve().parents[2] GUNICORN_CONF = REPO_ROOT / "gunicorn.conf.py" def _load_gunicorn_conf(): """Import gunicorn.conf.py as a module (not a package member). The file lives at the repo root and is loaded by gunicorn via exec(); we use importlib so the test can introspect `on_starting` and mutate `workers` between cases. """ spec = importlib.util.spec_from_file_location("gunicorn_conf_under_test", GUNICORN_CONF) assert spec is not None and spec.loader is not None, f"cannot load {GUNICORN_CONF}" module = importlib.util.module_from_spec(spec) # Make sure repo root is on sys.path so the module's `import os` etc. work. if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) spec.loader.exec_module(module) return module class _FakeServer: """Minimal stand-in for gunicorn's `server` argument to on_starting. gunicorn passes a `gunicorn.arbiter.Arbiter`-like object that exposes `server.log.warning(...)`. We only need the .log.warning sink. """ def __init__(self): self.warnings: list[str] = [] class _Log: def __init__(self, sink): self._sink = sink def warning(self, fmt: str, *args) -> None: # noqa: A002 - matches gunicorn API self._sink.append(fmt % args if args else fmt) @property def log(self) -> "_FakeServer._Log": return self._Log(self.warnings) @pytest.fixture def gunicorn_conf(): """Load gunicorn.conf.py once per test, then reset `workers` to the project default so other tests in the suite are not affected by us mutating the module global.""" mod = _load_gunicorn_conf() yield mod # Restore the env-driven default so a stale value doesn't leak out. import os mod.workers = int(os.environ.get("GUNICORN_WORKERS", "2")) def test_on_starting_hook_is_defined(gunicorn_conf): """The hook must exist — the bug fix lives here.""" assert callable(getattr(gunicorn_conf, "on_starting", None)), ( "gunicorn.conf.py is missing the on_starting hook. The MCP session-" "affinity warning is the only thing that keeps a multi-worker deploy " "from silently returning 'Session not found' for half its requests." ) def test_on_starting_silent_for_single_worker(gunicorn_conf): """workers == 1 is the safe config; no warning should fire.""" gunicorn_conf.workers = 1 server = _FakeServer() gunicorn_conf.on_starting(server) assert server.warnings == [], ( f"workers=1 should be silent (no MCP session affinity issue), got: {server.warnings}" ) def test_on_starting_warns_for_multi_worker(gunicorn_conf): """workers > 1 must produce a warning that names the symptom and the fix. The point of the hook is to be loud at deploy time, not at request time. """ gunicorn_conf.workers = 4 server = _FakeServer() gunicorn_conf.on_starting(server) assert len(server.warnings) == 1, ( f"expected exactly one warning when workers=4, got {len(server.warnings)}: {server.warnings}" ) msg = server.warnings[0] assert "GUNICORN_WORKERS=4" in msg, "warning should name the offending worker count" assert "Session not found" in msg, "warning should name the symptom users actually see" assert "GUNICORN_WORKERS=1" in msg, "warning should point at the easy fix" assert "sticky-session" in msg, "warning should mention the load-balancer escape hatch"