Pytest model of exceptions uses 'value' attribute instead of 'exception'.

This commit is contained in:
Matěj Cepl 2020-11-10 18:14:50 +01:00
parent 2705698d83
commit 6d364dc7aa
9 changed files with 75 additions and 82 deletions

View File

@ -1,18 +1,16 @@
from __future__ import unicode_literals
import os
import boto3
from freezegun import freeze_time
import sure # noqa
import uuid
import boto3
import pytest
import sure # noqa
from botocore.exceptions import ClientError
from freezegun import freeze_time
from moto import mock_acm, settings
from moto.core import ACCOUNT_ID
from nose import SkipTest
from nose.tools import assert_raises
from unittest import SkipTest
RESOURCE_FOLDER = os.path.join(os.path.dirname(__file__), "resources")
_GET_RESOURCE = lambda x: open(os.path.join(RESOURCE_FOLDER, x), "rb").read()
@ -404,7 +402,7 @@ def test_operations_with_invalid_tags():
client = boto3.client("acm", region_name="eu-central-1")
# request certificate with invalid tags
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.request_certificate(
DomainName="example.com",
Tags=[{"Key": "X" * 200, "Value": "Valid"}],
@ -415,7 +413,7 @@ def test_operations_with_invalid_tags():
)
# import certificate with invalid tags
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.import_certificate(
Certificate=SERVER_CRT,
PrivateKey=SERVER_KEY,
@ -434,7 +432,7 @@ def test_operations_with_invalid_tags():
arn = _import_cert(client)
# add invalid tags to existing certificate
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.add_tags_to_certificate(
CertificateArn=arn,
Tags=[{"Key": "aws:xxx", "Value": "Valid"}, {"Key": "key2"}],
@ -445,7 +443,7 @@ def test_operations_with_invalid_tags():
)
# try removing invalid tags from existing certificate
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.remove_tags_from_certificate(
CertificateArn=arn, Tags=[{"Key": "aws:xxx", "Value": "Valid"}]
)
@ -461,7 +459,7 @@ def test_add_too_many_tags():
arn = _import_cert(client)
# Add 51 tags
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.add_tags_to_certificate(
CertificateArn=arn,
Tags=[{"Key": "a-%d" % i, "Value": "abcd"} for i in range(1, 52)],
@ -478,7 +476,7 @@ def test_add_too_many_tags():
client.list_tags_for_certificate(CertificateArn=arn)["Tags"].should.have.length_of(
49
)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.add_tags_to_certificate(
CertificateArn=arn,
Tags=[{"Key": "x-1", "Value": "xyz"}, {"Key": "x-2", "Value": "xyz"}],

View File

@ -343,7 +343,7 @@ def test_put_scaling_policy():
PolicyType="ABCDEFG",
TargetTrackingScalingPolicyConfiguration=policy_body,
)
e.exception.response["Error"]["Message"].should.match(
e.value.response["Error"]["Message"].should.match(
r"Unknown policy type .* specified."
)
@ -450,7 +450,7 @@ def test_delete_scaling_policies():
ResourceId=resource_id,
ScalableDimension=scalable_dimension,
)
e.exception.response["Error"]["Message"].should.match(r"No scaling policy found .*")
e.value.response["Error"]["Message"].should.match(r"No scaling policy found .*")
response = client.put_scaling_policy(
PolicyName=policy_name,
@ -513,6 +513,6 @@ def test_deregister_scalable_target():
ResourceId=resource_id,
ScalableDimension=scalable_dimension,
)
e.exception.response["Error"]["Message"].should.match(
e.value.response["Error"]["Message"].should.match(
r"No scalable target found .*"
)

View File

@ -1,16 +1,14 @@
from __future__ import unicode_literals
# Ensure 'pytest.raises' context manager support for Python 2.6
import pytest
from moto.ec2 import ec2_backends
import boto
import boto3
from botocore.exceptions import ClientError
from boto.exception import EC2ResponseError
# Ensure 'pytest.raises' context manager support for Python 2.6
import pytest
import sure # noqa
from moto import mock_ec2_deprecated, mock_ec2
from boto.exception import EC2ResponseError
from botocore.exceptions import ClientError
from moto import mock_ec2, mock_ec2_deprecated
from moto.ec2 import ec2_backends
from moto.ec2.models import OWNER_ID
from moto.kms import mock_kms
@ -920,12 +918,12 @@ def test_search_for_many_snapshots():
@mock_ec2
def test_create_unencrypted_volume_with_kms_key_fails():
resource = boto3.resource("ec2", region_name="us-east-1")
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
resource.create_volume(
AvailabilityZone="us-east-1a", Encrypted=False, KmsKeyId="key", Size=10
)
ex.exception.response["Error"]["Code"].should.equal("InvalidParameterDependency")
ex.exception.response["Error"]["Message"].should.contain("KmsKeyId")
ex.value.response["Error"]["Code"].should.equal("InvalidParameterDependency")
ex.value.response["Error"]["Message"].should.contain("KmsKeyId")
@mock_kms
@ -933,9 +931,9 @@ def test_create_unencrypted_volume_with_kms_key_fails():
def test_create_encrypted_volume_without_kms_key_should_use_default_key():
kms = boto3.client("kms", region_name="us-east-1")
# Default master key for EBS does not exist until needed.
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
kms.describe_key(KeyId="alias/aws/ebs")
ex.exception.response["Error"]["Code"].should.equal("NotFoundException")
ex.value.response["Error"]["Code"].should.equal("NotFoundException")
# Creating an encrypted volume should create (and use) the default key.
resource = boto3.resource("ec2", region_name="us-east-1")
volume = resource.create_volume(

View File

@ -1,16 +1,16 @@
from __future__ import unicode_literals
import random
import boto
import boto3
import boto.vpc
# Ensure 'pytest.raises' context manager support for Python 2.6
import pytest
import boto3
import boto
import boto.vpc
from boto.exception import EC2ResponseError
from botocore.exceptions import ParamValidationError, ClientError
import sure # noqa
import random
from boto.exception import EC2ResponseError
from botocore.exceptions import ClientError, ParamValidationError
from moto import mock_ec2, mock_ec2_deprecated
@ -426,7 +426,7 @@ def test_create_subnet_with_invalid_cidr_range_multiple_vpc_cidr_blocks():
vpc.is_default.shouldnt.be.ok
subnet_cidr_block = "10.2.0.0/20"
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock=subnet_cidr_block)
str(ex.exception).should.equal(
"An error occurred (InvalidSubnet.Range) when calling the CreateSubnet "

View File

@ -1,13 +1,12 @@
from __future__ import unicode_literals
import boto3
import pytest
import sure # noqa
from botocore.exceptions import ClientError
from nose.tools import assert_raises
from parameterized import parameterized
from moto import mock_forecast
from moto.core import ACCOUNT_ID
from parameterized import parameterized
region = "us-east-1"
account_id = None
@ -40,7 +39,7 @@ def test_forecast_dataset_group_create_invalid_domain():
client = boto3.client("forecast", region_name=region)
invalid_domain = "INVALID"
with assert_raises(ClientError) as exc:
with pytest.raises(ClientError) as exc:
client.create_dataset_group(DatasetGroupName=name, Domain=invalid_domain)
exc.exception.response["Error"]["Code"].should.equal("ValidationException")
exc.exception.response["Error"]["Message"].should.equal(
@ -55,7 +54,7 @@ def test_forecast_dataset_group_create_invalid_domain():
def test_forecast_dataset_group_create_invalid_name(name):
client = boto3.client("forecast", region_name=region)
with assert_raises(ClientError) as exc:
with pytest.raises(ClientError) as exc:
client.create_dataset_group(DatasetGroupName=name, Domain="CUSTOM")
exc.exception.response["Error"]["Code"].should.equal("ValidationException")
exc.exception.response["Error"]["Message"].should.contain(
@ -70,7 +69,7 @@ def test_forecast_dataset_group_create_duplicate_fails():
client = boto3.client("forecast", region_name=region)
client.create_dataset_group(DatasetGroupName="name", Domain="RETAIL")
with assert_raises(ClientError) as exc:
with pytest.raises(ClientError) as exc:
client.create_dataset_group(DatasetGroupName="name", Domain="RETAIL")
exc.exception.response["Error"]["Code"].should.equal(
@ -122,7 +121,7 @@ def test_forecast_delete_dataset_group_missing():
"arn:aws:forecast:" + region + ":" + ACCOUNT_ID + ":dataset-group/missing"
)
with assert_raises(ClientError) as exc:
with pytest.raises(ClientError) as exc:
client.delete_dataset_group(DatasetGroupArn=missing_dsg_arn)
exc.exception.response["Error"]["Code"].should.equal("ResourceNotFoundException")
exc.exception.response["Error"]["Message"].should.equal(
@ -152,7 +151,7 @@ def test_forecast_update_dataset_group_not_found():
dataset_group_arn = (
"arn:aws:forecast:" + region + ":" + ACCOUNT_ID + ":dataset-group/" + "test"
)
with assert_raises(ClientError) as exc:
with pytest.raises(ClientError) as exc:
client.update_dataset_group(DatasetGroupArn=dataset_group_arn, DatasetArns=[])
exc.exception.response["Error"]["Code"].should.equal("ResourceNotFoundException")
exc.exception.response["Error"]["Message"].should.equal(
@ -180,7 +179,7 @@ def test_describe_dataset_group_missing():
dataset_group_arn = (
"arn:aws:forecast:" + region + ":" + ACCOUNT_ID + ":dataset-group/name"
)
with assert_raises(ClientError) as exc:
with pytest.raises(ClientError) as exc:
client.describe_dataset_group(DatasetGroupArn=dataset_group_arn)
exc.exception.response["Error"]["Code"].should.equal("ResourceNotFoundException")
exc.exception.response["Error"]["Message"].should.equal(
@ -192,7 +191,7 @@ def test_describe_dataset_group_missing():
def test_create_dataset_group_missing_datasets():
client = boto3.client("forecast", region_name=region)
dataset_arn = "arn:aws:forecast:" + region + ":" + ACCOUNT_ID + ":dataset/name"
with assert_raises(ClientError) as exc:
with pytest.raises(ClientError) as exc:
client.create_dataset_group(
DatasetGroupName="name", Domain="CUSTOM", DatasetArns=[dataset_arn]
)
@ -212,7 +211,7 @@ def test_update_dataset_group_missing_datasets():
client.create_dataset_group(DatasetGroupName=name, Domain="CUSTOM")
dataset_arn = "arn:aws:forecast:" + region + ":" + ACCOUNT_ID + ":dataset/name"
with assert_raises(ClientError) as exc:
with pytest.raises(ClientError) as exc:
client.update_dataset_group(
DatasetGroupArn=dataset_group_arn, DatasetArns=[dataset_arn]
)

View File

@ -608,7 +608,7 @@ def test_create_certificate_validation():
client.register_certificate(
certificatePem=cert["certificatePem"], setAsActive=False
)
e.exception.response["Error"]["Message"].should.contain(
e.value.response["Error"]["Message"].should.contain(
"The certificate is already provisioned or registered"
)
@ -616,7 +616,7 @@ def test_create_certificate_validation():
client.register_certificate_without_ca(
certificatePem=cert["certificatePem"], status="ACTIVE"
)
e.exception.response["Error"]["Message"].should.contain(
e.value.response["Error"]["Message"].should.contain(
"The certificate is already provisioned or registered"
)

View File

@ -631,17 +631,16 @@ def test_rotate_secret_rotation_period_too_long():
@mock_secretsmanager
def test_put_secret_value_on_non_existing_secret():
conn = boto3.client("secretsmanager", region_name="us-west-2")
with assert_raises(ClientError) as cm:
with pytest.raises(ClientError) as cm:
conn.put_secret_value(
SecretId=DEFAULT_SECRET_NAME,
SecretString="foosecret",
VersionStages=["AWSCURRENT"],
)
assert_equal(
"Secrets Manager can't find the specified secret.",
cm.exception.response["Error"]["Message"],
)
assert \
"Secrets Manager can't find the specified secret." == \
cm.exception.response["Error"]["Message"]
@mock_secretsmanager
@ -945,7 +944,7 @@ def test_tag_resource():
{"Key": "SecondTag", "Value": "AnotherValue"},
]
with assert_raises(ClientError) as cm:
with pytest.raises(ClientError) as cm:
conn.tag_resource(
SecretId="dummy-test-secret",
Tags=[
@ -953,10 +952,9 @@ def test_tag_resource():
],
)
assert_equal(
"Secrets Manager can't find the specified secret.",
cm.exception.response["Error"]["Message"],
)
assert \
"Secrets Manager can't find the specified secret." == \
cm.exception.response["Error"]["Message"]
@mock_secretsmanager

View File

@ -273,7 +273,7 @@ def test_message_send_with_attributes():
def test_message_with_invalid_attributes():
sqs = boto3.resource("sqs", region_name="us-east-1")
queue = sqs.create_queue(QueueName="blah")
with assert_raises(ClientError) as e:
with pytest.raises(ClientError) as e:
queue.send_message(
MessageBody="derp",
MessageAttributes={
@ -2249,7 +2249,7 @@ def test_maximum_message_size_attribute_default():
QueueName="test-queue",
)
int(queue.attributes["MaximumMessageSize"]).should.equal(MAXIMUM_MESSAGE_LENGTH)
with assert_raises(Exception) as e:
with pytest.raises(Exception) as e:
queue.send_message(MessageBody="a" * (MAXIMUM_MESSAGE_LENGTH + 1))
ex = e.exception
ex.response["Error"]["Code"].should.equal("InvalidParameterValue")
@ -2263,7 +2263,7 @@ def test_maximum_message_size_attribute_fails_for_invalid_values():
MAXIMUM_MESSAGE_SIZE_ATTR_UPPER_BOUND + 1,
]
for message_size in invalid_values:
with assert_raises(ClientError) as e:
with pytest.raises(ClientError) as e:
sqs.create_queue(
QueueName="test-queue",
Attributes={"MaximumMessageSize": str(message_size)},
@ -2281,7 +2281,7 @@ def test_send_message_fails_when_message_size_greater_than_max_message_size():
Attributes={"MaximumMessageSize": str(message_size_limit)},
)
int(queue.attributes["MaximumMessageSize"]).should.equal(message_size_limit)
with assert_raises(ClientError) as e:
with pytest.raises(ClientError) as e:
queue.send_message(MessageBody="a" * (message_size_limit + 1))
ex = e.exception
ex.response["Error"]["Code"].should.equal("InvalidParameterValue")

View File

@ -361,10 +361,10 @@ def test_state_machine_tagging_non_existent_resource_fails():
region=region, account=ACCOUNT_ID
)
)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.tag_resource(resourceArn=non_existent_arn, tags=[])
ex.exception.response["Error"]["Code"].should.equal("ResourceNotFound")
ex.exception.response["Error"]["Message"].should.contain(non_existent_arn)
ex.value.response["Error"]["Code"].should.equal("ResourceNotFound")
ex.value.response["Error"]["Message"].should.contain(non_existent_arn)
@mock_stepfunctions
@ -375,10 +375,10 @@ def test_state_machine_untagging_non_existent_resource_fails():
region=region, account=ACCOUNT_ID
)
)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.untag_resource(resourceArn=non_existent_arn, tagKeys=[])
ex.exception.response["Error"]["Code"].should.equal("ResourceNotFound")
ex.exception.response["Error"]["Message"].should.contain(non_existent_arn)
ex.value.response["Error"]["Code"].should.equal("ResourceNotFound")
ex.value.response["Error"]["Message"].should.contain(non_existent_arn)
@mock_stepfunctions
@ -664,7 +664,7 @@ def test_state_machine_list_executions_with_pagination():
for page in page_iterator:
page["executions"].should.have.length_of(25)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
resp = client.list_executions(
stateMachineArn=sm["stateMachineArn"], maxResults=10
)
@ -674,16 +674,16 @@ def test_state_machine_list_executions_with_pagination():
statusFilter="ABORTED",
nextToken=resp["nextToken"],
)
ex.exception.response["Error"]["Code"].should.equal("InvalidToken")
ex.exception.response["Error"]["Message"].should.contain(
ex.value.response["Error"]["Code"].should.equal("InvalidToken")
ex.value.response["Error"]["Message"].should.contain(
"Input inconsistent with page token"
)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.list_executions(
stateMachineArn=sm["stateMachineArn"], nextToken="invalid"
)
ex.exception.response["Error"]["Code"].should.equal("InvalidToken")
ex.value.response["Error"]["Code"].should.equal("InvalidToken")
@mock_stepfunctions
@ -867,10 +867,10 @@ def test_state_machine_cloudformation():
tag["value"].should.equal("value{}".format(i))
cf.Stack("test_stack").delete()
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
sf.describe_state_machine(stateMachineArn=output["StateMachineArn"])
ex.exception.response["Error"]["Code"].should.equal("StateMachineDoesNotExist")
ex.exception.response["Error"]["Message"].should.contain("Does Not Exist")
ex.value.response["Error"]["Code"].should.equal("StateMachineDoesNotExist")
ex.value.response["Error"]["Message"].should.contain("Does Not Exist")
@mock_stepfunctions
@ -941,10 +941,10 @@ def test_state_machine_cloudformation_update_with_replacement():
if tag["key"] == "key1":
tag["value"].should.equal("updated_value")
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
sf.describe_state_machine(stateMachineArn=original_machine_arn)
ex.exception.response["Error"]["Code"].should.equal("StateMachineDoesNotExist")
ex.exception.response["Error"]["Message"].should.contain(
ex.value.response["Error"]["Code"].should.equal("StateMachineDoesNotExist")
ex.value.response["Error"]["Message"].should.contain(
"State Machine Does Not Exist"
)