Implement update_user_pool_domain

Introduce the CognitoIDP's UpdateUserPoolDomain to update configuration
options of the associated domain to a Cognito IDP (e.g. ACM certificate).
This commit is contained in:
Carlos Aguado 2019-07-16 13:09:13 +10:00
parent e4ed14b131
commit 81980850d4
4 changed files with 93 additions and 14 deletions

View File

@ -928,6 +928,7 @@
- [ ] update_user_attributes
- [ ] update_user_pool
- [X] update_user_pool_client
- [X] update_user_pool_domain
- [ ] verify_software_token
- [ ] verify_user_attribute

View File

@ -2,6 +2,7 @@ from __future__ import unicode_literals
import datetime
import functools
import hashlib
import itertools
import json
import os
@ -154,20 +155,37 @@ class CognitoIdpUserPool(BaseModel):
class CognitoIdpUserPoolDomain(BaseModel):
def __init__(self, user_pool_id, domain):
def __init__(self, user_pool_id, domain, custom_domain_config=None):
self.user_pool_id = user_pool_id
self.domain = domain
self.custom_domain_config = custom_domain_config or {}
def to_json(self):
return {
"UserPoolId": self.user_pool_id,
"AWSAccountId": str(uuid.uuid4()),
"CloudFrontDistribution": None,
"Domain": self.domain,
"S3Bucket": None,
"Status": "ACTIVE",
"Version": None,
}
def _distribution_name(self):
if self.custom_domain_config and \
'CertificateArn' in self.custom_domain_config:
hash = hashlib.md5(
self.custom_domain_config['CertificateArn'].encode('utf-8')
).hexdigest()
return "{hash}.cloudfront.net".format(hash=hash[:16])
return None
def to_json(self, extended=True):
distribution = self._distribution_name()
if extended:
return {
"UserPoolId": self.user_pool_id,
"AWSAccountId": str(uuid.uuid4()),
"CloudFrontDistribution": distribution,
"Domain": self.domain,
"S3Bucket": None,
"Status": "ACTIVE",
"Version": None,
}
elif distribution:
return {
"CloudFrontDomain": distribution,
}
return None
class CognitoIdpUserPoolClient(BaseModel):
@ -338,11 +356,13 @@ class CognitoIdpBackend(BaseBackend):
del self.user_pools[user_pool_id]
# User pool domain
def create_user_pool_domain(self, user_pool_id, domain):
def create_user_pool_domain(self, user_pool_id, domain, custom_domain_config=None):
if user_pool_id not in self.user_pools:
raise ResourceNotFoundError(user_pool_id)
user_pool_domain = CognitoIdpUserPoolDomain(user_pool_id, domain)
user_pool_domain = CognitoIdpUserPoolDomain(
user_pool_id, domain, custom_domain_config=custom_domain_config
)
self.user_pool_domains[domain] = user_pool_domain
return user_pool_domain
@ -358,6 +378,14 @@ class CognitoIdpBackend(BaseBackend):
del self.user_pool_domains[domain]
def update_user_pool_domain(self, domain, custom_domain_config):
if domain not in self.user_pool_domains:
raise ResourceNotFoundError(domain)
user_pool_domain = self.user_pool_domains[domain]
user_pool_domain.custom_domain_config = custom_domain_config
return user_pool_domain
# User pool client
def create_user_pool_client(self, user_pool_id, extended_config):
user_pool = self.user_pools.get(user_pool_id)

View File

@ -50,7 +50,13 @@ class CognitoIdpResponse(BaseResponse):
def create_user_pool_domain(self):
domain = self._get_param("Domain")
user_pool_id = self._get_param("UserPoolId")
cognitoidp_backends[self.region].create_user_pool_domain(user_pool_id, domain)
custom_domain_config = self._get_param("CustomDomainConfig")
user_pool_domain = cognitoidp_backends[self.region].create_user_pool_domain(
user_pool_id, domain, custom_domain_config
)
domain_description = user_pool_domain.to_json(extended=False)
if domain_description:
return json.dumps(domain_description)
return ""
def describe_user_pool_domain(self):
@ -69,6 +75,17 @@ class CognitoIdpResponse(BaseResponse):
cognitoidp_backends[self.region].delete_user_pool_domain(domain)
return ""
def update_user_pool_domain(self):
domain = self._get_param("Domain")
custom_domain_config = self._get_param("CustomDomainConfig")
user_pool_domain = cognitoidp_backends[self.region].update_user_pool_domain(
domain, custom_domain_config
)
domain_description = user_pool_domain.to_json(extended=False)
if domain_description:
return json.dumps(domain_description)
return ""
# User pool client
def create_user_pool_client(self):
user_pool_id = self.parameters.pop("UserPoolId")

View File

@ -133,6 +133,22 @@ def test_create_user_pool_domain():
result["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
@mock_cognitoidp
def test_create_user_pool_domain_custom_domain_config():
conn = boto3.client("cognito-idp", "us-west-2")
domain = str(uuid.uuid4())
custom_domain_config = {
"CertificateArn": "arn:aws:acm:us-east-1:123456789012:certificate/123456789012",
}
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
result = conn.create_user_pool_domain(
UserPoolId=user_pool_id, Domain=domain, CustomDomainConfig=custom_domain_config
)
result["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
result["CloudFrontDomain"].should.equal("e2c343b3293ee505.cloudfront.net")
@mock_cognitoidp
def test_describe_user_pool_domain():
conn = boto3.client("cognito-idp", "us-west-2")
@ -162,6 +178,23 @@ def test_delete_user_pool_domain():
result["DomainDescription"].keys().should.have.length_of(0)
@mock_cognitoidp
def test_update_user_pool_domain():
conn = boto3.client("cognito-idp", "us-west-2")
domain = str(uuid.uuid4())
custom_domain_config = {
"CertificateArn": "arn:aws:acm:us-east-1:123456789012:certificate/123456789012",
}
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
conn.create_user_pool_domain(UserPoolId=user_pool_id, Domain=domain)
result = conn.update_user_pool_domain(
UserPoolId=user_pool_id, Domain=domain, CustomDomainConfig=custom_domain_config
)
result["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
result["CloudFrontDomain"].should.equal("e2c343b3293ee505.cloudfront.net")
@mock_cognitoidp
def test_create_user_pool_client():
conn = boto3.client("cognito-idp", "us-west-2")