use super::*;
use crate::dialects::Dialect;
use crate::operations::Operation;

fn basic_table(name: &str) -> Table {
    Table {
        name: name.to_string(),
        schema: None,
        columns: vec![],
        foreign_keys: vec![],
        indexes: vec![],
        constraints: vec![],
        triggers: vec![],
    }
}

fn text_col(name: &str) -> Column {
    Column {
        name: name.to_string(),
        col_type: "text".to_string(),
        nullable: false,
        default: None,
        primary_key: false,
        ..Default::default()
    }
}

fn apply_ok(state: &mut Schema, op: Operation) {
    state.apply(&op).expect("apply should succeed");
}

/// CreateTable inserts the table into the state.
#[test]
fn create_table() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    assert!(s.tables.contains_key("users"));
}

/// CreateTable on an existing table name returns TableAlreadyExists.
#[test]
fn create_table_duplicate() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    let err = s
        .apply(&Operation::CreateTable {
            table: basic_table("users"),
        })
        .unwrap_err();
    assert_eq!(err, ReplayError::TableAlreadyExists("users".to_string()));
}

/// DropTable removes the table from state.
#[test]
fn drop_table() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    apply_ok(
        &mut s,
        Operation::DropTable {
            table: basic_table("users"),
        },
    );
    assert!(!s.tables.contains_key("users"));
}

/// DropTable on a nonexistent table returns TableNotFound.
#[test]
fn drop_table_not_found() {
    let mut s = Schema::default();
    let err = s
        .apply(&Operation::DropTable {
            table: basic_table("ghost"),
        })
        .unwrap_err();
    assert_eq!(err, ReplayError::TableNotFound("ghost".to_string()));
}

/// RenameTable moves the table to the new name and updates its `name` field.
#[test]
fn rename_table() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    apply_ok(
        &mut s,
        Operation::RenameTable {
            old_name: "users".to_string(),
            new_name: "accounts".to_string(),
        },
    );
    assert!(!s.tables.contains_key("users"));
    assert_eq!(s.tables["accounts"].name, "accounts");
}

/// RenameTable errors if the target name already exists.
#[test]
fn rename_table_target_exists() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("accounts"),
        },
    );
    let err = s
        .apply(&Operation::RenameTable {
            old_name: "users".to_string(),
            new_name: "accounts".to_string(),
        })
        .unwrap_err();
    assert!(matches!(err, ReplayError::RenameTargetExists { .. }));
}

/// AddColumn appends the column to the table.
#[test]
fn add_column() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    apply_ok(
        &mut s,
        Operation::AddColumn {
            table_name: "users".to_string(),
            column: text_col("email"),
        },
    );
    assert_eq!(s.tables["users"].columns[0].name, "email");
}

/// AddColumn on an existing column name returns ColumnAlreadyExists.
#[test]
fn add_column_duplicate() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    apply_ok(
        &mut s,
        Operation::AddColumn {
            table_name: "users".to_string(),
            column: text_col("email"),
        },
    );
    let err = s
        .apply(&Operation::AddColumn {
            table_name: "users".to_string(),
            column: text_col("email"),
        })
        .unwrap_err();
    assert!(matches!(err, ReplayError::ColumnAlreadyExists { .. }));
}

/// DropColumn removes the named column.
#[test]
fn drop_column() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    apply_ok(
        &mut s,
        Operation::AddColumn {
            table_name: "users".to_string(),
            column: text_col("email"),
        },
    );
    apply_ok(
        &mut s,
        Operation::DropColumn {
            table_name: "users".to_string(),
            column: text_col("email"),
            cascade: false,
        },
    );
    assert!(s.tables["users"].columns.is_empty());
}

/// DropColumn on a missing column returns ColumnNotFound.
#[test]
fn drop_column_not_found() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    let err = s
        .apply(&Operation::DropColumn {
            table_name: "users".to_string(),
            column: text_col("ghost"),
            cascade: false,
        })
        .unwrap_err();
    assert!(matches!(err, ReplayError::ColumnNotFound { .. }));
}

/// RenameColumn changes the column's name in place.
#[test]
fn rename_column() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    apply_ok(
        &mut s,
        Operation::AddColumn {
            table_name: "users".to_string(),
            column: text_col("email"),
        },
    );
    apply_ok(
        &mut s,
        Operation::RenameColumn {
            table_name: "users".to_string(),
            old_name: "email".to_string(),
            new_name: "email_address".to_string(),
        },
    );
    assert_eq!(s.tables["users"].columns[0].name, "email_address");
}

/// AlterColumn replaces the column definition.
#[test]
fn alter_column() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    apply_ok(
        &mut s,
        Operation::AddColumn {
            table_name: "users".to_string(),
            column: text_col("bio"),
        },
    );
    let new_col = Column {
        name: "bio".to_string(),
        col_type: "varchar(500)".to_string(),
        nullable: true,
        default: None,
        primary_key: false,
        ..Default::default()
    };
    apply_ok(
        &mut s,
        Operation::AlterColumn {
            table_name: "users".to_string(),
            old: text_col("bio"),
            new: new_col,
            cast_expr: None,
        },
    );
    assert_eq!(s.tables["users"].columns[0].col_type, "varchar(500)");
    assert!(s.tables["users"].columns[0].nullable);
}

/// AddForeignKey attaches a FK to the table.
#[test]
fn add_foreign_key() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("posts"),
        },
    );
    let fk = ForeignKey {
        name: "fk_user".to_string(),
        from_column: "user_id".to_string(),
        to_table: "users".to_string(),
        to_column: "id".to_string(),
    };
    apply_ok(
        &mut s,
        Operation::AddForeignKey {
            table_name: "posts".to_string(),
            foreign_key: fk,
        },
    );
    assert_eq!(s.tables["posts"].foreign_keys[0].name, "fk_user");
}

/// AddForeignKey with a duplicate name returns ForeignKeyAlreadyExists.
#[test]
fn add_foreign_key_duplicate() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("posts"),
        },
    );
    let fk = ForeignKey {
        name: "fk_user".to_string(),
        from_column: "user_id".to_string(),
        to_table: "users".to_string(),
        to_column: "id".to_string(),
    };
    apply_ok(
        &mut s,
        Operation::AddForeignKey {
            table_name: "posts".to_string(),
            foreign_key: fk.clone(),
        },
    );
    let err = s
        .apply(&Operation::AddForeignKey {
            table_name: "posts".to_string(),
            foreign_key: fk,
        })
        .unwrap_err();
    assert!(matches!(err, ReplayError::ForeignKeyAlreadyExists { .. }));
}

/// DropForeignKey removes the FK from the table.
#[test]
fn drop_foreign_key() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("posts"),
        },
    );
    let fk = ForeignKey {
        name: "fk_user".to_string(),
        from_column: "user_id".to_string(),
        to_table: "users".to_string(),
        to_column: "id".to_string(),
    };
    apply_ok(
        &mut s,
        Operation::AddForeignKey {
            table_name: "posts".to_string(),
            foreign_key: fk,
        },
    );
    apply_ok(
        &mut s,
        Operation::DropForeignKey {
            table_name: "posts".to_string(),
            foreign_key: ForeignKey {
                name: "fk_user".to_string(),
                from_column: "user_id".to_string(),
                to_table: "users".to_string(),
                to_column: "id".to_string(),
            },
            cascade: false,
        },
    );
    assert!(s.tables["posts"].foreign_keys.is_empty());
}

/// AddIndex attaches an index to the table.
#[test]
fn add_index() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    let idx = Index {
        name: "idx_email".to_string(),
        columns: vec!["email".to_string()],
        unique: true,
        predicate: None,
    };
    apply_ok(
        &mut s,
        Operation::AddIndex {
            table_name: "users".to_string(),
            index: idx,
            concurrent: false,
        },
    );
    assert_eq!(s.tables["users"].indexes[0].name, "idx_email");
}

/// AddIndex with a duplicate name returns IndexAlreadyExists.
#[test]
fn add_index_duplicate() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    let idx = Index {
        name: "idx_email".to_string(),
        columns: vec!["email".to_string()],
        unique: true,
        predicate: None,
    };
    apply_ok(
        &mut s,
        Operation::AddIndex {
            table_name: "users".to_string(),
            index: idx.clone(),
            concurrent: false,
        },
    );
    let err = s
        .apply(&Operation::AddIndex {
            table_name: "users".to_string(),
            index: idx,
            concurrent: false,
        })
        .unwrap_err();
    assert!(matches!(err, ReplayError::IndexAlreadyExists { .. }));
}

/// DropIndex removes the index.
#[test]
fn drop_index() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    let idx = Index {
        name: "idx_email".to_string(),
        columns: vec!["email".to_string()],
        unique: true,
        predicate: None,
    };
    apply_ok(
        &mut s,
        Operation::AddIndex {
            table_name: "users".to_string(),
            index: idx,
            concurrent: false,
        },
    );
    apply_ok(
        &mut s,
        Operation::DropIndex {
            table_name: "users".to_string(),
            index: Index {
                name: "idx_email".to_string(),
                columns: vec!["email".to_string()],
                unique: true,
                predicate: None,
            },
            concurrent: false,
        },
    );
    assert!(s.tables["users"].indexes.is_empty());
}

/// AddConstraint attaches a check constraint to the table.
#[test]
fn add_constraint() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    let c = Constraint::Check {
        name: "chk_age".to_string(),
        expression: "age > 0".to_string(),
    };
    apply_ok(
        &mut s,
        Operation::AddConstraint {
            table_name: "users".to_string(),
            constraint: c,
        },
    );
    assert_eq!(s.tables["users"].constraints[0].name(), "chk_age");
}

/// AddConstraint with a duplicate name returns ConstraintAlreadyExists.
#[test]
fn add_constraint_duplicate() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    let c = Constraint::Unique {
        name: "uq_email".to_string(),
        columns: vec!["email".to_string()],
    };
    apply_ok(
        &mut s,
        Operation::AddConstraint {
            table_name: "users".to_string(),
            constraint: c.clone(),
        },
    );
    let err = s
        .apply(&Operation::AddConstraint {
            table_name: "users".to_string(),
            constraint: c,
        })
        .unwrap_err();
    assert!(matches!(err, ReplayError::ConstraintAlreadyExists { .. }));
}

/// DropConstraint removes the named constraint.
#[test]
fn drop_constraint() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    let c = Constraint::Unique {
        name: "uq_email".to_string(),
        columns: vec!["email".to_string()],
    };
    apply_ok(
        &mut s,
        Operation::AddConstraint {
            table_name: "users".to_string(),
            constraint: c,
        },
    );
    apply_ok(
        &mut s,
        Operation::DropConstraint {
            table_name: "users".to_string(),
            constraint: Constraint::Unique {
                name: "uq_email".to_string(),
                columns: vec!["email".to_string()],
            },
        },
    );
    assert!(s.tables["users"].constraints.is_empty());
}

/// Statement and Invoke are no-ops — state is unchanged.
#[test]
fn statement_and_invoke_are_noops() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::Statement {
            up: "SELECT 1".to_string(),
            down: None,
        },
    );
    apply_ok(
        &mut s,
        Operation::Invoke {
            up: "seed_data".to_string(),
            down: None,
        },
    );
    assert!(s.tables.is_empty());
}

/// Replaying a sequence of operations builds the expected state end-to-end.
#[test]
fn end_to_end_replay() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    apply_ok(
        &mut s,
        Operation::AddColumn {
            table_name: "users".to_string(),
            column: text_col("email"),
        },
    );
    apply_ok(
        &mut s,
        Operation::AddColumn {
            table_name: "users".to_string(),
            column: text_col("bio"),
        },
    );
    apply_ok(
        &mut s,
        Operation::DropColumn {
            table_name: "users".to_string(),
            column: text_col("bio"),
            cascade: false,
        },
    );
    apply_ok(
        &mut s,
        Operation::RenameColumn {
            table_name: "users".to_string(),
            old_name: "email".to_string(),
            new_name: "email_address".to_string(),
        },
    );
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("posts"),
        },
    );
    apply_ok(
        &mut s,
        Operation::AddColumn {
            table_name: "posts".to_string(),
            column: text_col("title"),
        },
    );

    assert_eq!(s.tables.len(), 2);
    assert_eq!(s.tables["users"].columns.len(), 1);
    assert_eq!(s.tables["users"].columns[0].name, "email_address");
    assert_eq!(s.tables["posts"].columns[0].name, "title");
}

/// Operations on a nonexistent table return TableNotFound.
#[test]
fn table_not_found_propagates() {
    let mut s = Schema::default();
    let err = s
        .apply(&Operation::AddColumn {
            table_name: "ghost".to_string(),
            column: text_col("x"),
        })
        .unwrap_err();
    assert_eq!(err, ReplayError::TableNotFound("ghost".to_string()));
}

/// CreateTable with two primary_key columns returns MultiplePrimaryKeys.
#[test]
fn create_table_with_multiple_pk_returns_error() {
    let mut s = Schema::default();
    let table = Table {
        name: "users".to_string(),
        schema: None,
        columns: vec![
            Column {
                name: "id".to_string(),
                col_type: "bigint".to_string(),
                nullable: false,
                default: None,
                primary_key: true,
                ..Default::default()
            },
            Column {
                name: "alt_id".to_string(),
                col_type: "bigint".to_string(),
                nullable: false,
                default: None,
                primary_key: true,
                ..Default::default()
            },
        ],
        foreign_keys: vec![],
        indexes: vec![],
        constraints: vec![],
        triggers: vec![],
    };
    let err = s.apply(&Operation::CreateTable { table }).unwrap_err();
    assert_eq!(err, ReplayError::MultiplePrimaryKeys("users".to_string()));
}

/// AddColumn with primary_key=true when one already exists returns MultiplePrimaryKeys.
#[test]
fn add_pk_column_when_pk_exists_returns_error() {
    let mut s = Schema::default();
    let pk_col = Column {
        name: "id".to_string(),
        col_type: "bigint".to_string(),
        nullable: false,
        default: None,
        primary_key: true,
        ..Default::default()
    };
    let table = Table {
        name: "users".to_string(),
        schema: None,
        columns: vec![pk_col],
        foreign_keys: vec![],
        indexes: vec![],
        constraints: vec![],
        triggers: vec![],
    };
    s.apply(&Operation::CreateTable { table }).unwrap();
    let second_pk = Column {
        name: "alt_id".to_string(),
        col_type: "bigint".to_string(),
        nullable: false,
        default: None,
        primary_key: true,
        ..Default::default()
    };
    let err = s
        .apply(&Operation::AddColumn {
            table_name: "users".to_string(),
            column: second_pk,
        })
        .unwrap_err();
    assert_eq!(err, ReplayError::MultiplePrimaryKeys("users".to_string()));
}

/// AlterColumn that promotes a column to primary_key when one already exists returns MultiplePrimaryKeys.
#[test]
fn alter_column_to_pk_when_pk_exists_returns_error() {
    let mut s = Schema::default();
    let pk_col = Column {
        name: "id".to_string(),
        col_type: "bigint".to_string(),
        nullable: false,
        default: None,
        primary_key: true,
        ..Default::default()
    };
    let other_col = Column {
        name: "other".to_string(),
        col_type: "bigint".to_string(),
        nullable: false,
        default: None,
        primary_key: false,
        ..Default::default()
    };
    let table = Table {
        name: "users".to_string(),
        schema: None,
        columns: vec![pk_col, other_col.clone()],
        foreign_keys: vec![],
        indexes: vec![],
        constraints: vec![],
        triggers: vec![],
    };
    s.apply(&Operation::CreateTable { table }).unwrap();
    let promoted = Column {
        name: "other".to_string(),
        col_type: "bigint".to_string(),
        nullable: false,
        default: None,
        primary_key: true,
        ..Default::default()
    };
    let err = s
        .apply(&Operation::AlterColumn {
            table_name: "users".to_string(),
            old: other_col,
            new: promoted,
            cast_expr: None,
        })
        .unwrap_err();
    assert_eq!(err, ReplayError::MultiplePrimaryKeys("users".to_string()));
}

/// validate() returns Ok for a state with at most one PK column per table.
#[test]
fn validate_single_pk_ok() {
    let mut s = Schema::default();
    let pk_col = Column {
        name: "id".to_string(),
        col_type: "bigint".to_string(),
        nullable: false,
        default: None,
        primary_key: true,
        ..Default::default()
    };
    let table = Table {
        name: "users".to_string(),
        schema: None,
        columns: vec![pk_col],
        foreign_keys: vec![],
        indexes: vec![],
        constraints: vec![],
        triggers: vec![],
    };
    s.tables.insert("users".to_string(), table);
    assert!(s.validate().is_ok());
}

/// validate() returns Err when a table has more than one primary key column.
#[test]
fn validate_multiple_pk_returns_err() {
    let mut s = Schema::default();
    let table = Table {
        name: "users".to_string(),
        schema: None,
        columns: vec![
            Column {
                name: "id".to_string(),
                col_type: "bigint".to_string(),
                nullable: false,
                default: None,
                primary_key: true,
                ..Default::default()
            },
            Column {
                name: "alt".to_string(),
                col_type: "bigint".to_string(),
                nullable: false,
                default: None,
                primary_key: true,
                ..Default::default()
            },
        ],
        foreign_keys: vec![],
        indexes: vec![],
        constraints: vec![],
        triggers: vec![],
    };
    s.tables.insert("users".to_string(), table);
    assert!(s.validate().is_err());
}

/// normalize() moves an inline `references` into `table.foreign_keys` and clears the column field.
/// The FK name is auto-generated as `{table}_{column}_fkey` when no explicit name is given.
#[test]
fn normalize_moves_inline_fk_to_foreign_keys() {
    let col = Column {
        name: "user_id".to_string(),
        col_type: "bigint".to_string(),
        nullable: false,
        default: None,
        primary_key: false,
        references: Some(ColumnRef {
            table: "users".to_string(),
            column: "id".to_string(),
            name: None,
        }),
        check: None,
        generated: None,
    };
    let table = Table {
        name: "posts".to_string(),
        schema: None,
        columns: vec![col],
        foreign_keys: vec![],
        indexes: vec![],
        constraints: vec![],
        triggers: vec![],
    };
    let mut s = Schema::default();
    s.tables.insert("posts".to_string(), table);
    s.normalize();
    let t = &s.tables["posts"];
    assert_eq!(t.foreign_keys.len(), 1);
    assert_eq!(t.foreign_keys[0].name, "posts_user_id_fkey");
    assert_eq!(t.foreign_keys[0].from_column, "user_id");
    assert_eq!(t.foreign_keys[0].to_table, "users");
    assert_eq!(t.foreign_keys[0].to_column, "id");
    assert!(t.columns[0].references.is_none());
}

/// normalize() uses the explicit `name` from ColumnRef instead of auto-generating when provided.
#[test]
fn normalize_inline_fk_uses_explicit_name_when_provided() {
    let col = Column {
        name: "user_id".to_string(),
        col_type: "bigint".to_string(),
        nullable: false,
        default: None,
        primary_key: false,
        references: Some(ColumnRef {
            table: "users".to_string(),
            column: "id".to_string(),
            name: Some("fk_posts_user".to_string()),
        }),
        check: None,
        generated: None,
    };
    let table = Table {
        name: "posts".to_string(),
        schema: None,
        columns: vec![col],
        foreign_keys: vec![],
        indexes: vec![],
        constraints: vec![],
        triggers: vec![],
    };
    let mut s = Schema::default();
    s.tables.insert("posts".to_string(), table);
    s.normalize();
    assert_eq!(s.tables["posts"].foreign_keys[0].name, "fk_posts_user");
}

/// normalize() moves an inline `check` into `table.constraints` and clears the column field.
/// The constraint name is auto-generated as `{table}_{column}_check`.
#[test]
fn normalize_moves_inline_check_to_constraints() {
    let col = Column {
        name: "score".to_string(),
        col_type: "integer".to_string(),
        nullable: false,
        default: None,
        primary_key: false,
        references: None,
        check: Some("score >= 0".to_string()),
        generated: None,
    };
    let table = Table {
        name: "results".to_string(),
        schema: None,
        columns: vec![col],
        foreign_keys: vec![],
        indexes: vec![],
        constraints: vec![],
        triggers: vec![],
    };
    let mut s = Schema::default();
    s.tables.insert("results".to_string(), table);
    s.normalize();
    let t = &s.tables["results"];
    assert_eq!(t.constraints.len(), 1);
    assert!(
        matches!(&t.constraints[0], Constraint::Check { name, expression } if name == "results_score_check" && expression == "score >= 0")
    );
    assert!(t.columns[0].check.is_none());
}

/// normalize() is idempotent — running it twice produces the same result.
#[test]
fn normalize_is_idempotent() {
    let col = Column {
        name: "user_id".to_string(),
        col_type: "bigint".to_string(),
        nullable: false,
        default: None,
        primary_key: false,
        references: Some(ColumnRef {
            table: "users".to_string(),
            column: "id".to_string(),
            name: None,
        }),
        check: None,
        generated: None,
    };
    let table = Table {
        name: "posts".to_string(),
        schema: None,
        columns: vec![col],
        foreign_keys: vec![],
        indexes: vec![],
        constraints: vec![],
        triggers: vec![],
    };
    let mut s = Schema::default();
    s.tables.insert("posts".to_string(), table);
    s.normalize();
    s.normalize();
    assert_eq!(s.tables["posts"].foreign_keys.len(), 1);
}

/// Column.col_type serializes as the key "type" in YAML, not "col_type".
#[test]
fn col_type_serializes_as_type_in_yaml() {
    let col = Column {
        name: "id".to_string(),
        col_type: "bigint".to_string(),
        nullable: false,
        default: None,
        primary_key: true,
        references: None,
        check: None,
        generated: None,
    };
    let yaml = serde_yaml::to_string(&col).expect("serialize");
    assert!(
        yaml.contains("type: bigint"),
        "expected 'type: bigint' in: {yaml}"
    );
    assert!(
        !yaml.contains("col_type"),
        "col_type should not appear in: {yaml}"
    );
    let back: Column = serde_yaml::from_str(&yaml).expect("deserialize");
    assert_eq!(back.col_type, "bigint");
}

fn basic_function(name: &str) -> FunctionDef {
    FunctionDef {
        name: name.to_string(),
        schema: None,
        arguments: String::new(),
        returns: "void".to_string(),
        language: "sql".to_string(),
        body: "SELECT 1".to_string(),
        volatility: Volatility::Volatile,
        security_definer: false,
    }
}

fn basic_trigger(name: &str) -> TriggerDef {
    TriggerDef {
        name: Some(name.to_string()),
        timing: TriggerTiming::After,
        events: vec![TriggerEvent::Insert],
        scope: TriggerScope::Row,
        function_name: Some("some_fn".to_string()),
        when: None,
        body: None,
        language: None,
    }
}

/// CreateFunction inserts the function into state.
#[test]
fn create_function() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateFunction {
            function: basic_function("notify"),
        },
    );
    assert!(s.functions.contains_key("notify"));
}

/// CreateFunction on an existing name returns FunctionAlreadyExists.
#[test]
fn create_function_duplicate() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateFunction {
            function: basic_function("notify"),
        },
    );
    let err = s
        .apply(&Operation::CreateFunction {
            function: basic_function("notify"),
        })
        .unwrap_err();
    assert_eq!(
        err,
        ReplayError::FunctionAlreadyExists("notify".to_string())
    );
}

/// DropFunction removes the function from state.
#[test]
fn drop_function() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateFunction {
            function: basic_function("notify"),
        },
    );
    apply_ok(
        &mut s,
        Operation::DropFunction {
            function: basic_function("notify"),
        },
    );
    assert!(!s.functions.contains_key("notify"));
}

/// DropFunction on a nonexistent function returns FunctionNotFound.
#[test]
fn drop_function_not_found() {
    let mut s = Schema::default();
    let err = s
        .apply(&Operation::DropFunction {
            function: basic_function("ghost"),
        })
        .unwrap_err();
    assert_eq!(err, ReplayError::FunctionNotFound("ghost".to_string()));
}

/// AlterFunction replaces the old function entry with the new one.
#[test]
fn alter_function() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateFunction {
            function: basic_function("notify"),
        },
    );
    let mut updated = basic_function("notify");
    updated.body = "SELECT 2".to_string();
    apply_ok(
        &mut s,
        Operation::AlterFunction {
            old: basic_function("notify"),
            new: updated.clone(),
        },
    );
    assert_eq!(s.functions["notify"].body, "SELECT 2");
}

/// CreateTrigger attaches a trigger to the table.
#[test]
fn create_trigger() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    apply_ok(
        &mut s,
        Operation::CreateTrigger {
            table_name: "users".to_string(),
            trigger: basic_trigger("audit_trg"),
        },
    );
    assert_eq!(s.tables["users"].triggers.len(), 1);
}

/// CreateTrigger with a duplicate name returns TriggerAlreadyExists.
#[test]
fn create_trigger_duplicate() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    apply_ok(
        &mut s,
        Operation::CreateTrigger {
            table_name: "users".to_string(),
            trigger: basic_trigger("audit_trg"),
        },
    );
    let err = s
        .apply(&Operation::CreateTrigger {
            table_name: "users".to_string(),
            trigger: basic_trigger("audit_trg"),
        })
        .unwrap_err();
    assert_eq!(
        err,
        ReplayError::TriggerAlreadyExists {
            table: "users".to_string(),
            trigger: "audit_trg".to_string()
        }
    );
}

/// DropTrigger removes the trigger from the table.
#[test]
fn drop_trigger() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    apply_ok(
        &mut s,
        Operation::CreateTrigger {
            table_name: "users".to_string(),
            trigger: basic_trigger("audit_trg"),
        },
    );
    apply_ok(
        &mut s,
        Operation::DropTrigger {
            table_name: "users".to_string(),
            trigger: basic_trigger("audit_trg"),
        },
    );
    assert!(s.tables["users"].triggers.is_empty());
}

/// DropTrigger on a nonexistent trigger returns TriggerNotFound.
#[test]
fn drop_trigger_not_found() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    let err = s
        .apply(&Operation::DropTrigger {
            table_name: "users".to_string(),
            trigger: basic_trigger("ghost_trg"),
        })
        .unwrap_err();
    assert_eq!(
        err,
        ReplayError::TriggerNotFound {
            table: "users".to_string(),
            trigger: "ghost_trg".to_string()
        }
    );
}

/// normalize() auto-generates a trigger name from table + events + timing.
#[test]
fn normalize_auto_generates_trigger_name() {
    let yaml = r#"
tables:
  users:
    name: users
    columns: []
    foreign_keys: []
    indexes: []
    constraints: []
    triggers:
      - timing: after
        events: [update, insert]
        scope: row
        function_name: some_fn
"#;
    let state = Schema::from_yaml_str(yaml).expect("parse");
    let trigger = &state.tables["users"].triggers[0];
    assert_eq!(
        trigger.name.as_deref(),
        Some("users_insert_update_after_trg")
    );
}

/// normalize() converts an inline trigger body into a FunctionDef and sets function_name.
#[test]
fn normalize_inline_trigger_body_to_function() {
    let yaml = r#"
tables:
  orders:
    name: orders
    columns: []
    foreign_keys: []
    indexes: []
    constraints: []
    triggers:
      - name: orders_insert_after_trg
        timing: after
        events: [insert]
        scope: row
        body: "BEGIN RETURN NEW; END;"
"#;
    let state = Schema::from_yaml_str(yaml).expect("parse");
    let trigger = &state.tables["orders"].triggers[0];
    assert_eq!(
        trigger.function_name.as_deref(),
        Some("orders_insert_after_trg_fn")
    );
    assert!(
        trigger.body.is_none(),
        "body should be cleared after normalize"
    );
    let func = state
        .functions
        .get("orders_insert_after_trg_fn")
        .expect("function should exist");
    assert_eq!(func.returns, "trigger");
    assert_eq!(func.language, "plpgsql");
    assert_eq!(func.body, "BEGIN RETURN NEW; END;");
}

/// validate() rejects a trigger with no events.
#[test]
fn validate_trigger_empty_events() {
    let mut s = Schema::default();
    let table = Table {
        name: "users".to_string(),
        schema: None,
        columns: vec![],
        foreign_keys: vec![],
        indexes: vec![],
        constraints: vec![],
        triggers: vec![TriggerDef {
            name: Some("t".to_string()),
            timing: TriggerTiming::After,
            events: vec![],
            scope: TriggerScope::Row,
            function_name: Some("fn".to_string()),
            when: None,
            body: None,
            language: None,
        }],
    };
    s.tables.insert("users".to_string(), table);
    assert!(s.validate().unwrap_err().contains("no events"));
}

/// validate() rejects a trigger with no function_name after normalize.
#[test]
fn validate_trigger_no_function_name() {
    let mut s = Schema::default();
    let table = Table {
        name: "users".to_string(),
        schema: None,
        columns: vec![],
        foreign_keys: vec![],
        indexes: vec![],
        constraints: vec![],
        triggers: vec![TriggerDef {
            name: Some("t".to_string()),
            timing: TriggerTiming::After,
            events: vec![TriggerEvent::Insert],
            scope: TriggerScope::Row,
            function_name: None,
            when: None,
            body: None,
            language: None,
        }],
    };
    s.tables.insert("users".to_string(), table);
    assert!(s.validate().unwrap_err().contains("no function_name"));
}

/// Tables are stored in a BTreeMap so keys are always alphabetical regardless of creation order.
#[test]
fn tables_always_in_alphabetical_order() {
    let mut s = Schema::default();
    for name in &["zebra", "alpha", "mango", "bravo"] {
        apply_ok(
            &mut s,
            Operation::CreateTable {
                table: basic_table(name),
            },
        );
    }
    let keys: Vec<&str> = s.tables.keys().map(String::as_str).collect();
    assert_eq!(keys, vec!["alpha", "bravo", "mango", "zebra"]);
}

/// Columns are stored in a Vec so insertion order is always preserved.
#[test]
fn columns_preserve_insertion_order() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    for name in &["c", "a", "b"] {
        apply_ok(
            &mut s,
            Operation::AddColumn {
                table_name: "users".to_string(),
                column: text_col(name),
            },
        );
    }
    let names: Vec<&str> = s.tables["users"]
        .columns
        .iter()
        .map(|c| c.name.as_str())
        .collect();
    assert_eq!(names, vec!["c", "a", "b"]);
}

/// Dropping the middle column preserves surrounding columns in their original positions.
#[test]
fn drop_middle_column_preserves_order() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    for name in &["first", "middle", "last"] {
        apply_ok(
            &mut s,
            Operation::AddColumn {
                table_name: "users".to_string(),
                column: text_col(name),
            },
        );
    }
    apply_ok(
        &mut s,
        Operation::DropColumn {
            table_name: "users".to_string(),
            column: text_col("middle"),
            cascade: false,
        },
    );
    let names: Vec<&str> = s.tables["users"]
        .columns
        .iter()
        .map(|c| c.name.as_str())
        .collect();
    assert_eq!(names, vec!["first", "last"]);
}

/// CreateTable with inline columns produces the same state as CreateTable + AddColumn for each.
#[test]
fn create_table_inline_vs_incremental_are_equal() {
    let cols = vec![text_col("name"), text_col("email"), text_col("bio")];
    let mut s1 = Schema::default();
    apply_ok(
        &mut s1,
        Operation::CreateTable {
            table: Table {
                name: "users".to_string(),
                schema: None,
                columns: cols.clone(),
                foreign_keys: vec![],
                indexes: vec![],
                constraints: vec![],
                triggers: vec![],
            },
        },
    );

    let mut s2 = Schema::default();
    apply_ok(
        &mut s2,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    for col in cols {
        apply_ok(
            &mut s2,
            Operation::AddColumn {
                table_name: "users".to_string(),
                column: col,
            },
        );
    }

    assert_eq!(s1, s2);
}

/// Applying inverse(op) immediately after op restores the original state for all invertible ops.
#[test]
fn inverse_restores_state_for_all_invertible_ops() {
    let fk = ForeignKey {
        name: "fk_x".to_string(),
        from_column: "col".to_string(),
        to_table: "other".to_string(),
        to_column: "id".to_string(),
    };
    let idx = Index {
        name: "idx_col".to_string(),
        columns: vec!["col".to_string()],
        unique: true,
        predicate: None,
    };
    let chk = Constraint::Check {
        name: "chk_col".to_string(),
        expression: "col != ''".to_string(),
    };

    let verify = |mut s: Schema, op: Operation| {
        let before = s.clone();
        s.apply(&op).expect("op should apply");
        s.apply(&op.inverse().expect("should be invertible"))
            .expect("inverse should apply");
        assert_eq!(
            s,
            before,
            "inverse of '{}' did not restore state",
            op.type_name()
        );
    };

    verify(
        Schema::default(),
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );

    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    verify(
        s.clone(),
        Operation::AddColumn {
            table_name: "users".to_string(),
            column: text_col("col"),
        },
    );

    apply_ok(
        &mut s,
        Operation::AddColumn {
            table_name: "users".to_string(),
            column: text_col("col"),
        },
    );
    verify(
        s.clone(),
        Operation::AddForeignKey {
            table_name: "users".to_string(),
            foreign_key: fk,
        },
    );

    verify(
        s.clone(),
        Operation::AddIndex {
            table_name: "users".to_string(),
            index: idx,
            concurrent: false,
        },
    );

    verify(
        s.clone(),
        Operation::AddConstraint {
            table_name: "users".to_string(),
            constraint: chk,
        },
    );

    verify(
        s.clone(),
        Operation::RenameTable {
            old_name: "users".to_string(),
            new_name: "accounts".to_string(),
        },
    );

    verify(
        s.clone(),
        Operation::AlterColumn {
            table_name: "users".to_string(),
            old: text_col("col"),
            new: Column {
                name: "col".to_string(),
                col_type: "varchar(100)".to_string(),
                nullable: true,
                ..Default::default()
            },
            cast_expr: None,
        },
    );

    verify(
        s.clone(),
        Operation::RenameColumn {
            table_name: "users".to_string(),
            old_name: "col".to_string(),
            new_name: "alias".to_string(),
        },
    );
}

/// RenameTable followed by its inverse restores original state.
#[test]
fn rename_table_inverse_restores_state() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    let before = s.clone();
    let op = Operation::RenameTable {
        old_name: "users".to_string(),
        new_name: "accounts".to_string(),
    };
    apply_ok(&mut s, op.clone());
    apply_ok(&mut s, op.inverse().unwrap());
    assert_eq!(s, before);
}

/// RenameColumn followed by its inverse restores original state.
#[test]
fn rename_column_inverse_restores_state() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    apply_ok(
        &mut s,
        Operation::AddColumn {
            table_name: "users".to_string(),
            column: text_col("email"),
        },
    );
    let before = s.clone();
    let op = Operation::RenameColumn {
        table_name: "users".to_string(),
        old_name: "email".to_string(),
        new_name: "email_address".to_string(),
    };
    apply_ok(&mut s, op.clone());
    apply_ok(&mut s, op.inverse().unwrap());
    assert_eq!(s, before);
}

/// AlterColumn followed by its inverse restores original state.
#[test]
fn alter_column_inverse_restores_state() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    apply_ok(
        &mut s,
        Operation::AddColumn {
            table_name: "users".to_string(),
            column: text_col("bio"),
        },
    );
    let before = s.clone();
    let new_col = Column {
        name: "bio".to_string(),
        col_type: "varchar(500)".to_string(),
        nullable: true,
        ..Default::default()
    };
    let op = Operation::AlterColumn {
        table_name: "users".to_string(),
        old: text_col("bio"),
        new: new_col,
        cast_expr: None,
    };
    apply_ok(&mut s, op.clone());
    apply_ok(&mut s, op.inverse().unwrap());
    assert_eq!(s, before);
}

/// Replay is idempotent: applying the same operation sequence on two fresh states gives identical results.
#[test]
fn replay_is_idempotent() {
    let ops = vec![
        Operation::CreateTable {
            table: basic_table("users"),
        },
        Operation::AddColumn {
            table_name: "users".to_string(),
            column: text_col("email"),
        },
        Operation::CreateTable {
            table: basic_table("posts"),
        },
        Operation::AddColumn {
            table_name: "posts".to_string(),
            column: text_col("title"),
        },
        Operation::AddForeignKey {
            table_name: "posts".to_string(),
            foreign_key: ForeignKey {
                name: "fk_posts_user".to_string(),
                from_column: "user_id".to_string(),
                to_table: "users".to_string(),
                to_column: "id".to_string(),
            },
        },
    ];
    let mut s1 = Schema::default();
    let mut s2 = Schema::default();
    for op in &ops {
        s1.apply(op).unwrap();
        s2.apply(op).unwrap();
    }
    assert_eq!(s1, s2);
}

/// Full multi-operation replay produces the exact expected state.
#[test]
fn full_replay_produces_exact_state() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    apply_ok(
        &mut s,
        Operation::AddColumn {
            table_name: "users".to_string(),
            column: text_col("email"),
        },
    );
    apply_ok(
        &mut s,
        Operation::AddColumn {
            table_name: "users".to_string(),
            column: Column {
                name: "age".to_string(),
                col_type: "integer".to_string(),
                nullable: true,
                ..Default::default()
            },
        },
    );
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("posts"),
        },
    );
    apply_ok(
        &mut s,
        Operation::AddColumn {
            table_name: "posts".to_string(),
            column: text_col("title"),
        },
    );
    apply_ok(
        &mut s,
        Operation::AddColumn {
            table_name: "posts".to_string(),
            column: text_col("user_id"),
        },
    );
    apply_ok(
        &mut s,
        Operation::AddForeignKey {
            table_name: "posts".to_string(),
            foreign_key: ForeignKey {
                name: "fk_posts_user_id".to_string(),
                from_column: "user_id".to_string(),
                to_table: "users".to_string(),
                to_column: "id".to_string(),
            },
        },
    );
    apply_ok(
        &mut s,
        Operation::AddIndex {
            table_name: "users".to_string(),
            index: Index {
                name: "users_email_idx".to_string(),
                columns: vec!["email".to_string()],
                unique: true,
                predicate: None,
            },
            concurrent: false,
        },
    );
    apply_ok(
        &mut s,
        Operation::DropColumn {
            table_name: "users".to_string(),
            column: Column {
                name: "age".to_string(),
                col_type: "integer".to_string(),
                nullable: true,
                ..Default::default()
            },
            cascade: false,
        },
    );

    assert_eq!(s.tables.len(), 2);
    let users = &s.tables["users"];
    assert_eq!(users.columns.len(), 1);
    assert_eq!(users.columns[0].name, "email");
    assert_eq!(users.indexes.len(), 1);

    let posts = &s.tables["posts"];
    assert_eq!(
        posts
            .columns
            .iter()
            .map(|c| c.name.as_str())
            .collect::<Vec<_>>(),
        vec!["title", "user_id"]
    );
    assert_eq!(posts.foreign_keys[0].name, "fk_posts_user_id");
}

/// Statement and Invoke operations are transparent — they do not mutate existing state.
#[test]
fn statement_and_invoke_are_transparent_to_existing_state() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateTable {
            table: basic_table("users"),
        },
    );
    apply_ok(
        &mut s,
        Operation::AddColumn {
            table_name: "users".to_string(),
            column: text_col("email"),
        },
    );
    let before = s.clone();

    apply_ok(
        &mut s,
        Operation::Statement {
            up: "UPDATE users SET email = 'x'".to_string(),
            down: None,
        },
    );
    apply_ok(
        &mut s,
        Operation::Invoke {
            up: "./seed.sh".to_string(),
            down: None,
        },
    );

    assert_eq!(s, before);
}

/// schema_qualified_key: None and "public" both produce the bare table name.
#[test]
fn schema_qualified_key_public_and_none_are_bare() {
    assert_eq!(schema_qualified_key("users", None), "users");
    assert_eq!(schema_qualified_key("users", Some("public")), "users");
}

/// schema_qualified_key: any non-public schema produces "schema.name".
#[test]
fn schema_qualified_key_non_public_is_qualified() {
    assert_eq!(schema_qualified_key("users", Some("myapp")), "myapp.users");
    assert_eq!(
        schema_qualified_key("orders", Some("billing")),
        "billing.orders"
    );
}

/// CreateView and DropView use schema_qualified_key for their map key.
#[test]
fn create_view_key_respects_schema() {
    let mut s = Schema::default();
    apply_ok(
        &mut s,
        Operation::CreateView {
            view: ViewDef {
                name: "active_users".to_string(),
                schema: None,
                definition: "SELECT 1".to_string(),
            },
        },
    );
    apply_ok(
        &mut s,
        Operation::CreateView {
            view: ViewDef {
                name: "summary".to_string(),
                schema: Some("reporting".to_string()),
                definition: "SELECT 2".to_string(),
            },
        },
    );
    assert!(
        s.views.contains_key("active_users"),
        "public view uses bare key"
    );
    assert!(
        s.views.contains_key("reporting.summary"),
        "non-public view uses qualified key"
    );
}

/// Applying all operations in order then replaying the chain produces an identical state.
#[test]
fn sequential_replay_matches_single_pass() {
    let ops = vec![
        Operation::CreateTable {
            table: basic_table("users"),
        },
        Operation::AddColumn {
            table_name: "users".to_string(),
            column: text_col("email"),
        },
        Operation::AddConstraint {
            table_name: "users".to_string(),
            constraint: Constraint::Unique {
                name: "uq_email".to_string(),
                columns: vec!["email".to_string()],
            },
        },
        Operation::CreateTable {
            table: basic_table("orders"),
        },
        Operation::RenameTable {
            old_name: "orders".to_string(),
            new_name: "purchases".to_string(),
        },
    ];

    let apply_all = |ops: &Vec<Operation>| {
        let mut s = Schema::default();
        for op in ops {
            s.apply(op).unwrap();
        }
        s
    };

    let s1 = apply_all(&ops);
    let s2 = apply_all(&ops);
    assert_eq!(s1, s2);
    assert!(
        !s1.tables.contains_key("orders"),
        "renamed table should not exist under old name"
    );
    assert!(s1.tables.contains_key("purchases"));
    assert_eq!(s1.tables["users"].constraints[0].name(), "uq_email");
}

/// builder().table::<T>() inserts the table using schema_qualified_key.
#[test]
fn builder_table_via_into_table() {
    struct Posts;
    impl IntoTable for Posts {
        fn into_table(_dialect: &Dialect) -> Table {
            TableBuilder::new("posts")
                .id()
                .column("title", "text", |c| c.not_null())
                .build()
        }
    }

    let state = Schema::builder(Dialect::Postgres).table::<Posts>().build();
    assert!(state.tables.contains_key("posts"));
    assert_eq!(state.tables["posts"].columns.len(), 2);
    assert_eq!(state.tables["posts"].columns[0].name, "id");
    assert!(state.tables["posts"].columns[0].primary_key);
}

/// builder().extension() registers an extension without a version.
#[test]
fn builder_extension() {
    let state = Schema::builder(Dialect::Postgres)
        .extension("pgcrypto")
        .build();
    assert!(state.extensions.contains_key("pgcrypto"));
    assert!(state.extensions["pgcrypto"].version.is_none());
}

/// builder().enum_type() registers a named enum with the given values.
#[test]
fn builder_enum_type() {
    let state = Schema::builder(Dialect::Postgres)
        .enum_type("status", &["active", "inactive"])
        .build();
    assert!(state.enums.contains_key("status"));
    assert_eq!(state.enums["status"].values, vec!["active", "inactive"]);
}

/// qualified_name() returns plain name when schema is None.
#[test]
fn qualified_name_no_schema() {
    let t = basic_table("users");
    assert_eq!(t.qualified_name(), "users");
}

/// qualified_name() returns plain name when schema is "public".
#[test]
fn qualified_name_public_schema() {
    let t = Table {
        schema: Some("public".to_string()),
        ..basic_table("users")
    };
    assert_eq!(t.qualified_name(), "users");
}

/// qualified_name() returns schema.name for non-public schemas.
#[test]
fn qualified_name_custom_schema() {
    let t = Table {
        schema: Some("analytics".to_string()),
        ..basic_table("events")
    };
    assert_eq!(t.qualified_name(), "analytics.events");
}

/// qualified_name() works on all entity types with schema fields.
#[test]
fn qualified_name_all_entities() {
    let f = FunctionDef {
        name: "my_fn".to_string(),
        schema: Some("utils".to_string()),
        arguments: String::new(),
        returns: "void".to_string(),
        language: "sql".to_string(),
        body: String::new(),
        volatility: Volatility::Volatile,
        security_definer: false,
    };
    assert_eq!(f.qualified_name(), "utils.my_fn");

    let v = ViewDef {
        name: "v1".to_string(),
        schema: None,
        definition: "SELECT 1".to_string(),
    };
    assert_eq!(v.qualified_name(), "v1");

    let e = ExtensionDef {
        name: "pgcrypto".to_string(),
        schema: Some("public".to_string()),
        version: None,
    };
    assert_eq!(e.qualified_name(), "pgcrypto");

    let en = EnumDef {
        name: "status".to_string(),
        schema: Some("core".to_string()),
        values: vec![],
    };
    assert_eq!(en.qualified_name(), "core.status");
}

/// canonicalize normalizes schema: Some("public") to None for tables.
#[test]
fn canonicalize_public_schema_to_none() {
    let mut s = Schema::default();
    s.tables.insert(
        "users".to_string(),
        Table {
            schema: Some("public".to_string()),
            ..basic_table("users")
        },
    );
    s.canonicalize(&Dialect::Postgres);
    assert_eq!(s.tables["users"].schema, None);
}

/// canonicalize normalizes schema: Some("pg_catalog") to None for extensions.
#[test]
fn canonicalize_pg_catalog_extension() {
    let mut s = Schema::default();
    s.extensions.insert(
        "plpgsql".to_string(),
        ExtensionDef {
            name: "plpgsql".to_string(),
            schema: Some("pg_catalog".to_string()),
            version: None,
        },
    );
    s.canonicalize(&Dialect::Postgres);
    assert_eq!(s.extensions["plpgsql"].schema, None);
}

/// canonicalize does NOT normalize pg_catalog for non-extension entities.
#[test]
fn canonicalize_pg_catalog_not_for_tables() {
    let mut s = Schema::default();
    s.tables.insert(
        "pg_catalog.sometable".to_string(),
        Table {
            name: "sometable".to_string(),
            schema: Some("pg_catalog".to_string()),
            ..basic_table("sometable")
        },
    );
    s.canonicalize(&Dialect::Postgres);
    assert_eq!(
        s.tables["pg_catalog.sometable"].schema,
        Some("pg_catalog".to_string())
    );
}

/// canonicalize re-keys BTreeMaps when schema normalization changes the qualified name.
#[test]
fn canonicalize_rekeys_btreemap() {
    let mut s = Schema::default();
    s.enums.insert(
        "public.status".to_string(),
        EnumDef {
            name: "status".to_string(),
            schema: Some("public".to_string()),
            values: vec!["active".to_string()],
        },
    );
    s.canonicalize(&Dialect::Postgres);
    assert!(!s.enums.contains_key("public.status"));
    assert!(s.enums.contains_key("status"));
    assert_eq!(s.enums["status"].schema, None);
}

/// apply(CreateTable) with non-public schema inserts under qualified key.
#[test]
fn apply_create_table_qualified_key() {
    let mut s = Schema::default();
    let table = Table {
        name: "events".to_string(),
        schema: Some("analytics".to_string()),
        ..basic_table("events")
    };
    apply_ok(&mut s, Operation::CreateTable { table });
    assert!(s.tables.contains_key("analytics.events"));
    assert!(!s.tables.contains_key("events"));
}
