Integration Testing
Integration testing validates that your Terraform modules work correctly with real cloud resources, handling the complexity of actual API interactions, network configurations, and service dependencies. While unit tests catch logic errors, integration tests ensure your infrastructure actually functions as intended in real environments.
This part covers comprehensive integration testing strategies using Terratest, custom testing frameworks, and cloud-native testing approaches.
Terratest Fundamentals
Terratest is the most popular framework for testing Terraform modules with real infrastructure:
// test/integration/vpc_test.go
package test
import (
"testing"
"github.com/gruntwork-io/terratest/modules/terraform"
"github.com/gruntwork-io/terratest/modules/aws"
"github.com/stretchr/testify/assert"
)
func TestVPCModule(t *testing.T) {
t.Parallel()
// Pick a random AWS region to test in
awsRegion := aws.GetRandomStableRegion(t, nil, nil)
terraformOptions := &terraform.Options{
TerraformDir: "../modules/vpc",
Vars: map[string]interface{}{
"name": "test-vpc",
"cidr_block": "10.0.0.0/16",
"availability_zones": []string{awsRegion + "a", awsRegion + "b"},
"environment": "test",
},
EnvVars: map[string]string{
"AWS_DEFAULT_REGION": awsRegion,
},
}
// Clean up resources with "defer"
defer terraform.Destroy(t, terraformOptions)
// Deploy the infrastructure
terraform.InitAndApply(t, terraformOptions)
// Validate the infrastructure
vpcId := terraform.Output(t, terraformOptions, "vpc_id")
assert.NotEmpty(t, vpcId)
publicSubnetIds := terraform.OutputList(t, terraformOptions, "public_subnet_ids")
assert.Len(t, publicSubnetIds, 2)
privateSubnetIds := terraform.OutputList(t, terraformOptions, "private_subnet_ids")
assert.Len(t, privateSubnetIds, 2)
// Validate VPC properties using AWS SDK
vpc := aws.GetVpcById(t, vpcId, awsRegion)
assert.Equal(t, "10.0.0.0/16", *vpc.CidrBlock)
assert.True(t, *vpc.EnableDnsHostnames)
assert.True(t, *vpc.EnableDnsSupport)
// Validate subnets
for _, subnetId := range publicSubnetIds {
subnet := aws.GetSubnetById(t, subnetId, awsRegion)
assert.True(t, *subnet.MapPublicIpOnLaunch)
assert.Contains(t, []string{awsRegion + "a", awsRegion + "b"}, *subnet.AvailabilityZone)
}
for _, subnetId := range privateSubnetIds {
subnet := aws.GetSubnetById(t, subnetId, awsRegion)
assert.False(t, *subnet.MapPublicIpOnLaunch)
}
}
Testing Complex Infrastructure
Test complete application stacks with multiple components:
// test/integration/complete_app_test.go
package test
import (
"fmt"
"testing"
"time"
"github.com/gruntwork-io/terratest/modules/terraform"
"github.com/gruntwork-io/terratest/modules/aws"
"github.com/gruntwork-io/terratest/modules/http-helper"
"github.com/gruntwork-io/terratest/modules/retry"
"github.com/stretchr/testify/assert"
)
func TestCompleteApplication(t *testing.T) {
t.Parallel()
awsRegion := aws.GetRandomStableRegion(t, nil, nil)
terraformOptions := &terraform.Options{
TerraformDir: "../examples/complete-app",
Vars: map[string]interface{}{
"name": "test-app",
"environment": "test",
"region": awsRegion,
},
EnvVars: map[string]string{
"AWS_DEFAULT_REGION": awsRegion,
},
}
defer terraform.Destroy(t, terraformOptions)
terraform.InitAndApply(t, terraformOptions)
// Test VPC
vpcId := terraform.Output(t, terraformOptions, "vpc_id")
assert.NotEmpty(t, vpcId)
// Test RDS
dbEndpoint := terraform.Output(t, terraformOptions, "database_endpoint")
assert.NotEmpty(t, dbEndpoint)
// Test Load Balancer
albDnsName := terraform.Output(t, terraformOptions, "load_balancer_dns_name")
assert.NotEmpty(t, albDnsName)
// Test application health endpoint
url := fmt.Sprintf("http://%s/health", albDnsName)
// Retry the health check as the application may take time to start
retry.DoWithRetry(t, "Check application health", 30, 10*time.Second, func() (string, error) {
statusCode, body := http_helper.HttpGet(t, url, nil)
if statusCode != 200 {
return "", fmt.Errorf("Expected status 200, got %d", statusCode)
}
assert.Contains(t, body, "healthy")
return body, nil
})
// Test database connectivity through application
dbTestUrl := fmt.Sprintf("http://%s/db-test", albDnsName)
statusCode, body := http_helper.HttpGet(t, dbTestUrl, nil)
assert.Equal(t, 200, statusCode)
assert.Contains(t, body, "database_connected")
}
Testing with Multiple Environments
Test modules across different environment configurations:
// test/integration/multi_environment_test.go
package test
import (
"testing"
"github.com/gruntwork-io/terratest/modules/terraform"
"github.com/gruntwork-io/terratest/modules/aws"
"github.com/stretchr/testify/assert"
)
func TestMultiEnvironmentDeployment(t *testing.T) {
environments := []struct {
name string
instanceType string
minSize int
maxSize int
}{
{"dev", "t3.micro", 1, 3},
{"staging", "t3.small", 2, 5},
{"prod", "t3.medium", 3, 10},
}
for _, env := range environments {
env := env // Capture range variable
t.Run(env.name, func(t *testing.T) {
t.Parallel()
awsRegion := aws.GetRandomStableRegion(t, nil, nil)
terraformOptions := &terraform.Options{
TerraformDir: "../modules/auto-scaling-group",
Vars: map[string]interface{}{
"name": fmt.Sprintf("test-asg-%s", env.name),
"environment": env.name,
"instance_type": env.instanceType,
"min_size": env.minSize,
"max_size": env.maxSize,
"vpc_id": getTestVpcId(t, awsRegion),
"subnet_ids": getTestSubnetIds(t, awsRegion),
},
EnvVars: map[string]string{
"AWS_DEFAULT_REGION": awsRegion,
},
}
defer terraform.Destroy(t, terraformOptions)
terraform.InitAndApply(t, terraformOptions)
// Validate Auto Scaling Group
asgName := terraform.Output(t, terraformOptions, "asg_name")
asg := aws.GetAsgByName(t, asgName, awsRegion)
assert.Equal(t, int64(env.minSize), *asg.MinSize)
assert.Equal(t, int64(env.maxSize), *asg.MaxSize)
// Validate Launch Template
launchTemplateId := terraform.Output(t, terraformOptions, "launch_template_id")
launchTemplate := aws.GetLaunchTemplate(t, launchTemplateId, awsRegion)
assert.Equal(t, env.instanceType, *launchTemplate.LaunchTemplateData.InstanceType)
})
}
}
Testing Security Configurations
Validate security group rules and IAM policies:
// test/integration/security_test.go
package test
import (
"testing"
"github.com/gruntwork-io/terratest/modules/terraform"
"github.com/gruntwork-io/terratest/modules/aws"
"github.com/stretchr/testify/assert"
)
func TestSecurityGroupConfiguration(t *testing.T) {
t.Parallel()
awsRegion := aws.GetRandomStableRegion(t, nil, nil)
terraformOptions := &terraform.Options{
TerraformDir: "../modules/web-security-group",
Vars: map[string]interface{}{
"name": "test-web-sg",
"vpc_id": getTestVpcId(t, awsRegion),
"allowed_cidr_blocks": []string{"10.0.0.0/8"},
},
EnvVars: map[string]string{
"AWS_DEFAULT_REGION": awsRegion,
},
}
defer terraform.Destroy(t, terraformOptions)
terraform.InitAndApply(t, terraformOptions)
// Get security group
sgId := terraform.Output(t, terraformOptions, "security_group_id")
sg := aws.GetSecurityGroupById(t, sgId, awsRegion)
// Validate ingress rules
assert.Len(t, sg.IpPermissions, 2) // HTTP and HTTPS
for _, rule := range sg.IpPermissions {
if *rule.FromPort == 80 {
assert.Equal(t, int64(80), *rule.ToPort)
assert.Equal(t, "tcp", *rule.IpProtocol)
assert.Len(t, rule.IpRanges, 1)
assert.Equal(t, "10.0.0.0/8", *rule.IpRanges[0].CidrIp)
} else if *rule.FromPort == 443 {
assert.Equal(t, int64(443), *rule.ToPort)
assert.Equal(t, "tcp", *rule.IpProtocol)
}
}
// Validate egress rules
assert.Len(t, sg.IpPermissionsEgress, 1)
egressRule := sg.IpPermissionsEgress[0]
assert.Equal(t, int64(0), *egressRule.FromPort)
assert.Equal(t, int64(0), *egressRule.ToPort)
assert.Equal(t, "-1", *egressRule.IpProtocol)
}
func TestIAMRoleConfiguration(t *testing.T) {
t.Parallel()
terraformOptions := &terraform.Options{
TerraformDir: "../modules/iam-role",
Vars: map[string]interface{}{
"role_name": "test-role",
"service": "ec2.amazonaws.com",
"policies": []string{
"arn:aws:iam::aws:policy/AmazonS3ReadOnlyAccess",
},
},
}
defer terraform.Destroy(t, terraformOptions)
terraform.InitAndApply(t, terraformOptions)
// Validate IAM role
roleName := terraform.Output(t, terraformOptions, "role_name")
role := aws.GetIamRole(t, roleName)
assert.Equal(t, roleName, *role.RoleName)
assert.Contains(t, *role.AssumeRolePolicyDocument, "ec2.amazonaws.com")
// Validate attached policies
attachedPolicies := aws.GetIamRoleAttachedPolicies(t, roleName)
assert.Len(t, attachedPolicies, 1)
assert.Equal(t, "AmazonS3ReadOnlyAccess", *attachedPolicies[0].PolicyName)
}
Performance and Load Testing
Test infrastructure under load:
// test/integration/performance_test.go
package test
import (
"fmt"
"testing"
"time"
"sync"
"github.com/gruntwork-io/terratest/modules/terraform"
"github.com/gruntwork-io/terratest/modules/http-helper"
"github.com/stretchr/testify/assert"
)
func TestLoadBalancerPerformance(t *testing.T) {
t.Parallel()
terraformOptions := &terraform.Options{
TerraformDir: "../examples/load-balanced-app",
Vars: map[string]interface{}{
"name": "perf-test-app",
"environment": "test",
"min_capacity": 3,
"max_capacity": 10,
},
}
defer terraform.Destroy(t, terraformOptions)
terraform.InitAndApply(t, terraformOptions)
// Get load balancer DNS name
albDnsName := terraform.Output(t, terraformOptions, "load_balancer_dns_name")
url := fmt.Sprintf("http://%s/", albDnsName)
// Wait for application to be ready
http_helper.HttpGetWithRetry(t, url, nil, 200, "OK", 30, 10*time.Second)
// Perform load test
concurrentRequests := 50
requestsPerWorker := 20
var wg sync.WaitGroup
results := make(chan int, concurrentRequests*requestsPerWorker)
for i := 0; i < concurrentRequests; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for j := 0; j < requestsPerWorker; j++ {
statusCode, _ := http_helper.HttpGet(t, url, nil)
results <- statusCode
}
}()
}
wg.Wait()
close(results)
// Analyze results
successCount := 0
totalRequests := 0
for statusCode := range results {
totalRequests++
if statusCode == 200 {
successCount++
}
}
successRate := float64(successCount) / float64(totalRequests)
assert.GreaterOrEqual(t, successRate, 0.95, "Success rate should be at least 95%")
t.Logf("Load test results: %d/%d requests successful (%.2f%%)",
successCount, totalRequests, successRate*100)
}
Custom Testing Framework
Build a custom testing framework for specific needs:
#!/usr/bin/env python3
# custom_terraform_tester.py
import boto3
import subprocess
import json
import time
import requests
from typing import Dict, List, Any, Optional
class TerraformIntegrationTester:
def __init__(self, terraform_dir: str, aws_region: str = "us-west-2"):
self.terraform_dir = terraform_dir
self.aws_region = aws_region
self.aws_session = boto3.Session(region_name=aws_region)
self.outputs = {}
def deploy(self, variables: Dict[str, Any]) -> bool:
"""Deploy infrastructure with Terraform"""
try:
# Create tfvars file
tfvars_content = "\n".join([
f'{key} = {json.dumps(value)}'
for key, value in variables.items()
])
with open(f"{self.terraform_dir}/test.tfvars", "w") as f:
f.write(tfvars_content)
# Initialize
subprocess.run(
["terraform", "init"],
cwd=self.terraform_dir,
check=True,
capture_output=True
)
# Apply
result = subprocess.run(
["terraform", "apply", "-var-file=test.tfvars", "-auto-approve"],
cwd=self.terraform_dir,
check=True,
capture_output=True,
text=True
)
# Get outputs
output_result = subprocess.run(
["terraform", "output", "-json"],
cwd=self.terraform_dir,
check=True,
capture_output=True,
text=True
)
self.outputs = json.loads(output_result.stdout)
return True
except subprocess.CalledProcessError as e:
print(f"Terraform deployment failed: {e.stderr}")
return False
def destroy(self) -> bool:
"""Destroy infrastructure"""
try:
subprocess.run(
["terraform", "destroy", "-var-file=test.tfvars", "-auto-approve"],
cwd=self.terraform_dir,
check=True,
capture_output=True
)
return True
except subprocess.CalledProcessError as e:
print(f"Terraform destroy failed: {e.stderr}")
return False
def get_output(self, key: str) -> Any:
"""Get Terraform output value"""
return self.outputs.get(key, {}).get("value")
def test_vpc_configuration(self, expected_cidr: str) -> bool:
"""Test VPC configuration"""
vpc_id = self.get_output("vpc_id")
if not vpc_id:
return False
ec2 = self.aws_session.client("ec2")
response = ec2.describe_vpcs(VpcIds=[vpc_id])
if not response["Vpcs"]:
return False
vpc = response["Vpcs"][0]
return vpc["CidrBlock"] == expected_cidr
def test_application_health(self, timeout: int = 300) -> bool:
"""Test application health endpoint"""
load_balancer_dns = self.get_output("load_balancer_dns_name")
if not load_balancer_dns:
return False
url = f"http://{load_balancer_dns}/health"
start_time = time.time()
while time.time() - start_time < timeout:
try:
response = requests.get(url, timeout=10)
if response.status_code == 200:
return True
except requests.RequestException:
pass
time.sleep(10)
return False
def test_database_connectivity(self) -> bool:
"""Test database connectivity"""
db_endpoint = self.get_output("database_endpoint")
if not db_endpoint:
return False
# This would typically involve connecting to the database
# and running a simple query
rds = self.aws_session.client("rds")
try:
response = rds.describe_db_instances()
for db in response["DBInstances"]:
if db["Endpoint"]["Address"] == db_endpoint:
return db["DBInstanceStatus"] == "available"
except Exception as e:
print(f"Database connectivity test failed: {e}")
return False
# Example usage
def test_complete_application():
tester = TerraformIntegrationTester("../examples/complete-app")
variables = {
"name": "integration-test",
"environment": "test",
"instance_type": "t3.micro",
"min_size": 2,
"max_size": 4
}
try:
# Deploy
assert tester.deploy(variables), "Deployment failed"
# Test VPC
assert tester.test_vpc_configuration("10.0.0.0/16"), "VPC test failed"
# Test application
assert tester.test_application_health(), "Application health test failed"
# Test database
assert tester.test_database_connectivity(), "Database test failed"
print("All integration tests passed!")
finally:
# Clean up
tester.destroy()
if __name__ == "__main__":
test_complete_application()
What’s Next
Integration testing validates that your infrastructure works correctly with real cloud resources, but ensuring compliance and implementing governance requires policy-based validation that goes beyond functional testing.
In the next part, we’ll explore policy as code using Open Policy Agent (OPA) and Sentinel to implement automated governance, compliance validation, and security policy enforcement for your Terraform configurations.