使用moto模拟跨账号EventBridge目标规则时遇NotImplementedError

问题背景 #

用moto模拟AWS服务时，同一账号内将SQS设为EventBridge目标没问题，但尝试把Account2的事件总线设为Account1中EventBridge的目标时，调用events_client.put_events抛出了NotImplementedError。

错误详情 #

def send_to_targets(self, event_bus_name, event):
    event_bus_name = event_bus_name.split("/")[-1]
    if event_bus_name != self.event_bus_name.split("/")[-1]:
        return

    if not self.event_pattern.matches_event(event):
        return

    # supported targets
    # - CloudWatch Log Group
    # - EventBridge Archive
    # - SQS Queue + FIFO Queue
    for target in self.targets:
        arn = self._parse_arn(target["Arn"])

        if arn.service == "logs" and arn.resource_type == "log-group":
            self._send_to_cw_log_group(arn.resource_id, event)
        elif arn.service == "events" and not arn.resource_type:
            input_template = json.loads(target["InputTransformer"]["InputTemplate"])
            archive_arn = self._parse_arn(input_template["archive-arn"])

            self._send_to_events_archive(archive_arn.resource_id, event)
        elif arn.service == "sqs":
            group_id = target.get("SqsParameters", {}).get("MessageGroupId")
            self._send_to_sqs_queue(arn.resource_id, event, group_id)
        else:
            raise NotImplementedError(f"Expr not defined for {type(self)}")
NotImplementedError: Expr not defined for <class 'moto.events.models.Rule'>

触发错误的测试代码 #

def test_put_events_event_bus_forwarding_rules(events_client, sqs_test, create_event_bus, roles):
pattern = {
    "source": ["source1"],
    "detail-type": ["test-detail-type"],
    "detail": {
        "test": [{"exists": True}],
    },
}

os.environ["MOTO_ACCOUNT_ID"] = "222222222222"

events_client.put_rule(
    Name='event_bus_2_rule',
    EventPattern=json.dumps(pattern),
    State="ENABLED",
    EventBusName='event_bus_2',
)

events_client.put_targets(
    Rule='event_bus_2_rule',
    EventBusName='event_bus_2',
    Targets=[
        {
            "Id": "sqs-dedup-fifo",
            "Arn": "arn:aws:sqs:us-east-1:222222222222:test-queue.fifo",
            "SqsParameters": {"MessageGroupId": "group-id"},
        }
    ],
)

policy = {
    "Statement": [
        {
            "Sid": "Account1",
            "Action": "events:PutEvents",
            "Principal": "123456789012",
            "StatementId": "Account1",
            "Effect": "n/a",
            "Resource": "n/a",
        }
    ]
}

events_client.put_permission(EventBusName='event_bus_2', Policy=json.dumps(policy))

del os.environ["MOTO_ACCOUNT_ID"]

events_client.put_rule(
    Name='event_bus_1_rule',
    RoleArn='arn:aws:iam::123456789012:role/Administrator',
    EventPattern=json.dumps(pattern),
    State="ENABLED",
    EventBusName='event_bus_1',
)

events_client.put_targets(
    Rule='event_bus_1_rule',
    EventBusName='event_bus_1',
    Targets=[
        {
            "Id": 'event_bus_2',
            "Arn": 'arn:aws:events:us-east-1:222222222222:event-bus/event_bus_2',
            "RoleArn": 'arn:aws:iam::123456789012:role/Administrator',
        },
    ],
)

test_events = [{
    'Source': 'source1',
    'DetailType': 'test-detail-type',
    'Detail': json.dumps({"test": "true"}),
    'EventBusName': 'event_bus_1'
}]

events_client.put_events(Entries=test_events)

os.environ["MOTO_ACCOUNT_ID"] = "222222222222"

sqs_client = MySQSClient()
queue_url = sqs_client.get_queue_url(queue_name="test-queue.fifo")

response = sqs_client.receive_message(
    queue_url=queue_url,
)

response["Messages"].should.have.length_of(1)

print(response)
del os.environ["MOTO_ACCOUNT_ID"]

解决办法 #

核心原因 #

从moto源码可以看出，当前版本的Rule.send_to_targets方法仅支持三类目标：CloudWatch日志组、EventBridge归档、SQS队列，完全没有处理事件总线类型的目标逻辑，因此遇到这类目标直接抛出异常。

1. 临时手动模拟转发（最快解决） #

既然moto不自动支持跨账号总线转发，就自己在测试代码中补全这个逻辑：发送事件到Account1的总线后，手动将事件推送到Account2的总线。修改后的测试代码片段如下：

# 原有的put_events调用
test_events = [{
    'Source': 'source1',
    'DetailType': 'test-detail-type',
    'Detail': json.dumps({"test": "true"}),
    'EventBusName': 'event_bus_1'
}]
events_client.put_events(Entries=test_events)

# 新增：手动模拟跨账号总线转发
os.environ["MOTO_ACCOUNT_ID"] = "222222222222"
# 初始化Account2的EventBridge客户端
account2_events = boto3.client('events', region_name='us-east-1')
# 把事件发送到Account2的event_bus_2
account2_events.put_events(Entries=[{
    'Source': 'source1',
    'DetailType': 'test-detail-type',
    'Detail': json.dumps({"test": "true"}),
    'EventBusName': 'event_bus_2'
}])

# 后续的SQS接收逻辑不变
sqs_client = MySQSClient()
queue_url = sqs_client.get_queue_url(queue_name="test-queue.fifo")
response = sqs_client.receive_message(queue_url=queue_url)
response["Messages"].should.have.length_of(1)

2. 升级moto到最新版 #

先检查当前使用的moto版本，说不定新版本已经支持了跨账号EventBridge目标功能。执行升级命令：

pip install --upgrade moto

3. 给moto提交PR扩展功能 #

如果需要完整的原生支持，可以给moto项目贡献代码：

• 在send_to_targets方法中添加分支，判断目标ARN是否为事件总线（arn.service == "events"且arn.resource_type == "event-bus"）
• 根据ARN中的账号ID切换moto的账号上下文
• 调用对应账号EventBus的put_events方法完成事件转发

内容的提问来源于stack exchange，提问作者ZZzzZZzz