Unit Testing Pulumi Programs with Mocks

Unit testing a Pulumi program with pulumi.runtime.set_mocks runs your resource graph in-process with the cloud provider replaced by a stub, letting you assert on resource inputs and outputs in milliseconds without credentials or state. This how-to belongs to Testing Python Infrastructure Code within Python IaC fundamentals and strategy, and it is the fastest, lowest layer of the testing pyramid for Pulumi.

Context

A Pulumi program builds a graph of resources whose values are wrapped in Output and resolved asynchronously against cloud APIs. Mocks intercept that resolution: every resource registration returns a fake id and a dictionary of state you control, so you can verify that your code passes the right inputs and wires outputs correctly. Because nothing reaches a provider, these tests are deterministic and need no AWS account — the same approach the Pulumi patterns and provider management section uses to test its component resources.

Mechanically, set_mocks swaps the gRPC resource monitor — the engine endpoint a Pulumi program normally talks to — for an in-process shim backed by your subclass. Two methods carry the whole contract. new_resource is invoked once per RegisterResource call and receives a MockResourceArgs carrying typ (the fully-qualified Pulumi type token, such as aws:s3/bucketV2:BucketV2), name (the logical name you passed), inputs (the serialised property bag after the SDK's own defaulting), provider, and id. It returns a (id, outputs) pair that becomes the resource's resolved state. call is invoked for provider functions — the get_* data sources — and receives a MockCallArgs with a token like aws:index/getAvailabilityZones:getAvailabilityZones, the arguments, and the provider reference.

Everything a mocked test can assert flows from those two hooks. You see exactly what your program declared: the inputs after SDK defaulting but before the provider has validated anything. You do not see what the provider would have done with them. That boundary is the single most useful thing to hold in mind while writing these tests, because it explains both what they catch instantly and what they will never catch.

The practical consequence: assert on decisions your code makes, not on cloud behaviour. "This bucket has versioning on" is a decision. "This bucket name is globally unique" is cloud behaviour, and a mock will happily accept a name S3 would reject.

Prerequisites

• Python 3.9+, pulumi>=3.0, and the relevant provider package (e.g. pulumi_aws>=6).
• pytest>=7 and pytest-asyncio>=0.21 (Output resolution is async).
• A pytest.ini or pyproject.toml setting asyncio_mode = "auto" or per-test @pytest.mark.asyncio.
• No cloud credentials — mocks never call a provider.

Implementation

1. Define a Mocks subclass and install it before importing infra code

set_mocks must run before the module under test constructs any resource, so install it in an autouse fixture or at import time.

# CLI: pytest tests/test_bucket.py -v
# State implication: new_resource returns fabricated state; no real bucket, no state file.
from typing import Any
import pulumi

class InfraMocks(pulumi.runtime.Mocks):
    def new_resource(
        self, args: pulumi.runtime.MockResourceArgs
    ) -> tuple[str, dict[str, Any]]:
        outputs = {**args.inputs}
        # Provider note: fabricate provider-computed fields the program reads downstream.
        if args.typ == "aws:s3/bucket:Bucket":
            outputs["arn"] = f"arn:aws:s3:::{args.name}"
        return f"{args.name}-id", outputs

    def call(self, args: pulumi.runtime.MockCallArgs) -> dict[str, Any]:
        return {}  # stub provider function (data source) calls

pulumi.runtime.set_mocks(InfraMocks(), preview=False)

2. Resolve outputs and assert on inputs

With mocks installed, import the module that defines your resources and await its outputs through the Output future.

# CLI: pytest tests/test_bucket.py::test_bucket_is_versioned -v
import pulumi
import pulumi_aws as aws
import pytest

def make_bucket(name: str) -> aws.s3.Bucket:
    return aws.s3.Bucket(name, versioning={"enabled": True})

@pytest.mark.asyncio
async def test_bucket_is_versioned() -> None:
    bucket = make_bucket("logs")

    def check(args: list[Any]) -> None:
        enabled, arn = args
        assert enabled is True            # asserts the input we passed
        assert arn.startswith("arn:aws:s3:::")  # asserts mocked output

    pulumi.Output.all(
        bucket.versioning.enabled, bucket.arn
    ).apply(check)

3. Toggle preview mode to test plan-time behavior

Pass preview=True to set_mocks when you need to verify behavior under pulumi preview, where some outputs are unknown.

# CLI: pytest tests/test_preview.py -v
import pulumi
pulumi.runtime.set_mocks(InfraMocks(), preview=True)
# State implication: under preview, computed outputs may be unknown — guard apply() accordingly.

4. Record every registration and assert across the whole graph

Per-resource tests get repetitive once a stack has thirty resources. A recording mock inverts the pattern: let the program build the entire graph, capture every registration, then write assertions as queries over the recording. This is where mock testing earns its keep, because rules like "every S3 bucket in this program has server-side encryption configured" cannot be expressed one resource at a time.

# tests/conftest.py — a recording mock installed once for the whole session
# CLI: pytest tests/ -q
from typing import Any
from dataclasses import dataclass, field
import pulumi
import pytest


@dataclass
class Registration:
    typ: str
    name: str
    inputs: dict[str, Any]


@dataclass
class RecordingMocks(pulumi.runtime.Mocks):
    seen: list[Registration] = field(default_factory=list)

    def new_resource(
        self, args: pulumi.runtime.MockResourceArgs
    ) -> tuple[str, dict[str, Any]]:
        self.seen.append(Registration(args.typ, args.name, dict(args.inputs)))
        outputs = {**args.inputs, "arn": f"arn:aws:{args.typ.split(':')[0]}:::{args.name}"}
        return f"{args.name}-id", outputs

    def call(self, args: pulumi.runtime.MockCallArgs) -> dict[str, Any]:
        return {}


@pytest.fixture(scope="session")
def graph() -> RecordingMocks:
    mocks = RecordingMocks()
    # State implication: mocks are process-global. Installing once per session and
    # importing the program once keeps the recording consistent across tests.
    pulumi.runtime.set_mocks(mocks, project="tests", stack="unit", preview=False)
    import infra.stack  # noqa: F401  — constructing the graph is the point
    return mocks


def test_every_bucket_is_encrypted(graph: RecordingMocks) -> None:
    buckets = [r for r in graph.seen if r.typ.startswith("aws:s3/bucketV2")]
    assert buckets, "no buckets were registered — did the import fail?"
    encrypted = {r.name for r in graph.seen
                 if r.typ.startswith("aws:s3/bucketServerSideEncryptionConfigurationV2")}
    missing = [b.name for b in buckets if not any(b.name in e for e in encrypted)]
    assert not missing, f"buckets without encryption config: {missing}"

The assert buckets guard is not decoration. If the import silently fails or the mock is installed after the module loads, the list comprehension returns empty and every "no resource violates the rule" assertion passes vacuously. A test suite full of vacuously-passing invariants is worse than no suite at all, so every graph-level assertion should first prove the graph is non-empty.

5. Stub provider functions so data sources resolve

Any program calling aws.get_availability_zones(), aws.ec2.get_ami(), or aws.get_caller_identity() routes through call, and the default return {} above makes those return empty structures. Code that then indexes into the result fails with a bare KeyError or IndexError that says nothing about the cause. Dispatch on the token instead.

# tests/mocks.py — realistic responses for the data sources this program uses
# CLI: pytest tests/test_network.py -v
from typing import Any
import pulumi

_RESPONSES: dict[str, dict[str, Any]] = {
    "aws:index/getAvailabilityZones:getAvailabilityZones": {
        "names": ["us-east-1a", "us-east-1b", "us-east-1c"],
        "zoneIds": ["use1-az1", "use1-az2", "use1-az4"],
    },
    "aws:index/getCallerIdentity:getCallerIdentity": {
        "accountId": "123456789012",
        "arn": "arn:aws:iam::123456789012:role/deployer",
        "userId": "AROAEXAMPLE:session",
    },
}


class NetworkMocks(pulumi.runtime.Mocks):
    def new_resource(
        self, args: pulumi.runtime.MockResourceArgs
    ) -> tuple[str, dict[str, Any]]:
        return f"{args.name}-id", {**args.inputs}

    def call(self, args: pulumi.runtime.MockCallArgs) -> dict[str, Any]:
        # Provider note: keys are camelCase here — `call` returns the wire shape,
        # which the SDK converts to snake_case before your program sees it.
        try:
            return _RESPONSES[args.token]
        except KeyError:
            raise AssertionError(f"unmocked provider function: {args.token}")

Raising on an unknown token is deliberate. It converts "this data source returned nothing and the test failed somewhere far away" into "you added a data source and forgot to mock it", which is a one-line fix instead of a debugging session. The camelCase note matters too: call operates below the SDK's naming translation, so returning zone_ids instead of zoneIds produces a response the SDK silently drops.

Verification

Run the suite; a passing run proves the resource inputs and the mocked outputs match your expectations without any provider call.

# CLI: confirm Pulumi unit tests are hermetic and green
pytest tests/ -q
# Provider note: no PULUMI_ACCESS_TOKEN or AWS creds required for this layer.

Gotchas & Edge Cases

Assertions inside apply can be swallowed. If a test only registers an apply callback and returns, pytest may finish before the callback runs. Prefer awaiting through pytest-asyncio and the Output future, or use pulumi.Output.all(...).apply(...) and ensure the test framework drives the event loop to completion.

set_mocks installed too late has no effect. Resources constructed at module import before the mock is installed register against the real runtime. Install mocks in conftest.py or before importing the infrastructure module.

call must be implemented for data sources. Programs that use aws.get_* functions will fail unless your Mocks call returns a dictionary matching the function's expected outputs.

Running the program without mocks fails with an engine error. Importing an infrastructure module in a plain python -c session, or in a test where the fixture did not run, raises Exception: Program run without the Pulumi engine available; re-run using the pulumi CLI. It is a clear message, but it appears at import time and pytest reports it as a collection error rather than a test failure, so it is easy to misread as a packaging problem.

Mocks are process-global, not per-test. set_mocks writes into pulumi.runtime.settings, and there is no supported teardown. Two tests that install different mocks in the same process interact through whichever ran last, and pytest-xdist distributing them across workers hides the problem until ordering changes. Install one mock set per session, or run genuinely different mock configurations as separate pytest invocations.

Under preview=True, apply callbacks may never run. Unknown outputs cause apply to skip the callback entirely rather than pass a sentinel. A test whose only assertion lives inside that callback then passes without asserting anything. When testing preview behaviour, assert on something outside the callback — that a resource was registered at all, or that a guard branch was taken.

Secrets are plain values in args.inputs. The mock monitor sits below the secret-wrapping layer, so a value passed through pulumi.Output.secret() arrives in the recording as an ordinary string. Do not print args.inputs wholesale in a failing assertion message if the program handles real secrets in configuration.

The type token changes between provider major versions. aws:s3/bucket:Bucket became aws:s3/bucketV2:BucketV2 in pulumi_aws 6, and a mock dispatching on the old token silently stops fabricating outputs — the resource still registers, so nothing errors, and downstream assertions fail with confusing values. Pin the provider and grep the tokens after any major upgrade.

Operational Notes

Mock-based tests run in milliseconds because no provider is ever called, which makes them cheap enough to run on every commit. Assert on the things that cause real incidents: that a bucket is encrypted, that a security group does not open a wide port, that a required tag is present. These are the same rules you might later promote to a CrossGuard policy, so writing them as unit tests first is a natural stepping stone.

Keep the program's resource-construction logic separate from any imperative lookups so the pure part is trivially testable. When a test needs a computed output, remember that under mocks outputs resolve synchronously to the values your mock returns, so structure assertions around pulumi.Output.all(...).apply(...) rather than expecting bare values.

Structure the code for testability rather than fighting the runtime. A module that builds resources at import time can only be exercised once per process, which is why the session-scoped fixture above imports it exactly once. A module that exposes a build(config: StackConfig) -> Outputs function instead can be called repeatedly with different configurations in the same session, and each call produces a fresh set of registrations you can record separately. That single refactor is usually what turns a stack with three tests into one with thirty.

Watch what these tests cost in maintenance. Every fabricated output in new_resource is a small lie about the provider, and lies drift. If a test asserts on an ARN your mock invented, it is really asserting on the mock. Restrict fabrication to fields the program genuinely reads downstream, and let everything else pass through from args.inputs untouched — that keeps the surface where the mock can diverge from reality as small as possible.

Finally, be explicit about where this layer stops. Mocks never validate provider schemas, so a required argument you forgot is not caught here; pulumi preview against a real provider is. Run the mocked suite on every commit and a preview on every pull request, and the two together cover the ground that neither covers alone. The testing pyramid discussion in the parent topic places the remaining layers — snapshot tests, property-based tests, and live integration runs — around these two.

FAQ

What is the difference between mocking Pulumi and mocking boto3 with moto? pulumi.runtime.set_mocks stubs the Pulumi engine's resource registration, so it tests how your program declares infrastructure. moto stubs the AWS SDK, so it tests boto3 calls your helper code makes outside the Pulumi graph. Use both at the same layer of the pyramid for different code paths.

Can I assert that a resource was created with a specific name? Yes. MockResourceArgs.name and args.inputs are available in new_resource, so you can record them into a list and assert on the captured calls after the program runs.

Do mocks work for ComponentResource subclasses? Yes. Components register their children through the same runtime, so each child hits new_resource. This is exactly how component tests in the Pulumi patterns and provider management section verify nested resource graphs.

Why does my test pass even though the assertion is wrong? Almost always because the assertion lives inside an apply callback that never ran, or because a list comprehension over the recorded graph returned empty. Add a positive assertion first — that the collection is non-empty, or that the callback executed — before asserting the property you care about.

How do I mock a ComponentResource that wraps other providers? You do not need to mock it specially. Its children register through the same monitor, so new_resource sees each one with the child's own type token. Assert on the children; the component itself appears as a registration with the component's type and no meaningful outputs.

Do mocks catch a missing required argument? No. The mock monitor accepts whatever the SDK serialises, and required-field validation lives in the provider. A resource missing a required argument registers cleanly in a mocked test and fails at pulumi preview against the real provider — which is why both layers belong in CI.

Related

• Testing Python Infrastructure Code — the testing pyramid this unit layer anchors.
• Mocking AWS Services with moto in pytest — mock the AWS SDK for boto3 calls outside the Pulumi graph.
• Pulumi Patterns & Provider Management — production component patterns that rely on this Mocks-based testing approach.