EC2 - Return custom error response for DescribeRouteTable/DescribeVPC (#4225)

This commit is contained in:
Bert Blommers 2021-08-26 18:53:26 +01:00 committed by GitHub
parent e865362791
commit 6a644850f6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 76 additions and 1 deletions

View File

@ -2,6 +2,21 @@ from __future__ import unicode_literals
from moto.core.exceptions import RESTError
# DescribeRouteTable has a custom root-tag - <Response> vs <ErrorResponse>
# TF complains if the roottag is incorrect
CUSTOM_ERROR_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Errors>
<Error>
<Code>{{error_type}}</Code>
<Message>{{message}}</Message>
</Error>
</Errors>
<{{request_id_tag}}>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</{{request_id_tag}}>
</Response>
"""
class EC2ClientError(RESTError):
code = 400
# EC2 uses <RequestID> as tag name in the XML response
@ -60,8 +75,14 @@ class InvalidKeyPairFormatError(EC2ClientError):
class InvalidVPCIdError(EC2ClientError):
def __init__(self, vpc_id):
kwargs = {}
kwargs.setdefault("template", "custom_response")
self.templates["custom_response"] = CUSTOM_ERROR_RESPONSE
super(InvalidVPCIdError, self).__init__(
"InvalidVpcID.NotFound", "VpcID {0} does not exist.".format(vpc_id)
"InvalidVpcID.NotFound",
"VpcID {0} does not exist.".format(vpc_id),
**kwargs
)
@ -176,9 +197,14 @@ class InvalidPermissionDuplicateError(EC2ClientError):
class InvalidRouteTableIdError(EC2ClientError):
def __init__(self, route_table_id):
kwargs = {}
kwargs.setdefault("template", "custom_response")
self.templates["custom_response"] = CUSTOM_ERROR_RESPONSE
super(InvalidRouteTableIdError, self).__init__(
"InvalidRouteTableID.NotFound",
"The routeTable ID '{0}' does not exist".format(route_table_id),
**kwargs
)

View File

@ -1,6 +1,7 @@
from __future__ import unicode_literals
import re
import sure # noqa
import xmltodict
import moto.server as server
from tests import EXAMPLE_AMI_ID
@ -24,3 +25,51 @@ def test_ec2_server_get():
res = test_client.get("/?Action=DescribeInstances")
res.data.decode("utf-8").should.contain(instance_id)
def test_ec2_get_unknown_vpc():
"""
Ensure that this call returns the error format in the right format
Terraform will throw errors when destroying a VPC otherwise
:return:
"""
backend = server.create_backend_app("ec2")
test_client = backend.test_client()
res = test_client.get(
"/?Action=DescribeVpcs&VpcId.1=vpc-unknown",
headers={"Host": "ec2.us-east-1.amazonaws.com"},
)
res.status_code.should.equal(400)
body = xmltodict.parse(res.data.decode("utf-8"), dict_constructor=dict)
body.should.have.key("Response")
body["Response"].should.have.key("Errors")
body["Response"]["Errors"].should.have.key("Error")
error = body["Response"]["Errors"]["Error"]
error["Code"].should.equal("InvalidVpcID.NotFound")
error["Message"].should.equal("VpcID {'vpc-unknown'} does not exist.")
def test_ec2_get_unknown_route_table():
"""
Ensure that this call returns the error format in the right format
Terraform will throw errors when destroying a RouteTable otherwise
:return:
"""
backend = server.create_backend_app("ec2")
test_client = backend.test_client()
res = test_client.get(
"/?Action=DescribeRouteTables&RouteTableId.1=rtb-unknown",
headers={"Host": "ec2.us-east-1.amazonaws.com"},
)
res.status_code.should.equal(400)
body = xmltodict.parse(res.data.decode("utf-8"), dict_constructor=dict)
body.should.have.key("Response")
body["Response"].should.have.key("Errors")
body["Response"]["Errors"].should.have.key("Error")
error = body["Response"]["Errors"]["Error"]
error["Code"].should.equal("InvalidRouteTableID.NotFound")
error["Message"].should.equal("The routeTable ID 'rtb-unknown' does not exist")