Testing Pulumi Dynamic Providers in Python

A dynamic provider is the only part of a Pulumi program you can call directly from a test without starting an engine, because it is just a class with five methods. This guide sits under Dynamic Providers and Custom Resources in Pulumi, itself part of Pulumi Patterns & Provider Management, and shows how to exercise the whole create/read/update/delete lifecycle offline, pin the diff replace decisions with parametrized cases, and keep a small live-API suite that runs on a schedule rather than on every commit.

Context

Most Pulumi resource code resists unit testing. A pulumi_aws.s3.BucketV2 declaration produces Output values that only resolve inside a running engine, which is why testing it means unit testing with mocks and asserting on registered resource properties. A dynamic.ResourceProvider is different. Its methods take plain dictionaries and return plain result objects — CreateResult, ReadResult, UpdateResult, DiffResult — with no Output anywhere in the signature. You can import the class in a test file, instantiate it, call create({...}), and assert on the return value in microseconds.

That property is worth exploiting, because a dynamic provider is also the riskiest code in the program. A first-class provider was written and regression-tested by the vendor; yours was written last Tuesday, and a wrong replaces list in diff will delete a production resource during what the operator believed was an in-place edit. The running example here is a webhook subscription on a monitoring SaaS: it has a target URL, a list of event types, a signing secret, and an endpoint_region that the vendor fixes at creation.

The provider is an ordinary class with a single impure edge, which is what makes it unit-testable.

Prerequisites

Python 3.9+ with pulumi>=3.0, pytest>=7.0, and dill (already a transitive dependency of the Pulumi SDK).
An existing provider class to test — the one built in writing a Pulumi dynamic provider in Python has the right shape.
A pytest.ini or pyproject.toml where custom markers can be registered, so -m contract is not a typo waiting to happen.
Read/write credentials on a throwaway account of the remote service, used only by the slow suite.

# CLI: install the test dependencies and register the marker
pip install "pulumi>=3.0" pytest dill
printf '[pytest]\nmarkers =\n    contract: hits the real remote API (slow, needs credentials)\n' > pytest.ini

Implementation

1. Give the provider a client seam

The single reason a provider is hard to test is that its methods reach for requests inline. Move the HTTP calls behind a typing.Protocol and accept an implementation in __init__, defaulting to the real one. The provider keeps working unchanged in a deployment, and a test can hand it an in-memory double.

Depending on a Protocol rather than the requests module puts the seam one layer below the lifecycle methods.

# providers/webhook.py — imported by __main__.py; deployed with: pulumi up
from typing import Any, Dict, List, Optional, Protocol

class WebhookApi(Protocol):
    def create_hook(self, body: Dict[str, Any]) -> Dict[str, Any]: ...
    def get_hook(self, hook_id: str) -> Optional[Dict[str, Any]]: ...
    def update_hook(self, hook_id: str, body: Dict[str, Any]) -> Dict[str, Any]: ...
    def delete_hook(self, hook_id: str) -> None: ...


class HttpWebhookApi:
    """The real transport. Holds no session object — see the pickle note below."""

    def __init__(self, base_url: str, token: str) -> None:
        self.base_url = base_url
        self.token = token

    def create_hook(self, body: Dict[str, Any]) -> Dict[str, Any]:
        import requests  # Provider note: import inside the method, not at module scope.
        r = requests.post(f"{self.base_url}/hooks", json=body,
                          headers={"Authorization": f"Bearer {self.token}"}, timeout=15)
        r.raise_for_status()
        return r.json()

The fake lives in conftest.py and is a dictionary with a counter. It should be strict: raise on an unknown id rather than returning None silently, so a test that passes the wrong id fails loudly.

# tests/conftest.py — CLI: pytest -q
from typing import Any, Dict, Optional
import pytest

class FakeWebhookApi:
    def __init__(self) -> None:
        self.hooks: Dict[str, Dict[str, Any]] = {}
        self.calls: list[str] = []
        self._next = 0

    def create_hook(self, body: Dict[str, Any]) -> Dict[str, Any]:
        self.calls.append("create")
        self._next += 1
        hook_id = f"hk_{self._next:04d}"
        record = {"id": hook_id, "endpoint_region": body.get("endpoint_region", "eu-1"), **body}
        self.hooks[hook_id] = record
        return record

    def get_hook(self, hook_id: str) -> Optional[Dict[str, Any]]:
        self.calls.append("get")
        return self.hooks.get(hook_id)

    def update_hook(self, hook_id: str, body: Dict[str, Any]) -> Dict[str, Any]:
        self.calls.append("update")
        if hook_id not in self.hooks:
            raise KeyError(f"404 no such hook: {hook_id}")
        self.hooks[hook_id].update(body)
        return self.hooks[hook_id]

    def delete_hook(self, hook_id: str) -> None:
        self.calls.append("delete")
        self.hooks.pop(hook_id, None)   # delete is idempotent on this API

@pytest.fixture
def api() -> FakeWebhookApi:
    return FakeWebhookApi()

2. Assert the CreateResult id and outs contract

create has a contract that the engine relies on and that nothing else enforces: id_ must be a non-empty string that read and delete can later use, and outs must contain every property that anyone might reference as an output — including values the remote API generated, which are not in props. Forget one and the stack output resolves to None at apply time with no error.

A create test drives the method directly and asserts on the id and the outs dictionary it returns.

# tests/test_webhook_provider.py — CLI: pytest tests/test_webhook_provider.py -q
from typing import Any, Dict
from providers.webhook import WebhookProvider

PROPS: Dict[str, Any] = {
    "target_url": "https://ops.internal/alerts",
    "event_types": ["incident.opened", "incident.resolved"],
    "secret": "s3cr3t",
    "endpoint_region": "eu-1",
}

def test_create_returns_usable_id_and_full_outs(api) -> None:
    provider = WebhookProvider(api=api)
    result = provider.create(PROPS)
    assert result.id_ == "hk_0001"
    # State implication: outs become the checkpoint; a missing key is an unresolvable output.
    assert result.outs["hook_id"] == "hk_0001"
    assert result.outs["target_url"] == PROPS["target_url"]
    assert set(PROPS).issubset(result.outs)

def test_create_then_read_round_trips(api) -> None:
    provider = WebhookProvider(api=api)
    created = provider.create(PROPS)
    read = provider.read(created.id_, created.outs)
    # State implication: refresh writes read.outs back over the checkpoint.
    assert read.id_ == created.id_
    assert read.outs["event_types"] == PROPS["event_types"]

def test_delete_is_idempotent(api) -> None:
    provider = WebhookProvider(api=api)
    created = provider.create(PROPS)
    provider.delete(created.id_, created.outs)
    provider.delete(created.id_, created.outs)   # a retried delete must not raise
    assert api.hooks == {}

The set(PROPS).issubset(result.outs) assertion is the cheapest guard against the most common regression: someone adds an input to the resource class, wires it into the API body, and forgets to merge it into outs.

3. Parametrize the diff replace decisions

diff is pure — two dictionaries in, a DiffResult out — so it deserves exhaustive coverage rather than one happy-path test. Write the table of expected outcomes first and treat it as the specification of which properties the remote API considers immutable.

Each row becomes one parametrized case; the replaces column is the one that destroys data when it is wrong.

# tests/test_webhook_diff.py — CLI: pytest tests/test_webhook_diff.py -q
from typing import Any, Dict, List
import pytest
from providers.webhook import WebhookProvider

OLD: Dict[str, Any] = {
    "target_url": "https://ops.internal/alerts",
    "event_types": ["incident.opened", "incident.resolved"],
    "secret": "s3cr3t",
    "endpoint_region": "eu-1",
}

@pytest.mark.parametrize("patch,changes,replaces", [
    ({}, False, []),
    ({"target_url": "https://ops.internal/v2"}, True, []),
    ({"event_types": ["incident.resolved", "incident.opened"]}, False, []),
    ({"secret": "rotated"}, True, []),
    ({"endpoint_region": "us-1"}, True, ["endpoint_region"]),
    ({"endpoint_region": "us-1", "secret": "rotated"}, True, ["endpoint_region"]),
])
def test_diff_replace_decisions(patch: Dict[str, Any], changes: bool,
                                replaces: List[str], api) -> None:
    provider = WebhookProvider(api=api)
    result = provider.diff("hk_0001", OLD, {**OLD, **patch})
    assert result.changes is changes
    # Provider note: a wrong entry here deletes and recreates the live subscription.
    assert sorted(result.replaces or []) == sorted(replaces)

Two rows carry most of the value. The reordered event_types row asserts that the provider normalizes an unordered list before comparing, which is the usual cause of a diff on every preview. The last row asserts that a mixed edit still reports only the immutable property in replaces — the engine replaces the resource once and applies the mutable change as part of that replacement, so listing extra keys is noise that misleads whoever reads the preview.

4. Test the serialization constraint explicitly

Pulumi serializes the provider instance into stack state (as the __provider output) using dill, and deserializes it on every subsequent operation. Anything the instance holds must survive that round trip. A requests.Session, a boto3 client, an open file, or a threading.Lock does not, and the failure only appears at pulumi up time with TypeError: cannot pickle '_thread.lock' object — long after the tests went green.

# tests/test_webhook_serialization.py — CLI: pytest tests/test_webhook_serialization.py -q
import dill
from providers.webhook import HttpWebhookApi, WebhookProvider

def test_provider_survives_a_dill_round_trip() -> None:
    provider = WebhookProvider(api=HttpWebhookApi("https://api.example.com", "tok"))
    # State implication: this is exactly what Pulumi stores under __provider.
    restored = dill.loads(dill.dumps(provider))
    assert isinstance(restored, WebhookProvider)
    assert restored.api.base_url == "https://api.example.com"

This is also where the closure trap bites. If a fixture builds the provider from a factory defined inside the test function, or monkeypatches a module attribute the provider reads at call time, the object under test is not the object that would be serialized. Keep the fake as a module-level class in conftest.py and inject it as an argument. Never reach for monkeypatch.setattr("providers.webhook.requests", fake) — it passes locally and proves nothing about the deployed provider.

Verification

Run the fast suite, confirm the contract tests were excluded, and check that diff has no unvisited branches.

The whole loop runs offline; only the separate contract suite touches the real service.

# CLI: fast suite only, with branch coverage on the provider module
pytest -m "not contract" -q --cov=providers.webhook --cov-branch --cov-report=term-missing

# CLI: prove the marker actually excluded something
pytest -m contract --collect-only -q | tail -1

# CLI: the end-to-end check — a second preview after an apply must be empty
pulumi up --yes && pulumi preview --diff

The pulumi preview --diff line is the real acceptance test for diff. If the second preview reports any change on an untouched program, the provider is comparing something it should be normalizing, and one of the parametrized rows above is missing.

Operational Notes

Contract tests answer a question the fakes cannot: does the remote API still behave the way the fake pretends it does? Keep them in tests/contract/, mark every one, and skip cleanly when credentials are absent so a fresh clone still runs pytest without failures.

The contract suite is two orders of magnitude slower, which is why it runs behind a marker.

# tests/contract/test_webhook_live.py — CLI: pytest -m contract -q
import os
from typing import Any, Dict
import pytest
from providers.webhook import HttpWebhookApi, WebhookProvider

pytestmark = [
    pytest.mark.contract,
    pytest.mark.skipif(not os.getenv("WEBHOOK_API_TOKEN"),
                       reason="WEBHOOK_API_TOKEN not set"),
]

def test_live_lifecycle() -> None:
    api = HttpWebhookApi("https://api.example.com", os.environ["WEBHOOK_API_TOKEN"])
    provider = WebhookProvider(api=api)
    props: Dict[str, Any] = {
        "target_url": "https://example.invalid/ci",
        "event_types": ["incident.opened"],
        "secret": "ci-secret",
        "endpoint_region": "eu-1",
    }
    created = provider.create(props)
    try:
        assert provider.read(created.id_, created.outs).outs["target_url"] == props["target_url"]
        updated = provider.update(created.id_, created.outs,
                                  {**created.outs, "target_url": "https://example.invalid/ci2"})
        assert updated.outs["target_url"] == "https://example.invalid/ci2"
    finally:
        # Provider note: teardown in finally, or a failed assert leaks a live subscription.
        provider.delete(created.id_, created.outs)

Run that suite nightly rather than per commit. When it fails and the fast suite passes, the fake has drifted from the vendor's behaviour — fix the fake, add the case to the parametrized diff table, and only then fix the provider. The same discipline applies to the SaaS resources covered in managing external SaaS resources, where the vendor changes the API without telling you.

Gotchas & Edge Cases

check failures are silent if untested. If the provider implements check, its CheckFailure(property=..., reason=...) entries are the only user-facing validation. Assert on them directly: assert provider.check({}, {"event_types": []}).failures[0].property == "event_types".

Secrets in outs leak into assertions. Anything the provider returns in outs lands in state. If a test asserts the plaintext signing secret appears there, the test is documenting a leak. Return a fingerprint instead, and assert the plaintext is absent.

A fake that never fails hides retry bugs. Add a mode to FakeWebhookApi that raises on the first call and succeeds on the second, and use it to prove create does not leave an orphaned remote object when a retry happens.

Comparing floats and datetimes. If the remote API returns "1.0" where the program supplies 1, diff sees a change forever. Coerce both sides in one helper and unit-test the helper — this is the single most common cause of a permanently dirty preview.

Coverage of delete during replacement. The engine calls delete with the old outs during a replacement, not the new inputs. A test that only ever passes freshly created outs to delete will not catch a provider that reads a key which exists only in the new shape.

FAQ

Can I test a dynamic provider without any Pulumi runtime at all?

Yes. Importing pulumi.dynamic is enough to get the result classes; nothing in the test path starts an engine or reads a stack. That is the whole advantage over testing resource declarations, which need pulumi.runtime.set_mocks.

Do I still need `pulumi.runtime.set_mocks` anywhere?

Only for the thin Resource subclass that wraps the provider, if you want to assert that inputs reach it correctly. The provider's own logic needs no mocks, so keep the two suites separate rather than pushing everything through the mock harness.

How do I test that a resource was actually replaced?

Assert on DiffResult.replaces in a unit test, then confirm the behaviour once end to end with pulumi preview --diff after editing the immutable property. The preview prints ++ pulumi-python:dynamic:Resource for a replacement, which is the string to look for in CI output.

Writing a Pulumi Dynamic Provider in Python — the provider implementation these tests exercise.
Handling Diffs and Updates in Pulumi Dynamic Providers — the semantics behind the replace decisions asserted above.
Unit Testing Pulumi Programs with Mocks — the complementary technique for ordinary resource declarations.
Testing Python IaC — the broader testing topic this guide plugs into.