run integration tests for CDK

.github/workflows/cdk.yml

@@ -68,6 +68,19 @@ jobs:
           awslocal dynamodb list-tables
           awslocal s3 ls
 
+      - name: Set up test dependencies
+        run: |
+          pip install requests boto3 pytest localstack-sdk-python
+
+      - name: Run Integration Tests
+        env:
+          AWS_DEFAULT_REGION: us-east-1
+          AWS_REGION: us-east-1
+          AWS_ACCESS_KEY_ID: test
+          AWS_SECRET_ACCESS_KEY: test
+        run: |
+          pytest tests/test_infra.py
+
       - name: Send a Slack notification
         if: failure() || github.event_name != 'pull_request'
         uses: ravsamhq/notify-slack-action@v2

cdk/quiz_app/quiz_app_stack.py

@@ -158,7 +158,7 @@ def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
         self.backend_api_url = rest_api.url
 
         # verify email identity for SES
-        for email in ["your.email@example.com", "admin@localstack.cloud"]:
+        for email in ["your.email@example.com", "admin@localstack.cloud", "sender@example.com"]:
             sanitised_email = email.replace(".", "-").replace("@", "-")
             cr.AwsCustomResource(
                 self,
@@ -262,6 +262,7 @@ def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
                 "../configurations/statemachine.json"
             ),
             role=state_machine_role,
+            state_machine_name="SendEmailStateMachine"
         )
 
         # set up lambda permissions

tests/test_infra.py

@@ -156,11 +156,26 @@ def test_quiz_workflow(api_endpoint):
 
     time.sleep(5)
 
-    response = requests.get(f"{api_endpoint}/getleaderboard?quiz_id={quiz_id}&top=3")
-    assert response.status_code == 200
-    leaderboard = response.json()
-    assert len(leaderboard) == 3
+    leaderboard_url = f"{api_endpoint}/getleaderboard?quiz_id={quiz_id}&top=3"
+    response = requests.get(leaderboard_url)
+    leaderboard = None
 
+    if response.json():
+        assert response.status_code == 200
+        leaderboard = response.json()
+    else:
+        # If the response is empty, retry it for 5 times with a 2 second delay.
+        # TODO: This is a hack to get around the fact that the leaderboard is not available immediately.
+        for _ in range(5):
+            time.sleep(2)
+            response = requests.get(leaderboard_url)
+            if response.json():
+                assert response.status_code == 200
+                leaderboard = response.json()
+                break
+
+    assert leaderboard is not None, "Failed to retrieve leaderboard data after retries"
+    assert len(leaderboard) == 3
     expected_scores = {
         "user1": None,
         "user2": None,