- Mock
- 测试
【免费下载链接】moto
A library that allows you to easily mock out tests based on AWS infrastructure.
本文基于 moto 仓库中docs/docs/services/cognito-identity.rst服务文档展开,系统梳理 moto 对 AWS Cognito Identity(身份池)服务的 mock 能力边界:哪些 API 已实现、哪些尚未支持、关键参数如何解析、身份 ID 与凭证如何生成,并结合 models.py、responses.py 源码与 test_cognitoidentity.py 测试用例,帮助你在单元测试中可靠地 mock 出 Identity Pool 的完整生命周期。
一、功能支持现状:以官方覆盖清单为准
moto 的官方文档 cognito-identity 服务页 明确列出了该服务各 API 的实现状态。下表完整继承文档中的覆盖清单(已实现 10 项,未实现 15 项):
| 操作 | 实现状态 | 文档备注 |
|---|---|---|
create_identity_pool | ✅ 已实现 | - |
delete_identities | ❌ 未实现 | - |
delete_identity_pool | ✅ 已实现 | - |
describe_identity | ❌ 未实现 | - |
describe_identity_pool | ✅ 已实现 | - |
get_credentials_for_identity | ✅ 已实现 | - |
get_id | ✅ 已实现 | - |
get_identity_pool_roles | ❌ 未实现 | - |
get_open_id_token | ✅ 已实现 | - |
get_open_id_token_for_developer_identity | ✅ 已实现 | - |
get_principal_tag_attribute_map | ❌ 未实现 | - |
list_identities | ✅ 已实现 | MaxResults参数尚未实现 |
list_identity_pools | ✅ 已实现 | MaxResults参数尚未实现 |
list_tags_for_resource | ❌ 未实现 | - |
lookup_developer_identity | ❌ 未实现 | - |
merge_developer_identities | ❌ 未实现 | - |
set_identity_pool_roles | ❌ 未实现 | - |
set_principal_tag_attribute_map | ❌ 未实现 | - |
tag_resource | ❌ 未实现 | - |
unlink_developer_identity | ❌ 未实现 | - |
unlink_identity | ❌ 未实现 | - |
untag_resource | ❌ 未实现 | - |
update_identity_pool | ✅ 已实现 | AllowClassic参数尚未实现 |
从源码可以印证这些备注:list_identities与list_identity_pools在 models.py 中的 docstring 均标注 "The MaxResults-parameter has not yet been implemented",即分页参数会被接受但不会生效,调用始终返回全部数据;update_identity_pool的 docstring 同样注明AllowClassic参数未实现(见 models.py)。此外,tag 系列 API(tag_resource/untag_resource/list_tags_for_resource)虽然整体未实现,但create_identity_pool时传入的IdentityPoolTags会被保存并在describe_identity_pool中返回,这一点在CognitoIdentityPool.to_json()的序列化字段中可见("IdentityPoolTags": self.tags,见 models.py)。
二、模块结构与请求路由
moto 的 cognito-identity 模拟由 4 个文件组成,结构典型地遵循 moto 各服务的统一范式:
| 文件 | 职责 |
|---|---|
| moto/cognitoidentity/models.py | 数据模型CognitoIdentityPool与后端CognitoIdentityBackend,承载全部业务逻辑 |
| moto/cognitoidentity/responses.py | CognitoIdentityResponse,从请求中提取参数并调用 backend |
| moto/cognitoidentity/urls.py | URL 匹配规则与 dispatch 入口 |
| moto/cognitoidentity/exceptions.py | ResourceNotFoundError与InvalidNameException两种异常 |
路由配置见 urls.py:
url_bases = [r"https?://cognito-identity\.(.+)\.amazonaws.com"] url_paths = {"{0}/$": CognitoIdentityResponse.dispatch}这意味着无论请求发往哪个区域的端点(cognito-identity.us-west-2.amazonaws.com、cognito-identity.eu-west-2.amazonaws.com等),只要 host 符合cognito-identity.{region}.amazonaws.com模式且路径为根路径,都会被CognitoIdentityResponse.dispatch接管。CognitoIdentityResponse继承自 moto 核心的BaseResponse,服务名注册为cognito-identity(见 responses.py),backend 实例通过cognitoidentity_backends[self.current_account][self.region]按「账户 + 区域」两级索引获取,即每个 (account, region) 组合拥有独立的 Identity Pool 存储。
三、Identity Pool 数据模型与名称校验规则
CognitoIdentityPool类(models.py)定义了池的全部可持久化字段:
| 属性 | 说明 |
|---|---|
identity_pool_name | 池名称,必须匹配正则[\w\s+=,.@-]+,否则抛出校验异常 |
allow_unauthenticated_identities | 是否允许未认证身份,默认为空字符串 |
supported_login_providers | 登录提供方映射,如{"graph.facebook.com": "123456789012345"},默认为{} |
developer_provider_name | 开发者身份提供方名称,默认为空字符串 |
open_id_connect_provider_arns | OIDC 提供方 ARN 列表 |
cognito_identity_providers | Cognito User Pool 提供方列表(每项含 ProviderName/ClientId/ServerSideTokenCheck) |
saml_provider_arns | SAML 提供方 ARN 列表 |
tags | 标签字典 |
identity_pool_id | 自动生成的 ID,格式为{region}:{uuid4} |
creation_time | 创建时间戳 |
名称校验是 create 时唯一会拒绝请求的约束。校验逻辑位于 models.py:
if not re.fullmatch(r"[\w\s+=,.@-]+", identity_pool_name): raise InvalidNameException(identity_pool_name)失败时抛出ValidationException,错误消息与真实 AWS 保持一致(见 exceptions.py):
1 validation error detected: Value '{name}' at 'identityPoolName' failed to satisfy constraint: Member must satisfy regular expression pattern: [\w\s+=,.@-]+测试用例 用三个非法名(pool#name、with!excl、with?quest)验证了该异常,并用四个合法名(x、pool-、pool_name、with space)验证通过分支——注意合法字符集包含字母数字、下划线、空格以及+=,.@-。
四、实战:装饰器模式下的完整生命周期
以下示例整合自 test_cognitoidentity.py 中的真实测试,展示创建 → 查询 → 更新 → 获取身份 → 列举 → 删除的完整链路。
4.1 创建与查询 Identity Pool
import boto3 from moto import mock_aws @mock_aws def create_and_describe_pool(): conn = boto3.client("cognito-identity", "us-west-2") res = conn.create_identity_pool( IdentityPoolName="TestPool", AllowUnauthenticatedIdentities=False, SupportedLoginProviders={"graph.facebook.com": "123456789012345"}, DeveloperProviderName="devname", OpenIdConnectProviderARNs=["arn:aws:rds:eu-west-2:123456789012:db:mysql-db"], CognitoIdentityProviders=[ { "ProviderName": "testprovider", "ClientId": "CLIENT12345", "ServerSideTokenCheck": True, } ], SamlProviderARNs=["arn:aws:rds:eu-west-2:123456789012:db:mysql-db"], ) assert res["IdentityPoolId"] != "" result = conn.describe_identity_pool(IdentityPoolId=res["IdentityPoolId"]) assert result["SupportedLoginProviders"] == res["SupportedLoginProviders"] assert result["DeveloperProviderName"] == res["DeveloperProviderName"] assert result["CognitoIdentityProviders"] == res["CognitoIdentityProviders"] assert result["SamlProviderARNs"] == res["SamlProviderARNs"]参数提取发生在 responses.py:create_identity_pool通过self._get_param(...)依次读取IdentityPoolName、AllowUnauthenticatedIdentities、SupportedLoginProviders、DeveloperProviderName、OpenIdConnectProviderARNs、CognitoIdentityProviders、SamlProviderARNs、IdentityPoolTags八个字段并透传给 backend。创建成功后,backend 除了把池存入identity_pools字典,还同步在pools_identities中初始化了空的Identities列表(models.py),为后续get_id/list_identities的写入做准备。
查询不存在的池会返回标准 400 错误,测试 test_describe_identity_pool_with_invalid_id_raises_error 验证了错误类型为ResourceNotFoundException,消息即传入的池 ID。delete_identity_pool内部先调用describe_identity_pool做存在性检查(models.py),因此删除不存在的池同样抛出该异常。
4.2 更新 Identity Pool 的“部分更新”语义
update_identity_pool(responses.py)的行为值得注意:未传入的参数保持原值不变,只有显式传入的参数才会被覆盖。这在 models.py 中通过一系列is not None判断实现:
if allow_unauthenticated is not None: pool.allow_unauthenticated_identities = allow_unauthenticated if login_providers is not None: pool.supported_login_providers = login_providers if provider_name: pool.developer_provider_name = provider_name # ... 其余字段同理测试 test_update_identity_pool 用参数化用例覆盖了SupportedLoginProviders的增量合并场景(从 1 个 provider 更新为 2 个)、清空场景(更新为{})以及DeveloperProviderName的替换(dev1→dev2),并断言更新后describe_identity_pool的返回值与更新响应一致。文档中标注的AllowClassic参数未实现,调用时传入该参数会被静默忽略。
4.3 获取身份 ID:区域前缀推断机制
get_id是已实现 API 中逻辑最特殊的一个(models.py):
def get_id(self, identity_pool_id: str) -> str: # This call does not have to be authenticated, which means we do not know # to which region it was sent originally # But the identity_pool_id is always prefixed with the region, # so we just that to determine the right region region = identity_pool_id.split(":")[0] backend: CognitoIdentityBackend = cognitoidentity_backends[self.account_id][region] identity_id = {"IdentityId": get_random_identity_id(self.region_name)} backend.pools_identities[identity_pool_id]["Identities"].append(identity_id) return json.dumps(identity_id)源码注释解释了这一设计的动机:GetId在实际 AWS 中可能是匿名(未经认证)调用,客户端未必把请求发往池所在的区域;而 moto 生成的IdentityPoolId总是以区域为前缀,因此可以用identity_pool_id.split(":")[0]还原出池的真实区域,再把新身份写入该区域的 backend 存储。ID 生成本身由 utils.py 完成:
def get_random_identity_id(region: str) -> str: return f"{region}:{mock_random.uuid4()}"即IdentityId为{region}:{uuid4}格式,其中mock_random是 moto 的可播种随机数源,因此在测试中 ID 生成具备可复现性。测试 同时验证了未认证调用场景(@set_initial_no_auth_action_count(1)允许 1 次免认证操作),且IdentityId以us-west-2开头。
这里有一个值得记住的适用前提(源码注释明确指出):如果用户跨区域请求——比如在us-west-2的客户端上请求us-west-1:...池的 ID——moto 仍能按前缀路由到正确区域,但这意味着某些真实 AWS 会报错的跨区域场景在 mock 中不会复现。
4.4 凭证与 OpenID Token 的返回内容
get_credentials_for_identity(models.py)返回一组固定的测试凭证:
{ "Credentials": { "AccessKeyId": "TESTACCESSKEY12345", "Expiration": <now + 90s 的 UNIX 时间戳>, "SecretKey": "ABCSECRETKEY", "SessionToken": "ABC12345" }, "IdentityId": "<传入的 IdentityId>" }有效期固定为 90 秒。测试 test_get_credentials_for_identity 验证Expiration反序列化为datetime类型、IdentityId原样回显。注意:该凭证是纯占位值,仅用于让你的业务代码在测试中走通「换取临时凭证」的代码路径,绝不能(也不应)被当作能访问真实 AWS 资源的凭证。
get_open_id_token与get_open_id_token_for_developer_identity(models.py)均返回{"IdentityId": ..., "Token": "{region}:{uuid4}"},Token 为随机值。一个细节是 responses.py 中对IdentityId的处理:若调用方未显式传入IdentityId,则自动生成一个随机值兜底,对应测试 test_get_open_id_token_for_developer_identity_when_no_explicit_identity_id 验证了此时返回的IdentityId非空。
4.5 列举与删除
list_identities直接序列化pools_identities[pool_id]整个结构,返回{"IdentityPoolId": ..., "Identities": [{"IdentityId": ...}, ...]}。测试 test_list_identities 展示完整链路:建池 →get_id→list_identities,断言刚生成的IdentityId出现在结果中。
list_identity_pools返回当前账户、当前区域内全部池的完整定义列表(models.py)。
delete_identity_pool删除后池从identity_pools中移除,测试 test_delete_identity_pool 用删除前后list_identity_pools的计数(1 → 0)验证。注意删除时pools_identities中对应的身份列表不会被级联清理,属于当前实现的隐含行为。
五、Server 模式:基于 X-Amz-Target 的 JSON 1.1 协议
除装饰器模式外,moto 也支持通过moto.server以独立 HTTP 服务方式运行。test_server.py 展示了 Server 模式下 cognito-identity 的请求协议:POST 到根路径/,通过X-Amz-Target头区分操作,Content-Type为application/x-amz-json-1.1(常量APPLICATION_AMZ_JSON_1_1,定义于 moto/utilities/constants.py):
import moto.server as server from moto import mock_aws from moto.utilities.constants import APPLICATION_AMZ_JSON_1_1 @mock_aws def test_server_get_id(): backend = server.create_backend_app("cognito-identity") test_client = backend.test_client() # 1. 创建池 res = test_client.post( "/", json={"IdentityPoolName": "test", "AllowUnauthenticatedIdentities": True}, headers={ "X-Amz-Target": "AWSCognitoIdentityService.CreateIdentityPool", "Content-Type": APPLICATION_AMZ_JSON_1_1, }, ) pool_id = json.loads(res.data.decode("utf-8"))["IdentityPoolId"] # 2. 获取身份 res = test_client.post( "/", json={"AccountId": "someaccount", "IdentityPoolId": pool_id, "Logins": {"someurl": "12345"}}, headers={ "X-Amz-Target": "AWSCognitoIdentityService.GetId", "Content-Type": APPLICATION_AMZ_JSON_1_1, }, ) assert ":" in json.loads(res.data.decode("utf-8"))["IdentityId"]该测试同时覆盖了ListIdentities的服务端调用(test_server.py)。这意味着如果你的被测代码以AWS_ENDPOINT_URL指向 moto server 实例,装饰器与 server 两种模式在 cognito-identity 上行为一致。
六、使用限制与注意事项小结
结合文档清单与源码,在实际项目中应留意以下约束:
- 分页参数不生效:
list_identities与list_identity_pools的MaxResults参数会被接受但被忽略,始终返回全量数据(文档明确标注,源码 docstring 印证)。如果你的测试依赖分页边界行为,需要另行处理。 - 15 个 API 未实现:角色绑定(
set_identity_pool_roles/get_identity_pool_roles)、开发者身份合并/解绑(merge_developer_identities/unlink_developer_identity/lookup_developer_identity)、describe_identity、标签资源级 API、principal tag 映射等均不在支持范围,调用会得到未知操作错误。设计测试时避免依赖这些路径。 AllowClassic未实现:update_identity_pool传入该参数会被忽略。- 凭证是占位值:
get_credentials_for_identity返回固定的TESTACCESSKEY12345等值,仅用于打通代码路径。 - 区域隔离:池数据按 (account, region) 隔离,
get_id依靠池 ID 的区域前缀做跨区域路由,这一 mock 行为与真实 AWS 的跨区域错误语义存在差异(源码注释已说明)。 - 名称校验可预期:池名称只允许
[\w\s+=,.@-]+字符集,测试可直接断言ValidationException错误码与消息。
七、延伸阅读路径
- 服务实现清单:docs/docs/services/cognito-identity.rst
- 数据模型与后端逻辑:moto/cognitoidentity/models.py
- 请求参数解析:moto/cognitoidentity/responses.py
- 路由规则:moto/cognitoidentity/urls.py
- 装饰器模式测试全集:tests/test_cognitoidentity/test_cognitoidentity.py
- Server 模式测试:tests/test_cognitoidentity/test_server.py
- 相关服务:moto 同时实现了 Cognito User Pool(
cognitoidp,位于 moto/cognitoidp/),Identity Pool 的CognitoIdentityProviders参数正是指向 User Pool 的 ClientId,两者常配合使用。
- Mock
- 测试
【免费下载链接】moto
A library that allows you to easily mock out tests based on AWS infrastructure.
相关推荐
使用 AWS CLI 的 cognito-identity create-identity-pool 命令创建 Cognito 身份池(Identity Pool)实战指南
使用 AWS CLI 的 cognito identity create identity pool 命令创建 Cognito 身份池(Identity Poo
开发工具云原生运维AWS CLI `cognito-identity delete-identity-pool` 详解:删除 Amazon Cognito 身份池的完整操作指南
AWS CLI cognito identity delete identity pool 详解:删除 Amazon Cognito 身份池的完整操作指南 de
开发工具云原生运维使用 AWS CLI 更新 Cognito 身份池(Identity Pool):update-identity-pool 完整实战指南
使用 AWS CLI 更新 Cognito 身份池(Identity Pool):update identity pool 完整实战指南 导读 本文基于 aws
开发工具云原生运维
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考