2.2Unit Testing with Pulumi Mocks

Pulumi's mock framework lets you test infrastructure logic without deploying anything. The mocks intercept all cloud API calls and return predictable responses.

TypeScript Unit Tests

import * as pulumi from "@pulumi/pulumi";
import * as aws from "@pulumi/aws";
import { describe, it, expect, beforeAll } from "vitest";

// Mock all Pulumi resource creation
pulumi.runtime.setMocks({
    newResource: (args: pulumi.runtime.MockResourceArgs) => {
        // Return mock values for each resource type
        switch (args.type) {
            case "aws:s3/bucket:Bucket":
                return {
                    id: `${args.name}-id`,
                    state: {
                        ...args.inputs,
                        arn: `arn:aws:s3:::${args.inputs.bucket || args.name}`,
                        bucket: args.inputs.bucket || args.name,
                    },
                };
            case "aws:rds/instance:Instance":
                return {
                    id: `${args.name}-id`,
                    state: {
                        ...args.inputs,
                        endpoint: "mock-db.cluster-abc123.us-east-1.rds.amazonaws.com",
                    },
                };
            default:
                return {
                    id: `${args.name}-id`,
                    state: args.inputs,
                };
        }
    },
    call: (args: pulumi.runtime.MockCallArgs) => {
        return args.inputs;
    },
});

describe("S3 Data Bucket", () => {
    it("must have encryption enabled", async () => {
        const bucket = new aws.s3.Bucket("test-bucket", {
            serverSideEncryptionConfiguration: {
                rule: {
                    applyServerSideEncryptionByDefault: {
                        sseAlgorithm: "aws:kms",
                    },
                },
            },
        });

        const encryption = await new Promise((resolve) =>
            bucket.serverSideEncryptionConfiguration.apply((config) =>
                resolve(config?.rule?.applyServerSideEncryptionByDefault?.sseAlgorithm)
            )
        );
        expect(encryption).toBe("aws:kms");
    });

    it("must block public access", async () => {
        const bucket = new aws.s3.Bucket("test-bucket", {});
        const publicAccessBlock = new aws.s3.BucketPublicAccessBlock("test-pab", {
            bucket: bucket.id,
            blockPublicAcls: true,
            blockPublicPolicy: true,
            ignorePublicAcls: true,
            restrictPublicBuckets: true,
        });

        const blockAcls = await new Promise((resolve) =>
            publicAccessBlock.blockPublicAcls.apply(resolve)
        );
        expect(blockAcls).toBe(true);
    });

    it("must have versioning enabled", async () => {
        const bucket = new aws.s3.Bucket("test-bucket", {
            versioning: { enabled: true },
        });

        const versioning = await new Promise((resolve) =>
            bucket.versioning.apply((v) => resolve(v?.enabled))
        );
        expect(versioning).toBe(true);
    });
});

Python Unit Tests

# tests/test_infrastructure.py
import pulumi
import pytest

class MockResource:
    """Mock Pulumi resource provider for unit testing."""

    def __init__(self):
        self.resources = {}

    def __call__(self, args):
        self.resources[args.name] = args.inputs
        return pulumi.runtime.MockResourceArgs(
            id_=f"{args.name}-id",
            state=args.inputs,
        )

@pytest.fixture
def mock_pulumi():
    mock = MockResource()
    pulumi.runtime.set_mocks(
        pulumi.runtime.Mocks(
            new_resource=mock,
            call=lambda args: {},
        )
    )
    return mock

def test_database_not_publicly_accessible(mock_pulumi):
    """RDS instances must never be publicly accessible."""
    import pulumi_aws as aws

    db = aws.rds.Instance(
        "test-db",
        instance_class="db.t4g.micro",
        allocated_storage=20,
        engine="postgres",
        publicly_accessible=False,
    )

    # Assert the publicly_accessible property
    def check(value):
        assert value is False, "Database must not be publicly accessible"

    db.publicly_accessible.apply(check)

def test_rds_has_encryption(mock_pulumi):
    """RDS instances must have storage encryption enabled."""
    import pulumi_aws as aws

    db = aws.rds.Instance(
        "test-db",
        instance_class="db.t4g.micro",
        allocated_storage=20,
        engine="postgres",
        storage_encrypted=True,
    )

    def check(value):
        assert value is True, "Database storage must be encrypted"

    db.storage_encrypted.apply(check)