connector-testing

# Testing Apache Iggy Connectors > **Universal connector rules** live in > [connectors-overview](../connectors-overview/SKILL.md). Repo-wide testing rules (`#[iggy_harness]`, BDD naming, harness layout) live in [AGENTS.md](../../../AGENTS.md#testing). This skill covers connector-specific test conventions. ## Contents - [STOP and ask the user before](#stop-and-ask-the-user-before) - [Unit tests](#unit-tests) - [Integration tests (real infra via Docker)](#integration-tests-real-infra-via-docker) - [SDK + runtime tests](#sdk--runtime-tests) - [CI / verification flow](#ci--verification-flow) - [Common pitfalls](#common-pitfalls) - [Examples worth copying](#examples-worth-copying) ## STOP and ask the user before - Mocking the database or backend in an integration test - we hit real infra via `testcontainers-modules` (mocks miss migration bugs). - Adding a unit test that requires Docker - that belongs under `core/integration/tests/connectors/<backend>/`, not in `<plugin>/src/lib.rs`. - Skipping the four canonical source state tests - they're mandatory for every source plugin. - Naming a test `test_foo` / `does_bar` - we use BDD `given_X_when_Y_should_Z`. There are **two distinct test layers**. Use both - they catch different bugs. | Layer | Location | What it covers | When required | | --------------------- | -------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------- | ---------------------------------------------------- | | **Unit tests** | `#[cfg(test)] mod tests` at EOF of `<plugin>/src/lib.rs` | Config parsing, defaults, fallbacks, payload conversion, query/document building, state ser/de | Every plugin | | **Integration tests** | `core/integration/tests/connectors/<name>/` (uses real backend via Docker) | End-to-end: runtime + plugin + real Postgres/ES/Iceberg/Mongo/etc. + Iggy server | Every plugin that integrates with an external system | ## Unit tests ### Naming - BDD `given_when_should` Match the convention used in the codebase (`sources/random_source/src/lib.rs::tests`, `sinks/postgres_sink/src/lib.rs::tests`): ```rust #[test] fn given_persisted_state_should_restore_messages_produced() { ... } #[test] fn given_no_state_should_start_fresh() { ... } #[test] fn given_invalid_duration_should_fall_back_to_default() { ... } #[test] fn given_payload_json_should_serialize_correctly() { ... } ``` You may **omit one part** when the test is small (`given_X_should_Y` instead of full `given_X_when_Y_should_Z`), but **stay consistent within a file**. Imperative `test_foo` / `does_bar` names are not the convention here. A test whose subject is a **type** rather than a scenario has no "given" to name, and drops the prefix: it reads `<subject>_should_<property>`. The canonical state round trip below is `state_should_be_serializable_and_deserializable` in all three shipped sources (`random_source`, `postgres_source`, `http_source`). Leave those as they are rather than bending them into the scenario form. ### Prove a test can fail, before trusting it A passing test says nothing until you have watched it fail. Break the behaviour it names, run the suite, and put the code back: ```bash # commit first: this reverts with `git checkout --`, which discards # every uncommitted change in the file, not only the mutation git status --porcelain -- <file> # must be empty <edit the guard out> cargo nextest run -p <crate> # expect a failure naming that test git checkout -- <file> ``` Two failures this catches, both seen in review of `http_source`: - **A fixture that makes the interesting case unreachable.** A guard test for endpoint resurrection passed `&[]` as the static config, so there was nothing to resurrect and the assertion held whatever `restore` did. - **An assertion aimed at something self-healing.** A test for a teardown guard asserted on gauges, which `Metrics::encode` rebuilds from the live instances on every scrape, so it passed with the guard deleted. Delete such a test rather than keep it; a test that cannot fail is worse than none, because it reads as coverage. Watch for an arm hidden behind a catch-all: `StatusClass::from` ends in `_ => ServerError`, so the redirect arm could be removed with every existing assertion still green. ### `test_config()` helper Every plugin's tests start with a small helper returning a tuned-down version of the production config. New tests extend it. don't construct from scratch each time. ```rust fn test_config() -> RandomSourceConfig { RandomSourceConfig { interval: Some("100ms".to_string()), max_count: Some(100), messages_range: Some((5, 10)), payload_size: Some(50), } } ``` ### Async testing pattern The trait methods are `async`. For unit tests, build a runtime locally - don't add `#[tokio::test]` in `src/lib.rs` (it pulls test macros into prod build): ```rust #[test] fn given_X_should_Y() { let runtime = tokio::runtime::Runtime::new().unwrap(); runtime.block_on(async { // await calls here }); } ``` ### Mandatory: four canonical source state tests Every source must have these. Copy from `sources/random_source/src/lib.rs::tests`: ```rust #[test] fn given_persisted_state_should_restore_<field>() { let state = State { /* fields */ }; let serialized = rmp_serde::to_vec(&state).expect("Failed to serialize state"); let connector_state = ConnectorState(serialized); let source = MySource::new(1, test_config(), Some(connector_state)); let runtime = tokio::runtime::Runtime::new().unwrap(); runtime.block_on(async { let restored = source.state.lock().await; assert_eq!(restored.<field>, /* expected */); }); } #[test] fn given_no_state_should_start_fresh() { let source = MySource::new(1, test_config(), None); /* assert default state */ } #[test] fn given_invalid_state_should_start_fresh() { let invalid = ConnectorState(b"not valid msgpack".to_vec()); let source = MySource::new(1, test_config(), Some(invalid)); /* assert default state */ } #[test] fn state_should_be_serializable_and_deserializable() { let original = State { /* fields */ }; let bytes = rmp_serde::to_vec(&original).unwrap(); let restored: State = rmp_serde::from_slice(&bytes).unwrap(); assert_eq!(original.<field>, restored.<field>); } ``` State is the only thing that survives a plugin restart. Silent corruption here means lost data on the next deploy. **The third test has two correct shapes, and which one applies is a design decision rather than a style choice.** Starting fresh is right when the state is a cursor. The worst case is re-reading from the beginning, which the at-least-once contract already permits, so discarding an undecodable cursor costs duplicates and nothing else. It is wrong when the state carries something whose *absence* is a downgrade rather than a repeat. `http_source` keeps its revocation tombstones in state, so starting fresh after a decode failure would serve every revoked endpoint again, with its secret. Its `restore` returns `Err` instead and the connector refuses to open. A source in that position asserts the refusal: ```rust #[test] fn given_invalid_state_should_refuse_to_open() { let invalid = ConnectorState(b"not valid msgpack".to_vec()); let restored = MyState::restore(&config_entries(), Some(invalid), 1); assert!( restored.is_err(), "discarding this state silently would re-serve what it was revoking" ); } ``` Name the test for whichever it is and say why in a comment. What is never right is starting fresh silently when the state that was discarded was load-bearing. ### Sink unit tests Cover pure per-message logic: - Config defaults + optional fields + invalid values → fallback. - Payload conversion for each `Payload` variant the sink claims to handle. - Header encoding (especially base64 for binary headers). - Query/document building (e.g., `postgres_sink::build_batch_insert_query`). - Transient vs permanent error classification (e.g., `postgres_sink::is_transient_error`). ### Transform tests Four branches must be covered (see `connector-transform`): 1. Happy path: matching input → expected output. 2. Filter path: input that should be dropped returns `Ok(None)`. 3. Pass-through path: non-JSON payload (or unmatched condition) returns `Ok(Some(unchanged))`. 4. Invalid config: `new()` returns `Err(Error::InvalidConfigValue(_))`. ### What NOT to mock - The underlying client lib (sqlx, reqwest, mongodb). Tests would exercise the mock, not the code. - The Iggy server. - The runtime. If you need a fake backend, use `wiremock` (a real local HTTP server) - see `core/integration/tests/connectors/http/`. ## Integration tests (real infra via Docker) The canonical pattern lives in `core/integration/tests/connectors/`. Plugins that integrate with external services (Postgres, Elasticsearch, Iceberg, MongoDB, InfluxDB, Quickwit, HTTP/wiremock) have a paired fixture + test file there. ### Layout ```text core/integration/tests/connectors/ ├── mod.rs Common helpers (TestMessage, create_test_messages) ├── fixtures/ One subdir per backend │ ├── postgres/ │ │ ├── container.rs PostgresContainer (testcontainers wrapper), env-var consts │ │ ├── sink.rs PostgresSinkFixture, PostgresSinkByteaFixture, PostgresSinkJsonFixture │ │ ├── source.rs PostgresSourceJsonFixture, PostgresSourceJsonbFixture, ... │ │ └── mod.rs re-exports │ ├── elasticsearch/ iceberg/ mongodb/ influxdb/ quickwit/ http/ delta/ │ └── wiremock.rs HTTP mocking for http_sink tests ├── postgres/ │ ├── postgres_sink.rs The actual #[iggy_harness] tests │ ├── postgres_source.rs │ ├── restart.rs Restart + state-survival tests │ ├── sink.toml Runtime config used by sink tests │ ├── source.toml Runtime config used by source tests │ └── mod.rs Shared constants (TEST_MESSAGE_COUNT, POLL_ATTEMPTS, ...) ├── delta/ elasticsearch/ iceberg/ mongodb/ influxdb/ random/ quickwit/ http/ stdout/ ├── http_config_provider/ HTTP config provider end-to-end ├── runtime/ Runtime-only tests (error isolation, missing plugin, ...) └── api/ HTTP control plane tests ``` ### Container naming + cleanup Every test container MUST be named with the `iggy-test-` prefix so a single `docker ps -aqf 'name=^iggy-test-'` sweep reaps them (`just clean-test-containers`, alias `ctc`; the `test`/`nextest` recipes also reap on exit). New fixtures get this for free by naming through `fixtures::unique_container_name("<svc>")`, which appends a uuid for parallel-safe ephemeral containers. Reuse fixtures use a fixed `iggy-test-<svc>` literal instead, because the stable name is what lets later nextest processes attach to the same container via `ReuseDirective::Always`. Do not hand-roll a different prefix - the sweep only finds `iggy-test-*`. As of now only `elasticsearch` and `doris` share a container. Both have a slow boot (JVM / Doris FE+BE, tens of seconds), so reusing one across the whole regression beats spinning a fresh one per test. Every other fixture boots a fresh, ephemeral container per test - cheap enough that the isolation is worth more than the reuse. ### The `#[iggy_harness]` proc macro Each integration test is annotated with this macro from the `integration` crate. It boots an in-process Iggy server (and optionally the connectors runtime) for the duration of the test, runs the seeds, then injects a `&TestHarness` and your fixture. ```rust use integration::harness::seeds; use integration::iggy_harness; #[iggy_harness( server(connectors_runtime(config_path = "tests/connectors/postgres/sink.toml")), seed = seeds::connector_stream )] async fn json_messages_sink_stores_as_bytea( harness: &TestHarness, fixture: PostgresSinkFixture, ) { let client = harness.root_client().await.unwrap(); let pool = fixture.create_pool().await.expect("Failed to create pool"); fixture.wait_for_table(&pool, "iggy_messages").await; // ... send messages, assert on rows } ``` Key pieces: - `server(connectors_runtime(config_path = ...))` boots the connectors runtime against the given TOML. - `seed = seeds::connector_stream` runs the named seed to create the test stream/topic. - `fixture: PostgresSinkFixture` - the test harness calls `PostgresSinkFixture::setup()` (which spins up the testcontainer) and injects the result.