1import pytest 2from botocore.exceptions import ClientError 3 4import boto3 5import boto 6from boto.exception import EC2ResponseError 7 8import sure # noqa # pylint: disable=unused-import 9import random 10 11from moto import mock_ec2, mock_ec2_deprecated 12from uuid import uuid4 13from .test_tags import retrieve_all_tagged 14 15SAMPLE_DOMAIN_NAME = "example.com" 16SAMPLE_NAME_SERVERS = ["10.0.0.6", "10.0.0.7"] 17 18 19# Has boto3 equivalent 20@mock_ec2_deprecated 21def test_vpcs(): 22 conn = boto.connect_vpc("the_key", "the_secret") 23 vpc = conn.create_vpc("10.0.0.0/16") 24 vpc.cidr_block.should.equal("10.0.0.0/16") 25 26 all_vpcs = conn.get_all_vpcs() 27 all_vpcs.should.have.length_of(2) 28 29 vpc.delete() 30 31 all_vpcs = conn.get_all_vpcs() 32 all_vpcs.should.have.length_of(1) 33 34 with pytest.raises(EC2ResponseError) as cm: 35 conn.delete_vpc("vpc-1234abcd") 36 cm.value.code.should.equal("InvalidVpcID.NotFound") 37 cm.value.status.should.equal(400) 38 cm.value.request_id.should_not.be.none 39 40 41@mock_ec2 42def test_create_and_delete_vpc(): 43 ec2 = boto3.resource("ec2", region_name="eu-north-1") 44 client = boto3.client("ec2", region_name="eu-north-1") 45 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 46 vpc.cidr_block.should.equal("10.0.0.0/16") 47 48 all_vpcs = retrieve_all_vpcs(client) 49 [v["VpcId"] for v in all_vpcs].should.contain(vpc.id) 50 51 vpc.delete() 52 53 all_vpcs = retrieve_all_vpcs(client) 54 [v["VpcId"] for v in all_vpcs].shouldnt.contain(vpc.id) 55 56 with pytest.raises(ClientError) as ex: 57 client.delete_vpc(VpcId="vpc-1234abcd") 58 ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 59 ex.value.response["ResponseMetadata"].should.have.key("RequestId") 60 ex.value.response["Error"]["Code"].should.equal("InvalidVpcID.NotFound") 61 62 63# Has boto3 equivalent 64@mock_ec2_deprecated 65def test_vpc_defaults(): 66 conn = boto.connect_vpc("the_key", "the_secret") 67 vpc = conn.create_vpc("10.0.0.0/16") 68 69 conn.get_all_vpcs().should.have.length_of(2) 70 conn.get_all_route_tables().should.have.length_of(2) 71 conn.get_all_security_groups(filters={"vpc-id": [vpc.id]}).should.have.length_of(1) 72 73 vpc.delete() 74 75 conn.get_all_vpcs().should.have.length_of(1) 76 conn.get_all_route_tables().should.have.length_of(1) 77 conn.get_all_security_groups(filters={"vpc-id": [vpc.id]}).should.have.length_of(0) 78 79 80@mock_ec2 81def test_vpc_defaults_boto3(): 82 ec2 = boto3.resource("ec2", region_name="eu-north-1") 83 client = boto3.client("ec2", region_name="eu-north-1") 84 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 85 86 filters = [{"Name": "vpc-id", "Values": [vpc.id]}] 87 88 client.describe_route_tables(Filters=filters)["RouteTables"].should.have.length_of( 89 1 90 ) 91 client.describe_security_groups(Filters=filters)[ 92 "SecurityGroups" 93 ].should.have.length_of(1) 94 95 vpc.delete() 96 97 client.describe_route_tables(Filters=filters)["RouteTables"].should.have.length_of( 98 0 99 ) 100 client.describe_security_groups(Filters=filters)[ 101 "SecurityGroups" 102 ].should.have.length_of(0) 103 104 105# Has boto3 equivalent 106@mock_ec2_deprecated 107def test_vpc_isdefault_filter(): 108 conn = boto.connect_vpc("the_key", "the_secret") 109 vpc = conn.create_vpc("10.0.0.0/16") 110 conn.get_all_vpcs(filters={"isDefault": "true"}).should.have.length_of(1) 111 vpc.delete() 112 conn.get_all_vpcs(filters={"isDefault": "true"}).should.have.length_of(1) 113 114 115@mock_ec2 116def test_vpc_isdefault_filter_boto3(): 117 ec2 = boto3.resource("ec2", region_name="eu-west-1") 118 client = boto3.client("ec2", region_name="eu-west-1") 119 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 120 client.describe_vpcs(Filters=[{"Name": "isDefault", "Values": ["true"]}])[ 121 "Vpcs" 122 ].should.have.length_of(1) 123 124 vpc.delete() 125 126 client.describe_vpcs(Filters=[{"Name": "isDefault", "Values": ["true"]}])[ 127 "Vpcs" 128 ].should.have.length_of(1) 129 130 131# Has boto3 equivalent 132@mock_ec2_deprecated 133def test_multiple_vpcs_default_filter(): 134 conn = boto.connect_vpc("the_key", "the_secret") 135 conn.create_vpc("10.8.0.0/16") 136 conn.create_vpc("10.0.0.0/16") 137 conn.create_vpc("192.168.0.0/16") 138 conn.get_all_vpcs().should.have.length_of(4) 139 vpc = conn.get_all_vpcs(filters={"isDefault": "true"}) 140 vpc.should.have.length_of(1) 141 vpc[0].cidr_block.should.equal("172.31.0.0/16") 142 143 144@mock_ec2 145def test_multiple_vpcs_default_filter_boto3(): 146 ec2 = boto3.resource("ec2", region_name="eu-west-1") 147 client = boto3.client("ec2", region_name="eu-west-1") 148 ec2.create_vpc(CidrBlock="10.8.0.0/16") 149 ec2.create_vpc(CidrBlock="10.0.0.0/16") 150 ec2.create_vpc(CidrBlock="192.168.0.0/16") 151 default_vpcs = retrieve_all_vpcs( 152 client, [{"Name": "isDefault", "Values": ["true"]}] 153 ) 154 [v["CidrBlock"] for v in default_vpcs].should.contain("172.31.0.0/16") 155 156 157# Has boto3 equivalent 158@mock_ec2_deprecated 159def test_vpc_state_available_filter(): 160 conn = boto.connect_vpc("the_key", "the_secret") 161 vpc = conn.create_vpc("10.0.0.0/16") 162 conn.create_vpc("10.1.0.0/16") 163 conn.get_all_vpcs(filters={"state": "available"}).should.have.length_of(3) 164 vpc.delete() 165 conn.get_all_vpcs(filters={"state": "available"}).should.have.length_of(2) 166 167 168@mock_ec2 169def test_vpc_state_available_filter_boto3(): 170 ec2 = boto3.resource("ec2", region_name="eu-west-1") 171 client = boto3.client("ec2", region_name="eu-west-1") 172 vpc1 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 173 vpc2 = ec2.create_vpc(CidrBlock="10.1.0.0/16") 174 175 available = retrieve_all_vpcs(client, [{"Name": "state", "Values": ["available"]}]) 176 [v["VpcId"] for v in available].should.contain(vpc1.id) 177 [v["VpcId"] for v in available].should.contain(vpc2.id) 178 179 vpc1.delete() 180 181 available = retrieve_all_vpcs(client, [{"Name": "state", "Values": ["available"]}]) 182 [v["VpcId"] for v in available].shouldnt.contain(vpc1.id) 183 [v["VpcId"] for v in available].should.contain(vpc2.id) 184 185 186def retrieve_all_vpcs(client, filters=[]): # pylint: disable=W0102 187 resp = client.describe_vpcs(Filters=filters) 188 all_vpcs = resp["Vpcs"] 189 token = resp.get("NextToken") 190 while token: 191 resp = client.describe_vpcs(Filters=filters, NextToken=token) 192 all_vpcs.extend(resp["Vpcs"]) 193 token = resp.get("NextToken") 194 return all_vpcs 195 196 197# Has boto3 equivalent 198@mock_ec2_deprecated 199def test_vpc_tagging(): 200 conn = boto.connect_vpc() 201 vpc = conn.create_vpc("10.0.0.0/16") 202 203 vpc.add_tag("a key", "some value") 204 tag = conn.get_all_tags()[0] 205 tag.name.should.equal("a key") 206 tag.value.should.equal("some value") 207 208 # Refresh the vpc 209 vpc = conn.get_all_vpcs(vpc_ids=[vpc.id])[0] 210 vpc.tags.should.have.length_of(1) 211 vpc.tags["a key"].should.equal("some value") 212 213 214@mock_ec2 215def test_vpc_tagging_boto3(): 216 ec2 = boto3.resource("ec2", region_name="eu-west-1") 217 client = boto3.client("ec2", region_name="eu-west-1") 218 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 219 220 vpc.create_tags(Tags=[{"Key": "a key", "Value": "some value"}]) 221 222 all_tags = retrieve_all_tagged(client) 223 ours = [t for t in all_tags if t["ResourceId"] == vpc.id][0] 224 ours.should.have.key("Key").equal("a key") 225 ours.should.have.key("Value").equal("some value") 226 227 # Refresh the vpc 228 vpc = client.describe_vpcs(VpcIds=[vpc.id])["Vpcs"][0] 229 vpc["Tags"].should.equal([{"Key": "a key", "Value": "some value"}]) 230 231 232# Has boto3 equivalent 233@mock_ec2_deprecated 234def test_vpc_get_by_id(): 235 conn = boto.connect_vpc() 236 vpc1 = conn.create_vpc("10.0.0.0/16") 237 vpc2 = conn.create_vpc("10.0.0.0/16") 238 conn.create_vpc("10.0.0.0/16") 239 240 vpcs = conn.get_all_vpcs(vpc_ids=[vpc1.id, vpc2.id]) 241 vpcs.should.have.length_of(2) 242 vpc_ids = tuple(map(lambda v: v.id, vpcs)) 243 vpc1.id.should.be.within(vpc_ids) 244 vpc2.id.should.be.within(vpc_ids) 245 246 with pytest.raises(EC2ResponseError) as cm: 247 conn.get_all_vpcs(vpc_ids=["vpc-does_not_exist"]) 248 cm.value.code.should.equal("InvalidVpcID.NotFound") 249 cm.value.status.should.equal(400) 250 cm.value.request_id.should_not.be.none 251 252 253@mock_ec2 254def test_vpc_get_by_id_boto3(): 255 ec2 = boto3.resource("ec2", region_name="eu-west-1") 256 client = boto3.client("ec2", region_name="eu-west-1") 257 vpc1 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 258 vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 259 ec2.create_vpc(CidrBlock="10.0.0.0/16") 260 261 vpcs = client.describe_vpcs(VpcIds=[vpc1.id, vpc2.id])["Vpcs"] 262 vpcs.should.have.length_of(2) 263 vpc_ids = tuple(map(lambda v: v["VpcId"], vpcs)) 264 vpc1.id.should.be.within(vpc_ids) 265 vpc2.id.should.be.within(vpc_ids) 266 267 with pytest.raises(ClientError) as ex: 268 client.describe_vpcs(VpcIds=["vpc-does_not_exist"]) 269 ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 270 ex.value.response["ResponseMetadata"].should.have.key("RequestId") 271 ex.value.response["Error"]["Code"].should.equal("InvalidVpcID.NotFound") 272 273 274# Has boto3 equivalent 275@mock_ec2_deprecated 276def test_vpc_get_by_cidr_block(): 277 conn = boto.connect_vpc() 278 vpc1 = conn.create_vpc("10.0.0.0/16") 279 vpc2 = conn.create_vpc("10.0.0.0/16") 280 conn.create_vpc("10.0.0.0/24") 281 282 vpcs = conn.get_all_vpcs(filters={"cidr": "10.0.0.0/16"}) 283 vpcs.should.have.length_of(2) 284 vpc_ids = tuple(map(lambda v: v.id, vpcs)) 285 vpc1.id.should.be.within(vpc_ids) 286 vpc2.id.should.be.within(vpc_ids) 287 288 289@mock_ec2 290def test_vpc_get_by_cidr_block_boto3(): 291 ec2 = boto3.resource("ec2", region_name="eu-west-1") 292 client = boto3.client("ec2", region_name="eu-west-1") 293 random_ip = ".".join(map(str, (random.randint(0, 99) for _ in range(4)))) 294 random_cidr = f"{random_ip}/16" 295 vpc1 = ec2.create_vpc(CidrBlock=random_cidr) 296 vpc2 = ec2.create_vpc(CidrBlock=random_cidr) 297 ec2.create_vpc(CidrBlock="10.0.0.0/24") 298 299 vpcs = client.describe_vpcs(Filters=[{"Name": "cidr", "Values": [random_cidr]}])[ 300 "Vpcs" 301 ] 302 set([vpc["VpcId"] for vpc in vpcs]).should.equal(set([vpc1.id, vpc2.id])) 303 304 305# Has boto3 equivalent 306@mock_ec2_deprecated 307def test_vpc_get_by_dhcp_options_id(): 308 conn = boto.connect_vpc() 309 dhcp_options = conn.create_dhcp_options(SAMPLE_DOMAIN_NAME, SAMPLE_NAME_SERVERS) 310 vpc1 = conn.create_vpc("10.0.0.0/16") 311 vpc2 = conn.create_vpc("10.0.0.0/16") 312 conn.create_vpc("10.0.0.0/24") 313 314 conn.associate_dhcp_options(dhcp_options.id, vpc1.id) 315 conn.associate_dhcp_options(dhcp_options.id, vpc2.id) 316 317 vpcs = conn.get_all_vpcs(filters={"dhcp-options-id": dhcp_options.id}) 318 vpcs.should.have.length_of(2) 319 vpc_ids = tuple(map(lambda v: v.id, vpcs)) 320 vpc1.id.should.be.within(vpc_ids) 321 vpc2.id.should.be.within(vpc_ids) 322 323 324@mock_ec2 325def test_vpc_get_by_dhcp_options_id_boto3(): 326 ec2 = boto3.resource("ec2", region_name="us-east-1") 327 client = boto3.client("ec2", region_name="us-east-1") 328 dhcp_options = ec2.create_dhcp_options( 329 DhcpConfigurations=[ 330 {"Key": "domain-name", "Values": [SAMPLE_DOMAIN_NAME]}, 331 {"Key": "domain-name-servers", "Values": SAMPLE_NAME_SERVERS}, 332 ] 333 ) 334 vpc1 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 335 vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 336 ec2.create_vpc(CidrBlock="10.0.0.0/24") 337 338 client.associate_dhcp_options(DhcpOptionsId=dhcp_options.id, VpcId=vpc1.id) 339 client.associate_dhcp_options(DhcpOptionsId=dhcp_options.id, VpcId=vpc2.id) 340 341 vpcs = client.describe_vpcs( 342 Filters=[{"Name": "dhcp-options-id", "Values": [dhcp_options.id]}] 343 )["Vpcs"] 344 vpcs.should.have.length_of(2) 345 vpc_ids = tuple(map(lambda v: v["VpcId"], vpcs)) 346 vpc1.id.should.be.within(vpc_ids) 347 vpc2.id.should.be.within(vpc_ids) 348 349 350# Has boto3 equivalent 351@mock_ec2_deprecated 352def test_vpc_get_by_tag(): 353 conn = boto.connect_vpc() 354 vpc1 = conn.create_vpc("10.0.0.0/16") 355 vpc2 = conn.create_vpc("10.0.0.0/16") 356 vpc3 = conn.create_vpc("10.0.0.0/24") 357 358 vpc1.add_tag("Name", "TestVPC") 359 vpc2.add_tag("Name", "TestVPC") 360 vpc3.add_tag("Name", "TestVPC2") 361 362 vpcs = conn.get_all_vpcs(filters={"tag:Name": "TestVPC"}) 363 vpcs.should.have.length_of(2) 364 vpc_ids = tuple(map(lambda v: v.id, vpcs)) 365 vpc1.id.should.be.within(vpc_ids) 366 vpc2.id.should.be.within(vpc_ids) 367 368 369@mock_ec2 370def test_vpc_get_by_tag_boto3(): 371 ec2 = boto3.resource("ec2", region_name="us-east-1") 372 client = boto3.client("ec2", region_name="us-east-1") 373 vpc1 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 374 vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 375 vpc3 = ec2.create_vpc(CidrBlock="10.0.0.0/24") 376 377 value1 = str(uuid4()) 378 vpc1.create_tags(Tags=[{"Key": "Name", "Value": value1}]) 379 vpc2.create_tags(Tags=[{"Key": "Name", "Value": value1}]) 380 vpc3.create_tags(Tags=[{"Key": "Name", "Value": "TestVPC2"}]) 381 382 vpcs = client.describe_vpcs(Filters=[{"Name": "tag:Name", "Values": [value1]}])[ 383 "Vpcs" 384 ] 385 vpcs.should.have.length_of(2) 386 set([vpc["VpcId"] for vpc in vpcs]).should.equal(set([vpc1.id, vpc2.id])) 387 388 389# Has boto3 equivalent 390@mock_ec2_deprecated 391def test_vpc_get_by_tag_key_superset(): 392 conn = boto.connect_vpc() 393 vpc1 = conn.create_vpc("10.0.0.0/16") 394 vpc2 = conn.create_vpc("10.0.0.0/16") 395 vpc3 = conn.create_vpc("10.0.0.0/24") 396 397 vpc1.add_tag("Name", "TestVPC") 398 vpc1.add_tag("Key", "TestVPC2") 399 vpc2.add_tag("Name", "TestVPC") 400 vpc2.add_tag("Key", "TestVPC2") 401 vpc3.add_tag("Key", "TestVPC2") 402 403 vpcs = conn.get_all_vpcs(filters={"tag-key": "Name"}) 404 vpcs.should.have.length_of(2) 405 vpc_ids = tuple(map(lambda v: v.id, vpcs)) 406 vpc1.id.should.be.within(vpc_ids) 407 vpc2.id.should.be.within(vpc_ids) 408 409 410@mock_ec2 411def test_vpc_get_by_tag_key_superset_boto3(): 412 ec2 = boto3.resource("ec2", region_name="us-east-1") 413 client = boto3.client("ec2", region_name="us-east-1") 414 vpc1 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 415 vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 416 vpc3 = ec2.create_vpc(CidrBlock="10.0.0.0/24") 417 418 tag_key = str(uuid4())[0:6] 419 vpc1.create_tags(Tags=[{"Key": tag_key, "Value": "TestVPC"}]) 420 vpc1.create_tags(Tags=[{"Key": "Key", "Value": "TestVPC2"}]) 421 vpc2.create_tags(Tags=[{"Key": tag_key, "Value": "TestVPC"}]) 422 vpc2.create_tags(Tags=[{"Key": "Key", "Value": "TestVPC2"}]) 423 vpc3.create_tags(Tags=[{"Key": "Key", "Value": "TestVPC2"}]) 424 425 vpcs = client.describe_vpcs(Filters=[{"Name": "tag-key", "Values": [tag_key]}])[ 426 "Vpcs" 427 ] 428 vpcs.should.have.length_of(2) 429 set([vpc["VpcId"] for vpc in vpcs]).should.equal(set([vpc1.id, vpc2.id])) 430 431 432# Has boto3 equivalent 433@mock_ec2_deprecated 434def test_vpc_get_by_tag_key_subset(): 435 conn = boto.connect_vpc() 436 vpc1 = conn.create_vpc("10.0.0.0/16") 437 vpc2 = conn.create_vpc("10.0.0.0/16") 438 vpc3 = conn.create_vpc("10.0.0.0/24") 439 440 vpc1.add_tag("Name", "TestVPC") 441 vpc1.add_tag("Key", "TestVPC2") 442 vpc2.add_tag("Name", "TestVPC") 443 vpc2.add_tag("Key", "TestVPC2") 444 vpc3.add_tag("Test", "TestVPC2") 445 446 vpcs = conn.get_all_vpcs(filters={"tag-key": ["Name", "Key"]}) 447 vpcs.should.have.length_of(2) 448 vpc_ids = tuple(map(lambda v: v.id, vpcs)) 449 vpc1.id.should.be.within(vpc_ids) 450 vpc2.id.should.be.within(vpc_ids) 451 452 453@mock_ec2 454def test_vpc_get_by_tag_key_subset_boto3(): 455 ec2 = boto3.resource("ec2", region_name="us-east-1") 456 client = boto3.client("ec2", region_name="us-east-1") 457 vpc1 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 458 vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 459 vpc3 = ec2.create_vpc(CidrBlock="10.0.0.0/24") 460 461 tag_key1 = str(uuid4())[0:6] 462 tag_key2 = str(uuid4())[0:6] 463 vpc1.create_tags(Tags=[{"Key": tag_key1, "Value": "TestVPC"}]) 464 vpc1.create_tags(Tags=[{"Key": tag_key2, "Value": "TestVPC2"}]) 465 vpc2.create_tags(Tags=[{"Key": tag_key1, "Value": "TestVPC"}]) 466 vpc2.create_tags(Tags=[{"Key": tag_key2, "Value": "TestVPC2"}]) 467 vpc3.create_tags(Tags=[{"Key": "Test", "Value": "TestVPC2"}]) 468 469 vpcs = client.describe_vpcs( 470 Filters=[{"Name": "tag-key", "Values": [tag_key1, tag_key2]}] 471 )["Vpcs"] 472 vpcs.should.have.length_of(2) 473 set([vpc["VpcId"] for vpc in vpcs]).should.equal(set([vpc1.id, vpc2.id])) 474 475 476# Has boto3 equivalent 477@mock_ec2_deprecated 478def test_vpc_get_by_tag_value_superset(): 479 conn = boto.connect_vpc() 480 vpc1 = conn.create_vpc("10.0.0.0/16") 481 vpc2 = conn.create_vpc("10.0.0.0/16") 482 vpc3 = conn.create_vpc("10.0.0.0/24") 483 484 vpc1.add_tag("Name", "TestVPC") 485 vpc1.add_tag("Key", "TestVPC2") 486 vpc2.add_tag("Name", "TestVPC") 487 vpc2.add_tag("Key", "TestVPC2") 488 vpc3.add_tag("Key", "TestVPC2") 489 490 vpcs = conn.get_all_vpcs(filters={"tag-value": "TestVPC"}) 491 vpcs.should.have.length_of(2) 492 vpc_ids = tuple(map(lambda v: v.id, vpcs)) 493 vpc1.id.should.be.within(vpc_ids) 494 vpc2.id.should.be.within(vpc_ids) 495 496 497@mock_ec2 498def test_vpc_get_by_tag_value_superset_boto3(): 499 ec2 = boto3.resource("ec2", region_name="us-east-1") 500 client = boto3.client("ec2", region_name="us-east-1") 501 vpc1 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 502 vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 503 vpc3 = ec2.create_vpc(CidrBlock="10.0.0.0/24") 504 505 tag_value = str(uuid4()) 506 vpc1.create_tags(Tags=[{"Key": "Name", "Value": tag_value}]) 507 vpc1.create_tags(Tags=[{"Key": "Key", "Value": "TestVPC2"}]) 508 vpc2.create_tags(Tags=[{"Key": "Name", "Value": tag_value}]) 509 vpc2.create_tags(Tags=[{"Key": "Key", "Value": "TestVPC2"}]) 510 vpc3.create_tags(Tags=[{"Key": "Key", "Value": "TestVPC2"}]) 511 512 vpcs = client.describe_vpcs(Filters=[{"Name": "tag-value", "Values": [tag_value]}])[ 513 "Vpcs" 514 ] 515 vpcs.should.have.length_of(2) 516 set([vpc["VpcId"] for vpc in vpcs]).should.equal(set([vpc1.id, vpc2.id])) 517 518 519# Has boto3 equivalent 520@mock_ec2_deprecated 521def test_vpc_get_by_tag_value_subset(): 522 conn = boto.connect_vpc() 523 vpc1 = conn.create_vpc("10.0.0.0/16") 524 vpc2 = conn.create_vpc("10.0.0.0/16") 525 conn.create_vpc("10.0.0.0/24") 526 527 vpc1.add_tag("Name", "TestVPC") 528 vpc1.add_tag("Key", "TestVPC2") 529 vpc2.add_tag("Name", "TestVPC") 530 vpc2.add_tag("Key", "TestVPC2") 531 532 vpcs = conn.get_all_vpcs(filters={"tag-value": ["TestVPC", "TestVPC2"]}) 533 vpcs.should.have.length_of(2) 534 vpc_ids = tuple(map(lambda v: v.id, vpcs)) 535 vpc1.id.should.be.within(vpc_ids) 536 vpc2.id.should.be.within(vpc_ids) 537 538 539@mock_ec2 540def test_vpc_get_by_tag_value_subset_boto3(): 541 ec2 = boto3.resource("ec2", region_name="us-east-1") 542 client = boto3.client("ec2", region_name="us-east-1") 543 vpc1 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 544 vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 545 ec2.create_vpc(CidrBlock="10.0.0.0/24") 546 547 value1 = str(uuid4())[0:6] 548 value2 = str(uuid4())[0:6] 549 vpc1.create_tags(Tags=[{"Key": "Name", "Value": value1}]) 550 vpc1.create_tags(Tags=[{"Key": "Key", "Value": value2}]) 551 vpc2.create_tags(Tags=[{"Key": "Name", "Value": value1}]) 552 vpc2.create_tags(Tags=[{"Key": "Key", "Value": value2}]) 553 554 vpcs = client.describe_vpcs( 555 Filters=[{"Name": "tag-value", "Values": [value1, value2]}] 556 )["Vpcs"] 557 vpcs.should.have.length_of(2) 558 vpc_ids = tuple(map(lambda v: v["VpcId"], vpcs)) 559 vpc1.id.should.be.within(vpc_ids) 560 vpc2.id.should.be.within(vpc_ids) 561 562 563@mock_ec2 564def test_default_vpc(): 565 ec2 = boto3.resource("ec2", region_name="us-west-1") 566 567 # Create the default VPC 568 default_vpc = list(ec2.vpcs.all())[0] 569 default_vpc.cidr_block.should.equal("172.31.0.0/16") 570 default_vpc.instance_tenancy.should.equal("default") 571 default_vpc.reload() 572 default_vpc.is_default.should.be.ok 573 574 # Test default values for VPC attributes 575 response = default_vpc.describe_attribute(Attribute="enableDnsSupport") 576 attr = response.get("EnableDnsSupport") 577 attr.get("Value").should.be.ok 578 579 response = default_vpc.describe_attribute(Attribute="enableDnsHostnames") 580 attr = response.get("EnableDnsHostnames") 581 attr.get("Value").should.be.ok 582 583 584@mock_ec2 585def test_non_default_vpc(): 586 ec2 = boto3.resource("ec2", region_name="us-west-1") 587 588 # Create the default VPC - this already exists when backend instantiated! 589 # ec2.create_vpc(CidrBlock='172.31.0.0/16') 590 591 # Create the non default VPC 592 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 593 vpc.reload() 594 vpc.is_default.shouldnt.be.ok 595 596 # Test default instance_tenancy 597 vpc.instance_tenancy.should.equal("default") 598 599 # Test default values for VPC attributes 600 response = vpc.describe_attribute(Attribute="enableDnsSupport") 601 attr = response.get("EnableDnsSupport") 602 attr.get("Value").should.be.ok 603 604 response = vpc.describe_attribute(Attribute="enableDnsHostnames") 605 attr = response.get("EnableDnsHostnames") 606 attr.get("Value").shouldnt.be.ok 607 608 # Check Primary CIDR Block Associations 609 cidr_block_association_set = next(iter(vpc.cidr_block_association_set), None) 610 cidr_block_association_set["CidrBlockState"]["State"].should.equal("associated") 611 cidr_block_association_set["CidrBlock"].should.equal(vpc.cidr_block) 612 cidr_block_association_set["AssociationId"].should.contain("vpc-cidr-assoc") 613 614 615@mock_ec2 616def test_vpc_dedicated_tenancy(): 617 ec2 = boto3.resource("ec2", region_name="us-west-1") 618 619 # Create the default VPC 620 ec2.create_vpc(CidrBlock="172.31.0.0/16") 621 622 # Create the non default VPC 623 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16", InstanceTenancy="dedicated") 624 vpc.reload() 625 vpc.is_default.shouldnt.be.ok 626 627 vpc.instance_tenancy.should.equal("dedicated") 628 629 630@mock_ec2 631def test_vpc_modify_tenancy_unknown(): 632 ec2 = boto3.resource("ec2", region_name="us-west-1") 633 ec2_client = boto3.client("ec2", region_name="us-west-1") 634 635 # Create the default VPC 636 ec2.create_vpc(CidrBlock="172.31.0.0/16") 637 638 # Create the non default VPC 639 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16", InstanceTenancy="dedicated") 640 vpc.instance_tenancy.should.equal("dedicated") 641 642 with pytest.raises(ClientError) as ex: 643 ec2_client.modify_vpc_tenancy(VpcId=vpc.id, InstanceTenancy="unknown") 644 err = ex.value.response["Error"] 645 err["Message"].should.equal("The tenancy value unknown is not supported.") 646 err["Code"].should.equal("UnsupportedTenancy") 647 648 ec2_client.modify_vpc_tenancy(VpcId=vpc.id, InstanceTenancy="default") 649 650 vpc.reload() 651 652 vpc.instance_tenancy.should.equal("default") 653 654 655@mock_ec2 656def test_vpc_modify_enable_dns_support(): 657 ec2 = boto3.resource("ec2", region_name="us-west-1") 658 659 # Create the default VPC 660 ec2.create_vpc(CidrBlock="172.31.0.0/16") 661 662 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 663 664 # Test default values for VPC attributes 665 response = vpc.describe_attribute(Attribute="enableDnsSupport") 666 attr = response.get("EnableDnsSupport") 667 attr.get("Value").should.be.ok 668 669 vpc.modify_attribute(EnableDnsSupport={"Value": False}) 670 671 response = vpc.describe_attribute(Attribute="enableDnsSupport") 672 attr = response.get("EnableDnsSupport") 673 attr.get("Value").shouldnt.be.ok 674 675 676@mock_ec2 677def test_vpc_modify_enable_dns_hostnames(): 678 ec2 = boto3.resource("ec2", region_name="us-west-1") 679 680 # Create the default VPC 681 ec2.create_vpc(CidrBlock="172.31.0.0/16") 682 683 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 684 # Test default values for VPC attributes 685 response = vpc.describe_attribute(Attribute="enableDnsHostnames") 686 attr = response.get("EnableDnsHostnames") 687 attr.get("Value").shouldnt.be.ok 688 689 vpc.modify_attribute(EnableDnsHostnames={"Value": True}) 690 691 response = vpc.describe_attribute(Attribute="enableDnsHostnames") 692 attr = response.get("EnableDnsHostnames") 693 attr.get("Value").should.be.ok 694 695 696# Has boto3 equivalent 697@mock_ec2_deprecated 698def test_vpc_associate_dhcp_options(): 699 conn = boto.connect_vpc() 700 dhcp_options = conn.create_dhcp_options(SAMPLE_DOMAIN_NAME, SAMPLE_NAME_SERVERS) 701 vpc = conn.create_vpc("10.0.0.0/16") 702 703 conn.associate_dhcp_options(dhcp_options.id, vpc.id) 704 705 vpc.update() 706 dhcp_options.id.should.equal(vpc.dhcp_options_id) 707 708 709@mock_ec2 710def test_vpc_associate_dhcp_options_boto3(): 711 ec2 = boto3.resource("ec2", region_name="us-west-1") 712 client = boto3.client("ec2", region_name="us-west-1") 713 dhcp_options = ec2.create_dhcp_options( 714 DhcpConfigurations=[ 715 {"Key": "domain-name", "Values": [SAMPLE_DOMAIN_NAME]}, 716 {"Key": "domain-name-servers", "Values": SAMPLE_NAME_SERVERS}, 717 ] 718 ) 719 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 720 721 client.associate_dhcp_options(DhcpOptionsId=dhcp_options.id, VpcId=vpc.id) 722 723 vpc.reload() 724 dhcp_options.id.should.equal(vpc.dhcp_options_id) 725 726 727@mock_ec2 728def test_associate_vpc_ipv4_cidr_block(): 729 ec2 = boto3.resource("ec2", region_name="us-west-1") 730 731 vpc = ec2.create_vpc(CidrBlock="10.10.42.0/24") 732 733 # Associate/Extend vpc CIDR range up to 5 ciders 734 for i in range(43, 47): 735 response = ec2.meta.client.associate_vpc_cidr_block( 736 VpcId=vpc.id, CidrBlock="10.10.{}.0/24".format(i) 737 ) 738 response["CidrBlockAssociation"]["CidrBlockState"]["State"].should.equal( 739 "associating" 740 ) 741 response["CidrBlockAssociation"]["CidrBlock"].should.equal( 742 "10.10.{}.0/24".format(i) 743 ) 744 response["CidrBlockAssociation"]["AssociationId"].should.contain( 745 "vpc-cidr-assoc" 746 ) 747 748 # Check all associations exist 749 vpc = ec2.Vpc(vpc.id) 750 vpc.cidr_block_association_set.should.have.length_of(5) 751 vpc.cidr_block_association_set[2]["CidrBlockState"]["State"].should.equal( 752 "associated" 753 ) 754 vpc.cidr_block_association_set[4]["CidrBlockState"]["State"].should.equal( 755 "associated" 756 ) 757 758 # Check error on adding 6th association. 759 with pytest.raises(ClientError) as ex: 760 response = ec2.meta.client.associate_vpc_cidr_block( 761 VpcId=vpc.id, CidrBlock="10.10.50.0/22" 762 ) 763 str(ex.value).should.equal( 764 "An error occurred (CidrLimitExceeded) when calling the AssociateVpcCidrBlock " 765 "operation: This network '{}' has met its maximum number of allowed CIDRs: 5".format( 766 vpc.id 767 ) 768 ) 769 770 771@mock_ec2 772def test_disassociate_vpc_ipv4_cidr_block(): 773 ec2 = boto3.resource("ec2", region_name="us-west-1") 774 775 vpc = ec2.create_vpc(CidrBlock="10.10.42.0/24") 776 ec2.meta.client.associate_vpc_cidr_block(VpcId=vpc.id, CidrBlock="10.10.43.0/24") 777 778 # Remove an extended cidr block 779 vpc = ec2.Vpc(vpc.id) 780 non_default_assoc_cidr_block = next( 781 iter( 782 [ 783 x 784 for x in vpc.cidr_block_association_set 785 if vpc.cidr_block != x["CidrBlock"] 786 ] 787 ), 788 None, 789 ) 790 response = ec2.meta.client.disassociate_vpc_cidr_block( 791 AssociationId=non_default_assoc_cidr_block["AssociationId"] 792 ) 793 response["CidrBlockAssociation"]["CidrBlockState"]["State"].should.equal( 794 "disassociating" 795 ) 796 response["CidrBlockAssociation"]["CidrBlock"].should.equal( 797 non_default_assoc_cidr_block["CidrBlock"] 798 ) 799 response["CidrBlockAssociation"]["AssociationId"].should.equal( 800 non_default_assoc_cidr_block["AssociationId"] 801 ) 802 803 # Error attempting to delete a non-existent CIDR_BLOCK association 804 with pytest.raises(ClientError) as ex: 805 response = ec2.meta.client.disassociate_vpc_cidr_block( 806 AssociationId="vpc-cidr-assoc-BORING123" 807 ) 808 str(ex.value).should.equal( 809 "An error occurred (InvalidVpcCidrBlockAssociationIdError.NotFound) when calling the " 810 "DisassociateVpcCidrBlock operation: The vpc CIDR block association ID " 811 "'vpc-cidr-assoc-BORING123' does not exist" 812 ) 813 814 # Error attempting to delete Primary CIDR BLOCK association 815 vpc_base_cidr_assoc_id = next( 816 iter( 817 [ 818 x 819 for x in vpc.cidr_block_association_set 820 if vpc.cidr_block == x["CidrBlock"] 821 ] 822 ), 823 {}, 824 )["AssociationId"] 825 826 with pytest.raises(ClientError) as ex: 827 response = ec2.meta.client.disassociate_vpc_cidr_block( 828 AssociationId=vpc_base_cidr_assoc_id 829 ) 830 str(ex.value).should.equal( 831 "An error occurred (OperationNotPermitted) when calling the DisassociateVpcCidrBlock operation: " 832 "The vpc CIDR block with association ID {} may not be disassociated. It is the primary " 833 "IPv4 CIDR block of the VPC".format(vpc_base_cidr_assoc_id) 834 ) 835 836 837@mock_ec2 838def test_cidr_block_association_filters(): 839 ec2 = boto3.resource("ec2", region_name="us-west-1") 840 vpc1 = ec2.create_vpc(CidrBlock="10.90.0.0/16") 841 vpc2 = ec2.create_vpc(CidrBlock="10.91.0.0/16") 842 ec2.meta.client.associate_vpc_cidr_block(VpcId=vpc2.id, CidrBlock="10.10.0.0/19") 843 vpc3 = ec2.create_vpc(CidrBlock="10.92.0.0/24") 844 ec2.meta.client.associate_vpc_cidr_block(VpcId=vpc3.id, CidrBlock="10.92.1.0/24") 845 ec2.meta.client.associate_vpc_cidr_block(VpcId=vpc3.id, CidrBlock="10.92.2.0/24") 846 vpc3_assoc_response = ec2.meta.client.associate_vpc_cidr_block( 847 VpcId=vpc3.id, CidrBlock="10.92.3.0/24" 848 ) 849 850 # Test filters for a cidr-block in all VPCs cidr-block-associations 851 filtered_vpcs = list( 852 ec2.vpcs.filter( 853 Filters=[ 854 { 855 "Name": "cidr-block-association.cidr-block", 856 "Values": ["10.10.0.0/19"], 857 } 858 ] 859 ) 860 ) 861 [vpc.id for vpc in filtered_vpcs].shouldnt.contain(vpc1.id) 862 [vpc.id for vpc in filtered_vpcs].should.contain(vpc2.id) 863 [vpc.id for vpc in filtered_vpcs].shouldnt.contain(vpc3.id) 864 865 # Test filter for association id in VPCs 866 association_id = vpc3_assoc_response["CidrBlockAssociation"]["AssociationId"] 867 filtered_vpcs = list( 868 ec2.vpcs.filter( 869 Filters=[ 870 { 871 "Name": "cidr-block-association.association-id", 872 "Values": [association_id], 873 } 874 ] 875 ) 876 ) 877 filtered_vpcs.should.be.length_of(1) 878 filtered_vpcs[0].id.should.equal(vpc3.id) 879 880 # Test filter for association state in VPC - this will never show anything in this test 881 filtered_vpcs = list( 882 ec2.vpcs.filter( 883 Filters=[ 884 {"Name": "cidr-block-association.association-id", "Values": ["failing"]} 885 ] 886 ) 887 ) 888 filtered_vpcs.should.be.length_of(0) 889 890 891@mock_ec2 892def test_vpc_associate_ipv6_cidr_block(): 893 ec2 = boto3.resource("ec2", region_name="us-west-1") 894 895 # Test create VPC with IPV6 cidr range 896 vpc = ec2.create_vpc(CidrBlock="10.10.42.0/24", AmazonProvidedIpv6CidrBlock=True) 897 ipv6_cidr_block_association_set = next( 898 iter(vpc.ipv6_cidr_block_association_set), None 899 ) 900 ipv6_cidr_block_association_set["Ipv6CidrBlockState"]["State"].should.equal( 901 "associated" 902 ) 903 ipv6_cidr_block_association_set["Ipv6CidrBlock"].should.contain("::/56") 904 ipv6_cidr_block_association_set["AssociationId"].should.contain("vpc-cidr-assoc") 905 906 # Test Fail on adding 2nd IPV6 association - AWS only allows 1 at this time! 907 with pytest.raises(ClientError) as ex: 908 response = ec2.meta.client.associate_vpc_cidr_block( 909 VpcId=vpc.id, AmazonProvidedIpv6CidrBlock=True 910 ) 911 str(ex.value).should.equal( 912 "An error occurred (CidrLimitExceeded) when calling the AssociateVpcCidrBlock " 913 "operation: This network '{}' has met its maximum number of allowed CIDRs: 1".format( 914 vpc.id 915 ) 916 ) 917 918 # Test associate ipv6 cidr block after vpc created 919 vpc = ec2.create_vpc(CidrBlock="10.10.50.0/24") 920 response = ec2.meta.client.associate_vpc_cidr_block( 921 VpcId=vpc.id, AmazonProvidedIpv6CidrBlock=True 922 ) 923 response["Ipv6CidrBlockAssociation"]["Ipv6CidrBlockState"]["State"].should.equal( 924 "associating" 925 ) 926 response["Ipv6CidrBlockAssociation"]["Ipv6CidrBlock"].should.contain("::/56") 927 response["Ipv6CidrBlockAssociation"]["AssociationId"].should.contain( 928 "vpc-cidr-assoc-" 929 ) 930 931 # Check on describe vpc that has ipv6 cidr block association 932 vpc = ec2.Vpc(vpc.id) 933 vpc.ipv6_cidr_block_association_set.should.be.length_of(1) 934 935 936@mock_ec2 937def test_vpc_disassociate_ipv6_cidr_block(): 938 ec2 = boto3.resource("ec2", region_name="us-west-1") 939 940 # Test create VPC with IPV6 cidr range 941 vpc = ec2.create_vpc(CidrBlock="10.10.42.0/24", AmazonProvidedIpv6CidrBlock=True) 942 # Test disassociating the only IPV6 943 assoc_id = vpc.ipv6_cidr_block_association_set[0]["AssociationId"] 944 response = ec2.meta.client.disassociate_vpc_cidr_block(AssociationId=assoc_id) 945 response["Ipv6CidrBlockAssociation"]["Ipv6CidrBlockState"]["State"].should.equal( 946 "disassociating" 947 ) 948 response["Ipv6CidrBlockAssociation"]["Ipv6CidrBlock"].should.contain("::/56") 949 response["Ipv6CidrBlockAssociation"]["AssociationId"].should.equal(assoc_id) 950 951 952@mock_ec2 953def test_ipv6_cidr_block_association_filters(): 954 ec2 = boto3.resource("ec2", region_name="us-west-1") 955 vpc1 = ec2.create_vpc(CidrBlock="10.90.0.0/16") 956 957 vpc2 = ec2.create_vpc(CidrBlock="10.91.0.0/16", AmazonProvidedIpv6CidrBlock=True) 958 vpc2_assoc_ipv6_assoc_id = vpc2.ipv6_cidr_block_association_set[0]["AssociationId"] 959 ec2.meta.client.associate_vpc_cidr_block(VpcId=vpc2.id, CidrBlock="10.10.0.0/19") 960 961 vpc3 = ec2.create_vpc(CidrBlock="10.92.0.0/24") 962 ec2.meta.client.associate_vpc_cidr_block(VpcId=vpc3.id, CidrBlock="10.92.1.0/24") 963 ec2.meta.client.associate_vpc_cidr_block(VpcId=vpc3.id, CidrBlock="10.92.2.0/24") 964 response = ec2.meta.client.associate_vpc_cidr_block( 965 VpcId=vpc3.id, AmazonProvidedIpv6CidrBlock=True 966 ) 967 vpc3_ipv6_cidr_block = response["Ipv6CidrBlockAssociation"]["Ipv6CidrBlock"] 968 969 vpc4 = ec2.create_vpc(CidrBlock="10.95.0.0/16") # Here for its looks 970 971 # Test filters for an ipv6 cidr-block in all VPCs cidr-block-associations 972 filtered_vpcs = list( 973 ec2.vpcs.filter( 974 Filters=[ 975 { 976 "Name": "ipv6-cidr-block-association.ipv6-cidr-block", 977 "Values": [vpc3_ipv6_cidr_block], 978 } 979 ] 980 ) 981 ) 982 filtered_vpcs.should.be.length_of(1) 983 filtered_vpcs[0].id.should.equal(vpc3.id) 984 985 # Test filter for association id in VPCs 986 filtered_vpcs = list( 987 ec2.vpcs.filter( 988 Filters=[ 989 { 990 "Name": "ipv6-cidr-block-association.association-id", 991 "Values": [vpc2_assoc_ipv6_assoc_id], 992 } 993 ] 994 ) 995 ) 996 filtered_vpcs.should.be.length_of(1) 997 filtered_vpcs[0].id.should.equal(vpc2.id) 998 999 # Test filter for association state in VPC - this will never show anything in this test 1000 assoc_vpcs = [ 1001 vpc.id 1002 for vpc in ec2.vpcs.filter( 1003 Filters=[ 1004 {"Name": "ipv6-cidr-block-association.state", "Values": ["associated"]} 1005 ] 1006 ) 1007 ] 1008 assoc_vpcs.shouldnt.contain(vpc1.id) 1009 assoc_vpcs.should.contain(vpc2.id) 1010 assoc_vpcs.should.contain(vpc3.id) 1011 assoc_vpcs.shouldnt.contain(vpc4.id) 1012 1013 1014@mock_ec2 1015def test_create_vpc_with_invalid_cidr_block_parameter(): 1016 ec2 = boto3.resource("ec2", region_name="us-west-1") 1017 1018 vpc_cidr_block = "1000.1.0.0/20" 1019 with pytest.raises(ClientError) as ex: 1020 ec2.create_vpc(CidrBlock=vpc_cidr_block) 1021 str(ex.value).should.equal( 1022 "An error occurred (InvalidParameterValue) when calling the CreateVpc " 1023 "operation: Value ({}) for parameter cidrBlock is invalid. This is not a valid CIDR block.".format( 1024 vpc_cidr_block 1025 ) 1026 ) 1027 1028 1029@mock_ec2 1030def test_create_vpc_with_invalid_cidr_range(): 1031 ec2 = boto3.resource("ec2", region_name="us-west-1") 1032 1033 vpc_cidr_block = "10.1.0.0/29" 1034 with pytest.raises(ClientError) as ex: 1035 ec2.create_vpc(CidrBlock=vpc_cidr_block) 1036 str(ex.value).should.equal( 1037 "An error occurred (InvalidVpc.Range) when calling the CreateVpc " 1038 "operation: The CIDR '{}' is invalid.".format(vpc_cidr_block) 1039 ) 1040 1041 1042@mock_ec2 1043def test_create_vpc_with_tags(): 1044 ec2 = boto3.resource("ec2", region_name="us-west-1") 1045 # Create VPC 1046 vpc = ec2.create_vpc( 1047 CidrBlock="10.0.0.0/16", 1048 TagSpecifications=[ 1049 {"ResourceType": "vpc", "Tags": [{"Key": "name", "Value": "some-vpc"}]} 1050 ], 1051 ) 1052 assert vpc.tags == [{"Key": "name", "Value": "some-vpc"}] 1053 1054 1055@mock_ec2 1056def test_enable_vpc_classic_link(): 1057 ec2 = boto3.resource("ec2", region_name="us-west-1") 1058 1059 # Create VPC 1060 vpc = ec2.create_vpc(CidrBlock="10.1.0.0/16") 1061 1062 response = ec2.meta.client.enable_vpc_classic_link(VpcId=vpc.id) 1063 assert response.get("Return").should.be.true 1064 1065 1066@mock_ec2 1067def test_enable_vpc_classic_link_failure(): 1068 ec2 = boto3.resource("ec2", region_name="us-west-1") 1069 1070 # Create VPC 1071 vpc = ec2.create_vpc(CidrBlock="10.90.0.0/16") 1072 1073 response = ec2.meta.client.enable_vpc_classic_link(VpcId=vpc.id) 1074 assert response.get("Return").should.be.false 1075 1076 1077@mock_ec2 1078def test_disable_vpc_classic_link(): 1079 ec2 = boto3.resource("ec2", region_name="us-west-1") 1080 1081 # Create VPC 1082 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 1083 1084 ec2.meta.client.enable_vpc_classic_link(VpcId=vpc.id) 1085 response = ec2.meta.client.disable_vpc_classic_link(VpcId=vpc.id) 1086 assert response.get("Return").should.be.false 1087 1088 1089@mock_ec2 1090def test_describe_classic_link_enabled(): 1091 ec2 = boto3.resource("ec2", region_name="us-west-1") 1092 1093 # Create VPC 1094 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 1095 1096 ec2.meta.client.enable_vpc_classic_link(VpcId=vpc.id) 1097 response = ec2.meta.client.describe_vpc_classic_link(VpcIds=[vpc.id]) 1098 assert response.get("Vpcs")[0].get("ClassicLinkEnabled").should.be.true 1099 1100 1101@mock_ec2 1102def test_describe_classic_link_disabled(): 1103 ec2 = boto3.resource("ec2", region_name="us-west-1") 1104 1105 # Create VPC 1106 vpc = ec2.create_vpc(CidrBlock="10.90.0.0/16") 1107 1108 response = ec2.meta.client.describe_vpc_classic_link(VpcIds=[vpc.id]) 1109 assert response.get("Vpcs")[0].get("ClassicLinkEnabled").should.be.false 1110 1111 1112@mock_ec2 1113def test_describe_classic_link_multiple(): 1114 ec2 = boto3.resource("ec2", region_name="us-west-1") 1115 1116 # Create VPC 1117 vpc1 = ec2.create_vpc(CidrBlock="10.90.0.0/16") 1118 vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 1119 1120 ec2.meta.client.enable_vpc_classic_link(VpcId=vpc2.id) 1121 response = ec2.meta.client.describe_vpc_classic_link(VpcIds=[vpc1.id, vpc2.id]) 1122 expected = [ 1123 {"VpcId": vpc1.id, "ClassicLinkDnsSupported": False}, 1124 {"VpcId": vpc2.id, "ClassicLinkDnsSupported": True}, 1125 ] 1126 1127 # Ensure response is sorted, because they can come in random order 1128 assert response.get("Vpcs").sort(key=lambda x: x["VpcId"]) == expected.sort( 1129 key=lambda x: x["VpcId"] 1130 ) 1131 1132 1133@mock_ec2 1134def test_enable_vpc_classic_link_dns_support(): 1135 ec2 = boto3.resource("ec2", region_name="us-west-1") 1136 1137 # Create VPC 1138 vpc = ec2.create_vpc(CidrBlock="10.1.0.0/16") 1139 1140 response = ec2.meta.client.enable_vpc_classic_link_dns_support(VpcId=vpc.id) 1141 assert response.get("Return").should.be.true 1142 1143 1144@mock_ec2 1145def test_disable_vpc_classic_link_dns_support(): 1146 ec2 = boto3.resource("ec2", region_name="us-west-1") 1147 1148 # Create VPC 1149 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 1150 1151 ec2.meta.client.enable_vpc_classic_link_dns_support(VpcId=vpc.id) 1152 response = ec2.meta.client.disable_vpc_classic_link_dns_support(VpcId=vpc.id) 1153 assert response.get("Return").should.be.false 1154 1155 1156@mock_ec2 1157def test_describe_classic_link_dns_support_enabled(): 1158 ec2 = boto3.resource("ec2", region_name="us-west-1") 1159 1160 # Create VPC 1161 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 1162 1163 ec2.meta.client.enable_vpc_classic_link_dns_support(VpcId=vpc.id) 1164 response = ec2.meta.client.describe_vpc_classic_link_dns_support(VpcIds=[vpc.id]) 1165 assert response.get("Vpcs")[0].get("ClassicLinkDnsSupported").should.be.true 1166 1167 1168@mock_ec2 1169def test_describe_classic_link_dns_support_disabled(): 1170 ec2 = boto3.resource("ec2", region_name="us-west-1") 1171 1172 # Create VPC 1173 vpc = ec2.create_vpc(CidrBlock="10.90.0.0/16") 1174 1175 response = ec2.meta.client.describe_vpc_classic_link_dns_support(VpcIds=[vpc.id]) 1176 assert response.get("Vpcs")[0].get("ClassicLinkDnsSupported").should.be.false 1177 1178 1179@mock_ec2 1180def test_describe_classic_link_dns_support_multiple(): 1181 ec2 = boto3.resource("ec2", region_name="us-west-1") 1182 1183 # Create VPC 1184 vpc1 = ec2.create_vpc(CidrBlock="10.90.0.0/16") 1185 vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 1186 1187 ec2.meta.client.enable_vpc_classic_link_dns_support(VpcId=vpc2.id) 1188 response = ec2.meta.client.describe_vpc_classic_link_dns_support( 1189 VpcIds=[vpc1.id, vpc2.id] 1190 ) 1191 expected = [ 1192 {"VpcId": vpc1.id, "ClassicLinkDnsSupported": False}, 1193 {"VpcId": vpc2.id, "ClassicLinkDnsSupported": True}, 1194 ] 1195 1196 # Ensure response is sorted, because they can come in random order 1197 assert response.get("Vpcs").sort(key=lambda x: x["VpcId"]) == expected.sort( 1198 key=lambda x: x["VpcId"] 1199 ) 1200 1201 1202@mock_ec2 1203def test_describe_vpc_end_points(): 1204 ec2 = boto3.client("ec2", region_name="us-west-1") 1205 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"] 1206 1207 route_table = ec2.create_route_table(VpcId=vpc["VpcId"])["RouteTable"] 1208 vpc_end_point = ec2.create_vpc_endpoint( 1209 VpcId=vpc["VpcId"], 1210 ServiceName="com.amazonaws.us-east-1.s3", 1211 RouteTableIds=[route_table["RouteTableId"]], 1212 VpcEndpointType="gateway", 1213 )["VpcEndpoint"] 1214 our_id = vpc_end_point["VpcEndpointId"] 1215 1216 all_endpoints = retrieve_all_endpoints(ec2) 1217 [e["VpcEndpointId"] for e in all_endpoints].should.contain(our_id) 1218 our_endpoint = [e for e in all_endpoints if e["VpcEndpointId"] == our_id][0] 1219 vpc_end_point["PrivateDnsEnabled"].should.be.true 1220 our_endpoint["PrivateDnsEnabled"].should.be.true 1221 1222 our_endpoint["VpcId"].should.equal(vpc["VpcId"]) 1223 our_endpoint["RouteTableIds"].should.equal([route_table["RouteTableId"]]) 1224 1225 our_endpoint.should.have.key("VpcEndpointType").equal("gateway") 1226 our_endpoint.should.have.key("ServiceName").equal("com.amazonaws.us-east-1.s3") 1227 our_endpoint.should.have.key("State").equal("available") 1228 1229 endpoint_by_id = ec2.describe_vpc_endpoints(VpcEndpointIds=[our_id])[ 1230 "VpcEndpoints" 1231 ][0] 1232 endpoint_by_id["VpcEndpointId"].should.equal(our_id) 1233 endpoint_by_id["VpcId"].should.equal(vpc["VpcId"]) 1234 endpoint_by_id["RouteTableIds"].should.equal([route_table["RouteTableId"]]) 1235 endpoint_by_id["VpcEndpointType"].should.equal("gateway") 1236 endpoint_by_id["ServiceName"].should.equal("com.amazonaws.us-east-1.s3") 1237 endpoint_by_id["State"].should.equal("available") 1238 1239 with pytest.raises(ClientError) as ex: 1240 ec2.describe_vpc_endpoints(VpcEndpointIds=[route_table["RouteTableId"]]) 1241 err = ex.value.response["Error"] 1242 err["Code"].should.equal("InvalidVpcEndpointId.NotFound") 1243 1244 1245def retrieve_all_endpoints(ec2): 1246 resp = ec2.describe_vpc_endpoints() 1247 all_endpoints = resp["VpcEndpoints"] 1248 next_token = resp.get("NextToken") 1249 while next_token: 1250 resp = ec2.describe_vpc_endpoints(NextToken=next_token) 1251 all_endpoints.extend(resp["VpcEndpoints"]) 1252 next_token = resp.get("NextToken") 1253 return all_endpoints 1254 1255 1256@mock_ec2 1257def test_delete_vpc_end_points(): 1258 ec2 = boto3.client("ec2", region_name="us-west-1") 1259 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"] 1260 1261 route_table = ec2.create_route_table(VpcId=vpc["VpcId"])["RouteTable"] 1262 vpc_end_point1 = ec2.create_vpc_endpoint( 1263 VpcId=vpc["VpcId"], 1264 ServiceName="com.amazonaws.us-west-1.s3", 1265 RouteTableIds=[route_table["RouteTableId"]], 1266 VpcEndpointType="gateway", 1267 )["VpcEndpoint"] 1268 vpc_end_point2 = ec2.create_vpc_endpoint( 1269 VpcId=vpc["VpcId"], 1270 ServiceName="com.amazonaws.us-west-1.s3", 1271 RouteTableIds=[route_table["RouteTableId"]], 1272 VpcEndpointType="gateway", 1273 )["VpcEndpoint"] 1274 1275 vpc_endpoints = retrieve_all_endpoints(ec2) 1276 all_ids = [e["VpcEndpointId"] for e in vpc_endpoints] 1277 all_ids.should.contain(vpc_end_point1["VpcEndpointId"]) 1278 all_ids.should.contain(vpc_end_point2["VpcEndpointId"]) 1279 1280 ec2.delete_vpc_endpoints(VpcEndpointIds=[vpc_end_point1["VpcEndpointId"]]) 1281 1282 vpc_endpoints = retrieve_all_endpoints(ec2) 1283 all_ids = [e["VpcEndpointId"] for e in vpc_endpoints] 1284 all_ids.should.contain(vpc_end_point1["VpcEndpointId"]) 1285 all_ids.should.contain(vpc_end_point2["VpcEndpointId"]) 1286 1287 ep1 = ec2.describe_vpc_endpoints(VpcEndpointIds=[vpc_end_point1["VpcEndpointId"]])[ 1288 "VpcEndpoints" 1289 ][0] 1290 ep1["State"].should.equal("deleted") 1291 1292 ep2 = ec2.describe_vpc_endpoints(VpcEndpointIds=[vpc_end_point2["VpcEndpointId"]])[ 1293 "VpcEndpoints" 1294 ][0] 1295 ep2["State"].should.equal("available") 1296 1297 1298@mock_ec2 1299def test_describe_vpcs_dryrun(): 1300 client = boto3.client("ec2", region_name="us-east-1") 1301 1302 with pytest.raises(ClientError) as ex: 1303 client.describe_vpcs(DryRun=True) 1304 ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412) 1305 ex.value.response["Error"]["Code"].should.equal("DryRunOperation") 1306 ex.value.response["Error"]["Message"].should.equal( 1307 "An error occurred (DryRunOperation) when calling the DescribeVpcs operation: Request would have succeeded, but DryRun flag is set" 1308 ) 1309