Struct TestContext 

pub struct TestContext {
    pub test_id: String,
    pub config: E2eConfig,
    pub postgres: PostgresResource,
    pub kafka: KafkaResource,
    pub clickhouse: Option<ClickHouseResource>,
    pub mysql: Option<MySqlResource>,
    pub prometheus: Option<PrometheusResource>,
    pub sqs: Option<SqsResource>,
    pub use_plugin: bool,
    pub temp_dir: TempDir,
    pub pg_database: String,
    pub kafka_topic: String,
}

Test context providing isolated resources for a single test

Fields

test_id: String

Unique identifier for this test run

config: E2eConfig

Configuration

postgres: PostgresResource

PostgreSQL resource manager

kafka: KafkaResource

Kafka resource manager

clickhouse: Option<ClickHouseResource>

ClickHouse resource manager (optional, only if clickhouse tests are needed)

mysql: Option<MySqlResource>

MySQL resource manager (optional, only if mysql tests are needed)

prometheus: Option<PrometheusResource>

Prometheus resource manager (optional, only if metrics tests are needed)

sqs: Option<SqsResource>

SQS resource manager (optional, only if SQS tests are needed)

use_plugin: bool

When true, Self::build_env_vars forwards STREAMLING__PLUGIN__PATH to the streamling subprocess if it is set in the current environment and points at an existing file.

temp_dir: TempDir

Temporary directory for pipeline files

pg_database: String

Name of the isolated PostgreSQL database

kafka_topic: String

Name of the isolated Kafka topic

Implementations

impl TestContext

pub async fn new() -> Result<Self>

Create a new test context with isolated resources

pub async fn with_options(options: TestContextOptions) -> Result<Self>

Create a new test context with custom options

pub fn pg_connection_string(&self) -> String

Get the PostgreSQL connection string for the isolated database

pub async fn run_streamling(&self, pipeline_yaml: &str) -> Result<ExitStatus>

Run streamling with the given pipeline YAML content

pub async fn run_streamling_file( &self, pipeline_path: &Path, ) -> Result<ExitStatus>

Run streamling with a pipeline file path

pub fn build_env_vars(&self) -> Vec<(String, String)>

Build environment variables for streamling execution

pub async fn create_sqs_queue(&self, queue_suffix: &str) -> Result<SqsResource>

Create an additional SQS queue for use in tests

pub async fn create_kafka_topic( &self, topic_suffix: &str, ) -> Result<KafkaResource>

Create an additional Kafka topic for use in tests (e.g., for sink output)

pub async fn consume_kafka_messages( &self, topic: &str, max_messages: usize, ) -> Result<Vec<(i64, String, String)>>

Consume messages from a Kafka topic and return them as decoded values

pub async fn run_pipeline( &self, pipeline_yaml: &str, record_limit: u64, ) -> Result<ExitStatus>

Run streamling with pipeline YAML and stop after N records

This is the most common pattern for e2e tests - run until a specific number of records have been processed.

pub async fn run_pipeline_with_opts( &self, pipeline_yaml: &str, opts: PipelineOpts, ) -> Result<ExitStatus>

Run streamling with pipeline YAML and custom options

pub async fn run_pipeline_with_capture( &self, pipeline_yaml: &str, opts: PipelineOpts, ) -> Result<PrintSinkOutput>

Run streamling with pipeline YAML and capture stdout/stderr for inspection

This is useful for tests that need to inspect the print sink output. Returns a PrintSinkOutput which can be used to verify the pipeline output.

pub async fn run_pipeline_raw( &self, pipeline_yaml: &str, opts: PipelineOpts, ) -> Result<StreamlingOutput>

Run streamling with pipeline YAML and return raw output (stdout, stderr, status)

This is useful for tests that need to inspect error messages or other raw output that doesn’t fit the PrintSinkOutput format. Unlike other run methods, this returns the output even if the pipeline fails, allowing tests to inspect error messages.