Pulumi Testing: Unit Tests, Mocks, and Integration Tests in Python and TypeScript

# infra/networking.py
import pulumi
import pulumi_aws as aws
from typing import Sequence

class NetworkingStack:
    def __init__(self, name: str, vpc_cidr: str, az_count: int = 2):
        self.name = name
        
        self.vpc = aws.ec2.Vpc(
            f"{name}-vpc",
            cidr_block=vpc_cidr,
            enable_dns_hostnames=True,
            enable_dns_support=True,
            tags={
                "Name": f"{name}-vpc",
                "ManagedBy": "pulumi",
            }
        )
        
        self.public_subnets = []
        for i in range(az_count):
            subnet = aws.ec2.Subnet(
                f"{name}-public-subnet-{i}",
                vpc_id=self.vpc.id,
                cidr_block=f"10.0.{i}.0/24",
                availability_zone=f"us-east-1{'abc'[i]}",
                map_public_ip_on_launch=True,
                tags={
                    "Name": f"{name}-public-{i}",
                    "Type": "public",
                }
            )
            self.public_subnets.append(subnet)
        
        # Export outputs
        pulumi.export("vpc_id", self.vpc.id)
        pulumi.export("public_subnet_ids", [s.id for s in self.public_subnets])

# test_networking.py
import unittest
import pulumi
from unittest.mock import MagicMock

class MyMocks(pulumi.runtime.Mocks):
    def new_resource(self, args: pulumi.runtime.MockResourceArgs):
        """Called for every resource constructor. Return mock outputs."""
        outputs = args.inputs.copy()
        
        # Add typical computed fields
        if args.type_ == "aws:ec2/vpc:Vpc":
            outputs["id"] = "vpc-mock12345"
            outputs["arn"] = f"arn:aws:ec2:us-east-1:123456789:vpc/vpc-mock12345"
        
        elif args.type_ == "aws:ec2/subnet:Subnet":
            outputs["id"] = f"subnet-{args.name[-1]}mock"
            outputs["arn"] = f"arn:aws:ec2:us-east-1:123456789:subnet/subnet-mock"
        
        return [f"{args.name}_id", outputs]
    
    def call(self, args: pulumi.runtime.MockCallArgs):
        """Called for SDK functions (data sources)."""
        return {}

pulumi.runtime.set_mocks(MyMocks())

# Import AFTER setting mocks
from infra.networking import NetworkingStack

class TestNetworkingStack(unittest.TestCase):
    
    @pulumi.runtime.test
    def test_vpc_dns_enabled(self):
        """VPC must have DNS hostnames and support enabled."""
        stack = NetworkingStack("test", vpc_cidr="10.0.0.0/16", az_count=2)
        
        def check_vpc(args):
            enable_dns_hostnames, enable_dns_support = args
            self.assertTrue(enable_dns_hostnames, "DNS hostnames must be enabled")
            self.assertTrue(enable_dns_support, "DNS support must be enabled")
        
        return pulumi.Output.all(
            stack.vpc.enable_dns_hostnames,
            stack.vpc.enable_dns_support
        ).apply(check_vpc)
    
    @pulumi.runtime.test
    def test_public_subnets_count(self):
        """Number of subnets must match az_count."""
        stack = NetworkingStack("test", vpc_cidr="10.0.0.0/16", az_count=3)
        self.assertEqual(len(stack.public_subnets), 3)
    
    @pulumi.runtime.test
    def test_subnets_have_required_tags(self):
        """All subnets must have Name and Type tags."""
        stack = NetworkingStack("test", vpc_cidr="10.0.0.0/16", az_count=2)
        
        checks = []
        for subnet in stack.public_subnets:
            def check_tags(tags, subnet_name=subnet._name):
                self.assertIn("Name", tags, f"Subnet {subnet_name} missing Name tag")
                self.assertIn("Type", tags, f"Subnet {subnet_name} missing Type tag")
                self.assertEqual(tags["Type"], "public", 
                                 f"Public subnet {subnet_name} must have Type=public")
            
            checks.append(subnet.tags.apply(check_tags))
        
        return pulumi.Output.all(*checks)
    
    @pulumi.runtime.test
    def test_public_subnets_auto_assign_ip(self):
        """Public subnets must auto-assign public IPs."""
        stack = NetworkingStack("test", vpc_cidr="10.0.0.0/16", az_count=2)
        
        checks = []
        for subnet in stack.public_subnets:
            def check_map_public(map_public_ip, name=subnet._name):
                self.assertTrue(map_public_ip, 
                    f"Public subnet {name} must have map_public_ip_on_launch=True")
            checks.append(subnet.map_public_ip_on_launch.apply(check_map_public))
        
        return pulumi.Output.all(*checks)

if __name__ == "__main__":
    unittest.main()

// infra/security-group.ts
import * as pulumi from "@pulumi/pulumi";
import * as aws from "@pulumi/aws";

export interface SecurityGroupArgs {
  name: string;
  vpcId: pulumi.Input<string>;
  allowedCidrs?: string[];
}

export class AppSecurityGroup extends pulumi.ComponentResource {
  public readonly sg: aws.ec2.SecurityGroup;
  public readonly sgId: pulumi.Output<string>;

  constructor(name: string, args: SecurityGroupArgs) {
    super("custom:security:AppSecurityGroup", name, {}, {});

    this.sg = new aws.ec2.SecurityGroup(
      `${name}-sg`,
      {
        vpcId: args.vpcId,
        description: `Security group for ${name}`,
        ingress: (args.allowedCidrs ?? []).map((cidr) => ({
          fromPort: 443,
          toPort: 443,
          protocol: "tcp",
          cidrBlocks: [cidr],
        })),
        egress: [
          {
            fromPort: 0,
            toPort: 0,
            protocol: "-1",
            cidrBlocks: ["0.0.0.0/0"],
          },
        ],
        tags: { Name: `${name}-sg`, ManagedBy: "pulumi" },
      },
      { parent: this }
    );

    this.sgId = this.sg.id;
    this.registerOutputs({ sgId: this.sgId });
  }
}

// test/security-group.test.ts
import * as pulumi from "@pulumi/pulumi";

// Set up mocks BEFORE importing infra code
pulumi.runtime.setMocks({
  newResource: (args: pulumi.runtime.MockResourceArgs): { id: string; state: Record<string, unknown> } => {
    const outputs: Record<string, unknown> = { ...args.inputs };
    
    if (args.type === "aws:ec2/securityGroup:SecurityGroup") {
      outputs["id"] = "sg-mock12345";
      outputs["arn"] = "arn:aws:ec2:us-east-1:123456789:security-group/sg-mock12345";
    }
    
    return { id: `${args.name}-id`, state: outputs };
  },
  call: (args: pulumi.runtime.MockCallArgs) => {
    return {};
  },
});

import { AppSecurityGroup } from "../infra/security-group";

describe("AppSecurityGroup", () => {
  let sg: AppSecurityGroup;
  
  beforeAll(() => {
    sg = new AppSecurityGroup("test", {
      name: "test-app",
      vpcId: "vpc-12345",
      allowedCidrs: ["10.0.0.0/8", "172.16.0.0/12"],
    });
  });

  test("security group has correct ingress rules", async () => {
    const ingress = await new Promise<aws.ec2.SecurityGroupArgs["ingress"]>(
      (resolve) => sg.sg.ingress.apply(resolve)
    );
    
    expect(ingress).toHaveLength(2);
    expect(ingress![0].fromPort).toBe(443);
    expect(ingress![0].toPort).toBe(443);
    expect(ingress![0].protocol).toBe("tcp");
  });

  test("security group does not allow 0.0.0.0/0 on ingress", async () => {
    const ingress = await new Promise<aws.ec2.SecurityGroupArgs["ingress"]>(
      (resolve) => sg.sg.ingress.apply(resolve)
    );
    
    for (const rule of ingress ?? []) {
      const cidrBlocks = rule.cidrBlocks ?? [];
      expect(cidrBlocks).not.toContain("0.0.0.0/0");
    }
  });

  test("security group has required tags", async () => {
    const tags = await new Promise<Record<string, string>>(
      (resolve) => sg.sg.tags.apply(resolve as any)
    );
    
    expect(tags).toHaveProperty("Name");
    expect(tags).toHaveProperty("ManagedBy", "pulumi");
  });
});

// package.json
{
  "scripts": {
    "test": "jest --testPathPattern='test/.*\\.test\\.ts$'"
  },
  "jest": {
    "preset": "ts-jest",
    "testEnvironment": "node",
    "globals": {
      "ts-jest": {
        "tsconfig": "tsconfig.json"
      }
    }
  }
}

# integration_test.py
import pytest
import random
import string
import boto3
from pulumi import automation as auto

def random_suffix(length: int = 8) -> str:
    return ''.join(random.choices(string.ascii_lowercase + string.digits, k=length))

class TestNetworkingIntegration:
    stack: auto.Stack
    stack_name: str
    aws_region: str = "us-east-1"
    
    @classmethod
    def setup_class(cls):
        """Deploy stack once for all tests in this class."""
        cls.stack_name = f"networking-test-{random_suffix()}"
        
        def pulumi_program():
            from infra.networking import NetworkingStack
            NetworkingStack("test", vpc_cidr="10.200.0.0/16", az_count=2)
        
        cls.stack = auto.create_or_select_stack(
            stack_name=cls.stack_name,
            project_name="terraform-tests",
            program=pulumi_program,
        )
        
        cls.stack.set_config("aws:region", auto.ConfigValue(value=cls.aws_region))
        
        # Deploy
        up_result = cls.stack.up(on_output=print)
        cls.outputs = up_result.outputs
    
    @classmethod
    def teardown_class(cls):
        """Destroy stack after all tests complete."""
        cls.stack.destroy(on_output=print)
        cls.stack.workspace.remove_stack(cls.stack_name)
    
    def test_vpc_created_successfully(self):
        assert "vpc_id" in self.outputs
        
        ec2 = boto3.client("ec2", region_name=self.aws_region)
        vpc_id = self.outputs["vpc_id"].value
        
        response = ec2.describe_vpcs(VpcIds=[vpc_id])
        vpc = response["Vpcs"][0]
        
        assert vpc["CidrBlock"] == "10.200.0.0/16"
        assert vpc["EnableDnsHostnames"] == True
        assert vpc["EnableDnsSupport"] == True
    
    def test_public_subnets_accessible(self):
        ec2 = boto3.client("ec2", region_name=self.aws_region)
        subnet_ids = self.outputs["public_subnet_ids"].value
        
        assert len(subnet_ids) == 2, "Expected 2 public subnets"
        
        response = ec2.describe_subnets(SubnetIds=subnet_ids)
        
        for subnet in response["Subnets"]:
            assert subnet["MapPublicIpOnLaunch"] == True, \
                f"Subnet {subnet['SubnetId']} must auto-assign public IPs"
    
    def test_no_public_cidr_in_security_groups(self):
        """Validate no security group allows unrestricted ingress."""
        ec2 = boto3.client("ec2", region_name=self.aws_region)
        vpc_id = self.outputs["vpc_id"].value
        
        sgs = ec2.describe_security_groups(
            Filters=[{"Name": "vpc-id", "Values": [vpc_id]}]
        )["SecurityGroups"]
        
        for sg in sgs:
            if sg["GroupName"] == "default":
                continue  # Skip default SG
            
            for rule in sg.get("IpPermissions", []):
                for ip_range in rule.get("IpRanges", []):
                    assert ip_range["CidrIp"] != "0.0.0.0/0", \
                        f"SG {sg['GroupId']} allows 0.0.0.0/0 ingress"

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

on:
  pull_request:
    paths:
      - 'infra/**'
      - 'test/**'
  schedule:
    - cron: '0 4 * * *'  # Nightly integration tests

jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      
      - name: Install dependencies
        run: pip install pulumi pulumi-aws pytest
      
      - name: Run unit tests
        run: pytest test/ -v -k "not integration"
  
  integration-tests:
    if: github.event_name == 'schedule' || contains(github.event.pull_request.labels.*.name, 'run-integration')
    runs-on: ubuntu-latest
    needs: unit-tests
    
    permissions:
      id-token: write
      contents: read
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: arn:aws:iam::${{ secrets.TEST_ACCOUNT_ID }}:role/PulumiTestRole
          aws-region: us-east-1
      
      - name: Install Pulumi CLI
        uses: pulumi/actions@v6
      
      - name: Run integration tests
        env:
          PULUMI_ACCESS_TOKEN: ${{ secrets.PULUMI_ACCESS_TOKEN }}
        run: pytest test/ -v -k "integration" --timeout=600

*** Test Cases ***
Infrastructure Endpoint Health
    [Documentation]    Verify load balancer is healthy post-deploy
    ${response}=    GET    ${ALB_DNS_NAME}/health
    Status Should Be    200    ${response}