sqlite-utils/tests/test_transform.py

import sqlite3

import pytest

from sqlite_utils import ANY
from sqlite_utils.db import Check, ForeignKey, TransactionError, TransformError
from sqlite_utils.utils import OperationalError


@pytest.mark.parametrize(
    "params,expected_sql",
    [
        # Identity transform - nothing changes
        (
            {},
            [
                'CREATE TABLE "dogs_new_suffix" (\n   "id" INTEGER PRIMARY KEY,\n   "name" TEXT,\n   "age" TEXT\n);',
                'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n   SELECT "rowid", "id", "name", "age" FROM "dogs";',
                'DROP TABLE "dogs";',
                "PRAGMA legacy_alter_table=ON;",
                'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
                "PRAGMA legacy_alter_table=OFF;",
            ],
        ),
        # Change column type
        (
            {"types": {"age": int}},
            [
                'CREATE TABLE "dogs_new_suffix" (\n   "id" INTEGER PRIMARY KEY,\n   "name" TEXT,\n   "age" INTEGER\n);',
                'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n   SELECT "rowid", "id", "name", NULLIF("age", \'\') FROM "dogs";',
                'DROP TABLE "dogs";',
                "PRAGMA legacy_alter_table=ON;",
                'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
                "PRAGMA legacy_alter_table=OFF;",
            ],
        ),
        # Rename a column
        (
            {"rename": {"age": "dog_age"}},
            [
                'CREATE TABLE "dogs_new_suffix" (\n   "id" INTEGER PRIMARY KEY,\n   "name" TEXT,\n   "dog_age" TEXT\n);',
                'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "dog_age")\n   SELECT "rowid", "id", "name", "age" FROM "dogs";',
                'DROP TABLE "dogs";',
                "PRAGMA legacy_alter_table=ON;",
                'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
                "PRAGMA legacy_alter_table=OFF;",
            ],
        ),
        # Drop a column
        (
            {"drop": ["age"]},
            [
                'CREATE TABLE "dogs_new_suffix" (\n   "id" INTEGER PRIMARY KEY,\n   "name" TEXT\n);',
                'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name")\n   SELECT "rowid", "id", "name" FROM "dogs";',
                'DROP TABLE "dogs";',
                "PRAGMA legacy_alter_table=ON;",
                'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
                "PRAGMA legacy_alter_table=OFF;",
            ],
        ),
        # Convert type AND rename column
        (
            {"types": {"age": int}, "rename": {"age": "dog_age"}},
            [
                'CREATE TABLE "dogs_new_suffix" (\n   "id" INTEGER PRIMARY KEY,\n   "name" TEXT,\n   "dog_age" INTEGER\n);',
                'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "dog_age")\n   SELECT "rowid", "id", "name", NULLIF("age", \'\') FROM "dogs";',
                'DROP TABLE "dogs";',
                "PRAGMA legacy_alter_table=ON;",
                'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
                "PRAGMA legacy_alter_table=OFF;",
            ],
        ),
        # Change primary key
        (
            {"pk": "age"},
            [
                'CREATE TABLE "dogs_new_suffix" (\n   "id" INTEGER,\n   "name" TEXT,\n   "age" TEXT PRIMARY KEY\n);',
                'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n   SELECT "rowid", "id", "name", "age" FROM "dogs";',
                'DROP TABLE "dogs";',
                "PRAGMA legacy_alter_table=ON;",
                'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
                "PRAGMA legacy_alter_table=OFF;",
            ],
        ),
        # Change primary key to a compound pk
        (
            {"pk": ("age", "name")},
            [
                'CREATE TABLE "dogs_new_suffix" (\n   "id" INTEGER,\n   "name" TEXT,\n   "age" TEXT,\n   PRIMARY KEY ("age", "name")\n);',
                'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n   SELECT "rowid", "id", "name", "age" FROM "dogs";',
                'DROP TABLE "dogs";',
                "PRAGMA legacy_alter_table=ON;",
                'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
                "PRAGMA legacy_alter_table=OFF;",
            ],
        ),
        # Remove primary key, creating a rowid table
        (
            {"pk": None},
            [
                'CREATE TABLE "dogs_new_suffix" (\n   "id" INTEGER,\n   "name" TEXT,\n   "age" TEXT\n);',
                'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n   SELECT "rowid", "id", "name", "age" FROM "dogs";',
                'DROP TABLE "dogs";',
                "PRAGMA legacy_alter_table=ON;",
                'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
                "PRAGMA legacy_alter_table=OFF;",
            ],
        ),
        # Keeping the table
        (
            {"drop": ["age"], "keep_table": "kept_table"},
            [
                'CREATE TABLE "dogs_new_suffix" (\n   "id" INTEGER PRIMARY KEY,\n   "name" TEXT\n);',
                'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name")\n   SELECT "rowid", "id", "name" FROM "dogs";',
                "PRAGMA legacy_alter_table=ON;",
                'ALTER TABLE "dogs" RENAME TO "kept_table";',
                'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
                "PRAGMA legacy_alter_table=OFF;",
            ],
        ),
    ],
)
@pytest.mark.parametrize("use_pragma_foreign_keys", [False, True])
def test_transform_sql_table_with_primary_key(
    fresh_db, params, expected_sql, use_pragma_foreign_keys
):
    captured = []

    def tracer(sql, params):
        return captured.append((sql, params))

    dogs = fresh_db.table("dogs")
    if use_pragma_foreign_keys:
        fresh_db.conn.execute("PRAGMA foreign_keys=ON")
    dogs.insert({"id": 1, "name": "Cleo", "age": "5"}, pk="id")
    sql = dogs.transform_sql(**{**params, "tmp_suffix": "suffix"})
    assert sql == expected_sql
    # Check that .transform() runs without exceptions:
    with fresh_db.tracer(tracer):
        dogs.transform(**params)
    # If use_pragma_foreign_keys, check that we did the right thing
    if use_pragma_foreign_keys:
        assert ("PRAGMA foreign_keys=0;", None) in captured
        assert captured[-2] == ("PRAGMA foreign_key_check;", None)
        assert captured[-1] == ("PRAGMA foreign_keys=1;", None)
    else:
        assert ("PRAGMA foreign_keys=0;", None) not in captured
        assert ("PRAGMA foreign_keys=1;", None) not in captured


@pytest.mark.parametrize(
    "params,expected_sql",
    [
        # Identity transform - nothing changes
        (
            {},
            [
                'CREATE TABLE "dogs_new_suffix" (\n   "id" INTEGER,\n   "name" TEXT,\n   "age" TEXT\n);',
                'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n   SELECT "rowid", "id", "name", "age" FROM "dogs";',
                'DROP TABLE "dogs";',
                "PRAGMA legacy_alter_table=ON;",
                'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
                "PRAGMA legacy_alter_table=OFF;",
            ],
        ),
        # Change column type
        (
            {"types": {"age": int}},
            [
                'CREATE TABLE "dogs_new_suffix" (\n   "id" INTEGER,\n   "name" TEXT,\n   "age" INTEGER\n);',
                'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n   SELECT "rowid", "id", "name", NULLIF("age", \'\') FROM "dogs";',
                'DROP TABLE "dogs";',
                "PRAGMA legacy_alter_table=ON;",
                'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
                "PRAGMA legacy_alter_table=OFF;",
            ],
        ),
        # Rename a column
        (
            {"rename": {"age": "dog_age"}},
            [
                'CREATE TABLE "dogs_new_suffix" (\n   "id" INTEGER,\n   "name" TEXT,\n   "dog_age" TEXT\n);',
                'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "dog_age")\n   SELECT "rowid", "id", "name", "age" FROM "dogs";',
                'DROP TABLE "dogs";',
                "PRAGMA legacy_alter_table=ON;",
                'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
                "PRAGMA legacy_alter_table=OFF;",
            ],
        ),
        # Make ID a primary key
        (
            {"pk": "id"},
            [
                'CREATE TABLE "dogs_new_suffix" (\n   "id" INTEGER PRIMARY KEY,\n   "name" TEXT,\n   "age" TEXT\n);',
                'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n   SELECT "rowid", "id", "name", "age" FROM "dogs";',
                'DROP TABLE "dogs";',
                "PRAGMA legacy_alter_table=ON;",
                'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
                "PRAGMA legacy_alter_table=OFF;",
            ],
        ),
    ],
)
@pytest.mark.parametrize("use_pragma_foreign_keys", [False, True])
def test_transform_sql_table_with_no_primary_key(
    fresh_db, params, expected_sql, use_pragma_foreign_keys
):
    captured = []

    def tracer(sql, params):
        return captured.append((sql, params))

    dogs = fresh_db.table("dogs")
    if use_pragma_foreign_keys:
        fresh_db.conn.execute("PRAGMA foreign_keys=ON")
    dogs.insert({"id": 1, "name": "Cleo", "age": "5"})
    sql = dogs.transform_sql(**{**params, "tmp_suffix": "suffix"})
    assert sql == expected_sql
    # Check that .transform() runs without exceptions:
    with fresh_db.tracer(tracer):
        dogs.transform(**params)
    # If use_pragma_foreign_keys, check that we did the right thing
    if use_pragma_foreign_keys:
        assert ("PRAGMA foreign_keys=0;", None) in captured
        assert captured[-2] == ("PRAGMA foreign_key_check;", None)
        assert captured[-1] == ("PRAGMA foreign_keys=1;", None)
    else:
        assert ("PRAGMA foreign_keys=0;", None) not in captured
        assert ("PRAGMA foreign_keys=1;", None) not in captured


def test_transform_sql_with_no_primary_key_to_primary_key_of_id(fresh_db):
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo", "age": "5"})
    assert (
        dogs.schema
        == 'CREATE TABLE "dogs" (\n   "id" INTEGER,\n   "name" TEXT,\n   "age" TEXT\n)'
    )
    dogs.transform(pk="id")
    # Slight oddity: [dogs] becomes "dogs" during the rename:
    assert (
        dogs.schema
        == 'CREATE TABLE "dogs" (\n   "id" INTEGER PRIMARY KEY,\n   "name" TEXT,\n   "age" TEXT\n)'
    )


def test_transform_rename_pk(fresh_db):
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo", "age": "5"}, pk="id")
    dogs.transform(rename={"id": "pk"})
    assert (
        dogs.schema
        == 'CREATE TABLE "dogs" (\n   "pk" INTEGER PRIMARY KEY,\n   "name" TEXT,\n   "age" TEXT\n)'
    )


def test_transform_rename_cannot_lose_data_by_colliding_columns(fresh_db):
    # Renaming one column onto the name of another used to silently collapse
    # the two columns into one, discarding the source column's data entirely.
    table = fresh_db.table("items")
    table.insert({"a": 1, "b": 2})
    with pytest.raises(TransformError, match="duplicate column name"):
        table.transform(rename={"b": "a"})
    # The failed transform must leave the table and its data untouched.
    assert table.count == 1
    assert [(r["a"], r["b"]) for r in table.rows_where(order_by="rowid")] == [(1, 2)]
    assert '"b"' in table.schema


def test_transform_rename_two_columns_to_same_name_errors(fresh_db):
    table = fresh_db.table("items")
    table.insert({"a": 1, "b": 2, "c": 3})
    with pytest.raises(TransformError, match="duplicate column name 'x'"):
        table.transform(rename={"b": "x", "c": "x"})


def test_transform_can_swap_two_column_names(fresh_db):
    # A straight swap a<->b is legal and must not be blocked.
    table = fresh_db.table("items")
    table.insert({"a": 1, "b": 2})
    table.transform(rename={"a": "b", "b": "a"})
    assert [(r["a"], r["b"]) for r in table.rows_where(order_by="rowid")] == [(2, 1)]


def test_transform_preserves_keyword_literal_defaults(fresh_db):
    # transform() used to requote keyword-literal defaults (DEFAULT TRUE became
    # DEFAULT 'TRUE'), so a default insert stored the text 'TRUE' instead of the
    # integer 1 -- silent value corruption on every rebuilt table.
    fresh_db.execute(
        "CREATE TABLE t ("
        " id INTEGER PRIMARY KEY,"
        " is_active INTEGER DEFAULT TRUE,"
        " flag INTEGER DEFAULT FALSE,"
        " note TEXT DEFAULT NULL"
        ")"
    )
    table = fresh_db.table("t")
    table.insert({"id": 1})
    before = fresh_db.execute("SELECT is_active, flag, note FROM t").fetchone()
    assert before == (1, 0, None)

    # Rebuild the table via an unrelated change.
    table.transform(rename={"note": "note2"})

    # The keyword literals stay unquoted in the schema ...
    assert "DEFAULT TRUE" in table.schema
    assert "DEFAULT FALSE" in table.schema
    assert "DEFAULT NULL" in table.schema
    assert "'TRUE'" not in table.schema

    # ... and a fresh default insert still yields 1 / 0 / NULL, not strings.
    table.insert({"id": 2})
    after = fresh_db.execute(
        "SELECT is_active, flag, note2 FROM t WHERE id = 2"
    ).fetchone()
    assert after == (1, 0, None)


def test_transform_not_null(fresh_db):
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo", "age": "5"}, pk="id")
    dogs.transform(not_null={"name"})
    assert (
        dogs.schema
        == 'CREATE TABLE "dogs" (\n   "id" INTEGER PRIMARY KEY,\n   "name" TEXT NOT NULL,\n   "age" TEXT\n)'
    )


def test_transform_remove_a_not_null(fresh_db):
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo", "age": "5"}, not_null={"age"}, pk="id")
    dogs.transform(not_null={"name": True, "age": False})
    assert (
        dogs.schema
        == 'CREATE TABLE "dogs" (\n   "id" INTEGER PRIMARY KEY,\n   "name" TEXT NOT NULL,\n   "age" TEXT\n)'
    )


@pytest.mark.parametrize("not_null", [{"age"}, {"age": True}])
def test_transform_add_not_null_with_rename(fresh_db, not_null):
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo", "age": "5"}, pk="id")
    dogs.transform(not_null=not_null, rename={"age": "dog_age"})
    assert (
        dogs.schema
        == 'CREATE TABLE "dogs" (\n   "id" INTEGER PRIMARY KEY,\n   "name" TEXT,\n   "dog_age" TEXT NOT NULL\n)'
    )


def test_transform_defaults(fresh_db):
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo", "age": 5}, pk="id")
    dogs.transform(defaults={"age": 1})
    assert (
        dogs.schema
        == 'CREATE TABLE "dogs" (\n   "id" INTEGER PRIMARY KEY,\n   "name" TEXT,\n   "age" INTEGER DEFAULT 1\n)'
    )


def test_transform_defaults_and_rename_column(fresh_db):
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo", "age": 5}, pk="id")
    dogs.transform(rename={"age": "dog_age"}, defaults={"age": 1})
    assert (
        dogs.schema
        == 'CREATE TABLE "dogs" (\n   "id" INTEGER PRIMARY KEY,\n   "name" TEXT,\n   "dog_age" INTEGER DEFAULT 1\n)'
    )


def test_remove_defaults(fresh_db):
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo", "age": 5}, defaults={"age": 1}, pk="id")
    dogs.transform(defaults={"age": None})
    assert (
        dogs.schema
        == 'CREATE TABLE "dogs" (\n   "id" INTEGER PRIMARY KEY,\n   "name" TEXT,\n   "age" INTEGER\n)'
    )


@pytest.fixture
def authors_db(fresh_db):
    books = fresh_db.table("books")
    authors = fresh_db.table("authors")
    authors.insert({"id": 5, "name": "Jane McGonical"}, pk="id")
    books.insert(
        {"id": 2, "title": "Reality is Broken", "author_id": 5},
        foreign_keys=("author_id",),
        pk="id",
    )
    return fresh_db


def test_transform_foreign_keys_persist(authors_db):
    assert authors_db.table("books").foreign_keys == [
        ForeignKey(
            table="books", column="author_id", other_table="authors", other_column="id"
        )
    ]
    authors_db.table("books").transform(rename={"title": "book_title"})
    assert authors_db.table("books").foreign_keys == [
        ForeignKey(
            table="books", column="author_id", other_table="authors", other_column="id"
        )
    ]


@pytest.mark.parametrize("use_pragma_foreign_keys", [False, True])
def test_transform_foreign_keys_survive_renamed_column(
    authors_db, use_pragma_foreign_keys
):
    if use_pragma_foreign_keys:
        authors_db.conn.execute("PRAGMA foreign_keys=ON")
    authors_db.table("books").transform(rename={"author_id": "author_id_2"})
    assert authors_db.table("books").foreign_keys == [
        ForeignKey(
            table="books",
            column="author_id_2",
            other_table="authors",
            other_column="id",
        )
    ]


def _add_country_city_continent(db):
    db.table("country").insert({"id": 1, "name": "France"}, pk="id")
    db.table("continent").insert({"id": 2, "name": "Europe"}, pk="id")
    db.table("city").insert({"id": 24, "name": "Paris"}, pk="id")


_CAVEAU = {
    "id": 32,
    "name": "Caveau de la Huchette",
    "country": 1,
    "continent": 2,
    "city": 24,
}


@pytest.mark.parametrize("use_pragma_foreign_keys", [False, True])
def test_transform_drop_foreign_keys(fresh_db, use_pragma_foreign_keys):
    if use_pragma_foreign_keys:
        fresh_db.conn.execute("PRAGMA foreign_keys=ON")
    # Create table with three foreign keys so we can drop two of them
    _add_country_city_continent(fresh_db)
    fresh_db.table("places").insert(
        _CAVEAU,
        foreign_keys=("country", "continent", "city"),
    )
    assert fresh_db.table("places").foreign_keys == [
        ForeignKey(
            table="places", column="city", other_table="city", other_column="id"
        ),
        ForeignKey(
            table="places",
            column="continent",
            other_table="continent",
            other_column="id",
        ),
        ForeignKey(
            table="places", column="country", other_table="country", other_column="id"
        ),
    ]
    # Drop two of those foreign keys
    fresh_db.table("places").transform(drop_foreign_keys=("country", "continent"))
    # Should be only one foreign key now
    assert fresh_db.table("places").foreign_keys == [
        ForeignKey(table="places", column="city", other_table="city", other_column="id")
    ]
    if use_pragma_foreign_keys:
        assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0]


def test_transform_verify_foreign_keys(fresh_db):
    fresh_db.conn.execute("PRAGMA foreign_keys=ON")
    fresh_db.table("authors").insert({"id": 3, "name": "Tina"}, pk="id")
    fresh_db.table("books").insert(
        {"id": 1, "title": "Book", "author_id": 3}, pk="id", foreign_keys={"author_id"}
    )
    # Renaming the id column on authors should break everything
    with pytest.raises(OperationalError) as e:
        fresh_db.table("authors").transform(rename={"id": "id2"})
    assert e.value.args[0] == 'foreign key mismatch - "books" referencing "authors"'
    # This should have rolled us back
    assert (
        fresh_db.table("authors").schema
        == 'CREATE TABLE "authors" (\n   "id" INTEGER PRIMARY KEY,\n   "name" TEXT\n)'
    )
    assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0]


@pytest.mark.parametrize("use_pragma_foreign_keys", [False, True])
def test_transform_on_delete_cascade_does_not_delete_records(
    fresh_db, use_pragma_foreign_keys
):
    # Transforming a table drops and recreates it - if another table references
    # it with ON DELETE CASCADE and PRAGMA foreign_keys is on, that drop must
    # not cascade and delete the referencing records
    if use_pragma_foreign_keys:
        fresh_db.conn.execute("PRAGMA foreign_keys=ON")
    fresh_db.executescript("""
        CREATE TABLE authors (id INTEGER PRIMARY KEY, name TEXT);
        CREATE TABLE books (
            id INTEGER PRIMARY KEY,
            title TEXT,
            author_id INTEGER REFERENCES authors(id) ON DELETE CASCADE
        );
        """)
    fresh_db.table("authors").insert({"id": 1, "name": "Ursula K. Le Guin"})
    fresh_db.table("books").insert(
        {"id": 1, "title": "The Dispossessed", "author_id": 1}
    )
    # Transform the table on the other end of the cascading foreign key
    fresh_db.table("authors").transform(rename={"name": "author_name"})
    assert list(fresh_db.table("authors").rows) == [
        {"id": 1, "author_name": "Ursula K. Le Guin"}
    ]
    assert list(fresh_db.table("books").rows) == [
        {"id": 1, "title": "The Dispossessed", "author_id": 1}
    ]
    # Transforming the table with the cascading foreign key should not
    # delete its records either
    fresh_db.table("books").transform(rename={"title": "book_title"})
    assert list(fresh_db.table("books").rows) == [
        {"id": 1, "book_title": "The Dispossessed", "author_id": 1}
    ]
    if use_pragma_foreign_keys:
        assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0]


@pytest.mark.parametrize("on_delete", ["CASCADE", "SET NULL", "SET DEFAULT", "cascade"])
def test_transform_in_transaction_refuses_destructive_on_delete(fresh_db, on_delete):
    # PRAGMA foreign_keys is a no-op inside a transaction, so transforming a
    # table referenced by ON DELETE CASCADE / SET NULL / SET DEFAULT foreign
    # keys inside an open transaction would fire those actions when the old
    # table is dropped - transform() should refuse instead
    fresh_db.conn.execute("PRAGMA foreign_keys=ON")
    fresh_db.executescript(f"""
        CREATE TABLE authors (id INTEGER PRIMARY KEY, name TEXT);
        CREATE TABLE books (
            id INTEGER PRIMARY KEY,
            title TEXT,
            author_id INTEGER REFERENCES authors(id) ON DELETE {on_delete}
        );
        """)
    fresh_db.table("authors").insert({"id": 1, "name": "Ursula K. Le Guin"})
    fresh_db.table("books").insert(
        {"id": 1, "title": "The Dispossessed", "author_id": 1}
    )
    previous_schema = fresh_db.table("authors").schema
    with fresh_db.atomic(), pytest.raises(TransactionError) as excinfo:
        fresh_db.table("authors").transform(rename={"name": "author_name"})
    message = str(excinfo.value)
    assert "books" in message
    assert f"ON DELETE {on_delete.upper()}" in message
    # Nothing should have changed
    assert fresh_db.table("authors").schema == previous_schema
    assert list(fresh_db.table("books").rows) == [
        {"id": 1, "title": "The Dispossessed", "author_id": 1}
    ]
    assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0]


def test_transform_in_transaction_refuses_self_referential_cascade(fresh_db):
    # The copied table carries a foreign key referencing the original table
    # name, so a self-referential cascade would wipe the copy too
    fresh_db.conn.execute("PRAGMA foreign_keys=ON")
    fresh_db.executescript("""
        CREATE TABLE categories (
            id INTEGER PRIMARY KEY,
            name TEXT,
            parent_id INTEGER REFERENCES categories(id) ON DELETE CASCADE
        );
        """)
    fresh_db.table("categories").insert_all(
        [
            {"id": 1, "name": "Fiction", "parent_id": None},
            {"id": 2, "name": "Science Fiction", "parent_id": 1},
        ]
    )
    with fresh_db.atomic(), pytest.raises(TransactionError) as excinfo:
        fresh_db.table("categories").transform(rename={"name": "title"})
    assert "categories" in str(excinfo.value)
    assert fresh_db.table("categories").count == 2


def test_transform_in_transaction_allowed_with_no_action_foreign_key(fresh_db):
    # An inbound foreign key without a destructive ON DELETE action is safe
    # inside a transaction thanks to PRAGMA defer_foreign_keys
    fresh_db.conn.execute("PRAGMA foreign_keys=ON")
    fresh_db.executescript("""
        CREATE TABLE authors (id INTEGER PRIMARY KEY, name TEXT);
        CREATE TABLE books (
            id INTEGER PRIMARY KEY,
            title TEXT,
            author_id INTEGER REFERENCES authors(id)
        );
        """)
    fresh_db.table("authors").insert({"id": 1, "name": "Ursula K. Le Guin"})
    fresh_db.table("books").insert(
        {"id": 1, "title": "The Dispossessed", "author_id": 1}
    )
    with fresh_db.atomic():
        fresh_db.table("authors").transform(rename={"name": "author_name"})
    assert list(fresh_db.table("authors").rows) == [
        {"id": 1, "author_name": "Ursula K. Le Guin"}
    ]
    assert list(fresh_db.table("books").rows) == [
        {"id": 1, "title": "The Dispossessed", "author_id": 1}
    ]
    assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0]


def test_transform_in_transaction_allowed_for_child_table(fresh_db):
    # The table being transformed only has an outbound foreign key - dropping
    # it fires no ON DELETE actions, so this is allowed inside a transaction
    fresh_db.conn.execute("PRAGMA foreign_keys=ON")
    fresh_db.executescript("""
        CREATE TABLE authors (id INTEGER PRIMARY KEY, name TEXT);
        CREATE TABLE books (
            id INTEGER PRIMARY KEY,
            title TEXT,
            author_id INTEGER REFERENCES authors(id) ON DELETE CASCADE
        );
        """)
    fresh_db.table("authors").insert({"id": 1, "name": "Ursula K. Le Guin"})
    fresh_db.table("books").insert(
        {"id": 1, "title": "The Dispossessed", "author_id": 1}
    )
    with fresh_db.atomic():
        fresh_db.table("books").transform(rename={"title": "book_title"})
    assert list(fresh_db.table("books").rows) == [
        {"id": 1, "book_title": "The Dispossessed", "author_id": 1}
    ]


def test_transform_in_transaction_allowed_with_foreign_keys_off(fresh_db):
    # With PRAGMA foreign_keys off (the default) no cascades can fire, so
    # transform inside a transaction is safe even with a CASCADE schema
    fresh_db.executescript("""
        CREATE TABLE authors (id INTEGER PRIMARY KEY, name TEXT);
        CREATE TABLE books (
            id INTEGER PRIMARY KEY,
            title TEXT,
            author_id INTEGER REFERENCES authors(id) ON DELETE CASCADE
        );
        """)
    fresh_db.table("authors").insert({"id": 1, "name": "Ursula K. Le Guin"})
    fresh_db.table("books").insert(
        {"id": 1, "title": "The Dispossessed", "author_id": 1}
    )
    with fresh_db.atomic():
        fresh_db.table("authors").transform(rename={"name": "author_name"})
    assert list(fresh_db.table("books").rows) == [
        {"id": 1, "title": "The Dispossessed", "author_id": 1}
    ]


def test_transform_add_foreign_keys_from_scratch(fresh_db):
    _add_country_city_continent(fresh_db)
    fresh_db.table("places").insert(_CAVEAU)
    # Should have no foreign keys
    assert fresh_db.table("places").foreign_keys == []
    # Now add them using .transform()
    fresh_db.table("places").transform(
        add_foreign_keys=("country", "continent", "city")
    )
    # Should now have all three:
    assert fresh_db.table("places").foreign_keys == [
        ForeignKey(
            table="places", column="city", other_table="city", other_column="id"
        ),
        ForeignKey(
            table="places",
            column="continent",
            other_table="continent",
            other_column="id",
        ),
        ForeignKey(
            table="places", column="country", other_table="country", other_column="id"
        ),
    ]
    assert fresh_db.table("places").schema == (
        'CREATE TABLE "places" (\n'
        '   "id" INTEGER,\n'
        '   "name" TEXT,\n'
        '   "country" INTEGER REFERENCES "country"("id"),\n'
        '   "continent" INTEGER REFERENCES "continent"("id"),\n'
        '   "city" INTEGER REFERENCES "city"("id")\n'
        ")"
    )


@pytest.mark.parametrize(
    "add_foreign_keys",
    (
        ("country", "continent"),
        # Fully specified
        (
            ("country", "country", "id"),
            ("continent", "continent", "id"),
        ),
    ),
)
def test_transform_add_foreign_keys_from_partial(fresh_db, add_foreign_keys):
    _add_country_city_continent(fresh_db)
    fresh_db.table("places").insert(
        _CAVEAU,
        foreign_keys=("city",),
    )
    # Should have one foreign keys
    assert fresh_db.table("places").foreign_keys == [
        ForeignKey(table="places", column="city", other_table="city", other_column="id")
    ]
    # Now add three more using .transform()
    fresh_db.table("places").transform(add_foreign_keys=add_foreign_keys)
    # Should now have all three:
    assert fresh_db.table("places").foreign_keys == [
        ForeignKey(
            table="places", column="city", other_table="city", other_column="id"
        ),
        ForeignKey(
            table="places",
            column="continent",
            other_table="continent",
            other_column="id",
        ),
        ForeignKey(
            table="places", column="country", other_table="country", other_column="id"
        ),
    ]


@pytest.mark.parametrize(
    "foreign_keys",
    (
        ("country", "continent"),
        # Fully specified
        (
            ("country", "country", "id"),
            ("continent", "continent", "id"),
        ),
    ),
)
def test_transform_replace_foreign_keys(fresh_db, foreign_keys):
    _add_country_city_continent(fresh_db)
    fresh_db.table("places").insert(
        _CAVEAU,
        foreign_keys=("city",),
    )
    assert len(fresh_db.table("places").foreign_keys) == 1
    # Replace with two different ones
    fresh_db.table("places").transform(foreign_keys=foreign_keys)
    assert fresh_db.table("places").schema == (
        'CREATE TABLE "places" (\n'
        '   "id" INTEGER,\n'
        '   "name" TEXT,\n'
        '   "country" INTEGER REFERENCES "country"("id"),\n'
        '   "continent" INTEGER REFERENCES "continent"("id"),\n'
        '   "city" INTEGER\n'
        ")"
    )


@pytest.mark.parametrize("table_type", ("id_pk", "rowid", "compound_pk"))
def test_transform_preserves_rowids(fresh_db, table_type):
    pk = None
    if table_type == "id_pk":
        pk = "id"
    elif table_type == "compound_pk":
        pk = ("id", "name")
    elif table_type == "rowid":
        pk = None
    fresh_db.table("places").insert_all(
        [
            {"id": "1", "name": "Paris", "country": "France"},
            {"id": "2", "name": "London", "country": "UK"},
            {"id": "3", "name": "New York", "country": "USA"},
        ],
        pk=pk,
    )
    # Now delete and insert a row to mix up the `rowid` sequence
    fresh_db.table("places").delete_where("id = ?", ["2"])
    fresh_db.table("places").insert({"id": "4", "name": "London", "country": "UK"})
    previous_rows = [
        tuple(row) for row in fresh_db.execute("select rowid, id, name from places")
    ]
    # Transform it
    fresh_db.table("places").transform(column_order=("country", "name"))
    # Should be the same
    next_rows = [
        tuple(row) for row in fresh_db.execute("select rowid, id, name from places")
    ]
    assert previous_rows == next_rows


@pytest.mark.parametrize(
    "initial_strict,transform_strict,expected_strict",
    (
        (False, None, False),
        (True, None, True),
        (False, True, True),
        (True, False, False),
    ),
)
def test_transform_strict(fresh_db, initial_strict, transform_strict, expected_strict):
    if not fresh_db.supports_strict:
        pytest.skip("SQLite version does not support strict tables")
    dogs = fresh_db.table("dogs", strict=initial_strict)
    dogs.insert({"id": 1, "name": "Cleo"})
    assert dogs.strict is initial_strict
    dogs.transform(strict=transform_strict)
    assert dogs.strict is expected_strict


def test_transform_to_strict_with_invalid_data(fresh_db):
    if not fresh_db.supports_strict:
        pytest.skip("SQLite version does not support strict tables")
    dogs = fresh_db.table("dogs")
    dogs.create({"id": int})
    dogs.insert({"id": "not-an-integer"})

    with pytest.raises(sqlite3.IntegrityError):
        dogs.transform(strict=True)

    assert dogs.strict is False
    assert list(dogs.rows) == [{"id": "not-an-integer"}]
    assert fresh_db.table_names() == ["dogs"]


def test_transform_strict_updates_default(fresh_db):
    if not fresh_db.supports_strict:
        pytest.skip("SQLite version does not support strict tables")
    table = fresh_db.table("items", strict=True)
    table.create({"id": int})

    table.transform(strict=False)
    assert table.strict is False

    table.create({"id": int}, replace=True)
    assert table.strict is False


@pytest.mark.parametrize("method_name", ("transform", "transform_sql"))
def test_transform_to_strict_not_supported(fresh_db, method_name):
    table = fresh_db.table("items")
    table.create({"id": int})
    fresh_db._supports_strict = False

    with pytest.raises(TransformError, match="SQLite does not support STRICT tables"):
        getattr(table, method_name)(strict=True)

    assert table.strict is False


def test_transform_preserves_any_column_in_strict_table(fresh_db):
    if not fresh_db.supports_strict:
        pytest.skip("SQLite version does not support strict tables")
    fresh_db.execute("create table items (id integer primary key, data any) strict")
    fresh_db.conn.executemany(
        "insert into items values (?, ?)",
        [
            (1, 42),
            (2, "000123"),
            (3, 3.14),
            (4, b"bytes"),
            (5, None),
        ],
    )
    table = fresh_db["items"]

    table.transform()

    assert table.strict is True
    assert table.columns_dict == {"id": int, "data": ANY}
    assert fresh_db.execute(
        "select typeof(data), data from items order by id"
    ).fetchall() == [
        ("integer", 42),
        ("text", "000123"),
        ("real", 3.14),
        ("blob", b"bytes"),
        ("null", None),
    ]


def test_transform_any_column_from_strict_to_non_strict(fresh_db):
    if not fresh_db.supports_strict:
        pytest.skip("SQLite version does not support strict tables")
    fresh_db.execute("create table items (data any) strict")
    fresh_db.execute("insert into items values (?)", ("000123",))
    table = fresh_db["items"]

    table.transform(strict=False)

    assert table.strict is False
    assert table.columns_dict == {"data": ANY}
    # Ordinary non-STRICT ANY columns apply NUMERIC affinity
    assert fresh_db.execute("select typeof(data), data from items").fetchone() == (
        "integer",
        123,
    )


@pytest.mark.parametrize(
    "indexes, transform_params",
    [
        ([["name"]], {"types": {"age": str}}),
        ([["name"], ["age", "breed"]], {"types": {"age": str}}),
        ([], {"types": {"age": str}}),
        ([["name"]], {"types": {"age": str}, "keep_table": "old_dogs"}),
    ],
)
def test_transform_indexes(fresh_db, indexes, transform_params):
    # https://github.com/simonw/sqlite-utils/issues/633
    # New table should have same indexes as old table after transformation
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo", "age": 5, "breed": "Labrador"}, pk="id")

    for index in indexes:
        dogs.create_index(index)

    indexes_before_transform = dogs.indexes

    dogs.transform(**transform_params)

    assert sorted(
        [
            {k: v for k, v in idx._asdict().items() if k != "seq"}
            for idx in dogs.indexes
        ],
        key=lambda x: x["name"],
    ) == sorted(
        [
            {k: v for k, v in idx._asdict().items() if k != "seq"}
            for idx in indexes_before_transform
        ],
        key=lambda x: x["name"],
    ), f"Indexes before transform: {indexes_before_transform}\nIndexes after transform: {dogs.indexes}"
    if "keep_table" in transform_params:
        assert all(
            index.origin == "pk"
            for index in fresh_db.table(transform_params["keep_table"]).indexes
        )


def test_transform_retains_indexes_with_foreign_keys(fresh_db):
    dogs = fresh_db.table("dogs")
    owners = fresh_db.table("owners")

    dogs.insert({"id": 1, "name": "Cleo", "owner_id": 1}, pk="id")
    owners.insert({"id": 1, "name": "Alice"}, pk="id")

    dogs.create_index(["name"])

    indexes_before_transform = dogs.indexes

    fresh_db.add_foreign_keys([("dogs", "owner_id", "owners", "id")])  # calls transform

    assert sorted(
        [
            {k: v for k, v in idx._asdict().items() if k != "seq"}
            for idx in dogs.indexes
        ],
        key=lambda x: x["name"],
    ) == sorted(
        [
            {k: v for k, v in idx._asdict().items() if k != "seq"}
            for idx in indexes_before_transform
        ],
        key=lambda x: x["name"],
    ), f"Indexes before transform: {indexes_before_transform}\nIndexes after transform: {dogs.indexes}"


def test_transform_with_indexes_errors(fresh_db):
    # Should error with a compound (name, age) index if age is dropped
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo", "age": 5}, pk="id")

    dogs.create_index(["name", "age"])

    with pytest.raises(TransformError) as excinfo:
        dogs.transform(drop=["age"])

    assert (
        "Index 'idx_dogs_name_age' column 'age' is not in updated table 'dogs'. "
        "You must manually drop this index prior to running this transformation"
        in str(excinfo.value)
    )


@pytest.mark.parametrize(
    ("table_name", "index_name"),
    (("name", "idx_name"), ("t", "name")),
)
def test_transform_rename_column_with_index(fresh_db, table_name, index_name):
    # https://github.com/simonw/sqlite-utils/issues/822
    # Use the same name for the table, column and index to ensure only the
    # indexed column changes.
    table = fresh_db.table(table_name)
    table.insert({"id": 1, "name": "Cleo"}, pk="id")
    table.create_index(["name"], index_name=index_name)

    sqls = table.transform_sql(rename={"name": "full_name"}, tmp_suffix="suffix")
    drop_index_sql = f'DROP INDEX IF EXISTS "{index_name}";'
    assert drop_index_sql in sqls
    assert sqls.index(drop_index_sql) < sqls.index(f'DROP TABLE "{table_name}";')

    table.transform(rename={"name": "full_name"})

    assert [column.name for column in table.columns] == ["id", "full_name"]
    assert [(index.name, index.columns) for index in table.indexes] == [
        (index_name, ["full_name"])
    ]


def test_transform_recreates_renamed_index_from_metadata(fresh_db):
    table = fresh_db.table("t")
    table.insert({"alpha": "one", "beta": "two"})
    # Deliberately use unquoted SQL and index details that need to survive the
    # reconstruction. Renaming both columns also guards against cascading
    # string substitutions.
    fresh_db.execute(
        "CREATE UNIQUE INDEX swap_idx ON t(alpha COLLATE NOCASE DESC, beta)"
    )

    table.transform(rename={"alpha": "beta", "beta": "alpha"})

    assert table.columns_dict == {"beta": str, "alpha": str}
    assert [(index.name, index.unique, index.columns) for index in table.indexes] == [
        ("swap_idx", 1, ["beta", "alpha"])
    ]
    key_columns = [column for column in table.xindexes[0].columns if column.key]
    assert [(column.name, column.desc, column.coll) for column in key_columns] == [
        ("beta", 1, "NOCASE"),
        ("alpha", 0, "BINARY"),
    ]


@pytest.mark.parametrize(
    "index_sql",
    (
        "CREATE INDEX idx_t_name ON t(lower(name))",
        "CREATE INDEX idx_t_name ON t(name) WHERE name IS NOT NULL",
    ),
)
def test_transform_rename_complex_index_errors(fresh_db, index_sql):
    table = fresh_db.table("t")
    table.insert({"id": 1, "name": "Cleo"}, pk="id")
    fresh_db.execute(index_sql)

    with pytest.raises(TransformError, match="partial or expression index"):
        table.transform(rename={"name": "full_name"})

    assert table.columns_dict == {"id": int, "name": str}
    assert [index.name for index in table.indexes] == ["idx_t_name"]


def test_transform_with_unique_constraint_implicit_index(fresh_db):
    dogs = fresh_db.table("dogs")
    # Create a table with a UNIQUE constraint on 'name', which creates an implicit index
    fresh_db.execute("""
        CREATE TABLE dogs (
            id INTEGER PRIMARY KEY,
            name TEXT UNIQUE ON CONFLICT IGNORE,
            age INTEGER
        );
    """)
    dogs.insert({"id": 1, "name": "Cleo", "age": 5})

    dogs.transform(types={"age": str}, rename={"name": "dog_name"})

    assert 'dog_name" TEXT UNIQUE ON CONFLICT IGNORE' in dogs.schema
    dogs.insert({"id": 2, "dog_name": "Cleo", "age": "6"})
    assert list(dogs.rows) == [{"id": 1, "dog_name": "Cleo", "age": "5"}]


def test_transform_preserves_composite_unique_constraint(fresh_db):
    fresh_db.execute("""
        CREATE TABLE memberships (
            account_id INTEGER,
            email TEXT,
            note TEXT,
            CONSTRAINT unique_membership
                UNIQUE (account_id DESC, email COLLATE NOCASE)
                ON CONFLICT ABORT
        )
    """)
    memberships = fresh_db.table("memberships")
    memberships.insert({"account_id": 1, "email": "one@example.com", "note": "x"})

    memberships.transform(rename={"account_id": "organization_id"}, types={"note": str})

    assert (
        'CONSTRAINT "unique_membership" UNIQUE '
        '("organization_id" DESC, "email" COLLATE "NOCASE") ON CONFLICT ABORT'
        in memberships.schema
    )
    with pytest.raises(sqlite3.IntegrityError):
        memberships.insert(
            {"organization_id": 1, "email": "ONE@example.com", "note": "y"}
        )


def test_transform_preserves_column_unique_collation(fresh_db):
    fresh_db.execute("""
        CREATE TABLE people (
            id INTEGER PRIMARY KEY,
            name TEXT COLLATE NOCASE UNIQUE
        )
    """)
    people = fresh_db.table("people")
    people.insert({"id": 1, "name": "Cleo"})

    people.transform(rename={"name": "full_name"})

    assert 'UNIQUE ("full_name" COLLATE "NOCASE")' in people.schema
    with pytest.raises(sqlite3.IntegrityError):
        people.insert({"id": 2, "full_name": "cleo"})


def test_transform_drops_entire_composite_unique_constraint(fresh_db):
    fresh_db.execute("""
        CREATE TABLE memberships (
            account_id INTEGER,
            email TEXT,
            UNIQUE (account_id, email)
        )
    """)
    memberships = fresh_db.table("memberships")
    memberships.insert({"account_id": 1, "email": "one@example.com"})

    memberships.transform(drop={"email"})

    assert "UNIQUE" not in memberships.schema
    memberships.insert({"account_id": 1})


def test_transform_preserves_autoincrement_and_sequence(fresh_db):
    fresh_db.execute(
        "CREATE TABLE entries (id INTEGER PRIMARY KEY AUTOINCREMENT, value TEXT)"
    )
    entries = fresh_db.table("entries")
    entries.insert_all(({"value": "one"}, {"value": "two"}))
    entries.delete(2)

    entries.transform(rename={"value": "label"})

    assert "PRIMARY KEY AUTOINCREMENT" in entries.schema
    entries.insert({"label": "three"})
    assert list(entries.rows) == [
        {"id": 1, "label": "one"},
        {"id": 3, "label": "three"},
    ]


@pytest.mark.parametrize(
    "new_type,expected_value,expected_type",
    [
        (int, 42, int),
        (float, 42.0, float),
        ("integer", 42, int),
        ("float", 42.0, float),
        ("REAL", 42.0, float),
    ],
)
def test_transform_empty_string_to_null_for_numeric_types(
    fresh_db, new_type, expected_value, expected_type
):
    fresh_db["test"].insert_all(
        [
            {"id": 1, "value": "42"},
            {"id": 2, "value": ""},
            {"id": 3, "value": None},
            {"id": 4, "value": " "},
        ]
    )
    fresh_db["test"].transform(types={"value": new_type})
    rows = {r["id"]: r["value"] for r in fresh_db["test"].rows}
    assert rows[1] == expected_value
    assert type(rows[1]) is expected_type
    assert rows[2] is None
    assert rows[3] is None
    assert rows[4] == " "


def test_transform_preserves_view(fresh_db):
    # https://github.com/simonw/sqlite-utils/issues/831
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo"}, pk="id")
    fresh_db.execute("create view dogs_view as select id, name from dogs")
    view_sql_before = fresh_db.execute(
        "select sql from sqlite_master where name = 'dogs_view'"
    ).fetchone()[0]
    dogs.transform(rename={"name": "title"})
    view_sql_after = fresh_db.execute(
        "select sql from sqlite_master where name = 'dogs_view'"
    ).fetchone()[0]
    assert view_sql_before == view_sql_after


@pytest.mark.parametrize(
    "transform_params",
    [
        {"types": {"name": int}},
        {"pk": "name"},
        {"add_foreign_keys": [("other_id", "other", "id")]},
        {"drop_foreign_keys": ["other_id"]},
    ],
)
def test_transform_variants_preserve_view(fresh_db, transform_params):
    # Covers retyping, changing primary key and foreign key modifications,
    # with a view whose columns are untouched by the transform
    fresh_db.table("other").insert({"id": 1}, pk="id")
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo", "other_id": 1}, pk="id")
    if "drop_foreign_keys" in transform_params:
        dogs.transform(add_foreign_keys=[("other_id", "other", "id")])
    fresh_db.execute("create view dogs_view as select id, name from dogs")
    view_sql_before = fresh_db.execute(
        "select sql from sqlite_master where name = 'dogs_view'"
    ).fetchone()[0]
    dogs.transform(**transform_params)
    view_sql_after = fresh_db.execute(
        "select sql from sqlite_master where name = 'dogs_view'"
    ).fetchone()[0]
    assert view_sql_before == view_sql_after
    assert list(fresh_db.view("dogs_view").rows) == [{"id": 1, "name": "Cleo"}]


def test_transform_view_referencing_renamed_column(fresh_db):
    # The view survives but querying it raises "no such column" - inherent
    # to SQLite views, whose SQL is stored as text
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo"}, pk="id")
    fresh_db.execute("create view dogs_view as select id, name from dogs")
    dogs.transform(rename={"name": "title"})
    with pytest.raises(OperationalError, match="no such column"):
        fresh_db.execute("select * from dogs_view")


def test_transform_view_on_view(fresh_db):
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo"}, pk="id")
    fresh_db.execute("create view v1 as select id, name from dogs")
    fresh_db.execute("create view v2 as select name from v1")
    sqls_before = fresh_db.execute(
        "select sql from sqlite_master where type = 'view' order by name"
    ).fetchall()
    dogs.transform(types={"id": str})
    sqls_after = fresh_db.execute(
        "select sql from sqlite_master where type = 'view' order by name"
    ).fetchall()
    assert sqls_before == sqls_after
    assert list(fresh_db.view("v2").rows) == [{"name": "Cleo"}]


def test_transform_keep_table_does_not_repoint_view(fresh_db):
    # Without legacy_alter_table the ALTER TABLE dogs RENAME TO dogs_backup
    # step would rewrite the view to select from "dogs_backup"
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo"}, pk="id")
    fresh_db.execute("create view dogs_view as select id, name from dogs")
    dogs.transform(types={"name": str}, keep_table="dogs_backup")
    view_sql = fresh_db.execute(
        "select sql from sqlite_master where name = 'dogs_view'"
    ).fetchone()[0]
    assert "dogs_backup" not in view_sql
    # View reads from the live table, not the frozen backup
    dogs.insert({"id": 2, "name": "Pancakes"})
    assert list(fresh_db.view("dogs_view").rows) == [
        {"id": 1, "name": "Cleo"},
        {"id": 2, "name": "Pancakes"},
    ]


def test_transform_sql_standalone_statements_work_with_view(fresh_db):
    # The documented "run these statements yourself" workflow should be
    # standalone-correct, so the pragmas must come from transform_sql()
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo"}, pk="id")
    fresh_db.execute("create view dogs_view as select id, name from dogs")
    sqls = dogs.transform_sql(types={"name": str}, tmp_suffix="suffix")
    assert sqls[-3] == "PRAGMA legacy_alter_table=ON;"
    assert sqls[-2] == 'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";'
    assert sqls[-1] == "PRAGMA legacy_alter_table=OFF;"
    for sql in sqls:
        fresh_db.execute(sql)
    assert list(fresh_db.view("dogs_view").rows) == [{"id": 1, "name": "Cleo"}]


def test_transform_with_view_in_open_transaction(fresh_db):
    fresh_db.conn.execute("PRAGMA foreign_keys=ON")
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo"}, pk="id")
    fresh_db.execute("create view dogs_view as select id, name from dogs")
    with fresh_db.conn:
        fresh_db.execute("insert into dogs (id, name) values (2, 'Pancakes')")
        dogs.transform(rename={"name": "title"})
    assert dogs.columns_dict == {"id": int, "title": str}
    view_sql = fresh_db.execute(
        "select sql from sqlite_master where name = 'dogs_view'"
    ).fetchone()[0]
    assert view_sql == "CREATE VIEW dogs_view as select id, name from dogs"


def test_transform_restores_legacy_alter_table_setting(fresh_db):
    if sqlite3.sqlite_version_info < (3, 25, 0):
        pytest.skip("legacy_alter_table pragma requires SQLite 3.25 or higher")
    dogs = fresh_db.table("dogs")
    dogs.insert({"id": 1, "name": "Cleo"}, pk="id")
    # Default is OFF, reset to OFF afterwards
    dogs.transform(types={"name": str})
    assert fresh_db.execute("PRAGMA legacy_alter_table").fetchone()[0] == 0
    # If the connection has it ON, it should be restored to ON
    fresh_db.execute("PRAGMA legacy_alter_table=ON")
    sqls = dogs.transform_sql(types={"name": str}, tmp_suffix="suffix")
    assert sqls[-1] == "PRAGMA legacy_alter_table=ON;"
    dogs.transform(types={"name": str})
    assert fresh_db.execute("PRAGMA legacy_alter_table").fetchone()[0] == 1


def test_transform_preserves_check_constraints(fresh_db):
    fresh_db.execute("""
        CREATE TABLE scores (
            id INTEGER PRIMARY KEY,
            score INTEGER CONSTRAINT valid_score CHECK(score BETWEEN 0 AND 100),
            CONSTRAINT nonzero_id CHECK(id != 0)
        )
    """)
    scores = fresh_db.table("scores")
    scores.insert({"id": 1, "score": 50})
    scores.transform()
    assert scores.checks == [
        Check("score BETWEEN 0 AND 100", name="valid_score", column="score"),
        Check("id != 0", name="nonzero_id"),
    ]
    with pytest.raises(sqlite3.IntegrityError, match="CHECK constraint failed"):
        scores.insert({"id": 2, "score": 101})


def test_transform_preserves_check_ending_in_line_comment(fresh_db):
    fresh_db.execute("""
        CREATE TABLE inventory (
            quantity INTEGER,
            CHECK (
                quantity >= 0 -- Quantity cannot be negative
            )
        )
    """)
    inventory = fresh_db.table("inventory")
    inventory.transform(types={"quantity": float})
    assert inventory.checks == [Check("quantity >= 0 -- Quantity cannot be negative")]
    with pytest.raises(sqlite3.IntegrityError, match="CHECK constraint failed"):
        inventory.insert({"quantity": -1})


def test_transform_preserves_comments_owned_by_columns(fresh_db):
    fresh_db.execute("""
        CREATE TABLE people (
            -- Primary identifier
            id INTEGER PRIMARY KEY /* IDs are stable */,
            /* Displayed to users */
            name TEXT /* May contain spaces */,
            -- Age in years
            age INTEGER -- May be NULL
        )
    """)
    people = fresh_db.table("people")
    people.insert({"id": 1, "name": "Cleo", "age": 5})
    people.transform(
        rename={"name": "display_name"},
        types={"age": float},
        column_order=("age", "id", "name"),
    )
    assert people.get(1) == {"age": 5.0, "id": 1, "display_name": "Cleo"}
    schema = people.schema
    assert schema.index("-- Age in years") < schema.index('"age" REAL')
    assert schema.index('"age" REAL') < schema.index("-- May be NULL")
    assert schema.index("-- Primary identifier") < schema.index('"id" INTEGER')
    assert schema.index('"id" INTEGER') < schema.index("/* IDs are stable */")
    assert schema.index("/* Displayed to users */") < schema.index(
        '"display_name" TEXT'
    )
    assert schema.index('"display_name" TEXT') < schema.index(
        "/* May contain spaces */"
    )


def test_transform_drops_comments_owned_by_dropped_column(fresh_db):
    fresh_db.execute("""
        CREATE TABLE t (
            /* Keep this explanation */
            id INTEGER,
            /* Drop this explanation */
            obsolete TEXT /* Drop this too */
        )
    """)
    fresh_db.table("t").transform(drop={"obsolete"})
    schema = fresh_db.table("t").schema
    assert "Keep this explanation" in schema
    assert "Drop this explanation" not in schema
    assert "Drop this too" not in schema


def test_transform_renames_columns_inside_check_constraints(fresh_db):
    fresh_db.execute("""
        CREATE TABLE inventory (
            quantity INTEGER CONSTRAINT positive
                CHECK(quantity > 0 AND 'quantity' != ''),
            maximum INTEGER,
            CONSTRAINT within_maximum CHECK(quantity <= maximum)
        )
    """)
    inventory = fresh_db.table("inventory")
    inventory.insert({"quantity": 2, "maximum": 3})
    inventory.transform(rename={"quantity": "amount"})
    assert inventory.checks == [
        Check(
            "amount > 0 AND 'quantity' != ''",
            name="positive",
            column="amount",
        ),
        Check("amount <= maximum", name="within_maximum"),
    ]
    with pytest.raises(sqlite3.IntegrityError, match="CHECK constraint failed"):
        inventory.insert({"amount": 4, "maximum": 3})


def test_transform_check_rewrite_preserves_functions_and_quotes(fresh_db):
    fresh_db.execute("""
        CREATE TABLE items (
            length TEXT,
            "old name" TEXT,
            CHECK(length("old name") > 0 AND length != '')
        )
    """)
    items = fresh_db.table("items")
    items.insert({"length": "label", "old name": "hello"})
    items.transform(rename={"length": "description", "old name": "new name"})
    assert items.checks == [Check("length(\"new name\") > 0 AND description != ''")]


def test_transform_check_rewrite_quotes_keyword_column(fresh_db):
    fresh_db.execute("CREATE TABLE t(old_name TEXT CHECK(old_name != ''))")
    fresh_db.table("t").insert({"old_name": "value"})
    fresh_db.table("t").transform(rename={"old_name": "select"})
    assert fresh_db.table("t").checks == [Check("\"select\" != ''", column="select")]


def test_transform_check_rewrite_does_not_rename_collations_or_cast_types(fresh_db):
    fresh_db.execute("""
        CREATE TABLE t (
            nocase TEXT,
            kind TEXT,
            other TEXT,
            CHECK(
                other COLLATE nocase != ''
                AND CAST(other AS kind) != ''
                AND nocase != ''
                AND kind != ''
            )
        )
    """)
    fresh_db.table("t").insert({"nocase": "n", "kind": "k", "other": "o"})
    fresh_db.table("t").transform(rename={"nocase": "label", "kind": "category"})
    check = fresh_db.table("t").checks[0].check
    assert "COLLATE nocase" in check
    assert "AS kind" in check
    assert "AND label != ''" in check
    assert "AND category != ''" in check


def test_transform_drops_check_owned_by_dropped_column(fresh_db):
    fresh_db.execute("""
        CREATE TABLE t (
            id INTEGER,
            obsolete INTEGER CHECK(obsolete > 0),
            CHECK(id > 0)
        )
    """)
    fresh_db.table("t").insert({"id": 1, "obsolete": 2})
    fresh_db.table("t").transform(drop={"obsolete"})
    assert fresh_db.table("t").checks == [Check("id > 0")]


def test_transform_refuses_to_drop_column_used_by_remaining_check(fresh_db):
    fresh_db.execute("""
        CREATE TABLE ranges (
            minimum INTEGER,
            maximum INTEGER,
            CHECK(minimum <= maximum)
        )
    """)
    ranges = fresh_db.table("ranges")
    ranges.insert({"minimum": 1, "maximum": 2})
    schema_before = ranges.schema
    with pytest.raises(
        TransformError,
        match="Cannot drop column 'maximum'.*CHECK constraint",
    ):
        ranges.transform(drop={"maximum"})
    assert ranges.schema == schema_before