AWS ECS and Fargate Integration Testing: Complete Guide
Testing containerized workloads on AWS ECS and Fargate requires a different mindset than testing traditional server-based applications. Your containers are ephemeral, networking is abstracted, and the control plane sits between you and your code. This guide covers practical integration testing patterns for ECS and Fargate workloads.
Why ECS/Fargate Testing Is Different
When you run containers on EC2, you own the host. You can SSH in, tail logs, inspect network interfaces. ECS Fargate removes all of that — no SSH, no host access, no persistent storage. The tradeoff (managed infrastructure) comes with a testing challenge: how do you verify your containers behave correctly in the actual AWS environment?
Local Docker testing catches a lot, but it misses:
- IAM role permissions (task execution role, task role)
- VPC networking and security group behavior
- Service discovery via AWS Cloud Map
- ECS-specific environment injection
- Container dependency ordering in task definitions
You need integration tests that run in or against the actual AWS environment.
Test Strategy: Four Layers
A complete ECS testing strategy has four layers:
Layer 4: End-to-end (deployed service)
Layer 3: Task integration (ECS task against real AWS services)
Layer 2: Container integration (local Docker + mocked AWS)
Layer 1: Unit (business logic, no AWS)Most teams skip layers 2 and 3, then wonder why layer 4 tests fail mysteriously. Don't skip them.
Layer 1: Unit Tests
Write unit tests for your application code without any AWS dependencies. If you're using boto3, abstract it:
class TaskProcessor:
def __init__(self, s3_client, dynamodb_resource):
self.s3 = s3_client
self.dynamodb = dynamodb_resource
def process_batch(self, bucket: str, key: str) -> int:
# business logic here
obj = self.s3.get_object(Bucket=bucket, Key=key)
# process...
return countNow unit test TaskProcessor with mocks:
def test_process_batch():
mock_s3 = Mock()
mock_s3.get_object.return_value = {"Body": BytesIO(b"data")}
mock_dynamo = Mock()
processor = TaskProcessor(mock_s3, mock_dynamo)
count = processor.process_batch("my-bucket", "key.json")
assert count > 0
mock_s3.get_object.assert_called_once_with(Bucket="my-bucket", Key="key.json")Layer 2: Container Integration with LocalStack
Use LocalStack to run AWS services locally and test your containerized application against them:
# docker-compose.test.yml
version: "3.8"
services:
localstack:
image: localstack/localstack:3.0
environment:
- SERVICES=s3,dynamodb,sqs,iam
- DEFAULT_REGION=us-east-1
ports:
- "4566:4566"
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:4566/_localstack/health"]
interval: 10s
timeout: 5s
retries: 5
app:
build: .
environment:
- AWS_ENDPOINT_URL=http://localstack:4566
- AWS_DEFAULT_REGION=us-east-1
- AWS_ACCESS_KEY_ID=test
- AWS_SECRET_ACCESS_KEY=test
depends_on:
localstack:
condition: service_healthyYour application code needs to support the AWS_ENDPOINT_URL override:
import boto3
import os
def get_s3_client():
kwargs = {"region_name": os.getenv("AWS_DEFAULT_REGION", "us-east-1")}
if endpoint_url := os.getenv("AWS_ENDPOINT_URL"):
kwargs["endpoint_url"] = endpoint_url
return boto3.client("s3", **kwargs)Run container integration tests:
docker-compose -f docker-compose.test.yml run --rm app pytest tests/integration/Layer 3: Task Integration Testing
This layer runs your actual ECS task against real AWS services in a test account. Use a dedicated test VPC and test IAM roles.
Setting Up Test Task Definitions
Create a test variant of your task definition with reduced resources and test-specific environment:
{
"family": "my-service-test",
"taskRoleArn": "arn:aws:iam::123456789:role/my-service-task-role-test",
"executionRoleArn": "arn:aws:iam::123456789:role/ecsTaskExecutionRole",
"networkMode": "awsvpc",
"requiresCompatibilities": ["FARGATE"],
"cpu": "256",
"memory": "512",
"containerDefinitions": [
{
"name": "app",
"image": "${IMAGE_URI}",
"environment": [
{"name": "ENV", "value": "test"},
{"name": "TABLE_NAME", "value": "my-table-test"}
],
"logConfiguration": {
"logDriver": "awslogs",
"options": {
"awslogs-group": "/ecs/my-service-test",
"awslogs-region": "us-east-1",
"awslogs-stream-prefix": "ecs"
}
}
}
]
}Running Tasks and Waiting for Completion
import boto3
import time
def run_ecs_task_and_wait(
cluster: str,
task_definition: str,
subnet_id: str,
security_group_id: str,
timeout: int = 300
) -> dict:
ecs = boto3.client("ecs", region_name="us-east-1")
response = ecs.run_task(
cluster=cluster,
taskDefinition=task_definition,
launchType="FARGATE",
networkConfiguration={
"awsvpcConfiguration": {
"subnets": [subnet_id],
"securityGroups": [security_group_id],
"assignPublicIp": "DISABLED"
}
}
)
task_arn = response["tasks"][0]["taskArn"]
# Wait for task to complete
deadline = time.time() + timeout
while time.time() < deadline:
tasks = ecs.describe_tasks(cluster=cluster, tasks=[task_arn])
task = tasks["tasks"][0]
if task["lastStatus"] == "STOPPED":
exit_code = task["containers"][0].get("exitCode", -1)
return {
"task_arn": task_arn,
"exit_code": exit_code,
"stop_reason": task.get("stoppedReason", "")
}
time.sleep(10)
raise TimeoutError(f"Task {task_arn} did not complete in {timeout}s")
def test_batch_processing_task():
# Seed test data
s3 = boto3.client("s3")
s3.put_object(
Bucket="my-bucket-test",
Key="input/test-batch.json",
Body=b'[{"id": 1}, {"id": 2}]'
)
result = run_ecs_task_and_wait(
cluster="my-cluster-test",
task_definition="my-service-test",
subnet_id="subnet-xxxxx",
security_group_id="sg-xxxxx"
)
assert result["exit_code"] == 0, f"Task failed: {result['stop_reason']}"
# Verify output
dynamo = boto3.resource("dynamodb")
table = dynamo.Table("my-table-test")
response = table.scan()
assert len(response["Items"]) == 2Pulling Logs After Task Completion
def get_task_logs(log_group: str, stream_prefix: str, task_id: str) -> list[str]:
logs = boto3.client("logs")
stream_name = f"{stream_prefix}/app/{task_id}"
try:
events = logs.get_log_events(
logGroupName=log_group,
logStreamName=stream_name,
startFromHead=True
)
return [e["message"] for e in events["events"]]
except logs.exceptions.ResourceNotFoundException:
return []Include log retrieval in your test assertions — when a task fails, you want to know why immediately rather than hunting through CloudWatch.
Layer 4: Service End-to-End Tests
For ECS services (as opposed to tasks), test via the load balancer or service endpoint. Don't test internal IPs — they change.
import httpx
import pytest
SERVICE_URL = "https://my-service.test.internal"
@pytest.fixture(scope="session")
def wait_for_service():
"""Wait for ECS service to be healthy after deploy."""
import time
deadline = time.time() + 120
while time.time() < deadline:
try:
r = httpx.get(f"{SERVICE_URL}/health", timeout=5)
if r.status_code == 200:
return
except httpx.ConnectError:
pass
time.sleep(10)
raise RuntimeError("Service did not become healthy")
def test_api_returns_valid_response(wait_for_service):
response = httpx.get(f"{SERVICE_URL}/api/v1/items")
assert response.status_code == 200
data = response.json()
assert "items" in data
assert isinstance(data["items"], list)
def test_health_check_endpoint(wait_for_service):
response = httpx.get(f"{SERVICE_URL}/health")
assert response.status_code == 200
body = response.json()
assert body["status"] == "ok"Testing ECS Service Scaling
Verify your service scales correctly under load:
def test_service_scales_up():
ecs = boto3.client("ecs")
autoscaling = boto3.client("application-autoscaling")
cluster = "my-cluster-test"
service = "my-service-test"
# Get initial task count
initial = ecs.describe_services(
cluster=cluster,
services=[service]
)["services"][0]["runningCount"]
# Generate load (simplified — use locust or k6 in practice)
# ... load generation code ...
# Wait for scaling event (up to 5 minutes)
deadline = time.time() + 300
while time.time() < deadline:
current = ecs.describe_services(
cluster=cluster,
services=[service]
)["services"][0]["runningCount"]
if current > initial:
print(f"Scaled from {initial} to {current} tasks")
return
time.sleep(15)
pytest.fail(f"Service did not scale up from {initial} tasks")Testing Task Definition Changes
When you update a task definition (new image, changed environment variables, updated IAM permissions), test the new revision before deploying it to production:
def register_test_task_definition(base_family: str, image_uri: str) -> str:
ecs = boto3.client("ecs")
# Get current definition
current = ecs.describe_task_definition(taskDefinition=base_family)["taskDefinition"]
# Update image
containers = current["containerDefinitions"]
containers[0]["image"] = image_uri
# Register new revision
response = ecs.register_task_definition(
family=f"{base_family}-candidate",
taskRoleArn=current["taskRoleArn"],
executionRoleArn=current["executionRoleArn"],
networkMode=current["networkMode"],
containerDefinitions=containers,
requiresCompatibilities=current["requiresCompatibilities"],
cpu=current["cpu"],
memory=current["memory"]
)
return response["taskDefinition"]["taskDefinitionArn"]CI/CD Integration
Wire these tests into your pipeline so they run automatically on every deploy:
# .github/workflows/test.yml
jobs:
integration-test:
runs-on: ubuntu-latest
needs: build
steps:
- uses: actions/checkout@v4
- name: Configure AWS credentials
uses: aws-actions/configure-aws-credentials@v4
with:
role-to-assume: arn:aws:iam::123456789:role/github-actions-test
aws-region: us-east-1
- name: Run Layer 2 tests (LocalStack)
run: |
docker-compose -f docker-compose.test.yml run --rm app \
pytest tests/integration/layer2/ -v
- name: Run Layer 3 tests (ECS task)
env:
IMAGE_URI: ${{ needs.build.outputs.image_uri }}
run: |
pytest tests/integration/layer3/ -v \
--timeout=300
- name: Cleanup test resources
if: always()
run: |
python scripts/cleanup_test_resources.pyCommon Failure Patterns
Task stops immediately with exit code 1: Check CloudWatch logs first. Usually a missing environment variable or failed IAM permission. Add the logs:GetLogEvents permission to your test runner role.
Task stuck in PENDING: Security group blocking outbound traffic, NAT Gateway misconfigured, or ECR pull failing. Enable VPC Flow Logs to diagnose.
IAM permission denied: The task role and execution role are different. Execution role is for ECS infrastructure (pulling images, writing logs). Task role is for your application code. Permissions denied in application code → fix the task role.
Flaky container dependency tests: ECS doesn't wait for HEALTHCHECK in Dockerfile by default — use healthCheck in the container definition JSON, and the condition: HEALTHY dependency.
Monitoring Test Costs
ECS Fargate tests cost money. Keep them fast and clean up after:
- Use
cpu: 256, memory: 512for test task definitions — don't run tests on production-sized tasks - Set task timeouts — a stuck test task running for hours costs ~$0.04/hour, but it adds up
- Tag all test resources with
Environment=testand enforce cleanup via Lambda + CloudWatch Events - Run layer 3/4 tests only on PRs targeting main, not on every feature branch push
ECS and Fargate testing is an investment that pays off at deployment time. The teams that skip it are the ones debugging production incidents at 2am because a missing IAM permission only shows up when the task actually runs in AWS. Build the test infrastructure once, and it catches real problems every deploy.
For monitoring your ECS services in production after deploy, HelpMeTest's health check system integrates with your pipeline to run continuous functional tests against your services — no code required.
