This commit is contained in:
Bert Blommers 2020-11-11 15:55:37 +00:00
parent cb6731f340
commit 273ca63d59
92 changed files with 515 additions and 1200 deletions

View File

@ -72,10 +72,7 @@ class ApplicationAutoscalingBackend(BaseBackend):
return applicationautoscaling_backends[self.region]
def describe_scalable_targets(
self,
namespace,
r_ids=None,
dimension=None,
self, namespace, r_ids=None, dimension=None,
):
""" Describe scalable targets. """
if r_ids is None:

View File

@ -21,10 +21,8 @@ class ApplicationAutoScalingResponse(BaseResponse):
scalable_dimension = self._get_param("ScalableDimension")
max_results = self._get_int_param("MaxResults", 50)
marker = self._get_param("NextToken")
all_scalable_targets = (
self.applicationautoscaling_backend.describe_scalable_targets(
service_namespace, resource_ids, scalable_dimension
)
all_scalable_targets = self.applicationautoscaling_backend.describe_scalable_targets(
service_namespace, resource_ids, scalable_dimension
)
start = int(marker) + 1 if marker else 0
next_token = None

View File

@ -82,12 +82,7 @@ class AthenaResponse(BaseResponse):
def error(self, msg, status):
return (
json.dumps(
{
"__type": "InvalidRequestException",
"Message": msg,
}
),
json.dumps({"__type": "InvalidRequestException", "Message": msg,}),
dict(status=status),
)

View File

@ -42,8 +42,8 @@ class AutoScalingResponse(BaseResponse):
def describe_launch_configurations(self):
names = self._get_multi_param("LaunchConfigurationNames.member")
all_launch_configurations = (
self.autoscaling_backend.describe_launch_configurations(names)
all_launch_configurations = self.autoscaling_backend.describe_launch_configurations(
names
)
marker = self._get_param("NextToken")
all_names = [lc.name for lc in all_launch_configurations]
@ -153,8 +153,8 @@ class AutoScalingResponse(BaseResponse):
@amzn_request_id
def describe_load_balancer_target_groups(self):
group_name = self._get_param("AutoScalingGroupName")
target_group_arns = (
self.autoscaling_backend.describe_load_balancer_target_groups(group_name)
target_group_arns = self.autoscaling_backend.describe_load_balancer_target_groups(
group_name
)
template = self.response_template(DESCRIBE_LOAD_BALANCER_TARGET_GROUPS)
return template.render(target_group_arns=target_group_arns)

View File

@ -254,8 +254,7 @@ def generate_resource_name(resource_type, stack_name, logical_id):
def parse_resource(
resource_json,
resources_map,
resource_json, resources_map,
):
resource_type = resource_json["Type"]
resource_class = resource_class_from_type(resource_type)
@ -276,9 +275,7 @@ def parse_resource(
def parse_resource_and_generate_name(
logical_id,
resource_json,
resources_map,
logical_id, resource_json, resources_map,
):
resource_tuple = parse_resource(resource_json, resources_map)
if not resource_tuple:
@ -698,10 +695,7 @@ class ResourceMap(collections_abc.Mapping):
]
parse_and_delete_resource(
resource_name,
resource_json,
self,
self._region_name,
resource_name, resource_json, self, self._region_name,
)
self._parsed_resources.pop(parsed_resource.logical_resource_id)

View File

@ -412,9 +412,7 @@ class CognitoIdpResponse(BaseResponse):
username = self._get_param("Username")
confirmation_code = self._get_param("ConfirmationCode")
cognitoidp_backends[self.region].confirm_sign_up(
client_id=client_id,
username=username,
confirmation_code=confirmation_code,
client_id=client_id, username=username, confirmation_code=confirmation_code,
)
return ""

View File

@ -101,10 +101,8 @@ class InvalidDeliveryChannelNameException(JsonRESTError):
code = 400
def __init__(self, name):
message = (
"The delivery channel name '{name}' is not valid, blank string.".format(
name=name
)
message = "The delivery channel name '{name}' is not valid, blank string.".format(
name=name
)
super(InvalidDeliveryChannelNameException, self).__init__(
"InvalidDeliveryChannelNameException", message
@ -289,10 +287,8 @@ class InvalidTagCharacters(JsonRESTError):
code = 400
def __init__(self, tag, param="tags.X.member.key"):
message = (
"1 validation error detected: Value '{}' at '{}' failed to satisfy ".format(
tag, param
)
message = "1 validation error detected: Value '{}' at '{}' failed to satisfy ".format(
tag, param
)
message += "constraint: Member must satisfy regular expression pattern: [\\\\p{L}\\\\p{Z}\\\\p{N}_.:/=+\\\\-@]+"

View File

@ -395,10 +395,8 @@ class OrganizationConformancePack(ConfigEmptyDictable):
self.delivery_s3_key_prefix = delivery_s3_key_prefix
self.excluded_accounts = excluded_accounts or []
self.last_update_time = datetime2int(datetime.utcnow())
self.organization_conformance_pack_arn = (
"arn:aws:config:{0}:{1}:organization-conformance-pack/{2}".format(
region, DEFAULT_ACCOUNT_ID, self._unique_pack_name
)
self.organization_conformance_pack_arn = "arn:aws:config:{0}:{1}:organization-conformance-pack/{2}".format(
region, DEFAULT_ACCOUNT_ID, self._unique_pack_name
)
self.organization_conformance_pack_name = name

View File

@ -190,10 +190,8 @@ class ConfigResponse(BaseResponse):
def get_organization_conformance_pack_detailed_status(self):
# 'Filters' parameter is not implemented yet
statuses = (
self.config_backend.get_organization_conformance_pack_detailed_status(
self._get_param("OrganizationConformancePackName")
)
statuses = self.config_backend.get_organization_conformance_pack_detailed_status(
self._get_param("OrganizationConformancePackName")
)
return json.dumps(statuses)

View File

@ -350,15 +350,11 @@ def tags_from_query_string(
tag_index = key.replace(prefix + ".", "").replace("." + key_suffix, "")
tag_key = querystring_dict.get(
"{prefix}.{index}.{key_suffix}".format(
prefix=prefix,
index=tag_index,
key_suffix=key_suffix,
prefix=prefix, index=tag_index, key_suffix=key_suffix,
)
)[0]
tag_value_key = "{prefix}.{index}.{value_suffix}".format(
prefix=prefix,
index=tag_index,
value_suffix=value_suffix,
prefix=prefix, index=tag_index, value_suffix=value_suffix,
)
if tag_value_key in querystring_dict:
response_values[tag_key] = querystring_dict.get(tag_value_key)[0]

View File

@ -1052,8 +1052,7 @@ class DynamoDBBackend(BaseBackend):
)
gsis_by_name[gsi_to_create["IndexName"]] = GlobalSecondaryIndex.create(
gsi_to_create,
table.table_key_attrs,
gsi_to_create, table.table_key_attrs,
)
# in python 3.6, dict.values() returns a dict_values object, but we expect it to be a list in other

View File

@ -340,9 +340,7 @@ class InvalidDependantParameterError(EC2ClientError):
super(InvalidDependantParameterError, self).__init__(
"InvalidParameter",
"{0} can't be empty if {1} is {2}.".format(
dependant_parameter,
parameter,
parameter_value,
dependant_parameter, parameter, parameter_value,
),
)
@ -352,9 +350,7 @@ class InvalidDependantParameterTypeError(EC2ClientError):
super(InvalidDependantParameterTypeError, self).__init__(
"InvalidParameter",
"{0} type must be {1} if {2} is provided.".format(
dependant_parameter,
parameter_value,
parameter,
dependant_parameter, parameter_value, parameter,
),
)
@ -362,8 +358,7 @@ class InvalidDependantParameterTypeError(EC2ClientError):
class InvalidAggregationIntervalParameterError(EC2ClientError):
def __init__(self, parameter):
super(InvalidAggregationIntervalParameterError, self).__init__(
"InvalidParameter",
"Invalid {0}".format(parameter),
"InvalidParameter", "Invalid {0}".format(parameter),
)

View File

@ -1491,12 +1491,7 @@ class AmiBackend(object):
# Limit by owner ids
if owners:
# support filtering by Owners=['self']
owners = list(
map(
lambda o: OWNER_ID if o == "self" else o,
owners,
)
)
owners = list(map(lambda o: OWNER_ID if o == "self" else o, owners,))
images = [ami for ami in images if ami.owner_id in owners]
# Generic filters
@ -3710,17 +3705,13 @@ class FlowLogsBackend(object):
):
if log_group_name is None and log_destination is None:
raise InvalidDependantParameterError(
"LogDestination",
"LogGroupName",
"not provided",
"LogDestination", "LogGroupName", "not provided",
)
if log_destination_type == "s3":
if log_group_name is not None:
raise InvalidDependantParameterTypeError(
"LogDestination",
"cloud-watch-logs",
"LogGroupName",
"LogDestination", "cloud-watch-logs", "LogGroupName",
)
elif log_destination_type == "cloud-watch-logs":
if deliver_logs_permission_arn is None:
@ -3868,8 +3859,7 @@ class FlowLogsBackend(object):
if non_existing:
raise InvalidFlowLogIdError(
len(flow_log_ids),
" ".join(x for x in flow_log_ids),
len(flow_log_ids), " ".join(x for x in flow_log_ids),
)
return True

View File

@ -70,8 +70,8 @@ class VPCs(BaseResponse):
def enable_vpc_classic_link_dns_support(self):
vpc_id = self._get_param("VpcId")
classic_link_dns_supported = (
self.ec2_backend.enable_vpc_classic_link_dns_support(vpc_id=vpc_id)
classic_link_dns_supported = self.ec2_backend.enable_vpc_classic_link_dns_support(
vpc_id=vpc_id
)
doc_date = self._get_doc_date()
template = self.response_template(ENABLE_VPC_CLASSIC_LINK_DNS_SUPPORT_RESPONSE)
@ -81,8 +81,8 @@ class VPCs(BaseResponse):
def disable_vpc_classic_link_dns_support(self):
vpc_id = self._get_param("VpcId")
classic_link_dns_supported = (
self.ec2_backend.disable_vpc_classic_link_dns_support(vpc_id=vpc_id)
classic_link_dns_supported = self.ec2_backend.disable_vpc_classic_link_dns_support(
vpc_id=vpc_id
)
doc_date = self._get_doc_date()
template = self.response_template(DISABLE_VPC_CLASSIC_LINK_DNS_SUPPORT_RESPONSE)

View File

@ -38,6 +38,5 @@ class ClusterNotFoundException(JsonRESTError):
def __init__(self):
super(ClusterNotFoundException, self).__init__(
error_type="ClientException",
message="Cluster not found",
error_type="ClientException", message="Cluster not found",
)

View File

@ -431,10 +431,8 @@ class ContainerInstance(BaseObject):
"type": "STRINGSET",
},
]
self.container_instance_arn = (
"arn:aws:ecs:{0}:012345678910:container-instance/{1}".format(
region_name, str(uuid.uuid4())
)
self.container_instance_arn = "arn:aws:ecs:{0}:012345678910:container-instance/{1}".format(
region_name, str(uuid.uuid4())
)
self.pending_tasks_count = 0
self.remaining_resources = [

View File

@ -8,11 +8,7 @@ from .exceptions import InvalidParameterValueError, ResourceNotFoundException
class FakeEnvironment(BaseModel):
def __init__(
self,
application,
environment_name,
solution_stack_name,
tags,
self, application, environment_name, solution_stack_name, tags,
):
self.application = weakref.proxy(
application
@ -53,10 +49,7 @@ class FakeApplication(BaseModel):
self.environments = dict()
def create_environment(
self,
environment_name,
solution_stack_name,
tags,
self, environment_name, solution_stack_name, tags,
):
if environment_name in self.environments:
raise InvalidParameterValueError
@ -93,10 +86,7 @@ class EBBackend(BaseBackend):
raise InvalidParameterValueError(
"Application {} already exists.".format(application_name)
)
new_app = FakeApplication(
backend=self,
application_name=application_name,
)
new_app = FakeApplication(backend=self, application_name=application_name,)
self.applications[application_name] = new_app
return new_app

View File

@ -18,16 +18,11 @@ class EBResponse(BaseResponse):
)
template = self.response_template(EB_CREATE_APPLICATION)
return template.render(
region_name=self.backend.region,
application=app,
)
return template.render(region_name=self.backend.region, application=app,)
def describe_applications(self):
template = self.response_template(EB_DESCRIBE_APPLICATIONS)
return template.render(
applications=self.backend.applications.values(),
)
return template.render(applications=self.backend.applications.values(),)
def create_environment(self):
application_name = self._get_param("ApplicationName")
@ -47,18 +42,13 @@ class EBResponse(BaseResponse):
)
template = self.response_template(EB_CREATE_ENVIRONMENT)
return template.render(
environment=env,
region=self.backend.region,
)
return template.render(environment=env, region=self.backend.region,)
def describe_environments(self):
envs = self.backend.describe_environments()
template = self.response_template(EB_DESCRIBE_ENVIRONMENTS)
return template.render(
environments=envs,
)
return template.render(environments=envs,)
def list_available_solution_stacks(self):
return EB_LIST_AVAILABLE_SOLUTION_STACKS
@ -78,10 +68,7 @@ class EBResponse(BaseResponse):
tags = self.backend.list_tags_for_resource(resource_arn)
template = self.response_template(EB_LIST_TAGS_FOR_RESOURCE)
return template.render(
tags=tags,
arn=resource_arn,
)
return template.render(tags=tags, arn=resource_arn,)
EB_CREATE_APPLICATION = """

View File

@ -125,12 +125,10 @@ class AssumedRoleAccessKey(object):
@property
def arn(self):
return (
"arn:aws:sts::{account_id}:assumed-role/{role_name}/{session_name}".format(
account_id=ACCOUNT_ID,
role_name=self._owner_role_name,
session_name=self._session_name,
)
return "arn:aws:sts::{account_id}:assumed-role/{role_name}/{session_name}".format(
account_id=ACCOUNT_ID,
role_name=self._owner_role_name,
session_name=self._session_name,
)
def create_credentials(self):

View File

@ -88,10 +88,8 @@ class InvalidTagCharacters(RESTError):
code = 400
def __init__(self, tag, param="tags.X.member.key"):
message = (
"1 validation error detected: Value '{}' at '{}' failed to satisfy ".format(
tag, param
)
message = "1 validation error detected: Value '{}' at '{}' failed to satisfy ".format(
tag, param
)
message += "constraint: Member must satisfy regular expression pattern: [\\p{L}\\p{Z}\\p{N}_.:/=+\\-@]+"

View File

@ -362,12 +362,7 @@ class InlinePolicy(CloudFormationModel):
self.update(policy_name, policy_document, group_names, role_names, user_names)
def update(
self,
policy_name,
policy_document,
group_names,
role_names,
user_names,
self, policy_name, policy_document, group_names, role_names, user_names,
):
self.policy_name = policy_name
self.policy_document = (
@ -409,11 +404,7 @@ class InlinePolicy(CloudFormationModel):
@classmethod
def update_from_cloudformation_json(
cls,
original_resource,
new_resource_name,
cloudformation_json,
region_name,
cls, original_resource, new_resource_name, cloudformation_json, region_name,
):
properties = cloudformation_json["Properties"]
@ -816,18 +807,11 @@ class AccessKey(CloudFormationModel):
user_name = properties.get("UserName")
status = properties.get("Status", "Active")
return iam_backend.create_access_key(
user_name,
status=status,
)
return iam_backend.create_access_key(user_name, status=status,)
@classmethod
def update_from_cloudformation_json(
cls,
original_resource,
new_resource_name,
cloudformation_json,
region_name,
cls, original_resource, new_resource_name, cloudformation_json, region_name,
):
properties = cloudformation_json["Properties"]
@ -1155,11 +1139,7 @@ class User(CloudFormationModel):
@classmethod
def update_from_cloudformation_json(
cls,
original_resource,
new_resource_name,
cloudformation_json,
region_name,
cls, original_resource, new_resource_name, cloudformation_json, region_name,
):
properties = cloudformation_json["Properties"]
@ -2577,11 +2557,7 @@ class IAMBackend(BaseBackend):
inline_policy = self.get_inline_policy(resource_name)
inline_policy.unapply_policy(self)
inline_policy.update(
policy_name,
policy_document,
group_names,
role_names,
user_names,
policy_name, policy_document, group_names, role_names, user_names,
)
inline_policy.apply_policy(self)
return inline_policy

View File

@ -343,10 +343,8 @@ class IAMPolicyDocumentValidator:
resource_partitions = resource.partition(":")
if resource_partitions[1] == "":
self._resource_error = (
'Resource {resource} must be in ARN format or "*".'.format(
resource=resource
)
self._resource_error = 'Resource {resource} must be in ARN format or "*".'.format(
resource=resource
)
return
@ -392,14 +390,15 @@ class IAMPolicyDocumentValidator:
service = resource_partitions[0]
if (
service in SERVICE_TYPE_REGION_INFORMATION_ERROR_ASSOCIATIONS.keys()
and not resource_partitions[2].startswith(":")
if service in SERVICE_TYPE_REGION_INFORMATION_ERROR_ASSOCIATIONS.keys() and not resource_partitions[
2
].startswith(
":"
):
self._resource_error = (
SERVICE_TYPE_REGION_INFORMATION_ERROR_ASSOCIATIONS[service].format(
resource=resource
)
self._resource_error = SERVICE_TYPE_REGION_INFORMATION_ERROR_ASSOCIATIONS[
service
].format(
resource=resource
)
return
@ -521,8 +520,8 @@ class IAMPolicyDocumentValidator:
assert 0 <= int(time_zone_minutes) <= 59
else:
seconds_with_decimal_fraction = time_parts[2]
seconds_with_decimal_fraction_partition = (
seconds_with_decimal_fraction.partition(".")
seconds_with_decimal_fraction_partition = seconds_with_decimal_fraction.partition(
"."
)
seconds = seconds_with_decimal_fraction_partition[0]
assert 0 <= int(seconds) <= 59

View File

@ -340,8 +340,7 @@ class IoTResponse(BaseResponse):
status = self._get_param("status")
cert = self.iot_backend.register_certificate_without_ca(
certificate_pem=certificate_pem,
status=status,
certificate_pem=certificate_pem, status=status,
)
return json.dumps(
dict(certificateId=cert.certificate_id, certificateArn=cert.arn)

View File

@ -261,11 +261,7 @@ class Stream(CloudFormationModel):
@classmethod
def update_from_cloudformation_json(
cls,
original_resource,
new_resource_name,
cloudformation_json,
region_name,
cls, original_resource, new_resource_name, cloudformation_json, region_name,
):
properties = cloudformation_json["Properties"]

View File

@ -20,6 +20,5 @@ class ResourceInUseException(KinesisvideoClientError):
def __init__(self, message):
self.code = 400
super(ResourceInUseException, self).__init__(
"ResourceInUseException",
message,
"ResourceInUseException", message,
)

View File

@ -32,8 +32,7 @@ class KinesisVideoResponse(BaseResponse):
stream_name = self._get_param("StreamName")
stream_arn = self._get_param("StreamARN")
stream_info = self.kinesisvideo_backend.describe_stream(
stream_name=stream_name,
stream_arn=stream_arn,
stream_name=stream_name, stream_arn=stream_arn,
)
return json.dumps(dict(StreamInfo=stream_info))
@ -52,8 +51,7 @@ class KinesisVideoResponse(BaseResponse):
stream_arn = self._get_param("StreamARN")
current_version = self._get_param("CurrentVersion")
self.kinesisvideo_backend.delete_stream(
stream_arn=stream_arn,
current_version=current_version,
stream_arn=stream_arn, current_version=current_version,
)
return json.dumps(dict())
@ -62,8 +60,6 @@ class KinesisVideoResponse(BaseResponse):
stream_arn = self._get_param("StreamARN")
api_name = self._get_param("APIName")
data_endpoint = self.kinesisvideo_backend.get_data_endpoint(
stream_name=stream_name,
stream_arn=stream_arn,
api_name=api_name,
stream_name=stream_name, stream_arn=stream_arn, api_name=api_name,
)
return json.dumps(dict(DataEndpoint=data_endpoint))

View File

@ -23,18 +23,16 @@ class KinesisVideoArchivedMediaResponse(BaseResponse):
max_media_playlist_fragment_results = self._get_param(
"MaxMediaPlaylistFragmentResults"
)
hls_streaming_session_url = (
self.kinesisvideoarchivedmedia_backend.get_hls_streaming_session_url(
stream_name=stream_name,
stream_arn=stream_arn,
playback_mode=playback_mode,
hls_fragment_selector=hls_fragment_selector,
container_format=container_format,
discontinuity_mode=discontinuity_mode,
display_fragment_timestamp=display_fragment_timestamp,
expires=expires,
max_media_playlist_fragment_results=max_media_playlist_fragment_results,
)
hls_streaming_session_url = self.kinesisvideoarchivedmedia_backend.get_hls_streaming_session_url(
stream_name=stream_name,
stream_arn=stream_arn,
playback_mode=playback_mode,
hls_fragment_selector=hls_fragment_selector,
container_format=container_format,
discontinuity_mode=discontinuity_mode,
display_fragment_timestamp=display_fragment_timestamp,
expires=expires,
max_media_playlist_fragment_results=max_media_playlist_fragment_results,
)
return json.dumps(dict(HLSStreamingSessionURL=hls_streaming_session_url))
@ -47,17 +45,15 @@ class KinesisVideoArchivedMediaResponse(BaseResponse):
dash_fragment_selector = self._get_param("DASHFragmentSelector")
expires = self._get_int_param("Expires")
max_manifest_fragment_results = self._get_param("MaxManifestFragmentResults")
dash_streaming_session_url = (
self.kinesisvideoarchivedmedia_backend.get_dash_streaming_session_url(
stream_name=stream_name,
stream_arn=stream_arn,
playback_mode=playback_mode,
display_fragment_timestamp=display_fragment_timestamp,
display_fragment_number=display_fragment_number,
dash_fragment_selector=dash_fragment_selector,
expires=expires,
max_manifest_fragment_results=max_manifest_fragment_results,
)
dash_streaming_session_url = self.kinesisvideoarchivedmedia_backend.get_dash_streaming_session_url(
stream_name=stream_name,
stream_arn=stream_arn,
playback_mode=playback_mode,
display_fragment_timestamp=display_fragment_timestamp,
display_fragment_number=display_fragment_number,
dash_fragment_selector=dash_fragment_selector,
expires=expires,
max_manifest_fragment_results=max_manifest_fragment_results,
)
return json.dumps(dict(DASHStreamingSessionURL=dash_streaming_session_url))

View File

@ -352,11 +352,7 @@ class ManagedBlockchainInvitation(BaseModel):
class ManagedBlockchainMember(BaseModel):
def __init__(
self,
id,
networkid,
member_configuration,
region,
self, id, networkid, member_configuration, region,
):
self.creationdate = datetime.datetime.utcnow()
self.id = id
@ -587,11 +583,7 @@ class ManagedBlockchainBackend(BaseBackend):
return self.networks.get(network_id)
def create_proposal(
self,
networkid,
memberid,
actions,
description=None,
self, networkid, memberid, actions, description=None,
):
# Check if network exists
if networkid not in self.networks:
@ -791,10 +783,7 @@ class ManagedBlockchainBackend(BaseBackend):
self.invitations.get(invitationid).reject_invitation()
def create_member(
self,
invitationid,
networkid,
member_configuration,
self, invitationid, networkid, member_configuration,
):
# Check if network exists
if networkid not in self.networks:
@ -999,8 +988,7 @@ class ManagedBlockchainBackend(BaseBackend):
chkregionpreregex = self.region_name + "[a-z]"
if re.match(chkregionpreregex, availabilityzone, re.IGNORECASE) is None:
raise InvalidRequestException(
"CreateNode",
"Availability Zone is not valid",
"CreateNode", "Availability Zone is not valid",
)
node_id = get_node_id()

View File

@ -134,10 +134,7 @@ class ManagedBlockchainResponse(BaseResponse):
description = json_body.get("Description", None)
response = self.backend.create_proposal(
network_id,
memberid,
actions,
description,
network_id, memberid, actions, description,
)
return 200, headers, json.dumps(response)
@ -201,10 +198,7 @@ class ManagedBlockchainResponse(BaseResponse):
vote = json_body["Vote"]
self.backend.vote_on_proposal(
network_id,
proposal_id,
votermemberid,
vote,
network_id, proposal_id, votermemberid, vote,
)
return 200, headers, ""
@ -284,9 +278,7 @@ class ManagedBlockchainResponse(BaseResponse):
member_configuration = json_body["MemberConfiguration"]
response = self.backend.create_member(
invitationid,
network_id,
member_configuration,
invitationid, network_id, member_configuration,
)
return 200, headers, json.dumps(response)
@ -325,9 +317,7 @@ class ManagedBlockchainResponse(BaseResponse):
def _memberid_response_patch(self, network_id, member_id, json_body, headers):
logpublishingconfiguration = json_body["LogPublishingConfiguration"]
self.backend.update_member(
network_id,
member_id,
logpublishingconfiguration,
network_id, member_id, logpublishingconfiguration,
)
return 200, headers, ""
@ -427,10 +417,7 @@ class ManagedBlockchainResponse(BaseResponse):
):
logpublishingconfiguration = json_body
self.backend.update_node(
network_id,
member_id,
node_id,
logpublishingconfiguration,
network_id, member_id, node_id, logpublishingconfiguration,
)
return 200, headers, ""

View File

@ -785,8 +785,7 @@ class OrganizationsBackend(BaseBackend):
)
admin = next(
(admin for admin in self.admins if admin.account.id == account_id),
None,
(admin for admin in self.admins if admin.account.id == account_id), None,
)
if admin is None:
account = next(
@ -842,8 +841,7 @@ class OrganizationsBackend(BaseBackend):
)
elif re.match(account_id_regex, target_id):
account = next(
(account for account in self.accounts if account.id == target_id),
None,
(account for account in self.accounts if account.id == target_id), None,
)
if account is not None:
if account in account.attached_policies:

View File

@ -88,10 +88,8 @@ class ResourceShare(BaseModel):
)
if root_id:
ous = (
self.organizations_backend.list_organizational_units_for_parent(
ParentId=root_id
)
ous = self.organizations_backend.list_organizational_units_for_parent(
ParentId=root_id
)
if any(principal == ou["Arn"] for ou in ous["OrganizationalUnits"]):
continue

View File

@ -523,10 +523,7 @@ class LifecycleAndFilter(BaseModel):
for key, value in self.tags.items():
data.append(
{
"type": "LifecycleTagPredicate",
"tag": {"key": key, "value": value},
}
{"type": "LifecycleTagPredicate", "tag": {"key": key, "value": value},}
)
return data
@ -1132,11 +1129,7 @@ class FakeBucket(CloudFormationModel):
@classmethod
def update_from_cloudformation_json(
cls,
original_resource,
new_resource_name,
cloudformation_json,
region_name,
cls, original_resource, new_resource_name, cloudformation_json, region_name,
):
properties = cloudformation_json["Properties"]
@ -1476,8 +1469,7 @@ class S3Backend(BaseBackend):
raise MissingKey(key_name)
self.tagger.delete_all_tags_for_resource(key.arn)
self.tagger.tag_resource(
key.arn,
[{"Key": k, "Value": v} for (k, v) in tags.items()],
key.arn, [{"Key": k, "Value": v} for (k, v) in tags.items()],
)
return key
@ -1489,8 +1481,7 @@ class S3Backend(BaseBackend):
bucket = self.get_bucket(bucket_name)
self.tagger.delete_all_tags_for_resource(bucket.arn)
self.tagger.tag_resource(
bucket.arn,
[{"Key": key, "Value": value} for key, value in tags.items()],
bucket.arn, [{"Key": key, "Value": value} for key, value in tags.items()],
)
def delete_bucket_tagging(self, bucket_name):

View File

@ -406,8 +406,8 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
template = self.response_template(S3_BUCKET_CORS_RESPONSE)
return template.render(cors=cors)
elif "notification" in querystring:
notification_configuration = (
self.backend.get_bucket_notification_configuration(bucket_name)
notification_configuration = self.backend.get_bucket_notification_configuration(
bucket_name
)
if not notification_configuration:
return 200, {}, ""

View File

@ -517,10 +517,8 @@ class FakeSageMakerNotebookInstanceLifecycleConfig(BaseObject):
self.creation_time = self.last_modified_time = datetime.now().strftime(
"%Y-%m-%d %H:%M:%S"
)
self.notebook_instance_lifecycle_config_arn = (
FakeSageMakerNotebookInstanceLifecycleConfig.arn_formatter(
self.notebook_instance_lifecycle_config_name, self.region_name
)
self.notebook_instance_lifecycle_config_arn = FakeSageMakerNotebookInstanceLifecycleConfig.arn_formatter(
self.notebook_instance_lifecycle_config_name, self.region_name
)
@staticmethod
@ -583,9 +581,7 @@ class SageMakerModelBackend(BaseBackend):
Model.arn_for_model_name(model_name, self.region_name)
)
raise RESTError(
error_type="ValidationException",
message=message,
template="error_json",
error_type="ValidationException", message=message, template="error_json",
)
def list_models(self):
@ -796,10 +792,7 @@ class SageMakerModelBackend(BaseBackend):
raise ValidationError(message=message)
def create_endpoint(
self,
endpoint_name,
endpoint_config_name,
tags,
self, endpoint_name, endpoint_config_name, tags,
):
try:
endpoint_config = self.describe_endpoint_config(endpoint_config_name)

View File

@ -243,14 +243,12 @@ class SageMakerResponse(BaseResponse):
@amzn_request_id
def create_notebook_instance_lifecycle_config(self):
try:
lifecycle_configuration = (
self.sagemaker_backend.create_notebook_instance_lifecycle_config(
notebook_instance_lifecycle_config_name=self._get_param(
"NotebookInstanceLifecycleConfigName"
),
on_create=self._get_param("OnCreate"),
on_start=self._get_param("OnStart"),
)
lifecycle_configuration = self.sagemaker_backend.create_notebook_instance_lifecycle_config(
notebook_instance_lifecycle_config_name=self._get_param(
"NotebookInstanceLifecycleConfigName"
),
on_create=self._get_param("OnCreate"),
on_start=self._get_param("OnStart"),
)
response = {
"NotebookInstanceLifecycleConfigArn": lifecycle_configuration.notebook_instance_lifecycle_config_arn,

View File

@ -340,14 +340,12 @@ class PlatformEndpoint(BaseModel):
@property
def arn(self):
return (
"arn:aws:sns:{region}:{AccountId}:endpoint/{platform}/{name}/{id}".format(
region=self.region,
AccountId=DEFAULT_ACCOUNT_ID,
platform=self.application.platform,
name=self.application.name,
id=self.id,
)
return "arn:aws:sns:{region}:{AccountId}:endpoint/{platform}/{name}/{id}".format(
region=self.region,
AccountId=DEFAULT_ACCOUNT_ID,
platform=self.application.platform,
name=self.application.name,
id=self.id,
)
def publish(self, message):

View File

@ -354,9 +354,7 @@ class SQSResponse(BaseResponse):
queue_name = self._get_queue_name()
message_attributes = self._get_multi_param("message_attributes")
if not message_attributes:
message_attributes = extract_input_message_attributes(
self.querystring,
)
message_attributes = extract_input_message_attributes(self.querystring,)
queue = self.sqs_backend.get_queue(queue_name)
@ -720,10 +718,8 @@ ERROR_TOO_LONG_RESPONSE = """<ErrorResponse xmlns="http://queue.amazonaws.com/do
<RequestId>6fde8d1e-52cd-4581-8cd9-c512f4c64223</RequestId>
</ErrorResponse>"""
ERROR_MAX_VISIBILITY_TIMEOUT_RESPONSE = (
"Invalid request, maximum visibility timeout is {0}".format(
MAXIMUM_VISIBILTY_TIMEOUT
)
ERROR_MAX_VISIBILITY_TIMEOUT_RESPONSE = "Invalid request, maximum visibility timeout is {0}".format(
MAXIMUM_VISIBILTY_TIMEOUT
)
ERROR_INEXISTENT_QUEUE = """<ErrorResponse xmlns="http://queue.amazonaws.com/doc/2012-11-05/">

View File

@ -148,9 +148,7 @@ class StateMachine(CloudFormationModel):
tags = cfn_to_api_tags(properties.get("Tags", []))
sf_backend = stepfunction_backends[region_name]
state_machine = sf_backend.update_state_machine(
original_resource.arn,
definition=definition,
role_arn=role_arn,
original_resource.arn, definition=definition, role_arn=role_arn,
)
state_machine.add_tags(tags)
return state_machine

View File

@ -48,12 +48,10 @@ class AssumedRole(BaseModel):
@property
def arn(self):
return (
"arn:aws:sts::{account_id}:assumed-role/{role_name}/{session_name}".format(
account_id=ACCOUNT_ID,
role_name=self.role_arn.split("/")[-1],
session_name=self.session_name,
)
return "arn:aws:sts::{account_id}:assumed-role/{role_name}/{session_name}".format(
account_id=ACCOUNT_ID,
role_name=self.role_arn.split("/")[-1],
session_name=self.session_name,
)

View File

@ -153,11 +153,7 @@ class FakeMedicalTranscriptionJob(BaseObject):
class FakeMedicalVocabulary(BaseObject):
def __init__(
self,
region_name,
vocabulary_name,
language_code,
vocabulary_file_uri,
self, region_name, vocabulary_name, language_code, vocabulary_file_uri,
):
self._region_name = region_name
self.vocabulary_name = vocabulary_name

View File

@ -20,10 +20,8 @@ SERVER_CRT = _GET_RESOURCE("star_moto_com.pem")
SERVER_COMMON_NAME = "*.moto.com"
SERVER_CRT_BAD = _GET_RESOURCE("star_moto_com-bad.pem")
SERVER_KEY = _GET_RESOURCE("star_moto_com.key")
BAD_ARN = (
"arn:aws:acm:us-east-2:{}:certificate/_0000000-0000-0000-0000-000000000000".format(
ACCOUNT_ID
)
BAD_ARN = "arn:aws:acm:us-east-2:{}:certificate/_0000000-0000-0000-0000-000000000000".format(
ACCOUNT_ID
)
@ -56,10 +54,7 @@ def test_import_certificate_with_tags():
Certificate=SERVER_CRT,
PrivateKey=SERVER_KEY,
CertificateChain=CA_CRT,
Tags=[
{"Key": "Environment", "Value": "QA"},
{"Key": "KeyOnly"},
],
Tags=[{"Key": "Environment", "Value": "QA"}, {"Key": "KeyOnly"},],
)
arn = resp["CertificateArn"]
@ -371,10 +366,7 @@ def test_request_certificate_with_tags():
DomainName="google.com",
IdempotencyToken=token,
SubjectAlternativeNames=["google.com", "www.google.com", "mail.google.com"],
Tags=[
{"Key": "Environment", "Value": "Prod"},
{"Key": "KeyOnly"},
],
Tags=[{"Key": "Environment", "Value": "Prod"}, {"Key": "KeyOnly"},],
)
arn_2 = resp["CertificateArn"]
@ -404,8 +396,7 @@ def test_operations_with_invalid_tags():
# request certificate with invalid tags
with pytest.raises(ClientError) as ex:
client.request_certificate(
DomainName="example.com",
Tags=[{"Key": "X" * 200, "Value": "Valid"}],
DomainName="example.com", Tags=[{"Key": "X" * 200, "Value": "Valid"}],
)
ex.value.response["Error"]["Code"].should.equal("ValidationException")
ex.value.response["Error"]["Message"].should.contain(
@ -567,7 +558,11 @@ def test_request_certificate_with_mutiple_times():
resp = client.request_certificate(
IdempotencyToken="test_token",
DomainName="google.com",
SubjectAlternativeNames=["google.com", "www.google.com", "mail.google.com"],
SubjectAlternativeNames=[
"google.com",
"www.google.com",
"mail.google.com",
],
)
arn = resp["CertificateArn"]
arn.should.equal(original_arn)

View File

@ -105,9 +105,7 @@ def test_create_rest_api_valid_apikeysources():
# 1. test creating rest api with HEADER apiKeySource
response = client.create_rest_api(
name="my_api",
description="this is my api",
apiKeySource="HEADER",
name="my_api", description="this is my api", apiKeySource="HEADER",
)
api_id = response["id"]
@ -116,9 +114,7 @@ def test_create_rest_api_valid_apikeysources():
# 2. test creating rest api with AUTHORIZER apiKeySource
response = client.create_rest_api(
name="my_api2",
description="this is my api",
apiKeySource="AUTHORIZER",
name="my_api2", description="this is my api", apiKeySource="AUTHORIZER",
)
api_id = response["id"]
@ -153,9 +149,7 @@ def test_create_rest_api_valid_endpointconfigurations():
response = client.get_rest_api(restApiId=api_id)
response["endpointConfiguration"].should.equal(
{
"types": ["PRIVATE"],
}
{"types": ["PRIVATE"],}
)
# 2. test creating rest api with REGIONAL endpointConfiguration
@ -168,9 +162,7 @@ def test_create_rest_api_valid_endpointconfigurations():
response = client.get_rest_api(restApiId=api_id)
response["endpointConfiguration"].should.equal(
{
"types": ["REGIONAL"],
}
{"types": ["REGIONAL"],}
)
# 3. test creating rest api with EDGE endpointConfiguration
@ -183,9 +175,7 @@ def test_create_rest_api_valid_endpointconfigurations():
response = client.get_rest_api(restApiId=api_id)
response["endpointConfiguration"].should.equal(
{
"types": ["EDGE"],
}
{"types": ["EDGE"],}
)
@ -231,11 +221,7 @@ def test_create_resource():
root_resource["ResponseMetadata"].pop("HTTPHeaders", None)
root_resource["ResponseMetadata"].pop("RetryAttempts", None)
root_resource.should.equal(
{
"path": "/",
"id": root_id,
"ResponseMetadata": {"HTTPStatusCode": 200},
}
{"path": "/", "id": root_id, "ResponseMetadata": {"HTTPStatusCode": 200},}
)
client.create_resource(restApiId=api_id, parentId=root_id, pathPart="users")
@ -1834,10 +1820,8 @@ def test_http_proxying_integration():
stage_name = "staging"
client.create_deployment(restApiId=api_id, stageName=stage_name)
deploy_url = (
"https://{api_id}.execute-api.{region_name}.amazonaws.com/{stage_name}".format(
api_id=api_id, region_name=region_name, stage_name=stage_name
)
deploy_url = "https://{api_id}.execute-api.{region_name}.amazonaws.com/{stage_name}".format(
api_id=api_id, region_name=region_name, stage_name=stage_name
)
if not settings.TEST_SERVER_MODE:

View File

@ -513,6 +513,4 @@ def test_deregister_scalable_target():
ResourceId=resource_id,
ScalableDimension=scalable_dimension,
)
e.value.response["Error"]["Message"].should.match(
r"No scalable target found .*"
)
e.value.response["Error"]["Message"].should.match(r"No scalable target found .*")

View File

@ -48,8 +48,7 @@ def test_describe_scalable_targets_with_invalid_scalable_dimension_should_return
with pytest.raises(ClientError) as err:
response = client.describe_scalable_targets(
ServiceNamespace=DEFAULT_SERVICE_NAMESPACE,
ScalableDimension="foo",
ServiceNamespace=DEFAULT_SERVICE_NAMESPACE, ScalableDimension="foo",
)
err.response["Error"]["Code"].should.equal("ValidationException")
err.response["Error"]["Message"].split(":")[0].should.look_like(
@ -64,8 +63,7 @@ def test_describe_scalable_targets_with_invalid_service_namespace_should_return_
with pytest.raises(ClientError) as err:
response = client.describe_scalable_targets(
ServiceNamespace="foo",
ScalableDimension=DEFAULT_SCALABLE_DIMENSION,
ServiceNamespace="foo", ScalableDimension=DEFAULT_SCALABLE_DIMENSION,
)
err.response["Error"]["Code"].should.equal("ValidationException")
err.response["Error"]["Message"].split(":")[0].should.look_like(
@ -80,8 +78,7 @@ def test_describe_scalable_targets_with_multiple_invalid_parameters_should_retur
with pytest.raises(ClientError) as err:
response = client.describe_scalable_targets(
ServiceNamespace="foo",
ScalableDimension="bar",
ServiceNamespace="foo", ScalableDimension="bar",
)
err.response["Error"]["Code"].should.equal("ValidationException")
err.response["Error"]["Message"].split(":")[0].should.look_like(
@ -105,12 +102,13 @@ def test_register_scalable_target_ecs_with_non_existent_service_should_return_va
err.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
@pytest.mark.parametrize("namespace,r_id,dimension,expected",
@pytest.mark.parametrize(
"namespace,r_id,dimension,expected",
[
("ecs", "service/default/test-svc", "ecs:service:DesiredCount", True),
("ecs", "banana/default/test-svc", "ecs:service:DesiredCount", False),
("rds", "service/default/test-svc", "ecs:service:DesiredCount", False),
]
],
)
def test_target_params_are_valid_success(namespace, r_id, dimension, expected):
if expected is True:

View File

@ -178,9 +178,7 @@ def test_create_named_query():
# craete named query
res = client.create_named_query(
Name="query-name",
Database="target_db",
QueryString="SELECT * FROM table1",
Name="query-name", Database="target_db", QueryString="SELECT * FROM table1",
)
assert "NamedQueryId" in res
@ -217,8 +215,6 @@ def create_basic_workgroup(client, name):
Name=name,
Description="Test work group",
Configuration={
"ResultConfiguration": {
"OutputLocation": "s3://bucket-name/prefix/",
}
"ResultConfiguration": {"OutputLocation": "s3://bucket-name/prefix/",}
},
)

View File

@ -961,8 +961,7 @@ def test_describe_autoscaling_groups_boto3_launch_config():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
client.create_launch_configuration(
LaunchConfigurationName="test_launch_configuration",
InstanceType="t2.micro",
LaunchConfigurationName="test_launch_configuration", InstanceType="t2.micro",
)
client.create_auto_scaling_group(
AutoScalingGroupName="test_asg",
@ -1041,8 +1040,7 @@ def test_describe_autoscaling_instances_boto3_launch_config():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
client.create_launch_configuration(
LaunchConfigurationName="test_launch_configuration",
InstanceType="t2.micro",
LaunchConfigurationName="test_launch_configuration", InstanceType="t2.micro",
)
client.create_auto_scaling_group(
AutoScalingGroupName="test_asg",
@ -2156,8 +2154,7 @@ def test_standby_exit_standby():
response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby")
response = client.exit_standby(
AutoScalingGroupName="test_asg",
InstanceIds=[instance_to_standby_exit_standby],
AutoScalingGroupName="test_asg", InstanceIds=[instance_to_standby_exit_standby],
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)

View File

@ -32,8 +32,7 @@ Outputs:
""".strip()
cf_client.create_stack(
StackName=stack_name,
TemplateBody=cf_template,
StackName=stack_name, TemplateBody=cf_template,
)
stack = cf_client.describe_stacks(StackName=stack_name)["Stacks"][0]
stack["Outputs"][0]["OutputValue"].should.be.equal("test_launch_configuration")
@ -57,8 +56,7 @@ Outputs:
""".strip()
cf_client.update_stack(
StackName=stack_name,
TemplateBody=cf_template,
StackName=stack_name, TemplateBody=cf_template,
)
stack = cf_client.describe_stacks(StackName=stack_name)["Stacks"][0]
stack["Outputs"][0]["OutputValue"].should.be.equal("test_launch_configuration")
@ -78,8 +76,7 @@ def test_autoscaling_group_from_launch_config():
client = boto3.client("autoscaling", region_name="us-east-1")
client.create_launch_configuration(
LaunchConfigurationName="test_launch_configuration",
InstanceType="t2.micro",
LaunchConfigurationName="test_launch_configuration", InstanceType="t2.micro",
)
stack_name = "test-auto-scaling-group"

View File

@ -208,9 +208,7 @@ def test_invoke_dryrun_function():
Runtime="python2.7",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={
"ZipFile": get_test_zip_file1(),
},
Code={"ZipFile": get_test_zip_file1(),},
Description="test lambda function",
Timeout=3,
MemorySize=128,
@ -1287,8 +1285,7 @@ def wait_for_log_msg(expected_msg, log_group):
for log_stream in log_streams:
result = logs_conn.get_log_events(
logGroupName=log_group,
logStreamName=log_stream["logStreamName"],
logGroupName=log_group, logStreamName=log_stream["logStreamName"],
)
received_messages.extend(
[event["message"] for event in result.get("events")]
@ -1727,9 +1724,7 @@ def test_remove_function_permission():
)
remove = conn.remove_permission(
FunctionName="testFunction",
StatementId="1",
Qualifier="2",
FunctionName="testFunction", StatementId="1", Qualifier="2",
)
remove["ResponseMetadata"]["HTTPStatusCode"].should.equal(204)
policy = conn.get_policy(FunctionName="testFunction", Qualifier="2")["Policy"]

View File

@ -738,8 +738,9 @@ def test_submit_job():
else:
raise RuntimeError("Batch job timed out")
resp = logs_client.describe_log_streams(logGroupName="/aws/batch/job",
logStreamNamePrefix="sayhellotomylittlefriend")
resp = logs_client.describe_log_streams(
logGroupName="/aws/batch/job", logStreamNamePrefix="sayhellotomylittlefriend"
)
len(resp["logStreams"]).should.equal(1)
ls_name = resp["logStreams"][0]["logStreamName"]

View File

@ -23,9 +23,7 @@ depends_on_template_list = {
},
"LaunchConfig": {
"Type": "AWS::AutoScaling::LaunchConfiguration",
"Properties": {
"LaunchConfigurationName": "test-launch-config",
},
"Properties": {"LaunchConfigurationName": "test-launch-config",},
},
},
}
@ -47,9 +45,7 @@ depends_on_template_string = {
},
"LaunchConfig": {
"Type": "AWS::AutoScaling::LaunchConfiguration",
"Properties": {
"LaunchConfigurationName": "test-launch-config",
},
"Properties": {"LaunchConfigurationName": "test-launch-config",},
},
},
}

View File

@ -1369,12 +1369,10 @@ def test_non_json_redrive_policy():
def test_boto3_create_duplicate_stack():
cf_conn = boto3.client("cloudformation", region_name="us-east-1")
cf_conn.create_stack(
StackName="test_stack",
TemplateBody=dummy_template_json,
StackName="test_stack", TemplateBody=dummy_template_json,
)
with pytest.raises(ClientError):
cf_conn.create_stack(
StackName="test_stack",
TemplateBody=dummy_template_json,
StackName="test_stack", TemplateBody=dummy_template_json,
)

View File

@ -2325,10 +2325,7 @@ def test_stack_dynamodb_resources_integration():
dynamodb_client = boto3.client("dynamodb", region_name="us-east-1")
table_desc = dynamodb_client.describe_table(TableName="myTableName")["Table"]
table_desc["StreamSpecification"].should.equal(
{
"StreamEnabled": True,
"StreamViewType": "KEYS_ONLY",
}
{"StreamEnabled": True, "StreamViewType": "KEYS_ONLY",}
)
dynamodb_conn = boto3.resource("dynamodb", region_name="us-east-1")
@ -2782,9 +2779,7 @@ def test_stack_events_get_attribute_integration():
@mock_dynamodb2
def test_dynamodb_table_creation():
CFN_TEMPLATE = {
"Outputs": {
"MyTableName": {"Value": {"Ref": "MyTable"}},
},
"Outputs": {"MyTableName": {"Value": {"Ref": "MyTable"}},},
"Resources": {
"MyTable": {
"Type": "AWS::DynamoDB::Table",

View File

@ -326,9 +326,7 @@ def test_update_pipeline():
"S3Bucket": "different-bucket",
"S3ObjectKey": "test-object",
},
"outputArtifacts": [
{"name": "artifact"},
],
"outputArtifacts": [{"name": "artifact"},],
},
],
},
@ -437,9 +435,7 @@ def test_update_pipeline_errors():
"S3Bucket": "test-bucket",
"S3ObjectKey": "test-object",
},
"outputArtifacts": [
{"name": "artifact"},
],
"outputArtifacts": [{"name": "artifact"},],
},
],
},
@ -700,9 +696,7 @@ def create_basic_codepipeline(client, name):
"S3Bucket": "test-bucket",
"S3ObjectKey": "test-object",
},
"outputArtifacts": [
{"name": "artifact"},
],
"outputArtifacts": [{"name": "artifact"},],
},
],
},

View File

@ -1272,20 +1272,15 @@ def user_authentication_flow(conn):
)["UserPoolClient"]["ClientId"]
conn.sign_up(
ClientId=client_id,
Username=username,
Password=password,
ClientId=client_id, Username=username, Password=password,
)
client_secret = conn.describe_user_pool_client(
UserPoolId=user_pool_id,
ClientId=client_id,
UserPoolId=user_pool_id, ClientId=client_id,
)["UserPoolClient"]["ClientSecret"]
conn.confirm_sign_up(
ClientId=client_id,
Username=username,
ConfirmationCode="123456",
ClientId=client_id, Username=username, ConfirmationCode="123456",
)
# generating secret hash
@ -1323,25 +1318,18 @@ def user_authentication_flow(conn):
)
conn.verify_software_token(
AccessToken=result["AuthenticationResult"]["AccessToken"],
UserCode="123456",
AccessToken=result["AuthenticationResult"]["AccessToken"], UserCode="123456",
)
conn.set_user_mfa_preference(
AccessToken=result["AuthenticationResult"]["AccessToken"],
SoftwareTokenMfaSettings={
"Enabled": True,
"PreferredMfa": True,
},
SoftwareTokenMfaSettings={"Enabled": True, "PreferredMfa": True,},
)
result = conn.initiate_auth(
ClientId=client_id,
AuthFlow="REFRESH_TOKEN",
AuthParameters={
"SECRET_HASH": secret_hash,
"REFRESH_TOKEN": refresh_token,
},
AuthParameters={"SECRET_HASH": secret_hash, "REFRESH_TOKEN": refresh_token,},
)
result["AuthenticationResult"]["IdToken"].should_not.be.none
@ -1595,8 +1583,7 @@ def test_sign_up():
conn = boto3.client("cognito-idp", "us-west-2")
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
client_id = conn.create_user_pool_client(
UserPoolId=user_pool_id,
ClientName=str(uuid.uuid4()),
UserPoolId=user_pool_id, ClientName=str(uuid.uuid4()),
)["UserPoolClient"]["ClientId"]
username = str(uuid.uuid4())
password = str(uuid.uuid4())
@ -1612,16 +1599,12 @@ def test_confirm_sign_up():
password = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
client_id = conn.create_user_pool_client(
UserPoolId=user_pool_id,
ClientName=str(uuid.uuid4()),
GenerateSecret=True,
UserPoolId=user_pool_id, ClientName=str(uuid.uuid4()), GenerateSecret=True,
)["UserPoolClient"]["ClientId"]
conn.sign_up(ClientId=client_id, Username=username, Password=password)
conn.confirm_sign_up(
ClientId=client_id,
Username=username,
ConfirmationCode="123456",
ClientId=client_id, Username=username, ConfirmationCode="123456",
)
result = conn.admin_get_user(UserPoolId=user_pool_id, Username=username)
@ -1635,19 +1618,14 @@ def test_initiate_auth_USER_SRP_AUTH():
password = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
client_id = conn.create_user_pool_client(
UserPoolId=user_pool_id,
ClientName=str(uuid.uuid4()),
GenerateSecret=True,
UserPoolId=user_pool_id, ClientName=str(uuid.uuid4()), GenerateSecret=True,
)["UserPoolClient"]["ClientId"]
conn.sign_up(ClientId=client_id, Username=username, Password=password)
client_secret = conn.describe_user_pool_client(
UserPoolId=user_pool_id,
ClientId=client_id,
UserPoolId=user_pool_id, ClientId=client_id,
)["UserPoolClient"]["ClientSecret"]
conn.confirm_sign_up(
ClientId=client_id,
Username=username,
ConfirmationCode="123456",
ClientId=client_id, Username=username, ConfirmationCode="123456",
)
key = bytes(str(client_secret).encode("latin-1"))
@ -1691,14 +1669,11 @@ def test_initiate_auth_for_unconfirmed_user():
password = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
client_id = conn.create_user_pool_client(
UserPoolId=user_pool_id,
ClientName=str(uuid.uuid4()),
GenerateSecret=True,
UserPoolId=user_pool_id, ClientName=str(uuid.uuid4()), GenerateSecret=True,
)["UserPoolClient"]["ClientId"]
conn.sign_up(ClientId=client_id, Username=username, Password=password)
client_secret = conn.describe_user_pool_client(
UserPoolId=user_pool_id,
ClientId=client_id,
UserPoolId=user_pool_id, ClientId=client_id,
)["UserPoolClient"]["ClientSecret"]
key = bytes(str(client_secret).encode("latin-1"))
@ -1730,19 +1705,14 @@ def test_initiate_auth_with_invalid_secret_hash():
password = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
client_id = conn.create_user_pool_client(
UserPoolId=user_pool_id,
ClientName=str(uuid.uuid4()),
GenerateSecret=True,
UserPoolId=user_pool_id, ClientName=str(uuid.uuid4()), GenerateSecret=True,
)["UserPoolClient"]["ClientId"]
conn.sign_up(ClientId=client_id, Username=username, Password=password)
client_secret = conn.describe_user_pool_client(
UserPoolId=user_pool_id,
ClientId=client_id,
UserPoolId=user_pool_id, ClientId=client_id,
)["UserPoolClient"]["ClientSecret"]
conn.confirm_sign_up(
ClientId=client_id,
Username=username,
ConfirmationCode="123456",
ClientId=client_id, Username=username, ConfirmationCode="123456",
)
invalid_secret_hash = str(uuid.uuid4())

View File

@ -1845,12 +1845,7 @@ def test_put_evaluations():
response["ResponseMetadata"].pop("HTTPHeaders", None)
response["ResponseMetadata"].pop("RetryAttempts", None)
response.should.equal(
{
"FailedEvaluations": [],
"ResponseMetadata": {
"HTTPStatusCode": 200,
},
}
{"FailedEvaluations": [], "ResponseMetadata": {"HTTPStatusCode": 200,},}
)

View File

@ -325,9 +325,7 @@ def test_access_denied_for_run_instances():
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(403)
ex.value.response["Error"]["Message"].should.equal(
"User: arn:aws:iam::{account_id}:user/{user_name} is not authorized to perform: {operation}".format(
account_id=ACCOUNT_ID,
user_name=user_name,
operation="ec2:RunInstances",
account_id=ACCOUNT_ID, user_name=user_name, operation="ec2:RunInstances",
)
)

View File

@ -1347,13 +1347,9 @@ def test_get_item_returns_consumed_capacity():
def test_put_empty_item():
dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
dynamodb.create_table(
AttributeDefinitions=[
{"AttributeName": "structure_id", "AttributeType": "S"},
],
AttributeDefinitions=[{"AttributeName": "structure_id", "AttributeType": "S"},],
TableName="test",
KeySchema=[
{"AttributeName": "structure_id", "KeyType": "HASH"},
],
KeySchema=[{"AttributeName": "structure_id", "KeyType": "HASH"},],
ProvisionedThroughput={"ReadCapacityUnits": 123, "WriteCapacityUnits": 123},
)
table = dynamodb.Table("test")
@ -1370,13 +1366,9 @@ def test_put_empty_item():
def test_put_item_nonexisting_hash_key():
dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
dynamodb.create_table(
AttributeDefinitions=[
{"AttributeName": "structure_id", "AttributeType": "S"},
],
AttributeDefinitions=[{"AttributeName": "structure_id", "AttributeType": "S"},],
TableName="test",
KeySchema=[
{"AttributeName": "structure_id", "KeyType": "HASH"},
],
KeySchema=[{"AttributeName": "structure_id", "KeyType": "HASH"},],
ProvisionedThroughput={"ReadCapacityUnits": 123, "WriteCapacityUnits": 123},
)
table = dynamodb.Table("test")
@ -2295,10 +2287,7 @@ def test_update_item_on_map():
table.update_item(
Key={"forum_name": "the-key", "subject": "123"},
UpdateExpression="SET body.#nested.#data = :tb",
ExpressionAttributeNames={
"#nested": "nested",
"#data": "data",
},
ExpressionAttributeNames={"#nested": "nested", "#data": "data",},
ExpressionAttributeValues={":tb": "new_value"},
)
# Running this against AWS DDB gives an exception so make sure it also fails.:
@ -3962,30 +3951,19 @@ def test_update_supports_nested_update_if_nested_value_not_exists():
table = dynamodb.Table(name)
table.put_item(
Item={
"user_id": "1234",
"friends": {"5678": {"name": "friend_5678"}},
},
Item={"user_id": "1234", "friends": {"5678": {"name": "friend_5678"}},},
)
table.update_item(
Key={"user_id": "1234"},
ExpressionAttributeNames={
"#friends": "friends",
"#friendid": "0000",
},
ExpressionAttributeValues={
":friend": {"name": "friend_0000"},
},
ExpressionAttributeNames={"#friends": "friends", "#friendid": "0000",},
ExpressionAttributeValues={":friend": {"name": "friend_0000"},},
UpdateExpression="SET #friends.#friendid = :friend",
ReturnValues="UPDATED_NEW",
)
item = table.get_item(Key={"user_id": "1234"})["Item"]
assert item == {
"user_id": "1234",
"friends": {
"5678": {"name": "friend_5678"},
"0000": {"name": "friend_0000"},
},
"friends": {"5678": {"name": "friend_5678"}, "0000": {"name": "friend_0000"},},
}
@ -4208,17 +4186,11 @@ def test_invalid_transact_get_items():
)
table = dynamodb.Table("test1")
table.put_item(
Item={
"id": "1",
"val": "1",
}
Item={"id": "1", "val": "1",}
)
table.put_item(
Item={
"id": "1",
"val": "2",
}
Item={"id": "1", "val": "2",}
)
client = boto3.client("dynamodb", region_name="us-east-1")
@ -4240,22 +4212,8 @@ def test_invalid_transact_get_items():
with pytest.raises(ClientError) as ex:
client.transact_get_items(
TransactItems=[
{
"Get": {
"Key": {
"id": {"S": "1"},
},
"TableName": "test1",
}
},
{
"Get": {
"Key": {
"id": {"S": "1"},
},
"TableName": "non_exists_table",
}
},
{"Get": {"Key": {"id": {"S": "1"},}, "TableName": "test1",}},
{"Get": {"Key": {"id": {"S": "1"},}, "TableName": "non_exists_table",}},
]
)
@ -4281,17 +4239,11 @@ def test_valid_transact_get_items():
)
table1 = dynamodb.Table("test1")
table1.put_item(
Item={
"id": "1",
"sort_key": "1",
}
Item={"id": "1", "sort_key": "1",}
)
table1.put_item(
Item={
"id": "1",
"sort_key": "2",
}
Item={"id": "1", "sort_key": "2",}
)
dynamodb.create_table(
@ -4308,10 +4260,7 @@ def test_valid_transact_get_items():
)
table2 = dynamodb.Table("test2")
table2.put_item(
Item={
"id": "1",
"sort_key": "1",
}
Item={"id": "1", "sort_key": "1",}
)
client = boto3.client("dynamodb", region_name="us-east-1")
@ -4425,10 +4374,7 @@ def test_valid_transact_get_items():
"TableName": "test1",
"CapacityUnits": 4.0,
"ReadCapacityUnits": 4.0,
"Table": {
"CapacityUnits": 4.0,
"ReadCapacityUnits": 4.0,
},
"Table": {"CapacityUnits": 4.0, "ReadCapacityUnits": 4.0,},
}
)
@ -4437,10 +4383,7 @@ def test_valid_transact_get_items():
"TableName": "test2",
"CapacityUnits": 2.0,
"ReadCapacityUnits": 2.0,
"Table": {
"CapacityUnits": 2.0,
"ReadCapacityUnits": 2.0,
},
"Table": {"CapacityUnits": 2.0, "ReadCapacityUnits": 2.0,},
}
)
@ -4456,9 +4399,7 @@ def test_gsi_verify_negative_number_order():
{"AttributeName": "gsiK1PartitionKey", "KeyType": "HASH"},
{"AttributeName": "gsiK1SortKey", "KeyType": "RANGE"},
],
"Projection": {
"ProjectionType": "KEYS_ONLY",
},
"Projection": {"ProjectionType": "KEYS_ONLY",},
}
],
"AttributeDefinitions": [
@ -4509,9 +4450,7 @@ def test_gsi_verify_negative_number_order():
def test_transact_write_items_put():
table_schema = {
"KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
"AttributeDefinitions": [
{"AttributeName": "id", "AttributeType": "S"},
],
"AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"},],
}
dynamodb = boto3.client("dynamodb", region_name="us-east-1")
dynamodb.create_table(
@ -4522,10 +4461,7 @@ def test_transact_write_items_put():
TransactItems=[
{
"Put": {
"Item": {
"id": {"S": "foo{}".format(str(i))},
"foo": {"S": "bar"},
},
"Item": {"id": {"S": "foo{}".format(str(i))}, "foo": {"S": "bar"},},
"TableName": "test-table",
}
}
@ -4541,19 +4477,14 @@ def test_transact_write_items_put():
def test_transact_write_items_put_conditional_expressions():
table_schema = {
"KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
"AttributeDefinitions": [
{"AttributeName": "id", "AttributeType": "S"},
],
"AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"},],
}
dynamodb = boto3.client("dynamodb", region_name="us-east-1")
dynamodb.create_table(
TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
)
dynamodb.put_item(
TableName="test-table",
Item={
"id": {"S": "foo2"},
},
TableName="test-table", Item={"id": {"S": "foo2"},},
)
# Put multiple items
with pytest.raises(ClientError) as ex:
@ -4591,9 +4522,7 @@ def test_transact_write_items_put_conditional_expressions():
def test_transact_write_items_conditioncheck_passes():
table_schema = {
"KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
"AttributeDefinitions": [
{"AttributeName": "id", "AttributeType": "S"},
],
"AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"},],
}
dynamodb = boto3.client("dynamodb", region_name="us-east-1")
dynamodb.create_table(
@ -4601,10 +4530,7 @@ def test_transact_write_items_conditioncheck_passes():
)
# Insert an item without email address
dynamodb.put_item(
TableName="test-table",
Item={
"id": {"S": "foo"},
},
TableName="test-table", Item={"id": {"S": "foo"},},
)
# Put an email address, after verifying it doesn't exist yet
dynamodb.transact_write_items(
@ -4638,9 +4564,7 @@ def test_transact_write_items_conditioncheck_passes():
def test_transact_write_items_conditioncheck_fails():
table_schema = {
"KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
"AttributeDefinitions": [
{"AttributeName": "id", "AttributeType": "S"},
],
"AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"},],
}
dynamodb = boto3.client("dynamodb", region_name="us-east-1")
dynamodb.create_table(
@ -4689,9 +4613,7 @@ def test_transact_write_items_conditioncheck_fails():
def test_transact_write_items_delete():
table_schema = {
"KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
"AttributeDefinitions": [
{"AttributeName": "id", "AttributeType": "S"},
],
"AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"},],
}
dynamodb = boto3.client("dynamodb", region_name="us-east-1")
dynamodb.create_table(
@ -4699,20 +4621,12 @@ def test_transact_write_items_delete():
)
# Insert an item
dynamodb.put_item(
TableName="test-table",
Item={
"id": {"S": "foo"},
},
TableName="test-table", Item={"id": {"S": "foo"},},
)
# Delete the item
dynamodb.transact_write_items(
TransactItems=[
{
"Delete": {
"Key": {"id": {"S": "foo"}},
"TableName": "test-table",
}
}
{"Delete": {"Key": {"id": {"S": "foo"}}, "TableName": "test-table",}}
]
)
# Assert the item is deleted
@ -4724,9 +4638,7 @@ def test_transact_write_items_delete():
def test_transact_write_items_delete_with_successful_condition_expression():
table_schema = {
"KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
"AttributeDefinitions": [
{"AttributeName": "id", "AttributeType": "S"},
],
"AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"},],
}
dynamodb = boto3.client("dynamodb", region_name="us-east-1")
dynamodb.create_table(
@ -4734,19 +4646,14 @@ def test_transact_write_items_delete_with_successful_condition_expression():
)
# Insert an item without email address
dynamodb.put_item(
TableName="test-table",
Item={
"id": {"S": "foo"},
},
TableName="test-table", Item={"id": {"S": "foo"},},
)
# ConditionExpression will pass - no email address has been specified yet
dynamodb.transact_write_items(
TransactItems=[
{
"Delete": {
"Key": {
"id": {"S": "foo"},
},
"Key": {"id": {"S": "foo"},},
"TableName": "test-table",
"ConditionExpression": "attribute_not_exists(#e)",
"ExpressionAttributeNames": {"#e": "email_address"},
@ -4763,9 +4670,7 @@ def test_transact_write_items_delete_with_successful_condition_expression():
def test_transact_write_items_delete_with_failed_condition_expression():
table_schema = {
"KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
"AttributeDefinitions": [
{"AttributeName": "id", "AttributeType": "S"},
],
"AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"},],
}
dynamodb = boto3.client("dynamodb", region_name="us-east-1")
dynamodb.create_table(
@ -4783,9 +4688,7 @@ def test_transact_write_items_delete_with_failed_condition_expression():
TransactItems=[
{
"Delete": {
"Key": {
"id": {"S": "foo"},
},
"Key": {"id": {"S": "foo"},},
"TableName": "test-table",
"ConditionExpression": "attribute_not_exists(#e)",
"ExpressionAttributeNames": {"#e": "email_address"},
@ -4806,9 +4709,7 @@ def test_transact_write_items_delete_with_failed_condition_expression():
def test_transact_write_items_update():
table_schema = {
"KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
"AttributeDefinitions": [
{"AttributeName": "id", "AttributeType": "S"},
],
"AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"},],
}
dynamodb = boto3.client("dynamodb", region_name="us-east-1")
dynamodb.create_table(
@ -4840,9 +4741,7 @@ def test_transact_write_items_update():
def test_transact_write_items_update_with_failed_condition_expression():
table_schema = {
"KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
"AttributeDefinitions": [
{"AttributeName": "id", "AttributeType": "S"},
],
"AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"},],
}
dynamodb = boto3.client("dynamodb", region_name="us-east-1")
dynamodb.create_table(
@ -5032,18 +4931,12 @@ def create_simple_table_and_return_client():
dynamodb.create_table(
TableName="moto-test",
KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
AttributeDefinitions=[
{"AttributeName": "id", "AttributeType": "S"},
],
AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"},],
ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 1},
)
dynamodb.put_item(
TableName="moto-test",
Item={
"id": {"S": "1"},
"myNum": {"N": "1"},
"MyStr": {"S": "1"},
},
Item={"id": {"S": "1"}, "myNum": {"N": "1"}, "MyStr": {"S": "1"},},
)
return dynamodb
@ -5107,11 +5000,7 @@ def test_update_expression_with_plus_in_attribute_name():
dynamodb.put_item(
TableName="moto-test",
Item={
"id": {"S": "1"},
"my+Num": {"S": "1"},
"MyStr": {"S": "aaa"},
},
Item={"id": {"S": "1"}, "my+Num": {"S": "1"}, "MyStr": {"S": "aaa"},},
)
try:
dynamodb.update_item(
@ -5138,11 +5027,7 @@ def test_update_expression_with_minus_in_attribute_name():
dynamodb.put_item(
TableName="moto-test",
Item={
"id": {"S": "1"},
"my-Num": {"S": "1"},
"MyStr": {"S": "aaa"},
},
Item={"id": {"S": "1"}, "my-Num": {"S": "1"}, "MyStr": {"S": "aaa"},},
)
try:
dynamodb.update_item(
@ -5169,11 +5054,7 @@ def test_update_expression_with_space_in_attribute_name():
dynamodb.put_item(
TableName="moto-test",
Item={
"id": {"S": "1"},
"my Num": {"S": "1"},
"MyStr": {"S": "aaa"},
},
Item={"id": {"S": "1"}, "my Num": {"S": "1"}, "MyStr": {"S": "aaa"},},
)
try:
@ -5356,8 +5237,7 @@ def test_update_item_atomic_counter_from_zero():
key = {"t_id": {"S": "item1"}}
ddb_mock.put_item(
TableName=table,
Item=key,
TableName=table, Item=key,
)
ddb_mock.update_item(
@ -5383,8 +5263,7 @@ def test_update_item_add_to_non_existent_set():
)
key = {"t_id": {"S": "item1"}}
ddb_mock.put_item(
TableName=table,
Item=key,
TableName=table, Item=key,
)
ddb_mock.update_item(
@ -5409,8 +5288,7 @@ def test_update_item_add_to_non_existent_number_set():
)
key = {"t_id": {"S": "item1"}}
ddb_mock.put_item(
TableName=table,
Item=key,
TableName=table, Item=key,
)
ddb_mock.update_item(
@ -5427,9 +5305,7 @@ def test_update_item_add_to_non_existent_number_set():
def test_transact_write_items_fails_with_transaction_canceled_exception():
table_schema = {
"KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
"AttributeDefinitions": [
{"AttributeName": "id", "AttributeType": "S"},
],
"AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"},],
}
dynamodb = boto3.client("dynamodb", region_name="us-east-1")
dynamodb.create_table(
@ -5481,9 +5357,7 @@ def test_gsi_projection_type_keys_only():
{"AttributeName": "gsiK1PartitionKey", "KeyType": "HASH"},
{"AttributeName": "gsiK1SortKey", "KeyType": "RANGE"},
],
"Projection": {
"ProjectionType": "KEYS_ONLY",
},
"Projection": {"ProjectionType": "KEYS_ONLY",},
}
],
"AttributeDefinitions": [
@ -5536,9 +5410,7 @@ def test_lsi_projection_type_keys_only():
{"AttributeName": "partitionKey", "KeyType": "HASH"},
{"AttributeName": "lsiK1SortKey", "KeyType": "RANGE"},
],
"Projection": {
"ProjectionType": "KEYS_ONLY",
},
"Projection": {"ProjectionType": "KEYS_ONLY",},
}
],
"AttributeDefinitions": [
@ -5563,8 +5435,7 @@ def test_lsi_projection_type_keys_only():
table.put_item(Item=item)
items = table.query(
KeyConditionExpression=Key("partitionKey").eq("pk-1"),
IndexName="LSI",
KeyConditionExpression=Key("partitionKey").eq("pk-1"), IndexName="LSI",
)["Items"]
items.should.have.length_of(1)
# Item should only include GSI Keys and Table Keys, as per the ProjectionType

View File

@ -212,11 +212,7 @@ def test_execution_of_remove_in_map():
"itemlist": {
"L": [
{"M": {"foo00": {"S": "bar1"}, "foo01": {"S": "bar2"}}},
{
"M": {
"foo10": {"S": "bar1"},
}
},
{"M": {"foo10": {"S": "bar1"},}},
]
}
}
@ -265,9 +261,7 @@ def test_execution_of_remove_in_list():
"itemmap": {
"M": {
"itemlist": {
"L": [
{"M": {"foo00": {"S": "bar1"}, "foo01": {"S": "bar2"}}},
]
"L": [{"M": {"foo00": {"S": "bar1"}, "foo01": {"S": "bar2"}}},]
}
}
},
@ -284,10 +278,7 @@ def test_execution_of_delete_element_from_set():
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={
"id": {"S": "foo2"},
"s": {"SS": ["value1", "value2", "value3"]},
},
attrs={"id": {"S": "foo2"}, "s": {"SS": ["value1", "value2", "value3"]},},
)
validated_ast = UpdateExpressionValidator(
update_expression_ast,
@ -301,10 +292,7 @@ def test_execution_of_delete_element_from_set():
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={
"id": {"S": "foo2"},
"s": {"SS": ["value1", "value3"]},
},
attrs={"id": {"S": "foo2"}, "s": {"SS": ["value1", "value3"]},},
)
assert expected_item == item
@ -317,10 +305,7 @@ def test_execution_of_add_number():
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={
"id": {"S": "foo2"},
"s": {"N": "5"},
},
attrs={"id": {"S": "foo2"}, "s": {"N": "5"},},
)
validated_ast = UpdateExpressionValidator(
update_expression_ast,
@ -347,10 +332,7 @@ def test_execution_of_add_set_to_a_number():
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={
"id": {"S": "foo2"},
"s": {"N": "5"},
},
attrs={"id": {"S": "foo2"}, "s": {"N": "5"},},
)
try:
validated_ast = UpdateExpressionValidator(
@ -381,10 +363,7 @@ def test_execution_of_add_to_a_set():
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={
"id": {"S": "foo2"},
"s": {"SS": ["value1", "value2", "value3"]},
},
attrs={"id": {"S": "foo2"}, "s": {"SS": ["value1", "value2", "value3"]},},
)
validated_ast = UpdateExpressionValidator(
update_expression_ast,
@ -406,37 +385,17 @@ def test_execution_of_add_to_a_set():
assert expected_item == item
@pytest.mark.parametrize("expression_attribute_values,unexpected_data_type",
@pytest.mark.parametrize(
"expression_attribute_values,unexpected_data_type",
[
(
{":value": {"S": "10"}},
"STRING",
),
(
{":value": {"N": "10"}},
"NUMBER",
),
(
{":value": {"B": "10"}},
"BINARY",
),
(
{":value": {"BOOL": True}},
"BOOLEAN",
),
(
{":value": {"NULL": True}},
"NULL",
),
(
{":value": {"M": {"el0": {"S": "10"}}}},
"MAP",
),
(
{":value": {"L": []}},
"LIST",
),
]
({":value": {"S": "10"}}, "STRING",),
({":value": {"N": "10"}}, "NUMBER",),
({":value": {"B": "10"}}, "BINARY",),
({":value": {"BOOL": True}}, "BOOLEAN",),
({":value": {"NULL": True}}, "NULL",),
({":value": {"M": {"el0": {"S": "10"}}}}, "MAP",),
({":value": {"L": []}}, "LIST",),
],
)
def test_execution_of__delete_element_from_set_invalid_value(
expression_attribute_values, unexpected_data_type
@ -449,10 +408,7 @@ def test_execution_of__delete_element_from_set_invalid_value(
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={
"id": {"S": "foo2"},
"s": {"SS": ["value1", "value2", "value3"]},
},
attrs={"id": {"S": "foo2"}, "s": {"SS": ["value1", "value2", "value3"]},},
)
try:
validated_ast = UpdateExpressionValidator(
@ -477,10 +433,7 @@ def test_execution_of_delete_element_from_a_string_attribute():
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={
"id": {"S": "foo2"},
"s": {"S": "5"},
},
attrs={"id": {"S": "foo2"}, "s": {"S": "5"},},
)
try:
validated_ast = UpdateExpressionValidator(

View File

@ -41,11 +41,8 @@ def test_validation_of_update_expression_with_keyword():
assert e.keyword == "path"
@pytest.mark.parametrize("update_expression",
[
"SET a = #b + :val2",
"SET a = :val2 + #b",
]
@pytest.mark.parametrize(
"update_expression", ["SET a = #b + :val2", "SET a = :val2 + #b",]
)
def test_validation_of_a_set_statement_with_incorrect_passed_value(update_expression):
"""
@ -101,12 +98,7 @@ def test_validation_of_update_expression_with_attribute_that_does_not_exist_in_i
assert True
@pytest.mark.parametrize("update_expression",
[
"SET a = #c",
"SET a = #c + #d",
]
)
@pytest.mark.parametrize("update_expression", ["SET a = #c", "SET a = #c + #d",])
def test_validation_of_update_expression_with_attribute_name_that_is_not_defined(
update_expression,
):

View File

@ -2,6 +2,7 @@ from __future__ import unicode_literals
import boto
import boto3
# Ensure 'pytest.raises' context manager support for Python 2.6
import pytest
import sure # noqa

View File

@ -144,9 +144,7 @@ def test_create_flow_log_create():
bucket = s3.create_bucket(
Bucket="test-flow-logs",
CreateBucketConfiguration={
"LocationConstraint": "us-west-1",
},
CreateBucketConfiguration={"LocationConstraint": "us-west-1",},
)
response = client.create_flow_logs(

View File

@ -211,9 +211,7 @@ def test_instance_detach_volume_wrong_path():
ImageId="ami-d3adb33f",
MinCount=1,
MaxCount=1,
BlockDeviceMappings=[
{"DeviceName": "/dev/sda1", "Ebs": {"VolumeSize": 50}},
],
BlockDeviceMappings=[{"DeviceName": "/dev/sda1", "Ebs": {"VolumeSize": 50}},],
)
instance = result[0]
for volume in instance.volumes.all():
@ -1585,9 +1583,7 @@ def test_create_instance_ebs_optimized():
instance.ebs_optimized.should.be(False)
instance = ec2_resource.create_instances(
ImageId="ami-12345678",
MaxCount=1,
MinCount=1,
ImageId="ami-12345678", MaxCount=1, MinCount=1,
)[0]
instance.load()
instance.ebs_optimized.should.be(False)

View File

@ -661,11 +661,7 @@ def test_run_instances_should_attach_to_default_subnet():
client = boto3.client("ec2", region_name="us-west-1")
ec2.create_security_group(GroupName="sg01", Description="Test security group sg01")
# run_instances
instances = client.run_instances(
MinCount=1,
MaxCount=1,
SecurityGroups=["sg01"],
)
instances = client.run_instances(MinCount=1, MaxCount=1, SecurityGroups=["sg01"],)
# Assert subnet is created appropriately
subnets = client.describe_subnets()["Subnets"]
default_subnet_id = subnets[0]["SubnetId"]

View File

@ -60,9 +60,7 @@ def test_create_vpn_connection_with_vpn_gateway():
vpn_gateway = client.create_vpn_gateway(Type="ipsec.1").get("VpnGateway", {})
customer_gateway = client.create_customer_gateway(
Type="ipsec.1",
PublicIp="205.251.242.54",
BgpAsn=65534,
Type="ipsec.1", PublicIp="205.251.242.54", BgpAsn=65534,
).get("CustomerGateway", {})
vpn_connection = client.create_vpn_connection(
Type="ipsec.1",

View File

@ -2531,9 +2531,7 @@ def test_describe_task_sets():
assert "tags" not in task_sets[0]
task_sets = client.describe_task_sets(
cluster=cluster_name,
service=service_name,
include=["TAGS"],
cluster=cluster_name, service=service_name, include=["TAGS"],
)["taskSets"]
cluster_arn = client.describe_clusters(clusters=[cluster_name])["clusters"][0][
@ -2593,39 +2591,29 @@ def test_delete_task_set():
)
task_set = client.create_task_set(
cluster=cluster_name,
service=service_name,
taskDefinition=task_def_name,
cluster=cluster_name, service=service_name, taskDefinition=task_def_name,
)["taskSet"]
task_sets = client.describe_task_sets(
cluster=cluster_name,
service=service_name,
taskSets=[task_set["taskSetArn"]],
cluster=cluster_name, service=service_name, taskSets=[task_set["taskSetArn"]],
)["taskSets"]
assert len(task_sets) == 1
response = client.delete_task_set(
cluster=cluster_name,
service=service_name,
taskSet=task_set["taskSetArn"],
cluster=cluster_name, service=service_name, taskSet=task_set["taskSetArn"],
)
assert response["taskSet"]["taskSetArn"] == task_set["taskSetArn"]
task_sets = client.describe_task_sets(
cluster=cluster_name,
service=service_name,
taskSets=[task_set["taskSetArn"]],
cluster=cluster_name, service=service_name, taskSets=[task_set["taskSetArn"]],
)["taskSets"]
assert len(task_sets) == 0
with pytest.raises(ClientError):
_ = client.delete_task_set(
cluster=cluster_name,
service=service_name,
taskSet=task_set["taskSetArn"],
cluster=cluster_name, service=service_name, taskSet=task_set["taskSetArn"],
)
@ -2661,9 +2649,7 @@ def test_update_service_primary_task_set():
)
task_set = client.create_task_set(
cluster=cluster_name,
service=service_name,
taskDefinition=task_def_name,
cluster=cluster_name, service=service_name, taskDefinition=task_def_name,
)["taskSet"]
service = client.describe_services(cluster=cluster_name, services=[service_name],)[
@ -2683,9 +2669,7 @@ def test_update_service_primary_task_set():
assert service["taskDefinition"] == service["taskSets"][0]["taskDefinition"]
another_task_set = client.create_task_set(
cluster=cluster_name,
service=service_name,
taskDefinition=task_def_name,
cluster=cluster_name, service=service_name, taskDefinition=task_def_name,
)["taskSet"]
service = client.describe_services(cluster=cluster_name, services=[service_name],)[
"services"
@ -2737,15 +2721,11 @@ def test_update_task_set():
)
task_set = client.create_task_set(
cluster=cluster_name,
service=service_name,
taskDefinition=task_def_name,
cluster=cluster_name, service=service_name, taskDefinition=task_def_name,
)["taskSet"]
another_task_set = client.create_task_set(
cluster=cluster_name,
service=service_name,
taskDefinition=task_def_name,
cluster=cluster_name, service=service_name, taskDefinition=task_def_name,
)["taskSet"]
assert another_task_set["scale"]["unit"] == "PERCENT"
assert another_task_set["scale"]["value"] == 100.0
@ -2758,9 +2738,7 @@ def test_update_task_set():
)
updated_task_set = client.describe_task_sets(
cluster=cluster_name,
service=service_name,
taskSets=[task_set["taskSetArn"]],
cluster=cluster_name, service=service_name, taskSets=[task_set["taskSetArn"]],
)["taskSets"][0]
assert updated_task_set["scale"]["value"] == 25.0
assert updated_task_set["scale"]["unit"] == "PERCENT"
@ -2806,13 +2784,11 @@ def test_list_tasks_with_filters():
}
_ = ecs.register_task_definition(
family="test_task_def_1",
containerDefinitions=[test_container_def],
family="test_task_def_1", containerDefinitions=[test_container_def],
)
_ = ecs.register_task_definition(
family="test_task_def_2",
containerDefinitions=[test_container_def],
family="test_task_def_2", containerDefinitions=[test_container_def],
)
_ = ecs.start_task(

View File

@ -9,30 +9,24 @@ from moto import mock_elasticbeanstalk
def test_create_application():
# Create Elastic Beanstalk Application
conn = boto3.client("elasticbeanstalk", region_name="us-east-1")
app = conn.create_application(
ApplicationName="myapp",
)
app = conn.create_application(ApplicationName="myapp",)
app["Application"]["ApplicationName"].should.equal("myapp")
@mock_elasticbeanstalk
def test_create_application_dup():
conn = boto3.client("elasticbeanstalk", region_name="us-east-1")
conn.create_application(
ApplicationName="myapp",
conn.create_application(ApplicationName="myapp",)
conn.create_application.when.called_with(ApplicationName="myapp",).should.throw(
ClientError
)
conn.create_application.when.called_with(
ApplicationName="myapp",
).should.throw(ClientError)
@mock_elasticbeanstalk
def test_describe_applications():
# Create Elastic Beanstalk Application
conn = boto3.client("elasticbeanstalk", region_name="us-east-1")
conn.create_application(
ApplicationName="myapp",
)
conn.create_application(ApplicationName="myapp",)
apps = conn.describe_applications()
len(apps["Applications"]).should.equal(1)
@ -43,13 +37,8 @@ def test_describe_applications():
def test_create_environment():
# Create Elastic Beanstalk Environment
conn = boto3.client("elasticbeanstalk", region_name="us-east-1")
app = conn.create_application(
ApplicationName="myapp",
)
env = conn.create_environment(
ApplicationName="myapp",
EnvironmentName="myenv",
)
app = conn.create_application(ApplicationName="myapp",)
env = conn.create_environment(ApplicationName="myapp", EnvironmentName="myenv",)
env["EnvironmentName"].should.equal("myenv")
@ -57,12 +46,9 @@ def test_create_environment():
def test_describe_environments():
# List Elastic Beanstalk Envs
conn = boto3.client("elasticbeanstalk", region_name="us-east-1")
conn.create_application(
ApplicationName="myapp",
)
conn.create_application(ApplicationName="myapp",)
conn.create_environment(
ApplicationName="myapp",
EnvironmentName="myenv",
ApplicationName="myapp", EnvironmentName="myenv",
)
envs = conn.describe_environments()
@ -89,9 +75,7 @@ def tags_list_to_dict(tag_list):
@mock_elasticbeanstalk
def test_create_environment_tags():
conn = boto3.client("elasticbeanstalk", region_name="us-east-1")
conn.create_application(
ApplicationName="myapp",
)
conn.create_application(ApplicationName="myapp",)
env_tags = {"initial key": "initial value"}
env = conn.create_environment(
ApplicationName="myapp",
@ -99,9 +83,7 @@ def test_create_environment_tags():
Tags=tags_dict_to_list(env_tags),
)
tags = conn.list_tags_for_resource(
ResourceArn=env["EnvironmentArn"],
)
tags = conn.list_tags_for_resource(ResourceArn=env["EnvironmentArn"],)
tags["ResourceArn"].should.equal(env["EnvironmentArn"])
tags_list_to_dict(tags["ResourceTags"]).should.equal(env_tags)
@ -109,9 +91,7 @@ def test_create_environment_tags():
@mock_elasticbeanstalk
def test_update_tags():
conn = boto3.client("elasticbeanstalk", region_name="us-east-1")
conn.create_application(
ApplicationName="myapp",
)
conn.create_application(ApplicationName="myapp",)
env_tags = {
"initial key": "initial value",
"to remove": "delete me",
@ -137,9 +117,7 @@ def test_update_tags():
total_env_tags.update(extra_env_tags)
del total_env_tags["to remove"]
tags = conn.list_tags_for_resource(
ResourceArn=env["EnvironmentArn"],
)
tags = conn.list_tags_for_resource(ResourceArn=env["EnvironmentArn"],)
tags["ResourceArn"].should.equal(env["EnvironmentArn"])
tags_list_to_dict(tags["ResourceTags"]).should.equal(total_env_tags)

View File

@ -525,10 +525,8 @@ def test_run_job_flow_with_instance_groups_with_autoscaling():
if "AutoScalingPolicy" in y:
x["AutoScalingPolicy"]["Status"]["State"].should.equal("ATTACHED")
returned_policy = deepcopy(x["AutoScalingPolicy"])
auto_scaling_policy_with_cluster_id = (
_patch_cluster_id_placeholder_in_autoscaling_policy(
y["AutoScalingPolicy"], cluster_id
)
auto_scaling_policy_with_cluster_id = _patch_cluster_id_placeholder_in_autoscaling_policy(
y["AutoScalingPolicy"], cluster_id
)
del returned_policy["Status"]
returned_policy.should.equal(auto_scaling_policy_with_cluster_id)
@ -554,10 +552,8 @@ def test_put_remove_auto_scaling_policy():
AutoScalingPolicy=auto_scaling_policy,
)
auto_scaling_policy_with_cluster_id = (
_patch_cluster_id_placeholder_in_autoscaling_policy(
auto_scaling_policy, cluster_id
)
auto_scaling_policy_with_cluster_id = _patch_cluster_id_placeholder_in_autoscaling_policy(
auto_scaling_policy, cluster_id
)
del resp["AutoScalingPolicy"]["Status"]
resp["AutoScalingPolicy"].should.equal(auto_scaling_policy_with_cluster_id)
@ -804,7 +800,11 @@ def test_instance_groups():
x["AutoScalingPolicy"]["Status"]["State"].should.equal("ATTACHED")
returned_policy = dict(x["AutoScalingPolicy"])
del returned_policy["Status"]
policy = json.loads(json.dumps(y["AutoScalingPolicy"]).replace("${emr.clusterId}", cluster_id))
policy = json.loads(
json.dumps(y["AutoScalingPolicy"]).replace(
"${emr.clusterId}", cluster_id
)
)
returned_policy.should.equal(policy)
if "EbsConfiguration" in y:
_do_assertion_ebs_configuration(x, y)

View File

@ -71,9 +71,7 @@ def test_forecast_dataset_group_create_duplicate_fails():
with pytest.raises(ClientError) as exc:
client.create_dataset_group(DatasetGroupName="name", Domain="RETAIL")
exc.value.response["Error"]["Code"].should.equal(
"ResourceAlreadyExistsException"
)
exc.value.response["Error"]["Code"].should.equal("ResourceAlreadyExistsException")
@mock_forecast

View File

@ -207,9 +207,7 @@ def test_remove_role_from_instance_profile():
def test_delete_instance_profile():
conn = boto3.client("iam", region_name="us-east-1")
conn.create_role(
RoleName="my-role",
AssumeRolePolicyDocument="some policy",
Path="/my-path/",
RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/",
)
conn.create_instance_profile(InstanceProfileName="my-profile")
conn.add_role_to_instance_profile(
@ -259,9 +257,7 @@ def test_delete_role():
# Test deletion failure with a managed policy
conn.create_role(
RoleName="my-role",
AssumeRolePolicyDocument="some policy",
Path="/my-path/",
RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/",
)
response = conn.create_policy(
PolicyName="my-managed-policy", PolicyDocument=MOCK_POLICY
@ -277,14 +273,10 @@ def test_delete_role():
# Test deletion failure with an inline policy
conn.create_role(
RoleName="my-role",
AssumeRolePolicyDocument="some policy",
Path="/my-path/",
RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/",
)
conn.put_role_policy(
RoleName="my-role",
PolicyName="my-role-policy",
PolicyDocument=MOCK_POLICY,
RoleName="my-role", PolicyName="my-role-policy", PolicyDocument=MOCK_POLICY,
)
with pytest.raises(conn.exceptions.DeleteConflictException):
conn.delete_role(RoleName="my-role")
@ -295,9 +287,7 @@ def test_delete_role():
# Test deletion failure with attachment to an instance profile
conn.create_role(
RoleName="my-role",
AssumeRolePolicyDocument="some policy",
Path="/my-path/",
RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/",
)
conn.create_instance_profile(InstanceProfileName="my-profile")
conn.add_role_to_instance_profile(
@ -314,9 +304,7 @@ def test_delete_role():
# Test deletion with no conflicts
conn.create_role(
RoleName="my-role",
AssumeRolePolicyDocument="some policy",
Path="/my-path/",
RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/",
)
conn.delete_role(RoleName="my-role")
with pytest.raises(conn.exceptions.NoSuchEntityException):
@ -343,9 +331,7 @@ def test_list_instance_profiles_for_role():
conn = boto.connect_iam()
conn.create_role(
role_name="my-role",
assume_role_policy_document="some policy",
path="my-path",
role_name="my-role", assume_role_policy_document="some policy", path="my-path",
)
conn.create_role(
role_name="my-role2",
@ -357,8 +343,7 @@ def test_list_instance_profiles_for_role():
profile_path_list = ["my-path", "my-path2"]
for profile_count in range(0, 2):
conn.create_instance_profile(
profile_name_list[profile_count],
path=profile_path_list[profile_count],
profile_name_list[profile_count], path=profile_path_list[profile_count],
)
for profile_count in range(0, 2):
@ -424,9 +409,7 @@ def test_put_role_policy():
def test_get_role_policy():
conn = boto3.client("iam", region_name="us-east-1")
conn.create_role(
RoleName="my-role",
AssumeRolePolicyDocument="some policy",
Path="my-path",
RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="my-path",
)
with pytest.raises(conn.exceptions.NoSuchEntityException):
conn.get_role_policy(RoleName="my-role", PolicyName="does-not-exist")
@ -1039,8 +1022,7 @@ def test_create_virtual_mfa_device_errors():
client.create_virtual_mfa_device.when.called_with(
VirtualMFADeviceName="test-device"
).should.throw(
ClientError,
"MFADevice entity at the same path and name already exists.",
ClientError, "MFADevice entity at the same path and name already exists.",
)
client.create_virtual_mfa_device.when.called_with(
@ -1229,9 +1211,7 @@ def test_delete_user():
# Test deletion failure with an inline policy
conn.create_user(UserName="my-user")
conn.put_user_policy(
UserName="my-user",
PolicyName="my-user-policy",
PolicyDocument=MOCK_POLICY,
UserName="my-user", PolicyName="my-user-policy", PolicyDocument=MOCK_POLICY,
)
with pytest.raises(conn.exceptions.DeleteConflictException):
conn.delete_user(UserName="my-user")
@ -1416,9 +1396,7 @@ def test_managed_policy():
role_name = "my-role"
conn.create_role(
role_name,
assume_role_policy_document={"policy": "test"},
path="my-path",
role_name, assume_role_policy_document={"policy": "test"}, path="my-path",
)
for policy_name in [
"AmazonElasticMapReduceRole",
@ -1445,8 +1423,7 @@ def test_managed_policy():
].should.have.length_of(2)
conn.detach_role_policy(
"arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceRole",
role_name,
"arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceRole", role_name,
)
rows = conn.list_policies(only_attached=True)["list_policies_response"][
"list_policies_result"
@ -1586,9 +1563,7 @@ def test_get_ssh_public_key():
with pytest.raises(ClientError):
client.get_ssh_public_key(
UserName=username,
SSHPublicKeyId="xxnon-existent-keyxx",
Encoding="SSH",
UserName=username, SSHPublicKeyId="xxnon-existent-keyxx", Encoding="SSH",
)
resp = client.upload_ssh_public_key(UserName=username, SSHPublicKeyBody=public_key)
@ -1629,9 +1604,7 @@ def test_update_ssh_public_key():
with pytest.raises(ClientError):
client.update_ssh_public_key(
UserName=username,
SSHPublicKeyId="xxnon-existent-keyxx",
Status="Inactive",
UserName=username, SSHPublicKeyId="xxnon-existent-keyxx", Status="Inactive",
)
resp = client.upload_ssh_public_key(UserName=username, SSHPublicKeyBody=public_key)
@ -1709,9 +1682,7 @@ def test_get_account_authorization_details():
UserName="testUser", PolicyName="testPolicy", PolicyDocument=test_policy
)
conn.put_group_policy(
GroupName="testGroup",
PolicyName="testPolicy",
PolicyDocument=test_policy,
GroupName="testGroup", PolicyName="testPolicy", PolicyDocument=test_policy,
)
conn.attach_user_policy(
@ -2011,9 +1982,7 @@ def test_create_role_with_tags():
map(lambda x: {"Key": str(x), "Value": str(x)}, range(0, 51))
)
conn.create_role(
RoleName="my-role3",
AssumeRolePolicyDocument="{}",
Tags=too_many_tags,
RoleName="my-role3", AssumeRolePolicyDocument="{}", Tags=too_many_tags,
)
assert (
"failed to satisfy constraint: Member must have length less than or equal to 50."
@ -2279,9 +2248,7 @@ def test_update_role_description():
conn.delete_role(RoleName="my-role")
conn.create_role(
RoleName="my-role",
AssumeRolePolicyDocument="some policy",
Path="/my-path/",
RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/",
)
response = conn.update_role_description(RoleName="my-role", Description="test")
@ -2296,9 +2263,7 @@ def test_update_role():
conn.delete_role(RoleName="my-role")
conn.create_role(
RoleName="my-role",
AssumeRolePolicyDocument="some policy",
Path="/my-path/",
RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/",
)
response = conn.update_role_description(RoleName="my-role", Description="test")
assert response["Role"]["RoleName"] == "my-role"
@ -2312,9 +2277,7 @@ def test_update_role():
conn.delete_role(RoleName="my-role")
conn.create_role(
RoleName="my-role",
AssumeRolePolicyDocument="some policy",
Path="/my-path/",
RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/",
)
response = conn.update_role(RoleName="my-role", Description="test")
assert len(response.keys()) == 1
@ -2355,9 +2318,7 @@ def test_list_entities_for_policy():
conn = boto3.client("iam", region_name="us-east-1")
conn.create_role(
RoleName="my-role",
AssumeRolePolicyDocument="some policy",
Path="/my-path/",
RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/",
)
conn.create_user(Path="/", UserName="testUser")
conn.create_group(Path="/", GroupName="testGroup")
@ -2373,9 +2334,7 @@ def test_list_entities_for_policy():
UserName="testUser", PolicyName="testPolicy", PolicyDocument=test_policy
)
conn.put_group_policy(
GroupName="testGroup",
PolicyName="testPolicy",
PolicyDocument=test_policy,
GroupName="testGroup", PolicyName="testPolicy", PolicyDocument=test_policy,
)
conn.attach_user_policy(
@ -2438,9 +2397,7 @@ def test_list_entities_for_policy():
def test_create_role_no_path():
conn = boto3.client("iam", region_name="us-east-1")
resp = conn.create_role(
RoleName="my-role",
AssumeRolePolicyDocument="some policy",
Description="test",
RoleName="my-role", AssumeRolePolicyDocument="some policy", Description="test",
)
resp.get("Role").get("Arn").should.equal(
"arn:aws:iam::{}:role/my-role".format(ACCOUNT_ID)
@ -2496,9 +2453,7 @@ def test_create_role_with_same_name_should_fail():
iam = boto3.client("iam", region_name="us-east-1")
test_role_name = str(uuid4())
iam.create_role(
RoleName=test_role_name,
AssumeRolePolicyDocument="policy",
Description="test",
RoleName=test_role_name, AssumeRolePolicyDocument="policy", Description="test",
)
# Create the role again, and verify that it fails
with pytest.raises(ClientError) as err:
@ -2585,24 +2540,14 @@ def test_create_open_id_connect_provider_errors():
client.create_open_id_connect_provider.when.called_with(
Url="http://example.org",
ThumbprintList=[
"a" * 40,
"b" * 40,
"c" * 40,
"d" * 40,
"e" * 40,
"f" * 40,
],
ThumbprintList=["a" * 40, "b" * 40, "c" * 40, "d" * 40, "e" * 40, "f" * 40,],
).should.throw(ClientError, "Thumbprint list must contain fewer than 5 entries.")
too_many_client_ids = ["{}".format(i) for i in range(101)]
client.create_open_id_connect_provider.when.called_with(
Url="http://example.org",
ThumbprintList=[],
ClientIDList=too_many_client_ids,
Url="http://example.org", ThumbprintList=[], ClientIDList=too_many_client_ids,
).should.throw(
ClientError,
"Cannot exceed quota for ClientIdsPerOpenIdConnectProvider: 100",
ClientError, "Cannot exceed quota for ClientIdsPerOpenIdConnectProvider: 100",
)
too_long_url = "b" * 256
@ -2643,8 +2588,7 @@ def test_delete_open_id_connect_provider():
client.get_open_id_connect_provider.when.called_with(
OpenIDConnectProviderArn=open_id_arn
).should.throw(
ClientError,
"OpenIDConnect Provider not found for arn {}".format(open_id_arn),
ClientError, "OpenIDConnect Provider not found for arn {}".format(open_id_arn),
)
# deleting a non existing provider should be successful
@ -2736,9 +2680,7 @@ def test_update_account_password_policy_errors():
client = boto3.client("iam", region_name="us-east-1")
client.update_account_password_policy.when.called_with(
MaxPasswordAge=1096,
MinimumPasswordLength=129,
PasswordReusePrevention=25,
MaxPasswordAge=1096, MinimumPasswordLength=129, PasswordReusePrevention=25,
).should.throw(
ClientError,
"3 validation errors detected: "
@ -2816,8 +2758,7 @@ def test_delete_account_password_policy_errors():
client = boto3.client("iam", region_name="us-east-1")
client.delete_account_password_policy.when.called_with().should.throw(
ClientError,
"The account policy with name PasswordPolicy cannot be found.",
ClientError, "The account policy with name PasswordPolicy cannot be found.",
)
@ -2945,8 +2886,7 @@ def test_list_user_tags():
conn = boto3.client("iam", region_name="us-east-1")
conn.create_user(UserName="kenny-bania")
conn.create_user(
UserName="jackie-chiles",
Tags=[{"Key": "Sue-Allen", "Value": "Oh-Henry"}],
UserName="jackie-chiles", Tags=[{"Key": "Sue-Allen", "Value": "Oh-Henry"}],
)
conn.create_user(
UserName="cosmo",
@ -2965,10 +2905,7 @@ def test_list_user_tags():
response = conn.list_user_tags(UserName="cosmo")
response["Tags"].should.equal(
[
{"Key": "Stan", "Value": "The Caddy"},
{"Key": "like-a", "Value": "glove"},
]
[{"Key": "Stan", "Value": "The Caddy"}, {"Key": "like-a", "Value": "glove"},]
)
response["IsTruncated"].should_not.be.ok
@ -3011,8 +2948,7 @@ def test_delete_account_password_policy_errors():
client = boto3.client("iam", region_name="us-east-1")
client.delete_account_password_policy.when.called_with().should.throw(
ClientError,
"The account policy with name PasswordPolicy cannot be found.",
ClientError, "The account policy with name PasswordPolicy cannot be found.",
)
@ -3041,10 +2977,7 @@ def test_role_list_config_discovered_resources():
max_session_duration=3600,
)
roles.append(
{
"id": this_role.id,
"name": this_role.name,
}
{"id": this_role.id, "name": this_role.name,}
)
assert len(roles) == num_roles
@ -3102,11 +3035,7 @@ def test_role_config_dict():
basic_assume_role = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": "*"},
"Action": "sts:AssumeRole",
}
{"Effect": "Allow", "Principal": {"AWS": "*"}, "Action": "sts:AssumeRole",}
],
}
@ -3423,9 +3352,7 @@ def test_role_config_client():
# Test non-aggregated pagination
assert (
config_client.list_discovered_resources(
resourceType="AWS::IAM::Role",
limit=1,
nextToken=result["nextToken"],
resourceType="AWS::IAM::Role", limit=1, nextToken=result["nextToken"],
)["resourceIdentifiers"][0]["resourceId"]
) != first_result
@ -3461,18 +3388,14 @@ def test_role_config_client():
# Test non-aggregated resource name/id filter
assert (
config_client.list_discovered_resources(
resourceType="AWS::IAM::Role",
resourceName=roles[1]["name"],
limit=1,
resourceType="AWS::IAM::Role", resourceName=roles[1]["name"], limit=1,
)["resourceIdentifiers"][0]["resourceName"]
== roles[1]["name"]
)
assert (
config_client.list_discovered_resources(
resourceType="AWS::IAM::Role",
resourceIds=[roles[0]["id"]],
limit=1,
resourceType="AWS::IAM::Role", resourceIds=[roles[0]["id"]], limit=1,
)["resourceIdentifiers"][0]["resourceName"]
== roles[0]["name"]
)
@ -3518,17 +3441,13 @@ def test_role_config_client():
# Test non-aggregated resource name/id filter
assert (
config_client.list_discovered_resources(
resourceType="AWS::IAM::Role",
resourceName=roles[1]["name"],
limit=1,
resourceType="AWS::IAM::Role", resourceName=roles[1]["name"], limit=1,
)["resourceIdentifiers"][0]["resourceName"]
== roles[1]["name"]
)
assert (
config_client.list_discovered_resources(
resourceType="AWS::IAM::Role",
resourceIds=[roles[0]["id"]],
limit=1,
resourceType="AWS::IAM::Role", resourceIds=[roles[0]["id"]], limit=1,
)["resourceIdentifiers"][0]["resourceName"]
== roles[0]["name"]
)
@ -3638,10 +3557,7 @@ def test_policy_list_config_discovered_resources():
policy_name="policy{}".format(ix),
)
policies.append(
{
"id": this_policy.id,
"name": this_policy.name,
}
{"id": this_policy.id, "name": this_policy.name,}
)
assert len(policies) == num_policies
@ -3866,9 +3782,7 @@ def test_policy_config_client():
# Test non-aggregated pagination
assert (
config_client.list_discovered_resources(
resourceType="AWS::IAM::Policy",
limit=1,
nextToken=result["nextToken"],
resourceType="AWS::IAM::Policy", limit=1, nextToken=result["nextToken"],
)["resourceIdentifiers"][0]["resourceId"]
) != first_result
@ -3905,18 +3819,14 @@ def test_policy_config_client():
# Test non-aggregated resource name/id filter
assert (
config_client.list_discovered_resources(
resourceType="AWS::IAM::Policy",
resourceName=policies[1]["name"],
limit=1,
resourceType="AWS::IAM::Policy", resourceName=policies[1]["name"], limit=1,
)["resourceIdentifiers"][0]["resourceName"]
== policies[1]["name"]
)
assert (
config_client.list_discovered_resources(
resourceType="AWS::IAM::Policy",
resourceIds=[policies[0]["id"]],
limit=1,
resourceType="AWS::IAM::Policy", resourceIds=[policies[0]["id"]], limit=1,
)["resourceIdentifiers"][0]["resourceName"]
== policies[0]["name"]
)
@ -3997,10 +3907,7 @@ def test_policy_config_client():
assert (
config_client.batch_get_resource_config(
resourceKeys=[
{
"resourceType": "AWS::IAM::Policy",
"resourceId": policies[7]["id"],
}
{"resourceType": "AWS::IAM::Policy", "resourceId": policies[7]["id"],}
]
)["baseConfigurationItems"][0]["resourceName"]
== policies[7]["name"]

View File

@ -1017,9 +1017,7 @@ def test_delete_thing_group():
group_name_1a = "my-group-name-1a"
group_name_2a = "my-group-name-2a"
tree_dict = {
group_name_1a: {
group_name_2a: {},
},
group_name_1a: {group_name_2a: {},},
}
group_catalog = generate_thing_group_tree(client, tree_dict)

View File

@ -24,9 +24,7 @@ def test_get_hls_streaming_session_url():
region_name=region_name,
endpoint_url=data_endpoint,
)
res = client.get_hls_streaming_session_url(
StreamName=stream_name,
)
res = client.get_hls_streaming_session_url(StreamName=stream_name,)
reg_exp = "^{}/hls/v1/getHLSMasterPlaylist.m3u8\?SessionToken\=.+$".format(
data_endpoint
)
@ -50,9 +48,7 @@ def test_get_dash_streaming_session_url():
region_name=region_name,
endpoint_url=data_endpoint,
)
res = client.get_dash_streaming_session_url(
StreamName=stream_name,
)
res = client.get_dash_streaming_session_url(StreamName=stream_name,)
reg_exp = "^{}/dash/v1/getDASHManifest.mpd\?SessionToken\=.+$".format(data_endpoint)
res.should.have.key("DASHStreamingSessionURL").which.should.match(reg_exp)

View File

@ -15,7 +15,11 @@ from moto.kms.models import KmsBackend
from moto.kms.exceptions import NotFoundException as MotoNotFoundException
from moto import mock_kms_deprecated, mock_kms
PLAINTEXT_VECTORS = [b"some encodeable plaintext", b"some unencodeable plaintext \xec\x8a\xcf\xb6r\xe9\xb5\xeb\xff\xa23\x16", "some unicode characters ø˚∆øˆˆ∆ßçøˆˆçßøˆ¨¥"]
PLAINTEXT_VECTORS = [
b"some encodeable plaintext",
b"some unencodeable plaintext \xec\x8a\xcf\xb6r\xe9\xb5\xeb\xff\xa23\x16",
"some unicode characters ø˚∆øˆˆ∆ßçøˆˆçßøˆ¨¥",
]
def _get_encoded_value(plaintext):
@ -570,10 +574,8 @@ def test__delete_alias__raises_if_alias_is_not_found():
with pytest.raises(NotFoundException) as err:
kms.delete_alias(alias_name)
expected_message_match = (
r"Alias arn:aws:kms:{region}:[0-9]{{12}}:{alias_name} is not found.".format(
region=region, alias_name=alias_name
)
expected_message_match = r"Alias arn:aws:kms:{region}:[0-9]{{12}}:{alias_name} is not found.".format(
region=region, alias_name=alias_name
)
ex = err.value
ex.body["__type"].should.equal("NotFoundException")

View File

@ -14,9 +14,11 @@ import pytest
from moto import mock_kms
PLAINTEXT_VECTORS = [b"some encodeable plaintext",
b"some unencodeable plaintext \xec\x8a\xcf\xb6r\xe9\xb5\xeb\xff\xa23\x16",
"some unicode characters ø˚∆øˆˆ∆ßçøˆˆçßøˆ¨¥"]
PLAINTEXT_VECTORS = [
b"some encodeable plaintext",
b"some unencodeable plaintext \xec\x8a\xcf\xb6r\xe9\xb5\xeb\xff\xa23\x16",
"some unicode characters ø˚∆øˆˆ∆ßçøˆˆçßøˆ¨¥",
]
def _get_encoded_value(plaintext):
@ -52,20 +54,14 @@ def test_create_key():
key["KeyMetadata"]["Origin"].should.equal("AWS_KMS")
key["KeyMetadata"].should_not.have.key("SigningAlgorithms")
key = conn.create_key(
KeyUsage="ENCRYPT_DECRYPT",
CustomerMasterKeySpec="RSA_2048",
)
key = conn.create_key(KeyUsage="ENCRYPT_DECRYPT", CustomerMasterKeySpec="RSA_2048",)
sorted(key["KeyMetadata"]["EncryptionAlgorithms"]).should.equal(
["RSAES_OAEP_SHA_1", "RSAES_OAEP_SHA_256"]
)
key["KeyMetadata"].should_not.have.key("SigningAlgorithms")
key = conn.create_key(
KeyUsage="SIGN_VERIFY",
CustomerMasterKeySpec="RSA_2048",
)
key = conn.create_key(KeyUsage="SIGN_VERIFY", CustomerMasterKeySpec="RSA_2048",)
key["KeyMetadata"].should_not.have.key("EncryptionAlgorithms")
sorted(key["KeyMetadata"]["SigningAlgorithms"]).should.equal(
@ -80,24 +76,21 @@ def test_create_key():
)
key = conn.create_key(
KeyUsage="SIGN_VERIFY",
CustomerMasterKeySpec="ECC_SECG_P256K1",
KeyUsage="SIGN_VERIFY", CustomerMasterKeySpec="ECC_SECG_P256K1",
)
key["KeyMetadata"].should_not.have.key("EncryptionAlgorithms")
key["KeyMetadata"]["SigningAlgorithms"].should.equal(["ECDSA_SHA_256"])
key = conn.create_key(
KeyUsage="SIGN_VERIFY",
CustomerMasterKeySpec="ECC_NIST_P384",
KeyUsage="SIGN_VERIFY", CustomerMasterKeySpec="ECC_NIST_P384",
)
key["KeyMetadata"].should_not.have.key("EncryptionAlgorithms")
key["KeyMetadata"]["SigningAlgorithms"].should.equal(["ECDSA_SHA_384"])
key = conn.create_key(
KeyUsage="SIGN_VERIFY",
CustomerMasterKeySpec="ECC_NIST_P521",
KeyUsage="SIGN_VERIFY", CustomerMasterKeySpec="ECC_NIST_P521",
)
key["KeyMetadata"].should_not.have.key("EncryptionAlgorithms")
@ -107,10 +100,7 @@ def test_create_key():
@mock_kms
def test_describe_key():
client = boto3.client("kms", region_name="us-east-1")
response = client.create_key(
Description="my key",
KeyUsage="ENCRYPT_DECRYPT",
)
response = client.create_key(Description="my key", KeyUsage="ENCRYPT_DECRYPT",)
key_id = response["KeyMetadata"]["KeyId"]
response = client.describe_key(KeyId=key_id)
@ -129,7 +119,14 @@ def test_describe_key():
response["KeyMetadata"].should_not.have.key("SigningAlgorithms")
@pytest.mark.parametrize("key_id", ["alias/does-not-exist", "arn:aws:kms:us-east-1:012345678912:alias/does-not-exist", "invalid"])
@pytest.mark.parametrize(
"key_id",
[
"alias/does-not-exist",
"arn:aws:kms:us-east-1:012345678912:alias/does-not-exist",
"invalid",
],
)
@mock_kms
def test_describe_key_via_alias_invalid_alias(key_id):
client = boto3.client("kms", region_name="us-east-1")
@ -204,8 +201,15 @@ def test_decrypt(plaintext):
decrypt_response["KeyId"].should.equal(key_arn)
@pytest.mark.parametrize("key_id",
["not-a-uuid", "alias/DoesNotExist", "arn:aws:kms:us-east-1:012345678912:alias/DoesNotExist", "d25652e4-d2d2-49f7-929a-671ccda580c6", "arn:aws:kms:us-east-1:012345678912:key/d25652e4-d2d2-49f7-929a-671ccda580c6"]
@pytest.mark.parametrize(
"key_id",
[
"not-a-uuid",
"alias/DoesNotExist",
"arn:aws:kms:us-east-1:012345678912:alias/DoesNotExist",
"d25652e4-d2d2-49f7-929a-671ccda580c6",
"arn:aws:kms:us-east-1:012345678912:key/d25652e4-d2d2-49f7-929a-671ccda580c6",
],
)
@mock_kms
def test_invalid_key_ids(key_id):
@ -352,14 +356,15 @@ def test_list_resource_tags():
assert response["Tags"][0]["TagValue"] == "string"
@pytest.mark.parametrize("kwargs,expected_key_length",
@pytest.mark.parametrize(
"kwargs,expected_key_length",
(
(dict(KeySpec="AES_256"), 32),
(dict(KeySpec="AES_128"), 16),
(dict(NumberOfBytes=64), 64),
(dict(NumberOfBytes=1), 1),
(dict(NumberOfBytes=1024), 1024),
)
),
)
@mock_kms
def test_generate_data_key_sizes(kwargs, expected_key_length):
@ -384,8 +389,15 @@ def test_generate_data_key_decrypt():
assert resp1["Plaintext"] == resp2["Plaintext"]
@pytest.mark.parametrize("kwargs",
[dict(KeySpec="AES_257"), dict(KeySpec="AES_128", NumberOfBytes=16), dict(NumberOfBytes=2048), dict(NumberOfBytes=0), dict()]
@pytest.mark.parametrize(
"kwargs",
[
dict(KeySpec="AES_257"),
dict(KeySpec="AES_128", NumberOfBytes=16),
dict(NumberOfBytes=2048),
dict(NumberOfBytes=0),
dict(),
],
)
@mock_kms
def test_generate_data_key_invalid_size_params(kwargs):
@ -398,8 +410,14 @@ def test_generate_data_key_invalid_size_params(kwargs):
client.generate_data_key(KeyId=key["KeyMetadata"]["KeyId"], **kwargs)
@pytest.mark.parametrize("key_id",
["alias/DoesNotExist", "arn:aws:kms:us-east-1:012345678912:alias/DoesNotExist", "d25652e4-d2d2-49f7-929a-671ccda580c6", "arn:aws:kms:us-east-1:012345678912:key/d25652e4-d2d2-49f7-929a-671ccda580c6"]
@pytest.mark.parametrize(
"key_id",
[
"alias/DoesNotExist",
"arn:aws:kms:us-east-1:012345678912:alias/DoesNotExist",
"d25652e4-d2d2-49f7-929a-671ccda580c6",
"arn:aws:kms:us-east-1:012345678912:key/d25652e4-d2d2-49f7-929a-671ccda580c6",
],
)
@mock_kms
def test_generate_data_key_invalid_key(key_id):
@ -409,8 +427,14 @@ def test_generate_data_key_invalid_key(key_id):
client.generate_data_key(KeyId=key_id, KeySpec="AES_256")
@pytest.mark.parametrize("prefix,append_key_id",
[("alias/DoesExist", False), ("arn:aws:kms:us-east-1:012345678912:alias/DoesExist", False), ("", True), ("arn:aws:kms:us-east-1:012345678912:key/", True)]
@pytest.mark.parametrize(
"prefix,append_key_id",
[
("alias/DoesExist", False),
("arn:aws:kms:us-east-1:012345678912:alias/DoesExist", False),
("", True),
("arn:aws:kms:us-east-1:012345678912:key/", True),
],
)
@mock_kms
def test_generate_data_key_all_valid_key_ids(prefix, append_key_id):
@ -512,8 +536,15 @@ def test_generate_random(number_of_bytes):
len(response["Plaintext"]).should.equal(number_of_bytes)
@pytest.mark.parametrize("number_of_bytes,error_type",
[(2048, botocore.exceptions.ClientError), (1025, botocore.exceptions.ClientError), (0, botocore.exceptions.ParamValidationError), (-1, botocore.exceptions.ParamValidationError), (-1024, botocore.exceptions.ParamValidationError)]
@pytest.mark.parametrize(
"number_of_bytes,error_type",
[
(2048, botocore.exceptions.ClientError),
(1025, botocore.exceptions.ClientError),
(0, botocore.exceptions.ParamValidationError),
(-1, botocore.exceptions.ParamValidationError),
(-1024, botocore.exceptions.ParamValidationError),
],
)
@mock_kms
def test_generate_random_invalid_number_of_bytes(number_of_bytes, error_type):

View File

@ -98,7 +98,9 @@ def test_deserialize_ciphertext_blob(raw, serialized):
test.should.equal(raw)
@pytest.mark.parametrize("encryption_context", [ec[0] for ec in ENCRYPTION_CONTEXT_VECTORS])
@pytest.mark.parametrize(
"encryption_context", [ec[0] for ec in ENCRYPTION_CONTEXT_VECTORS]
)
def test_encrypt_decrypt_cycle(encryption_context):
plaintext = b"some secret plaintext"
master_key = Key("nop", "nop", "nop", "nop", "nop")

View File

@ -205,8 +205,7 @@ def test_delete_subscription_filter_errors():
# when
client_logs.delete_subscription_filter(
logGroupName="/test",
filterName="test",
logGroupName="/test", filterName="test",
)
# then
@ -244,8 +243,7 @@ def test_delete_subscription_filter_errors():
# when
with pytest.raises(ClientError) as e:
client_logs.delete_subscription_filter(
logGroupName="not-existing-log-group",
filterName="test",
logGroupName="not-existing-log-group", filterName="test",
)
# then
@ -260,8 +258,7 @@ def test_delete_subscription_filter_errors():
# when
with pytest.raises(ClientError) as e:
client_logs.delete_subscription_filter(
logGroupName="/test",
filterName="wrong-filter-name",
logGroupName="/test", filterName="wrong-filter-name",
)
# then
@ -345,9 +342,7 @@ def _get_role_name(region_name):
return iam.get_role(RoleName="test-role")["Role"]["Arn"]
except ClientError:
return iam.create_role(
RoleName="test-role",
AssumeRolePolicyDocument="test policy",
Path="/",
RoleName="test-role", AssumeRolePolicyDocument="test policy", Path="/",
)["Role"]["Arn"]
@ -377,8 +372,7 @@ def _wait_for_log_msg(client, log_group_name, expected_msg_part):
for log_stream in log_streams:
result = client.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream["logStreamName"],
logGroupName=log_group_name, logStreamName=log_stream["logStreamName"],
)
received_messages.extend(
[event["message"] for event in result.get("events")]

View File

@ -448,9 +448,7 @@ def test_describe_subscription_filters_errors():
# when
with pytest.raises(ClientError) as e:
client.describe_subscription_filters(
logGroupName="not-existing-log-group",
)
client.describe_subscription_filters(logGroupName="not-existing-log-group",)
# then
ex = e.value

View File

@ -183,8 +183,7 @@ def test_create_another_member_withopts():
# But cannot get
response = conn.get_member.when.called_with(
NetworkId=network_id,
MemberId=member_id2,
NetworkId=network_id, MemberId=member_id2,
).should.throw(Exception, "Member {0} not found".format(member_id2))
# Delete member 1
@ -256,9 +255,7 @@ def test_invite_and_remove_member():
# Create proposal (invite and remove member)
response = conn.create_proposal(
NetworkId=network_id,
MemberId=member_id,
Actions=both_policy_actions,
NetworkId=network_id, MemberId=member_id, Actions=both_policy_actions,
)
proposal_id2 = response["ProposalId"]
@ -371,10 +368,7 @@ def test_create_too_many_members():
MemberConfiguration=helpers.create_member_configuration(
"testmember6", "admin", "Admin12345", False, "Test Member 6"
),
).should.throw(
Exception,
"is the maximum number of members allowed in a",
)
).should.throw(Exception, "is the maximum number of members allowed in a",)
@mock_managedblockchain
@ -600,8 +594,7 @@ def test_get_member_badmember():
network_id = response["NetworkId"]
response = conn.get_member.when.called_with(
NetworkId=network_id,
MemberId="m-ABCDEFGHIJKLMNOP0123456789",
NetworkId=network_id, MemberId="m-ABCDEFGHIJKLMNOP0123456789",
).should.throw(Exception, "Member m-ABCDEFGHIJKLMNOP0123456789 not found")
@ -631,8 +624,7 @@ def test_delete_member_badmember():
network_id = response["NetworkId"]
response = conn.delete_member.when.called_with(
NetworkId=network_id,
MemberId="m-ABCDEFGHIJKLMNOP0123456789",
NetworkId=network_id, MemberId="m-ABCDEFGHIJKLMNOP0123456789",
).should.throw(Exception, "Member m-ABCDEFGHIJKLMNOP0123456789 not found")

View File

@ -58,9 +58,7 @@ def test_create_node():
# Delete node
conn.delete_node(
NetworkId=network_id,
MemberId=member_id,
NodeId=node_id,
NetworkId=network_id, MemberId=member_id, NodeId=node_id,
)
# Find node in full list
@ -79,9 +77,7 @@ def test_create_node():
# But cannot get
response = conn.get_node.when.called_with(
NetworkId=network_id,
MemberId=member_id,
NodeId=node_id,
NetworkId=network_id, MemberId=member_id, NodeId=node_id,
).should.throw(Exception, "Node {0} not found".format(node_id))
@ -107,9 +103,7 @@ def test_create_node_standard_edition():
logconfigbad = dict(helpers.default_nodeconfiguration)
logconfigbad["InstanceType"] = "bc.t3.large"
response = conn.create_node(
NetworkId=network_id,
MemberId=member_id,
NodeConfiguration=logconfigbad,
NetworkId=network_id, MemberId=member_id, NodeConfiguration=logconfigbad,
)
node_id = response["NodeId"]
@ -152,8 +146,7 @@ def test_create_node_standard_edition():
# Should now be an exception
response = conn.list_nodes.when.called_with(
NetworkId=network_id,
MemberId=member_id,
NetworkId=network_id, MemberId=member_id,
).should.throw(Exception, "Member {0} not found".format(member_id))
@ -199,8 +192,7 @@ def test_create_too_many_nodes():
MemberId=member_id,
NodeConfiguration=helpers.default_nodeconfiguration,
).should.throw(
Exception,
"Maximum number of nodes exceeded in member {0}".format(member_id),
Exception, "Maximum number of nodes exceeded in member {0}".format(member_id),
)
@ -257,18 +249,14 @@ def test_create_node_badnodeconfig():
logconfigbad = dict(helpers.default_nodeconfiguration)
logconfigbad["InstanceType"] = "foo"
response = conn.create_node.when.called_with(
NetworkId=network_id,
MemberId=member_id,
NodeConfiguration=logconfigbad,
NetworkId=network_id, MemberId=member_id, NodeConfiguration=logconfigbad,
).should.throw(Exception, "Requested instance foo isn't supported.")
# Incorrect instance type for edition
logconfigbad = dict(helpers.default_nodeconfiguration)
logconfigbad["InstanceType"] = "bc.t3.large"
response = conn.create_node.when.called_with(
NetworkId=network_id,
MemberId=member_id,
NodeConfiguration=logconfigbad,
NetworkId=network_id, MemberId=member_id, NodeConfiguration=logconfigbad,
).should.throw(
Exception,
"Instance type bc.t3.large is not supported with STARTER Edition networks",
@ -278,9 +266,7 @@ def test_create_node_badnodeconfig():
logconfigbad = dict(helpers.default_nodeconfiguration)
logconfigbad["AvailabilityZone"] = "us-east-11"
response = conn.create_node.when.called_with(
NetworkId=network_id,
MemberId=member_id,
NodeConfiguration=logconfigbad,
NetworkId=network_id, MemberId=member_id, NodeConfiguration=logconfigbad,
).should.throw(Exception, "Availability Zone is not valid")
@ -310,8 +296,7 @@ def test_list_nodes_badmember():
network_id = response["NetworkId"]
response = conn.list_nodes.when.called_with(
NetworkId=network_id,
MemberId="m-ABCDEFGHIJKLMNOP0123456789",
NetworkId=network_id, MemberId="m-ABCDEFGHIJKLMNOP0123456789",
).should.throw(Exception, "Member m-ABCDEFGHIJKLMNOP0123456789 not found")

View File

@ -131,9 +131,7 @@ def test_create_proposal_badinvitationacctid():
member_id = response["MemberId"]
response = conn.create_proposal.when.called_with(
NetworkId=network_id,
MemberId=member_id,
Actions=actions,
NetworkId=network_id, MemberId=member_id, Actions=actions,
).should.throw(Exception, "Account ID format specified in proposal is not valid")
@ -157,9 +155,7 @@ def test_create_proposal_badremovalmemid():
member_id = response["MemberId"]
response = conn.create_proposal.when.called_with(
NetworkId=network_id,
MemberId=member_id,
Actions=actions,
NetworkId=network_id, MemberId=member_id, Actions=actions,
).should.throw(Exception, "Member ID format specified in proposal is not valid")
@ -198,6 +194,5 @@ def test_get_proposal_badproposal():
network_id = response["NetworkId"]
response = conn.get_proposal.when.called_with(
NetworkId=network_id,
ProposalId="p-ABCDEFGHIJKLMNOP0123456789",
NetworkId=network_id, ProposalId="p-ABCDEFGHIJKLMNOP0123456789",
).should.throw(Exception, "Proposal p-ABCDEFGHIJKLMNOP0123456789 not found")

View File

@ -666,6 +666,5 @@ def test_list_proposal_votes_badproposal():
member_id = response["MemberId"]
response = conn.list_proposal_votes.when.called_with(
NetworkId=network_id,
ProposalId="p-ABCDEFGHIJKLMNOP0123456789",
NetworkId=network_id, ProposalId="p-ABCDEFGHIJKLMNOP0123456789",
).should.throw(Exception, "Proposal p-ABCDEFGHIJKLMNOP0123456789 not found")

View File

@ -931,10 +931,7 @@ def test_tag_resource_errors():
with pytest.raises(ClientError) as e:
client.tag_resource(
ResourceId="000000000000",
Tags=[
{"Key": "key", "Value": "value"},
],
ResourceId="000000000000", Tags=[{"Key": "key", "Value": "value"},],
)
ex = e.value
ex.operation_name.should.equal("TagResource")

View File

@ -2397,9 +2397,7 @@ def test_boto3_get_object_if_match():
with pytest.raises(botocore.exceptions.ClientError) as err:
s3.get_object(
Bucket=bucket_name,
Key=key,
IfMatch='"hello"',
Bucket=bucket_name, Key=key, IfMatch='"hello"',
)
e = err.value
e.response["Error"]["Code"].should.equal("PreconditionFailed")
@ -2418,9 +2416,7 @@ def test_boto3_get_object_if_none_match():
with pytest.raises(botocore.exceptions.ClientError) as err:
s3.get_object(
Bucket=bucket_name,
Key=key,
IfNoneMatch=etag,
Bucket=bucket_name, Key=key, IfNoneMatch=etag,
)
e = err.value
e.response["Error"].should.equal({"Code": "304", "Message": "Not Modified"})
@ -2478,9 +2474,7 @@ def test_boto3_head_object_if_match():
with pytest.raises(botocore.exceptions.ClientError) as err:
s3.head_object(
Bucket=bucket_name,
Key=key,
IfMatch='"hello"',
Bucket=bucket_name, Key=key, IfMatch='"hello"',
)
e = err.value
e.response["Error"].should.equal({"Code": "412", "Message": "Precondition Failed"})
@ -2498,9 +2492,7 @@ def test_boto3_head_object_if_none_match():
with pytest.raises(botocore.exceptions.ClientError) as err:
s3.head_object(
Bucket=bucket_name,
Key=key,
IfNoneMatch=etag,
Bucket=bucket_name, Key=key, IfNoneMatch=etag,
)
e = err.value
e.response["Error"].should.equal({"Code": "304", "Message": "Not Modified"})
@ -4037,8 +4029,8 @@ def test_leading_slashes_not_removed(bucket_name):
e.value.response["Error"]["Code"].should.equal("NoSuchKey")
@pytest.mark.parametrize("key",
["foo/bar/baz", "foo", "foo/run_dt%3D2019-01-01%252012%253A30%253A00"]
@pytest.mark.parametrize(
"key", ["foo/bar/baz", "foo", "foo/run_dt%3D2019-01-01%252012%253A30%253A00"]
)
@mock_s3
def test_delete_objects_with_url_encoded_key(key):

View File

@ -14,12 +14,7 @@ def test_s3_bucket_cloudformation_basic():
template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"testInstance": {
"Type": "AWS::S3::Bucket",
"Properties": {},
}
},
"Resources": {"testInstance": {"Type": "AWS::S3::Bucket", "Properties": {},}},
"Outputs": {"Bucket": {"Value": {"Ref": "testInstance"}}},
}
template_json = json.dumps(template)

View File

@ -93,7 +93,8 @@ def test_parse_region_from_url():
parse_region_from_url(url).should.equal(expected)
@pytest.mark.parametrize("key,expected",
@pytest.mark.parametrize(
"key,expected",
[
("foo/bar/baz", "foo/bar/baz"),
("foo", "foo"),
@ -101,13 +102,14 @@ def test_parse_region_from_url():
"foo/run_dt%3D2019-01-01%252012%253A30%253A00",
"foo/run_dt=2019-01-01%2012%3A30%3A00",
),
]
],
)
def test_clean_key_name(key, expected):
clean_key_name(key).should.equal(expected)
@pytest.mark.parametrize("key,expected",
@pytest.mark.parametrize(
"key,expected",
[
("foo/bar/baz", "foo/bar/baz"),
("foo", "foo"),
@ -115,7 +117,7 @@ def test_clean_key_name(key, expected):
"foo/run_dt%3D2019-01-01%252012%253A30%253A00",
"foo/run_dt%253D2019-01-01%25252012%25253A30%25253A00",
),
]
],
)
def test_undo_clean_key_name(key, expected):
undo_clean_key_name(key).should.equal(expected)

View File

@ -638,7 +638,9 @@ def test_put_secret_value_on_non_existing_secret():
VersionStages=["AWSCURRENT"],
)
cm.value.response["Error"]["Message"].should.equal("Secrets Manager can't find the specified secret.")
cm.value.response["Error"]["Message"].should.equal(
"Secrets Manager can't find the specified secret."
)
@mock_secretsmanager
@ -923,17 +925,11 @@ def test_tag_resource():
conn = boto3.client("secretsmanager", region_name="us-west-2")
conn.create_secret(Name="test-secret", SecretString="foosecret")
conn.tag_resource(
SecretId="test-secret",
Tags=[
{"Key": "FirstTag", "Value": "SomeValue"},
],
SecretId="test-secret", Tags=[{"Key": "FirstTag", "Value": "SomeValue"},],
)
conn.tag_resource(
SecretId="test-secret",
Tags=[
{"Key": "SecondTag", "Value": "AnotherValue"},
],
SecretId="test-secret", Tags=[{"Key": "SecondTag", "Value": "AnotherValue"},],
)
secrets = conn.list_secrets()
@ -945,14 +941,13 @@ def test_tag_resource():
with pytest.raises(ClientError) as cm:
conn.tag_resource(
SecretId="dummy-test-secret",
Tags=[
{"Key": "FirstTag", "Value": "SomeValue"},
],
Tags=[{"Key": "FirstTag", "Value": "SomeValue"},],
)
assert \
"Secrets Manager can't find the specified secret." == \
cm.value.response["Error"]["Message"]
assert (
"Secrets Manager can't find the specified secret."
== cm.value.response["Error"]["Message"]
)
@mock_secretsmanager

View File

@ -89,9 +89,7 @@ def test_send_email_when_verify_source():
conn = boto3.client("ses", region_name="us-east-1")
kwargs = dict(
Destination={
"ToAddresses": ["test_to@example.com"],
},
Destination={"ToAddresses": ["test_to@example.com"],},
Message={
"Subject": {"Data": "test subject"},
"Body": {"Text": {"Data": "test body"}},
@ -278,16 +276,7 @@ def test_send_email_notification_with_encoded_sender():
response = conn.send_email(
Source=sender,
Destination={"ToAddresses": ["your.friend@hotmail.com"]},
Message={
"Subject": {
"Data": "hi",
},
"Body": {
"Text": {
"Data": "there",
}
},
},
Message={"Subject": {"Data": "hi",}, "Body": {"Text": {"Data": "there",}},},
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
@ -302,9 +291,7 @@ def test_create_configuration_set():
EventDestination={
"Name": "snsEvent",
"Enabled": True,
"MatchingEventTypes": [
"send",
],
"MatchingEventTypes": ["send",],
"SNSDestination": {
"TopicARN": "arn:aws:sns:us-east-1:123456789012:myTopic"
},
@ -317,9 +304,7 @@ def test_create_configuration_set():
EventDestination={
"Name": "snsEvent",
"Enabled": True,
"MatchingEventTypes": [
"send",
],
"MatchingEventTypes": ["send",],
"SNSDestination": {
"TopicARN": "arn:aws:sns:us-east-1:123456789012:myTopic"
},
@ -334,9 +319,7 @@ def test_create_configuration_set():
EventDestination={
"Name": "snsEvent",
"Enabled": True,
"MatchingEventTypes": [
"send",
],
"MatchingEventTypes": ["send",],
"SNSDestination": {
"TopicARN": "arn:aws:sns:us-east-1:123456789012:myTopic"
},

View File

@ -152,9 +152,7 @@ def test_publish_to_sqs_msg_attr_byte_value():
sqs = boto3.resource("sqs", region_name="us-east-1")
queue = sqs.create_queue(QueueName="test-queue")
conn.subscribe(
TopicArn=topic_arn,
Protocol="sqs",
Endpoint=queue.attributes["QueueArn"],
TopicArn=topic_arn, Protocol="sqs", Endpoint=queue.attributes["QueueArn"],
)
queue_raw = sqs.create_queue(QueueName="test-queue-raw")
conn.subscribe(

View File

@ -525,9 +525,7 @@ def test_untag_resource_error():
@mock_sns
def test_topic_kms_master_key_id_attribute():
client = boto3.client("sns", region_name="us-west-2")
resp = client.create_topic(
Name="test-sns-no-key-attr",
)
resp = client.create_topic(Name="test-sns-no-key-attr",)
topic_arn = resp["TopicArn"]
resp = client.get_topic_attributes(TopicArn=topic_arn)
resp["Attributes"].should_not.have.key("KmsMasterKeyId")
@ -540,10 +538,7 @@ def test_topic_kms_master_key_id_attribute():
resp["Attributes"]["KmsMasterKeyId"].should.equal("test-key")
resp = client.create_topic(
Name="test-sns-with-key-attr",
Attributes={
"KmsMasterKeyId": "key-id",
},
Name="test-sns-with-key-attr", Attributes={"KmsMasterKeyId": "key-id",},
)
topic_arn = resp["TopicArn"]
resp = client.get_topic_attributes(TopicArn=topic_arn)

View File

@ -719,10 +719,7 @@ def test_send_receive_message_with_attributes_with_labels():
response = queue.send_message(
MessageBody="test message",
MessageAttributes={
"somevalue": {
"StringValue": "somevalue",
"DataType": "String.custom",
}
"somevalue": {"StringValue": "somevalue", "DataType": "String.custom",}
},
)
@ -2245,9 +2242,7 @@ def test_invoke_function_from_sqs_exception():
@mock_sqs
def test_maximum_message_size_attribute_default():
sqs = boto3.resource("sqs", region_name="eu-west-3")
queue = sqs.create_queue(
QueueName="test-queue",
)
queue = sqs.create_queue(QueueName="test-queue",)
int(queue.attributes["MaximumMessageSize"]).should.equal(MAXIMUM_MESSAGE_LENGTH)
with pytest.raises(Exception) as e:
queue.send_message(MessageBody="a" * (MAXIMUM_MESSAGE_LENGTH + 1))

View File

@ -309,29 +309,25 @@ def test_put_parameter_invalid_names():
client.put_parameter.when.called_with(
Name="ssm_test", Value="value", Type="String"
).should.throw(
ClientError,
invalid_prefix_err,
ClientError, invalid_prefix_err,
)
client.put_parameter.when.called_with(
Name="SSM_TEST", Value="value", Type="String"
).should.throw(
ClientError,
invalid_prefix_err,
ClientError, invalid_prefix_err,
)
client.put_parameter.when.called_with(
Name="aws_test", Value="value", Type="String"
).should.throw(
ClientError,
invalid_prefix_err,
ClientError, invalid_prefix_err,
)
client.put_parameter.when.called_with(
Name="AWS_TEST", Value="value", Type="String"
).should.throw(
ClientError,
invalid_prefix_err,
ClientError, invalid_prefix_err,
)
ssm_path = "/ssm_test/path/to/var"
@ -358,16 +354,14 @@ def test_put_parameter_invalid_names():
client.put_parameter.when.called_with(
Name=aws_path, Value="value", Type="String"
).should.throw(
ClientError,
"No access to reserved parameter name: {}.".format(aws_path),
ClientError, "No access to reserved parameter name: {}.".format(aws_path),
)
aws_path = "/AWS/PATH/TO/VAR"
client.put_parameter.when.called_with(
Name=aws_path, Value="value", Type="String"
).should.throw(
ClientError,
"No access to reserved parameter name: {}.".format(aws_path),
ClientError, "No access to reserved parameter name: {}.".format(aws_path),
)

View File

@ -356,10 +356,8 @@ def test_state_machine_can_deleted_nonexisting_machine():
@mock_stepfunctions
def test_state_machine_tagging_non_existent_resource_fails():
client = boto3.client("stepfunctions", region_name=region)
non_existent_arn = (
"arn:aws:states:{region}:{account}:stateMachine:non-existent".format(
region=region, account=ACCOUNT_ID
)
non_existent_arn = "arn:aws:states:{region}:{account}:stateMachine:non-existent".format(
region=region, account=ACCOUNT_ID
)
with pytest.raises(ClientError) as ex:
client.tag_resource(resourceArn=non_existent_arn, tags=[])
@ -370,10 +368,8 @@ def test_state_machine_tagging_non_existent_resource_fails():
@mock_stepfunctions
def test_state_machine_untagging_non_existent_resource_fails():
client = boto3.client("stepfunctions", region_name=region)
non_existent_arn = (
"arn:aws:states:{region}:{account}:stateMachine:non-existent".format(
region=region, account=ACCOUNT_ID
)
non_existent_arn = "arn:aws:states:{region}:{account}:stateMachine:non-existent".format(
region=region, account=ACCOUNT_ID
)
with pytest.raises(ClientError) as ex:
client.untag_resource(resourceArn=non_existent_arn, tagKeys=[])
@ -390,9 +386,7 @@ def test_state_machine_tagging():
{"key": "tag_key2", "value": "tag_value2"},
]
machine = client.create_state_machine(
name="test",
definition=str(simple_definition),
roleArn=_get_default_role(),
name="test", definition=str(simple_definition), roleArn=_get_default_role(),
)
client.tag_resource(resourceArn=machine["stateMachineArn"], tags=tags)
resp = client.list_tags_for_resource(resourceArn=machine["stateMachineArn"])
@ -944,9 +938,7 @@ def test_state_machine_cloudformation_update_with_replacement():
with pytest.raises(ClientError) as ex:
sf.describe_state_machine(stateMachineArn=original_machine_arn)
ex.value.response["Error"]["Code"].should.equal("StateMachineDoesNotExist")
ex.value.response["Error"]["Message"].should.contain(
"State Machine Does Not Exist"
)
ex.value.response["Error"]["Message"].should.contain("State Machine Does Not Exist")
@mock_stepfunctions

View File

@ -17,9 +17,7 @@ def test_run_medical_transcription_job_minimal_params():
args = {
"MedicalTranscriptionJobName": job_name,
"LanguageCode": "en-US",
"Media": {
"MediaFileUri": "s3://my-bucket/my-media-file.wav",
},
"Media": {"MediaFileUri": "s3://my-bucket/my-media-file.wav",},
"OutputBucketName": "my-output-bucket",
"Specialty": "PRIMARYCARE",
"Type": "CONVERSATION",
@ -100,9 +98,7 @@ def test_run_medical_transcription_job_all_params():
"LanguageCode": "en-US",
"MediaSampleRateHertz": 48000,
"MediaFormat": "flac",
"Media": {
"MediaFileUri": "s3://my-bucket/my-media-file.dat",
},
"Media": {"MediaFileUri": "s3://my-bucket/my-media-file.dat",},
"OutputBucketName": "my-output-bucket",
"OutputEncryptionKMSKeyId": "arn:aws:kms:us-east-1:012345678901:key/37111b5e-8eff-4706-ae3a-d4f9d1d559fc",
"Settings": {
@ -203,9 +199,7 @@ def test_run_medical_transcription_job_with_existing_job_name():
args = {
"MedicalTranscriptionJobName": job_name,
"LanguageCode": "en-US",
"Media": {
"MediaFileUri": "s3://my-bucket/my-media-file.wav",
},
"Media": {"MediaFileUri": "s3://my-bucket/my-media-file.wav",},
"OutputBucketName": "my-output-bucket",
"Specialty": "PRIMARYCARE",
"Type": "CONVERSATION",
@ -228,9 +222,7 @@ def test_run_medical_transcription_job_nonexistent_vocabulary():
args = {
"MedicalTranscriptionJobName": job_name,
"LanguageCode": "en-US",
"Media": {
"MediaFileUri": "s3://my-bucket/my-media-file.dat",
},
"Media": {"MediaFileUri": "s3://my-bucket/my-media-file.dat",},
"OutputBucketName": "my-output-bucket",
"Settings": {"VocabularyName": "NonexistentVocabulary"},
"Specialty": "PRIMARYCARE",
@ -252,9 +244,7 @@ def test_list_medical_transcription_jobs():
args = {
"MedicalTranscriptionJobName": job_name,
"LanguageCode": "en-US",
"Media": {
"MediaFileUri": "s3://my-bucket/my-media-file.wav",
},
"Media": {"MediaFileUri": "s3://my-bucket/my-media-file.wav",},
"OutputBucketName": "my-output-bucket",
"Specialty": "PRIMARYCARE",
"Type": "CONVERSATION",