Pulumi Automation API: Testing Infrastructure as Real Code

# Python
pip install pulumi pulumi-aws

# TypeScript
npm install @pulumi/pulumi @pulumi/aws

# Go
go get github.com/pulumi/pulumi/sdk/v3/go/auto

# test_s3_bucket.py
import pytest
import pulumi
import pulumi_aws as aws
from pulumi import automation as auto
import boto3
import os

def create_s3_program(bucket_name: str):
    """Pulumi program that creates an S3 bucket."""
    def program():
        bucket = aws.s3.BucketV2(
            "test-bucket",
            bucket=bucket_name,
            tags={"Environment": "test", "ManagedBy": "pulumi-test"},
        )
        
        versioning = aws.s3.BucketVersioningV2(
            "test-bucket-versioning",
            bucket=bucket.id,
            versioning_configuration=aws.s3.BucketVersioningV2VersioningConfigurationArgs(
                status="Enabled",
            ),
        )
        
        pulumi.export("bucket_name", bucket.id)
        pulumi.export("bucket_arn", bucket.arn)
    
    return program

@pytest.fixture(scope="module")
def deployed_stack():
    """Create a stack, yield it for tests, then destroy it."""
    stack_name = f"test-s3-{os.urandom(4).hex()}"
    bucket_name = f"test-bucket-{os.urandom(4).hex()}"
    
    stack = auto.create_stack(
        stack_name=stack_name,
        project_name="infra-tests",
        program=create_s3_program(bucket_name),
    )
    
    # Set AWS config
    stack.set_config("aws:region", auto.ConfigValue("us-east-1"))
    
    # Deploy
    up_result = stack.up(on_output=print)
    
    yield {
        "stack": stack,
        "outputs": up_result.outputs,
        "bucket_name": bucket_name,
    }
    
    # Cleanup — always runs even if test fails
    stack.destroy(on_output=print)
    stack.workspace.remove_stack(stack_name)

def test_bucket_created(deployed_stack):
    """Bucket should be created with the right name."""
    bucket_name = deployed_stack["outputs"]["bucket_name"].value
    assert bucket_name == deployed_stack["bucket_name"]

def test_bucket_exists_in_aws(deployed_stack):
    """Verify bucket actually exists in AWS (not just in Pulumi state)."""
    bucket_name = deployed_stack["bucket_name"]
    
    s3 = boto3.client("s3", region_name="us-east-1")
    response = s3.head_bucket(Bucket=bucket_name)
    
    assert response["ResponseMetadata"]["HTTPStatusCode"] == 200

def test_bucket_versioning_enabled(deployed_stack):
    """Bucket versioning should be enabled."""
    bucket_name = deployed_stack["bucket_name"]
    
    s3 = boto3.client("s3", region_name="us-east-1")
    versioning = s3.get_bucket_versioning(Bucket=bucket_name)
    
    assert versioning.get("Status") == "Enabled"

def test_bucket_tags(deployed_stack):
    """Bucket should have required tags."""
    bucket_name = deployed_stack["bucket_name"]
    
    s3 = boto3.client("s3", region_name="us-east-1")
    tagging = s3.get_bucket_tagging(Bucket=bucket_name)
    
    tags = {t["Key"]: t["Value"] for t in tagging["TagSet"]}
    assert tags.get("Environment") == "test"
    assert tags.get("ManagedBy") == "pulumi"

# test_s3_unit.py
import pytest
import pulumi

class MyMocks(pulumi.runtime.Mocks):
    def new_resource(self, args: pulumi.runtime.MockResourceArgs):
        outputs = args.inputs
        if args.typ == "aws:s3/bucketV2:BucketV2":
            outputs = {**args.inputs, "id": args.inputs["bucket"], "arn": f"arn:aws:s3:::{args.inputs['bucket']}"}
        return [args.name + "_id", outputs]
    
    def call(self, args: pulumi.runtime.MockCallArgs):
        return {}

pulumi.runtime.set_mocks(MyMocks())

# Import the Pulumi program AFTER setting mocks
from my_infra.s3_module import create_bucket

@pulumi.runtime.test
def test_bucket_has_versioning():
    def check(args):
        bucket_name, versioning_status = args
        assert bucket_name is not None
        assert versioning_status == "Enabled"
    
    bucket, versioning = create_bucket("test-bucket")
    return pulumi.Output.all(bucket.id, versioning.versioning_configuration.status).apply(check)

@pulumi.runtime.test
def test_bucket_has_required_tags():
    def check(tags):
        assert tags.get("Environment") is not None
        assert tags.get("ManagedBy") == "pulumi"
    
    bucket, _ = create_bucket("test-bucket")
    return bucket.tags.apply(check)

import concurrent.futures

def test_module_parametrized_parallel():
    """Test the same module with different configs in parallel."""
    test_cases = [
        {"environment": "dev", "instance_type": "t3.micro"},
        {"environment": "staging", "instance_type": "t3.small"},
    ]
    
    def run_test(case):
        stack_name = f"test-{case['environment']}-{os.urandom(4).hex()}"
        
        stack = auto.create_stack(
            stack_name=stack_name,
            project_name="infra-tests",
            program=lambda: create_ec2_program(**case),
        )
        stack.set_config("aws:region", auto.ConfigValue("us-east-1"))
        
        try:
            result = stack.up()
            return {"case": case, "success": True, "outputs": result.outputs}
        finally:
            stack.destroy()
            stack.workspace.remove_stack(stack_name)
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        futures = [executor.submit(run_test, case) for case in test_cases]
        results = [f.result() for f in concurrent.futures.as_completed(futures)]
    
    assert all(r["success"] for r in results)

def test_preview_produces_expected_changes():
    """Validate what will be created without deploying."""
    stack = auto.create_or_select_stack(
        stack_name="preview-test",
        project_name="infra-tests",
        program=create_s3_program("preview-test-bucket"),
    )
    stack.set_config("aws:region", auto.ConfigValue("us-east-1"))
    
    preview_result = stack.preview()
    
    # Verify expected resources will be created
    changes = preview_result.change_summary
    assert changes.get(auto.OpType.CREATE, 0) >= 2  # bucket + versioning
    assert changes.get(auto.OpType.DELETE, 0) == 0   # no deletions

// tests/s3.test.ts
import * as automation from "@pulumi/pulumi/automation";
import * as aws from "@pulumi/aws";
import * as pulumi from "@pulumi/pulumi";
import { S3Client, HeadBucketCommand } from "@aws-sdk/client-s3";

const bucketName = `test-bucket-${Math.random().toString(36).substr(2, 8)}`;

const program = async () => {
  const bucket = new aws.s3.BucketV2("test-bucket", {
    bucket: bucketName,
    tags: { Environment: "test" },
  });
  
  return { bucketName: bucket.id };
};

describe("S3 Bucket Module", () => {
  let stack: automation.Stack;
  let outputs: Record<string, automation.OutputValue>;

  beforeAll(async () => {
    stack = await automation.LocalWorkspace.createStack({
      stackName: `test-${Math.random().toString(36).substr(2, 8)}`,
      projectName: "infra-tests",
      program,
    });
    
    await stack.setConfig("aws:region", { value: "us-east-1" });
    const result = await stack.up({ onOutput: console.log });
    outputs = result.outputs;
  }, 120_000);

  afterAll(async () => {
    await stack.destroy({ onOutput: console.log });
    await stack.workspace.removeStack(stack.name);
  }, 120_000);

  test("bucket is created with correct name", () => {
    expect(outputs.bucketName.value).toBe(bucketName);
  });

  test("bucket exists in AWS", async () => {
    const s3 = new S3Client({ region: "us-east-1" });
    const response = await s3.send(new HeadBucketCommand({ Bucket: bucketName }));
    expect(response.$metadata.httpStatusCode).toBe(200);
  });
});

# .github/workflows/pulumi-tests.yml
name: Pulumi Infrastructure Tests

on:
  push:
    paths: ["infra/**"]
  pull_request:

jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install pulumi pulumi-aws pytest
      - name: Unit tests (mocked)
        run: pytest tests/unit/ -v

  integration-tests:
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    environment: aws-test
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install pulumi pulumi-aws pytest boto3
      - name: Integration tests (real AWS)
        run: pytest tests/integration/ -v --timeout=300
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          PULUMI_ACCESS_TOKEN: ${{ secrets.PULUMI_ACCESS_TOKEN }}