Test improvements

Author: michaelhtm

test/e2e/tests/helper.py

@@ -15,6 +15,7 @@
 """
 
 from typing import Union, Dict
+import time
 
 
 class EC2Validator:
@@ -214,6 +215,23 @@ 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:
+                    print(f'try: {tries}')
+                    print(transit_gateway['TransitGateways'][0]['State'])
+                    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])

test/e2e/tests/test_transitgateway_vpc_attachment.py

@@ -36,12 +36,18 @@
 MODIFY_WAIT_AFTER_SECONDS = 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