faucet-conformance

A reusable connector conformance test battery for the faucet-stream ecosystem. Any faucet-source-* / faucet-sink-* crate calls it from its own tests/ to prove it upholds the connector contract.

Passing this battery is the Tier-1 (supported) criterion for a connector. There is no separate tiering scheme — a connector is "supported" exactly when it invokes and passes these checks in CI; connectors that don't are Tier-2 (experimental).

Checks

Every check ships with both a passing and a #[should_panic] failing test in this crate — a check that cannot fail is worthless.

Declared-capability checks (RFC 0013)

Each proves a sink behaves as a capability it declares says it does:

Checks 20 and 21 are required for a sink built on a family driver (faucet_core::driver, faucet_common_sql::SqlDriver); they run on any sink.

Which checks a connector must run

faucet_conformance::required::REQUIREMENTS maps each declared capability (a descriptor flag, a write mode, a trait-method override, a buffering flush, a declared BatchAtomicity) to the assertions that prove it. For the built-in connectors scripts/conformance-coverage.py (CI Doc counts job) enforces the table from the committed declared-capabilities.json snapshot, each crate's src/ and its tests/conformance.rs (or crates/interop-tests/tests/<kind>_conformance.rs). Checks that cannot run in CI are listed with a reason in conformance-coverage-exempt.txt (call-deadline ones in call-deadline-exempt.txt). Regenerate the snapshot after changing a connector's declared capabilities:

cargo test -p faucet-cli --features full --test declared_capabilities -- --ignored regenerate

A test driving a cloud client whose futures are deeper than the default test stack runs its body through block_on_large_stack.

Descriptor checks the built-in connectors cannot run offline

Every built-in connector calls check 14 from its tests/conformance.rs for sample configs covering each config-dependent capability. These connectors open a connection inside new(), so they cannot be built without a backend; their src/descriptor.rs unit tests pin every capability for every config branch instead (and faucet-cli's registry_capability_parity test probes them whenever they do build):

The Spanner source and sink run check 14 against the emulator container, like their other conformance checks.

Integration-level checks

These drive a live backend or the real pipeline, so they belong in a connector's testcontainers/tempfile conformance test rather than against the synthetic doubles:

Usage

use faucet_conformance as conf;

#[test]
fn config_schema_is_valid() {
    let source = MySource::new(/* … */);
    conf::assert_config_schema_valid(&source);
}

#[tokio::test]
async fn streams_with_bounded_memory() {
    let source = MySource::over_n_rows(10_000);
    conf::assert_bounded_memory(&source, 500, 10_000).await;
}

Sinks use the value form of check 1: conf::assert_config_schema_valid_value(&sink.config_schema(), sink.connector_name()).

Value fidelity

fidelity holds the corpus of values that have broken a pipeline somewhere — integers past 2^53, -0.0, 1e±300, quote/backslash/padding strings, offset and pre-epoch timestamps, nested and empty containers, and the canonical renderings of faucet_core::fidelity (exact decimals, big integers, base64) — plus assert_round_trip / diff_round_trip and the one-sided helpers assert_corpus_read (a source over a seeded backend) and assert_corpus_written (a sink, read back). Every built-in connector runs it (scripts/fidelity-coverage.py); a Tolerance names what a destination genuinely cannot hold.

Reliability suites

tests/reliability_*.rs run the real Pipeline::run against scripted doubles to pin the engine's guarantees — bookmark ordering, exactly-once, lifecycle, cleanup safety, the DLQ boundary, panic isolation, state migration and (reliability_governance) the governed DLQ, row accounting and per-row policy quarantine on every write path (RFC 0013). See the reliability testing guide.

Forwarding harnesses (for wrappers)

A wrapper (a source or sink that holds another one) must forward every defaulted hook, or the feature behind it turns off for everything it wraps. Write wrappers with faucet_core::forward_source! / forward_sink!, then prove the result with faucet_conformance::forwarding:

use faucet_conformance::forwarding::{HookSink, assert_sink_forwards_every_hook};

#[tokio::test]
async fn my_wrapper_forwards_every_hook() {
    let probe = HookSink::default();
    let wrapped = MyWrapper::new(Box::new(probe.clone()));
    // List the hooks the wrapper deliberately does not forward.
    assert_sink_forwards_every_hook(&wrapped, &probe, &["load_native"]).await;
}

HookSource / HookSink answer every hook differently from the trait default and record which hooks were reached; assert_source_forwards_every_hook / assert_sink_forwards_every_hook call every hook through the wrapper and name the ones that never reached the inner connector. A name in the skip list that is not a hook fails the assertion, so the list cannot go stale. SOURCE_HOOKS and SINK_HOOKS are the canonical hook lists; a test in this crate checks them against the trait definitions, so a hook added to either trait without being added to the harness fails CI.

Call deadlines and cancellation

A connector that runs its backend calls through faucet_core::io::CallContext (#880) proves it with faucet_conformance::calls: point the connector at a backend that never answers (a wiremock server with a long set_delay) and call assert_source_call_deadline(|| make_source()) or assert_sink_call_deadline(|| make_sink(), &records). Each installs a context with a short deadline, then one with a cancel token, and fails if the read or write does not end within CALL_BOUND.

faucet_conformance::hung::StallingBodyServer answers HTTP requests through a route closure and can stop a response body mid-stream (Reply::stall_body): the same assertions then prove the connector reads the body under the context too (CallContext::reader), not only the request.

Doubles

faucet_conformance::doubles provides a CountingSource (lazily emits N synthetic records in pages — genuinely bounded) and a TestSink (append or keyed/upsert recording sink) for use in your own tests.

Starting test containers

With the containers feature, faucet_conformance::containers starts a testcontainers backend the same way in every integration test: a 180 s start-up budget, up to three attempts for start-up failures only (logged, with backoff, the failed container dropped first), an optional readiness probe (TCP, HTTP or a custom check) and the FAUCET_REQUIRE_BACKENDS rule — skip locally, fail where the backends are required.

[dev-dependencies]
faucet-conformance = { version = "1", features = ["containers"] }

use faucet_conformance::containers::{self, ReadyProbe, StartOptions};

let opts = StartOptions::default().ready(ReadyProbe::tcp(6379));
let Some(redis) = containers::start_or_skip(Redis::default, &opts).await else {
    return;
};

License

Licensed under either of Apache-2.0 or MIT at your option.