Chapter 6: CDK Testing and Debugging

Knowledge Summary

CDK Testing Overview

CDK supports multiple levels of testing, from unit tests to integration tests, ensuring the quality and reliability of infrastructure code.

graph TD
    A[CDK Testing Pyramid] --> B[Unit Tests]
    A --> C[Integration Tests]
    A --> D[End-to-End Tests]

    B --> B1[Snapshot Tests]
    B --> B2[Fine-Grained Tests]
    B --> B3[Validation Tests]

    C --> C1[Stack Integration Tests]
    C --> C2[Cross-Service Tests]

    D --> D1[Deployment Tests]
    D --> D2[Functional Tests]

    style B fill:#e8f5e8
    style C fill:#fff2cc
    style D fill:#fce4ec

Test Type Description

Unit Testing Practice

Test Environment Setup

# requirements-dev.txt
pytest>=7.0.0
aws-cdk-lib>=2.50.0
constructs>=10.0.0
pytest-mock>=3.8.0

# Install test dependencies
# pip install -r requirements-dev.txt

# tests/__init__.py
# Empty file, makes tests a Python package

Snapshot Testing

Snapshot testing is the most basic type of test, used to detect unexpected changes in CloudFormation templates.

# tests/unit/test_snapshot.py
import aws_cdk as core
import aws_cdk.assertions as assertions
from my_cdk_app.web_app_stack import WebAppStack

def test_snapshot():
    """Snapshot test - detect template changes"""
    app = core.App()
    stack = WebAppStack(app, "TestWebAppStack")

    # Generate template
    template = assertions.Template.from_stack(stack)

    # Snapshot assertion - will generate a snapshot file for comparison
    template.template_matches({
        "AWSTemplateFormatVersion": "2010-09-09",
        "Resources": {
            # This will contain the complete resource definitions
            # First run creates the snapshot, subsequent runs compare differences
        }
    })

def test_s3_bucket_snapshot():
    """S3 bucket snapshot test"""
    app = core.App()
    stack = WebAppStack(app, "TestStack")
    template = assertions.Template.from_stack(stack)

    # Verify S3 bucket is included
    template.has_resource_properties("AWS::S3::Bucket", {
        "VersioningConfiguration": {
            "Status": "Enabled"
        }
    })

Fine-Grained Testing

Fine-grained testing is used to verify specific properties and configurations of resources.

# tests/unit/test_fine_grained.py
import aws_cdk as core
import aws_cdk.assertions as assertions
from my_cdk_app.web_app_stack import WebAppStack

class TestWebAppStack:
    def setup_method(self):
        """Test initialization"""
        self.app = core.App()
        self.stack = WebAppStack(self.app, "TestWebAppStack")
        self.template = assertions.Template.from_stack(self.stack)

    def test_s3_bucket_properties(self):
        """Test S3 bucket properties"""
        self.template.has_resource_properties("AWS::S3::Bucket", {
            "VersioningConfiguration": {
                "Status": "Enabled"
            },
            "BucketEncryption": {
                "ServerSideEncryptionConfiguration": [
                    {
                        "ServerSideEncryptionByDefault": {
                            "SSEAlgorithm": "AES256"
                        }
                    }
                ]
            }
        })

    def test_lambda_function_configuration(self):
        """Test Lambda function configuration"""
        self.template.has_resource_properties("AWS::Lambda::Function", {
            "Runtime": "python3.9",
            "Handler": "index.handler",
            "Timeout": 30,
            "Environment": {
                "Variables": {
                    "BUCKET_NAME": assertions.Match.any_value()
                }
            }
        })

    def test_resource_count(self):
        """Test resource count"""
        self.template.resource_count_is("AWS::S3::Bucket", 1)
        self.template.resource_count_is("AWS::Lambda::Function", 2)
        self.template.resource_count_is("AWS::IAM::Role", 2)

    def test_iam_role_policies(self):
        """Test IAM role policies"""
        self.template.has_resource_properties("AWS::IAM::Role", {
            "AssumeRolePolicyDocument": {
                "Statement": [
                    {
                        "Effect": "Allow",
                        "Principal": {
                            "Service": "lambda.amazonaws.com"
                        },
                        "Action": "sts:AssumeRole"
                    }
                ]
            }
        })

Validation Testing

Validation testing is used to check references and dependencies between resources.

# tests/unit/test_validation.py
import aws_cdk as core
import aws_cdk.assertions as assertions
from my_cdk_app.web_app_stack import WebAppStack

def test_lambda_s3_permissions():
    """Verify Lambda function has S3 access permissions"""
    app = core.App()
    stack = WebAppStack(app, "TestStack")
    template = assertions.Template.from_stack(stack)

    # Check for Lambda execution role
    template.has_resource_properties("AWS::IAM::Role", {
        "AssumeRolePolicyDocument": {
            "Statement": assertions.Match.array_with([
                {
                    "Effect": "Allow",
                    "Principal": {
                        "Service": "lambda.amazonaws.com"
                    },
                    "Action": "sts:AssumeRole"
                }
            ])
        }
    })

    # Check for S3 access policy
    template.has_resource_properties("AWS::IAM::Policy", {
        "PolicyDocument": {
            "Statement": assertions.Match.array_with([
                {
                    "Effect": "Allow",
                    "Action": [
                        "s3:GetObject",
                        "s3:PutObject"
                    ],
                    "Resource": assertions.Match.any_value()
                }
            ])
        }
    })

def test_api_gateway_lambda_integration():
    """Verify API Gateway and Lambda integration"""
    app = core.App()
    stack = WebAppStack(app, "TestStack")
    template = assertions.Template.from_stack(stack)

    # Check Lambda permissions
    template.has_resource_properties("AWS::Lambda::Permission", {
        "Action": "lambda:InvokeFunction",
        "Principal": "apigateway.amazonaws.com"
    })

Advanced Testing Techniques

Using Mocks and Stubs

# tests/unit/test_with_mocks.py
import pytest
from unittest.mock import patch, Mock
import aws_cdk as core
from my_cdk_app.custom_construct import DatabaseConstruct

def test_database_construct_with_mock():
    """Test custom construct using Mock"""
    app = core.App()
    stack = core.Stack(app, "TestStack")

    with patch('boto3.client') as mock_client:
        # Mock RDS client response
        mock_rds = Mock()
        mock_rds.describe_db_instances.return_value = {
            'DBInstances': [
                {'DBInstanceStatus': 'available'}
            ]
        }
        mock_client.return_value = mock_rds

        # Create construct
        db_construct = DatabaseConstruct(
            stack,
            "TestDatabase",
            instance_type="db.t3.micro"
        )

        # Verify construct creation succeeded
        assert db_construct is not None

Parameterized Testing

# tests/unit/test_parameterized.py
import pytest
import aws_cdk as core
import aws_cdk.assertions as assertions
from my_cdk_app.web_app_stack import WebAppStack

@pytest.mark.parametrize("environment,expected_instance_count", [
    ("dev", 1),
    ("staging", 2),
    ("prod", 3)
])
def test_instance_count_by_environment(environment, expected_instance_count):
    """Parameterized test for instance count by environment"""
    app = core.App()
    stack = WebAppStack(
        app,
        f"TestStack-{environment}",
        env_name=environment
    )

    template = assertions.Template.from_stack(stack)
    template.resource_count_is(
        "AWS::EC2::Instance",
        expected_instance_count
    )

@pytest.mark.parametrize("runtime,handler", [
    ("python3.9", "index.handler"),
    ("nodejs18.x", "index.handler"),
    ("java11", "com.example.Handler")
])
def test_lambda_runtime_configurations(runtime, handler):
    """Test Lambda configurations for different runtimes"""
    app = core.App()
    stack = WebAppStack(
        app,
        "TestStack",
        lambda_runtime=runtime,
        lambda_handler=handler
    )

    template = assertions.Template.from_stack(stack)
    template.has_resource_properties("AWS::Lambda::Function", {
        "Runtime": runtime,
        "Handler": handler
    })

Integration Testing

Stack-Level Integration Testing

# tests/integration/test_stack_integration.py
import aws_cdk as core
import aws_cdk.assertions as assertions
from my_cdk_app.web_app_stack import WebAppStack
from my_cdk_app.database_stack import DatabaseStack

class TestStackIntegration:
    def setup_method(self):
        """Integration test initialization"""
        self.app = core.App()
        self.db_stack = DatabaseStack(self.app, "TestDatabaseStack")
        self.web_stack = WebAppStack(
            self.app,
            "TestWebStack",
            database=self.db_stack.database
        )

    def test_cross_stack_references(self):
        """Test cross-stack references"""
        web_template = assertions.Template.from_stack(self.web_stack)

        # Verify Web Stack references the database
        web_template.has_resource_properties("AWS::Lambda::Function", {
            "Environment": {
                "Variables": {
                    "DB_HOST": assertions.Match.any_value(),
                    "DB_PORT": "5432"
                }
            }
        })

    def test_security_group_rules(self):
        """Test security group rules"""
        db_template = assertions.Template.from_stack(self.db_stack)
        web_template = assertions.Template.from_stack(self.web_stack)

        # Verify database security group allows access from web tier
        db_template.has_resource_properties("AWS::EC2::SecurityGroupIngress", {
            "IpProtocol": "tcp",
            "FromPort": 5432,
            "ToPort": 5432
        })

Application-Level Integration Testing

# tests/integration/test_app_integration.py
import aws_cdk as core
from my_cdk_app.app import create_app

def test_complete_application():
    """Test complete application"""
    app = create_app()

    # Synthesize all Stacks
    assembly = app.synth()

    # Verify expected Stacks were generated
    stack_names = [stack.stack_name for stack in assembly.stacks]
    expected_stacks = ["DatabaseStack", "WebAppStack", "MonitoringStack"]

    for expected in expected_stacks:
        assert any(expected in name for name in stack_names), f"Missing stack: {expected}"

    # Verify no synthesis errors
    assert len(assembly.messages) == 0, f"Synthesis errors: {assembly.messages}"

Debugging Techniques

CDK Debugging Tools

# debug_utils.py
import json
import aws_cdk as core
from aws_cdk import assertions

def debug_stack_template(stack: core.Stack, output_file: str = None):
    """Debug tool: output Stack template"""
    template = assertions.Template.from_stack(stack)
    template_json = template.to_json()

    if output_file:
        with open(output_file, 'w') as f:
            json.dump(template_json, f, indent=2)
        print(f"Template saved to {output_file}")
    else:
        print(json.dumps(template_json, indent=2))

def find_resource_by_type(stack: core.Stack, resource_type: str):
    """Find resources of a specific type"""
    template = assertions.Template.from_stack(stack)
    template_dict = template.to_json()

    resources = template_dict.get('Resources', {})
    found_resources = []

    for logical_id, resource in resources.items():
        if resource.get('Type') == resource_type:
            found_resources.append({
                'LogicalId': logical_id,
                'Type': resource_type,
                'Properties': resource.get('Properties', {})
            })

    return found_resources

# Usage example
def test_debug_resources():
    app = core.App()
    stack = WebAppStack(app, "DebugStack")

    # Output complete template
    debug_stack_template(stack, "debug_template.json")

    # Find all Lambda functions
    lambdas = find_resource_by_type(stack, "AWS::Lambda::Function")
    print(f"Found {len(lambdas)} Lambda functions:")
    for lamb in lambdas:
        print(f"  - {lamb['LogicalId']}: {lamb['Properties'].get('Handler')}")

Conditional Breakpoints and Logging

# tests/unit/test_with_debugging.py
import logging
import aws_cdk as core
import aws_cdk.assertions as assertions
from my_cdk_app.web_app_stack import WebAppStack

# Configure logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

def test_with_debugging():
    """Test with debugging information"""
    logger.info("Starting test Stack creation")

    app = core.App()
    stack = WebAppStack(app, "DebugTestStack")
    template = assertions.Template.from_stack(stack)

    logger.info("Stack creation complete, starting resource verification")

    # Get all resources for debugging
    template_dict = template.to_json()
    resources = template_dict.get('Resources', {})

    logger.debug(f"Total of {len(resources)} resources created")
    for logical_id, resource in resources.items():
        logger.debug(f"  - {logical_id}: {resource['Type']}")

    # Execute actual test
    template.resource_count_is("AWS::Lambda::Function", 1)
    logger.info("Test passed")

Error Diagnostic Tools

# tests/utils/diagnostic_tools.py
import aws_cdk as core
import aws_cdk.assertions as assertions

class CDKDiagnostics:
    """CDK diagnostics tool class"""

    def __init__(self, stack: core.Stack):
        self.stack = stack
        self.template = assertions.Template.from_stack(stack)

    def check_common_issues(self):
        """Check common issues"""
        issues = []
        template_dict = self.template.to_json()

        # Check for hardcoded ARNs
        issues.extend(self._check_hardcoded_arns(template_dict))

        # Check for insecure permissions
        issues.extend(self._check_overpermissive_policies(template_dict))

        # Check resource naming
        issues.extend(self._check_resource_naming(template_dict))

        return issues

    def _check_hardcoded_arns(self, template_dict):
        """Check for hardcoded ARNs"""
        issues = []
        # Implement hardcoded ARN check logic
        return issues

    def _check_overpermissive_policies(self, template_dict):
        """Check for overly permissive policies"""
        issues = []
        resources = template_dict.get('Resources', {})

        for logical_id, resource in resources.items():
            if resource.get('Type') == 'AWS::IAM::Policy':
                statements = resource.get('Properties', {}).get('PolicyDocument', {}).get('Statement', [])
                for statement in statements:
                    if statement.get('Resource') == '*' and statement.get('Effect') == 'Allow':
                        issues.append(f"Policy {logical_id} has overly broad permissions")

        return issues

    def _check_resource_naming(self, template_dict):
        """Check resource naming conventions"""
        issues = []
        # Implement naming convention check logic
        return issues

# Usage example
def test_diagnostic_check():
    app = core.App()
    stack = WebAppStack(app, "DiagnosticTestStack")

    diagnostics = CDKDiagnostics(stack)
    issues = diagnostics.check_common_issues()

    if issues:
        print("Issues found:")
        for issue in issues:
            print(f"  - {issue}")

        # If there are critical issues, fail the test
        critical_issues = [i for i in issues if "overly broad permissions" in i]
        assert len(critical_issues) == 0, f"Critical security issues found: {critical_issues}"
    else:
        print("No issues found")

Testing Best Practices

1. Test Organization Structure

tests/
├── unit/                 # Unit tests
│   ├── test_stacks.py   # Stack tests
│   ├── test_constructs.py # Construct tests
│   └── test_utils.py    # Utility function tests
├── integration/         # Integration tests
│   ├── test_app.py      # Application-level tests
│   └── test_cross_stack.py # Cross-stack tests
├── e2e/                 # End-to-end tests
│   └── test_deployment.py
├── fixtures/            # Test data
│   └── sample_config.json
└── utils/               # Test utilities
    ├── __init__.py
    └── test_helpers.py

2. Test Configuration

# pytest.ini
[tool:pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts =
    -v
    --tb=short
    --strict-markers
    --disable-warnings
markers =
    unit: Unit tests
    integration: Integration tests
    e2e: End-to-end tests
    slow: Slow running tests

3. Test Data Management

# tests/fixtures/test_data.py
from dataclasses import dataclass
from typing import Dict, Any

@dataclass
class TestConfiguration:
    """Test configuration data class"""
    environment: str
    instance_count: int
    lambda_memory: int
    database_instance_type: str

    @classmethod
    def dev_config(cls):
        return cls(
            environment="dev",
            instance_count=1,
            lambda_memory=128,
            database_instance_type="db.t3.micro"
        )

    @classmethod
    def prod_config(cls):
        return cls(
            environment="prod",
            instance_count=3,
            lambda_memory=512,
            database_instance_type="db.r5.large"
        )

# Using test data
def test_dev_configuration():
    config = TestConfiguration.dev_config()
    app = core.App()
    stack = WebAppStack(app, "TestStack", config=config)
    # Test logic...

4. Continuous Integration Configuration

# .github/workflows/test.yml
name: CDK Tests

on:
  push:
    branches: [ main, develop ]
  pull_request:
    branches: [ main ]

jobs:
  test:
    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v3

    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.9'

    - name: Set up Node.js
      uses: actions/setup-node@v3
      with:
        node-version: '18'

    - name: Install dependencies
      run: |
        npm install -g aws-cdk
        pip install -r requirements.txt
        pip install -r requirements-dev.txt

    - name: Run unit tests
      run: pytest tests/unit -v --cov=my_cdk_app

    - name: Run integration tests
      run: pytest tests/integration -v
      if: github.event_name == 'push'

    - name: CDK synth check
      run: cdk synth --all
      env:
        AWS_DEFAULT_REGION: us-east-1

Troubleshooting Guide

Common Testing Issues

Debugging Steps

1. Output Template: Use cdk synth to view generated CloudFormation template
2. Step-by-Step Verification: Start with simple resource counts, gradually increase assertion complexity
3. Use Logging: Add detailed log output in tests
4. Isolate Issues: Create minimal test cases to reproduce the problem
5. Consult Documentation: Refer to CDK and CloudFormation documentation to confirm expected behavior

Through this chapter, you should be able to write comprehensive tests for CDK applications, ensuring the quality and reliability of infrastructure code.