## Integration testing

URL: https://docs.atlan.com/product/capabilities/build-apps/sdks/application-sdk/how-tos/integration-tests

> Write and run integration tests for App Framework connectors using scenarios, assertions, and locally running Dapr and Temporal services.

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](#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:

```bash
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:

```python

# 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:

```python
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:

```python

# tests/integration/my_connector/scenarios.py

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:

```python

# 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:

```python
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:

```bash
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:

```bash

# .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:

```bash

# .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

```bash

# 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:

```python

# 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.

<!-- vale Microsoft.Headings = NO -->

## Assertion DSL

<!-- vale Microsoft.Headings = YES -->

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

### Basic assertions

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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:

```python
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:

```python
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:

```python
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:

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

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

```bash
env | grep MY_DB_
```

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

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

**"Timeout"**: increase the class-level timeout:

```python
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):

```bash

# 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:

```bash

# Terminal 1 — Dapr + Temporal

uv run poe start-deps

# Terminal 2 — app server

uv run python main.py
```

**Run tests** in a third terminal:

```bash
pytest tests/integration/ -v
```

## See also

- [Unit testing](./unit-tests.md): mock-based tests for task logic and handlers, no external services required
- [Test your apps](../concepts/test-framework.md): overview of the three testing tiers

---