Chapter 6: CDK Testing and Debugging

Learning Objectives

• Master how to write unit tests for CDK applications
• Understand the differences and uses of snapshot testing and fine-grained testing
• Learn to use CDK's built-in testing tools and assertions
• Master debugging techniques and troubleshooting methods for CDK applications
• Understand integration testing and end-to-end testing strategies

Knowledge Point Summary

CDK Testing Overview

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

Test Type Descriptions

Test Type	Purpose	Features	Execution Speed
Snapshot Test	Detect unexpected changes	Compares the generated CloudFormation template	Fast
Fine-grained Test	Validate specific properties	Checks the specific configuration of resources	Fast
Integration Test	Test component interactions	Verifies relationships between multiple resources	Medium
End-to-End Test	Full functional validation	Actual deployment and functional testing	Slow

Unit Testing in Practice

Setting up the Test Environment

# 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 to make 'tests' a Python package

Snapshot Testing

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

# 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 - detects template changes"""
    app = core.App()
    stack = WebAppStack(app, "TestWebAppStack")
    
    # Generate the 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 full resource definitions
            # The first run creates a snapshot, subsequent runs compare against it
        }
    })

def test_s3_bucket_snapshot():
    """S3 bucket snapshot test"""
    app = core.App()
    stack = WebAppStack(app, "TestStack")
    template = assertions.Template.from_stack(stack)
    
    # Verify that an S3 bucket is included
    template.has_resource_properties("AWS::S3::Bucket", {
        "VersioningConfiguration": {
            "Status": "Enabled"
        }
    })

Fine-grained Testing

Fine-grained testing is used to validate 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 that the Lambda function has S3 access permissions"""
    app = core.App()
    stack = WebAppStack(app, "TestStack")
    template = assertions.Template.from_stack(stack)
    
    # Check for the 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 the 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 the integration between API Gateway and Lambda"""
    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 a custom construct using mocks"""
    app = core.App()
    stack = core.Stack(app, "TestStack")
    
    with patch('boto3.client') as mock_client:
        # Mock the RDS client response
        mock_rds = Mock()
        mock_rds.describe_db_instances.return_value = {
            'DBInstances': [
                {'DBInstanceStatus': 'available'}
            ]
        }
        mock_client.return_value = mock_rds
        
        # Create the construct
        db_construct = DatabaseConstruct(
            stack, 
            "TestDatabase",
            instance_type="db.t3.micro"
        )
        
        # Verify that the construct was created successfully
        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 based on 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 that the 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 that the database security group allows access from the 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 the complete application"""
    app = create_app()
    
    # Synthesize all Stacks
    assembly = app.synth()
    
    # Verify that the 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 that there are 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):
    """Debugging tool: output the 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

# Example usage
def test_debug_resources():
    app = core.App()
    stack = WebAppStack(app, "DebugStack")
    
    # Output the full 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 to create the test Stack")
    
    app = core.App()
    stack = WebAppStack(app, "DebugTestStack")
    template = assertions.Template.from_stack(stack)
    
    logger.info("Stack creation complete, starting resource validation")
    
    # Get all resources for debugging
    template_dict = template.to_json()
    resources = template_dict.get('Resources', {})
    
    logger.debug(f"A total of {len(resources)} resources were created")
    for logical_id, resource in resources.items():
        logger.debug(f"  - {logical_id}: {resource['Type']}")
    
    # Execute the 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 diagnostic tools class"""
    
    def __init__(self, stack: core.Stack):
        self.stack = stack
        self.template = assertions.Template.from_stack(stack)
    
    def check_common_issues(self):
        """Check for common issues"""
        issues = []
        template_dict = self.template.to_json()
        
        # Check for hard-coded 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 hard-coded ARNs"""
        issues = []
        # Implement logic to check for hard-coded ARNs
        return issues
    
    def _check_overpermissive_policies(self, template_dict):
        """Check for over-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 logic to check naming conventions
        return issues

# Example usage
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

Summary of Testing Best Practices

1. Testing Pyramid: Numerous unit tests + a moderate number of integration tests + a few end-to-end tests
2. Fast Feedback: Prioritize running fast unit tests
3. Test Isolation: Each test should be independent and not rely on the state of other tests
4. Meaningful Assertions: Tests should validate business logic, not just technical implementation
5. Continuous Maintenance: Update test cases as the code changes

Troubleshooting Guide

Common Testing Issues

Issue	Cause	Solution
Template comparison fails	Resource order or format mismatch	Use Match.any_value() or broader matching
Incorrect resource count	Implicitly created resources	Check for resources automatically created by CDK (e.g., IAM roles)
Permission test fails	Complex policy document format	Use assertions.Match.array_with()
Slow test execution	Numerous resource creations	Use mocks and stubs to reduce real resource creation

Debugging Steps

1. Output the template: Use cdk synth to view the generated CloudFormation template
2. Validate step-by-step: Start with simple resource counts and gradually increase assertion complexity
3. Use logging: Add detailed log output in your tests
4. Isolate the problem: Create a minimal test case to reproduce the issue
5. Consult the documentation: Refer to CDK and CloudFormation documentation to confirm expected behavior

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