Merge remote-tracking branch 'upstream/master'

This commit is contained in:
Stephan Huber 2019-12-23 08:38:53 +01:00
commit 0527e88d46
541 changed files with 75504 additions and 51429 deletions

View file

@ -4,11 +4,12 @@ import boto3
from botocore.exceptions import ClientError
from moto import mock_sns
import sure # noqa
from moto.core import ACCOUNT_ID
@mock_sns
def test_create_platform_application():
conn = boto3.client('sns', region_name='us-east-1')
conn = boto3.client("sns", region_name="us-east-1")
response = conn.create_platform_application(
Name="my-application",
Platform="APNS",
@ -17,14 +18,15 @@ def test_create_platform_application():
"PlatformPrincipal": "platform_principal",
},
)
application_arn = response['PlatformApplicationArn']
application_arn = response["PlatformApplicationArn"]
application_arn.should.equal(
'arn:aws:sns:us-east-1:123456789012:app/APNS/my-application')
"arn:aws:sns:us-east-1:{}:app/APNS/my-application".format(ACCOUNT_ID)
)
@mock_sns
def test_get_platform_application_attributes():
conn = boto3.client('sns', region_name='us-east-1')
conn = boto3.client("sns", region_name="us-east-1")
platform_application = conn.create_platform_application(
Name="my-application",
Platform="APNS",
@ -33,25 +35,29 @@ def test_get_platform_application_attributes():
"PlatformPrincipal": "platform_principal",
},
)
arn = platform_application['PlatformApplicationArn']
attributes = conn.get_platform_application_attributes(
PlatformApplicationArn=arn)['Attributes']
attributes.should.equal({
"PlatformCredential": "platform_credential",
"PlatformPrincipal": "platform_principal",
})
arn = platform_application["PlatformApplicationArn"]
attributes = conn.get_platform_application_attributes(PlatformApplicationArn=arn)[
"Attributes"
]
attributes.should.equal(
{
"PlatformCredential": "platform_credential",
"PlatformPrincipal": "platform_principal",
}
)
@mock_sns
def test_get_missing_platform_application_attributes():
conn = boto3.client('sns', region_name='us-east-1')
conn = boto3.client("sns", region_name="us-east-1")
conn.get_platform_application_attributes.when.called_with(
PlatformApplicationArn="a-fake-arn").should.throw(ClientError)
PlatformApplicationArn="a-fake-arn"
).should.throw(ClientError)
@mock_sns
def test_set_platform_application_attributes():
conn = boto3.client('sns', region_name='us-east-1')
conn = boto3.client("sns", region_name="us-east-1")
platform_application = conn.create_platform_application(
Name="my-application",
Platform="APNS",
@ -60,291 +66,249 @@ def test_set_platform_application_attributes():
"PlatformPrincipal": "platform_principal",
},
)
arn = platform_application['PlatformApplicationArn']
conn.set_platform_application_attributes(PlatformApplicationArn=arn,
Attributes={
"PlatformPrincipal": "other"}
)
attributes = conn.get_platform_application_attributes(
PlatformApplicationArn=arn)['Attributes']
attributes.should.equal({
"PlatformCredential": "platform_credential",
"PlatformPrincipal": "other",
})
arn = platform_application["PlatformApplicationArn"]
conn.set_platform_application_attributes(
PlatformApplicationArn=arn, Attributes={"PlatformPrincipal": "other"}
)
attributes = conn.get_platform_application_attributes(PlatformApplicationArn=arn)[
"Attributes"
]
attributes.should.equal(
{"PlatformCredential": "platform_credential", "PlatformPrincipal": "other"}
)
@mock_sns
def test_list_platform_applications():
conn = boto3.client('sns', region_name='us-east-1')
conn = boto3.client("sns", region_name="us-east-1")
conn.create_platform_application(
Name="application1",
Platform="APNS",
Attributes={},
Name="application1", Platform="APNS", Attributes={}
)
conn.create_platform_application(
Name="application2",
Platform="APNS",
Attributes={},
Name="application2", Platform="APNS", Attributes={}
)
applications_repsonse = conn.list_platform_applications()
applications = applications_repsonse['PlatformApplications']
applications = applications_repsonse["PlatformApplications"]
applications.should.have.length_of(2)
@mock_sns
def test_delete_platform_application():
conn = boto3.client('sns', region_name='us-east-1')
conn = boto3.client("sns", region_name="us-east-1")
conn.create_platform_application(
Name="application1",
Platform="APNS",
Attributes={},
Name="application1", Platform="APNS", Attributes={}
)
conn.create_platform_application(
Name="application2",
Platform="APNS",
Attributes={},
Name="application2", Platform="APNS", Attributes={}
)
applications_repsonse = conn.list_platform_applications()
applications = applications_repsonse['PlatformApplications']
applications = applications_repsonse["PlatformApplications"]
applications.should.have.length_of(2)
application_arn = applications[0]['PlatformApplicationArn']
application_arn = applications[0]["PlatformApplicationArn"]
conn.delete_platform_application(PlatformApplicationArn=application_arn)
applications_repsonse = conn.list_platform_applications()
applications = applications_repsonse['PlatformApplications']
applications = applications_repsonse["PlatformApplications"]
applications.should.have.length_of(1)
@mock_sns
def test_create_platform_endpoint():
conn = boto3.client('sns', region_name='us-east-1')
conn = boto3.client("sns", region_name="us-east-1")
platform_application = conn.create_platform_application(
Name="my-application",
Platform="APNS",
Attributes={},
Name="my-application", Platform="APNS", Attributes={}
)
application_arn = platform_application['PlatformApplicationArn']
application_arn = platform_application["PlatformApplicationArn"]
endpoint = conn.create_platform_endpoint(
PlatformApplicationArn=application_arn,
Token="some_unique_id",
CustomUserData="some user data",
Attributes={
"Enabled": 'false',
},
Attributes={"Enabled": "false"},
)
endpoint_arn = endpoint['EndpointArn']
endpoint_arn = endpoint["EndpointArn"]
endpoint_arn.should.contain(
"arn:aws:sns:us-east-1:123456789012:endpoint/APNS/my-application/")
"arn:aws:sns:us-east-1:{}:endpoint/APNS/my-application/".format(ACCOUNT_ID)
)
@mock_sns
def test_create_duplicate_platform_endpoint():
conn = boto3.client('sns', region_name='us-east-1')
conn = boto3.client("sns", region_name="us-east-1")
platform_application = conn.create_platform_application(
Name="my-application",
Platform="APNS",
Attributes={},
Name="my-application", Platform="APNS", Attributes={}
)
application_arn = platform_application['PlatformApplicationArn']
application_arn = platform_application["PlatformApplicationArn"]
endpoint = conn.create_platform_endpoint(
PlatformApplicationArn=application_arn,
Token="some_unique_id",
CustomUserData="some user data",
Attributes={
"Enabled": 'false',
},
Attributes={"Enabled": "false"},
)
endpoint = conn.create_platform_endpoint.when.called_with(
PlatformApplicationArn=application_arn,
Token="some_unique_id",
CustomUserData="some user data",
Attributes={
"Enabled": 'false',
},
Attributes={"Enabled": "false"},
).should.throw(ClientError)
@mock_sns
def test_get_list_endpoints_by_platform_application():
conn = boto3.client('sns', region_name='us-east-1')
conn = boto3.client("sns", region_name="us-east-1")
platform_application = conn.create_platform_application(
Name="my-application",
Platform="APNS",
Attributes={},
Name="my-application", Platform="APNS", Attributes={}
)
application_arn = platform_application['PlatformApplicationArn']
application_arn = platform_application["PlatformApplicationArn"]
endpoint = conn.create_platform_endpoint(
PlatformApplicationArn=application_arn,
Token="some_unique_id",
CustomUserData="some user data",
Attributes={
"CustomUserData": "some data",
},
Attributes={"CustomUserData": "some data"},
)
endpoint_arn = endpoint['EndpointArn']
endpoint_arn = endpoint["EndpointArn"]
endpoint_list = conn.list_endpoints_by_platform_application(
PlatformApplicationArn=application_arn
)['Endpoints']
)["Endpoints"]
endpoint_list.should.have.length_of(1)
endpoint_list[0]['Attributes']['CustomUserData'].should.equal('some data')
endpoint_list[0]['EndpointArn'].should.equal(endpoint_arn)
endpoint_list[0]["Attributes"]["CustomUserData"].should.equal("some data")
endpoint_list[0]["EndpointArn"].should.equal(endpoint_arn)
@mock_sns
def test_get_endpoint_attributes():
conn = boto3.client('sns', region_name='us-east-1')
conn = boto3.client("sns", region_name="us-east-1")
platform_application = conn.create_platform_application(
Name="my-application",
Platform="APNS",
Attributes={},
Name="my-application", Platform="APNS", Attributes={}
)
application_arn = platform_application['PlatformApplicationArn']
application_arn = platform_application["PlatformApplicationArn"]
endpoint = conn.create_platform_endpoint(
PlatformApplicationArn=application_arn,
Token="some_unique_id",
CustomUserData="some user data",
Attributes={
"Enabled": 'false',
"CustomUserData": "some data",
},
Attributes={"Enabled": "false", "CustomUserData": "some data"},
)
endpoint_arn = endpoint['EndpointArn']
endpoint_arn = endpoint["EndpointArn"]
attributes = conn.get_endpoint_attributes(
EndpointArn=endpoint_arn)['Attributes']
attributes.should.equal({
"Token": "some_unique_id",
"Enabled": 'false',
"CustomUserData": "some data",
})
attributes = conn.get_endpoint_attributes(EndpointArn=endpoint_arn)["Attributes"]
attributes.should.equal(
{"Token": "some_unique_id", "Enabled": "false", "CustomUserData": "some data"}
)
@mock_sns
def test_get_missing_endpoint_attributes():
conn = boto3.client('sns', region_name='us-east-1')
conn = boto3.client("sns", region_name="us-east-1")
conn.get_endpoint_attributes.when.called_with(
EndpointArn="a-fake-arn").should.throw(ClientError)
EndpointArn="a-fake-arn"
).should.throw(ClientError)
@mock_sns
def test_set_endpoint_attributes():
conn = boto3.client('sns', region_name='us-east-1')
conn = boto3.client("sns", region_name="us-east-1")
platform_application = conn.create_platform_application(
Name="my-application",
Platform="APNS",
Attributes={},
Name="my-application", Platform="APNS", Attributes={}
)
application_arn = platform_application['PlatformApplicationArn']
application_arn = platform_application["PlatformApplicationArn"]
endpoint = conn.create_platform_endpoint(
PlatformApplicationArn=application_arn,
Token="some_unique_id",
CustomUserData="some user data",
Attributes={
"Enabled": 'false',
"CustomUserData": "some data",
},
Attributes={"Enabled": "false", "CustomUserData": "some data"},
)
endpoint_arn = endpoint['EndpointArn']
endpoint_arn = endpoint["EndpointArn"]
conn.set_endpoint_attributes(EndpointArn=endpoint_arn,
Attributes={"CustomUserData": "other data"}
)
attributes = conn.get_endpoint_attributes(
EndpointArn=endpoint_arn)['Attributes']
attributes.should.equal({
"Token": "some_unique_id",
"Enabled": 'false',
"CustomUserData": "other data",
})
conn.set_endpoint_attributes(
EndpointArn=endpoint_arn, Attributes={"CustomUserData": "other data"}
)
attributes = conn.get_endpoint_attributes(EndpointArn=endpoint_arn)["Attributes"]
attributes.should.equal(
{"Token": "some_unique_id", "Enabled": "false", "CustomUserData": "other data"}
)
@mock_sns
def test_publish_to_platform_endpoint():
conn = boto3.client('sns', region_name='us-east-1')
conn = boto3.client("sns", region_name="us-east-1")
platform_application = conn.create_platform_application(
Name="my-application",
Platform="APNS",
Attributes={},
Name="my-application", Platform="APNS", Attributes={}
)
application_arn = platform_application['PlatformApplicationArn']
application_arn = platform_application["PlatformApplicationArn"]
endpoint = conn.create_platform_endpoint(
PlatformApplicationArn=application_arn,
Token="some_unique_id",
CustomUserData="some user data",
Attributes={
"Enabled": 'true',
},
Attributes={"Enabled": "true"},
)
endpoint_arn = endpoint['EndpointArn']
endpoint_arn = endpoint["EndpointArn"]
conn.publish(Message="some message",
MessageStructure="json", TargetArn=endpoint_arn)
conn.publish(
Message="some message", MessageStructure="json", TargetArn=endpoint_arn
)
@mock_sns
def test_publish_to_disabled_platform_endpoint():
conn = boto3.client('sns', region_name='us-east-1')
conn = boto3.client("sns", region_name="us-east-1")
platform_application = conn.create_platform_application(
Name="my-application",
Platform="APNS",
Attributes={},
Name="my-application", Platform="APNS", Attributes={}
)
application_arn = platform_application['PlatformApplicationArn']
application_arn = platform_application["PlatformApplicationArn"]
endpoint = conn.create_platform_endpoint(
PlatformApplicationArn=application_arn,
Token="some_unique_id",
CustomUserData="some user data",
Attributes={
"Enabled": 'false',
},
Attributes={"Enabled": "false"},
)
endpoint_arn = endpoint['EndpointArn']
endpoint_arn = endpoint["EndpointArn"]
conn.publish.when.called_with(
Message="some message",
MessageStructure="json",
TargetArn=endpoint_arn,
Message="some message", MessageStructure="json", TargetArn=endpoint_arn
).should.throw(ClientError)
@mock_sns
def test_set_sms_attributes():
conn = boto3.client('sns', region_name='us-east-1')
conn = boto3.client("sns", region_name="us-east-1")
conn.set_sms_attributes(attributes={'DefaultSMSType': 'Transactional', 'test': 'test'})
conn.set_sms_attributes(
attributes={"DefaultSMSType": "Transactional", "test": "test"}
)
response = conn.get_sms_attributes()
response.should.contain('attributes')
response['attributes'].should.contain('DefaultSMSType')
response['attributes'].should.contain('test')
response['attributes']['DefaultSMSType'].should.equal('Transactional')
response['attributes']['test'].should.equal('test')
response.should.contain("attributes")
response["attributes"].should.contain("DefaultSMSType")
response["attributes"].should.contain("test")
response["attributes"]["DefaultSMSType"].should.equal("Transactional")
response["attributes"]["test"].should.equal("test")
@mock_sns
def test_get_sms_attributes_filtered():
conn = boto3.client('sns', region_name='us-east-1')
conn = boto3.client("sns", region_name="us-east-1")
conn.set_sms_attributes(attributes={'DefaultSMSType': 'Transactional', 'test': 'test'})
conn.set_sms_attributes(
attributes={"DefaultSMSType": "Transactional", "test": "test"}
)
response = conn.get_sms_attributes(attributes=['DefaultSMSType'])
response.should.contain('attributes')
response['attributes'].should.contain('DefaultSMSType')
response['attributes'].should_not.contain('test')
response['attributes']['DefaultSMSType'].should.equal('Transactional')
response = conn.get_sms_attributes(attributes=["DefaultSMSType"])
response.should.contain("attributes")
response["attributes"].should.contain("DefaultSMSType")
response["attributes"].should_not.contain("test")
response["attributes"]["DefaultSMSType"].should.equal("Transactional")