任务调度后端消息队列【免费下载链接】celeryDistributed Task Queue (development branch)项目地址https://gitcode.com/gh_mirrors/ce/celery点击查看免费下载导读在编写 Celery 单元测试或集成测试时直接复用生产环境的Celery实例往往会把 broker、backend、日志等配置耦合进测试代码导致测试不稳定且难以隔离。celery.contrib.testing.app正是为此提供的官方测试辅助模块它内置了基于memory://与cachememory://的零依赖默认配置提供TestApp()工厂函数快速创建测试应用并通过set_trap()与setup_default_app()两个上下文管理器严格隔离全局应用状态防止测试间相互污染。读完本文你将掌握如何用该模块搭建隔离、可复现的 Celery 测试环境理解其底层全局状态管理机制并能将其与celery.contrib.pytest的 fixture 体系无缝衔接。模块定位与设计目标该模块对应的 API 参考文档位于 docs/reference/celery.contrib.testing.app.rst其实际实现位于 celery/contrib/testing/app.py。模块 docstring 只有一句话“Create Celery app instances used for testing.”创建用于测试的 Celery 应用实例但它承担着三项核心职责快速构造测试应用通过TestApp()工厂函数基于一套内置的测试默认配置生成独立的应用实例隔离全局状态通过set_trap()阻止代码隐式依赖current_app/default_app通过setup_default_app()在测试结束后恢复全局状态配套辅助类Trap陷阱类与UnitLogging日志类为上述目标提供支撑。该模块是 Celery 官方 pytest 插件 celery/contrib/pytest.py 的底层依赖也是项目自身单元测试如 t/unit/conftest.py与集成测试如 t/integration/test_backend_leak_in_fixture_teardown.py的基础设施。内置默认配置DEFAULT_TEST_CONFIG模块顶部定义了一个模块级字典DEFAULT_TEST_CONFIG它浓缩了“测试友好”的配置取向DEFAULT_TEST_CONFIG { worker_hijack_root_logger: False, worker_log_color: False, accept_content: {json}, enable_utc: True, timezone: UTC, broker_url: memory://, result_backend: cachememory://, broker_heartbeat: 0, }逐项解读其设计意图配置项默认值测试场景下的意义worker_hijack_root_loggerFalse不让 worker 劫持根 logger避免污染测试框架的日志输出worker_log_colorFalse关闭彩色日志保证日志输出在 CI 与测试捕获器中可读、稳定accept_content{json}只接受 JSON 序列化内容规避其他序列化器带来的依赖与兼容问题enable_utc/timezoneTrue/UTC固定时区为 UTC保证时间相关断言可复现broker_urlmemory://使用内存 brokerkombu 提供无需启动 RabbitMQ/Redis 即可收发消息result_backendcachememory://结果后端使用内存缓存任务结果无需落盘或依赖外部服务broker_heartbeat0关闭 broker 心跳避免测试期间产生额外的心跳线程与超时干扰从源码结构看这套配置刻意将外部依赖降到最低无需任何消息队列与存储服务即可运行任务调用与结果获取的完整链路这是它能在 CI 与单测中大规模使用的前提。测试应用工厂TestApp()TestApp()是模块对外最核心的工厂函数签名如下def TestApp(nameNone, configNone, enable_loggingFalse, set_as_currentFalse, logUnitLogging, backendNone, brokerNone, **kwargs): App used for testing. from . import tasks # noqa config dict(deepcopy(DEFAULT_TEST_CONFIG), **config or {}) if broker is not None: config.pop(broker_url, None) if backend is not None: config.pop(result_backend, None) log None if enable_logging else log test_app Celery( name or celery.tests, set_as_currentset_as_current, loglog, brokerbroker, backendbackend, **kwargs) test_app.add_defaults(config) return test_app关键行为拆解默认应用名未指定name时使用celery.tests保证各测试应用拥有稳定的标识配置合并先用deepcopy复制DEFAULT_TEST_CONFIG再用调用方传入的config字典覆盖config or {}处理None入参因此config{result_backend: ...}之类的局部覆盖非常轻量broker/backend 覆盖显式传入broker或backend时会先移除默认配置中对应的broker_url/result_backend避免二者冲突随后作为Celery(...)的构造参数传入日志开关enable_loggingFalse时默认注入UnitLogging日志类见下文置为True则传logNone使用标准日志行为副作用函数体内部from . import tasks会导入 celery/contrib/testing/tasks.py从而注册celery.ping共享任务——这是集成测试中用于探活 worker 的辅助任务其实现为返回字符串pong。使用示例from celery.contrib.testing.app import TestApp # 最基本的测试应用内存 broker 内存结果后端 app TestApp() # 覆盖部分配置仍继承其余测试默认值 app TestApp(config{ task_always_eager: True, task_eager_propagates: True, }) # 显式指定结果后端 app TestApp(config{result_backend: TEST_BACKEND})最后一种写法正是集成测试 t/integration/test_backend_leak_in_fixture_teardown.py 中的实际用法它先用TestApp(config{result_backend: TEST_BACKEND})创建应用再用setup_default_app(app)包裹用于验证 fixture teardown 后 backend 连接是否被正确释放。隔离日志UnitLoggingUnitLogging继承自symbol_by_name(Celery.log_cls)——即 Celery 默认的日志类celery.app.log.Logging通过symbol_by_name按名称动态解析而来class UnitLogging(symbol_by_name(Celery.log_cls)): Sets up logging for the test application. def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.already_setup True它的唯一特殊之处是在构造时直接置self.already_setup True。从源码结构看Celery 的日志系统以_setup/already_setup之类标志防止重复初始化在 celery/contrib/testing/worker.py 的setup_app_for_worker中可见type(app.log)._setup False的复位操作UnitLogging通过预先标记“已完成初始化”避免测试应用在日志尚未就绪时触发重复设置流程从而让日志保持安静、可控。陷阱机制Trap 与 set_trap()Trap是一个“伪装成 app、但一用就抛异常”的类class Trap: Trap that pretends to be an app but raises an exception instead. def __getattr__(self, name): if name _is_coroutine or name __func__: return None print(name) raise RuntimeError(Test depends on current_app)其设计动机在 docstring 中写得很清楚保护那些没有正确传递 app 实例、而是隐式回退到current_app的代码。凡是访问该对象的任意属性都会抛出RuntimeError(Test depends on current_app)从而在测试中立刻暴露对全局current_app的隐性依赖。其中两个特殊属性名_is_coroutine、__func__返回None是用于兼容unittest.mock在 Python 3.8 上对该对象进行 patch 的场景。set_trap()则是安装陷阱的上下文管理器contextmanager def set_trap(app): trap Trap() prev_tls _state._tls _state.set_default_app(trap) class NonTLS: current_app trap _state._tls NonTLS() try: yield finally: _state._tls prev_tls它做了两件事把_state.default_app换成Trap实例把线程局部存储_state._tls替换为一个current_app同样指向Trap的内部类NonTLS的实例。这样一来测试期间任何读取默认应用或当前应用的代码都会触发RuntimeError从而强制开发者显式传递 app。退出上下文后恢复原先的_state._tls。相关的全局状态定义位于 celery/_state.py_tls是线程局部存储第 71 行set_default_app第 86 行、get_current_app、_tls.current_app等共同构成了 Celery 的全局应用状态机set_trap正是通过替换这些全局量实现“陷阱”效果。状态隔离核心setup_default_app()setup_default_app()是模块中用于测试隔离的最重要上下文管理器负责“进入测试前备份全局状态退出测试后彻底恢复”contextmanager def setup_default_app(app, use_trapFalse): prev_current_app _state.get_current_app() prev_default_app _state.default_app prev_finalizers set(_state._on_app_finalizers) prev_apps weakref.WeakSet(_state._apps) try: if use_trap: with set_trap(app): yield else: yield finally: _state.set_default_app(prev_default_app) _state._tls.current_app prev_current_app if app is not prev_current_app: app.close() _state._on_app_finalizers prev_finalizers _state._apps prev_apps if app._backend is not None: app._backend_cache None app._local.backend None其隔离策略包含四个层面全局应用引用备份并恢复_state中的当前应用与默认应用finalizer 集合备份_state._on_app_finalizers测试期间新增的应用 finalizer 不会泄漏到后续测试应用注册表用weakref.WeakSet(_state._apps)快照应用集合测试后恢复避免弱引用集合持续膨胀backend 资源清理退出时若app._backend存在则清空app._backend_cache与app._local.backend使 GC 能够回收 backend 的连接。代码注释明确引用了 GitHub issue #6382——这正是为防止 backend 连接泄漏而加入的修复也是 t/integration/test_backend_leak_in_fixture_teardown.py 这个专门回归测试存在的原因。此外use_trapTrue时会在进入阶段嵌套调用set_trap(app)实现“默认/当前应用一律禁止访问”的最强隔离。与 pytest 插件体系的集成celery.contrib.testing.app并非孤立存在它被官方 pytest 插件 celery/contrib/pytest.py 直接消费。插件内部的_create_app()上下文管理器是二者的粘合点contextmanager def _create_app(enable_loggingFalse, use_trapFalse, parametersNone, **config): from .testing.app import TestApp, setup_default_app parameters {} if not parameters else parameters test_app TestApp( set_as_currentFalse, enable_loggingenable_logging, configconfig, **parameters ) with setup_default_app(test_app, use_trapuse_trap): yield test_app由此派生出系列 fixturecelery_app函数级每个测试创建一个TestApp实例测试结束自动清理celery_session_app会话级整个测试会话共享一个测试应用退出时通过setup_default_app恢复状态celery_config/celery_parameters/celery_enable_logging分别控制传入TestApp的config、Celery构造参数与日志开关use_celery_app_trap返回True即可为整个测试会话开启set_trap强制所有测试显式传递 appdepends_on_current_app显式调用celery_app.set_current()为确实依赖当前应用的测试提供出口celery_worker/celery_session_worker基于测试应用启动嵌入式 workersolo 池并通过celery.ping任务做就绪探测。项目自身的单元测试基建 t/unit/conftest.py 也直接使用了TestApp与Trap例如在第 223 行用TestApp为测试实例注入Celery属性并在第 91 行定义use_celery_app_trap的覆写可见该模块是 Celery 自测体系的公共地基。完整实战示例综合上述 API一个典型的测试场景可以这样组织from celery.contrib.testing.app import TestApp, setup_default_app, set_trap # 1. 创建隔离的测试应用 app TestApp(config{ task_always_eager: True, broker_connection_retry_on_startup: True, }) # 2. 用 setup_default_app 包裹确保测试结束后全局状态恢复 with setup_default_app(app): # 3. 在隔离环境中执行任务 result some_task.apply(args[1, 2]) assert result.get() 3 # 4. 如需强制检查代码是否隐式依赖 current_app开启陷阱 with set_trap(app): try: from celery import current_app _ current_app.conf # 此处将抛出 RuntimeError: Test depends on current_app except RuntimeError: pass # 陷阱生效说明代码存在隐性全局依赖在集成测试层面更常见的做法是与celery.contrib.testing.worker配合start_worker(app)在独立线程中启动 worker并先执行ping.delay().get()探活见 celery/contrib/testing/worker.py其中ping正是 celery/contrib/testing/tasks.py 提供的辅助任务。小结celery.contrib.testing.app用不到一百五十行代码浓缩了 Celery 官方测试隔离的全部核心机制DEFAULT_TEST_CONFIG提供零外部依赖的默认配置TestApp()负责快速实例化UnitLogging保持日志安静Trap/set_trap()揪出隐性全局依赖setup_default_app()则在四个层面保证全局状态与 backend 连接的彻底回收。无论你是直接使用这些 API 编写测试还是通过celery.contrib.pytest的 fixture 间接受益理解本模块都是掌握 Celery 测试体系的关键一步。赞分享任务调度后端消息队列【免费下载链接】celeryDistributed Task Queue (development branch)项目地址https://gitcode.com/gh_mirrors/ce/celery点击查看免费下载相关推荐Celery 应用日志配置解析celery.app.log 模块源码级指南Celery 应用日志配置解析celery.app.log 模块源码级指南 Celery 作为分布式任务队列其 worker、beat、命令行工具等进程的运任务调度后端消息队列browserify模块化设计模式从单例到工厂的实战应用browserify模块化设计模式从单例到工厂的实战应用 在前端开发中你是否经常遇到代码复用困难、全局变量污染、依赖关系混乱等问题这些痛点不仅降低开发效率前端构建CLI开发工具告别选择器地狱Midscene.js 视觉自动化测试从入门到实战告别选择器地狱Midscene.js 视觉自动化测试从入门到实战 前端重构、UI 改版、产品迭代……每一次变化背后测试代码里那几十个脆弱的 CSS 选择人工智能AI Agent测试GUI 自动化浏览器控制测试智能体上一篇终极指南PointNet在建筑信息模型中的5大创新应用——从构件智能分类到工程量自动估算下一篇KernelSU项目对非GKI内核设备的支持现状与技术解析创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考