Made some changes for server testing and added another get_id test.
This commit is contained in:
parent
2455de8282
commit
229d453b99
@ -12,21 +12,6 @@ from moto.core.utils import iso_8601_datetime_with_milliseconds
|
||||
from .utils import get_random_identity_id
|
||||
|
||||
|
||||
class CognitoIdentityObject(BaseModel):
|
||||
|
||||
def __init__(self, object_id, name, fields):
|
||||
self.object_id = object_id
|
||||
self.name = name
|
||||
self.fields = fields
|
||||
|
||||
def to_json(self):
|
||||
return {
|
||||
"fields": self.fields,
|
||||
"id": self.object_id,
|
||||
"name": self.name,
|
||||
}
|
||||
|
||||
|
||||
class CognitoIdentity(BaseModel):
|
||||
|
||||
def __init__(self, region, identity_pool_name, **kwargs):
|
||||
@ -89,7 +74,7 @@ class CognitoIdentityBackend(BaseBackend):
|
||||
now = datetime.datetime.utcnow()
|
||||
expiration = now + datetime.timedelta(seconds=duration)
|
||||
expiration_str = str(iso_8601_datetime_with_milliseconds(expiration))
|
||||
return json.dumps(
|
||||
response = json.dumps(
|
||||
{
|
||||
"Credentials":
|
||||
{
|
||||
@ -100,6 +85,7 @@ class CognitoIdentityBackend(BaseBackend):
|
||||
},
|
||||
"IdentityId": identity_id
|
||||
})
|
||||
return response
|
||||
|
||||
def get_open_id_token_for_developer_identity(self, identity_id):
|
||||
duration = 90
|
||||
|
@ -1,7 +1,5 @@
|
||||
from __future__ import unicode_literals
|
||||
|
||||
import json
|
||||
|
||||
from moto.core.responses import BaseResponse
|
||||
|
||||
from .models import cognitoidentity_backends
|
||||
@ -9,18 +7,6 @@ from .models import cognitoidentity_backends
|
||||
|
||||
class CognitoIdentityResponse(BaseResponse):
|
||||
|
||||
@property
|
||||
def parameters(self):
|
||||
# TODO this should really be moved to core/responses.py
|
||||
if self.body:
|
||||
return json.loads(self.body)
|
||||
else:
|
||||
return self.querystring
|
||||
|
||||
@property
|
||||
def cognitoidentity_backend(self):
|
||||
return cognitoidentity_backends[self.region]
|
||||
|
||||
def create_identity_pool(self):
|
||||
identity_pool_name = self._get_param('IdentityPoolName')
|
||||
allow_unauthenticated_identities = self._get_param('AllowUnauthenticatedIdentities')
|
||||
|
@ -5,6 +5,8 @@ import boto3
|
||||
from moto import mock_cognitoidentity
|
||||
import sure # noqa
|
||||
|
||||
from moto.cognitoidentity.utils import get_random_identity_id
|
||||
|
||||
|
||||
@mock_cognitoidentity
|
||||
def test_create_identity_pool():
|
||||
@ -26,8 +28,14 @@ def test_create_identity_pool():
|
||||
assert result['IdentityPoolId'] != ''
|
||||
|
||||
|
||||
# testing a helper function
|
||||
def test_get_random_identity_id():
|
||||
assert len(get_random_identity_id('us-west-2')) > 0
|
||||
|
||||
|
||||
@mock_cognitoidentity
|
||||
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')
|
||||
result = conn.get_id(AccountId='someaccount',
|
||||
IdentityPoolId='us-west-2:12345',
|
||||
@ -35,14 +43,15 @@ def test_get_id():
|
||||
'someurl': '12345'
|
||||
})
|
||||
print(result)
|
||||
assert result['IdentityId'].startswith('us-west-2')
|
||||
assert result.get('IdentityId', "").startswith('us-west-2') or result.get('ResponseMetadata').get('HTTPStatusCode') == 200
|
||||
|
||||
|
||||
@mock_cognitoidentity
|
||||
def test_get_credentials_for_identity():
|
||||
conn = boto3.client('cognito-identity', 'us-west-2')
|
||||
result = conn.get_credentials_for_identity(IdentityId='12345')
|
||||
assert result['IdentityId'] == '12345'
|
||||
|
||||
assert result.get('IdentityId') == '12345' or result.get('ResponseMetadata').get('HTTPStatusCode') == 200
|
||||
|
||||
|
||||
@mock_cognitoidentity
|
||||
|
@ -25,3 +25,21 @@ def test_create_identity_pool():
|
||||
|
||||
json_data = json.loads(res.data.decode("utf-8"))
|
||||
assert json_data['IdentityPoolName'] == "test"
|
||||
|
||||
|
||||
@mock_cognitoidentity
|
||||
def test_get_id():
|
||||
backend = server.create_backend_app("cognito-identity")
|
||||
test_client = backend.test_client()
|
||||
|
||||
res = test_client.post('/',
|
||||
data=json.dumps({'AccountId': 'someaccount',
|
||||
'IdentityPoolId': 'us-west-2:12345',
|
||||
'Logins': {'someurl': '12345'}}),
|
||||
headers={
|
||||
"X-Amz-Target": "com.amazonaws.cognito.identity.model.AWSCognitoIdentityService.GetId"},
|
||||
)
|
||||
|
||||
print(res.data)
|
||||
json_data = json.loads(res.data.decode("utf-8"))
|
||||
assert ':' in json_data['IdentityId']
|
||||
|
Loading…
Reference in New Issue
Block a user