qzl
46 lines
1.2 KiB
Python
46 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
from celery import Celery
|
|
from pytest import MonkeyPatch
|
|
|
|
from core.logging import celery as celery_logging
|
|
from core.logging.context import clear_context, get_context
|
|
|
|
|
|
class DummyTask:
|
|
name: str = "tasks.sample"
|
|
|
|
|
|
def test_celery_prerun_binds_task_context() -> None:
|
|
handlers = celery_logging.build_celery_signal_handlers()
|
|
|
|
handlers.on_task_prerun(task_id="task-123", task=DummyTask())
|
|
context = get_context()
|
|
|
|
assert context["task_id"] == "task-123"
|
|
assert context["task_name"] == "tasks.sample"
|
|
|
|
clear_context()
|
|
|
|
|
|
def test_celery_setup_logging_calls_configure(monkeypatch: MonkeyPatch) -> None:
|
|
called = {"value": False}
|
|
|
|
def fake_configure_logging(settings: object | None = None) -> None:
|
|
called["value"] = True
|
|
|
|
monkeypatch.setattr(celery_logging, "configure_logging", fake_configure_logging)
|
|
handlers = celery_logging.build_celery_signal_handlers()
|
|
|
|
handlers.on_setup_logging()
|
|
|
|
assert called["value"] is True
|
|
|
|
|
|
def test_configure_celery_app_disables_hijack() -> None:
|
|
app = Celery("test")
|
|
|
|
celery_logging.configure_celery_app(app)
|
|
|
|
assert app.conf.worker_hijack_root_logger is False
|