Exercise PostgreSQL ClickPipe creation through the CLI in live CI

.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

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.
 

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"

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

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 <path>.";
 
 #[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<std::process::Output> {
+            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 <redacted>");
@@ -472,6 +506,23 @@ fn log_step(message: &str) {
     eprintln!("  step: {message}");
 }
 
+fn clickhousectl_binary() -> TestResult<PathBuf> {
+    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<Duration> {
     match std::env::var(name) {
         Ok(value) => Ok(Duration::from_secs(value.parse()?)),
@@ -613,7 +664,7 @@ impl ClickHouseQuery {
     async fn count_rows(&self, table: &str) -> TestResult<i64> {
         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::<i64>()?)