End-to-end testing

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:

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

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

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

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:

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

import json, pathlib

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))

Add GitHub Actions workflow

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

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.

How end-to-end tests work​

Project structure​

Create test class​

Non-SQL connectors: BaseE2ETest​

SQL connectors: SQLAppE2ETest​

Key attributes​

include_filter for SQL connectors​

Configure Compose overlay​

Configure objectstore binding​

Write secrets script​

Add GitHub Actions workflow​

Required CI secrets​

See also​