Amazon ECS — moto Gaps, Smoke Tests & CI¶

# tests/test_moto_gaps.py (excerpt)
def test_run_task_crashes_when_no_container_has_memory(ecs, cluster, network, unique):
    family = unique("nomem")
    ecs.register_task_definition(family=family, requiresCompatibilities=["FARGATE"], networkMode="awsvpc", cpu="256", memory="512", containerDefinitions=PLAIN)
    with pytest.raises(TypeError):                                                    # moto sums container-level memory; Fargate does not need it
        ecs.run_task(cluster=cluster, taskDefinition=family, launchType="FARGATE", networkConfiguration=awsvpc(network))


def test_services_never_start_tasks(ecs, cluster, taskdef, network):
    ecs.create_service(cluster=cluster, serviceName="api", taskDefinition=taskdef, desiredCount=2, launchType="FARGATE", networkConfiguration=awsvpc(network))
    svc = ecs.describe_services(cluster=cluster, services=["api"])["services"][0]
    assert (svc["desiredCount"], svc["runningCount"], svc["pendingCount"]) == (2, 0, 2)
    assert ecs.list_tasks(cluster=cluster)["taskArns"] == []
    with pytest.raises(WaiterError, match="Max attempts"):                            # services_stable can never succeed
        ecs.get_waiter("services_stable").wait(cluster=cluster, services=["api"], WaiterConfig={"Delay": 0, "MaxAttempts": 2})


def test_task_state_is_thin(ecs, cluster, taskdef, network):
    arn = ecs.run_task(cluster=cluster, taskDefinition=taskdef, launchType="FARGATE", networkConfiguration=awsvpc(network))["tasks"][0]["taskArn"]
    for _ in range(5):
        task = ecs.describe_tasks(cluster=cluster, tasks=[arn])["tasks"][0]
    assert task["lastStatus"] == "STOPPED" and task["desiredStatus"] == "RUNNING"     # desiredStatus is never updated
    assert task["containers"][0]["exitCode"] == 0                                      # every task "succeeds"

# tests/smoke/test_ecs_smoke.py (excerpt)
pytestmark = [pytest.mark.aws_smoke, pytest.mark.skipif(os.environ.get("RUN_AWS_SMOKE") != "1", reason="set RUN_AWS_SMOKE=1 in the sandbox account")]
TAGS = {"purpose": "smoke", "owner": "qa-automation"}

@pytest.fixture
def taskdef(ecs, cfg):
    family = f"smoke-{uuid.uuid4().hex[:8]}"                      # unique: never reuse a fixed name
    td = build_task_definition(family, "public.ecr.aws/docker/library/busybox:1.37", cpu=256, memory=512,
                               execution_role_arn=cfg["execution_role"], task_role_arn=cfg["execution_role"], log_group="/ecs/smoke")
    td["containerDefinitions"][0].update(entryPoint=["sh", "-c"], memory=256)
    td["tags"] = [{"key": k, "value": v} for k, v in TAGS.items()]
    arn = ecs.register_task_definition(**td)["taskDefinition"]["taskDefinitionArn"]
    try:
        yield arn
    finally:
        ecs.deregister_task_definition(taskDefinition=arn)
        ecs.delete_task_definitions(taskDefinitions=[arn])       # deregister first; deletion needs an INACTIVE revision

def test_task_runs_and_exits_zero(ecs, cfg, taskdef):
    assert run(ecs, cfg, taskdef, "echo smoke-ok").ok

def test_nonzero_exit_code_is_reported(ecs, cfg, taskdef):
    with pytest.raises(TaskFailedError) as ei:
        run(ecs, cfg, taskdef, "exit 3")
    assert ei.value.result.exit_codes == {"app": 3}

# app/janitor.py (excerpt)
def stop_leftover_tasks(ecs, cluster: str, started_by: str = "qa-harness") -> list[str]:
    arns = [a for page in ecs.get_paginator("list_tasks").paginate(cluster=cluster, startedBy=started_by) for a in page["taskArns"]]
    for arn in arns:
        ecs.stop_task(cluster=cluster, task=arn, reason="janitor: leftover from a test run")
    return arns

def delete_task_definitions(ecs, family_prefix: str = "smoke-") -> int:
    families = [f for page in ecs.get_paginator("list_task_definition_families").paginate(familyPrefix=family_prefix, status="ACTIVE") for f in page["families"]]
    arns = [a for fam in families                                      # list_task_definitions wants the FULL family name, not a prefix
            for page in ecs.get_paginator("list_task_definitions").paginate(familyPrefix=fam, status="ACTIVE") for a in page["taskDefinitionArns"]]
    for arn in arns:
        ecs.deregister_task_definition(taskDefinition=arn)
    for i in range(0, len(arns), 10):                                  # delete_task_definitions takes up to 10 INACTIVE revisions
        ecs.delete_task_definitions(taskDefinitions=arns[i:i + 10])
    return len(arns)

# .github/workflows/ecs.yml
name: ecs
on:
  push:
    branches: [main]
  pull_request:

permissions:
  contents: read

jobs:
  tests:                                         # unit, Stubber, moto, container tests: no AWS credentials
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v4
      - run: uv sync --locked
      - run: docker pull python:3.13-slim-bookworm            # the SIGTERM tests skip themselves when this image is missing
      - run: uv run pytest -n 4 -m "not aws_smoke"          # includes the Docker tests: Docker is available on ubuntu-latest

  deploy:
    if: github.ref == 'refs/heads/main'
    needs: tests
    runs-on: ubuntu-latest
    environment: staging
    permissions:
      id-token: write                             # lets the job request an OIDC token
      contents: read
    env:
      AWS_REGION: us-east-1
      CLUSTER: staging
      HEALTH_URL: https://staging.example.com/health
      EXECUTION_ROLE_ARN: arn:aws:iam::123456789012:role/ecsTaskExecutionRole     # placeholders
      TASK_ROLE_ARN: arn:aws:iam::123456789012:role/orders-task
    steps:
      - uses: actions/checkout@v4
      - uses: aws-actions/configure-aws-credentials@v6
        with:
          role-to-assume: arn:aws:iam::123456789012:role/github-ecs-deploy
          aws-region: ${{ env.AWS_REGION }}
      - id: ecr
        uses: aws-actions/amazon-ecr-login@v2         # outputs the registry host
      - name: Build and push (immutable tag = commit SHA)
        run: |
          IMAGE="${{ steps.ecr.outputs.registry }}/orders:${{ github.sha }}"
          docker build -t "$IMAGE" .
          docker push "$IMAGE"
          echo "IMAGE=$IMAGE" >> "$GITHUB_ENV"
      - uses: astral-sh/setup-uv@v4
      - run: uv sync --locked
      - name: Register, deploy, verify, roll back on failure
        run: uv run python -m scripts.deploy
      - name: Smoke test (real Fargate task)
        run: RUN_AWS_SMOKE=1 uv run pytest tests/smoke -m aws_smoke
        env:
          SMOKE_CLUSTER: ${{ env.CLUSTER }}
          SMOKE_SUBNETS: ${{ vars.SMOKE_SUBNETS }}
          SMOKE_SECURITY_GROUPS: ${{ vars.SMOKE_SECURITY_GROUPS }}
          SMOKE_EXECUTION_ROLE_ARN: ${{ env.EXECUTION_ROLE_ARN }}

{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Principal": {"Federated": "arn:aws:iam::123456789012:oidc-provider/token.actions.githubusercontent.com"},
    "Action": "sts:AssumeRoleWithWebIdentity",
    "Condition": {
      "StringEquals": {"token.actions.githubusercontent.com:aud": "sts.amazonaws.com"},
      "StringLike": {"token.actions.githubusercontent.com:sub": "repo:your-org/your-repo:ref:refs/heads/main"}
    }
  }]
}

# app/deploy.py (the core)
def deploy_and_verify(ecs, *, cluster, service, task_definition_arn, health_check, timeout_s=600, poll_s=15) -> DeployResult:
    previous = _service(ecs, cluster, service)["taskDefinition"]
    ecs.update_service(cluster=cluster, service=service, taskDefinition=task_definition_arn, forceNewDeployment=True)

    def roll_back(reason: str):
        ecs.update_service(cluster=cluster, service=service, taskDefinition=previous, forceNewDeployment=True)
        try:
            _wait_stable(ecs, cluster, service, timeout_s, poll_s)
        except WaiterError:
            reason += " (rollback did not stabilise either)"
        raise DeployFailed(reason, previous) from None

    try:
        _wait_stable(ecs, cluster, service, timeout_s, poll_s)
    except WaiterError as exc:
        roll_back(f"service did not become stable: {exc}")

    svc = _service(ecs, cluster, service)
    if svc["taskDefinition"] != task_definition_arn:                  # circuit breaker rolled back; "stable" then means "stable on the OLD revision"
        raise DeployFailed("deployment circuit breaker rolled the service back", svc["taskDefinition"])
    if not health_check():
        roll_back("health check failed after the deployment completed")
    return DeployResult(previous=previous, current=task_definition_arn)

# app/verify.py: the endpoint must answer with the version we just deployed
def wait_for_version(url, expected_version, *, timeout_s=120, interval_s=3) -> bool:
    deadline = time.monotonic() + timeout_s
    while True:
        try:
            with urllib.request.urlopen(url, timeout=5) as resp:
                if resp.status == 200 and json.load(resp).get("version") == expected_version:
                    return True
        except (urllib.error.URLError, TimeoutError, ValueError):
            pass                                                       # not up yet, or a 5xx during the switch-over
        if time.monotonic() >= deadline:
            return False
        time.sleep(interval_s)