Added tests for auto create

Author: harshil-goel

src/config.rs

@@ -792,6 +792,92 @@ mod tests {
         );
     }
 
+    fn auto_create_set(
+        base: &EmitterConfig,
+        overlay: &ConfigOverlay,
+    ) -> std::collections::HashSet<String> {
+        use crate::emit::ch_ddl::DdlConfig;
+        let (r, _) = ConfigResolver::resolve(
+            base,
+            overlay,
+            &CliOverrides::default(),
+            &OptInState::default(),
+            &HashMap::new(),
+        );
+        DdlConfig::from_resolved(&r, "db".into(), false).auto_create_namespaces
+    }
+
+    #[test]
+    fn ddl_config_auto_create_from_toml() {
+        let base = EmitterConfig::from_toml_str(
+            "[ch]\n[namespace.s1]\nauto_create = true\n[namespace.s2]\nauto_create = false\n",
+        )
+        .unwrap();
+        let ns = auto_create_set(&base, &ConfigOverlay::default());
+        assert!(ns.contains("s1"), "TOML auto_create=true enables");
+        assert!(!ns.contains("s2"), "TOML auto_create=false stays off");
+    }
+
+    #[test]
+    fn ddl_config_auto_create_from_overlay() {
+        // No TOML namespace; the overlay alone turns it on.
+        let base = base_with("retain");
+        let mut overlay = ConfigOverlay::default();
+        overlay.namespaces.insert(
+            "s1".into(),
+            NamespaceRow {
+                auto_create: Some(true),
+                ..Default::default()
+            },
+        );
+        let ns = auto_create_set(&base, &overlay);
+        assert!(ns.contains("s1"), "overlay auto_create=true enables");
+    }
+
+    #[test]
+    fn overlay_auto_create_overrides_toml() {
+        // TOML false, overlay true → enabled.
+        let base =
+            EmitterConfig::from_toml_str("[ch]\n[namespace.s1]\nauto_create = false\n").unwrap();
+        let mut overlay = ConfigOverlay::default();
+        overlay.namespaces.insert(
+            "s1".into(),
+            NamespaceRow {
+                auto_create: Some(true),
+                ..Default::default()
+            },
+        );
+        assert!(
+            auto_create_set(&base, &overlay).contains("s1"),
+            "overlay true beats TOML false"
+        );
+
+        // TOML true, overlay false → disabled.
+        let base =
+            EmitterConfig::from_toml_str("[ch]\n[namespace.s1]\nauto_create = true\n").unwrap();
+        let mut overlay = ConfigOverlay::default();
+        overlay.namespaces.insert(
+            "s1".into(),
+            NamespaceRow {
+                auto_create: Some(false),
+                ..Default::default()
+            },
+        );
+        assert!(
+            !auto_create_set(&base, &overlay).contains("s1"),
+            "overlay false beats TOML true"
+        );
+    }
+
+    #[test]
+    fn ddl_config_no_auto_create_when_unset() {
+        let base = base_with("retain");
+        assert!(
+            auto_create_set(&base, &ConfigOverlay::default()).is_empty(),
+            "no source sets auto_create → empty set"
+        );
+    }
+
     fn rel_desc(namespace: &str, name: &str) -> RelDescriptor {
         use crate::schema::{RelAttr, ReplIdent};
         use walrus::pg::walparser::RelFileNode;

src/emit/ch_emitter.rs

@@ -2126,6 +2126,28 @@ mod tests {
         assert!(c.soft_delete);
     }
 
+    #[test]
+    fn namespace_toml_parses_auto_create() {
+        let c = EmitterConfig::from_toml_str(
+            "[ch]\n\
+             [namespace.s1]\n\
+             auto_create = true\n\
+             [namespace.s2]\n\
+             auto_create = false\n\
+             [namespace.s3]\n\
+             target_database = \"warehouse\"\n",
+        )
+        .unwrap();
+        assert!(c.namespaces["s1"].auto_create, "explicit true");
+        assert!(!c.namespaces["s2"].auto_create, "explicit false");
+        // Key absent defaults off (unwrap_or(false)).
+        assert!(!c.namespaces["s3"].auto_create, "absent defaults off");
+        assert_eq!(
+            c.namespaces["s3"].target_database.as_deref(),
+            Some("warehouse")
+        );
+    }
+
     /// Dotted names stay inside their TOML key level: schema `a.b` table `c`
     /// and schema `a` table `b.c` are distinct rels, distinct targets.
     #[test]

tests/ddl_replicates.rs

@@ -28,6 +28,7 @@ mod fx;
 
 use std::time::Duration;
 
+use walshadow::ch_emitter::EmitterConfig;
 use walshadow::mapping::TableTarget;
 use walshadow::mapping::{ColumnMapping, NamespaceMapping};
 use walshadow::schema::RelName;
@@ -634,3 +635,110 @@ async fn auto_create_honors_per_namespace_target_database() {
         "table must not be created in the global database",
     );
 }
+
+/// Same as `create_table_auto_replicates_in_namespace`, but the auto-create
+/// namespace comes from a parsed TOML `[namespace.<ns>]` block rather than a
+/// programmatic `NamespaceMapping` — exercising the real
+/// `EmitterConfig::from_toml_str` → resolve → `CREATE TABLE` path end-to-end.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn create_table_auto_replicates_from_toml_namespace() {
+    if !fx::pg_available() || !fx::pg_basebackup_available() || !fx::clickhouse_available() {
+        eprintln!("skip: missing initdb / pg_basebackup / clickhouse");
+        return;
+    }
+
+    let tmp = tempfile::tempdir().unwrap();
+    let source_port = SOURCE_PORT + 60;
+    let shadow_port = SHADOW_PORT + 60;
+    let ch_tcp_port = CH_TCP_PORT + 60;
+    let ch_http_port = CH_HTTP_PORT + 60;
+    let walsender_port = WALSENDER_PORT + 60;
+    let (
+        fx::BootstrappedClusters {
+            source,
+            shadow,
+            shadow_filter_dir,
+        },
+        shadow_stream_state,
+    ) = fx::bootstrap_clusters(
+        &tmp,
+        "CREATE SCHEMA s15toml;\n",
+        source_port,
+        shadow_port,
+        walsender_port,
+    )
+    .await;
+    let _src_stop = fx::StopOnDrop { sh: &source };
+    let _shd_stop = fx::StopOnDrop { sh: &shadow };
+
+    let ch_tmp = tempfile::tempdir().unwrap();
+    let ch = fx::ChServer::spawn(ch_tmp, ch_tcp_port, ch_http_port).expect("spawn ch");
+    ch.query("CREATE DATABASE IF NOT EXISTS walshadow_test")
+        .expect("create db");
+
+    // Empty programmatic namespaces — the TOML parsed in `tune` is the only
+    // source of the auto-create flag.
+    let mut pipeline = fx::build_pipeline_with(
+        fx::BuildPipelineArgs {
+            tmp: &tmp,
+            source: &source,
+            shadow: &shadow,
+            shadow_filter_dir: &shadow_filter_dir,
+            shadow_stream_state,
+            ch_database: "walshadow_test",
+            ch_tcp_port,
+            mappings: vec![],
+            app_name: "walshadow-ddl-create-auto-toml",
+            ddl: Some(fx::DdlPipelineArgs::default()),
+        },
+        |cfg| {
+            cfg.namespaces = EmitterConfig::from_toml_str(
+                "[ch]\n\
+                 [namespace.s15toml]\n\
+                 target_database = \"walshadow_test\"\n\
+                 auto_create = true\n",
+            )
+            .expect("namespace toml parses")
+            .namespaces;
+        },
+    )
+    .await;
+
+    let driver = fx::spawn_workload(
+        &source,
+        vec![
+            "CREATE TABLE s15toml.new_t (id bigint PRIMARY KEY, body text)".into(),
+            "INSERT INTO s15toml.new_t (id, body) VALUES (1, 'toml-auto')".into(),
+            "SELECT pg_switch_wal()".into(),
+        ],
+    );
+
+    let shipped = fx::pump_segments(&mut pipeline, 1, Duration::from_secs(60)).await;
+    let _ = driver.join();
+    assert!(shipped >= 1, "no segments shipped in 60s");
+
+    let target = pipeline.stream.dispatched_lsn();
+    let observed = shadow
+        .wait_for_replay(target, Duration::from_secs(30))
+        .expect("shadow replay");
+    assert!(observed >= target);
+    pipeline.shutdown().await.expect("pipeline drains clean");
+
+    let tbls = ch
+        .query(
+            "SELECT name FROM system.tables WHERE database = 'walshadow_test' AND name = 'new_t'",
+        )
+        .expect("ch table existence");
+    assert_eq!(
+        tbls, "new_t",
+        "TOML [namespace] auto_create must create the CH table"
+    );
+
+    let body = ch
+        .query(
+            "SELECT argMax(body, _lsn) FROM walshadow_test.new_t \
+             WHERE _is_deleted = 0 AND id = 1",
+        )
+        .expect("ch body");
+    assert_eq!(body, "toml-auto");
+}

tests/runtime_config_e2e.rs

@@ -46,6 +46,13 @@
 //!      stored `123.45` (vs a scale-0 `123`) proves the override reached
 //!      the projection (plans/config.md §Column overrides).
 //!
+//! 7. `auto_create_namespace_via_config_namespace`
+//!    * Operator inserts `config_namespace (auto_create=true)`, no
+//!      `config_table` row and no TOML mapping.
+//!    * Source `CREATE TABLE` in the namespace + INSERT.
+//!    * Expect: the namespace flag alone auto-creates the CH table and the
+//!      row lands.
+//!
 //! Source-side `config_*` install runs the real `sql/runtime_config_install.sql`
 //! inside the bootstrap schema dump, so the drills double as install-script
 //! coverage (psql `\if` default-schema guard included).
@@ -641,6 +648,100 @@ async fn opt_in_then_alter_add_column_reaches_ch() {
     assert_eq!(pre, "pre-alter\t1");
 }
 
+/// Drill 7: `config_namespace.auto_create = true` alone (no `config_table`
+/// row, no TOML mapping) authorises namespace-wide auto-create. A source
+/// `CREATE TABLE` in the flagged namespace must run `CREATE TABLE` on CH and
+/// the trailing INSERT must land — proving the overlay's namespace layer
+/// drives auto-create, not just the per-table `replicate=true` opt-in.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn auto_create_namespace_via_config_namespace() {
+    if !fx::pg_available() || !fx::pg_basebackup_available() || !fx::clickhouse_available() {
+        eprintln!("skip: missing initdb / pg_basebackup / clickhouse");
+        return;
+    }
+
+    let tmp = tempfile::tempdir().unwrap();
+    let source_port = SOURCE_PORT + 60;
+    let shadow_port = SHADOW_PORT + 60;
+    let ch_tcp_port = CH_TCP_PORT + 60;
+    let ch_http_port = CH_HTTP_PORT + 60;
+    let walsender_port = WALSENDER_PORT + 60;
+    let schema_sql = format!("{INSTALL_SQL}\nCREATE SCHEMA app;\n");
+    let (
+        fx::BootstrappedClusters {
+            source,
+            shadow,
+            shadow_filter_dir,
+        },
+        shadow_stream_state,
+    ) = fx::bootstrap_clusters(&tmp, &schema_sql, source_port, shadow_port, walsender_port).await;
+    let _src_stop = fx::StopOnDrop { sh: &source };
+    let _shd_stop = fx::StopOnDrop { sh: &shadow };
+
+    let ch_tmp = tempfile::tempdir().unwrap();
+    let ch = fx::ChServer::spawn(ch_tmp, ch_tcp_port, ch_http_port).expect("spawn ch");
+    ch.query("CREATE DATABASE IF NOT EXISTS walshadow_test")
+        .expect("create db");
+
+    // No TOML namespaces — the config_namespace row alone authorises it.
+    let mut pipeline = fx::build_pipeline(fx::BuildPipelineArgs {
+        tmp: &tmp,
+        source: &source,
+        shadow: &shadow,
+        shadow_filter_dir: &shadow_filter_dir,
+        shadow_stream_state,
+        ch_database: "walshadow_test",
+        ch_tcp_port,
+        mappings: vec![],
+        app_name: "walshadow-config-ns-auto-create",
+        ddl: Some(overlay_ddl_args()),
+    })
+    .await;
+
+    // The auto_create row commits before the CREATE TABLE, so `apply_added`
+    // sees the namespace in `auto_create_namespaces` when the DDL drains.
+    let driver = fx::spawn_workload(
+        &source,
+        vec![
+            "INSERT INTO walshadow.config_namespace (namespace, target_database, auto_create) \
+             VALUES ('app', 'walshadow_test', true)"
+                .into(),
+            "CREATE TABLE app.thing (id bigint PRIMARY KEY, body text)".into(),
+            "INSERT INTO app.thing (id, body) VALUES (1, 'ns-auto')".into(),
+            "SELECT pg_switch_wal()".into(),
+        ],
+    );
+
+    let shipped = fx::pump_segments(&mut pipeline, 1, Duration::from_secs(45)).await;
+    let _ = driver.join();
+    assert!(shipped >= 1, "no segments shipped in 45s");
+
+    let target = pipeline.stream.dispatched_lsn();
+    let observed = shadow
+        .wait_for_replay(target, Duration::from_secs(30))
+        .expect("shadow replay");
+    assert!(observed >= target);
+    pipeline.shutdown().await.expect("pipeline drains clean");
+
+    let tbls = ch
+        .query(
+            "SELECT name FROM system.tables WHERE database = 'walshadow_test' AND name = 'thing'",
+        )
+        .expect("ch table existence");
+    assert_eq!(
+        tbls, "thing",
+        "config_namespace.auto_create must create the CH table"
+    );
+
+    let body = ch
+        .query(
+            "SELECT argMax(body, _lsn) FROM walshadow_test.thing \
+             WHERE _is_deleted = 0 AND id = 1",
+        )
+        .expect("ch body");
+    assert_eq!(body, "ns-auto");
+}
+
 /// Drill 6: `config_column.target_type` reaches the emitted projection
 /// (plans/config.md §Column overrides). CH dest pre-created with
 /// `Decimal(38, 2)` while TOML deliberately maps the stale bridge default