Add empty VPC and e2e tests

Author: Nicholas Thomson

test/e2e/bootstrap_resources.py

@@ -21,6 +21,7 @@
 @dataclass
 class BootstrapResources(Resources):
     SharedTestVPC: VPC
+    EmptyVPC: VPC
 
 _bootstrap_resources = None
 

test/e2e/resources/internet_gateway_vpc_attachment.yaml

@@ -0,0 +1,6 @@
+apiVersion: ec2.services.k8s.aws/v1alpha1
+kind: InternetGateway
+metadata:
+  name: $INTERNET_GATEWAY_NAME
+spec:
+  vpc: $VPC_ID

test/e2e/service_bootstrap.py

@@ -24,7 +24,8 @@ def service_bootstrap() -> Resources:
     logging.getLogger().setLevel(logging.INFO)
 
     resources = BootstrapResources(
-        SharedTestVPC=VPC(name_prefix="e2e-test-vpc", num_public_subnet=1, num_private_subnet=0)
+        SharedTestVPC=VPC(name_prefix="e2e-test-vpc", num_public_subnet=1, num_private_subnet=0),
+        EmptyVPC=VPC(name_prefix="e2e-empty-vpc", num_public_subnet=0, num_private_subnet=0)
     )
 
     try:

test/e2e/tests/helper.py

@@ -30,14 +30,17 @@ def assert_dhcp_options(self, dhcp_options_id: str, exists=True):
             pass
         assert res_found is exists
 
-    def assert_internet_gateway(self, ig_id: str, exists=True):
-        res_found = False
+    def get_internet_gateway(self, igw_id: str):
         try:
-            aws_res = self.ec2_client.describe_internet_gateways(InternetGatewayIds=[ig_id])
-            res_found = len(aws_res["InternetGateways"]) > 0
+            aws_res = self.ec2_client.describe_internet_gateways(InternetGatewayIds=[igw_id])
+            if len(aws_res["InternetGateways"]) > 0:
+                return aws_res["InternetGateways"][0]
+            return None
         except self.ec2_client.exceptions.ClientError:
-            pass
-        assert res_found is exists
+            return None
+
+    def assert_internet_gateway(self, igw_id: str, exists=True):
+        assert (self.get_internet_gateway(igw_id) is not None) == exists
 
     def assert_nat_gateway(self, ngw_id: str, exists=True):
         res_found = False

test/e2e/tests/test_internet_gateway.py

@@ -23,10 +23,12 @@
 from e2e import service_marker, CRD_GROUP, CRD_VERSION, load_ec2_resource
 from e2e.replacement_values import REPLACEMENT_VALUES
 from e2e.tests.helper import EC2Validator
+from e2e.bootstrap_resources import get_bootstrap_resources
 
 RESOURCE_PLURAL = "internetgateways"
 
 CREATE_WAIT_AFTER_SECONDS = 10
+MODIFY_WAIT_AFTER_SECONDS = 10
 DELETE_WAIT_AFTER_SECONDS = 10
 
 @service_marker
@@ -70,5 +72,75 @@ def test_create_delete(self, ec2_client):
 
         time.sleep(DELETE_WAIT_AFTER_SECONDS)
 
+        # Check Internet Gateway no longer exists in AWS
+        ec2_validator.assert_internet_gateway(resource_id, exists=False)
+
+    def test_vpc_association(self, ec2_client):
+        resource_name = random_suffix_name("ig-ack-test", 24)
+
+        test_vpc = get_bootstrap_resources().EmptyVPC
+        vpc_id = test_vpc.vpc_id
+
+        replacements = REPLACEMENT_VALUES.copy()
+        replacements["INTERNET_GATEWAY_NAME"] = resource_name
+        replacements["VPC_ID"] = vpc_id
+
+        # Load Internet Gateway CR
+        resource_data = load_ec2_resource(
+            "internet_gateway_vpc_attachment",
+            additional_replacements=replacements,
+        )
+        logging.debug(resource_data)
+
+        # Create k8s resource
+        ref = k8s.CustomResourceReference(
+            CRD_GROUP, CRD_VERSION, RESOURCE_PLURAL,
+            resource_name, namespace="default",
+        )
+        k8s.create_custom_resource(ref, resource_data)
+        cr = k8s.wait_resource_consumed_by_controller(ref)
+
+        assert cr is not None
+        assert k8s.get_resource_exists(ref)
+
+        resource = k8s.get_resource(ref)
+        resource_id = resource["status"]["internetGatewayID"]
+
+        time.sleep(CREATE_WAIT_AFTER_SECONDS)
+
+        # Check Internet Gateway exists in AWS
+        ec2_validator = EC2Validator(ec2_client)
+        ec2_validator.assert_internet_gateway(resource_id)
+
+        # Check attachments appear on Internet Gateway
+        igw = ec2_validator.get_internet_gateway(resource_id)
+        assert len(igw["Attachments"]) == 1
+        assert igw["Attachments"][0]["VpcId"] == vpc_id
+        assert igw["Attachments"][0]["State"] == "available"
+
+        # Patch the IGW, removing the attachment
+        updates = {
+            "spec": {"vpc": None},
+        }
+        k8s.patch_custom_resource(ref, updates)
+        time.sleep(MODIFY_WAIT_AFTER_SECONDS)
+
+        # Check there are no attachments on Internet Gateway
+        igw = ec2_validator.get_internet_gateway(resource_id)
+
+        # In the case where it shows the attachment as being in detached state
+        if len(igw["Attachments"]) == 1:
+            assert igw["Attachments"][0]["VpcId"] == vpc_id
+            assert igw["Attachments"][0]["State"] == "detached"
+        else:
+            # Otherwise there should be no attachment on the IGW
+            assert len(igw["Attachments"]) == 0
+
+        # Delete k8s resource
+        _, deleted = k8s.delete_custom_resource(ref, 2, 5)
+        assert deleted is True
+
+        time.sleep(DELETE_WAIT_AFTER_SECONDS)
+
         # Check Internet Gateway no longer exists in AWS
         ec2_validator.assert_internet_gateway(resource_id, exists=False)