☰
moto 中 cognito-identity 服务 mock 全解:Identity Pool 操作覆盖、实现原理与测试实战
2026/9/25 2:59:42 网站建设 项目流程
  • Mock
  • 测试

【免费下载链接】moto

A library that allows you to easily mock out tests based on AWS infrastructure.

项目地址:https://gitcode.com/gh_mirrors/mo/moto
点击查看免费下载

本文基于 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.pyCognitoIdentityResponse,从请求中提取参数并调用 backend
moto/cognitoidentity/urls.pyURL 匹配规则与 dispatch 入口
moto/cognitoidentity/exceptions.pyResourceNotFoundError与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_arnsOIDC 提供方 ARN 列表
cognito_identity_providersCognito User Pool 提供方列表(每项含 ProviderName/ClientId/ServerSideTokenCheck)
saml_provider_arnsSAML 提供方 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 上行为一致。

六、使用限制与注意事项小结

结合文档清单与源码,在实际项目中应留意以下约束:

  1. 分页参数不生效:list_identities与list_identity_pools的MaxResults参数会被接受但被忽略,始终返回全量数据(文档明确标注,源码 docstring 印证)。如果你的测试依赖分页边界行为,需要另行处理。
  2. 15 个 API 未实现:角色绑定(set_identity_pool_roles/get_identity_pool_roles)、开发者身份合并/解绑(merge_developer_identities/unlink_developer_identity/lookup_developer_identity)、describe_identity、标签资源级 API、principal tag 映射等均不在支持范围,调用会得到未知操作错误。设计测试时避免依赖这些路径。
  3. AllowClassic未实现:update_identity_pool传入该参数会被忽略。
  4. 凭证是占位值:get_credentials_for_identity返回固定的TESTACCESSKEY12345等值,仅用于打通代码路径。
  5. 区域隔离:池数据按 (account, region) 隔离,get_id依靠池 ID 的区域前缀做跨区域路由,这一 mock 行为与真实 AWS 的跨区域错误语义存在差异(源码注释已说明)。
  6. 名称校验可预期:池名称只允许[\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.

项目地址:https://gitcode.com/gh_mirrors/mo/moto
点击查看免费下载
上一篇:深入解析 VS Code MCP 扩展示例:用 `registerMcpServerDefinitionProvider` 将 Gist 中的 MCP Server 接入 Copilot Chat
下一篇:如何快速安装wangEditor v5:完整配置指南

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询