Integration testing

Integration tests exercise the full request-to-app run path of your connector with locally running infrastructure. Unlike unit tests, which use mock infrastructure and run without any external services, integration tests start a real app server and make actual HTTP requests through the handler API, validating responses with a declarative assertion DSL. These tests typically run in CI; see Run locally if you want to run them on your machine.

How integration tests work

Integration tests are built on BaseIntegrationTest, a pytest-based class that drives HTTP calls to a running app server. You define scenarios: lightweight data objects that specify which API to call, what inputs to send, and what the response must contain. The framework runs each scenario as a separate pytest test.

Project structure

Organize integration tests to mirror your source code layout—when every source module has a predictable test counterpart, finding and adding tests is straightforward:

tests/integration/
├── __init__.py
├── conftest.py           # shared fixtures
├── _example/             # reference implementation (copy this)
│   ├── __init__.py
│   ├── scenarios.py
│   └── test_integration.py
└── my_connector/
    ├── __init__.py
    ├── scenarios.py
    └── test_integration.py

Start from the reference implementation:

cp -r tests/integration/_example tests/integration/my_connector

Define scenarios

Edit scenarios.py to describe the test cases for your connector. Each Scenario object declares one API call, the inputs it receives, and the assertions that must pass on the response.

Use names that describe what's being tested and the expected outcome—these become the pytest test identifiers:

# Clear: what is being tested and why it should fail
"auth_wrong_password"
"preflight_nonexistent_database"
"workflow_empty_schema_filter"

# Unclear: gives no information about the scenario
"test_1"
"scenario_b"
"negative_case"

For non-obvious scenarios, add a description field to document the condition being tested:

Scenario(
    name="preflight_read_only_user",
    description=(
        "User has SELECT on all tables but no CREATE permission. "
        "Preflight should pass with a warning, not fail."
    ),
    api="preflight",
    args=lazy(lambda: {...}),
    assert_that={
        "success": equals(True),
        "data.status": equals("ready"),
    },
)

A complete scenarios.py example:

# tests/integration/my_connector/scenarios.py
import os

from application_sdk.testing.integration import (
    Scenario,
    lazy,
    equals,
    exists,
    is_true,
    one_of,
)


def load_credentials():
    return {
        "host": os.getenv("MY_DB_HOST"),
        "username": os.getenv("MY_DB_USER"),
        "password": os.getenv("MY_DB_PASSWORD"),
    }


scenarios = [
    # Authentication — valid credentials
    Scenario(
        name="auth_valid",
        api="auth",
        args=lazy(lambda: {"credentials": load_credentials()}),
        assert_that={
            "success": equals(True),
            "data.status": equals("success"),
        },
    ),

    # Authentication — wrong password
    Scenario(
        name="auth_invalid_password",
        api="auth",
        args=lazy(lambda: {
            "credentials": {**load_credentials(), "password": "wrong"}
        }),
        assert_that={"success": equals(False)},
    ),

    # Preflight — valid configuration
    Scenario(
        name="preflight_valid",
        api="preflight",
        args=lazy(lambda: {
            "credentials": load_credentials(),
            "metadata": {"databases": ["MY_DB"]},
        }),
        assert_that={
            "success": equals(True),
            "data.status": equals("ready"),
            "data.checks.0.passed": equals(True),
        },
    ),

    # Workflow — start extraction
    Scenario(
        name="workflow_extraction",
        api="workflow",
        args=lazy(lambda: {
            "credentials": load_credentials(),
            "metadata": {"databases": ["MY_DB"]},
            "connection": {"name": "test_conn"},
        }),
        assert_that={
            "success": equals(True),
            "data.workflow_id": exists(),
            "data.run_id": exists(),
        },
    ),
]

Supported APIs

Value	Endpoint	Purpose
"auth"	/workflows/v1/auth	Test authentication
"preflight"	/workflows/v1/check	Validate configuration
"metadata"	/workflows/v1/metadata	Fetch metadata list
"workflow"	/workflows/v1/start	Start a workflow
"config"	/workflows/v1/config/{id}	Get or update a credential config

Create test class

Edit test_integration.py to declare a test class that inherits from BaseIntegrationTest and references your scenarios:

# tests/integration/my_connector/test_integration.py
from application_sdk.testing.integration import BaseIntegrationTest
from .scenarios import scenarios


class MyConnectorTest(BaseIntegrationTest):
    scenarios = scenarios
    server_host = "http://localhost:8000"

BaseIntegrationTest generates one pytest test per scenario. Each test calls the specified API, validates all assertions, and reports a clear failure message if any assertion doesn't pass.

Set environment variables

Never hardcode credentials in scenario files or commit real credentials to version control. Reference environment variables and load them at test-run time:

def load_credentials():
    return {
        "host": os.getenv("MY_DB_HOST"),
        "username": os.getenv("MY_DB_USER"),
        "password": os.getenv("MY_DB_PASSWORD"),
    }

Set the variables before running tests:

export MY_DB_HOST=localhost
export MY_DB_USER=test_user
export MY_DB_PASSWORD=YOUR_PASSWORD_HERE

Use a .env file for local development—never commit real credentials:

# .env — excluded from version control
MY_DB_HOST=localhost
MY_DB_USER=test_user
MY_DB_PASSWORD=YOUR_PASSWORD_HERE

Commit a .env.example with placeholder values as documentation:

# .env.example — committed as documentation
MY_DB_HOST=your-host-here
MY_DB_USER=your-username-here
MY_DB_PASSWORD=your-password-here

In CI, populate these from repository secrets.

Run tests

# All integration tests
pytest tests/integration/ -v

# One connector
pytest tests/integration/my_connector/ -v

# One scenario by name
pytest tests/integration/my_connector/ -v -k "auth_valid"

# With debug logging to see full API responses
pytest tests/integration/my_connector/ -v --log-cli-level=DEBUG

Lazy evaluation for credentials

Wrap credential loading in lazy() so the lambda is evaluated at test-run time, not at import time. This prevents failures when environment variables aren't set in the environment where scenarios are imported:

# Wrong: evaluated at import time — fails if env vars are missing
args={"credentials": load_credentials()}

# Correct: evaluated when the test runs
args=lazy(lambda: {"credentials": load_credentials()})

lazy() caches the result after the first evaluation, so credentials are loaded once per test run.

Assertion DSL

The assertion DSL provides composable predicate functions for validating nested response fields using dot-notation paths.

Basic assertions

from application_sdk.testing.integration import (
    equals, not_equals, exists, is_none, is_true, is_false
)

assert_that = {
    "success": equals(True),
    "data.status": equals("success"),
    "data.error": is_none(),
    "data.workflow_id": exists(),
}

Collection assertions

from application_sdk.testing.integration import (
    one_of, not_one_of, contains, not_contains,
    has_length, is_empty, is_not_empty
)

assert_that = {
    "data.status": one_of(["RUNNING", "COMPLETED"]),
    "message": contains("successful"),
    "data.checks": is_not_empty(),
}

Numeric assertions

from application_sdk.testing.integration import (
    greater_than, greater_than_or_equal, less_than, between
)

assert_that = {
    "data.count": greater_than(0),
    "data.duration_ms": between(1, 30000),
}

Type assertions

from application_sdk.testing.integration import is_dict, is_list, is_string, is_type

assert_that = {
    "data": is_dict(),
    "data.checks": is_list(),
    "data.workflow_id": is_string(),
}

Combinators

from application_sdk.testing.integration import all_of, any_of, none_of

assert_that = {
    "data.name": all_of(exists(), is_string(), is_not_empty()),
    "data.role": any_of(equals("admin"), equals("superuser")),
    "message": none_of(contains("error"), contains("fail")),
}

Custom assertions

from application_sdk.testing.integration import custom

assert_that = {
    "data.count": custom(lambda x: x % 2 == 0, "must_be_even"),
    "data.value": lambda x: 0 < x < 100,
}

Advanced test class configuration

Custom app run endpoint

class MyConnectorTest(BaseIntegrationTest):
    scenarios = scenarios
    server_host = "http://localhost:8000"
    workflow_endpoint = "/extract"   # overrides the default /start
    timeout = 60                     # seconds; default is 30

Override the endpoint per-scenario:

Scenario(
    name="mine_queries",
    api="workflow",
    endpoint="/mine",
    args={...},
    assert_that={...},
)

Setup and teardown hooks

Use hooks to create and clean up test data around the entire test run:

class MyConnectorTest(BaseIntegrationTest):
    scenarios = scenarios

    @classmethod
    def setup_test_environment(cls):
        """Called once before any scenario runs."""
        cls.db = create_test_database_connection()
        cls.db.execute("CREATE SCHEMA IF NOT EXISTS test_schema")

    @classmethod
    def cleanup_test_environment(cls):
        """Called once after all scenarios finish."""
        cls.db.execute("DROP SCHEMA test_schema CASCADE")
        cls.db.close()

    def before_scenario(self, scenario):
        """Called before each individual scenario."""
        print(f"Starting: {scenario.name}")

    def after_scenario(self, scenario, result):
        """Called after each individual scenario."""
        status = "PASSED" if result.success else "FAILED"
        print(f"{scenario.name}: {status}")

Skip slow scenarios

Mark individual scenarios to skip in routine runs:

Scenario(
    name="workflow_full_extraction",
    api="workflow",
    args={...},
    assert_that={...},
    skip=True,
    skip_reason="Full extraction takes >10 minutes; run manually only",
)

Troubleshooting

"Server not available": confirm the app server is running and accessible:

curl http://localhost:8000/server/health

"Credentials not loading": check environment variables are exported in the current shell:

env | grep MY_DB_

"Assertion failed": run with debug logging to see the raw API response:

pytest -v --log-cli-level=DEBUG -k "failing_scenario_name"

"Timeout": increase the class-level timeout:

class MyConnectorTest(BaseIntegrationTest):
    timeout = 120

Run locally

Integration tests require Dapr and Temporal to be installed and running, plus the app server. In CI these are provided by the workflow; locally you need to set them up yourself.

Install the CLIs (one-time):

# Dapr CLI
curl -fsSL https://raw.githubusercontent.com/dapr/cli/master/install/install.sh | /bin/bash
dapr init --slim

# Temporal CLI
curl -sSf https://temporal.download/cli.sh | sh

Start services and the app server, each in its own terminal:

# Terminal 1 — Dapr + Temporal
uv run poe start-deps

# Terminal 2 — app server
uv run python main.py

Run tests in a third terminal:

pytest tests/integration/ -v

See also

• Unit testing: mock-based tests for task logic and handlers, no external services required
• Test your apps: overview of the three testing tiers