Test improvements

Author: michaelhtm

test/e2e/tests/helper.py

@@ -15,6 +15,7 @@
 """
 
 from typing import Union, Dict
+import time
 
 
 class EC2Validator:
@@ -214,11 +215,26 @@ def assert_transit_gateway(self, tgw_id: str, exists=True):
             pass
         assert res_found is exists
 
+    def wait_transit_gateway_state(self, tgw_id: str, state: str):
+        is_state = False
+        max_tries = 5
+        try:
+            for tries in range(max_tries):
+                transit_gateway = self.ec2_client.describe_transit_gateways(TransitGatewayIds=[tgw_id])
+                if transit_gateway['TransitGateways'][0]['State'] == state:
+                    is_state=True
+                    break
+                else:
+                    time.sleep(30)
+        except:
+            pass
+        return is_state
+
     def get_transit_gateway_vpc_attachment(self, attachment_id: str) -> Union[None, Dict]:
         try:
             aws_res = self.ec2_client.describe_transit_gateway_vpc_attachments(TransitGatewayAttachmentIds=[attachment_id])
-            if len(aws_res["TransitGateways"]) > 0:
-                return aws_res["TransitGateways"][0]
+            if len(aws_res["TransitGatewayVpcAttachments"]) > 0:
+                return aws_res["TransitGatewayVpcAttachments"][0]
             return None
         except self.ec2_client.exceptions.ClientError:
             return None

test/e2e/tests/test_transitgateway_vpc_attachment.py

@@ -34,14 +34,21 @@
 DELETE_WAIT_AFTER_SECONDS = 180
 MAX_WAIT_FOR_SYNCED_SECONDS = 180
 MODIFY_WAIT_AFTER_SECONDS = 30
+WAIT_PERIOD = 30
 
 @pytest.fixture(scope="module")
-def simple_tgw_attachment(request):
+def simple_tgw_attachment(request, ec2_client):
     test_resource_values = REPLACEMENT_VALUES.copy()
     resource_name = random_suffix_name("tgw-attachment-test", 24)
 
     test_vpc = get_bootstrap_resources().SharedTestVPC
     test_tgw = get_bootstrap_resources().TestTransitGateway
+
+    tgw_id = test_tgw.transit_gateway_id
+
+    ec2_validator = EC2Validator(ec2_client)
+    is_available = ec2_validator.wait_transit_gateway_state(tgw_id=tgw_id, state='available')
+    assert is_available
 
     test_resource_values["TGWVA_NAME"] = resource_name
     test_resource_values["VPC_ID"] = test_vpc.vpc_id
@@ -96,7 +103,7 @@ def test_crud_tags(self, ec2_client, simple_tgw_attachment):
             ref,
             "ACK.ResourceSynced",
             "True",
-            wait_periods=MAX_WAIT_FOR_SYNCED_SECONDS,
+            wait_periods=WAIT_PERIOD,
         )
 
         time.sleep(CREATE_WAIT_AFTER_SECONDS)
@@ -143,20 +150,6 @@ def test_crud_tags(self, ec2_client, simple_tgw_attachment):
             actual=attachment["Tags"],
         )
 
-    @pytest.mark.resource_data({'tag_key': 'initialtagkey', 'tag_value': 'initialtagvalue'})
-    def test_update_options(self, ec2_client, simple_tgw_attachment):
-        (ref, cr) = simple_tgw_attachment
-        
-
-        assert k8s.wait_on_condition(
-            ref,
-            "ACK.ResourceSynced",
-            "True",
-            wait_periods=MAX_WAIT_FOR_SYNCED_SECONDS,
-        )
-
-        attachment_id = cr["status"]["id"]
-
         # Update options
         updates = {
             "spec": {
@@ -171,7 +164,7 @@ def test_update_options(self, ec2_client, simple_tgw_attachment):
         time.sleep(MODIFY_WAIT_AFTER_SECONDS)
 
         # Check resource synced successfully
-        assert k8s.wait_on_condition(ref, "ACK.ResourceSynced", "True", wait_periods=5)
+        assert k8s.wait_on_condition(ref, "ACK.ResourceSynced", "True", wait_periods=WAIT_PERIOD)
 
         # Verify the update in AWS
         ec2_validator = EC2Validator(ec2_client)
@@ -180,20 +173,6 @@ def test_update_options(self, ec2_client, simple_tgw_attachment):
         assert attachment["Options"]["DnsSupport"] == "disable"
         assert attachment["Options"]["Ipv6Support"] == "enable"
 
-    @pytest.mark.resource_data({'tag_key': 'initialtagkey', 'tag_value': 'initialtagvalue'})
-    def test_update_subnet_ids(self, ec2_client, simple_tgw_attachment):
-        (ref, cr) = simple_tgw_attachment
-        
-        
-        assert k8s.wait_on_condition(
-            ref,
-            "ACK.ResourceSynced",
-            "True",
-            wait_periods=MAX_WAIT_FOR_SYNCED_SECONDS,
-        )
-
-        attachment_id = cr["status"]["id"]
-
         # Update subnet ids
         test_vpc = get_bootstrap_resources().SharedTestVPC
         updates = {
@@ -206,10 +185,10 @@ def test_update_subnet_ids(self, ec2_client, simple_tgw_attachment):
         time.sleep(MODIFY_WAIT_AFTER_SECONDS)
 
         # Check resource synced successfully
-        assert k8s.wait_on_condition(ref, "ACK.ResourceSynced", "True", wait_periods=5)
+        assert k8s.wait_on_condition(ref, "ACK.ResourceSynced", "True", wait_periods=WAIT_PERIOD)
 
         # Verify the update in AWS
         ec2_validator = EC2Validator(ec2_client)
         attachment = ec2_validator.get_transit_gateway_vpc_attachment(attachment_id)
 
-        assert attachment["SubnetIds"] == test_vpc.public_subnets.subnet_ids
+        assert set(attachment["SubnetIds"]) == set(test_vpc.public_subnets.subnet_ids)