1import random 2 3import boto 4import boto3 5import boto.vpc 6 7import pytest 8import sure # noqa # pylint: disable=unused-import 9from boto.exception import EC2ResponseError 10from botocore.exceptions import ClientError 11from moto import mock_ec2, mock_ec2_deprecated, settings 12from tests import EXAMPLE_AMI_ID 13from uuid import uuid4 14from unittest import SkipTest 15 16 17# Has boto3 equivalent 18@mock_ec2_deprecated 19def test_subnets(): 20 ec2 = boto.connect_ec2("the_key", "the_secret") 21 conn = boto.connect_vpc("the_key", "the_secret") 22 vpc = conn.create_vpc("10.0.0.0/16") 23 subnet = conn.create_subnet(vpc.id, "10.0.0.0/18") 24 25 all_subnets = conn.get_all_subnets() 26 all_subnets.should.have.length_of(1 + len(ec2.get_all_zones())) 27 28 conn.delete_subnet(subnet.id) 29 30 all_subnets = conn.get_all_subnets() 31 all_subnets.should.have.length_of(0 + len(ec2.get_all_zones())) 32 33 with pytest.raises(EC2ResponseError) as cm: 34 conn.delete_subnet(subnet.id) 35 cm.value.code.should.equal("InvalidSubnetID.NotFound") 36 cm.value.status.should.equal(400) 37 cm.value.request_id.should_not.be.none 38 39 40@mock_ec2 41def test_subnets_boto3(): 42 ec2 = boto3.resource("ec2", region_name="us-east-1") 43 client = boto3.client("ec2", region_name="us-east-1") 44 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 45 subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock="10.0.0.0/18") 46 47 ours = client.describe_subnets(SubnetIds=[subnet.id])["Subnets"] 48 ours.should.have.length_of(1) 49 50 client.delete_subnet(SubnetId=subnet.id) 51 52 with pytest.raises(ClientError) as ex: 53 client.describe_subnets(SubnetIds=[subnet.id]) 54 err = ex.value.response["Error"] 55 err["Code"].should.equal("InvalidSubnetID.NotFound") 56 57 with pytest.raises(ClientError) as ex: 58 client.delete_subnet(SubnetId=subnet.id) 59 ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 60 ex.value.response["ResponseMetadata"].should.have.key("RequestId") 61 ex.value.response["Error"]["Code"].should.equal("InvalidSubnetID.NotFound") 62 63 64# Has boto3 equivalent 65@mock_ec2_deprecated 66def test_subnet_create_vpc_validation(): 67 conn = boto.connect_vpc("the_key", "the_secret") 68 69 with pytest.raises(EC2ResponseError) as cm: 70 conn.create_subnet("vpc-abcd1234", "10.0.0.0/18") 71 cm.value.code.should.equal("InvalidVpcID.NotFound") 72 cm.value.status.should.equal(400) 73 cm.value.request_id.should_not.be.none 74 75 76@mock_ec2 77def test_subnet_create_vpc_validation_boto3(): 78 ec2 = boto3.resource("ec2", region_name="us-east-1") 79 80 with pytest.raises(ClientError) as ex: 81 ec2.create_subnet(VpcId="vpc-abcd1234", CidrBlock="10.0.0.0/18") 82 ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 83 ex.value.response["ResponseMetadata"].should.have.key("RequestId") 84 ex.value.response["Error"]["Code"].should.equal("InvalidVpcID.NotFound") 85 86 87# Has boto3 equivalent 88@mock_ec2_deprecated 89def test_subnet_tagging(): 90 conn = boto.connect_vpc("the_key", "the_secret") 91 vpc = conn.create_vpc("10.0.0.0/16") 92 subnet = conn.create_subnet(vpc.id, "10.0.0.0/18") 93 94 subnet.add_tag("a key", "some value") 95 96 tag = conn.get_all_tags()[0] 97 tag.name.should.equal("a key") 98 tag.value.should.equal("some value") 99 100 # Refresh the subnet 101 subnet = conn.get_all_subnets(subnet_ids=[subnet.id])[0] 102 subnet.tags.should.have.length_of(1) 103 subnet.tags["a key"].should.equal("some value") 104 105 106@mock_ec2 107def test_subnet_tagging_boto3(): 108 ec2 = boto3.resource("ec2", region_name="us-east-1") 109 client = boto3.client("ec2", region_name="us-east-1") 110 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 111 subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock="10.0.0.0/18") 112 113 subnet.create_tags(Tags=[{"Key": "a key", "Value": "some value"}]) 114 115 tag = client.describe_tags( 116 Filters=[{"Name": "resource-id", "Values": [subnet.id]}] 117 )["Tags"][0] 118 tag["Key"].should.equal("a key") 119 tag["Value"].should.equal("some value") 120 121 # Refresh the subnet 122 subnet = client.describe_subnets(SubnetIds=[subnet.id])["Subnets"][0] 123 subnet["Tags"].should.equal([{"Key": "a key", "Value": "some value"}]) 124 125 126# Has boto3 equivalent 127@mock_ec2_deprecated 128def test_subnet_should_have_proper_availability_zone_set(): 129 conn = boto.vpc.connect_to_region("us-west-1") 130 vpcA = conn.create_vpc("10.0.0.0/16") 131 subnetA = conn.create_subnet(vpcA.id, "10.0.0.0/24", availability_zone="us-west-1b") 132 subnetA.availability_zone.should.equal("us-west-1b") 133 134 135@mock_ec2 136def test_subnet_should_have_proper_availability_zone_set_boto3(): 137 ec2 = boto3.resource("ec2", region_name="us-west-1") 138 vpcA = ec2.create_vpc(CidrBlock="10.0.0.0/16") 139 subnetA = ec2.create_subnet( 140 VpcId=vpcA.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-1b" 141 ) 142 subnetA.availability_zone.should.equal("us-west-1b") 143 144 145@mock_ec2 146def test_availability_zone_in_create_subnet(): 147 ec2 = boto3.resource("ec2", region_name="us-west-1") 148 149 vpc = ec2.create_vpc(CidrBlock="172.31.0.0/16") 150 151 subnet = ec2.create_subnet( 152 VpcId=vpc.id, CidrBlock="172.31.48.0/20", AvailabilityZoneId="use1-az6" 153 ) 154 subnet.availability_zone_id.should.equal("use1-az6") 155 156 157@mock_ec2 158def test_default_subnet(): 159 if settings.TEST_SERVER_MODE: 160 raise SkipTest("ServerMode will have conflicting CidrBlocks") 161 ec2 = boto3.resource("ec2", region_name="us-west-1") 162 163 default_vpc = list(ec2.vpcs.all())[0] 164 default_vpc.cidr_block.should.equal("172.31.0.0/16") 165 default_vpc.reload() 166 default_vpc.is_default.should.be.ok 167 168 subnet = ec2.create_subnet( 169 VpcId=default_vpc.id, CidrBlock="172.31.48.0/20", AvailabilityZone="us-west-1a" 170 ) 171 subnet.reload() 172 subnet.map_public_ip_on_launch.shouldnt.be.ok 173 174 175# Has boto3 equivalent 176@mock_ec2_deprecated 177def test_non_default_subnet(): 178 vpc_cli = boto.vpc.connect_to_region("us-west-1") 179 180 # Create the non default VPC 181 vpc = vpc_cli.create_vpc("10.0.0.0/16") 182 vpc.is_default.shouldnt.be.ok 183 184 subnet = vpc_cli.create_subnet(vpc.id, "10.0.0.0/24") 185 subnet = vpc_cli.get_all_subnets(subnet_ids=[subnet.id])[0] 186 subnet.mapPublicIpOnLaunch.should.equal("false") 187 188 189@mock_ec2 190def test_boto3_non_default_subnet(): 191 ec2 = boto3.resource("ec2", region_name="us-west-1") 192 193 # Create the non default VPC 194 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 195 vpc.reload() 196 vpc.is_default.shouldnt.be.ok 197 198 subnet = ec2.create_subnet( 199 VpcId=vpc.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-1a" 200 ) 201 subnet.reload() 202 subnet.map_public_ip_on_launch.shouldnt.be.ok 203 204 205@mock_ec2 206def test_modify_subnet_attribute_public_ip_on_launch(): 207 ec2 = boto3.resource("ec2", region_name="us-west-1") 208 client = boto3.client("ec2", region_name="us-west-1") 209 210 random_ip = ".".join(map(str, (random.randint(0, 99) for _ in range(4)))) 211 vpc = ec2.create_vpc(CidrBlock=f"{random_ip}/16") 212 213 random_subnet_cidr = f"{random_ip}/20" # Same block as the VPC 214 215 subnet = ec2.create_subnet( 216 VpcId=vpc.id, CidrBlock=random_subnet_cidr, AvailabilityZone="us-west-1a" 217 ) 218 219 # 'map_public_ip_on_launch' is set when calling 'DescribeSubnets' action 220 subnet.reload() 221 222 # For non default subnet, attribute value should be 'False' 223 subnet.map_public_ip_on_launch.shouldnt.be.ok 224 225 client.modify_subnet_attribute( 226 SubnetId=subnet.id, MapPublicIpOnLaunch={"Value": False} 227 ) 228 subnet.reload() 229 subnet.map_public_ip_on_launch.shouldnt.be.ok 230 231 client.modify_subnet_attribute( 232 SubnetId=subnet.id, MapPublicIpOnLaunch={"Value": True} 233 ) 234 subnet.reload() 235 subnet.map_public_ip_on_launch.should.be.ok 236 237 238@mock_ec2 239def test_modify_subnet_attribute_assign_ipv6_address_on_creation(): 240 ec2 = boto3.resource("ec2", region_name="us-west-1") 241 client = boto3.client("ec2", region_name="us-west-1") 242 243 random_ip = ".".join(map(str, (random.randint(0, 99) for _ in range(4)))) 244 vpc = ec2.create_vpc(CidrBlock=f"{random_ip}/16") 245 246 random_subnet_cidr = f"{random_ip}/20" # Same block as the VPC 247 248 subnet = ec2.create_subnet( 249 VpcId=vpc.id, CidrBlock=random_subnet_cidr, AvailabilityZone="us-west-1a" 250 ) 251 252 # 'map_public_ip_on_launch' is set when calling 'DescribeSubnets' action 253 subnet.reload() 254 client.describe_subnets() 255 256 # For non default subnet, attribute value should be 'False' 257 subnet.assign_ipv6_address_on_creation.shouldnt.be.ok 258 259 client.modify_subnet_attribute( 260 SubnetId=subnet.id, AssignIpv6AddressOnCreation={"Value": False} 261 ) 262 subnet.reload() 263 subnet.assign_ipv6_address_on_creation.shouldnt.be.ok 264 265 client.modify_subnet_attribute( 266 SubnetId=subnet.id, AssignIpv6AddressOnCreation={"Value": True} 267 ) 268 subnet.reload() 269 subnet.assign_ipv6_address_on_creation.should.be.ok 270 271 272@mock_ec2 273def test_modify_subnet_attribute_validation(): 274 # TODO: implement some actual logic 275 ec2 = boto3.resource("ec2", region_name="us-west-1") 276 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 277 ec2.create_subnet( 278 VpcId=vpc.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-1a" 279 ) 280 281 282# Has boto3 equivalent 283@mock_ec2_deprecated 284def test_subnet_get_by_id(): 285 conn = boto.vpc.connect_to_region("us-west-1") 286 vpcA = conn.create_vpc("10.0.0.0/16") 287 subnetA = conn.create_subnet(vpcA.id, "10.0.0.0/24", availability_zone="us-west-1a") 288 vpcB = conn.create_vpc("10.0.0.0/16") 289 subnetB1 = conn.create_subnet( 290 vpcB.id, "10.0.0.0/24", availability_zone="us-west-1a" 291 ) 292 conn.create_subnet(vpcB.id, "10.0.1.0/24", availability_zone="us-west-1b") 293 294 subnets_by_id = conn.get_all_subnets(subnet_ids=[subnetA.id, subnetB1.id]) 295 subnets_by_id.should.have.length_of(2) 296 subnets_by_id = tuple(map(lambda s: s.id, subnets_by_id)) 297 subnetA.id.should.be.within(subnets_by_id) 298 subnetB1.id.should.be.within(subnets_by_id) 299 300 with pytest.raises(EC2ResponseError) as cm: 301 conn.get_all_subnets(subnet_ids=["subnet-does_not_exist"]) 302 cm.value.code.should.equal("InvalidSubnetID.NotFound") 303 cm.value.status.should.equal(400) 304 cm.value.request_id.should_not.be.none 305 306 307@mock_ec2 308def test_subnet_get_by_id_boto3(): 309 ec2 = boto3.resource("ec2", region_name="us-west-1") 310 client = boto3.client("ec2", region_name="us-west-1") 311 vpcA = ec2.create_vpc(CidrBlock="10.0.0.0/16") 312 subnetA = ec2.create_subnet( 313 VpcId=vpcA.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-1a" 314 ) 315 vpcB = ec2.create_vpc(CidrBlock="10.0.0.0/16") 316 subnetB1 = ec2.create_subnet( 317 VpcId=vpcB.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-1a" 318 ) 319 ec2.create_subnet( 320 VpcId=vpcB.id, CidrBlock="10.0.1.0/24", AvailabilityZone="us-west-1b" 321 ) 322 323 subnets_by_id = client.describe_subnets(SubnetIds=[subnetA.id, subnetB1.id])[ 324 "Subnets" 325 ] 326 subnets_by_id.should.have.length_of(2) 327 subnets_by_id = tuple(map(lambda s: s["SubnetId"], subnets_by_id)) 328 subnetA.id.should.be.within(subnets_by_id) 329 subnetB1.id.should.be.within(subnets_by_id) 330 331 with pytest.raises(ClientError) as ex: 332 client.describe_subnets(SubnetIds=["subnet-does_not_exist"]) 333 ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 334 ex.value.response["ResponseMetadata"].should.have.key("RequestId") 335 ex.value.response["Error"]["Code"].should.equal("InvalidSubnetID.NotFound") 336 337 338# Has boto3 equivalent 339@mock_ec2_deprecated 340def test_get_subnets_filtering(): 341 ec2 = boto.ec2.connect_to_region("us-west-1") 342 conn = boto.vpc.connect_to_region("us-west-1") 343 vpcA = conn.create_vpc("10.0.0.0/16") 344 subnetA = conn.create_subnet(vpcA.id, "10.0.0.0/24", availability_zone="us-west-1a") 345 vpcB = conn.create_vpc("10.0.0.0/16") 346 subnetB1 = conn.create_subnet( 347 vpcB.id, "10.0.0.0/24", availability_zone="us-west-1a" 348 ) 349 subnetB2 = conn.create_subnet( 350 vpcB.id, "10.0.1.0/24", availability_zone="us-west-1b" 351 ) 352 353 all_subnets = conn.get_all_subnets() 354 all_subnets.should.have.length_of(3 + len(ec2.get_all_zones())) 355 356 # Filter by VPC ID 357 subnets_by_vpc = conn.get_all_subnets(filters={"vpc-id": vpcB.id}) 358 subnets_by_vpc.should.have.length_of(2) 359 set([subnet.id for subnet in subnets_by_vpc]).should.equal( 360 set([subnetB1.id, subnetB2.id]) 361 ) 362 363 # Filter by CIDR variations 364 subnets_by_cidr1 = conn.get_all_subnets(filters={"cidr": "10.0.0.0/24"}) 365 subnets_by_cidr1.should.have.length_of(2) 366 set([subnet.id for subnet in subnets_by_cidr1]).should.equal( 367 set([subnetA.id, subnetB1.id]) 368 ) 369 370 subnets_by_cidr2 = conn.get_all_subnets(filters={"cidr-block": "10.0.0.0/24"}) 371 subnets_by_cidr2.should.have.length_of(2) 372 set([subnet.id for subnet in subnets_by_cidr2]).should.equal( 373 set([subnetA.id, subnetB1.id]) 374 ) 375 376 subnets_by_cidr3 = conn.get_all_subnets(filters={"cidrBlock": "10.0.0.0/24"}) 377 subnets_by_cidr3.should.have.length_of(2) 378 set([subnet.id for subnet in subnets_by_cidr3]).should.equal( 379 set([subnetA.id, subnetB1.id]) 380 ) 381 382 # Filter by VPC ID and CIDR 383 subnets_by_vpc_and_cidr = conn.get_all_subnets( 384 filters={"vpc-id": vpcB.id, "cidr": "10.0.0.0/24"} 385 ) 386 subnets_by_vpc_and_cidr.should.have.length_of(1) 387 set([subnet.id for subnet in subnets_by_vpc_and_cidr]).should.equal( 388 set([subnetB1.id]) 389 ) 390 391 # Filter by subnet ID 392 subnets_by_id = conn.get_all_subnets(filters={"subnet-id": subnetA.id}) 393 subnets_by_id.should.have.length_of(1) 394 set([subnet.id for subnet in subnets_by_id]).should.equal(set([subnetA.id])) 395 396 # Filter by availabilityZone 397 subnets_by_az = conn.get_all_subnets( 398 filters={"availabilityZone": "us-west-1a", "vpc-id": vpcB.id} 399 ) 400 subnets_by_az.should.have.length_of(1) 401 set([subnet.id for subnet in subnets_by_az]).should.equal(set([subnetB1.id])) 402 403 # Filter by defaultForAz 404 405 subnets_by_az = conn.get_all_subnets(filters={"defaultForAz": "true"}) 406 subnets_by_az.should.have.length_of(len(conn.get_all_zones())) 407 408 # Unsupported filter 409 conn.get_all_subnets.when.called_with( 410 filters={"not-implemented-filter": "foobar"} 411 ).should.throw(NotImplementedError) 412 413 414@mock_ec2 415def test_get_subnets_filtering_boto3(): 416 ec2 = boto3.resource("ec2", region_name="us-west-1") 417 client = boto3.client("ec2", region_name="us-west-1") 418 vpcA = ec2.create_vpc(CidrBlock="10.0.0.0/16") 419 subnetA = ec2.create_subnet( 420 VpcId=vpcA.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-1a" 421 ) 422 vpcB = ec2.create_vpc(CidrBlock="10.0.0.0/16") 423 subnetB1 = ec2.create_subnet( 424 VpcId=vpcB.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-1a" 425 ) 426 subnetB2 = ec2.create_subnet( 427 VpcId=vpcB.id, CidrBlock="10.0.1.0/24", AvailabilityZone="us-west-1b" 428 ) 429 430 nr_of_a_zones = len(client.describe_availability_zones()["AvailabilityZones"]) 431 all_subnets = client.describe_subnets()["Subnets"] 432 if settings.TEST_SERVER_MODE: 433 # ServerMode may have other tests running that are creating subnets 434 all_subnet_ids = [s["SubnetId"] for s in all_subnets] 435 all_subnet_ids.should.contain(subnetA.id) 436 all_subnet_ids.should.contain(subnetB1.id) 437 all_subnet_ids.should.contain(subnetB2.id) 438 else: 439 all_subnets.should.have.length_of(3 + nr_of_a_zones) 440 441 # Filter by VPC ID 442 subnets_by_vpc = client.describe_subnets( 443 Filters=[{"Name": "vpc-id", "Values": [vpcB.id]}] 444 )["Subnets"] 445 subnets_by_vpc.should.have.length_of(2) 446 set([subnet["SubnetId"] for subnet in subnets_by_vpc]).should.equal( 447 set([subnetB1.id, subnetB2.id]) 448 ) 449 450 # Filter by CIDR variations 451 subnets_by_cidr1 = client.describe_subnets( 452 Filters=[{"Name": "cidr", "Values": ["10.0.0.0/24"]}] 453 )["Subnets"] 454 subnets_by_cidr1 = [s["SubnetId"] for s in subnets_by_cidr1] 455 subnets_by_cidr1.should.contain(subnetA.id) 456 subnets_by_cidr1.should.contain(subnetB1.id) 457 subnets_by_cidr1.shouldnt.contain(subnetB2.id) 458 459 subnets_by_cidr2 = client.describe_subnets( 460 Filters=[{"Name": "cidr-block", "Values": ["10.0.0.0/24"]}] 461 )["Subnets"] 462 subnets_by_cidr2 = [s["SubnetId"] for s in subnets_by_cidr2] 463 subnets_by_cidr2.should.contain(subnetA.id) 464 subnets_by_cidr2.should.contain(subnetB1.id) 465 subnets_by_cidr2.shouldnt.contain(subnetB2.id) 466 467 subnets_by_cidr3 = client.describe_subnets( 468 Filters=[{"Name": "cidrBlock", "Values": ["10.0.0.0/24"]}] 469 )["Subnets"] 470 subnets_by_cidr3 = [s["SubnetId"] for s in subnets_by_cidr3] 471 subnets_by_cidr3.should.contain(subnetA.id) 472 subnets_by_cidr3.should.contain(subnetB1.id) 473 subnets_by_cidr3.shouldnt.contain(subnetB2.id) 474 475 # Filter by VPC ID and CIDR 476 subnets_by_vpc_and_cidr = client.describe_subnets( 477 Filters=[ 478 {"Name": "vpc-id", "Values": [vpcB.id]}, 479 {"Name": "cidr", "Values": ["10.0.0.0/24"]}, 480 ] 481 )["Subnets"] 482 subnets_by_vpc_and_cidr.should.have.length_of(1) 483 subnets_by_vpc_and_cidr[0]["SubnetId"].should.equal(subnetB1.id) 484 485 # Filter by subnet ID 486 subnets_by_id = client.describe_subnets( 487 Filters=[{"Name": "subnet-id", "Values": [subnetA.id]}] 488 )["Subnets"] 489 subnets_by_id.should.have.length_of(1) 490 subnets_by_id[0]["SubnetId"].should.equal(subnetA.id) 491 492 # Filter by availabilityZone 493 subnets_by_az = client.describe_subnets( 494 Filters=[ 495 {"Name": "availabilityZone", "Values": ["us-west-1a"]}, 496 {"Name": "vpc-id", "Values": [vpcB.id]}, 497 ] 498 )["Subnets"] 499 subnets_by_az.should.have.length_of(1) 500 subnets_by_az[0]["SubnetId"].should.equal(subnetB1.id) 501 502 if not settings.TEST_SERVER_MODE: 503 # Filter by defaultForAz 504 subnets_by_az = client.describe_subnets( 505 Filters=[{"Name": "defaultForAz", "Values": ["true"]}] 506 )["Subnets"] 507 subnets_by_az.should.have.length_of(nr_of_a_zones) 508 509 # Unsupported filter 510 filters = [{"Name": "not-implemented-filter", "Values": ["foobar"]}] 511 client.describe_subnets.when.called_with(Filters=filters).should.throw( 512 NotImplementedError 513 ) 514 515 516@mock_ec2 517def test_create_subnet_response_fields(): 518 ec2 = boto3.resource("ec2", region_name="us-west-1") 519 client = boto3.client("ec2", region_name="us-west-1") 520 521 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 522 subnet = client.create_subnet( 523 VpcId=vpc.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-1a" 524 )["Subnet"] 525 526 subnet.should.have.key("AvailabilityZone") 527 subnet.should.have.key("AvailabilityZoneId") 528 subnet.should.have.key("AvailableIpAddressCount") 529 subnet.should.have.key("CidrBlock") 530 subnet.should.have.key("State") 531 subnet.should.have.key("SubnetId") 532 subnet.should.have.key("VpcId") 533 subnet.should.have.key("Tags") 534 subnet.should.have.key("DefaultForAz").which.should.equal(False) 535 subnet.should.have.key("MapPublicIpOnLaunch").which.should.equal(False) 536 subnet.should.have.key("OwnerId") 537 subnet.should.have.key("AssignIpv6AddressOnCreation").which.should.equal(False) 538 539 subnet_arn = "arn:aws:ec2:{region}:{owner_id}:subnet/{subnet_id}".format( 540 region=subnet["AvailabilityZone"][0:-1], 541 owner_id=subnet["OwnerId"], 542 subnet_id=subnet["SubnetId"], 543 ) 544 subnet.should.have.key("SubnetArn").which.should.equal(subnet_arn) 545 subnet.should.have.key("Ipv6CidrBlockAssociationSet").which.should.equal([]) 546 547 548@mock_ec2 549def test_describe_subnet_response_fields(): 550 ec2 = boto3.resource("ec2", region_name="us-west-1") 551 client = boto3.client("ec2", region_name="us-west-1") 552 553 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 554 subnet_object = ec2.create_subnet( 555 VpcId=vpc.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-1a" 556 ) 557 558 subnets = client.describe_subnets(SubnetIds=[subnet_object.id])["Subnets"] 559 subnets.should.have.length_of(1) 560 subnet = subnets[0] 561 562 subnet.should.have.key("AvailabilityZone") 563 subnet.should.have.key("AvailabilityZoneId") 564 subnet.should.have.key("AvailableIpAddressCount") 565 subnet.should.have.key("CidrBlock") 566 subnet.should.have.key("State") 567 subnet.should.have.key("SubnetId") 568 subnet.should.have.key("VpcId") 569 subnet.shouldnt.have.key("Tags") 570 subnet.should.have.key("DefaultForAz").which.should.equal(False) 571 subnet.should.have.key("MapPublicIpOnLaunch").which.should.equal(False) 572 subnet.should.have.key("OwnerId") 573 subnet.should.have.key("AssignIpv6AddressOnCreation").which.should.equal(False) 574 575 subnet_arn = "arn:aws:ec2:{region}:{owner_id}:subnet/{subnet_id}".format( 576 region=subnet["AvailabilityZone"][0:-1], 577 owner_id=subnet["OwnerId"], 578 subnet_id=subnet["SubnetId"], 579 ) 580 subnet.should.have.key("SubnetArn").which.should.equal(subnet_arn) 581 subnet.should.have.key("Ipv6CidrBlockAssociationSet").which.should.equal([]) 582 583 584@mock_ec2 585def test_create_subnet_with_invalid_availability_zone(): 586 ec2 = boto3.resource("ec2", region_name="us-west-1") 587 client = boto3.client("ec2", region_name="us-west-1") 588 589 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 590 591 subnet_availability_zone = "asfasfas" 592 with pytest.raises(ClientError) as ex: 593 client.create_subnet( 594 VpcId=vpc.id, 595 CidrBlock="10.0.0.0/24", 596 AvailabilityZone=subnet_availability_zone, 597 ) 598 assert str(ex.value).startswith( 599 "An error occurred (InvalidParameterValue) when calling the CreateSubnet " 600 "operation: Value ({}) for parameter availabilityZone is invalid. Subnets can currently only be created in the following availability zones: ".format( 601 subnet_availability_zone 602 ) 603 ) 604 605 606@mock_ec2 607def test_create_subnet_with_invalid_cidr_range(): 608 ec2 = boto3.resource("ec2", region_name="us-west-1") 609 610 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 611 vpc.reload() 612 vpc.is_default.shouldnt.be.ok 613 614 subnet_cidr_block = "10.1.0.0/20" 615 with pytest.raises(ClientError) as ex: 616 ec2.create_subnet(VpcId=vpc.id, CidrBlock=subnet_cidr_block) 617 str(ex.value).should.equal( 618 "An error occurred (InvalidSubnet.Range) when calling the CreateSubnet " 619 "operation: The CIDR '{}' is invalid.".format(subnet_cidr_block) 620 ) 621 622 623@mock_ec2 624def test_create_subnet_with_invalid_cidr_range_multiple_vpc_cidr_blocks(): 625 ec2 = boto3.resource("ec2", region_name="us-west-1") 626 627 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 628 ec2.meta.client.associate_vpc_cidr_block(CidrBlock="10.1.0.0/16", VpcId=vpc.id) 629 vpc.reload() 630 vpc.is_default.shouldnt.be.ok 631 632 subnet_cidr_block = "10.2.0.0/20" 633 with pytest.raises(ClientError) as ex: 634 ec2.create_subnet(VpcId=vpc.id, CidrBlock=subnet_cidr_block) 635 str(ex.value).should.equal( 636 "An error occurred (InvalidSubnet.Range) when calling the CreateSubnet " 637 "operation: The CIDR '{}' is invalid.".format(subnet_cidr_block) 638 ) 639 640 641@mock_ec2 642def test_create_subnet_with_invalid_cidr_block_parameter(): 643 ec2 = boto3.resource("ec2", region_name="us-west-1") 644 645 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 646 vpc.reload() 647 vpc.is_default.shouldnt.be.ok 648 649 subnet_cidr_block = "1000.1.0.0/20" 650 with pytest.raises(ClientError) as ex: 651 ec2.create_subnet(VpcId=vpc.id, CidrBlock=subnet_cidr_block) 652 str(ex.value).should.equal( 653 "An error occurred (InvalidParameterValue) when calling the CreateSubnet " 654 "operation: Value ({}) for parameter cidrBlock is invalid. This is not a valid CIDR block.".format( 655 subnet_cidr_block 656 ) 657 ) 658 659 660@mock_ec2 661def test_create_subnets_with_multiple_vpc_cidr_blocks(): 662 ec2 = boto3.resource("ec2", region_name="us-west-1") 663 client = boto3.client("ec2", region_name="us-west-1") 664 665 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 666 ec2.meta.client.associate_vpc_cidr_block(CidrBlock="10.1.0.0/16", VpcId=vpc.id) 667 vpc.reload() 668 vpc.is_default.shouldnt.be.ok 669 670 subnet_cidr_block_primary = "10.0.0.0/24" 671 subnet_primary = ec2.create_subnet( 672 VpcId=vpc.id, CidrBlock=subnet_cidr_block_primary 673 ) 674 675 subnet_cidr_block_secondary = "10.1.0.0/24" 676 subnet_secondary = ec2.create_subnet( 677 VpcId=vpc.id, CidrBlock=subnet_cidr_block_secondary 678 ) 679 680 subnets = client.describe_subnets( 681 SubnetIds=[subnet_primary.id, subnet_secondary.id] 682 )["Subnets"] 683 subnets.should.have.length_of(2) 684 685 for subnet in subnets: 686 subnet.should.have.key("AvailabilityZone") 687 subnet.should.have.key("AvailabilityZoneId") 688 subnet.should.have.key("AvailableIpAddressCount") 689 subnet.should.have.key("CidrBlock") 690 subnet.should.have.key("State") 691 subnet.should.have.key("SubnetId") 692 subnet.should.have.key("VpcId") 693 subnet.shouldnt.have.key("Tags") 694 subnet.should.have.key("DefaultForAz").which.should.equal(False) 695 subnet.should.have.key("MapPublicIpOnLaunch").which.should.equal(False) 696 subnet.should.have.key("OwnerId") 697 subnet.should.have.key("AssignIpv6AddressOnCreation").which.should.equal(False) 698 699 700@mock_ec2 701def test_create_subnets_with_overlapping_cidr_blocks(): 702 ec2 = boto3.resource("ec2", region_name="us-west-1") 703 704 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 705 vpc.reload() 706 vpc.is_default.shouldnt.be.ok 707 708 subnet_cidr_block = "10.0.0.0/24" 709 with pytest.raises(ClientError) as ex: 710 ec2.create_subnet(VpcId=vpc.id, CidrBlock=subnet_cidr_block) 711 ec2.create_subnet(VpcId=vpc.id, CidrBlock=subnet_cidr_block) 712 str(ex.value).should.equal( 713 "An error occurred (InvalidSubnet.Conflict) when calling the CreateSubnet " 714 "operation: The CIDR '{}' conflicts with another subnet".format( 715 subnet_cidr_block 716 ) 717 ) 718 719 720@mock_ec2 721def test_create_subnet_with_tags(): 722 ec2 = boto3.resource("ec2", region_name="us-west-1") 723 vpc = ec2.create_vpc(CidrBlock="172.31.0.0/16") 724 725 random_ip = "172.31." + ".".join( 726 map(str, (random.randint(10, 40) for _ in range(2))) 727 ) 728 random_cidr = f"{random_ip}/20" 729 730 subnet = ec2.create_subnet( 731 VpcId=vpc.id, 732 CidrBlock=random_cidr, 733 AvailabilityZoneId="use1-az6", 734 TagSpecifications=[ 735 {"ResourceType": "subnet", "Tags": [{"Key": "name", "Value": "some-vpc"}]} 736 ], 737 ) 738 739 assert subnet.tags == [{"Key": "name", "Value": "some-vpc"}] 740 741 742@mock_ec2 743def test_available_ip_addresses_in_subnet(): 744 if settings.TEST_SERVER_MODE: 745 raise SkipTest( 746 "ServerMode is not guaranteed to be empty - other subnets will affect the count" 747 ) 748 ec2 = boto3.resource("ec2", region_name="us-west-1") 749 client = boto3.client("ec2", region_name="us-west-1") 750 751 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 752 cidr_range_addresses = [ 753 ("10.0.0.0/16", 65531), 754 ("10.0.0.0/17", 32763), 755 ("10.0.0.0/18", 16379), 756 ("10.0.0.0/19", 8187), 757 ("10.0.0.0/20", 4091), 758 ("10.0.0.0/21", 2043), 759 ("10.0.0.0/22", 1019), 760 ("10.0.0.0/23", 507), 761 ("10.0.0.0/24", 251), 762 ("10.0.0.0/25", 123), 763 ("10.0.0.0/26", 59), 764 ("10.0.0.0/27", 27), 765 ("10.0.0.0/28", 11), 766 ] 767 for (cidr, expected_count) in cidr_range_addresses: 768 validate_subnet_details(client, vpc, cidr, expected_count) 769 770 771@mock_ec2 772def test_available_ip_addresses_in_subnet_with_enis(): 773 if settings.TEST_SERVER_MODE: 774 raise SkipTest( 775 "ServerMode is not guaranteed to be empty - other ENI's will affect the count" 776 ) 777 ec2 = boto3.resource("ec2", region_name="us-west-1") 778 client = boto3.client("ec2", region_name="us-west-1") 779 780 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 781 # Verify behaviour for various CIDR ranges (...) 782 # Don't try to assign ENIs to /27 and /28, as there are not a lot of IP addresses to go around 783 cidr_range_addresses = [ 784 ("10.0.0.0/16", 65531), 785 ("10.0.0.0/17", 32763), 786 ("10.0.0.0/18", 16379), 787 ("10.0.0.0/19", 8187), 788 ("10.0.0.0/20", 4091), 789 ("10.0.0.0/21", 2043), 790 ("10.0.0.0/22", 1019), 791 ("10.0.0.0/23", 507), 792 ("10.0.0.0/24", 251), 793 ("10.0.0.0/25", 123), 794 ("10.0.0.0/26", 59), 795 ] 796 for (cidr, expected_count) in cidr_range_addresses: 797 validate_subnet_details_after_creating_eni(client, vpc, cidr, expected_count) 798 799 800def validate_subnet_details(client, vpc, cidr, expected_ip_address_count): 801 subnet = client.create_subnet( 802 VpcId=vpc.id, CidrBlock=cidr, AvailabilityZone="us-west-1b" 803 )["Subnet"] 804 subnet["AvailableIpAddressCount"].should.equal(expected_ip_address_count) 805 client.delete_subnet(SubnetId=subnet["SubnetId"]) 806 807 808def validate_subnet_details_after_creating_eni( 809 client, vpc, cidr, expected_ip_address_count 810): 811 subnet = client.create_subnet( 812 VpcId=vpc.id, CidrBlock=cidr, AvailabilityZone="us-west-1b" 813 )["Subnet"] 814 # Create a random number of Elastic Network Interfaces 815 nr_of_eni_to_create = random.randint(0, 5) 816 ip_addresses_assigned = 0 817 enis_created = [] 818 for _ in range(0, nr_of_eni_to_create): 819 # Create a random number of IP addresses per ENI 820 nr_of_ip_addresses = random.randint(1, 5) 821 if nr_of_ip_addresses == 1: 822 # Pick the first available IP address (First 4 are reserved by AWS) 823 private_address = "10.0.0." + str(ip_addresses_assigned + 4) 824 eni = client.create_network_interface( 825 SubnetId=subnet["SubnetId"], PrivateIpAddress=private_address 826 )["NetworkInterface"] 827 enis_created.append(eni) 828 ip_addresses_assigned = ip_addresses_assigned + 1 829 else: 830 # Assign a list of IP addresses 831 private_addresses = [ 832 "10.0.0." + str(4 + ip_addresses_assigned + i) 833 for i in range(0, nr_of_ip_addresses) 834 ] 835 eni = client.create_network_interface( 836 SubnetId=subnet["SubnetId"], 837 PrivateIpAddresses=[ 838 {"PrivateIpAddress": address} for address in private_addresses 839 ], 840 )["NetworkInterface"] 841 enis_created.append(eni) 842 ip_addresses_assigned = ip_addresses_assigned + nr_of_ip_addresses # 843 844 # Verify that the nr of available IP addresses takes these ENIs into account 845 updated_subnet = client.describe_subnets(SubnetIds=[subnet["SubnetId"]])["Subnets"][ 846 0 847 ] 848 849 private_addresses = [] 850 for eni in enis_created: 851 private_addresses.extend( 852 [address["PrivateIpAddress"] for address in eni["PrivateIpAddresses"]] 853 ) 854 error_msg = ( 855 "Nr of IP addresses for Subnet with CIDR {0} is incorrect. Expected: {1}, Actual: {2}. " 856 "Addresses: {3}" 857 ) 858 with sure.ensure( 859 error_msg, 860 cidr, 861 str(expected_ip_address_count), 862 updated_subnet["AvailableIpAddressCount"], 863 str(private_addresses), 864 ): 865 updated_subnet["AvailableIpAddressCount"].should.equal( 866 expected_ip_address_count - ip_addresses_assigned 867 ) 868 # Clean up, as we have to create a few more subnets that shouldn't interfere with each other 869 for eni in enis_created: 870 client.delete_network_interface(NetworkInterfaceId=eni["NetworkInterfaceId"]) 871 client.delete_subnet(SubnetId=subnet["SubnetId"]) 872 873 874@mock_ec2 875def test_run_instances_should_attach_to_default_subnet(): 876 # https://github.com/spulec/moto/issues/2877 877 ec2 = boto3.resource("ec2", region_name="sa-east-1") 878 client = boto3.client("ec2", region_name="sa-east-1") 879 sec_group_name = str(uuid4())[0:6] 880 ec2.create_security_group( 881 GroupName=sec_group_name, Description="Test security group sg01" 882 ) 883 # run_instances 884 instances = client.run_instances( 885 ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1, SecurityGroups=[sec_group_name], 886 ) 887 # Assert subnet is created appropriately 888 subnets = client.describe_subnets( 889 Filters=[{"Name": "defaultForAz", "Values": ["true"]}] 890 )["Subnets"] 891 default_subnet_id = subnets[0]["SubnetId"] 892 if len(subnets) > 1: 893 default_subnet_id1 = subnets[1]["SubnetId"] 894 assert ( 895 instances["Instances"][0]["NetworkInterfaces"][0]["SubnetId"] 896 == default_subnet_id 897 or instances["Instances"][0]["NetworkInterfaces"][0]["SubnetId"] 898 == default_subnet_id1 899 ) 900 901 if not settings.TEST_SERVER_MODE: 902 # Available IP addresses will depend on other resources that might be created in parallel 903 assert ( 904 subnets[0]["AvailableIpAddressCount"] == 4090 905 or subnets[1]["AvailableIpAddressCount"] == 4090 906 ) 907 908 909@mock_ec2 910def test_describe_subnets_by_vpc_id(): 911 ec2 = boto3.resource("ec2", region_name="us-west-1") 912 client = boto3.client("ec2", region_name="us-west-1") 913 914 vpc1 = ec2.create_vpc(CidrBlock="10.0.0.0/16") 915 subnet1 = ec2.create_subnet( 916 VpcId=vpc1.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-1a" 917 ) 918 vpc2 = ec2.create_vpc(CidrBlock="172.31.0.0/16") 919 subnet2 = ec2.create_subnet( 920 VpcId=vpc2.id, CidrBlock="172.31.48.0/20", AvailabilityZone="us-west-1b" 921 ) 922 923 subnets = client.describe_subnets( 924 Filters=[{"Name": "vpc-id", "Values": [vpc1.id]}] 925 ).get("Subnets", []) 926 subnets.should.have.length_of(1) 927 subnets[0]["SubnetId"].should.equal(subnet1.id) 928 929 subnets = client.describe_subnets( 930 Filters=[{"Name": "vpc-id", "Values": [vpc2.id]}] 931 ).get("Subnets", []) 932 subnets.should.have.length_of(1) 933 subnets[0]["SubnetId"].should.equal(subnet2.id) 934 935 # Specify multiple VPCs in Filter. 936 subnets = client.describe_subnets( 937 Filters=[{"Name": "vpc-id", "Values": [vpc1.id, vpc2.id]}] 938 ).get("Subnets", []) 939 subnets.should.have.length_of(2) 940 941 # Specify mismatched SubnetIds/Filters. 942 subnets = client.describe_subnets( 943 SubnetIds=[subnet1.id], Filters=[{"Name": "vpc-id", "Values": [vpc2.id]}] 944 ).get("Subnets", []) 945 subnets.should.have.length_of(0) 946 947 948@mock_ec2 949def test_describe_subnets_by_state(): 950 ec2 = boto3.resource("ec2", region_name="us-west-1") 951 client = boto3.client("ec2", region_name="us-west-1") 952 953 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 954 ec2.create_subnet( 955 VpcId=vpc.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-1a" 956 ) 957 958 subnets = client.describe_subnets( 959 Filters=[{"Name": "state", "Values": ["available"]}] 960 ).get("Subnets", []) 961 for subnet in subnets: 962 subnet["State"].should.equal("available") 963 964 965@mock_ec2 966def test_associate_subnet_cidr_block(): 967 ec2 = boto3.resource("ec2", region_name="us-west-1") 968 client = boto3.client("ec2", region_name="us-west-1") 969 970 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 971 subnet_object = ec2.create_subnet( 972 VpcId=vpc.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-1a" 973 ) 974 975 subnets = client.describe_subnets(SubnetIds=[subnet_object.id])["Subnets"] 976 association_set = subnets[0]["Ipv6CidrBlockAssociationSet"] 977 association_set.should.equal([]) 978 979 res = client.associate_subnet_cidr_block( 980 Ipv6CidrBlock="1080::1:200C:417A/112", SubnetId=subnet_object.id 981 ) 982 res.should.have.key("Ipv6CidrBlockAssociation") 983 association = res["Ipv6CidrBlockAssociation"] 984 association.should.have.key("AssociationId").match("subnet-cidr-assoc-[a-z0-9]+") 985 association.should.have.key("Ipv6CidrBlock").equals("1080::1:200C:417A/112") 986 association.should.have.key("Ipv6CidrBlockState").equals({"State": "associated"}) 987 988 subnets = client.describe_subnets(SubnetIds=[subnet_object.id])["Subnets"] 989 association_set = subnets[0]["Ipv6CidrBlockAssociationSet"] 990 association_set.should.have.length_of(1) 991 association_set[0].should.have.key("AssociationId").equal( 992 association["AssociationId"] 993 ) 994 association_set[0].should.have.key("Ipv6CidrBlock").equals("1080::1:200C:417A/112") 995 996 997@mock_ec2 998def test_disassociate_subnet_cidr_block(): 999 ec2 = boto3.resource("ec2", region_name="us-west-1") 1000 client = boto3.client("ec2", region_name="us-west-1") 1001 1002 vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") 1003 subnet_object = ec2.create_subnet( 1004 VpcId=vpc.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-1a" 1005 ) 1006 1007 client.associate_subnet_cidr_block( 1008 Ipv6CidrBlock="1080::1:200C:417A/111", SubnetId=subnet_object.id 1009 ) 1010 association_id = client.associate_subnet_cidr_block( 1011 Ipv6CidrBlock="1080::1:200C:417A/999", SubnetId=subnet_object.id 1012 )["Ipv6CidrBlockAssociation"]["AssociationId"] 1013 1014 subnets = client.describe_subnets(SubnetIds=[subnet_object.id])["Subnets"] 1015 association_set = subnets[0]["Ipv6CidrBlockAssociationSet"] 1016 association_set.should.have.length_of(2) 1017 1018 client.disassociate_subnet_cidr_block(AssociationId=association_id) 1019 1020 subnets = client.describe_subnets(SubnetIds=[subnet_object.id])["Subnets"] 1021 association_set = subnets[0]["Ipv6CidrBlockAssociationSet"] 1022 association_set.should.have.length_of(1) 1023 association_set[0]["Ipv6CidrBlock"].should.equal("1080::1:200C:417A/111") 1024 1025 1026@mock_ec2 1027def test_describe_subnets_dryrun(): 1028 client = boto3.client("ec2", region_name="us-east-1") 1029 1030 with pytest.raises(ClientError) as ex: 1031 client.describe_subnets(DryRun=True) 1032 ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412) 1033 ex.value.response["Error"]["Code"].should.equal("DryRunOperation") 1034 ex.value.response["Error"]["Message"].should.equal( 1035 "An error occurred (DryRunOperation) when calling the DescribeSubnets operation: Request would have succeeded, but DryRun flag is set" 1036 ) 1037