From f6b375ac785add72e116aae880d8f3901fbe3202 Mon Sep 17 00:00:00 2001 From: sdairs Date: Mon, 24 Aug 2026 16:52:55 +0100 Subject: [PATCH 1/3] Exercise PostgreSQL ClickPipe creation through CLI --- Cargo.lock | 1 + crates/clickhouse-cloud-api/Cargo.toml | 1 + .../tests/clickpipes/postgres_cdc_test.rs | 175 +++++++++++------- 3 files changed, 115 insertions(+), 62 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2c5b3205..3596f81e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -985,6 +985,7 @@ dependencies = [ "rustls-pemfile", "serde", "serde_json", + "tempfile", "thiserror", "tokio", "tokio-postgres", diff --git a/crates/clickhouse-cloud-api/Cargo.toml b/crates/clickhouse-cloud-api/Cargo.toml index 92faacdc..6995d5ba 100644 --- a/crates/clickhouse-cloud-api/Cargo.toml +++ b/crates/clickhouse-cloud-api/Cargo.toml @@ -45,6 +45,7 @@ rcgen = "0.14.8" rustls = { version = "0.23.40", default-features = false, features = ["ring", "std", "tls12"] } rustls-native-certs = "0.8.3" rustls-pemfile = "2.2.0" +tempfile = "3.27.0" tokio = { version = "1", features = ["macros", "rt-multi-thread"] } tokio-postgres = "0.7.17" tokio-postgres-rustls = "0.13.0" diff --git a/crates/clickhouse-cloud-api/tests/clickpipes/postgres_cdc_test.rs b/crates/clickhouse-cloud-api/tests/clickpipes/postgres_cdc_test.rs index 82f5018e..c09f4753 100644 --- a/crates/clickhouse-cloud-api/tests/clickpipes/postgres_cdc_test.rs +++ b/crates/clickhouse-cloud-api/tests/clickpipes/postgres_cdc_test.rs @@ -1,7 +1,16 @@ +//! Managed-Postgres CDC E2E. Resource setup, state polling, and cleanup use the +//! API client, while both ClickPipe creation attempts run the built CLI binary. +//! +//! The managed Postgres fixture has a private CA. Mandatory CI has no +//! publicly-trusted PostgreSQL fixture, so successful creation without an +//! uploaded CA remains a residual gap. + #[path = "../common/mod.rs"] mod common; mod support; +use std::path::PathBuf; +use std::process::Command; use std::str::FromStr; use std::time::Duration; @@ -19,16 +28,24 @@ const POST_SEED_ROW_COUNT: i64 = 8; const DEFAULT_CLICKPIPE_READY_TIMEOUT_SECS: u64 = 600; const DEFAULT_CDC_LAG_TIMEOUT_SECS: u64 = 300; +const CLICKHOUSECTL_BINARY_ENV: &str = "CLICKHOUSECTL_TEST_BINARY"; +const PRIVATE_CA_API_ERROR: &str = "x509: certificate signed by unknown authority"; +const PRIVATE_CA_HINT: &str = + "Hint: Provide the PostgreSQL source's PEM CA bundle with --ca-certificate ."; #[tokio::test] #[ignore = "requires live ClickHouse Cloud credentials and provisions real resources"] -async fn cloud_clickpipe_postgres_cdc() -> TestResult<()> { +async fn cloud_clickpipe_postgres_cli_cdc() -> TestResult<()> { // rustls 0.23 requires a default CryptoProvider be installed before any // ClientConfig is constructed. install_default returns an Err if one was // already set by another test in the same process, which is harmless here. let _ = rustls::crypto::ring::default_provider().install_default(); let ctx = TestContext::from_env()?; + let clickhousectl = clickhousectl_binary()?; + let cli_workspace = tempfile::tempdir()?; + let cli_home = cli_workspace.path().join("home"); + std::fs::create_dir(&cli_home)?; let clickpipe_ready_timeout = duration_from_env_or( "CLICKHOUSE_CLOUD_TEST_TIMEOUT_CLICKPIPE_READY_SECS", DEFAULT_CLICKPIPE_READY_TIMEOUT_SECS, @@ -42,7 +59,7 @@ async fn cloud_clickpipe_postgres_cdc() -> TestResult<()> { let mut cleanup = CleanupRegistry::default(); let test_result = async { - log_run_header("cloud_clickpipe_postgres_cdc", &ctx); + log_run_header("cloud_clickpipe_postgres_cli_cdc", &ctx); let mut failures = FailureRecorder::default(); // ── Preflight ─────────────────────────────────────────────── @@ -252,81 +269,98 @@ async fn cloud_clickpipe_postgres_cdc() -> TestResult<()> { let pg_ca_pem = client .postgres_service_certs_get(&ctx.org_id, &postgres_id) - .await - .ok(); + .await?; + if pg_ca_pem.trim().is_empty() { + return Err("postgres CA endpoint returned an empty bundle".into()); + } + let pg_ca_path = cli_workspace.path().join("postgres-ca.pem"); + std::fs::write(&pg_ca_path, pg_ca_pem.as_bytes())?; let pg_client = connect_postgres( &pg_hostname, pg_port, &pg_database, &pg_username, &pg_password, - pg_ca_pem.as_deref(), + Some(&pg_ca_pem), ) .await?; configure_pg_for_cdc(&pg_client).await?; - // ── Create ClickPipe ──────────────────────────────────────── + // ── Create ClickPipe through the real CLI ─────────────────── log_phase("Create ClickPipe"); - let pipe_request = ClickPipePostRequest { - name: format!("cdc-{}", ctx.run_id), - destination: ClickPipeMutateDestination { - // Postgres is a "database pipe" — only `database` is valid at - // the top level. The per-mapping `targetTable` carries the - // destination table name. - database: "default".to_string(), - ..Default::default() - }, - source: ClickPipePostSource { - postgres: Some(ClickPipeMutatePostgresSource { - authentication: ClickPipeMutatePostgresSourceAuthentication::Basic, - ca_certificate: pg_ca_pem.clone(), - credentials: PLAIN { - username: pg_username.clone(), - password: pg_password.clone(), - }, - database: pg_database.clone(), - host: pg_hostname.clone(), - port: pg_port as i64, - settings: ClickPipePostgresPipeSettings { - // `cdc` does snapshot + ongoing replication. The API - // manages the replication slot itself in this mode and - // rejects an explicit replicationSlotName. - replication_mode: ClickPipePostgresPipeSettingsReplicationmode::Cdc, - publication_name: Some(PUBLICATION.to_string()), - // The API rejects 0 for numeric fields ("Value must be >= 1"); - // the Default impl gives every i64 a 0, so we set sensible - // values explicitly. (Same issue exists in the CLI's - // clickpipe create postgres handler — track separately.) - sync_interval_seconds: Some(60), - pull_batch_size: Some(100_000), - initial_load_parallelism: Some(4), - snapshot_num_rows_per_partition: Some(100_000), - snapshot_number_of_parallel_tables: Some(4), - ..Default::default() - }, - table_mappings: vec![ClickPipePostgresPipeTableMapping { - source_schema_name: SOURCE_SCHEMA.to_string(), - source_table: SOURCE_TABLE.to_string(), - target_table: TARGET_TABLE.to_string(), - table_engine: - ClickPipePostgresPipeTableMappingTableengine::ReplacingMergeTree, - ..Default::default() - }], - r#type: Some(ClickPipeMutatePostgresSourceType::Postgres), - ..Default::default() - }), - ..Default::default() - }, - ..Default::default() + let run_cli_create = |name: &str, + ca_path: Option<&std::path::Path>| + -> TestResult { + let mut command = Command::new(&clickhousectl); + command + .env("DO_NOT_TRACK", "1") + .env("HOME", &cli_home) + .current_dir(cli_workspace.path()) + .arg("cloud"); + if let Some(base_url) = std::env::var("CLICKHOUSE_CLOUD_API_BASE_URL") + .ok() + .filter(|value| !value.is_empty()) + { + command.args(["--url", &base_url]); + } + command + .arg("--json") + .args(["clickpipe", "create", "postgres"]) + .arg(&clickhouse_id) + .args(["--name", name]) + .args(["--host", &pg_hostname]) + .arg("--port") + .arg(pg_port.to_string()) + .args(["--pg-database", &pg_database]) + .args(["--username", &pg_username]) + .args(["--password", &pg_password]) + .args([ + "--table-mapping", + &format!("{SOURCE_SCHEMA}.{SOURCE_TABLE}:{TARGET_TABLE}"), + ]) + .args(["--publication-name", PUBLICATION]) + .args(["--org-id", &ctx.org_id]); + if let Some(path) = ca_path { + command.arg("--ca-certificate").arg(path); + } + Ok(command.output()?) }; - let pipe = client - .click_pipe_create(&ctx.org_id, &clickhouse_id, &pipe_request) - .await? - .result - .ok_or("clickpipe create returned no result")?; + let without_ca = run_cli_create(&format!("private-ca-check-{}", ctx.run_id), None)?; + if without_ca.status.code() != Some(1) { + return Err(format!( + "clickhousectl without --ca-certificate exited {:?}, expected 1", + without_ca.status.code() + ) + .into()); + } + let without_ca_stderr = String::from_utf8_lossy(&without_ca.stderr); + if !without_ca_stderr.contains(PRIVATE_CA_API_ERROR) { + return Err(format!( + "private-CA failure did not preserve API detail `{PRIVATE_CA_API_ERROR}`: {without_ca_stderr}" + ) + .into()); + } + if !without_ca_stderr.contains(PRIVATE_CA_HINT) { + return Err(format!( + "private-CA failure did not include the CA flag guidance: {without_ca_stderr}" + ) + .into()); + } + + let with_ca = run_cli_create(&format!("cdc-{}", ctx.run_id), Some(&pg_ca_path))?; + if !with_ca.status.success() { + return Err(format!( + "clickhousectl with --ca-certificate exited {:?}: {}", + with_ca.status.code(), + String::from_utf8_lossy(&with_ca.stderr) + ) + .into()); + } + let pipe: ClickPipe = serde_json::from_slice(&with_ca.stdout) + .map_err(|error| format!("clickhousectl returned invalid ClickPipe JSON: {error}"))?; let clickpipe_id = clickpipe_id(&pipe)?; cleanup.register_clickpipe(clickhouse_id.clone(), clickpipe_id.clone()); eprintln!(" provisioned clickpipe id "); @@ -472,6 +506,23 @@ fn log_step(message: &str) { eprintln!(" step: {message}"); } +fn clickhousectl_binary() -> TestResult { + let configured = std::env::var_os(CLICKHOUSECTL_BINARY_ENV).ok_or_else(|| { + format!( + "{CLICKHOUSECTL_BINARY_ENV} must point to a built clickhousectl binary; run `cargo build -p clickhousectl` first" + ) + })?; + let binary = std::fs::canonicalize(configured)?; + if !binary.is_file() { + return Err(format!( + "{CLICKHOUSECTL_BINARY_ENV} does not point to a file: {}", + binary.display() + ) + .into()); + } + Ok(binary) +} + fn duration_from_env_or(name: &str, default_secs: u64) -> TestResult { match std::env::var(name) { Ok(value) => Ok(Duration::from_secs(value.parse()?)), From 6f1d8be48489257a07fa783e290dede923c12706 Mon Sep 17 00:00:00 2001 From: sdairs Date: Mon, 24 Aug 2026 16:52:59 +0100 Subject: [PATCH 2/3] Run CLI-backed ClickPipe test in live CI --- .github/workflows/cloud-integration.yml | 8 ++++++++ AGENTS.md | 2 +- crates/clickhouse-cloud-api/README.md | 5 ++++- 3 files changed, 13 insertions(+), 2 deletions(-) diff --git a/.github/workflows/cloud-integration.yml b/.github/workflows/cloud-integration.yml index 36db509a..12631548 100644 --- a/.github/workflows/cloud-integration.yml +++ b/.github/workflows/cloud-integration.yml @@ -254,10 +254,18 @@ jobs: needs.plan.outputs.organization == 'true' }} run: cargo test -p clickhouse-cloud-api --test integration_org_test -- --ignored --nocapture + - name: Build CLI for ClickPipe Postgres CDC integration test + if: >- + ${{ !cancelled() && + needs.plan.outputs.clickpipes == 'true' }} + run: cargo build -p clickhousectl + - name: Run ClickPipe Postgres CDC integration test if: >- ${{ !cancelled() && needs.plan.outputs.clickpipes == 'true' }} + env: + CLICKHOUSECTL_TEST_BINARY: ${{ github.workspace }}/target/debug/clickhousectl run: cargo test -p clickhouse-cloud-api --test clickpipe_postgres_cdc_test -- --ignored --nocapture # ClickPipe create smoke tests run against a long-lived ClickHouse Cloud diff --git a/AGENTS.md b/AGENTS.md index 9ff0aef8..6c13a7e1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -186,7 +186,7 @@ Real cloud integration tests, 100% OpenAPI spec coverage. Cost is not a reason t - `tests/common/support.rs` — generic test infra (polling, logging, env helpers, ClickHouse provisioning & cleanup, HTTP query helper). Used by every integration binary. Call `Client` directly from Rust. - `tests/integration_test.rs`, `tests/integration_postgres_test.rs`, `tests/integration_org_test.rs` — cloud-service, Postgres-service, and organization lifecycle tests. -- `tests/clickpipes/` — ClickPipes E2E suite, including external cloud services. Only Postgres CDC (uses ClickHouse & Postgres inside ClickHouse Cloud) is run in CI. Tests for third party services must be executed manually. CI also optionally runs `clickpipe_smoke_test` against a long-lived service when the `CLICKHOUSE_CLOUD_TEST_CLICKPIPE_SERVICE_ID` repo variable is set (see `.github/workflows/cloud-integration.yml`); the step is skipped when the variable is unset. +- `tests/clickpipes/` — ClickPipes E2E suite, including external cloud services. Only Postgres CDC (uses ClickHouse & Postgres inside ClickHouse Cloud and the built `clickhousectl` binary for ClickPipe creation) is run in CI. Tests for third party services must be executed manually. CI also optionally runs `clickpipe_smoke_test` against a long-lived service when the `CLICKHOUSE_CLOUD_TEST_CLICKPIPE_SERVICE_ID` repo variable is set (see `.github/workflows/cloud-integration.yml`); the step is skipped when the variable is unset. - `spec_coverage_test.rs`: runs the shared analyzer against the vendored OpenAPI snapshot and requires an actionable-drift-free report. - Labeled internal PRs classify the exact base-to-head diff with `scripts/classify-cloud-integration.py` and run only affected `service`, `postgres`, `organization`, and `clickpipes` suites. New or renamed API source/test files must be added to its explicit mappings; unknown paths fail closed to all suites. Scheduled runs still select all suites, while manual runs use the requested scope. diff --git a/crates/clickhouse-cloud-api/README.md b/crates/clickhouse-cloud-api/README.md index ba8ce1ab..5cd82d34 100644 --- a/crates/clickhouse-cloud-api/README.md +++ b/crates/clickhouse-cloud-api/README.md @@ -80,12 +80,15 @@ cargo test --test clickpipe_kinesis_test -- --ignored --nocapture # per-s cargo test --test clickpipe_mongo_test -- --ignored --nocapture # per-source: MongoDB cargo test --test clickpipe_mysql_test -- --ignored --nocapture # per-source: MySQL cargo test --test clickpipe_postgres_ec2_test -- --ignored --nocapture # per-source: Postgres-on-EC2 -cargo test --test clickpipe_postgres_cdc_test -- --ignored --nocapture # CHC-managed Postgres CDC +cargo build -p clickhousectl +CLICKHOUSECTL_TEST_BINARY=../../target/debug/clickhousectl cargo test --test clickpipe_postgres_cdc_test -- --ignored --nocapture # CHC-managed Postgres CDC through the CLI cargo test --test clickpipe_smoke_test -- --ignored --nocapture # create-only smoke against a shared service ``` All require `CLICKHOUSE_CLOUD_API_KEY`, `CLICKHOUSE_CLOUD_API_SECRET`, `CLICKHOUSE_CLOUD_TEST_ORG_ID`, `CLICKHOUSE_CLOUD_TEST_PROVIDER`, and `CLICKHOUSE_CLOUD_TEST_REGION` in the environment, and are wired into the scheduled `Cloud Integration` GitHub Actions workflow. The ClickPipes E2E suites additionally need AWS credentials and an `eu-west-1` region quota; `clickpipe_smoke_test` reads a pre-provisioned service ID from `CLICKHOUSE_CLOUD_TEST_CLICKPIPE_SERVICE_ID`. +The managed-Postgres CDC test fetches its private CA, writes it to a temporary file, and exercises failed CLI creation without `--ca-certificate` followed by successful snapshot and CDC with the flag. The mandatory CI fixtures do not currently include a PostgreSQL source with a publicly trusted certificate, so successful creation without an uploaded CA remains untested. + Applying `run-cloud-integration` to an eligible PR selects suites from that PR's merge-base-to-head diff; known changes without a live suite finish without entering the environment-bearing job, while unknown API source or test paths select all suites. Changes to the classifier, its tests, or the workflow also select all suites through an independent workflow guard. Manual `Cloud Integration` dispatches accept `scope=all`, `service`, `postgres`, `organization`, or `clickpipes`. The focused scopes run only their corresponding suite; `clickpipes` runs Postgres CDC plus the fixture-gated smoke test, while `all` runs all four mandatory suites plus that optional smoke test. Because stacked-PR diffs exclude inherited changes, manually run `scope=all` against the top stack branch for full-stack validation. `spec_coverage_test` sends the checked-in sources and snapshot through the From 41a0cc950acbc42293eb2c1aabc91f05b8351286 Mon Sep 17 00:00:00 2001 From: sdairs Date: Mon, 24 Aug 2026 17:29:03 +0100 Subject: [PATCH 3/3] Verify CLI-created tables without FINAL --- .../clickhouse-cloud-api/tests/clickpipes/postgres_cdc_test.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/clickhouse-cloud-api/tests/clickpipes/postgres_cdc_test.rs b/crates/clickhouse-cloud-api/tests/clickpipes/postgres_cdc_test.rs index c09f4753..bbdccce5 100644 --- a/crates/clickhouse-cloud-api/tests/clickpipes/postgres_cdc_test.rs +++ b/crates/clickhouse-cloud-api/tests/clickpipes/postgres_cdc_test.rs @@ -664,7 +664,7 @@ impl ClickHouseQuery { async fn count_rows(&self, table: &str) -> TestResult { let body = self .run_query(&format!( - "SELECT count() FROM default.{table} FINAL FORMAT TabSeparated" + "SELECT count() FROM default.{table} FORMAT TabSeparated" )) .await?; Ok(body.trim().parse::()?)