Testing Pulumi Component Resources in Isolation

A component resource has no behaviour you can observe by calling it — it is a constructor whose entire output is a set of resource registrations sent to the Pulumi engine. That makes it awkward to test the way you would test a function, and it is why so many shared components ship with no tests at all and are validated by whoever deploys them next. This guide belongs to Pulumi Component Resources inside Pulumi Patterns & Provider Management, and it shows how to put a single component on a bench, feed it arguments, and assert on exactly what it registered — no account, no stack, no state file.

Context

pulumi.runtime.set_mocks replaces the engine's resource monitor with an object you control. Every resource constructor in the program stops talking to a provider and instead calls your new_resource method, which returns an id and a state dictionary you invent. The general mechanics of that swap are covered in unit testing Pulumi programs with mocks; this page assumes it and spends its time on the part that is specific to components.

The distinction matters because a whole Pulumi program and a single component are different units under test. A program test usually asks "did the stack end up with an encrypted bucket?" A component test asks a narrower and more useful question: given these constructor arguments, which children did this class create, what did it pass them, did it attach them to itself, and what did it refuse to build? Those four questions cover almost every defect a shared component actually has — a forgotten parent, an argument that never reaches the child that needs it, a default that silently disappears when an optional argument is supplied, a validation branch that was written but never taken.

Nothing about this requires the component to be part of a stack. You instantiate the class directly in the test body, exactly as you would instantiate any other Python object, and the registrations land in your recorder instead of AWS.

Prerequisites

• Python 3.9+, pulumi>=3.0, and the provider package the component depends on (pulumi_aws>=6 in the examples below).
• pytest>=7. No pytest-asyncio is needed for the idiom shown here, because pulumi.runtime.test drives the event loop itself.
• A component that takes its configuration as an explicit, annotated argument object rather than **kwargs — see typing Pulumi component inputs and outputs.
• No credentials of any kind. If a test in this file needs AWS_REGION set, something in the component is reaching outside the graph and that is itself the bug.

# CLI: create the test environment for component tests only
python3 -m venv .venv && .venv/bin/pip install "pulumi>=3.0" "pulumi-aws>=6" "pytest>=7"
.venv/bin/pytest tests/ -q

The component used throughout is QueueWorker: it builds a dead-letter queue, a main queue with a redrive policy pointing at it, an execution role, and an inline policy granting that role access to both queues.

Implementation

Step 1 — Install recording mocks before anything constructs a resource

The mock object is where the test gets its evidence. A Mocks subclass that only returns fake ids tells you nothing; one that appends every registration to a list turns the invisible side effects into data you can assert on.

# tests/conftest.py — recording mocks shared by every component test
# CLI: pytest tests/ -q
from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple

import pulumi
import pytest

ACCOUNT = "123456789012"
REGION = "eu-west-1"


@dataclass(frozen=True)
class Registration:
    typ: str
    name: str
    inputs: Dict[str, Any]


class RecordingMocks(pulumi.runtime.Mocks):
    """Records every registration and fabricates the state a provider would return."""

    def __init__(self) -> None:
        self.registrations: List[Registration] = []

    def new_resource(
        self, args: pulumi.runtime.MockResourceArgs
    ) -> Tuple[Optional[str], Dict[str, Any]]:
        self.registrations.append(Registration(args.typ, args.name, dict(args.inputs)))
        state: Dict[str, Any] = dict(args.inputs)
        if args.typ == "aws:sqs/queue:Queue":
            state.setdefault("arn", f"arn:aws:sqs:{REGION}:{ACCOUNT}:{args.name}")
            state.setdefault("url", f"https://sqs.{REGION}.amazonaws.com/{ACCOUNT}/{args.name}")
        if args.typ == "aws:iam/role:Role":
            state.setdefault("arn", f"arn:aws:iam::{ACCOUNT}:role/{args.name}")
        # State implication: this state is returned to the program, never persisted anywhere.
        return f"{args.name}-id", state

    def call(self, args: pulumi.runtime.MockCallArgs) -> Dict[str, Any]:
        if args.token == "aws:index/getCallerIdentity:getCallerIdentity":
            return {"accountId": ACCOUNT, "arn": f"arn:aws:iam::{ACCOUNT}:root", "id": ACCOUNT}
        raise NotImplementedError(f"unmocked provider call: {args.token}")


MOCKS = RecordingMocks()
# Provider note: install at import time so no resource can register against the real monitor.
pulumi.runtime.set_mocks(MOCKS, project="unit-test", stack="test", preview=False)


@pytest.fixture()
def registrations() -> List[Registration]:
    MOCKS.registrations.clear()
    return MOCKS.registrations

Two details earn their place. The state dictionary starts as a copy of the inputs, so any property the component set is readable back — that is what makes the arn-shaped values below deterministic rather than None. And call raises instead of returning {}: a component that quietly invokes aws.get_caller_identity() you did not know about should fail the test loudly, not receive an empty dictionary and produce a policy document with an empty account id in it.

The registrations fixture clears the list per test. Mocks are process-global — one installation serves the whole session — so without that reset the second test in a file sees the first test's children too.

Step 2 — Instantiate the component and assert on the children it created

With the recorder in place, the test is ordinary Python: build the argument object, construct the component, then read the list.

# components/queue_worker.py — the unit under test (abridged)
# CLI: imported by tests/test_queue_worker.py and by __main__.py
from __future__ import annotations

import json
from dataclasses import dataclass
from typing import Optional

import pulumi
import pulumi_aws as aws


@dataclass(frozen=True)
class QueueWorkerArgs:
    visibility_timeout_seconds: int = 60
    max_receive_count: int = 5
    retention_days: int = 4


class QueueWorker(pulumi.ComponentResource):
    def __init__(
        self, name: str, args: QueueWorkerArgs, opts: Optional[pulumi.ResourceOptions] = None
    ) -> None:
        super().__init__("acme:messaging:QueueWorker", name, None, opts)
        if args.max_receive_count < 1:
            raise ValueError("max_receive_count must be at least 1")
        child = pulumi.ResourceOptions(parent=self)

        self.dlq = aws.sqs.Queue(
            f"{name}-dlq", message_retention_seconds=1209600, opts=child
        )
        self.queue = aws.sqs.Queue(
            f"{name}-main",
            visibility_timeout_seconds=args.visibility_timeout_seconds,
            message_retention_seconds=args.retention_days * 86400,
            redrive_policy=self.dlq.arn.apply(
                lambda arn: json.dumps(
                    {"deadLetterTargetArn": arn, "maxReceiveCount": args.max_receive_count}
                )
            ),
            opts=child,
        )
        self.register_outputs({"queue_url": self.queue.url, "dlq_arn": self.dlq.arn})

# tests/test_queue_worker.py — which children, with which inputs
# CLI: pytest tests/test_queue_worker.py -q
from typing import Any, Dict, List

import pulumi

from components.queue_worker import QueueWorker, QueueWorkerArgs
from tests.conftest import Registration


def types_of(regs: List[Registration]) -> List[str]:
    return [r.typ for r in regs if r.typ.startswith("aws:")]


@pulumi.runtime.test
def test_creates_a_main_queue_and_a_dead_letter_queue(registrations: List[Registration]) -> None:
    QueueWorker("orders", QueueWorkerArgs())
    queues = [r for r in registrations if r.typ == "aws:sqs/queue:Queue"]
    assert [r.name for r in queues] == ["orders-dlq", "orders-main"]
    assert types_of(registrations).count("aws:sqs/queue:Queue") == 2


@pulumi.runtime.test
def test_retention_days_are_converted_to_seconds(registrations: List[Registration]) -> None:
    QueueWorker("orders", QueueWorkerArgs(retention_days=14))
    main = next(r for r in registrations if r.name == "orders-main")
    assert main.inputs["messageRetentionSeconds"] == 14 * 86400

The second test is the shape most component bugs take. The component accepts days, the provider wants seconds, and the conversion lives in one line that nobody reads again. Note the key: messageRetentionSeconds, not message_retention_seconds. Inputs reach new_resource after the SDK has translated Python names into the provider's wire names, so assertions are written in camel case. Getting this wrong produces a KeyError that looks like the component never set the property at all.

Filtering on args.typ keeps the assertions honest. The component's own registration carries the type token passed to super().__init__ — acme:messaging:QueueWorker — while the provider-backed children carry tokens beginning aws:, so a count of "how many queues" never accidentally includes the component itself.

Step 3 — Assert the parent/child relationship explicitly

A child constructed without opts=pulumi.ResourceOptions(parent=self) still works. It deploys, the outputs resolve, the tests in step two all pass — and the resource is attached to the root stack rather than the component, so deleting the component orphans it and two instances of the component in one stack fight over the same flat namespace. Nothing in the registration inputs reveals this. The evidence is in the URN.

A URN's third segment is the type chain: the parent's type token, a $, then the resource's own type. A properly parented queue reads acme:messaging:QueueWorker$aws:sqs/queue:Queue. A forgotten parent reads pulumi:pulumi:Stack$aws:sqs/queue:Queue instead, because the engine defaults the parent to the root stack.

# tests/test_parenting.py — catch a child that escaped the component
# CLI: pytest tests/test_parenting.py -q
from typing import Any, List

import pulumi

from components.queue_worker import QueueWorker, QueueWorkerArgs

CHAIN = "acme:messaging:QueueWorker$"


@pulumi.runtime.test
def test_every_child_is_parented_to_the_component() -> None:
    worker = QueueWorker("orders", QueueWorkerArgs())
    urns = pulumi.Output.all(worker.queue.urn, worker.dlq.urn)

    def check(values: List[str]) -> None:
        for urn in values:
            type_chain = urn.split("::")[2]
            assert type_chain.startswith(CHAIN), f"child not parented: {urn}"

    return urns.apply(check)

Splitting on :: and taking index two is deliberate rather than a substring search: it fails if the chain is right but nested one level deeper than intended, which is what happens when someone passes parent=self.queue by mistake and buries the role under the main queue.

Step 4 — Wait for Output assertions instead of firing and forgetting

Everything interesting on a Pulumi resource is an Output, and an assertion inside .apply() runs later — or never. The failure is silent and it is the single most common reason a component test suite is green and worthless.

# tests/test_outputs.py — the returned-Output idiom, and the trap it avoids
# CLI: pytest tests/test_outputs.py -q
import json
from typing import List

import pulumi

from components.queue_worker import QueueWorker, QueueWorkerArgs


@pulumi.runtime.test
def test_redrive_policy_points_at_the_dead_letter_queue() -> None:
    worker = QueueWorker("orders", QueueWorkerArgs(max_receive_count=3))

    def check(values: List[str]) -> None:
        policy_json, dlq_arn = values
        policy = json.loads(policy_json)
        assert policy["deadLetterTargetArn"] == dlq_arn
        assert policy["maxReceiveCount"] == 3

    # The decorator awaits whatever the function RETURNS. Dropping the return
    # here makes the test pass without ever running check().
    return pulumi.Output.all(worker.queue.redrive_policy, worker.dlq.arn).apply(check)

pulumi.runtime.test wraps the function, runs it inside the runtime's event loop, and awaits the value it returns. Return the Output produced by apply and the callback is guaranteed to have completed — with any AssertionError propagating — before pytest records the result. Omit the return and pytest sees a function that returned None and reports a pass.

A cheap way to make that mistake impossible to repeat is a lint rule, or a helper that both applies and returns:

# tests/support.py — one place where the return can be forgotten
# CLI: pytest tests/ -q
from typing import Any, Callable, List

import pulumi


def assert_all(*outputs: pulumi.Output[Any], check: Callable[[List[Any]], None]) -> pulumi.Output[None]:
    return pulumi.Output.all(*outputs).apply(check)

Step 5 — Prove the component rejects arguments it cannot honour

Validation that raises before super().__init__ finishes never reaches the engine, so it is testable with plain pytest.raises and no Output machinery at all. Test both directions: the rejection, and the boundary value that must still be accepted.

# tests/test_validation.py — the guard clauses, both sides of the boundary
# CLI: pytest tests/test_validation.py -q
import pulumi
import pytest

from components.queue_worker import QueueWorker, QueueWorkerArgs


@pulumi.runtime.test
def test_zero_receive_count_is_rejected() -> None:
    with pytest.raises(ValueError, match="max_receive_count must be at least 1"):
        QueueWorker("orders", QueueWorkerArgs(max_receive_count=0))


@pulumi.runtime.test
def test_one_receive_count_is_accepted(registrations: list) -> None:
    QueueWorker("orders", QueueWorkerArgs(max_receive_count=1))
    assert any(r.name == "orders-main" for r in registrations)

Matching on the message text, not just the exception class, is what stops the test from passing for the wrong reason — a ValueError raised by an unrelated int() conversion three lines earlier satisfies a bare pytest.raises(ValueError).

Verification

Run the suite and confirm two things: that it is green, and that it is actually asserting.

# CLI: run the component tests and prove the assertions execute
.venv/bin/pytest tests/ -q
.venv/bin/pytest tests/ -q --collect-only | tail -n 3
# Provider note: no AWS credentials and no PULUMI_ACCESS_TOKEN are read by any of these tests.

Then break the component on purpose. Delete opts=child from the dead-letter queue and re-run — test_every_child_is_parented_to_the_component must fail. Change retention_days * 86400 to retention_days * 3600 — the conversion test must fail. A component test suite that survives both mutations is measuring nothing, and finding that out takes ninety seconds.

Finally, check for the silent-pass failure mode across the whole file:

# CLI: find apply-based tests that forget to return their Output
grep -rn "\.apply(" tests/ | grep -v "return" | grep -v "support.py"

Any hit is a test whose callback may never run. Some are legitimate — an apply used to build a value rather than to assert — but each one needs a human to say so.

Gotchas & Edge Cases

Mocks installed after the first resource is constructed do nothing. If a module builds resources at import time and pytest imports it before conftest.py runs set_mocks, the program registers against a monitor that is not there and raises Exception: Program run without the Pulumi engine available; re-run using the pulumi CLI. Keep component construction inside test bodies or fixtures, never at module scope.

preview=True makes assertions evaporate. Under preview, computed outputs are unknown, and Output.apply skips its callback for unknown values rather than passing None into it. A test written against preview=True therefore returns an unknown Output, runs no assertions, and passes. Use preview=False for component tests and reach for preview mode only when the behaviour under test is specifically "what happens when this value is unknown".

Duplicate names are not caught. The real engine rejects two resources with the same URN; the mock monitor computes URNs and hands them back without a uniqueness check. A component that hardcodes a child name instead of deriving it from name looks fine in tests and fails on the second instantiation in a real stack. Cover it with a test that builds the component twice under different names and asserts the child names differ.

Wire names, not Python names. args.inputs uses the provider's property names — visibilityTimeoutSeconds, assumeRolePolicy, messageRetentionSeconds. Assert with main.inputs["visibilityTimeoutSeconds"], and expect a bare KeyError: 'visibility_timeout_seconds' if you slip.

A misspelled provider argument fails at construction, not in the assertion. Writing visibility_timout_seconds=60 raises TypeError: Queue._internal_init() got an unexpected keyword argument 'visibility_timout_seconds' before any mock is consulted. That is the SDK's own typing doing useful work, and it is a reason to keep generated provider classes rather than wrapping every child in a helper that takes a dictionary.

Secretness is invisible from inside new_resource. Inputs arrive there as plain values, so a recorder cannot tell you whether the component wrapped a password in pulumi.Output.secret. Assert on the resource attribute instead, using pulumi.Output.is_secret(worker.queue.redrive_policy) and the same returned-Output idiom.

Operational Notes

Keep component tests in the package that ships the component, not in the stacks that consume it. A component published as a library is versioned and released independently; its tests belong to the same release so a consumer upgrading from 2.3.0 to 2.4.0 can see that the wiring assertions still held.

Decide deliberately what these tests are not for. They prove the component asked for the right resources with the right arguments. They cannot tell you the IAM policy grants the permissions the worker actually needs at runtime, that the CIDR ranges do not overlap an existing network, that the region supports the instance family, or that a change is an update rather than a replacement. Those belong to a preview-based check or an ephemeral stack, and pretending mock tests cover them is how a suite becomes a false sense of safety.

Run them in the fast lane of the pipeline. A recorder-based suite for a handful of components runs in under a second, needs no credentials, and can therefore live in the same job as mypy and ruff — well before anything that needs a cloud role. If your component tests take long enough that someone proposes moving them to a nightly job, something in the component is doing real work at construction time and should be moved out.

Record the type token in one constant. Every parenting assertion depends on the exact string passed to super().__init__, and a token typo is a real defect: it changes every child URN and forces a replacement of the whole tree on the next deployment. Importing the token from the component module rather than retyping it in the test makes that impossible to get wrong silently — and makes an intentional rename show up as one failing test rather than a quiet redeployment of production.

FAQ

Do I need pytest-asyncio to test component resources?

No, if you use pulumi.runtime.test. That decorator runs the test function inside the runtime's own event loop and awaits whatever it returns, which covers Outputs and coroutines. Adding pytest-asyncio on top usually produces two competing event loops and an attached to a different loop error.

How do I test a component that reads a data source?

Implement call in the mocks and return the shape the provider function documents, keyed by the same names the generated Python function returns. Raise NotImplementedError for every other token so an unmocked lookup shows up as an explicit failure rather than an empty dictionary.

Can one Mocks instance serve the whole test suite?

Yes, and it usually should — set_mocks is process-global, so installing it once in conftest.py is simpler than per-test installation. Just clear the recorder between tests, or every assertion after the first sees a growing pile of registrations from earlier tests.

How do I assert on the value of register_outputs?

Read the component's public attributes instead. register_outputs publishes values to the engine for display and for stack references, and the mock monitor does not hand them back to you; the attributes you assigned in __init__ hold the same Outputs and are what consumers use anyway.

Should the mocked state include realistic ARNs?

Yes, when anything in the component parses them. A component that builds a policy document by slicing a queue ARN will happily produce nonsense from "arn" as a literal string, and the test that checks the document will pass. Fabricate values with the real shape and the parsing logic gets tested for free.

What is left to test after all of this?

Whether the component deploys. Stand up one ephemeral stack per release that instantiates the component with its default arguments, runs pulumi up, checks a runtime signal, and destroys itself. Mock tests catch wiring mistakes in seconds; only an apply catches the provider rejecting the combination of arguments you assembled.

Related

• Unit Testing Pulumi Programs with Mocks — the general mocking technique this page specialises for components.
• Typing Pulumi Component Inputs and Outputs — the annotated argument object that makes these tests readable.
• Building a Reusable VPC Component in Pulumi (Python) — a fuller component to point this harness at.
• Pulumi Component Resources — the parent topic covering construction, parenting and packaging.