Testing

Testing Pulumi Infrastructure Code with Go and Python

HelpMeTest

HelpMeTest

— 7 min read
Testing Pulumi Infrastructure Code with Go and Python

Pulumi infrastructure code is real programming-language code, so it deserves real tests. Pulumi's SDK provides a mock framework for unit tests (no cloud required) and an Automation API for integration tests against ephemeral stacks. This guide covers both layers in Go and Python, with CI patterns for each.

Key Takeaways

Pulumi's mocks replace cloud calls entirely. Unit tests run in milliseconds with no cloud credentials—the mock framework intercepts resource registrations and lets you assert on inputs before anything is deployed.

Test resource properties, not existence. Checking that an S3 bucket was created is weak. Check that it has versioning enabled, ACL set to private, and the correct tags.

Use ephemeral stacks for integration tests. The Automation API lets you create, deploy, and destroy a complete stack programmatically—ideal for CI without manual state management.

Mock outputs control downstream resources. When resource A feeds resource B, set mock outputs on A to control what B receives—this lets you test the full dependency chain without a real cloud.

Keep stacks short-lived. Integration test stacks should be created fresh per test run and destroyed on completion (even on failure). Never reuse a test stack across runs.

Why Pulumi Testing Is Different

Terraform users write HCL—a declarative config language with limited expressibility. Pulumi users write real programs in Go, Python, TypeScript, or C#. This is a superpower for complex infrastructure, but it also means your infrastructure code can have all the same bugs as application code: wrong conditionals, off-by-one errors, incorrect string formatting, broken loops.

Pulumi's testing framework reflects this. You test Pulumi programs the same way you test any program: unit tests with mocks for fast iteration, integration tests against real infrastructure for confidence.

Unit Testing in Go

Pulumi's Go SDK includes pulumi/pulumix and the integration package. For unit tests, you use pulumi.RunErr with mocks:

// infra/stack_test.go
package main

import (
    "sync"
    "testing"

    "github.com/pulumi/pulumi-aws/sdk/v6/go/aws/s3"
    "github.com/pulumi/pulumi/sdk/v3/go/common/resource"
    "github.com/pulumi/pulumi/sdk/v3/go/pulumi"
    "github.com/stretchr/testify/assert"
)

type mocks struct{}

// NewResource is called every time a resource is registered
func (m *mocks) NewResource(args pulumi.MockResourceArgs) (string, resource.PropertyMap, error) {
    outputs := args.Inputs.Copy()

    switch args.TypeToken {
    case "aws:s3/bucket:Bucket":
        outputs["id"] = resource.NewStringProperty("my-test-bucket")
        outputs["arn"] = resource.NewStringProperty("arn:aws:s3:::my-test-bucket")
        outputs["bucketDomainName"] = resource.NewStringProperty("my-test-bucket.s3.amazonaws.com")
    case "aws:s3/bucketVersioningV2:BucketVersioningV2":
        outputs["id"] = resource.NewStringProperty("my-test-bucket")
    }

    return args.Name + "_id", outputs, nil
}

// Call is invoked for provider function calls
func (m *mocks) Call(args pulumi.MockCallArgs) (resource.PropertyMap, error) {
    return resource.PropertyMap{}, nil
}

func TestS3BucketConfiguration(t *testing.T) {
    err := pulumi.RunErr(func(ctx *pulumi.Context) error {
        infra, err := createInfrastructure(ctx)
        if err != nil {
            return err
        }

        var wg sync.WaitGroup
        wg.Add(3)

        // Assert bucket has versioning enabled
        pulumi.All(infra.bucket.Bucket, infra.versioning.VersioningConfiguration).ApplyT(
            func(all []interface{}) error {
                defer wg.Done()
                bucketName := all[0].(string)
                config := all[1].(s3.BucketVersioningV2VersioningConfiguration)

                assert.Equal(t, "Enabled", config.Status, "versioning should be enabled")
                assert.NotEmpty(t, bucketName)
                return nil
            },
        )

        // Assert bucket has correct tags
        infra.bucket.Tags.ApplyT(func(tags map[string]string) error {
            defer wg.Done()
            assert.Equal(t, "production", tags["Environment"])
            assert.Equal(t, "platform", tags["Team"])
            return nil
        })

        // Assert server-side encryption is configured
        infra.bucket.ServerSideEncryptionConfiguration.ApplyT(
            func(enc *s3.BucketServerSideEncryptionConfiguration) error {
                defer wg.Done()
                assert.NotNil(t, enc, "server-side encryption must be configured")
                return nil
            },
        )

        wg.Wait()
        return nil
    }, pulumi.WithMocks("project", "stack", &mocks{}))

    assert.NoError(t, err)
}

Your infrastructure program is just a normal Go function:

// infra/main.go
type Infrastructure struct {
    bucket     *s3.Bucket
    versioning *s3.BucketVersioningV2
}

func createInfrastructure(ctx *pulumi.Context) (*Infrastructure, error) {
    bucket, err := s3.NewBucket(ctx, "app-bucket", &s3.BucketArgs{
        Tags: pulumi.StringMap{
            "Environment": pulumi.String("production"),
            "Team":        pulumi.String("platform"),
        },
        ServerSideEncryptionConfiguration: &s3.BucketServerSideEncryptionConfigurationArgs{
            Rule: &s3.BucketServerSideEncryptionConfigurationRuleArgs{
                ApplyServerSideEncryptionByDefault: &s3.BucketServerSideEncryptionConfigurationRuleApplyServerSideEncryptionByDefaultArgs{
                    SseAlgorithm: pulumi.String("aws:kms"),
                },
            },
        },
    })
    if err != nil {
        return nil, err
    }

    versioning, err := s3.NewBucketVersioningV2(ctx, "app-bucket-versioning", &s3.BucketVersioningV2Args{
        Bucket: bucket.Bucket,
        VersioningConfiguration: &s3.BucketVersioningV2VersioningConfigurationArgs{
            Status: pulumi.String("Enabled"),
        },
    })
    if err != nil {
        return nil, err
    }

    return &Infrastructure{bucket: bucket, versioning: versioning}, nil
}

Unit Testing in Python

Python uses pytest. Pulumi provides pulumi.runtime.set_mocks:

# test_stack.py
import pytest
import pulumi

class MyMocks(pulumi.runtime.Mocks):
    def new_resource(self, args: pulumi.runtime.MockResourceArgs):
        outputs = args.inputs.copy()

        if args.typ == "aws:ec2/vpc:Vpc":
            outputs["id"] = "vpc-12345678"
            outputs["arn"] = "arn:aws:ec2:us-east-1:123456789:vpc/vpc-12345678"
            outputs["defaultRouteTableId"] = "rtb-12345678"
        elif args.typ == "aws:ec2/subnet:Subnet":
            outputs["id"] = f"subnet-{args.name}"
            outputs["arn"] = f"arn:aws:ec2:us-east-1:123456789:subnet/subnet-{args.name}"

        return [args.name + "_id", outputs]

    def call(self, args: pulumi.runtime.MockCallArgs):
        return {}

pulumi.runtime.set_mocks(
    MyMocks(),
    project="myproject",
    stack="testing",
    preview=False,
)

# Import AFTER set_mocks — this is critical
import infra

@pulumi.runtime.test
def test_vpc_has_correct_cidr():
    def check_cidr(args):
        cidr_block, = args
        assert cidr_block == "10.0.0.0/16", f"Expected 10.0.0.0/16, got {cidr_block}"

    return pulumi.Output.all(infra.vpc.cidr_block).apply(check_cidr)

@pulumi.runtime.test
def test_vpc_has_dns_enabled():
    def check_dns(args):
        dns_hostnames, dns_support = args
        assert dns_hostnames is True, "DNS hostnames must be enabled"
        assert dns_support is True, "DNS support must be enabled"

    return pulumi.Output.all(
        infra.vpc.enable_dns_hostnames,
        infra.vpc.enable_dns_support
    ).apply(check_dns)

@pulumi.runtime.test
def test_subnets_are_tagged():
    def check_tags(args):
        tags, = args
        assert "Environment" in tags, "subnet must have Environment tag"
        assert tags["Environment"] == "staging"

    return pulumi.Output.all(infra.private_subnet.tags).apply(check_tags)

@pulumi.runtime.test
def test_nat_gateway_has_eip():
    def check_eip(args):
        allocation_id, = args
        assert allocation_id is not None, "NAT gateway must have an EIP"

    return pulumi.Output.all(infra.nat_gateway.allocation_id).apply(check_eip)

Run with:

python -m pytest test_stack.py -v

Mocking Output Values for Dependency Testing

When resource B depends on output from resource A, your mocks need to return the right output so B receives correct inputs:

class NetworkMocks(pulumi.runtime.Mocks):
    def new_resource(self, args):
        outputs = args.inputs.copy()

        if args.typ == "aws:ec2/vpc:Vpc":
            outputs["id"] = "vpc-test-12345"

        elif args.typ == "aws:ec2/subnet:Subnet":
            # Verify the subnet is using the correct VPC
            assert args.inputs.get("vpcId") == "vpc-test-12345", \
                f"Subnet must reference correct VPC, got: {args.inputs.get('vpcId')}"
            outputs["id"] = f"subnet-test-{args.name}"

        elif args.typ == "aws:ec2/routeTableAssociation:RouteTableAssociation":
            assert "subnet-test-" in args.inputs.get("subnetId", ""), \
                "Route table association must reference a test subnet"

        return [args.name + "_id", outputs]

This pattern lets you test the full dependency chain without a real cloud.

Integration Tests with the Automation API

The Automation API lets you manage Pulumi stacks programmatically—create, deploy, assert, destroy:

// test/integration_test.go
package test

import (
    "context"
    "fmt"
    "os"
    "testing"

    "github.com/pulumi/pulumi/sdk/v3/go/auto"
    "github.com/pulumi/pulumi/sdk/v3/go/auto/optdestroy"
    "github.com/pulumi/pulumi/sdk/v3/go/auto/optup"
    "github.com/stretchr/testify/assert"
    "github.com/stretchr/testify/require"
)

func TestVpcIntegration(t *testing.T) {
    if testing.Short() {
        t.Skip("skipping integration test in short mode")
    }

    ctx := context.Background()
    runID := os.Getenv("GITHUB_RUN_ID")
    if runID == "" {
        runID = "local"
    }
    stackName := fmt.Sprintf("test-%s", runID)

    s, err := auto.NewStackLocalSource(ctx, stackName, "../infra")
    require.NoError(t, err)

    s.SetConfig(ctx, "aws:region", auto.ConfigValue{Value: "us-east-1"})
    s.SetConfig(ctx, "environment", auto.ConfigValue{Value: "test"})

    defer func() {
        s.Destroy(ctx, optdestroy.ProgressStreams(os.Stdout))
        s.Workspace().RemoveStack(ctx, stackName)
    }()

    res, err := s.Up(ctx, optup.ProgressStreams(os.Stdout))
    require.NoError(t, err, "stack up failed")
    assert.Equal(t, "succeeded", res.Summary.Result)

    vpcID, ok := res.Outputs["vpcId"]
    require.True(t, ok, "vpcId output must exist")
    assert.Regexp(t, `^vpc-[a-f0-9]+$`, vpcID.Value.(string))

    subnetIDs, ok := res.Outputs["privateSubnetIds"]
    require.True(t, ok, "privateSubnetIds output must exist")
    subnets := subnetIDs.Value.([]interface{})
    assert.Equal(t, 3, len(subnets), "should have 3 private subnets")
}

Python Integration Tests with pytest

# test/test_integration.py
import os
import pytest
import pulumi.automation as auto

@pytest.fixture(scope="module")
def stack():
    run_id = os.environ.get("GITHUB_RUN_ID", "local")
    stack_name = f"test-{run_id}"

    s = auto.create_or_select_stack(
        stack_name=stack_name,
        work_dir="../infra",
    )

    s.set_config("aws:region", auto.ConfigValue(value="us-east-1"))
    s.set_config("environment", auto.ConfigValue(value="test"))
    s.workspace.install_plugin("aws", "v6.0.0")

    up_result = s.up(on_output=print)
    assert up_result.summary.result == "succeeded"

    yield up_result

    s.destroy(on_output=print)
    s.workspace.remove_stack(stack_name)

def test_vpc_output_exists(stack):
    assert "vpcId" in stack.outputs
    assert stack.outputs["vpcId"].value.startswith("vpc-")

def test_correct_subnet_count(stack):
    subnet_ids = stack.outputs["privateSubnetIds"].value
    assert len(subnet_ids) == 3, f"Expected 3 subnets, got {len(subnet_ids)}"

def test_load_balancer_dns_accessible(stack):
    import urllib.request
    lb_dns = stack.outputs["loadBalancerDns"].value
    response = urllib.request.urlopen(f"http://{lb_dns}/health", timeout=10)
    assert response.status == 200

CI Pipeline

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

on:
  pull_request:
    paths: ['infra/**']
  push:
    branches: [main]

jobs:
  unit-tests-go:
    name: Go Unit Tests
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-go@v5
        with:
          go-version: "1.21"
      - name: Run unit tests
        working-directory: infra
        run: go test -v -short ./...

  unit-tests-python:
    name: Python Unit Tests
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install pulumi pulumi-aws pytest
      - name: Run unit tests
        working-directory: infra
        run: python -m pytest tests/unit/ -v

  integration-tests:
    name: Integration Tests
    runs-on: ubuntu-latest
    needs: [unit-tests-go, unit-tests-python]
    if: github.ref == 'refs/heads/main'
    environment: aws-test
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-go@v5
        with:
          go-version: "1.21"
      - name: Run integration tests
        working-directory: test
        run: go test -v -run Integration -timeout 30m ./...
        env:
          PULUMI_ACCESS_TOKEN: ${{ secrets.PULUMI_ACCESS_TOKEN }}
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          AWS_DEFAULT_REGION: us-east-1
          GITHUB_RUN_ID: ${{ github.run_id }}

Compliance Testing Across All Resources

One powerful Pulumi testing pattern is asserting compliance across every resource of a given type. Define a helper that collects all resources and checks required tags:

@pulumi.runtime.test
def test_all_ec2_instances_use_approved_amis():
    approved_amis = {
        "ami-0abcdef1234567890",  # Amazon Linux 2023
        "ami-0fedcba9876543210",  # Ubuntu 22.04 hardened
    }

    def check_ami(args):
        ami_id, instance_name = args
        assert ami_id in approved_amis, \
            f"Instance '{instance_name}' uses unapproved AMI: {ami_id}. " \
            f"Approved: {approved_amis}"

    checks = []
    for instance in infra.all_ec2_instances:
        checks.append(
            pulumi.Output.all(instance.ami, instance._name).apply(check_ami)
        )

    return pulumi.Output.all(*checks)

This approach scales to any compliance requirement: approved regions, required tags, encryption settings, or security group rules.

Conclusion

Pulumi's testing story is stronger than most IaC tools precisely because infrastructure code is real code. Unit tests with mocks give you millisecond feedback on logic errors, property constraints, and dependency wiring. Integration tests with ephemeral stacks give you confidence that the actual cloud resources are provisioned correctly. Use both, run unit tests on every commit, and gate merges on integration tests passing. Your infrastructure deserves the same quality bar as your application code.

Read more

Start now free