Sentry Outbox 模式测试编写实战指南：覆盖记录生成、跨 silo 排水与信号派发的完整测试方法论

导读

在 Sentry 的 hybrid-cloud（混合云）架构中，控制面（Control Silo）与各 Cell Silo 之间通过数据库外发箱（Outbox）模式传递状态变更，从而实现异步、可靠、可重放的跨 silo 数据同步。本文以仓库内 outbox-tests.md 为骨架，系统讲解如何为这套机制编写高质量测试：从验证 outbox 记录是否正确生成，到用 outbox_runner() 排水并断言跨 silo 副作用，再到校验信号参数、shard 调度与删除传播。你将获得一套可直接套用的测试模板、工具函数的语义边界（例如为什么 factory 绝不能包裹在 assume_test_silo_mode 中），以及底层实现依据，可直接迁移到对 ControlOutbox / CellOutbox 相关模块的测试编写中。

一、先读懂被测对象：Outbox 模式的运行时模型

编写 outbox 测试之前，需要理解被测机制的实现位置。Sentry 将 outbox 核心全部收敛在 src/sentry/hybridcloud 目录下：

• outbox.py：OutboxBase 抽象模型及 CellOutbox、ControlOutbox 两张具体表，以及 outbox_context 上下文管理器；
• category.py：OutboxCategory（消息类别）与 OutboxScope（分片作用域）的枚举定义；
• base.py：CellOutboxProducingModel / ControlOutboxProducingModel / ReplicatedCellModel / ReplicatedControlModel 等模型 Mixin，负责在 save/update/delete 时自动产出 outbox；
• signals.py：process_cell_outbox 与 process_control_outbox 两个 Signal，排水（drain）时触发。

从源码看（outbox.py），每条 outbox 记录都携带几组关键列：

列	含义	说明
shard_scope + shard_identifier	分片键	决定并行粒度，不同分片可并行处理。如 ORGANIZATION_SCOPE(0) 下以 organization_id 分片
category + object_identifier	消息类别与对象 ID	配合分片键构成“coalesced”合并键（见 coalesced_columns），同键多条消息在排水时只取最新一条处理
payload	JSON 负载	用于需要在 outbox 中快照状态的消息（如 provisioning、审计日志）
scheduled_for / scheduled_from	调度时间	失败后按 next_schedule（outbox.py，last_delay * 2，封顶 1 小时）指数退避重排
cell_name	目标 Cell	仅 ControlOutbox（控制面 → Cell 方向）携带

方向性非常重要：CellOutbox（表 sentry_regionoutbox）承载 Cell → Control 方向，其 send_signal 发送 process_cell_outbox；ControlOutbox（表 sentry_controloutbox）承载 Control → Cell 方向，其 send_signal 发送 process_control_outbox（参见 outbox.py）。因此测试装饰器也要按方向选择：聚焦 ControlOutbox 用 @control_silo_test，聚焦 CellOutbox 用 @cell_silo_test。

二、标准 Import Block：测试文件的导入骨架

outbox-tests.md 给出了开箱即用的导入块，几乎覆盖了所有 outbox 测试会用到符号：

from unittest.mock import Mock, call, patch

import pytest

from sentry.hybridcloud.models.outbox import (
    ControlOutbox,
    CellOutbox,
    outbox_context,
)
from sentry.hybridcloud.outbox.category import OutboxCategory, OutboxScope
from sentry.models.organization import Organization
from sentry.models.organizationmember import OrganizationMember
from sentry.silo.base import SiloMode
from sentry.testutils.cases import TestCase
from sentry.testutils.outbox import outbox_runner
from sentry.testutils.silo import (
    assume_test_silo_mode,
    assume_test_silo_mode_of,
    control_silo_test,
    cell_silo_test,
)

可以按被测模型裁剪：如果被测对象只是“产生 CellOutbox 的复制模型”，那么只需要 CellOutbox、outbox_context、outbox_runner、cell_silo_test、assume_test_silo_mode_of 即可。

三、测试脚手架：装饰器、基类与运行方式

仓库级 skill 文档 SKILL.md 给出了与参考文档一致的脚手架规则，推荐组合如下：

场景	装饰器	基类
控制 outbox 测试	@control_silo_test	TestCase
Cell outbox 测试	@cell_silo_test	TestCase
线程/并发场景	无装饰器或按需	TransactionTestCase

SiloModeTestDecorator 的实现位于 silo.py，支持 cells=、include_monolith_run=True 等扩展参数（例如同时运行 Monolith 模式则自动生成 XxxTest__InMonolithMode 副本）。

TestCase vs TransactionTestCase 的选择

这是新手最容易出错的地方。outbox_runner() 在成功退出时会同步执行所有到期 outbox 任务，其底层 drain_shard 明确断言“必须在任何活动事务之外调用”（见 outbox.py 中的 in_test_assert_no_transaction）。这意味着：标准的 outbox 排水测试用 TestCase 就够了，因为 outbox_runner 内部以同步方式（concurrency=1）通过任务 Runner 与 assume_test_silo_mode(SiloMode.MONOLITH) 完成排水（见 testutils/outbox.py 的实现：它循环 enqueue_outbox_jobs / enqueue_outbox_jobs_control，直到没有任何 find_scheduled_shards() 命中，最多 10 轮，超限抛 OutboxRecursionLimitError）。

只有真正的线程/并发测试（例如借助 threading.Barrier 同时触发多个 drain，验证锁冲突与收敛）才需要 TransactionTestCase 提供真实提交的事务语义。仓库中 OutboxDrainTest(TransactionTestCase) 正是此类用例的代表（见 test_outbox.py）。

两把“切换 silo”的工具

• assume_test_silo_mode_of(Model)：根据模型的 @cell_silo_model/@control_silo_model 装饰自动推导所在 silo，最适合“只查单个模型的跨 silo 状态”；
• assume_test_silo_mode(SiloMode.X)：用于一个代码块里同时访问多个模型或非模型资源时，手动指定 SiloMode.CONTROL / SiloMode.CELL。

它们的共同铁律是：只包裹直接 ORM 查询（objects.filter/get/count/exists），绝不包裹 factory 调用（见下文“红线规则”）。

四、模板一：验证 outbox 记录创建

第一条测试要回答的问题是：保存一个模型，到底有没有按预期写入 outbox 表？核心工具是 outbox_context(flush=False)——它会临时关闭“提交后自动排水”行为（源码见 outbox.py，flushing_enabled 由 thread-local 的 OutboxContext 管理），从而让你在保存后、排水前检查落库记录。

@control_silo_test
class Test{Feature}Outbox(TestCase):

    def test_outbox_created_on_save(self):
        """Verify that saving a model creates the expected outbox record."""
        with outbox_context(flush=False):
            {Model}(id=10).outbox_for_update().save()

        assert {OutboxModel}.objects.count() == 1
        outbox = {OutboxModel}.objects.first()
        assert outbox.shard_scope == OutboxScope.{SCOPE}.value
        assert outbox.shard_identifier == 10
        assert outbox.category == OutboxCategory.{CATEGORY}.value

    def test_multiple_outboxes_created(self):
        """Verify multiple outbox records are created for batch operations."""
        with outbox_context(flush=False):
            {Model}(id=10).outbox_for_update().save()
            {Model}(id=20).outbox_for_update().save()

        assert {OutboxModel}.objects.count() == 2

模板中的 {Model}、{OutboxModel}、{SCOPE}、{CATEGORY} 需要在具体场景中替换。例如在仓库自己的单测中，Organization（Cell 侧复制源模型）对应 OutboxCategory.ORGANIZATION_UPDATE 与 OutboxScope.ORGANIZATION_SCOPE，真实用例 test_creating_org_outboxes 如下（test_outbox.py）：

with outbox_context(flush=False):
    Organization(id=10).outbox_for_update().save()
    OrganizationMember(organization_id=12, id=15).outbox_for_update().save()
assert CellOutbox.objects.count() == 2

with outbox_runner():
    pass  # drain outboxes
assert CellOutbox.objects.count() == 0

断言取值背后的依据

• OutboxScope.{SCOPE}：枚举值定义在 category.py。参考文档模板中常见 ORGANIZATION_SCOPE(0)、USER_SCOPE(1) 等；每个 Scope 通过 scope_categories(...) 静态注册其允许的 Category 集合。
• OutboxCategory.{CATEGORY}：见 category.py，是 IntEnum，因此模板中比较的是 .value。
• category 与 scope 的合法性：OutboxBase.save() 会调用 OutboxScope.scope_has_category() 校验，若类别未注册到该作用域直接抛 InvalidOutboxError（outbox.py）。测试中若看到该异常，通常不是测试写错，而是类别/作用域配对本身不合法。

补充：这条记录到底是谁“造”出来的？

手动 outbox_for_update().save() 是测试的显式写法。生产路径上，复制模型继承 ReplicatedCellModel（Cell 侧，见 base.py）或 ReplicatedControlModel（Control 侧，base.py），其 save/update/delete 会在同一个原子事务里自动调用 outbox_for_update() 或 outboxes_for_update()。例如 OrganizationMember 声明为 @cell_silo_model 并继承 ReplicatedCellModel（见 organizationmember.py）。理解这层关系后，测试就可以从“验证模型行为”自然过渡到“验证 outbox 产出”。

五、模板二：验证排水与跨 silo 副作用

光有记录不算数，outbox 的核心价值在于排水后在对端产生真实的副作用。这一步用 outbox_runner() 同步驱动。

class Test{Feature}OutboxProcessing(TestCase):

    def test_outbox_drains_and_produces_side_effect(self):
        """Verify outbox processing produces the expected cross-silo effect."""
        # Create source objects using factories (no silo wrapper needed)
        org = self.create_organization()
        member = self.create_member(
            organization=org,
            user=self.create_user(),
        )

        # Drain outboxes
        with outbox_runner():
            pass

        # Verify cross-silo effect (silo wrapper needed for ORM query)
        with assume_test_silo_mode_of({ReplicaModel}):
            assert {ReplicaModel}.objects.filter(
                organization_id=org.id,
            ).exists()

    def test_outbox_drain_is_idempotent(self):
        """Verify draining the same shard twice produces no duplicates."""
        org = self.create_organization()

        with outbox_runner():
            pass

        with assume_test_silo_mode_of({ReplicaModel}):
            count_after_first = {ReplicaModel}.objects.count()

        # Drain again — should be a no-op
        with outbox_runner():
            pass

        with assume_test_silo_mode_of({ReplicaModel}):
            assert {ReplicaModel}.objects.count() == count_after_first

两个细节值得展开：

1. factory 与 ORM 查询的分工：造数用 self.create_organization() 等 factory（silo-aware，内部处理 silo 上下文），因此无需包裹；只有直接读对端副本表（如 {ReplicaModel}）时才需要 assume_test_silo_mode_of({ReplicaModel})，让查询落到模型实际所在的 silo。

2. 幂等性是业务要求而非测试技巧：从实现看，outbox 是会被合并（coalesced）且可能重试的——payload_for_update 的注释明确指出“not every payload generated is guaranteed to be processed”，handle_async_replication / handle_async_deletion 也要求在重复调用下安全（base.py）。因此“排水两次无重复副作用”不是可有可无的断言，而是设计契约。仓库对排水收敛性有专门的验证用例 test_outbox_converges（test_outbox.py）：即使同一条消息被重复写入多次，最终也只派发一次信号。

六、模板三：验证 outbox 信号以正确参数派发

drain_shard 最终会调用 process_cell_outbox 或 process_control_outbox（见 outbox.py 的 process() 与 send_signal 实现）。因此可以 mock 掉信号发送方法，精确断言“传给信号接收方”的参数：

    @patch("sentry.hybridcloud.models.outbox.process_cell_outbox.send")
    def test_outbox_sends_correct_signal(self, mock_send):
        """Verify the outbox signal fires with correct arguments."""
        org = self.create_organization()

        with outbox_context(flush=False):
            Organization(id=org.id).outbox_for_update().save()

        CellOutbox.objects.filter(
            shard_identifier=org.id,
        ).first().drain_shard()

        mock_send.assert_called_with(
            sender=OutboxCategory.{CATEGORY},
            payload=None,
            object_identifier=org.id,
            shard_identifier=org.id,
            shard_scope=OutboxScope.{SCOPE},
        )

对照实现，CellOutboxBase.send_signal()（outbox.py）发送的正是这组关键字参数。注意两点：

• sender 传的是 OutboxCategory 枚举本身（不是 .value），断言时不要画蛇添足；
• ControlOutboxBase.send_signal() 会额外携带 cell_name、date_added、scheduled_for（outbox.py），写 Control 方向测试时若用 @patch("sentry.hybridcloud.models.outbox.process_control_outbox.send")，断言参数需相应增加。

该模式的仓库级证据可参考 test_outbox_rescheduling 中的 assert_called_for_org（test_outbox.py），它同样断言了 sender=OutboxCategory.ORGANIZATION_UPDATE, payload=None, object_identifier=org, shard_identifier=org, shard_scope=OutboxScope.ORGANIZATION_SCOPE 的精确组合。

七、模板四：验证 shard 调度

因为 outbox 按 (shard_scope, shard_identifier) 分片、且不同分片可并行，测试里经常需要确认“恰好哪些分片被安排处理”。find_scheduled_shards()（outbox.py）会聚合所有 scheduled_for <= now 的记录，按分片键去重并返回行映射。

    def test_scheduled_shards(self):
        """Verify correct shards are scheduled for processing."""
        org1 = self.create_organization()
        org2 = self.create_organization()

        with outbox_context(flush=False):
            Organization(id=org1.id).outbox_for_update().save()
            Organization(id=org2.id).outbox_for_update().save()

        shards = {
            (row["shard_scope"], row["shard_identifier"])
            for row in CellOutbox.find_scheduled_shards()
        }
        assert shards == {
            (OutboxScope.ORGANIZATION_SCOPE.value, org1.id),
            (OutboxScope.ORGANIZATION_SCOPE.value, org2.id),
        }

这里有一个很有价值的语义：两个不同组织产生两条 outbox，但 shard 集合只有两个（各自组织）而非四条；反之，若同一 shard 内积压 3 条消息，调度结果仍只有一个 shard——这正是“coalesced + 按 shard 排水”设计的体现。真实用例见 test_outbox.py（test_cell_sharding_keys）。如果写的是 ControlOutbox 的测试，由于 Control 侧的分片列多一个 cell_name（见 outbox.py），find_scheduled_shards() 返回的 key 中会额外出现 cell_name。

八、模板五：验证删除通过 outbox 传播

“删除传播”是 outbox 模式在数据一致性上最重要的使用场景之一：源端对象删除后，对端副本或映射记录应当被异步清理。此时要先建立映射、再删除、再验证，形成完整闭环：

    def test_delete_propagates_via_outbox(self):
        """Verify deleting an object propagates to the other silo via outbox."""
        # Create objects using factories (no silo wrapper needed)
        org = self.create_organization()
        member = self.create_member(
            organization=org,
            user=self.create_user(),
        )

        # Ensure mapping exists first
        with outbox_runner():
            pass
        with assume_test_silo_mode_of({MappingModel}):
            assert {MappingModel}.objects.filter(
                organizationmember_id=member.id,
            ).exists()

        # Delete and drain
        with outbox_runner():
            member.delete()

        # Verify mapping is gone
        with assume_test_silo_mode_of({MappingModel}):
            assert not {MappingModel}.objects.filter(
                organizationmember_id=member.id,
            ).exists()

注意 member.delete() 被包在 outbox_runner() 里面执行——删除与排水必须严格先后发生，因此要放在同一 with 块内按顺序驱动。这里工厂创建的 member 是真实落库对象，delete() 经由 ReplicatedCellModel 的 delete 覆写（见 base.py，outbox_before_super=True）先生成删除类 outbox，排水时接收方通过 maybe_process_tombstone 判断对象已不存在，进而走 handle_async_deletion 清理对端资源（接线逻辑见 category.py 的 connect_cell_model_updates / connect_control_model_updates）。

九、关键模式速查与红线规则

outbox-tests.md 末尾提炼的结论，几乎每条都能在上文源码中找到对应依据，值得作为日常编写的 checklist 常备：

• outbox_context(flush=False)：只造 outbox 记录、不处理它们，用于验证“创建”环节。
• outbox_runner()：同步处理所有待处理 outbox。配合 TestCase 使用即可，不需要升级为 TransactionTestCase。
• assume_test_silo_mode_of(Model)：跨 silo 检查单个模型状态的首选，自动探测模型所属 silo（依据 @cell_silo_model / @control_silo_model，见 silo.py）。
• assume_test_silo_mode(SiloMode.X)：仅当代码块涉及多个模型或非模型资源时才手动指定 silo。
• Factory 调用永不包裹：self.create_organization()、self.create_member()、self.create_user() 内部自行处理 silo 上下文，套上 assume_test_silo_mode 反而会破坏其上下文切换逻辑。仓库级约束见 SKILL.md 的 Critical Constraints 与 Step 5。
• 装饰器方向匹配：@control_silo_test 管 ControlOutbox 场景，@cell_silo_test 管 CellOutbox 场景，二者均以 TestCase 为基类。
• TransactionTestCase 的唯一用途是线程/并发测试（如 threading.Barrier），常规排水测试用它反而会带来不必要的复杂度。
• 隔离性 fixture：当多个测试共享环境、担心脏数据残留时，可声明 autouse fixture，在每个用例开始前先把积压 outbox 排空：

  @pytest.fixture(autouse=True, scope="function")
  def setup_clear_outbox():
      with outbox_runner():
          pass

十、命名与运行：让测试可维护、可执行

• 类与方法命名：采用仓库风格 Test{Feature}Outbox / Test{Feature}OutboxProcessing，方法用 test_<action>_<scenario> 描述性命名（如 test_outbox_drain_is_idempotent）。这一步可以在 Step 7 验证清单中核对（SKILL.md）。
• 放置位置：遵循镜像路径约定 src/sentry/foo/bar.py → tests/sentry/foo/test_bar.py。若被测模块已有测试文件，优先追加而非新建。
• 运行命令：

pytest -svv --reuse-db tests/sentry/path/to/test_file.py

总结

Outbox 测试的完整闭环可以浓缩为四个动作：在 outbox_context(flush=False) 下验证记录生成 → 在 outbox_runner() 下驱动排水 → 用 assume_test_silo_mode_of(Model) 检查对端副作用 → 用 mock 断言信号参数或再次排水验证幂等。而这一切之所以能够成立，源于 outbox.py 中“分片 + 合并 + 指数退避调度”的实现设计与 testutils/outbox.py 中 outbox_runner 对任务派发队列的同步封装。无论你是要为新接入的复制模型补齐覆盖，还是排查一个“为什么排水后对端没变化”的疑难测试，都可以直接以本文模板为起点，再对照 test_outbox.py 中的既有真实用例校准细节。