moto 中 AWS Config 服务的实现全解：已支持 API、Recorder/Aggregator 机制与源码级剖析

Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载本文以 docs/docs/services/config.rst 这份服务实现清单为核心系统梳理 moto 对 AWS ConfigConfiguration服务的模拟能力哪些 API 已实现、哪些尚未覆盖、各 API 在源码中的行为约束命名规则、区域校验、分页上限、异常语义以及list_discovered_resources、select_resource_config等查询类API 如何跨服务读取其他 moto 后端的资源。读完本文你可以准确评估 moto Config 模拟对测试场景的支撑边界并能直接编写可运行的mock_aws测试代码。已实现 / 未实现 API 总览docs/docs/services/config.rst 以复选框形式列出 AWS Config 全部 API 的实现状态。这是判断某个测试场景能否被 mock的第一依据。已实现文档中标记 [X]的 APIAPI文档中的行为说明put_configuration_recorder创建/更新配置记录器默认名称defaultdescribe_configuration_recorders返回指定名称或全部记录器describe_configuration_recorder_status返回记录器的录制状态start_configuration_recorder/stop_configuration_recorder启动/停止录制delete_configuration_recorder删除记录器put_delivery_channel创建投递通道describe_delivery_channels/delete_delivery_channel查询/删除投递通道put_configuration_aggregator/delete_configuration_aggregator配置聚合器describe_configuration_aggregators查询聚合器分页put_aggregation_authorization/delete_aggregation_authorization/describe_aggregation_authorizations聚合授权put_config_rule添加/更新合规规则目前只做规则的记账不产生事件、不触发评估、与 recorder 无交互describe_config_rules/delete_config_rule查询/删除合规规则list_discovered_resources非聚合资源列表查询资源类型后端必须已实现对应 listinglist_aggregate_discovered_resources聚合资源列表查询要求事先创建 Config Aggregator且可按资源区域过滤get_resource_config_history返回单条资源配置AWS Config 格式。注意并不返回历史——moto 暂不支持历史later_time、earlier_time、limit、next_token均被忽略只返回 1 条查不到则抛异常batch_get_resource_config批量返回当前区域后端的资源配置AWS Config 格式参数resource_keys、backend_regionbatch_get_aggregate_resource_config聚合版批量查询要求事先有 Aggregatormoto忽略查询中的资源账户 IDselect_resource_config不真正执行 SQL而是使用可通过 moto API 配置好的预定义结果仿照 AWS Athena 的思路文档注明还需实现分页put_evaluations文档未附说明源码中当前仅支持 TestMode见下文put_organization_conformance_pack/describe_organization_conformance_packs/describe_organization_conformance_pack_statuses/get_organization_conformance_pack_detailed_status/delete_organization_conformance_pack组织合规包管理put_resource_config/delete_resource_config写入/删除自定义资源配置put_retention_configuration/describe_retention_configurations/delete_retention_configuration保留期配置describe 最多接收 1 个名称多于 1 个抛ValidationExceptiontag_resource/untag_resource/list_tags_for_resource按 ARN 匹配为 Config 资源打标签/删标签/列标签untag_resource中若tag_keys与资源已有键不匹配则直接忽略尚未实现[ ]的 API包括associate_resource_types、deliver_config_snapshot、各类describe_*_compliance*合规评估类 API如describe_compliance_by_config_rule、describe_config_rule_evaluation_status、get_discovered_resource_counts、get_stored_query、put_conformance_pack、put_connector、remediation 系列put_remediation_configurations、start_remediation_execution等、select_aggregate_resource_config、start_config_rules_evaluation、start_resource_evaluation、put_service_linked_configuration_recorder等。如果你的测试依赖合规评估结果compliance details/summary或 remediation 流程当前无法用 moto 覆盖。请求入口从 URL 路由到后端实例Config 服务的请求经 moto/config/urls.py 路由url_bases [rhttps?://config\.(.)\.amazonaws\.com] url_paths {{0}/$: ConfigResponse.dispatch}moto/config/responses.py 中的ConfigResponse是请求分发层每个 boto3 方法名对应一个同名 handler如put_configuration_recorder、batch_get_aggregate_resource_confighandler 用self._get_param(...)解析 JSON 参数后委托给 moto/config/models.py 中ConfigBackend的方法。注意两个细节后端是按账户 区域隔离的config_backends BackendDict(ConfigBackend, config)moto/config/models.pyConfigResponse.config_backend属性返回config_backends[self.current_account][self.region]moto/config/responses.py#L13-L15。因此多区域测试中在us-west-2创建的 recorder 不会出现在eu-west-1。describe_delivery_channel_status在 responses 层直接raise NotImplementedError()moto/config/responses.py#L93-L94与文档中该 API 未标记实现的清单一致。配置记录器与投递通道状态机与前置约束ConfigBackend在初始化时维护两组单例性质的字典self.recorders与self.delivery_channelsmoto/config/models.py#L912-L913对应 AWS Config每账户各限 1 个的约束。put_configuration_recorder 的校验链put_configuration_recordermoto/config/models.py#L1153的校验逻辑非常细直接决定了测试中请求参数怎么填名称缺省名抛InvalidConfigurationRecorderNameException长度超过 256 抛ValidationExceptionNameTooLongException。数量上限len(self.recorders) 1且新名称不同时抛MaxNumberOfConfigurationRecordersExceededException。RecordingGroup 三种互斥策略源码 moto/config/models.py#L1185-L1278allSupported: True不允许同时提供resourceTypes或非空的exclusionByResourceTypesrecordingStrategy.useOnly必须是ALL_SUPPORTED_RESOURCE_TYPES或省略resourceTypes: [...]不允许提供includeGlobalResourceTypes: True或排除清单策略对应INCLUSION_BY_RESOURCE_TYPES且每个资源类型必须通过_validate_resource_types——它用 boto3 服务模型中ResourceType形状的enum校验moto/config/models.py#L925-L934不合法类型抛InvalidResourceTypeExceptionexclusionByResourceTypes必须同时提供recordingStrategy: {useOnly: EXCLUSION_BY_RESOURCE_TYPES}且排除列表非空。任何字段组合冲突抛InvalidRecordingGroupException。若未提供recordingGroup则使用ALL_SUPPORTED_RESOURCE_TYPES全量策略兜底。tests/test_config/test_config.py 的test_put_configuration_recorder用 15 种非法recordingGroup组合逐一断言InvalidRecordingGroupException是编写参数边界测试的直接参照。启动/删除的前置依赖start_configuration_recordermoto/config/models.py#L1407-L1416必须先有投递通道否则抛NoAvailableDeliveryChannelException启动后ConfigRecorderStatus.start()将recordingTrue、lastStatusPENDING并记录时间戳。put_delivery_channelmoto/config/models.py#L1325必须先存在配置记录器NoAvailableConfigurationRecorderExceptions3BucketName为空抛NoSuchBucketExceptions3KeyPrefix/snsTopicARN/s3KmsKeyArn显式传空字符串分别抛InvalidS3KeyPrefixException/InvalidSNSTopicARNException/InvalidS3KmsKeyArnException源码注释特别提到 SNS 用全大写 ARN 而 KMS 用 Arn 的大小写差异configSnapshotDeliveryProperties.deliveryFrequency必须命中服务模型枚举否则InvalidDeliveryFrequency同样限 1 个通道。delete_delivery_channelmoto/config/models.py#L1431-L1440若有 recorder 正处于 recording 状态删除失败抛LastDeliveryChannelDeleteFailedException——即必须stop_configuration_recorder之后才能删通道。基于以上约束一个可运行的最小工作流如下参数命名与取值均来自源码校验逻辑与 tests/test_config/test_config.py 中的用例import boto3 from moto import mock_aws mock_aws def test_config_flow(): s3 boto3.client(s3, region_nameus-west-2) s3.create_bucket(Bucketmy-bucket) config boto3.client(config, region_nameus-west-2) # 1. 记录器全量支持策略可省略 recordingGroup使用默认 ALL_SUPPORTED_RESOURCE_TYPES config.put_configuration_recorder(ConfigurationRecorder{ name: default, roleARN: arn:aws:iam::123456789012:role/ConfigRole, recordingGroup: {allSupported: True}, }) # 2. 投递通道s3BucketName 不能为空 config.put_delivery_channel(DeliveryChannel{ name: default, s3BucketName: my-bucket, configSnapshotDeliveryProperties: {deliveryFrequency: Daily}, }) # 3. 启动此时通道必须已存在 config.start_configuration_recorder(default) # 4. 列出已发现的 S3 桶 resp config.list_discovered_resources(resourceTypeAWS::S3::Bucket) assert any(i[resourceId] my-bucket for i in resp[resourceIdentifiers])聚合器与聚合查询跨区域资源的全量聚合假设put_configuration_aggregator 的互斥校验put_configuration_aggregatormoto/config/models.py#L948-L1032要求名称不超过 256AccountAggregationSources与OrganizationAggregationSource二选一且必须提供其一同时提供抛InvalidParameterValueExceptionAccountAggregationSources当前最多 1 个超过抛TooManyAccountSources每个 source 的AwsRegions与AllAwsRegions互斥且必居其一见AccountAggregatorSource/OrganizationAggregationSource构造器moto/config/models.py#L339-L351同名再次调用时就地更新tags、sources 并刷新last_updated_time而不是新建。聚合器的 ARN 由 moto 随机生成 8 位小写字母后缀config-aggregator-{random_string()}moto/config/models.py#L407。聚合查询的语义差异文档中强调的两条语义在源码中一一对应必须事先有 Aggregatorlist_aggregate_discovered_resources与batch_get_aggregate_resource_config开头都是if not self.config_aggregators.get(aggregator_name): raise NoSuchConfigurationAggregatorException()moto/config/models.py#L1551-L1552、moto/config/models.py#L1727-L1728。moto 忽略资源账户 IDbatch_get_aggregate_resource_config的 docstring 明确 moto will IGNORE the resource account ID in the search query源码中只取SourceRegion/ResourceId/ResourceName进行查询。区域过滤方面list_aggregate_discovered_resources只从Filters中取Region、ResourceId、ResourceName三个键moto/config/models.py#L1566-L1569并调用资源类型的list_config_service_resources(..., resource_region..., aggregator...)。从 moto/core/common_models.py 中ConfigQueryModel基类的契约文档看聚合查询在 moto 中的假设是该资源类型在所有区域后端的全量聚合聚合方先汇集所有区域后端的全部资源再按resource_region过滤——这解释了为什么聚合 API 不需要真实的跨账户配置。资源类型到后端的桥接RESOURCE_MAP 与 ConfigQueryModelConfig 服务之所以能看见其他服务的资源靠的是 moto/config/models.py#L75-L81 中的资源类型注册表RESOURCE_MAP: dict[str, ConfigQueryModel[Any]] { AWS::S3::Bucket: s3_config_query, AWS::S3::AccountPublicAccessBlock: s3_account_public_access_block_query, AWS::IAM::Role: role_config_query, AWS::IAM::Policy: policy_config_query, AWS::SNS::Topic: sns_config_query, }每个条目指向对应服务包中的ConfigQueryModel子类如moto/s3/config.py、moto/iam/config.py、moto/sns/config.py它们实现了两个抽象方法定义于 moto/core/common_models.py#L99-L199list_config_service_resources(account_id, partition, resource_ids, resource_name, limit, next_token, backend_regionNone, resource_regionNone, aggregatorNone)——返回{type, name, id, region}结构的标识列表 下一页 tokenget_config_resource(account_id, partition, resource_id, resource_nameNone, backend_regionNone, resource_regionNone)——返回单条 AWS Config 格式的资源配置。这直接决定了查询类 API 的能力边界list_discovered_resources、get_resource_config_history、batch_get_*只对RESOURCE_MAP中登记的 5 种资源类型有数据查询未登记类型例如AWS::EC2::Instance时list_discovered_resources返回空列表而get_resource_config_history抛ResourceNotDiscoveredExceptionmoto/config/models.py#L1618-L1620。list_discovered_resourcesmoto/config/models.py#L1442-L1527的其余校验值得在测试中利用limit缺省 100DEFAULT_PAGE_SIZE超过 100 抛InvalidLimitExceptionresource_ids与resource_name不能同时提供InvalidResourceParametersresource_ids最多 20 个TooManyResourceIds全局资源类型如 IAM会自动改用 partition 作为后端区域backend_region self.partition。batch_get_resource_config与聚合版的差异moto/config/models.py#L1651-L1772行为batch_get_resource_configbatch_get_aggregate_resource_config上限resource_keys超过 100 抛TooManyResourceKeysresource_identifiers超过 100 同样抛TooManyResourceKeys未命中项静默跳过unprocessedResourceKeys恒为[]收集进UnprocessedResourceIdentifierstags保留会pop(tags, None)源码注释聚合结果中不返回 tags区域用请求到达区域全局类型用 partition用identifier[SourceRegion]Config Rule受管规则校验与 Lambda 关联put_config_rulemoto/config/models.py#L2003-L2065支持按名称、ARN 或 ID 定位既有规则进行更新创建新规则时若数量达到ConfigRule.MAX_RULES 150抛MaxNumberOfConfigRulesExceededException。ConfigRule构造与modify_fieldsmoto/config/models.py#L726-L839的关键校验创建时不允许自带ConfigRuleArn/ConfigRuleId由服务生成config-rule-{6位随机}ConfigRuleState仅接受ACTIVERULE_STATES枚举含DELETING等状态用于内部流转CreatedBy字段不能由用户提供InputParameters必须是合法 JSON否则InvalidParameterValueException。受管规则OwnerAWSSource.owner必须是AWS或CUSTOM_LAMBDAmoto/config/models.py#L640-L677。AWS 受管规则要求sourceIdentifier存在于aws_managed_rules.json由 scripts/pull_down_aws_managed_rules.py 从 AWS 拉取存放于 moto/config/resources/aws_managed_rules.json且SourceDetails必须为空随后validate_managed_rule()moto/config/models.py#L841会校验InputParameters的参数名集合未知参数名或必填参数缺失都会抛InvalidParameterValueException。自定义规则OwnerCUSTOM_LAMBDA必须提供SourceDetails且sourceIdentifierLambda ARN/名称会被真的拿去查 moto 的 Lambda 后端——get_backend(account_id, region).get_function(source_identifier)查不到时抛InsufficientPermissionsExceptionmoto/config/models.py#L688-L697。SourceDetailmoto/config/models.py#L542-L634进一步校验EventSource只允许aws.configMessageType必须是ConfigurationItemChangeNotification/ConfigurationSnapshotDeliveryCompleted/OversizedConfigurationItemChangeNotification/ScheduledNotification之一maximumExecutionFrequency限定在One_Hour…TwentyFour_Hours五个枚举且变更通知类消息类型不允许设频率抛InvalidParameterValueException快照/定时类缺省时默认TwentyFour_Hours。与文档TBD注记一致规则目前只做记账CRUD describedescribe_config_rules每页固定 25 条CONFIG_RULE_PAGE_SIZE不会与 recorder 联动、不产生评估事件。put_evaluations、select_resource_config 与自定义资源put_evaluationsmoto/config/models.py#L1774-L1798Evaluations为空抛参数异常ResultToken缺失抛InvalidResultTokenException非 TestMode 调用直接NotImplementedErrorTestMode 下返回空的FailedEvaluations。select_resource_configmoto/config/models.py#L2175-L2204不解析也不执行 SQL而是按以下顺序取预定义结果若expression命中后端的_expression_results字典表达式 → 结果行列表直接返回否则从query_results_queueFIFO 队列弹出一个结果集都没有则返回空结果。结果行的列名由首行 dict 的键推导为QueryInfo.SelectFields并生成一个随机query_id记账。tests/test_config/test_config.py 的test_select_resource_config_with_expression_results展示了用法先通过from moto.config.models import config_backends拿到config_backends[account][us-east-1]backend.query_results_queue.append(predefined_results)再调用select_resource_config(Expression...)断言返回注意该测试在 server 模式下会被SkipTest跳过因为装饰器模式下才能直接访问与客户端相同的后端实例。put_resource_config / delete_resource_configmoto/config/models.py#L2206-L2245用于写入自定义资源类型如MyCorp::WidgetResourceType前缀不允许是amzn/amazon/alexa/custom抛InvalidResourceType资源以{resource_type}:{resource_id}为键保存在_custom_resources中。保留期配置与标签管理保留期moto/config/models.py#L2114-L2173put_retention_configuration要求retentionPeriodInDays在30 ~ 2557天之间越界抛带 AWS 原样措辞的ValidationException后端同一时间只保存一个RetentionConfiguration名称缺省defaultdescribe_retention_configurations与文档描述一致最多接受 1 个名称传多个抛ValidationException名称不存在抛NoSuchRetentionConfigurationException不传名称则返回当前配置无配置返回空列表delete_retention_configuration名称不匹配同样抛NoSuchRetentionConfigurationException。标签moto/config/models.py#L1958-L2001tag_resource/untag_resource通过_match_arn在ConfigRule、ConfigurationAggregator、AggregationAuthorization三类资源的 ARN 中匹配moto/config/models.py#L1933-L1956匹配不到抛ResourceNotFoundExceptiontag_resource是合并语义tags.updateuntag_resource对不存在的键静默忽略。标签校验函数validate_tagsmoto/config/models.py#L151-L168统一执行单次最多 50 个标签TooManyTags、key 不超过 128 字符且字符集受限TagKeyTooBig/InvalidTagCharacters、value 不超过 256TagValueTooBig、重复 key 抛DuplicateTags。list_tags_for_resource返回按 key 排序的标签limit上限 100。组织合规包put_organization_conformance_packmoto/config/models.py#L1800-L1846校验TemplateS3Uri与TemplateBody至少提供一个TemplateS3Uri必须匹配正则s3://.*源码注释注明目前缺少对模板内容的真实校验。创建/更新返回OrganizationConformancePackArn状态字段CREATE_SUCCESSFUL/UPDATE_SUCCESSFUL由get_organization_conformance_pack_detailed_status返回目前详细状态只包含当前账户一条记录。查询类 API 对不存在的包名抛NoSuchOrganizationConformancePackException。分页、异常与测试验证分页参数行为在全文档中统一DEFAULT_PAGE_SIZE 100describe_config_rules固定 25 条/页CONFIG_RULE_PAGE_SIZE 25NextToken的语义是下一项的名称aggregator/authorization/rule 的 describe 方法中直接sorted_list.index(token)定位非法 token 抛InvalidNextTokenException资源列表类 API 的limit超 100 抛InvalidLimitException。所有异常类集中在 moto/config/exceptions.py错误码均对齐 AWS 真实 API如InvalidRecordingGroupException、NoSuchConfigurationAggregatorException。配套测试位于 tests/test_config/test_config.pyrecorder/delivery channel/aggregator/资源查询/retention/select_resource_config 等主流程与异常路径、test_config_rules.py、test_config_rules_integration.py规则与 Lambda 联动、test_config_tags.py标签管理。编写新测试时建议直接参考这些文件中对ClientError错误码与消息断言的写法。实践要点小结先查清单再写测试以 docs/docs/services/config.rst 的 [X]/[ ] 标记为准合规评估、remediation、connector 等场景当前不可 mock对象创建顺序有硬约束recorder → delivery channel → start recorder删除顺序相反先 stop 再删 channel查询类 API 依赖 RESOURCE_MAP目前仅 S3 桶/S3 账户公网访问块/IAM Role/IAM Policy/SNS Topic 五种类型可查其余类型列表为空或抛ResourceNotDiscoveredException聚合 API 必须先建 Aggregator且 moto 假设全区域全量聚合、忽略账户 ID、聚合批量结果不带 tagsget_resource_config_history 不返回历史select_resource_config不执行 SQL——两者都需要通过后端预置数据query_results_queue/_expression_results来驱动断言区域与账户隔离由BackendDict保证跨区域测试要使用对应区域的客户端与后端。赞分享Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载相关推荐moto 中 CloudFront Mock 的实现全景:已支持 API、内部机制与已知限制moto 中 CloudFront Mock 的实现全景:已支持 API、内部机制与已知限制 本文以 moto 仓库中 CloudFront 服务的实现清单文档Mock测试moto 中的 AWS CodeDeploy 模拟已实现 API 清单、源码实现解析与 boto3 测试实战moto 中的 AWS CodeDeploy 模拟已实现 API 清单、源码实现解析与 boto3 测试实战 本文以 moto 官方服务文档 codedeplMock测试moto 中 CodePipeline 服务的 Mock 实现功能覆盖清单与源码级行为解析moto 中 CodePipeline 服务的 Mock 实现功能覆盖清单与源码级行为解析 本篇基于 moto 仓库中的服务文档 codepipeline.rMock测试上一篇Drawnix深度解析从插件化架构到企业级白板应用的实战指南下一篇常见问题解决方案rumqtt 项目创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考