moto/tests/test_ec2/test_dhcp_options.py
Waldemar Hummer f4f8527955
Merge LocalStack changes into upstream moto (#4082)
* fix OPTIONS requests on non-existing API GW integrations

* add cloudformation models for API Gateway deployments

* bump version

* add backdoor to return CloudWatch metrics

* Updating implementation coverage

* Updating implementation coverage

* add cloudformation models for API Gateway deployments

* Updating implementation coverage

* Updating implementation coverage

* Implemented get-caller-identity returning real data depending on the access key used.

* bump version

* minor fixes

* fix Number data_type for SQS message attribute

* fix handling of encoding errors

* bump version

* make CF stack queryable before starting to initialize its resources

* bump version

* fix integration_method for API GW method integrations

* fix undefined status in CF FakeStack

* Fix apigateway issues with terraform v0.12.21
* resource_methods -> add handle for "DELETE" method
* integrations -> fix issue that "httpMethod" wasn't included in body request (this value was set as the value from refer method resource)

* bump version

* Fix setting http method for API gateway integrations (#6)

* bump version

* remove duplicate methods

* add storage class to S3 Key when completing multipart upload (#7)

* fix SQS performance issues; bump version

* add pagination to SecretsManager list-secrets (#9)

* fix default parameter groups in RDS

* fix adding S3 metadata headers with names containing dots (#13)

* Updating implementation coverage

* Updating implementation coverage

* add cloudformation models for API Gateway deployments

* Updating implementation coverage

* Updating implementation coverage

* Implemented get-caller-identity returning real data depending on the access key used.

* make CF stack queryable before starting to initialize its resources

* bump version

* remove duplicate methods

* fix adding S3 metadata headers with names containing dots (#13)

* Update amis.json to support EKS AMI mocks (#15)

* fix PascalCase for boolean value in ListMultipartUploads response (#17); fix _get_multi_param to parse nested list/dict query params

* determine non-zero container exit code in Batch API

* support filtering by dimensions in CW get_metric_statistics

* fix storing attributes for ELBv2 Route entities; API GW refactorings for TF tests

* add missing fields for API GW resources

* fix error messages for Route53 (TF-compat)

* various fixes for IAM resources (tf-compat)

* minor fixes for API GW models (tf-compat)

* minor fixes for API GW responses (tf-compat)

* add s3 exception for bucket notification filter rule validation

* change the way RESTErrors generate the response body and content-type header

* fix lint errors and disable "black" syntax enforcement

* remove return type hint in RESTError.get_body

* add RESTError XML template for IAM exceptions

* add support for API GW minimumCompressionSize

* fix casing getting PrivateDnsEnabled API GW attribute

* minor fixes for error responses

* fix escaping special chars for IAM role descriptions (tf-compat)

* minor fixes and tagging support for API GW and ELB v2 (tf-compat)

* Merge branch 'master' into localstack

* add "AlarmRule" attribute to enable support for composite CloudWatch metrics

* fix recursive parsing of complex/nested query params

* bump version

* add API to delete S3 website configurations (#18)

* use dict copy to allow parallelism and avoid concurrent modification exceptions in S3

* fix precondition check for etags in S3 (#19)

* minor fix for user filtering in Cognito

* fix API Gateway error response; avoid returning empty response templates (tf-compat)

* support tags and tracingEnabled attribute for API GW stages

* fix boolean value in S3 encryption response (#20)

* fix connection arn structure

* fix api destination arn structure

* black format

* release 2.0.3.37

* fix s3 exception tests

see botocore/parsers.py:1002 where RequestId is removed from parsed

* remove python 2 from build action

* add test failure annotations in build action

* fix events test arn comparisons

* fix s3 encryption response test

* return default value "0" if EC2 availableIpAddressCount is empty

* fix extracting SecurityGroupIds for EC2 VPC endpoints

* support deleting/updating API Gateway DomainNames

* fix(events): Return empty string instead of null when no pattern is specified in EventPattern (tf-compat) (#22)

* fix logic and revert CF changes to get tests running again (#21)

* add support for EC2 customer gateway API (#25)

* add support for EC2 Transit Gateway APIs (#24)

* feat(logs): add `kmsKeyId` into `LogGroup` entity (#23)

* minor change in ELBv2 logic to fix tests

* feat(events): add APIs to describe and delete CloudWatch Events connections (#26)

* add support for EC2 transit gateway route tables (#27)

* pass transit gateway route table ID in Describe API, minor refactoring (#29)

* add support for EC2 Transit Gateway Routes (#28)

* fix region on ACM certificate import (#31)

* add support for EC2 transit gateway attachments (#30)

* add support for EC2 Transit Gateway VPN attachments (#32)

* fix account ID for logs API

* add support for DeleteOrganization API

* feat(events): store raw filter representation for CloudWatch events patterns (tf-compat) (#36)

* feat(events): add support to describe/update/delete CloudWatch API destinations (#35)

* add Cognito UpdateIdentityPool, CW Logs PutResourcePolicy

* feat(events): add support for tags in EventBus API (#38)

* fix parameter validation for Batch compute environments (tf-compat)

* revert merge conflicts in IMPLEMENTATION_COVERAGE.md

* format code using black

* restore original README; re-enable and fix CloudFormation tests

* restore tests and old logic for CF stack parameters from SSM

* parameterize RequestId/RequestID in response messages and revert related test changes

* undo LocalStack-specific adaptations

* minor fix

* Update CodeCov config to reflect removal of Py2

* undo change related to CW metric filtering; add additional test for CW metric statistics with dimensions

* Terraform - Extend whitelist of running tests

Co-authored-by: acsbendi <acsbendi28@gmail.com>
Co-authored-by: Phan Duong <duongpv@outlook.com>
Co-authored-by: Thomas Rausch <thomas@thrau.at>
Co-authored-by: Macwan Nevil <macnev2013@gmail.com>
Co-authored-by: Dominik Schubert <dominik.schubert91@gmail.com>
Co-authored-by: Gonzalo Saad <saad.gonzalo.ale@gmail.com>
Co-authored-by: Mohit Alonja <monty16597@users.noreply.github.com>
Co-authored-by: Miguel Gagliardo <migag9@gmail.com>
Co-authored-by: Bert Blommers <info@bertblommers.nl>
2021-07-26 15:21:17 +01:00

336 lines
11 KiB
Python

from __future__ import unicode_literals
import pytest
import boto3
import boto
from boto.exception import EC2ResponseError
import sure # noqa
from moto import mock_ec2, mock_ec2_deprecated
SAMPLE_DOMAIN_NAME = "example.com"
SAMPLE_NAME_SERVERS = ["10.0.0.6", "10.0.0.7"]
@mock_ec2_deprecated
def test_dhcp_options_associate():
"""associate dhcp option"""
conn = boto.connect_vpc("the_key", "the_secret")
dhcp_options = conn.create_dhcp_options(SAMPLE_DOMAIN_NAME, SAMPLE_NAME_SERVERS)
vpc = conn.create_vpc("10.0.0.0/16")
rval = conn.associate_dhcp_options(dhcp_options.id, vpc.id)
rval.should.be.equal(True)
@mock_ec2_deprecated
def test_dhcp_options_associate_invalid_dhcp_id():
"""associate dhcp option bad dhcp options id"""
conn = boto.connect_vpc("the_key", "the_secret")
vpc = conn.create_vpc("10.0.0.0/16")
with pytest.raises(EC2ResponseError) as cm:
conn.associate_dhcp_options("foo", vpc.id)
cm.value.code.should.equal("InvalidDhcpOptionID.NotFound")
cm.value.status.should.equal(400)
cm.value.request_id.should_not.be.none
@mock_ec2_deprecated
def test_dhcp_options_associate_invalid_vpc_id():
"""associate dhcp option invalid vpc id"""
conn = boto.connect_vpc("the_key", "the_secret")
dhcp_options = conn.create_dhcp_options(SAMPLE_DOMAIN_NAME, SAMPLE_NAME_SERVERS)
with pytest.raises(EC2ResponseError) as cm:
conn.associate_dhcp_options(dhcp_options.id, "foo")
cm.value.code.should.equal("InvalidVpcID.NotFound")
cm.value.status.should.equal(400)
cm.value.request_id.should_not.be.none
@mock_ec2_deprecated
def test_dhcp_options_delete_with_vpc():
"""Test deletion of dhcp options with vpc"""
conn = boto.connect_vpc("the_key", "the_secret")
dhcp_options = conn.create_dhcp_options(SAMPLE_DOMAIN_NAME, SAMPLE_NAME_SERVERS)
dhcp_options_id = dhcp_options.id
vpc = conn.create_vpc("10.0.0.0/16")
rval = conn.associate_dhcp_options(dhcp_options_id, vpc.id)
rval.should.be.equal(True)
with pytest.raises(EC2ResponseError) as cm:
conn.delete_dhcp_options(dhcp_options_id)
cm.value.code.should.equal("DependencyViolation")
cm.value.status.should.equal(400)
cm.value.request_id.should_not.be.none
vpc.delete()
with pytest.raises(EC2ResponseError) as cm:
conn.get_all_dhcp_options([dhcp_options_id])
cm.value.code.should.equal("InvalidDhcpOptionID.NotFound")
cm.value.status.should.equal(400)
cm.value.request_id.should_not.be.none
@mock_ec2_deprecated
def test_create_dhcp_options():
"""Create most basic dhcp option"""
conn = boto.connect_vpc("the_key", "the_secret")
dhcp_option = conn.create_dhcp_options(SAMPLE_DOMAIN_NAME, SAMPLE_NAME_SERVERS)
dhcp_option.options["domain-name"][0].should.be.equal(SAMPLE_DOMAIN_NAME)
dhcp_option.options["domain-name-servers"][0].should.be.equal(
SAMPLE_NAME_SERVERS[0]
)
dhcp_option.options["domain-name-servers"][1].should.be.equal(
SAMPLE_NAME_SERVERS[1]
)
@mock_ec2_deprecated
def test_create_dhcp_options_invalid_options():
"""Create invalid dhcp options"""
conn = boto.connect_vpc("the_key", "the_secret")
servers = ["f", "f", "f", "f", "f"]
with pytest.raises(EC2ResponseError) as cm:
conn.create_dhcp_options(ntp_servers=servers)
cm.value.code.should.equal("InvalidParameterValue")
cm.value.status.should.equal(400)
cm.value.request_id.should_not.be.none
with pytest.raises(EC2ResponseError) as cm:
conn.create_dhcp_options(netbios_node_type="0")
cm.value.code.should.equal("InvalidParameterValue")
cm.value.status.should.equal(400)
cm.value.request_id.should_not.be.none
@mock_ec2_deprecated
def test_describe_dhcp_options():
"""Test dhcp options lookup by id"""
conn = boto.connect_vpc("the_key", "the_secret")
dhcp_option = conn.create_dhcp_options()
dhcp_options = conn.get_all_dhcp_options([dhcp_option.id])
dhcp_options.should.be.length_of(1)
dhcp_options = conn.get_all_dhcp_options()
dhcp_options.should.be.length_of(1)
@mock_ec2_deprecated
def test_describe_dhcp_options_invalid_id():
"""get error on invalid dhcp_option_id lookup"""
conn = boto.connect_vpc("the_key", "the_secret")
with pytest.raises(EC2ResponseError) as cm:
conn.get_all_dhcp_options(["1"])
cm.value.code.should.equal("InvalidDhcpOptionID.NotFound")
cm.value.status.should.equal(400)
cm.value.request_id.should_not.be.none
@mock_ec2_deprecated
def test_delete_dhcp_options():
"""delete dhcp option"""
conn = boto.connect_vpc("the_key", "the_secret")
dhcp_option = conn.create_dhcp_options()
dhcp_options = conn.get_all_dhcp_options([dhcp_option.id])
dhcp_options.should.be.length_of(1)
conn.delete_dhcp_options(dhcp_option.id) # .should.be.equal(True)
with pytest.raises(EC2ResponseError) as cm:
conn.get_all_dhcp_options([dhcp_option.id])
cm.value.code.should.equal("InvalidDhcpOptionID.NotFound")
cm.value.status.should.equal(400)
cm.value.request_id.should_not.be.none
@mock_ec2_deprecated
def test_delete_dhcp_options_invalid_id():
conn = boto.connect_vpc("the_key", "the_secret")
conn.create_dhcp_options()
with pytest.raises(EC2ResponseError) as cm:
conn.delete_dhcp_options("dopt-abcd1234")
cm.value.code.should.equal("InvalidDhcpOptionID.NotFound")
cm.value.status.should.equal(400)
cm.value.request_id.should_not.be.none
@mock_ec2_deprecated
def test_delete_dhcp_options_malformed_id():
conn = boto.connect_vpc("the_key", "the_secret")
conn.create_dhcp_options()
with pytest.raises(EC2ResponseError) as cm:
conn.delete_dhcp_options("foo-abcd1234")
cm.value.code.should.equal("InvalidDhcpOptionsId.Malformed")
cm.value.status.should.equal(400)
cm.value.request_id.should_not.be.none
@mock_ec2_deprecated
def test_dhcp_tagging():
conn = boto.connect_vpc("the_key", "the_secret")
dhcp_option = conn.create_dhcp_options()
dhcp_option.add_tag("a key", "some value")
tag = conn.get_all_tags()[0]
tag.name.should.equal("a key")
tag.value.should.equal("some value")
# Refresh the DHCP options
dhcp_option = conn.get_all_dhcp_options()[0]
dhcp_option.tags.should.have.length_of(1)
dhcp_option.tags["a key"].should.equal("some value")
@mock_ec2_deprecated
def test_dhcp_options_get_by_tag():
conn = boto.connect_vpc("the_key", "the_secret")
dhcp1 = conn.create_dhcp_options("example.com", ["10.0.10.2"])
dhcp1.add_tag("Name", "TestDhcpOptions1")
dhcp1.add_tag("test-tag", "test-value")
dhcp2 = conn.create_dhcp_options("example.com", ["10.0.20.2"])
dhcp2.add_tag("Name", "TestDhcpOptions2")
dhcp2.add_tag("test-tag", "test-value")
filters = {"tag:Name": "TestDhcpOptions1", "tag:test-tag": "test-value"}
dhcp_options_sets = conn.get_all_dhcp_options(filters=filters)
dhcp_options_sets.should.have.length_of(1)
dhcp_options_sets[0].options["domain-name"][0].should.be.equal("example.com")
dhcp_options_sets[0].options["domain-name-servers"][0].should.be.equal("10.0.10.2")
dhcp_options_sets[0].tags["Name"].should.equal("TestDhcpOptions1")
dhcp_options_sets[0].tags["test-tag"].should.equal("test-value")
filters = {"tag:Name": "TestDhcpOptions2", "tag:test-tag": "test-value"}
dhcp_options_sets = conn.get_all_dhcp_options(filters=filters)
dhcp_options_sets.should.have.length_of(1)
dhcp_options_sets[0].options["domain-name"][0].should.be.equal("example.com")
dhcp_options_sets[0].options["domain-name-servers"][0].should.be.equal("10.0.20.2")
dhcp_options_sets[0].tags["Name"].should.equal("TestDhcpOptions2")
dhcp_options_sets[0].tags["test-tag"].should.equal("test-value")
filters = {"tag:test-tag": "test-value"}
dhcp_options_sets = conn.get_all_dhcp_options(filters=filters)
dhcp_options_sets.should.have.length_of(2)
@mock_ec2_deprecated
def test_dhcp_options_get_by_id():
conn = boto.connect_vpc("the_key", "the_secret")
dhcp1 = conn.create_dhcp_options("test1.com", ["10.0.10.2"])
dhcp1.add_tag("Name", "TestDhcpOptions1")
dhcp1.add_tag("test-tag", "test-value")
dhcp1_id = dhcp1.id
dhcp2 = conn.create_dhcp_options("test2.com", ["10.0.20.2"])
dhcp2.add_tag("Name", "TestDhcpOptions2")
dhcp2.add_tag("test-tag", "test-value")
dhcp2_id = dhcp2.id
dhcp_options_sets = conn.get_all_dhcp_options()
dhcp_options_sets.should.have.length_of(2)
dhcp_options_sets = conn.get_all_dhcp_options(filters={"dhcp-options-id": dhcp1_id})
dhcp_options_sets.should.have.length_of(1)
dhcp_options_sets[0].options["domain-name"][0].should.be.equal("test1.com")
dhcp_options_sets[0].options["domain-name-servers"][0].should.be.equal("10.0.10.2")
dhcp_options_sets = conn.get_all_dhcp_options(filters={"dhcp-options-id": dhcp2_id})
dhcp_options_sets.should.have.length_of(1)
dhcp_options_sets[0].options["domain-name"][0].should.be.equal("test2.com")
dhcp_options_sets[0].options["domain-name-servers"][0].should.be.equal("10.0.20.2")
@mock_ec2
def test_dhcp_options_get_by_value_filter():
ec2 = boto3.resource("ec2", region_name="us-west-1")
ec2.create_dhcp_options(
DhcpConfigurations=[
{"Key": "domain-name", "Values": ["example.com"]},
{"Key": "domain-name-servers", "Values": ["10.0.10.2"]},
]
)
ec2.create_dhcp_options(
DhcpConfigurations=[
{"Key": "domain-name", "Values": ["example.com"]},
{"Key": "domain-name-servers", "Values": ["10.0.20.2"]},
]
)
ec2.create_dhcp_options(
DhcpConfigurations=[
{"Key": "domain-name", "Values": ["example.com"]},
{"Key": "domain-name-servers", "Values": ["10.0.30.2"]},
]
)
filters = [{"Name": "value", "Values": ["10.0.10.2"]}]
dhcp_options_sets = list(ec2.dhcp_options_sets.filter(Filters=filters))
dhcp_options_sets.should.have.length_of(1)
@mock_ec2
def test_dhcp_options_get_by_key_filter():
ec2 = boto3.resource("ec2", region_name="us-west-1")
ec2.create_dhcp_options(
DhcpConfigurations=[
{"Key": "domain-name", "Values": ["example.com"]},
{"Key": "domain-name-servers", "Values": ["10.0.10.2"]},
]
)
ec2.create_dhcp_options(
DhcpConfigurations=[
{"Key": "domain-name", "Values": ["example.com"]},
{"Key": "domain-name-servers", "Values": ["10.0.20.2"]},
]
)
ec2.create_dhcp_options(
DhcpConfigurations=[
{"Key": "domain-name", "Values": ["example.com"]},
{"Key": "domain-name-servers", "Values": ["10.0.30.2"]},
]
)
filters = [{"Name": "key", "Values": ["domain-name"]}]
dhcp_options_sets = list(ec2.dhcp_options_sets.filter(Filters=filters))
dhcp_options_sets.should.have.length_of(3)
@mock_ec2_deprecated
def test_dhcp_options_get_by_invalid_filter():
conn = boto.connect_vpc("the_key", "the_secret")
conn.create_dhcp_options(SAMPLE_DOMAIN_NAME, SAMPLE_NAME_SERVERS)
filters = {"invalid-filter": "invalid-value"}
conn.get_all_dhcp_options.when.called_with(filters=filters).should.throw(
NotImplementedError
)