test: implement comprehensive unit and integration test suite

Author: DessimA

tests/e2e/__init__.py

@@ -0,0 +1 @@
+# E2E tests

tests/integration/__init__.py

@@ -0,0 +1 @@
+# Integration tests

tests/integration/test_infrastructure.py

@@ -0,0 +1,44 @@
+import boto3
+import pytest
+import os
+
+# Use LocalStack endpoint if running in Docker/LocalStack env
+ENDPOINT_URL = os.environ.get("LOCALSTACK_ENDPOINT", "http://localhost:4566")
+
+@pytest.fixture
+def s3_client():
+    return boto3.client("s3", endpoint_url=ENDPOINT_URL, region_name="us-east-1")
+
+@pytest.fixture
+def dynamodb_client():
+    return boto3.client("dynamodb", endpoint_url=ENDPOINT_URL, region_name="us-east-1")
+
+@pytest.fixture
+def iam_client():
+    return boto3.client("iam", endpoint_url=ENDPOINT_URL, region_name="us-east-1")
+
+def test_s3_bucket_exists(s3_client):
+    # This assumes the bucket name from our terraform/main.tf default
+    bucket_name = "portfolio-cloud-dev-dev-portfolio-uploads"
+    try:
+        response = s3_client.head_bucket(Bucket=bucket_name)
+        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
+    except Exception as e:
+        pytest.fail(f"Bucket {bucket_name} does not exist or is not accessible: {e}")
+
+def test_dynamodb_table_exists(dynamodb_client):
+    table_name = "portfolio-cloud-dev-dev-metadata"
+    try:
+        response = dynamodb_client.describe_table(TableName=table_name)
+        assert response["Table"]["TableStatus"] == "ACTIVE"
+    except Exception as e:
+        pytest.fail(f"Table {table_name} does not exist: {e}")
+
+def test_iam_roles_exist(iam_client):
+    roles = ["upload_lambda_role", "metadata_lambda_role"]
+    for role_name in roles:
+        try:
+            response = iam_client.get_role(RoleName=role_name)
+            assert response["Role"]["RoleName"] == role_name
+        except Exception as e:
+            pytest.fail(f"IAM Role {role_name} does not exist: {e}")

tests/unit/test_sample.py

@@ -0,0 +1,13 @@
+import boto3
+import pytest
+from moto import mock_s3
+
+@mock_s3
+def test_s3_bucket_creation():
+    s3 = boto3.client("s3", region_name="us-east-1")
+    bucket_name = "test-bucket"
+    s3.create_bucket(Bucket=bucket_name)
+    
+    response = s3.list_buckets()
+    buckets = [bucket["Name"] for bucket in response["Buckets"]]
+    assert bucket_name in buckets

tests/unit/test_upload_lambda.py

@@ -0,0 +1,73 @@
+import json
+import base64
+import pytest
+import boto3
+from moto import mock_s3, mock_dynamodb
+from src.lambda.upload.handler import handler
+
+@pytest.fixture
+def aws_credentials():
+    """Mocked AWS Credentials for moto."""
+    import os
+    os.environ["AWS_ACCESS_KEY_ID"] = "testing"
+    os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
+    os.environ["AWS_SECURITY_TOKEN"] = "testing"
+    os.environ["AWS_SESSION_TOKEN"] = "testing"
+    os.environ["AWS_DEFAULT_REGION"] = "us-east-1"
+    os.environ["BUCKET_NAME"] = "test-bucket"
+    os.environ["DYNAMODB_TABLE"] = "test-table"
+
+@mock_s3
+@mock_dynamodb
+def test_upload_handler_success(aws_credentials):
+    # Setup mock S3
+    s3 = boto3.client("s3", region_name="us-east-1")
+    s3.create_bucket(Bucket="test-bucket")
+
+    # Setup mock DynamoDB
+    db = boto3.resource("dynamodb", region_name="us-east-1")
+    db.create_table(
+        TableName="test-table",
+        KeySchema=[
+            {"AttributeName": "project_id", "KeyType": "HASH"},
+            {"AttributeName": "timestamp", "KeyType": "RANGE"}
+        ],
+        AttributeDefinitions=[
+            {"AttributeName": "project_id", "AttributeType": "S"},
+            {"AttributeName": "timestamp", "AttributeType": "N"}
+        ],
+        ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 1}
+    )
+
+    # Mock event
+    file_content = b"hello world"
+    encoded_content = base64.b64encode(file_content).decode("utf-8")
+    event = {
+        "body": encoded_content,
+        "isBase64Encoded": True,
+        "headers": {
+            "file-name": "test.txt",
+            "Content-Type": "text/plain",
+            "project-id": "proj-123"
+        }
+    }
+
+    # Call handler
+    response = handler(event, None)
+
+    # Verify response
+    assert response["statusCode"] == 200
+    body = json.loads(response["body"])
+    assert body["message"] == "Upload e registro concluídos"
+
+    # Verify S3 content
+    obj = s3.get_object(Bucket="test-bucket", Key="uploads/proj-123/test.txt")
+    assert obj["Body"].read() == file_content
+
+    # Verify DynamoDB metadata
+    table = db.Table("test-table")
+    items = table.scan()["Items"]
+    assert len(items) == 1
+    assert items[0]["project_id"] == "proj-123"
+    assert items[0]["filename"] == "test.txt"
+    assert items[0]["status"] == "UPLOADED"