Amazon EC2 — moto Gaps, Smoke Tests & CI¶

def test_client_token_is_not_honoured(ec2):
    a = run(ec2, ClientToken="same")["Instances"][0]["InstanceId"]
    b = run(ec2, ClientToken="same")["Instances"][0]["InstanceId"]
    assert a != b                                                                      # real: the retry returns the first instance
    run(ec2, ClientToken="same", InstanceType="m5.large")                              # real: IdempotentParameterMismatch


def test_dry_run_always_reports_permission_granted(ec2, net):
    p = build_run_params(ami_id=AMI, instance_type="t3.micro", subnet_id=net["subnet"], security_group_ids=[net["sg"]],
                         tags={"owner": "qa", "env": "test"}, run_id="t", name="n")
    assert can_launch(ec2, **p) is True                                                # IAM is not evaluated: UnauthorizedOperation never happens


# --- time and state ------------------------------------------------------------------------------------------
def test_instances_are_running_and_healthy_immediately(ec2):
    r = run(ec2)["Instances"][0]
    assert r["State"]["Name"] == "pending"                                             # only the run_instances response says pending
    iid = r["InstanceId"]
    assert ec2.describe_instances(InstanceIds=[iid])["Reservations"][0]["Instances"][0]["State"]["Name"] == "running"
    status = ec2.describe_instance_status(InstanceIds=[iid])["InstanceStatuses"][0]
    assert status["InstanceStatus"]["Status"] == status["SystemStatus"]["Status"] == "ok"      # real: "initializing" first
    ec2.get_waiter("instance_status_ok").wait(InstanceIds=[iid], WaiterConfig={"Delay": 1, "MaxAttempts": 1})   # first poll passes


def test_a_terminated_instance_can_be_started(ec2):
    iid = run(ec2)["Instances"][0]["InstanceId"]
    ec2.terminate_instances(InstanceIds=[iid])
    out = ec2.start_instances(InstanceIds=[iid])["StartingInstances"][0]["CurrentState"]["Name"]
    assert out == "pending"                                                            # real: IncorrectInstanceState

"""Real-account smoke test. Skipped unless RUN_AWS_SMOKE=1.
IMPORTANT: the autouse `aws_env` fixture in tests/conftest.py sets FAKE credentials. This module overrides it."""
import os
import uuid

import boto3
import pytest
from botocore.config import Config

from app.compliance import imdsv1_enabled, iter_instances, missing_tags
from app.launch import build_run_params, can_launch, latest_al2023_ami, launch_and_wait, terminate_run
from app.userdata import TEMPLATE, render_user_data

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"),
]
REAL = Config(retries={"mode": "standard", "max_attempts": 5}, connect_timeout=5, read_timeout=30)


@pytest.fixture(autouse=True)
def aws_env(monkeypatch):                              # same name as the root fixture: pytest uses the closest one
    """Do NOT set fake credentials here. Take the real ones (OIDC role in CI) but pin region and refuse a wrong account."""
    monkeypatch.setenv("AWS_DEFAULT_REGION", os.environ.get("SMOKE_REGION", "eu-west-1"))


@pytest.fixture(scope="module")
def clients():
    sts = boto3.client("sts", config=REAL)
    account = sts.get_caller_identity()["Account"]
    assert account == os.environ["SMOKE_ACCOUNT_ID"], f"refusing to run in account {account}"
    return {"ec2": boto3.client("ec2", config=REAL), "ssm": boto3.client("ssm", config=REAL)}


@pytest.fixture
def smoke_run(clients):
    rid = f"smoke-{uuid.uuid4().hex[:8]}"
    try:
        yield rid
    finally:
        terminate_run(clients["ec2"], rid)             # runs even when the test body fails


def test_launch_ssm_agent_boot_and_iam(clients, smoke_run):
    ec2, ssm = clients["ec2"], clients["ssm"]
    params = build_run_params(
        ami_id=latest_al2023_ami(ssm, "arm64"), instance_type="t4g.nano",           # a tiny Graviton type: very low hourly cost
        subnet_id=os.environ["SMOKE_SUBNET_ID"], security_group_ids=[os.environ["SMOKE_SG_ID"]],
        tags={"owner": "qa-smoke", "env": "test"}, run_id=smoke_run, name=smoke_run,
        user_data=render_user_data(TEMPLATE, role="smoke", packages="jq", service="amazon-ssm-agent"),
        profile_name=os.environ["SMOKE_PROFILE"],
    )
    assert can_launch(ec2, **params), "the smoke role may not launch instances: fix IAM before debugging anything else"
    iid = launch_and_wait(ec2, params, timeout_s=300)
    ec2.get_waiter("instance_status_ok").wait(InstanceIds=[iid], WaiterConfig={"Delay": 15, "MaxAttempts": 40})
    inst = next(iter_instances(ec2.describe_instances(InstanceIds=[iid])))
    assert missing_tags(inst) == [] and imdsv1_enabled([inst]) == []           # IMDSv2 really applied by the real service

"""Scheduled cleanup: terminate test instances that outlived their welcome. Safe by default (dry run)."""
import datetime as dt

SCOPE_TAG = ("env", "test")                    # only ever touch instances that carry env=test


def expired(instances, now, max_age=dt.timedelta(hours=8), keep_tag="keep"):
    """Instance IDs older than max_age, unless tagged keep=true. Pure: feed it describe_instances output."""
    out = []
    for inst in instances:
        tags = {t["Key"]: t["Value"] for t in inst.get("Tags", [])}
        if tags.get(keep_tag) != "true" and now - inst["LaunchTime"] > max_age:
            out.append(inst["InstanceId"])
    return out


def janitor(ec2, now, max_age=dt.timedelta(hours=8), dry_run=True):
    """Find (and, with dry_run=False, terminate) expired instances in scope. Returns the IDs."""
    pages = ec2.get_paginator("describe_instances").paginate(Filters=[
        {"Name": f"tag:{SCOPE_TAG[0]}", "Values": [SCOPE_TAG[1]]},
        {"Name": "instance-state-name", "Values": ["pending", "running", "stopping", "stopped"]},
    ])
    candidates = [i for p in pages for r in p["Reservations"] for i in r["Instances"]]
    ids = expired(candidates, now, max_age)
    if ids and not dry_run:
        ec2.terminate_instances(InstanceIds=ids)
    return ids


if __name__ == "__main__":                      # python -m app.janitor [--delete]
    import sys

    import boto3

    delete = "--delete" in sys.argv
    found = janitor(boto3.client("ec2"), dt.datetime.now(dt.timezone.utc), dry_run=not delete)
    print(("terminated" if delete else "would terminate"), found)

import datetime as dt

from app.janitor import expired, janitor
from tests.conftest import AMI

HOURS = dt.timedelta(hours=1)


def launch(ec2, net, name, tags):
    resp = ec2.run_instances(ImageId=AMI, MinCount=1, MaxCount=1, SubnetId=net["subnet"], TagSpecifications=[
        {"ResourceType": "instance", "Tags": [{"Key": "Name", "Value": name}] + [{"Key": k, "Value": v} for k, v in tags.items()]}])
    return resp["Instances"][0]["InstanceId"]


def test_expired_is_a_pure_function_of_age_and_keep_tag():
    now = dt.datetime(2026, 10, 21, 12, tzinfo=dt.timezone.utc)
    sample = [
        {"InstanceId": "i-old", "LaunchTime": now - 20 * HOURS},
        {"InstanceId": "i-new", "LaunchTime": now - 1 * HOURS},
        {"InstanceId": "i-keep", "LaunchTime": now - 99 * HOURS, "Tags": [{"Key": "keep", "Value": "true"}]},
    ]
    assert expired(sample, now) == ["i-old"]


def test_janitor_is_scoped_dry_by_default_and_terminates_on_request(ec2, net):
    mine = launch(ec2, net, "mine", {"env": "test"})
    protected = launch(ec2, net, "protected", {"env": "test", "keep": "true"})
    prod = launch(ec2, net, "prod", {"env": "prod"})                      # out of scope: never touched
    later = dt.datetime.now(dt.timezone.utc) + 10 * HOURS                 # pretend ten hours have passed
    assert janitor(ec2, later) == [mine]                                  # dry run: reports only
    state = lambda i: ec2.describe_instances(InstanceIds=[i])["Reservations"][0]["Instances"][0]["State"]["Name"]   # noqa: E731
    assert state(mine) == "running"
    assert janitor(ec2, later, dry_run=False) == [mine]
    assert (state(mine), state(protected), state(prod)) == ("terminated", "running", "running")
    assert janitor(ec2, later, dry_run=False) == []                       # idempotent

{
  "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:acme/app:ref:refs/heads/main"}
    }
  }]
}

# .github/workflows/aws-ec2-smoke.yml  (needs an IAM role that trusts GitHub's OIDC provider)
name: aws-ec2-smoke
on:
  schedule: [{cron: "0 5 * * 1"}]               # weekly; never on every push
  workflow_dispatch: {}
permissions:
  id-token: write                               # lets the job request an OIDC token
  contents: read
concurrency:
  group: aws-ec2-smoke                          # one smoke run at a time: no quota races, no double bills
  cancel-in-progress: false

jobs:
  smoke:
    runs-on: ubuntu-latest
    timeout-minutes: 20
    env:
      RUN_AWS_SMOKE: "1"
      SMOKE_REGION: eu-west-1
      SMOKE_ACCOUNT_ID: ${{ vars.SMOKE_ACCOUNT_ID }}       # the test refuses to run in any other account
      SMOKE_SUBNET_ID: ${{ vars.SMOKE_SUBNET_ID }}
      SMOKE_SG_ID: ${{ vars.SMOKE_SG_ID }}                 # a group with no inbound rules: Session Manager only
      SMOKE_PROFILE: smoke-instance-profile
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v4
      - uses: aws-actions/configure-aws-credentials@v6
        with:
          role-to-assume: arn:aws:iam::123456789012:role/ci-ec2-smoke     # placeholder account ID
          aws-region: eu-west-1
      - run: uv sync --locked
      - run: uv run pytest -m aws_smoke -q
      - name: Sweep leftovers even if the test crashed
        if: always()
        run: uv run python -m app.janitor --delete

# .github/workflows/aws-ec2-janitor.yml
name: aws-ec2-janitor
on:
  schedule: [{cron: "17 * * * *"}]              # hourly
permissions:
  id-token: write
  contents: read
jobs:
  janitor:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v4
      - uses: aws-actions/configure-aws-credentials@v6
        with:
          role-to-assume: arn:aws:iam::123456789012:role/ci-ec2-janitor   # may only terminate instances tagged env=test
          aws-region: eu-west-1
      - run: uv sync --locked
      - run: uv run python -m app.janitor --delete