hybrid-cloud-test-gen

混合云测试生成

本技能为 Sentry 的混合云架构生成测试。它涵盖 RPC 服务、API 网关代理、发件箱模式和端点 silo 装饰器。

关键约束

始终使用工厂方法（self.create_user()、self.create_organization()）——切勿使用 Model.objects.create()。

切勿将工厂方法调用包裹在 assume_test_silo_mode 或 assume_test_silo_mode_of 中。工厂方法可感知 silo，并在内部处理 silo 模式。仅对直接 ORM 查询（Model.objects.get/filter/count/exists/delete）使用 silo 模式上下文管理器。

始终使用 pytest 风格的断言（assert x == y）——切勿使用 self.assertEqual()。

始终将测试添加到现有测试文件，而不是创建新文件，除非该模块不存在任何文件。

对于跨 silo 的 ORM 访问：在访问单个模型时使用 assume_test_silo_mode_of(Model)（自动检测 silo）。当代码块涵盖多个模型或非模型操作时，使用 assume_test_silo_mode(SiloMode.X)。

对大多数测试使用 TestCase，包括使用 outbox_runner() 的测试。仅当测试需要真正已提交的事务（线程、并发、多进程场景）时才使用 TransactionTestCase。

切勿在处理 RPC 模型的测试文件中使用 from __future__ import annotations。

步骤 1：确定测试类别

根据用户的请求确定要生成哪一类 HC 测试：

信号	类别	前往
RPC 服务、服务方法、序列化往返、分发	RPC 服务测试	步骤 3
API 网关、代理、中间件、转发	API 网关测试	步骤 4
发件箱、跨 silo 消息、ControlOutbox、CellOutbox、发件箱排空	发件箱模式测试	步骤 5
带 silo 装饰器的 API 端点、端点测试、权限检查	端点 Silo 测试	步骤 6

如果信号不明确，请让用户澄清属于哪个类别。

步骤 2：收集上下文

在生成任何测试之前：

1. 阅读被测试的源模块。通过检查 @cell_silo_endpoint、@control_silo_endpoint、local_mode = SiloMode.X 或 @cell_silo_model/@control_silo_model 装饰器来确定其 silo 模式。

2. 使用镜像路径约定找到现有测试文件：

   • src/sentry/foo/bar.py → tests/sentry/foo/test_bar.py
   • src/sentry/foo/services/bar/service.py → tests/sentry/foo/services/test_bar.py
   • src/sentry/foo/services/bar/impl.py → tests/sentry/foo/services/test_bar.py

3. 阅读现有测试文件，了解已测试的内容、使用的基类以及已建立的模式。

4. 阅读源方法签名，了解参数、返回类型以及涉及哪些 RPC 模型。

步骤 3：生成 RPC 服务测试

加载 references/rpc-service-tests.md 以获取完整的模板和模式。

RPC 服务测试必须涵盖：

• Silo 兼容性：@all_silo_test 确保服务在所有 silo 模式下均可工作
• 序列化往返：dispatch_to_local_service 验证参数/返回值在序列化后保持不变
• 字段准确性：将 RPC 模型与 ORM 对象逐字段比较
• 错误处理：未找到返回、已禁用方法、远程异常包装
• 跨 silo 效应：outbox_runner() + assume_test_silo_mode 用于传播检查

快速参考 — 装饰器与基类

场景	装饰器	基类
标准 RPC 服务	@all_silo_test	TestCase
带命名 cell 的 RPC	@all_silo_test(cells=create_test_cells("us"))	TestCase
带成员映射断言的 RPC	@all_silo_test	TestCase, HybridCloudTestMixin

步骤 4：生成 API Gateway 测试

加载 references/api-gateway-tests.md 获取完整模板和模式。

API gateway 测试验证发往 control-silo 端点的请求是否被正确代理到相应的 cell。它们必须覆盖：

• 代理透传：请求转发时带有正确的 params、headers、body
• 查询参数转发：多值参数被保留
• 错误代理：上游错误被正确转发
• 流式响应：使用 close_streaming_response() 读取被代理的响应体

快速参考 — 装饰器与基类

场景	装饰器	基类
标准 gateway 测试	@control_silo_test(cells=[ApiGatewayTestCase.CELL], include_monolith_run=True)	ApiGatewayTestCase

步骤 5：生成 Outbox 模式测试

加载 references/outbox-tests.md 获取完整模板和模式。

Outbox 测试验证跨 silo 消息是否被创建、排空并产生预期的副作用。它们必须覆盖：

• Outbox 创建：使用 outbox_context(flush=False) 验证正确的 outbox 记录
• Outbox 处理：outbox_runner() 排空待处理消息
• 跨 silo 副作用：使用 assume_test_silo_mode_of(Model) 检查副本/映射状态
• 幂等性：对同一 shard 排空两次不会产生重复

快速参考 — 装饰器与基类

场景	装饰器	基类
Control outbox 测试	@control_silo_test	TestCase
Cell outbox 测试	@cell_silo_test	TestCase
带线程/并发的 Outbox	(无)	TransactionTestCase

步骤 6：生成端点 Silo 测试

加载 references/endpoint-silo-tests.md 获取完整模板和模式。

端点 silo 测试验证 API 端点在声明的 silo 模式下是否正常工作。它们必须覆盖：

• 正确的 silo 装饰器：匹配端点 → 测试装饰器
• 跨 silo 数据设置：使用工厂方法创建数据（无需 silo 包装器）
• 权限检查：验证未授权访问的 401/403
• 响应准确性：验证响应体与预期数据匹配

快速参考 — 装饰器映射

端点装饰器	测试装饰器
@cell_silo_endpoint	@cell_silo_test
@control_silo_endpoint	@control_silo_test
@control_silo_endpoint（带代理）	@control_silo_test(cells=create_test_cells("us"))
无装饰器（仅 monolith）	@no_silo_test

步骤 7：验证

在提交生成的测试之前，请对照以下检查清单进行验证：

• 测试类上的 silo 装饰器正确
• 单模型 ORM 访问使用 assume_test_silo_mode_of(Model)；多模型/非模型 ORM 块使用 assume_test_silo_mode(SiloMode.X)
• 工厂方法（self.create_*）绝不包裹在 assume_test_silo_mode 中
• 使用工厂方法 —— 绝不使用 Model.objects.create()
• 仅使用 pytest 风格的断言（assert x == y）
• 基类正确（大多数测试使用 TestCase；仅线程/并发场景使用 TransactionTestCase）
• 导入正确且精简
• 测试文件位于正确的镜像路径
• 测试方法具有描述性名称（test_<action>_<scenario>）
• 运行命令：pytest -svv --reuse-db tests/sentry/path/to/test_file.py

关键导入速查

# Silo decorators
from sentry.testutils.silo import (
    all_silo_test,
    control_silo_test,
    cell_silo_test,
    no_silo_test,
    assume_test_silo_mode,
    assume_test_silo_mode_of,
    create_test_cells,
)

# Base classes
from sentry.testutils.cases import TestCase, TransactionTestCase, APITestCase

# Cross-silo utilities
from sentry.testutils.outbox import outbox_runner
from sentry.testutils.hybrid_cloud import HybridCloudTestMixin
from sentry.silo.base import SiloMode

# RPC testing
from sentry.hybridcloud.rpc.service import dispatch_to_local_service

# API gateway testing
from sentry.testutils.helpers.apigateway import ApiGatewayTestCase, verify_request_params

# Outbox models
from sentry.hybridcloud.models.outbox import ControlOutbox, CellOutbox, outbox_context
from sentry.hybridcloud.outbox.category import OutboxCategory, OutboxScope

上下文管理器速查

# Use ONLY for direct ORM queries — never for factory calls
assume_test_silo_mode(SiloMode.CONTROL)     # Switch to control silo for ORM access
assume_test_silo_mode(SiloMode.CELL)        # Switch to cell silo for ORM access
assume_test_silo_mode_of(ModelClass)        # Switch to silo matching model's silo mode

outbox_runner()                             # Drain all pending outboxes on exit
outbox_context(flush=False)                 # Create outboxes without flushing
override_cells(cells)                       # Override active cell config
override_settings(SILO_MODE=SiloMode.X)     # Override Django settings
override_options({"key": value})            # Override Sentry options