add infra tests using pytest and github actions (#3)

Author: HarshCasper

.github/workflows/integration-test.yml

@@ -0,0 +1,95 @@
+name: Run Integration Tests
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+      - main
+  schedule:
+    # “At 00:00 on Sunday.”
+    - cron: "0 0 * * 0"
+  workflow_dispatch:
+    inputs:
+      runner-os:
+        default: ubuntu-latest
+        type: choice
+        options:
+          - ubuntu-latest
+
+
+jobs:
+  run-it-tests-job:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Set up Python 3.11
+        id: setup-python
+        uses: actions/setup-python@v2
+        with:
+          python-version: 3.11
+
+      - name: Setup Node.js
+        uses: actions/setup-node@v4
+        with:
+          node-version: 20
+
+      - name: Set up Dependencies
+        run: |
+          pip install requests boto3 pytest
+
+      - name: Start LocalStack
+        uses: LocalStack/setup-localstack@v0.2.3
+        with:
+          image-tag: 'latest'
+          use-pro: 'true'
+          configuration: LS_LOG=trace
+          install-awslocal: 'true'
+        env:
+          LOCALSTACK_API_KEY: ${{ secrets.LOCALSTACK_API_KEY }}
+
+      - name: Deploy infrastructure
+        run: |
+          bash bin/deploy.sh
+
+      - name: Run 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 -v
+
+      - name: Show localstack logs
+        if: always()
+        run: |
+          localstack logs
+
+      - name: Send a Slack notification
+        if: failure() || github.event_name != 'pull_request'
+        uses: ravsamhq/notify-slack-action@v2
+        with:
+          status: ${{ job.status }}
+          token: ${{ secrets.GITHUB_TOKEN }}
+          notification_title: "{workflow} has {status_message}"
+          message_format: "{emoji} *{workflow}* {status_message} in <{repo_url}|{repo}>"
+          footer: "Linked Repo <{repo_url}|{repo}> | <{run_url}|View Workflow run>"
+          notify_when: "failure"
+        env:
+          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
+
+      - name: Generate a Diagnostic Report
+        if: failure()
+        run: |
+          curl -s localhost:4566/_localstack/diagnose | gzip -cf > diagnose.json.gz
+
+      - name: Upload the Diagnostic Report
+        if: failure()
+        uses: actions/upload-artifact@v3
+        with:
+          name: diagnose.json.gz
+          path: ./diagnose.json.gz

.gitignore

@@ -12,3 +12,4 @@ dist/
 .venv
 env/
 venv/
+__pycache__

tests/test_infra.py

@@ -0,0 +1,233 @@
+import pytest
+import boto3
+import requests
+import json
+import time
+
+@pytest.fixture(scope='module')
+def api_endpoint():
+    apigateway_client = boto3.client('apigateway', endpoint_url='http://localhost:4566')
+    lambda_client = boto3.client('lambda', endpoint_url='http://localhost:4566')
+
+    API_NAME = 'QuizAPI'
+    response = apigateway_client.get_rest_apis()
+    api_list = response.get('items', [])
+    api = next((item for item in api_list if item['name'] == API_NAME), None)
+
+    if not api:
+        raise Exception(f"API {API_NAME} not found.")
+
+    API_ID = api['id']
+    API_ENDPOINT = f"http://localhost:4566/restapis/{API_ID}/test/_user_request_"
+
+    print(f"API Endpoint: {API_ENDPOINT}")
+
+    time.sleep(2)
+
+    return API_ENDPOINT
+
+def test_quiz_workflow(api_endpoint):
+    create_quiz_payload = {
+        "Title": "Sample Quiz",
+        "Visibility": "Public",
+        "EnableTimer": True,
+        "TimerSeconds": 10,
+        "Questions": [
+            {
+                "QuestionText": "What is the capital of France?",
+                "Options": ["A. Berlin", "B. London", "C. Madrid", "D. Paris"],
+                "CorrectAnswer": "D. Paris",
+                "Trivia": "Paris is known as the City of Light."
+            },
+            {
+                "QuestionText": "Who wrote Hamlet?",
+                "Options": ["A. Dickens", "B. Shakespeare", "C. Twain", "D. Hemingway"],
+                "CorrectAnswer": "B. Shakespeare",
+                "Trivia": "Shakespeare is often called England's national poet."
+            },
+            {
+                "QuestionText": "What is the largest planet in our solar system?",
+                "Options": ["A. Earth", "B. Mars", "C. Jupiter", "D. Saturn"],
+                "CorrectAnswer": "C. Jupiter",
+                "Trivia": "Jupiter is so large that all the other planets in the solar system could fit inside it."
+            },
+            {
+                "QuestionText": "Which element has the chemical symbol 'O'?",
+                "Options": ["A. Gold", "B. Oxygen", "C. Silver", "D. Iron"],
+                "CorrectAnswer": "B. Oxygen",
+                "Trivia": "Oxygen makes up about 21% of the Earth's atmosphere."
+            },
+            {
+                "QuestionText": "In which year did World War II end?",
+                "Options": ["A. 1943", "B. 1945", "C. 1947", "D. 1950"],
+                "CorrectAnswer": "B. 1945",
+                "Trivia": "The war ended with the surrender of Japan on September 2, 1945."
+            }
+        ]
+    }
+
+    response = requests.post(
+        f"{api_endpoint}/createquiz",
+        headers={"Content-Type": "application/json"},
+        data=json.dumps(create_quiz_payload)
+    )
+
+    assert response.status_code == 200
+    quiz_creation_response = response.json()
+    assert 'QuizID' in quiz_creation_response
+    quiz_id = quiz_creation_response['QuizID']
+
+    print(f"Quiz created with ID: {quiz_id}")
+
+    response = requests.get(f"{api_endpoint}/listquizzes")
+    assert response.status_code == 200
+    quizzes_list = response.json().get('Quizzes', [])
+    quiz_titles = [quiz['Title'] for quiz in quizzes_list]
+    assert "Sample Quiz" in quiz_titles
+
+    response = requests.get(f"{api_endpoint}/getquiz?quiz_id={quiz_id}")
+    assert response.status_code == 200
+    quiz_details = response.json()
+    assert quiz_details['Title'] == "Sample Quiz"
+    assert len(quiz_details['Questions']) == 5
+
+    submissions = []
+    users = [
+        {
+            "Username": "user1",
+            "Answers": {
+                "0": {"Answer": "D. Paris", "TimeTaken": 8},
+                "1": {"Answer": "B. Shakespeare", "TimeTaken": 5},
+                "2": {"Answer": "C. Jupiter", "TimeTaken": 6},
+                "3": {"Answer": "B. Oxygen", "TimeTaken": 7},
+                "4": {"Answer": "B. 1945", "TimeTaken": 9}
+            }
+        },
+        {
+            "Username": "user2",
+            "Email": "user@example.com",
+            "Answers": {
+                "0": {"Answer": "D. Paris", "TimeTaken": 7},
+                "1": {"Answer": "B. Shakespeare", "TimeTaken": 6},
+                "2": {"Answer": "D. Saturn", "TimeTaken": 5},  # Incorrect
+                "3": {"Answer": "B. Oxygen", "TimeTaken": 8},
+                "4": {"Answer": "B. 1945", "TimeTaken": 10}
+            }
+        },
+        {
+            "Username": "user3",
+            "Answers": {
+                "0": {"Answer": "A. Berlin", "TimeTaken": 9},  # Incorrect
+                "1": {"Answer": "D. Hemingway", "TimeTaken": 4},  # Incorrect
+                "2": {"Answer": "C. Jupiter", "TimeTaken": 11},  # Exceeds time
+                "3": {"Answer": "B. Oxygen", "TimeTaken": 12},  # Exceeds time
+                "4": {"Answer": "B. 1945", "TimeTaken": 13}     # Exceeds time
+            }
+        }
+    ]
+
+    for user in users:
+        submission_payload = {
+            "Username": user["Username"],
+            "QuizID": quiz_id,
+            "Answers": user["Answers"]
+        }
+        if "Email" in user:
+            submission_payload["Email"] = user["Email"]
+
+        response = requests.post(
+            f"{api_endpoint}/submitquiz",
+            headers={"Content-Type": "application/json"},
+            data=json.dumps(submission_payload)
+        )
+        assert response.status_code == 200
+        submission_response = response.json()
+        assert 'SubmissionID' in submission_response
+        submissions.append({
+            "Username": user["Username"],
+            "SubmissionID": submission_response["SubmissionID"]
+        })
+
+        print(f"{user['Username']} submitted quiz with SubmissionID: {submission_response['SubmissionID']}")
+
+    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
+
+    expected_scores = {
+        "user1": None,
+        "user2": None,
+        "user3": None
+    }
+
+    max_score = 100
+    timer_seconds = 10
+
+    correct_answers = ["D. Paris", "B. Shakespeare", "C. Jupiter", "B. Oxygen", "B. 1945"]
+
+    def calculate_user_score(user_answers):
+        score = 0
+        for idx, correct_answer in enumerate(correct_answers):
+            user_answer = user_answers[str(idx)]["Answer"]
+            time_taken = user_answers[str(idx)]["TimeTaken"]
+            if user_answer == correct_answer and time_taken <= timer_seconds:
+                question_score = max_score * (1 - (time_taken / timer_seconds))
+                score += max(0, question_score)
+            else:
+                pass
+        return score
+
+    for user in users:
+        username = user["Username"]
+        expected_scores[username] = calculate_user_score(user["Answers"])
+
+    for entry in leaderboard:
+        username = entry["Username"]
+        actual_score = entry["Score"]
+        expected_score = expected_scores[username]
+        assert actual_score == pytest.approx(expected_score, abs=0.01)
+
+        print(f"{username} - Expected Score: {expected_score}, Actual Score: {actual_score}")
+
+    for submission in submissions:
+        response = requests.get(f"{api_endpoint}/getsubmission?submission_id={submission['SubmissionID']}")
+        assert response.status_code == 200
+        submission_data = response.json()
+        assert submission_data['Username'] == submission['Username']
+        assert submission_data['QuizID'] == quiz_id
+        assert 'Score' in submission_data
+        assert 'UserAnswers' in submission_data
+
+        expected_score = expected_scores[submission['Username']]
+        actual_score = submission_data['Score']
+        assert actual_score == pytest.approx(expected_score, abs=0.01)
+
+        print(f"Verified submission for {submission['Username']} with Score: {actual_score}")
+
+    ses_endpoint = "http://localhost:4566/_aws/ses"
+    ses_response = requests.get(ses_endpoint)
+    assert ses_response.status_code == 200
+    ses_data = ses_response.json()
+
+    messages = ses_data.get('messages', [])
+    sender_email = "sender@example.com"
+    email_found = False
+
+    for message in messages:
+        if message.get('Source') == sender_email:
+            email_found = True
+            assert 'Id' in message
+            assert 'Region' in message
+            assert 'Timestamp' in message
+            assert 'Destination' in message
+            assert 'Subject' in message
+            assert 'Body' in message
+
+            body = message['Body']
+            assert 'html_part' in body
+            html_content = body['html_part']
+
+    assert email_found, f"No email found sent from {sender_email}"