Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add cognito-identity list identities method (#4525)

Browse Source
This commit is contained in:
Antonio Caparrós 2021-11-05 14:17:30 +01:00 committed by GitHub
parent 7b375195bd
commit 48fbe0db70
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 101 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
moto/cognitoidentity/models.py
View File

@ -54,6 +54,7 @@ class CognitoIdentityBackend(BaseBackend):
super(CognitoIdentityBackend, self).__init__()
self.region = region
self.identity_pools = OrderedDict()
self.pools_identities = {}
def reset(self):
region = self.region
@ -105,7 +106,14 @@ class CognitoIdentityBackend(BaseBackend):
tags=tags,
)
self.identity_pools[new_identity.identity_pool_id] = new_identity
self.pools_identities.update(
{
new_identity.identity_pool_id: {
"IdentityPoolId": new_identity.identity_pool_id,
"Identities": [],
}
}
)
response = new_identity.to_json()
return response
@ -142,8 +150,9 @@ class CognitoIdentityBackend(BaseBackend):
response = pool.to_json()
return response
def get_id(self):
def get_id(self, identity_pool_id: str):
identity_id = {"IdentityId": get_random_identity_id(self.region)}
self.pools_identities[identity_pool_id]["Identities"].append(identity_id)
return json.dumps(identity_id)
def get_credentials_for_identity(self, identity_id):
@ -176,6 +185,10 @@ class CognitoIdentityBackend(BaseBackend):
)
return response
def list_identities(self, identity_pool_id, max_results=123):
response = json.dumps(self.pools_identities[identity_pool_id])
return response
cognitoidentity_backends = {}
for region in Session().get_available_regions("cognito-identity"):

9
moto/cognitoidentity/responses.py
View File

@ -53,7 +53,9 @@ class CognitoIdentityResponse(BaseResponse):
)
def get_id(self):
return cognitoidentity_backends[self.region].get_id()
return cognitoidentity_backends[self.region].get_id(
identity_pool_id=self._get_param("IdentityPoolId"),
)
def describe_identity_pool(self):
return cognitoidentity_backends[self.region].describe_identity_pool(
@ -76,3 +78,8 @@ class CognitoIdentityResponse(BaseResponse):
return cognitoidentity_backends[self.region].get_open_id_token(
self._get_param("IdentityId") or get_random_identity_id(self.region)
)
def list_identities(self):
return cognitoidentity_backends[self.region].list_identities(
self._get_param("IdentityPoolId") or get_random_identity_id(self.region)
)

23
tests/test_cognitoidentity/test_cognitoidentity.py
View File

@ -156,9 +156,12 @@ def test_get_random_identity_id():
def test_get_id():
# These two do NOT work in server mode. They just don't return the data from the model.
conn = boto3.client("cognito-identity", "us-west-2")
identity_pool_data = conn.create_identity_pool(
IdentityPoolName="test_identity_pool", AllowUnauthenticatedIdentities=True
)
result = conn.get_id(
AccountId="someaccount",
IdentityPoolId="us-west-2:12345",
IdentityPoolId=identity_pool_data["IdentityPoolId"],
Logins={"someurl": "12345"},
)
assert (
@ -212,3 +215,21 @@ def test_get_open_id_token():
result = conn.get_open_id_token(IdentityId="12345", Logins={"someurl": "12345"})
assert len(result["Token"]) > 0
assert result["IdentityId"] == "12345"
@mock_cognitoidentity
def test_list_identities():
conn = boto3.client("cognito-identity", "us-west-2")
identity_pool_data = conn.create_identity_pool(
IdentityPoolName="test_identity_pool", AllowUnauthenticatedIdentities=True
)
identity_pool_id = identity_pool_data["IdentityPoolId"]
identity_data = conn.get_id(
AccountId="someaccount",
IdentityPoolId=identity_pool_id,
Logins={"someurl": "12345"},
)
identity_id = identity_data["IdentityId"]
identities = conn.list_identities(IdentityPoolId=identity_pool_id, MaxResults=123)
assert "IdentityPoolId" in identities and "Identities" in identities
assert identity_id in [x["IdentityId"] for x in identities["Identities"]]

57
tests/test_cognitoidentity/test_server.py
View File

@ -1,4 +1,5 @@
import json
import sure # noqa # pylint: disable=unused-import
import moto.server as server
@ -32,12 +33,21 @@ def test_get_id():
backend = server.create_backend_app("cognito-identity")
test_client = backend.test_client()
res = test_client.post(
"/",
data={"IdentityPoolName": "test", "AllowUnauthenticatedIdentities": True},
headers={
"X-Amz-Target": "com.amazonaws.cognito.identity.model.AWSCognitoIdentityService.CreateIdentityPool"
},
)
json_data = json.loads(res.data.decode("utf-8"))
res = test_client.post(
"/",
data=json.dumps(
{
"AccountId": "someaccount",
"IdentityPoolId": "us-west-2:12345",
"IdentityPoolId": json_data["IdentityPoolId"],
"Logins": {"someurl": "12345"},
}
),
@ -48,3 +58,48 @@ def test_get_id():
json_data = json.loads(res.data.decode("utf-8"))
assert ":" in json_data["IdentityId"]
@mock_cognitoidentity
def test_list_identities():
backend = server.create_backend_app("cognito-identity")
test_client = backend.test_client()
res = test_client.post(
"/",
data={"IdentityPoolName": "test", "AllowUnauthenticatedIdentities": True},
headers={
"X-Amz-Target": "com.amazonaws.cognito.identity.model.AWSCognitoIdentityService.CreateIdentityPool"
},
)
json_data = json.loads(res.data.decode("utf-8"))
identity_pool_id = json_data["IdentityPoolId"]
res = test_client.post(
"/",
data=json.dumps(
{
"AccountId": "someaccount",
"IdentityPoolId": identity_pool_id,
"Logins": {"someurl": "12345"},
}
),
headers={
"X-Amz-Target": "com.amazonaws.cognito.identity.model.AWSCognitoIdentityService.GetId"
},
)
json_data = json.loads(res.data.decode("utf-8"))
identity_id = json_data["IdentityId"]
res = test_client.post(
"/",
data=json.dumps({"IdentityPoolId": identity_pool_id,}),
headers={
"X-Amz-Target": "com.amazonaws.cognito.identity.model.AWSCognitoIdentityService.ListIdentities"
},
)
json_data = json.loads(res.data.decode("utf-8"))
assert "IdentityPoolId" in json_data and "Identities" in json_data
assert identity_id in [x["IdentityId"] for x in json_data["Identities"]]