101 lines
2.5 KiB
Python
101 lines
2.5 KiB
Python
import importlib.util
|
|
import sys
|
|
import types
|
|
from pathlib import Path
|
|
|
|
from werkzeug.local import LocalProxy
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
GUNICORN_CONFIG_PATH = ROOT / "gunicorn.conf.py"
|
|
|
|
|
|
class _Log:
|
|
def info(self, *_args, **_kwargs):
|
|
return None
|
|
|
|
def exception(self, *_args, **_kwargs):
|
|
return None
|
|
|
|
|
|
class _Server:
|
|
log = _Log()
|
|
|
|
|
|
class _Worker:
|
|
pid = 123
|
|
|
|
|
|
def _load_gunicorn_config():
|
|
spec = importlib.util.spec_from_file_location(
|
|
"gunicorn_conf_under_test",
|
|
GUNICORN_CONFIG_PATH,
|
|
)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec.loader is not None
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_post_fork_skips_request_bound_local_proxy(monkeypatch):
|
|
config = _load_gunicorn_config()
|
|
fake_module = types.ModuleType("services.fake_request_bound_module")
|
|
|
|
def _missing_request_context():
|
|
raise RuntimeError("Working outside of request context.")
|
|
|
|
fake_module.request_bound = LocalProxy(_missing_request_context)
|
|
monkeypatch.setitem(sys.modules, fake_module.__name__, fake_module)
|
|
|
|
config.post_fork(_Server(), _Worker())
|
|
|
|
|
|
def test_post_fork_disposes_database_manager_instance_cache(monkeypatch):
|
|
from sqlalchemy import create_engine
|
|
|
|
config = _load_gunicorn_config()
|
|
fake_module = types.ModuleType("database.manager")
|
|
engine = create_engine("sqlite:///:memory:")
|
|
before_pool = id(engine.pool)
|
|
|
|
class FakeDatabaseManager:
|
|
_instance_cache = {
|
|
("sqlite", "memory"): {
|
|
"engine": engine,
|
|
"Session": object(),
|
|
}
|
|
}
|
|
|
|
fake_module.DatabaseManager = FakeDatabaseManager
|
|
monkeypatch.setitem(sys.modules, fake_module.__name__, fake_module)
|
|
|
|
config.post_fork(_Server(), _Worker())
|
|
|
|
assert id(engine.pool) != before_pool
|
|
|
|
|
|
def test_gunicorn_disables_preload_for_hup_hot_reload():
|
|
config = _load_gunicorn_config()
|
|
|
|
assert config.preload_app is False
|
|
|
|
|
|
def test_gunicorn_uses_thread_worker_for_health_resilience(monkeypatch):
|
|
monkeypatch.delenv("GUNICORN_WORKER_CLASS", raising=False)
|
|
monkeypatch.delenv("GUNICORN_THREADS", raising=False)
|
|
|
|
config = _load_gunicorn_config()
|
|
|
|
assert config.worker_class == "gthread"
|
|
assert config.threads == 4
|
|
|
|
|
|
def test_gunicorn_thread_worker_can_be_overridden(monkeypatch):
|
|
monkeypatch.setenv("GUNICORN_WORKER_CLASS", "sync")
|
|
monkeypatch.setenv("GUNICORN_THREADS", "2")
|
|
|
|
config = _load_gunicorn_config()
|
|
|
|
assert config.worker_class == "sync"
|
|
assert config.threads == 2
|