tests.platform package

Submodules

tests.platform.conftest module

pytest configuration for platform tests.

Provides shared fixtures for OIDC authentication caching across pytest workers. Uses pytest-xdist hooks to ensure OIDC token fetching happens only once on the master node.

tests.platform.conftest.is_master(config) → bool

True if the code running is in the xdist master node or not using xdist at all.

tests.platform.conftest.oidc_token_fixture(request)

Session-scoped fixture that verifies OIDC token setup.

The actual token fetching is done by pytest_configure hooks and stored in environment variables for cross-process access.

tests.platform.conftest.pytest_configure(config)

Configure hook: fetch OIDC token on master node only.

tests.platform.conftest.pytest_configure_node(node)

xdist hook: pass token data to worker nodes via workerinput for fixture access.

tests.platform.helper module

Helper utilities for platform tests.

Provides:

• Lightweight REST wrappers (no SDK abstraction where raw status codes matter)

• Polling helpers (compilation, generic condition)

• Simple selector/object helpers

No automatic cleanup; pipelines are left in place for inspection after failures.

class tests.platform.helper.HTTPStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: IntEnum

HTTP status codes and reason phrases

Status codes from the following RFCs are all observed:

• RFC 7231: Hypertext Transfer Protocol (HTTP/1.1), obsoletes 2616

• RFC 6585: Additional HTTP Status Codes

• RFC 3229: Delta encoding in HTTP

• RFC 4918: HTTP Extensions for WebDAV, obsoletes 2518

• RFC 5842: Binding Extensions to WebDAV

• RFC 7238: Permanent Redirect

• RFC 2295: Transparent Content Negotiation in HTTP

• RFC 2774: An HTTP Extension Framework

• RFC 7725: An HTTP Status Code to Report Legal Obstacles

• RFC 7540: Hypertext Transfer Protocol Version 2 (HTTP/2)

• RFC 2324: Hyper Text Coffee Pot Control Protocol (HTCPCP/1.0)

• RFC 8297: An HTTP Status Code for Indicating Hints

• RFC 8470: Using Early Data in HTTP

ACCEPTED = 202

ALREADY_REPORTED = 208

BAD_GATEWAY = 502

BAD_REQUEST = 400

CONFLICT = 409

CONTINUE = 100

CREATED = 201

EARLY_HINTS = 103

EXPECTATION_FAILED = 417

FAILED_DEPENDENCY = 424

FORBIDDEN = 403

FOUND = 302

GATEWAY_TIMEOUT = 504

GONE = 410

HTTP_VERSION_NOT_SUPPORTED = 505

IM_A_TEAPOT = 418

IM_USED = 226

INSUFFICIENT_STORAGE = 507

INTERNAL_SERVER_ERROR = 500

LENGTH_REQUIRED = 411

LOCKED = 423

LOOP_DETECTED = 508

METHOD_NOT_ALLOWED = 405

MISDIRECTED_REQUEST = 421

MOVED_PERMANENTLY = 301

MULTIPLE_CHOICES = 300

MULTI_STATUS = 207

NETWORK_AUTHENTICATION_REQUIRED = 511

NON_AUTHORITATIVE_INFORMATION = 203

NOT_ACCEPTABLE = 406

NOT_EXTENDED = 510

NOT_FOUND = 404

NOT_IMPLEMENTED = 501

NOT_MODIFIED = 304

NO_CONTENT = 204

OK = 200

PARTIAL_CONTENT = 206

PAYMENT_REQUIRED = 402

PERMANENT_REDIRECT = 308

PRECONDITION_FAILED = 412

PRECONDITION_REQUIRED = 428

PROCESSING = 102

PROXY_AUTHENTICATION_REQUIRED = 407

REQUESTED_RANGE_NOT_SATISFIABLE = 416

REQUEST_ENTITY_TOO_LARGE = 413

REQUEST_HEADER_FIELDS_TOO_LARGE = 431

REQUEST_TIMEOUT = 408

REQUEST_URI_TOO_LONG = 414

RESET_CONTENT = 205

SEE_OTHER = 303

SERVICE_UNAVAILABLE = 503

SWITCHING_PROTOCOLS = 101

TEMPORARY_REDIRECT = 307

TOO_EARLY = 425

TOO_MANY_REQUESTS = 429

UNAUTHORIZED = 401

UNAVAILABLE_FOR_LEGAL_REASONS = 451

UNPROCESSABLE_ENTITY = 422

UNSUPPORTED_MEDIA_TYPE = 415

UPGRADE_REQUIRED = 426

USE_PROXY = 305

VARIANT_ALSO_NEGOTIATES = 506

property is_client_error

property is_informational

property is_redirection

property is_server_error

property is_success

tests.platform.helper.connector_action(pipeline: str, table: str, connector: str, action: str)

tests.platform.helper.delete(path: str, **kw) → Response

tests.platform.helper.extract_object_by_name(collection: Iterable[Dict[str, Any]], name: str) → Dict[str, Any]

tests.platform.helper.gen_pipeline_name(func)

Decorator for pytest functions that automatically generates a unique pipeline name. The decorated function will receive a ‘pipeline_name’ parameter.

After the test completes, attempts to delete the pipeline but ignores any errors.

tests.platform.helper.get(path: str, **kw) → Response

tests.platform.helper.http_request(method: str, path: str, **kwargs) → Response

Low-level request wrapper (no retries). Raises only on network errors.

Args:

method: HTTP method (GET, POST, etc.) path: URL path base_headers: Optional override for base headers. If None (default), uses _base_headers().

If empty dict {}, no base headers are applied (for testing unauthenticated requests).

**kwargs: Additional arguments passed to requests.request()

tests.platform.helper.patch_json(path: str, body: Dict[str, Any], **kw) → Response

tests.platform.helper.post_json(path: str, body: Dict[str, Any], **kw) → Response

tests.platform.helper.put_json(path: str, body: Dict[str, Any], **kw) → Response

tests.platform.helper.unique_pipeline_name(base_name: str) → str

In CI, multiple tests of different runs can run against the same Feldera instance, we make sure the pipeline names they use are unique by appending the first 5 characters of the commit SHA or ‘local’ if not in CI.

tests.platform.helper.wait_for_condition(description: str, predicate, timeout_s: float = 30.0, sleep_s: float = 0.2) → None

Generic polling helper. predicate: callable returning truthy when condition met (can be sync or async).

tests.platform.helper.wait_for_program_success(pipeline_name: str, expected_program_version: int, timeout_s: float = 1800.0, sleep_s: float = 0.5) → WaitResult

Poll until the pipeline’s program_status is Success and program_version >= expected_program_version.

Mirrors semantics of the Rust wait_for_compiled_program helper.

Returns a WaitResult. Raises AssertionError on compile error, TimeoutError on timeout.

tests.platform.negative_test module

class tests.platform.negative_test.NegativeCompilationTests(methodName='runTest')

Bases: TestCase

test_errors0()

test_initialization_error()

test_program_error0()

test_program_error1()

test_rust_error()

test_sql_error()

tests.platform.test_bootstrapping module

tests.platform.test_checkpoint_suspend module

tests.platform.test_checkpoint_sync module

tests.platform.test_cluster_health module

tests.platform.test_cluster_health.test_cluster_health_check()

Poll the cluster health endpoint until both runner and compiler are healthy.

• Repeatedly query the cluster health endpoint.

• If both runner and compiler report healthy=true, expect HTTP 200 and stop.

• Otherwise expect HTTP 503 while waiting.

• Timeout after 300 seconds.

tests.platform.test_cluster_health.test_health_check()

Poll the /healthz endpoint until it reports overall healthy or timeouts.

Success condition:

status code 200 and body == {“status”: “healthy”}

Acceptable transient condition (database not ready yet):

status code 500 and body == {

“status”: “unhealthy: unable to reach database (see logs for further details)”

}

tests.platform.test_completion_tokens module

tests.platform.test_completion_tokens.test_completion_tokens(pipeline_name)

• Pipeline without output connectors

• Ingest many single-row JSON events, each returning a completion token

• Poll completion_status for each token

• Validate the row becomes visible exactly once

tests.platform.test_completion_tokens.test_completion_tokens_with_outputs(pipeline_name)

• Pipeline with multiple file_output connectors on materialized views.

• Ingest multiple records, verify completion tokens, and validate counts.

• Start a paused datagen input connector, obtain a completion token through its endpoint, wait for completion, and validate resulting counts.

tests.platform.test_ingress_formats module

class tests.platform.test_ingress_formats.JsonLineReader(resp)

Bases: object

read_events(n, timeout_s=10.0)

tests.platform.test_ingress_formats.test_case_sensitive_tables(pipeline_name)

• Distinguish between quoted and unquoted identifiers.

• Validate streaming outputs for two views.

tests.platform.test_ingress_formats.test_duplicate_outputs(pipeline_name)

multiple inserts producing duplicate output values.

tests.platform.test_ingress_formats.test_json_ingress(pipeline_name)

Exercise raw inserts, insert_delete format, array format, parse errors, debezium update, and CSV ingestion with parse error.

tests.platform.test_ingress_formats.test_map_column(pipeline_name)

Table with column of type MAP

tests.platform.test_ingress_formats.test_parse_datetime(pipeline_name)

tests.platform.test_ingress_formats.test_primary_keys(pipeline_name)

Port of primary_keys: test insert/update/delete semantics with primary key.

tests.platform.test_ingress_formats.test_quoted_columns(pipeline_name)

tests.platform.test_ingress_formats.test_upsert(pipeline_name)

• Insert several rows with composite PK.

• Perform updates/inserts overwriting existing rows.

• Perform no-op updates and deletes of non-existing keys.

tests.platform.test_metrics_logs module

tests.platform.test_metrics_logs.test_pipeline_logs(pipeline_name)

• Logs 404 before pipeline creation.

• Create pipeline; poll until logs return 200.

• Pause / start / stop / clear transitions keep logs accessible (200).

• After delete, logs eventually return 404 again.

tests.platform.test_metrics_logs.test_pipeline_metrics(pipeline_name)

Tests that circuit metrics can be retrieved from the pipeline.

tests.platform.test_metrics_logs.test_pipeline_stats(pipeline_name)

Tests retrieving pipeline statistics via /stats.

tests.platform.test_nowstream module

class tests.platform.test_nowstream.TestNowStream(methodName='runTest')

Bases: TestCase

test_nowstream()

Test the now() function: pipeline should produce outputs even if no new inputs are supplied.

tests.platform.test_nowstream.get_result(pipeline) → str

tests.platform.test_oidc_auth_simple module

Simplified OIDC Authentication Integration Tests

Tests OIDC authentication using Resource Owner Password Flow only.

Environment Variables Required: - OIDC_TEST_ISSUER: The issuer URL for the OIDC provider - OIDC_TEST_CLIENT_ID: Client ID for the test application - OIDC_TEST_CLIENT_SECRET: Client secret for Resource Owner Password Flow - OIDC_TEST_USERNAME: Username for test user - OIDC_TEST_PASSWORD: Password for test user

tests.platform.test_oidc_auth_simple.test_authenticated_endpoint_access()

Test that valid token allows access to protected endpoints

tests.platform.test_oidc_auth_simple.test_authentication_config_discovery()

Test that the authentication configuration endpoint returns GenericOidc configuration

tests.platform.test_oidc_auth_simple.test_malformed_token_rejection()

Test that malformed JWT tokens are rejected

tests.platform.test_oidc_auth_simple.test_obtain_access_token()

Test obtaining access token via Resource Owner Password Flow

tests.platform.test_oidc_auth_simple.test_oidc_discovery_endpoint()

Test that the remote OIDC provider’s discovery endpoint is accessible

tests.platform.test_oidc_auth_simple.test_unauthenticated_request_rejection()

Test that requests without authentication are rejected

tests.platform.test_orchestration module

tests.platform.test_orchestration.test_pipeline_orchestration_basic(pipeline_name)

Tests the orchestration of the pipeline, which means the starting and pausing of the pipeline itself as well as its connectors individually. This tests the basic processing of data and handling of case sensitivity and special characters.

tests.platform.test_orchestration.test_pipeline_orchestration_errors(pipeline_name)

Port of Rust pipeline_orchestration_errors: - Validate return codes for valid/invalid pipeline & connector actions.

tests.platform.test_orchestration.test_pipeline_orchestration_scenarios(pipeline_name)

Tests for orchestration that the effects (i.e., pipeline and connector state) are indeed as expected after each scenario consisting of various start and pause steps.

tests.platform.test_pipeline_builder module

class tests.platform.test_pipeline_builder.TestPipelineBuilder(methodName='runTest')

Bases: TestCase

test_connector_orchestration()

tests.platform.test_pipeline_configs module

tests.platform.test_pipeline_configs.test_pipeline_program_config(pipeline_name)

Tests that the pipeline program configuration is validated and stored correctly, and that patching works on the field as whole.

tests.platform.test_pipeline_configs.test_pipeline_runtime_config(pipeline_name)

Tests that the pipeline runtime configuration is validated and stored correctly, and that patching works on the field as whole.

tests.platform.test_pipeline_crud module

tests.platform.test_pipeline_crud.test_pipeline_connector_endpoint_naming(pipeline_name)

tests.platform.test_pipeline_crud.test_pipeline_create_compile_delete(pipeline_name)

tests.platform.test_pipeline_crud.test_pipeline_get(pipeline_name)

tests.platform.test_pipeline_crud.test_pipeline_get_selector(pipeline_name)

tests.platform.test_pipeline_crud.test_pipeline_name_conflict(pipeline_name)

tests.platform.test_pipeline_crud.test_pipeline_name_invalid()

tests.platform.test_pipeline_crud.test_pipeline_post(pipeline_name)

tests.platform.test_pipeline_crud.test_refresh_version(pipeline_name)

tests.platform.test_pipeline_lifecycle module

tests.platform.test_shared_pipeline module

tests.platform.test_shared_pipeline_stress module

class tests.platform.test_shared_pipeline_stress.TestPipeline(methodName='runTest')

Bases: SharedTestPipeline

test_create_pipeline()

CREATE TABLE tbl(id INT) WITH (‘materialized’ = ‘true’); CREATE MATERIALIZED VIEW v0 AS SELECT * FROM tbl;

test_get_pipeline_logs()

tests.platform.test_update_runtime module

class tests.platform.test_update_runtime.TestPipeline(methodName='runTest')

Bases: TestCase

test_update_runtime()

Test the /update_runtime API.

Pipeline’s platform version should only be updated to the current version on-demand (by calling /update_runtime) or when updating the pipeline code.

Module contents

Platform tests package.

This file makes the tests.platform directory a proper Python package so that relative imports like from .helper import … work when running pytest with PYTHONPATH pointing at the repository python directory.