- Mock
- 测试
【免费下载链接】moto
A library that allows you to easily mock out tests based on AWS infrastructure.
本文以 docs/docs/services/config.rst 这份服务实现清单为核心,系统梳理 moto 对 AWS Config(Configuration)服务的模拟能力:哪些 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])的 API:
| API | 文档中的行为说明 |
|---|---|
put_configuration_recorder | 创建/更新配置记录器(默认名称default) |
describe_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 | 非聚合资源列表查询,资源类型后端必须已实现对应 listing |
list_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_region |
batch_get_aggregate_resource_config | 聚合版批量查询,要求事先有 Aggregator;moto忽略查询中的资源账户 ID |
select_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 个抛ValidationException |
tag_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 = [r"https?://config\.(.+)\.amazonaws\.com"] url_paths = {"{0}/$": ConfigResponse.dispatch}moto/config/responses.py 中的ConfigResponse是请求分发层,每个 boto3 方法名对应一个同名 handler(如put_configuration_recorder、batch_get_aggregate_resource_config),handler 用self._get_param(...)解析 JSON 参数后委托给 moto/config/models.py 中ConfigBackend的方法。注意两个细节:
- 后端是按账户 + 区域隔离的:
config_backends = BackendDict(ConfigBackend, "config")(moto/config/models.py),ConfigResponse.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_channels(moto/config/models.py#L912-L913),对应 AWS Config"每账户各限 1 个"的约束。
put_configuration_recorder 的校验链
put_configuration_recorder(moto/config/models.py#L1153)的校验逻辑非常细,直接决定了测试中请求参数怎么填:
- 名称:缺省名抛
InvalidConfigurationRecorderNameException;长度超过 256 抛ValidationException(NameTooLongException)。 - 数量上限:
len(self.recorders) == 1且新名称不同时抛MaxNumberOfConfigurationRecordersExceededException。 - RecordingGroup 三种互斥策略(源码 moto/config/models.py#L1185-L1278):
allSupported: True:不允许同时提供resourceTypes或非空的exclusionByResourceTypes,recordingStrategy.useOnly必须是ALL_SUPPORTED_RESOURCE_TYPES(或省略);resourceTypes: [...]:不允许提供includeGlobalResourceTypes: True或排除清单,策略对应INCLUSION_BY_RESOURCE_TYPES,且每个资源类型必须通过_validate_resource_types——它用 boto3 服务模型中ResourceType形状的enum校验(moto/config/models.py#L925-L934),不合法类型抛InvalidResourceTypeException;exclusionByResourceTypes:必须同时提供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_recorder(moto/config/models.py#L1407-L1416):必须先有投递通道,否则抛NoAvailableDeliveryChannelException;启动后ConfigRecorderStatus.start()将recording=True、lastStatus="PENDING"并记录时间戳。put_delivery_channel(moto/config/models.py#L1325):必须先存在配置记录器(NoAvailableConfigurationRecorderException);s3BucketName为空抛NoSuchBucketException;s3KeyPrefix/snsTopicARN/s3KmsKeyArn显式传空字符串分别抛InvalidS3KeyPrefixException/InvalidSNSTopicARNException/InvalidS3KmsKeyArnException(源码注释特别提到 SNS 用全大写 "ARN" 而 KMS 用 "Arn" 的大小写差异);configSnapshotDeliveryProperties.deliveryFrequency必须命中服务模型枚举,否则InvalidDeliveryFrequency;同样限 1 个通道。delete_delivery_channel(moto/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_name="us-west-2") s3.create_bucket(Bucket="my-bucket") config = boto3.client("config", region_name="us-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(resourceType="AWS::S3::Bucket") assert any(i["resourceId"] == "my-bucket" for i in resp["resourceIdentifiers"])聚合器与聚合查询:跨区域资源的"全量聚合"假设
put_configuration_aggregator 的互斥校验
put_configuration_aggregator(moto/config/models.py#L948-L1032)要求:
- 名称不超过 256;
AccountAggregationSources与OrganizationAggregationSource二选一,且必须提供其一,同时提供抛InvalidParameterValueException;AccountAggregationSources当前最多 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)。
聚合查询的语义差异
文档中强调的两条语义在源码中一一对应:
- 必须事先有 Aggregator:
list_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 忽略资源账户 ID:
batch_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 与 ConfigQueryModel
Config 服务之所以能"看见"其他服务的资源,靠的是 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-L199):
list_config_service_resources(account_id, partition, resource_ids, resource_name, limit, next_token, backend_region=None, resource_region=None, aggregator=None)——返回{'type', 'name', 'id', 'region'}结构的标识列表 + 下一页 token;get_config_resource(account_id, partition, resource_id, resource_name=None, backend_region=None, resource_region=None)——返回单条 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抛ResourceNotDiscoveredException(moto/config/models.py#L1618-L1620)。
list_discovered_resources(moto/config/models.py#L1442-L1527)的其余校验值得在测试中利用:
limit缺省 100(DEFAULT_PAGE_SIZE),超过 100 抛InvalidLimitException;resource_ids与resource_name不能同时提供(InvalidResourceParameters);resource_ids最多 20 个(TooManyResourceIds);- 全局资源类型(如 IAM)会自动改用 partition 作为后端区域(
backend_region = self.partition)。
batch_get_resource_config与聚合版的差异(moto/config/models.py#L1651-L1772):
| 行为 | batch_get_resource_config | batch_get_aggregate_resource_config |
|---|---|---|
| 上限 | resource_keys超过 100 抛TooManyResourceKeys | resource_identifiers超过 100 同样抛TooManyResourceKeys |
| 未命中项 | 静默跳过,unprocessedResourceKeys恒为[] | 收集进UnprocessedResourceIdentifiers |
| tags | 保留 | 会pop("tags", None)(源码注释:聚合结果中不返回 tags) |
| 区域 | 用请求到达区域;全局类型用 partition | 用identifier["SourceRegion"] |
Config Rule:受管规则校验与 Lambda 关联
put_config_rule(moto/config/models.py#L2003-L2065)支持按名称、ARN 或 ID 定位既有规则进行更新;创建新规则时若数量达到ConfigRule.MAX_RULES = 150抛MaxNumberOfConfigRulesExceededException。ConfigRule构造与modify_fields(moto/config/models.py#L726-L839)的关键校验:
- 创建时不允许自带
ConfigRuleArn/ConfigRuleId(由服务生成config-rule-{6位随机}); ConfigRuleState仅接受ACTIVE(RULE_STATES枚举含DELETING等状态用于内部流转);CreatedBy字段不能由用户提供;InputParameters必须是合法 JSON,否则InvalidParameterValueException。
受管规则(Owner=AWS):Source.owner必须是AWS或CUSTOM_LAMBDA(moto/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。
自定义规则(Owner=CUSTOM_LAMBDA):必须提供SourceDetails,且sourceIdentifier(Lambda ARN/名称)会被真的拿去查 moto 的 Lambda 后端——get_backend(account_id, region).get_function(source_identifier)查不到时抛InsufficientPermissionsException(moto/config/models.py#L688-L697)。SourceDetail(moto/config/models.py#L542-L634)进一步校验:EventSource只允许aws.config;MessageType必须是ConfigurationItemChangeNotification/ConfigurationSnapshotDeliveryCompleted/OversizedConfigurationItemChangeNotification/ScheduledNotification之一;maximumExecutionFrequency限定在One_Hour…TwentyFour_Hours五个枚举,且变更通知类消息类型不允许设频率(抛InvalidParameterValueException),快照/定时类缺省时默认TwentyFour_Hours。
与文档"TBD"注记一致:规则目前只做"记账"(CRUD + describe,describe_config_rules每页固定 25 条CONFIG_RULE_PAGE_SIZE),不会与 recorder 联动、不产生评估事件。
put_evaluations、select_resource_config 与自定义资源
put_evaluations(moto/config/models.py#L1774-L1798):Evaluations为空抛参数异常,ResultToken缺失抛InvalidResultTokenException;非 TestMode 调用直接NotImplementedError,TestMode 下返回空的FailedEvaluations。
select_resource_config(moto/config/models.py#L2175-L2204)不解析也不执行 SQL,而是按以下顺序取"预定义结果":
- 若
expression命中后端的_expression_results字典(表达式 → 结果行列表),直接返回; - 否则从
query_results_queue(FIFO 队列)弹出一个结果集; - 都没有则返回空结果。
结果行的列名由首行 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_config(moto/config/models.py#L2206-L2245):用于写入自定义资源类型(如MyCorp::Widget);ResourceType前缀不允许是amzn/amazon/alexa/custom(抛InvalidResourceType),资源以"{resource_type}:{resource_id}"为键保存在_custom_resources中。
保留期配置与标签管理
保留期(moto/config/models.py#L2114-L2173):
put_retention_configuration要求retentionPeriodInDays在30 ~ 2557天之间,越界抛带 AWS 原样措辞的ValidationException;后端同一时间只保存一个RetentionConfiguration(名称缺省default);describe_retention_configurations与文档描述一致:最多接受 1 个名称,传多个抛ValidationException,名称不存在抛NoSuchRetentionConfigurationException,不传名称则返回当前配置(无配置返回空列表);delete_retention_configuration名称不匹配同样抛NoSuchRetentionConfigurationException。
标签(moto/config/models.py#L1958-L2001):tag_resource/untag_resource通过_match_arn在ConfigRule、ConfigurationAggregator、AggregationAuthorization三类资源的 ARN 中匹配(moto/config/models.py#L1933-L1956),匹配不到抛ResourceNotFoundException;tag_resource是合并语义(tags.update),untag_resource对不存在的键静默忽略。标签校验函数validate_tags(moto/config/models.py#L151-L168)统一执行:单次最多 50 个标签(TooManyTags)、key 不超过 128 字符且字符集受限(TagKeyTooBig/InvalidTagCharacters)、value 不超过 256(TagValueTooBig)、重复 key 抛DuplicateTags。list_tags_for_resource返回按 key 排序的标签,limit上限 100。
组织合规包
put_organization_conformance_pack(moto/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 = 100,describe_config_rules固定 25 条/页(CONFIG_RULE_PAGE_SIZE = 25);NextToken的语义是"下一项的名称"(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.py(recorder/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、聚合批量结果不带 tags;
- get_resource_config_history 不返回历史,
select_resource_config不执行 SQL——两者都需要通过后端预置数据(query_results_queue/_expression_results)来驱动断言; - 区域与账户隔离由
BackendDict保证,跨区域测试要使用对应区域的客户端与后端。
- Mock
- 测试
【免费下载链接】moto
A library that allows you to easily mock out tests based on AWS infrastructure.
相关推荐
moto 中 CloudFront Mock 的实现全景:已支持 API、内部机制与已知限制
moto 中 CloudFront Mock 的实现全景:已支持 API、内部机制与已知限制 本文以 moto 仓库中 CloudFront 服务的实现清单文档
Mock测试moto 中的 AWS CodeDeploy 模拟:已实现 API 清单、源码实现解析与 boto3 测试实战
moto 中的 AWS CodeDeploy 模拟:已实现 API 清单、源码实现解析与 boto3 测试实战 本文以 moto 官方服务文档 codedepl
Mock测试moto 中 CodePipeline 服务的 Mock 实现:功能覆盖清单与源码级行为解析
moto 中 CodePipeline 服务的 Mock 实现:功能覆盖清单与源码级行为解析 本篇基于 moto 仓库中的服务文档 codepipeline.r
Mock测试
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考