Unit testing

Unit tests for App Framework applications use in-memory mocks for every infrastructure dependency. Tests run in milliseconds with no credentials, no network calls, and no running services. This page covers the mock module, how to write task and handler tests, and recommended practices.

How mock infrastructure works

The application_sdk.testing module provides drop-in replacements for every infrastructure service the framework uses in production. These mocks implement the same Protocol interfaces as the production services, so your @task and Handler methods call them identically—the only difference is that no network traffic is made.

Wire the mocks into the framework using InfrastructureContext and set_infrastructure():

import pytest
from application_sdk.infrastructure import InfrastructureContext, set_infrastructure, clear_infrastructure
from application_sdk.testing import MockSecretStore, MockStateStore

@pytest.fixture
def infra():
    ctx = InfrastructureContext(
        secret_store=MockSecretStore({"db-password": "test-secret"}),
        state_store=MockStateStore(),
    )
    set_infrastructure(ctx)
    yield ctx
    clear_infrastructure()

The yield + clear_infrastructure() pattern resets the module-level singleton after each test, preventing state from leaking between tests. Don't put clear_infrastructure() at the end of test functions—that call is skipped if the test raises.

Pre-populate MockSecretStore with the secrets your task reads so tests never depend on environment variables or external vaults.

Unit testing @task methods

Given an App subclass with a @task method that reads from the secret store:

from application_sdk.app import App, task
from application_sdk.contracts import Input, Output

class FetchInput(Input):
    connection_id: str

class FetchOutput(Output):
    rows_fetched: int

class MyConnector(App):
    @task
    async def fetch_data(self, input: FetchInput) -> FetchOutput:
        password = await self.context.get_secret("db-password")
        # ... use password to connect and fetch rows
        return FetchOutput(rows_fetched=42)

The test injects mock infrastructure and calls the task directly—no orchestration layer involved:

import pytest
from application_sdk.infrastructure import InfrastructureContext, set_infrastructure, clear_infrastructure
from application_sdk.testing import MockSecretStore, MockStateStore
from application_sdk.testing.fixtures import clean_app_registry  # noqa: F401

@pytest.fixture
def infra():
    ctx = InfrastructureContext(
        secret_store=MockSecretStore({"db-password": "test-secret"}),
        state_store=MockStateStore(),
    )
    set_infrastructure(ctx)
    yield ctx
    clear_infrastructure()

@pytest.mark.asyncio
async def test_fetch_data(infra):
    connector = MyConnector()
    output = await connector.fetch_data(FetchInput(connection_id="test-conn"))
    assert output.rows_fetched > 0

Always test both the happy path and failure paths. A test suite that only verifies success cases is incomplete:

from application_sdk.infrastructure import SecretNotFoundError

@pytest.mark.asyncio
async def test_fetch_raises_on_missing_secret(infra):
    # Secret not populated — SecretNotFoundError should propagate
    connector = MyConnector()
    with pytest.raises(SecretNotFoundError):
        await connector.fetch_data(FetchInput(connection_id="test"))

Test handlers

Handler methods can be tested in isolation without starting a server by calling them directly with a MockSecretStore providing any credentials the handler reads:

import pytest
from application_sdk.handler.base import Handler
from application_sdk.handler.contracts import AuthInput, AuthOutput, AuthStatus
from application_sdk.infrastructure import InfrastructureContext, set_infrastructure, clear_infrastructure
from application_sdk.testing import MockSecretStore, MockStateStore

class MyHandler(Handler):
    async def test_auth(self, input: AuthInput) -> AuthOutput:
        api_key = await self.context.get_secret("api-key")
        if api_key == "valid-key":
            return AuthOutput(status=AuthStatus.SUCCESS, message="OK")
        return AuthOutput(status=AuthStatus.FAILED, message="Invalid key")

@pytest.fixture
def infra():
    ctx = InfrastructureContext(
        secret_store=MockSecretStore({"api-key": "valid-key"}),
        state_store=MockStateStore(),
    )
    set_infrastructure(ctx)
    yield ctx
    clear_infrastructure()

@pytest.mark.asyncio
async def test_auth_succeeds_with_valid_key(infra):
    handler = MyHandler()
    result = await handler.test_auth(AuthInput(credentials={}))
    assert result.status == AuthStatus.SUCCESS

@pytest.mark.asyncio
async def test_auth_fails_with_wrong_key(infra):
    infra.secret_store.set("api-key", "wrong-key")
    handler = MyHandler()
    result = await handler.test_auth(AuthInput(credentials={}))
    assert result.status == AuthStatus.FAILED

MockSecretStore exposes a set(name, value) method for adding or overriding secrets within a test.

Call tracking

All mock classes record every call made to them. Use these records to verify your task interacted with infrastructure correctly—not just that it returned the right output:

@pytest.mark.asyncio
async def test_fetch_persists_state(infra):
    connector = MyConnector()
    await connector.fetch_data(FetchInput(connection_id="test-conn"))

    save_calls = infra.state_store.get_save_calls()
    assert len(save_calls) == 1
    key, value = save_calls[0]
    assert key == "last_run"

This catches bugs where a task returns the correct value but fails to persist state, which can cause incorrect behavior on retry.

Available call-tracking methods by mock:

Mock	Tracking methods
`MockSecretStore`	`get_get_calls()`, `get_get_optional_calls()`
`MockStateStore`	`get_save_calls()`, `get_load_calls()`, `get_delete_calls()`, `get_list_keys_calls()`
`MockPubSub`	`get_publish_calls(topic=None)`, `get_published_messages(topic=None)`
`MockBinding`	`get_invocations(operation=None)`

`clean_app_registry` fixture

App subclasses register themselves in the global AppRegistry at class-definition time. Without resetting the registry between tests, registrations accumulate and can cause unexpected behavior—particularly when testing multiple App subclasses in the same session.

Add clean_app_registry (and clean_task_registry) to your top-level conftest.py so they apply automatically:

# tests/conftest.py
from application_sdk.testing.fixtures import (  # noqa: F401
    clean_app_registry,
    clean_task_registry,
)

The # noqa: F401 comment suppresses the "imported but unused" linter warning—pytest discovers fixtures through the import, not through explicit use.

Available mocks and fixtures

The application_sdk.testing module exports all mocks and fixtures from a single location:

from application_sdk.testing import (
    # Mocks
    MockSecretStore,
    MockStateStore,
    MockPubSub,
    MockBinding,
    MockCredentialStore,
    MockHeartbeatController,
    # Pytest fixtures
    app_context,
    clean_app_registry,
    clean_task_registry,
    mock_secret_store,
    mock_state_store,
    mock_pubsub,
    mock_binding,
    mock_credential_store,
    mock_heartbeat,
)

The fixture methods (mock_secret_store, mock_state_store, etc.) return a fresh instance for each test. Use them when you don't need to configure the mock before the test runs; use the class directly when you need to pre-populate data (such as seeding secrets in MockSecretStore).

How mock infrastructure works​

Unit testing @task methods​

Test handlers​

Call tracking​

clean_app_registry fixture​

Available mocks and fixtures​

See also​