## End-to-end testing

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

> Validate the complete production data flow for App Framework connectors with a live Atlan tenant using BaseE2ETest or SQLAppE2ETest.

End-to-end tests validate the complete production data flow on a live Atlan tenant. The connector worker runs inside a Docker compose container on the CI runner, registered on a unique per-run queue. The Automation Engine drives the full DAG—extract → query intelligence → publish → lineage—while the rest of the system apps run inside the live tenant cluster. The test then queries Atlan to verify that the expected assets and lineage edges were created.

These tests are triggered by the `e2e` PR label or `workflow_dispatch`, and take 10–20 minutes to complete.

## How end-to-end tests work

The test framework submits an Automation Engine workflow to the live tenant, routing extraction to your connector worker via a unique per-run queue name. Once the AE workflow completes, it queries Atlan using pyatlan to assert that the expected entity counts and lineage edges are present.

The connector worker runs inside Docker compose on the CI runner. For connectors with a source system, a sibling container runs alongside it hosting the source system. The rest of the DAG (publish, query intelligence, lineage) runs in the tenant cluster—nothing else is simulated.

## Project structure

```
tests/
 e2e/
 test_<connector>_e2e.py
.github/
 e2e/
 e2e-full-components/
 objectstore.yaml # S3-via-tenant objectstore binding
 e2e-full-docker-compose.yaml # per-run deployment name + OAuth env
 make-secrets-e2e-full.py # flat secret bundle for AGENT credential flow
 workflows/
 e2e-full.yaml # delegates to SDK reusable workflow
```

## Create test class

Choose the base class that matches your connector type.

### Non-SQL connectors: `BaseE2ETest`

For connectors that aren't SQL databases (APIs, SaaS tools, etc.), extend `BaseE2ETest` directly:

```python

# tests/e2e/test_openapi_e2e.py

from application_sdk.testing.e2e import BaseE2ETest, RunMode

class TestOpenAPIE2E(BaseE2ETest):
 connector_short_name = "openapi"
 connection_type = "api" # overrides connector_short_name in the connection QN
 argo_package_name = "@atlan/openapi"
 argo_template_name = "atlan-openapi"
 mode = RunMode.AGENT

 expected_min_asset_counts = {
 "APISpec": 1,
 "APIPath": 10,
 }
 expect_lineage = False
```

### SQL connectors: `SQLAppE2ETest`

For SQL database connectors, extend `SQLAppE2ETest`, which adds SQL-specific wiring on top of `BaseE2ETest`: automatic agent and connection spec generation, SQL filter handling, and query intelligence knobs:

```python

# tests/e2e/test_mysql_e2e.py

from application_sdk.testing.e2e import RunMode, SQLAppE2ETest
from application_sdk.testing.e2e.payload import DatabaseSpec

class TestMySQLE2E(SQLAppE2ETest):
 connector_short_name = "mysql"
 argo_package_name = "@atlan/mysql"
 argo_template_name = "atlan-mysql"
 mode = RunMode.AGENT

 include_filter = r"^def\.e2e_main$"
 exclude_filter = ""

 expected_min_asset_counts = {
 "Database": 1,
 "Schema": 1,
 "Table": 2,
 "View": 1,
 "Column": 10,
 }
 expect_lineage = True

 def database_spec(self) -> DatabaseSpec:
 return DatabaseSpec(
 host="mysql",
 port=3306,
 username="e2e_user",
 password="e2e_pass",
 connector_config_name="atlan-connectors-mysql",
 )
```

### Key attributes

| Attribute | Class | Purpose |
|-----------|-------|---------|
| `connector_short_name` | Both | Connector identifier (for example, `"mysql"`, `"openapi"`) |
| `argo_package_name` | Both | Argo package name for AE workflow submission |
| `argo_template_name` | Both | Argo template name |
| `mode` | Both | `RunMode.AGENT` for credential-routed extraction |
| `connection_type` | Both | Override when the Atlan connection type differs from `connector_short_name` (for example, OpenAPI uses `"api"`) |
| `expected_min_asset_counts` | Both | Minimum entity counts to assert in Atlan after the DAG completes |
| `expect_lineage` | Both | Whether to assert lineage edges were created |
| `include_filter` | `SQLAppE2ETest` | Anchored regex scoped to the hermetic seed dataset |
| `database_spec()` | `SQLAppE2ETest` | Host, port, and credentials for the sibling DB container |

### `include_filter` for SQL connectors

SQL connector templates substitute `include_filter` directly into a `REGEXP` SQL clause. Pass an anchored regex string—not a JSON dict:

```python

# Correct: anchored regex string

include_filter = r"^def\.e2e_main$"

# Wrong: JSON dict shape (causes SQL syntax error)

include_filter = {"schema": "e2e_main"}
```

## Configure Compose overlay

Create `.github/e2e/e2e-full-docker-compose.yaml` to set the per-run deployment name and OAuth credentials on the connector container:

```yaml
services:
 atlan-app:
 environment:
 ATLAN_DEPLOYMENT_NAME: "e2e-full-ci-${GITHUB_RUN_ID}"
 ATLAN_AUTH_CLIENT_ID: "${ATLAN_AUTH_CLIENT_ID}"
 ATLAN_AUTH_CLIENT_SECRET: "${ATLAN_AUTH_CLIENT_SECRET}"
 mysql:
 image: mysql:8.0
 environment:
 MYSQL_ROOT_PASSWORD: root
 MYSQL_DATABASE: e2e_main
 MYSQL_USER: e2e_user
 MYSQL_PASSWORD: e2e_pass
 volumes:
 - .github/e2e/seed.sql:/docker-entrypoint-initdb.d/seed.sql
```

The `ATLAN_DEPLOYMENT_NAME` value registers the worker on a unique Temporal queue per CI run, so concurrent runs don't interfere with each other.

## Configure objectstore binding

Create `.github/e2e/e2e-full-components/objectstore.yaml` to replace the default local storage with the tenant S3 proxy:

```yaml
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
 name: bindings.aws.s3
spec:
 type: bindings.aws.s3
 version: v1
 metadata:
 - name: bucket
 value: "atlan-bucket"
 - name: endpoint
 value: "https://<tenant>/api/blobstorage"
 - name: authType
 value: "sigv4"
```

## Write secrets script

Create `.github/e2e/make-secrets-e2e-full.py` to produce a flat secret bundle for the AGENT credential flow (`key-type: single-key`):

```python

secrets = {
 "MY_CONNECTOR_USERNAME": "e2e_user",
 "MY_CONNECTOR_PASSWORD": "e2e_pass",
}

out = pathlib.Path(".github/e2e/secrets")
out.mkdir(parents=True, exist_ok=True)
(out / "credentials.json").write_text(json.dumps(secrets))
```

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

## Add GitHub Actions workflow

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

Create `.github/workflows/e2e-full.yaml`, which delegates entirely to the SDK reusable workflow:

```yaml
name: E2E Full-DAG

on:
 pull_request:
 types: [labeled]
 workflow_dispatch:

jobs:
 e2e-full:
 if: github.event_name == 'workflow_dispatch' || contains(github.event.pull_request.labels.*.name, 'e2e')
 uses: atlanhq/application-sdk/.github/workflows/e2e-full-reusable.yaml@main
 secrets: inherit
 with:
 app-name: my-connector
 app-image-name: atlan-my-connector-app
 secrets-script: .github/e2e/make-secrets-e2e-full.py
 compose-overlay: .github/e2e/e2e-full-docker-compose.yaml
 components-dir: .github/e2e/e2e-full-components
 test-path: tests/e2e/
```

## Required CI secrets

| Secret | Purpose |
|--------|---------|
| `SDR_TEST_TENANT` | Tenant domain |
| `SDR_CLIENT_ID` | OAuth client ID |
| `SDR_CLIENT_SECRET` | OAuth client secret |
| `ATLAN_BASE_URL` | Tenant base URL |
| `ATLAN_API_KEY` | API key for AE management endpoints |
| `SDR_OAUTH_CLIENT_ID` | OAuth client ID for objectstore proxy |
| `SDR_OAUTH_CLIENT_SECRET` | OAuth client secret for objectstore proxy |

`ATLAN_API_KEY` is separate from the OAuth pair—the OAuth client lacks the realm-admin permissions needed for Automation Engine management endpoints.

## See also

- [Unit testing](./unit-tests.md): mock-based tests for task logic and handlers, no external services required
- [Integration testing](./integration-tests.md): scenario-based tests with a running app server
- [Test your apps](../concepts/test-framework.md): overview of all three testing tiers

---