EC2 - create_default_vpc (#5212)

moto/ec2/exceptions.py

@@ -28,6 +28,14 @@ class EC2ClientError(RESTError):
         super().__init__(*args, **kwargs)
 
 
+class DefaultVpcAlreadyExists(EC2ClientError):
+    def __init__(self):
+        super().__init__(
+            "DefaultVpcAlreadyExists",
+            "A Default VPC already exists for this account in this region.",
+        )
+
+
 class DependencyViolationError(EC2ClientError):
     def __init__(self, message):
         super().__init__("DependencyViolation", message)

moto/ec2/models/__init__.py

@@ -151,7 +151,7 @@ class EC2Backend(
         #   docs.aws.amazon.com/AmazonVPC/latest/UserGuide/default-vpc.html
         #
         if not self.vpcs:
-            vpc = self.create_vpc("172.31.0.0/16")
+            vpc = self.create_default_vpc()
         else:
             # For now this is included for potential
             # backward-compatibility issues

moto/ec2/models/vpcs.py

@@ -10,6 +10,7 @@ from .core import TaggedEC2Resource
 from ..exceptions import (
     CidrLimitExceeded,
     UnsupportedTenancy,
+    DefaultVpcAlreadyExists,
     DependencyViolationError,
     InvalidCIDRBlockParameterError,
     InvalidServiceName,
@@ -332,12 +333,20 @@ class VPCBackend:
         self.vpc_end_points = {}
         self.vpc_refs[self.__class__].add(weakref.ref(self))
 
+    def create_default_vpc(self):
+        default_vpc = self.describe_vpcs(filters={"is-default": "true"})
+        if default_vpc:
+            raise DefaultVpcAlreadyExists
+        cidr_block = "172.31.0.0/16"
+        return self.create_vpc(cidr_block=cidr_block, is_default=True)
+
     def create_vpc(
         self,
         cidr_block,
         instance_tenancy="default",
         amazon_provided_ipv6_cidr_block=False,
         tags=None,
+        is_default=False,
     ):
         vpc_id = random_vpc_id()
         try:
@@ -350,9 +359,9 @@ class VPCBackend:
             self,
             vpc_id,
             cidr_block,
-            len(self.vpcs) == 0,
-            instance_tenancy,
-            amazon_provided_ipv6_cidr_block,
+            is_default=is_default,
+            instance_tenancy=instance_tenancy,
+            amazon_provided_ipv6_cidr_block=amazon_provided_ipv6_cidr_block,
         )
 
         for tag in tags or []:

moto/ec2/responses/vpcs.py

@@ -12,6 +12,12 @@ class VPCs(EC2BaseResponse):
             else "2016-11-15"
         )
 
+    def create_default_vpc(self):
+        vpc = self.ec2_backend.create_default_vpc()
+        doc_date = self._get_doc_date()
+        template = self.response_template(CREATE_VPC_RESPONSE)
+        return template.render(vpc=vpc, doc_date=doc_date)
+
     def create_vpc(self):
         cidr_block = self._get_param("CidrBlock")
         tags = self._get_multi_param("TagSpecification")

tests/test_ec2/test_vpcs.py

@@ -6,7 +6,8 @@ import boto3
 import sure  # noqa # pylint: disable=unused-import
 import random
 
-from moto import mock_ec2
+from moto import mock_ec2, settings
+from unittest import SkipTest
 from uuid import uuid4
 from .test_tags import retrieve_all_tagged
 
@@ -14,6 +15,52 @@ SAMPLE_DOMAIN_NAME = "example.com"
 SAMPLE_NAME_SERVERS = ["10.0.0.6", "10.0.0.7"]
 
 
+@mock_ec2
+def test_creating_a_vpc_in_empty_region_does_not_make_this_vpc_the_default():
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest("Lets not start deleting VPC's while other tests are using it")
+    # Delete VPC that's created by default
+    client = boto3.client("ec2", region_name="eu-north-1")
+    all_vpcs = retrieve_all_vpcs(client)
+    for vpc in all_vpcs:
+        client.delete_vpc(VpcId=vpc["VpcId"])
+    # create vpc
+    client.create_vpc(CidrBlock="10.0.0.0/16")
+    # verify this is not the default
+    all_vpcs = retrieve_all_vpcs(client)
+    all_vpcs.should.have.length_of(1)
+    all_vpcs[0].should.have.key("IsDefault").equals(False)
+
+
+@mock_ec2
+def test_create_default_vpc():
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest("Lets not start deleting VPC's while other tests are using it")
+    # Delete VPC that's created by default
+    client = boto3.client("ec2", region_name="eu-north-1")
+    all_vpcs = retrieve_all_vpcs(client)
+    for vpc in all_vpcs:
+        client.delete_vpc(VpcId=vpc["VpcId"])
+    # create default vpc
+    client.create_default_vpc()
+    # verify this is the default
+    all_vpcs = retrieve_all_vpcs(client)
+    all_vpcs.should.have.length_of(1)
+    all_vpcs[0].should.have.key("IsDefault").equals(True)
+
+
+@mock_ec2
+def test_create_multiple_default_vpcs():
+    client = boto3.client("ec2", region_name="eu-north-1")
+    with pytest.raises(ClientError) as exc:
+        client.create_default_vpc()
+    err = exc.value.response["Error"]
+    err["Code"].should.equal("DefaultVpcAlreadyExists")
+    err["Message"].should.equal(
+        "A Default VPC already exists for this account in this region."
+    )
+
+
 @mock_ec2
 def test_create_and_delete_vpc():
     ec2 = boto3.resource("ec2", region_name="eu-north-1")