Pipeline E2E and Smoke Testing Across Environments Guide¶

End-to-end (E2E) and smoke tests execute pipelines across multiple execution modes and storage backends.

• Specification-driven smoke suite: tests/cdm_data_loaders/pipelines/test_pipeline_smoke_spec.py
• Unified test runner: tests/cdm_data_loaders/pipelines/framework.py (PipelineTestRunner)
• Environment resolver: tests/pipeline_env.py (SmokeEnvConfig, resolve_smoke_env)
• dlt config isolation: tests/dlt_config_isolation.py (write_dlt_config_dir)

1. Execution Modes and Storage Backends¶

PipelineTestRunner executes a pipeline scenario in one of three modes:

Built-in storage environments in tests/cdm_data_loaders/pipelines/test_pipeline_smoke_spec.py:

2. Running Built-In Environments¶

Run All Smoke Scenarios (local_fs and s3_moto)¶

uv run pytest tests/cdm_data_loaders/pipelines/test_pipeline_smoke_spec.py -m "not requires_ceph"

Filter by Pipeline (--smoke-pipeline)¶

Pass --smoke-pipeline once or multiple times to restrict the suite to specific pipelines:

uv run pytest tests/cdm_data_loaders/pipelines/test_pipeline_smoke_spec.py \
  --smoke-pipeline pubmed \
  -m "not requires_ceph"

Filter by Environment or Execution Mode (-k)¶

Run only local_fs in core mode:

uv run pytest tests/cdm_data_loaders/pipelines/test_pipeline_smoke_spec.py -k "local_fs and core"

Run only s3_moto in cli mode:

uv run pytest tests/cdm_data_loaders/pipelines/test_pipeline_smoke_spec.py -k "s3_moto and cli"

Run Against Local Ceph (s3_ceph)¶

When a local Ceph Nano container is running, select the requires_ceph marker:

uv run pytest tests/cdm_data_loaders/pipelines/test_pipeline_smoke_spec.py -m requires_ceph

3. Running Against a Custom Environment¶

When --dlt-config-dir or --settings-file is provided on the pytest command line, the test parametrization switches from the built-in local_fs / s3_moto / s3_ceph fixtures to a single custom environment.

Step 1: Create a dlt Configuration Directory (--dlt-config-dir)¶

Create a directory (for example, /tmp/my-env/dlt_conf) containing config.toml and secrets.toml:

# /tmp/my-env/dlt_conf/config.toml
[destination.filesystem]
bucket_url = "s3://my-staging-lake/smoke-tests"

# /tmp/my-env/dlt_conf/secrets.toml
[destination.filesystem.credentials]
endpoint_url = "https://s3.staging.example.org"
aws_access_key_id = "YOUR_ACCESS_KEY"
aws_secret_access_key = "YOUR_SECRET_KEY"

Any destination supported by dlt (for example, filesystem, duckdb, snowflake, postgres) or Iceberg catalog configuration can be specified inside config.toml and secrets.toml.

Step 2: Create a Settings File (--settings-file)¶

Create a TOML, YAML, or JSON settings file (for example, /tmp/my-env/settings.toml) with top-level shared settings or per-pipeline sections under [pipelines.<pipeline_name>]:

# /tmp/my-env/settings.toml
use_destination = "filesystem"
output_dir = "/tmp/my-env/output"
loader_file_format = "parquet"

[pipelines.pubmed]
buffer_size = 500

[pipelines.ncbi_datasets]
buffer_size = 100

Settings Precedence (lowest to highest): 1. Scenario settings defined in specs/pipelines/<pipeline>/<case>/settings.toml and case.yaml. 2. Top-level keys in --settings-file (filtered to valid fields of the target pipeline's Settings model). 3. Pipeline-specific keys in [pipelines.<pipeline_name>] inside --settings-file. 4. Per-run isolated input_dir and output_dir generated by the test runner.

Step 3: Execute the Smoke Test Suite¶

uv run pytest tests/cdm_data_loaders/pipelines/test_pipeline_smoke_spec.py \
  --dlt-config-dir /tmp/my-env/dlt_conf \
  --settings-file /tmp/my-env/settings.toml \
  --smoke-pipeline ncbi_datasets

4. Running with Apache Iceberg (loader_file_format = "iceberg" and .pyiceberg.yaml)¶

When testing Iceberg output against a catalog (such as a Polaris REST catalog or SQL catalog) configured via .pyiceberg.yaml:

1. Place .pyiceberg.yaml in $HOME (~/.pyiceberg.yaml):
2. Standard PyIceberg configuration lives at $HOME/.pyiceberg.yaml.
3. load_pipeline_env_from_dir() and PipelineTestRunner ignore any .pyiceberg.yaml placed inside --dlt-config-dir.
4. In Docker mode (--run-docker), PipelineTestRunner stages $HOME/.pyiceberg.yaml from the host into /input_dir/dlt_conf/.pyiceberg.yaml, and scripts/entrypoint.sh copies it to $HOME/.pyiceberg.yaml inside the container.
5. Manage Table Lifecycle Across Runs in a Scratch Catalog:
6. Because preset pipelines such as ncbi_datasets and pubmed use a fixed dataset_name (ncbi_datasets or pubmed), consecutive test cases writing to a persistent catalog (such as Polaris) will append to existing catalog tables unless tables are dropped before or after each run.
7. Use either pytest CLI flags or settings.toml keys to control catalog table cleanup:

Both drop_existing_tables and drop_tables_after can be set at the top level of settings.toml or under [pipelines.<pipeline_name>].

Example settings.toml for running ncbi_datasets with Iceberg and a Polaris REST catalog configured in ~/.pyiceberg.yaml:

# /tmp/my-env/settings.toml
use_destination = "s3"
loader_file_format = "iceberg"
drop_existing_tables = true
drop_tables_after = true

Example pytest invocation:

uv run pytest tests/cdm_data_loaders/pipelines/test_pipeline_smoke_spec.py \
  --dlt-config-dir /tmp/my-env/dlt_conf \
  --settings-file /tmp/my-env/settings.toml \
  --smoke-pipeline ncbi_datasets \
  --drop-existing-tables \
  --drop-tables-after

5. Running in Docker Mode (--run-docker)¶

Enable the docker execution mode by passing --run-docker and --docker-image (defaults to cdm-data-loaders:latest):

# Build the image first
docker build -t cdm-data-loaders:latest .

# Run smoke tests in Docker against a custom S3 environment
uv run pytest tests/cdm_data_loaders/pipelines/test_pipeline_smoke_spec.py \
  --dlt-config-dir /tmp/my-env/dlt_conf \
  --settings-file /tmp/my-env/settings.toml \
  --run-docker \
  --docker-image cdm-data-loaders:latest \
  -k "docker"

How PipelineTestRunner handles Docker mounts and environment variables: - Local paths: When input_dir, output_dir, or dlt bucket_url are local paths, PipelineTestRunner bind-mounts them into /mnt/input, /mnt/output, and /mnt/dlt_dest inside the container. - Remote URLs (s3://...): When input_dir or output_dir are s3:// URLs, PipelineTestRunner passes the remote URLs to the CLI command inside the container without bind-mounting local directories. - dlt configuration: PipelineTestRunner mounts the isolated dlt_conf directory at /mnt/dlt_conf and sets DLT_PROJECT_DIR=/mnt/dlt_conf and DLT_DATA_DIR=/mnt/dlt_data inside the container. - S3 credentials: AWS/S3 environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_S3_ENDPOINT, FSSPEC_S3_ENDPOINT_URL, FSSPEC_S3_KEY, FSSPEC_S3_SECRET) are forwarded into the container when set.

6. Per-Run Isolation on Shared Buckets¶

When running against a shared S3 bucket or shared local directory, test_pipeline_smoke_spec.py isolates each test invocation using a deterministic run_slug:

<pipeline>_<run_id>_<mode>_<8-char-uuid>

• Input files: Staged under <input_base>/<run_slug>/ (both local paths and s3:// URLs via fsspec).
• Output reports and logs: Written under <output_base>/<run_slug>/.
• dlt destination: Written under <bucket_url>/<run_slug> with preserve_dlt_destination=True.

This prevents table collisions, schema pollution, or report overwrites when multiple scenarios or pipelines target the same bucket.

7. What Every Run Verifies¶

Regardless of execution mode (core, cli, docker) or storage backend (local_fs, s3_moto, s3_ceph, custom), assert_smoke_run_output() in tests/cdm_data_loaders/pipelines/helpers.py verifies:

1. Exit Code Contract: Even when a run records rejected files or failed records (status: "completed with errors"), the CLI and Docker process must exit with code 0 so the KBase Container Task Service (CTS) preserves output artifacts and logs.
2. Summary Report (report.txt): Parses status, Metrics (files_processed, records_extracted, records_failed, files_rejected), Output Tables, and Errors from <output_dir>/<run_id>/report.txt and matches them against expected_report in rules.yaml.
3. Audit Table (_load_info): Reads the _load_info table from the dlt destination and asserts that the audit row exists for run_id.
4. Destination Tables (expected/*.json): Reads every output table from the dlt dataset via read_destination_tables(), normalizes volatile dlt metadata (_dlt_load_id, _dlt_id), strips null columns that dlt omits from materialization, and asserts row-level equality against specs/pipelines/<pipeline>/expected/<scenario>_<table>.json.