remote/sqlite-utils
1
0
Fork 0
mirror of https://github.com/simonw/sqlite-utils.git synced 2026-07-24 01:44:31 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge branch 'main' into say-5/document-memory-attrs

Browse source
This commit is contained in:
Simon Willison 2026-07-07 19:35:27 -07:00 • committed by GitHub
commit eee6c49547
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
62 changed files with 7227 additions and 816 deletions
Download patch file Download diff file Expand all files Collapse all files

31
tests/conftest.py
View file

@ -8,11 +8,36 @@ create table Gosh2 (c1 text, c2 text, c3 text);
"""
def pytest_addoption(parser):
parser.addoption(
"--sqlite-autocommit",
action="store_true",
default=False,
help=(
"Run every test against connections created with the Python 3.12+ "
"sqlite3.connect(autocommit=True) mode"
),
)
def pytest_configure(config):
import sys
sys._called_from_test = True # type: ignore[attr-defined]
if config.getoption("--sqlite-autocommit"):
if sys.version_info < (3, 12):
raise pytest.UsageError(
"--sqlite-autocommit requires Python 3.12 or higher"
)
real_connect = sqlite3.connect
def autocommit_connect(*args, **kwargs):
kwargs.setdefault("autocommit", True)
return real_connect(*args, **kwargs)
sqlite3.connect = autocommit_connect
@pytest.fixture(autouse=True)
def close_all_databases():
@ -42,14 +67,12 @@ def fresh_db():
@pytest.fixture
def existing_db():
database = Database(memory=True)
database.executescript(
"""
database.executescript("""
CREATE TABLE foo (text TEXT);
INSERT INTO foo (text) values ("one");
INSERT INTO foo (text) values ("two");
INSERT INTO foo (text) values ("three");
"""
)
""")
return database

9
tests/test_analyze_tables.py
View file

@ -143,10 +143,7 @@ def db_to_analyze_path(db_to_analyze, tmpdir):
def test_analyze_table(db_to_analyze_path):
result = CliRunner().invoke(cli.cli, ["analyze-tables", db_to_analyze_path])
assert (
result.output.strip()
== (
"""
assert result.output.strip() == ("""
stuff.id: (1/3)
Total rows: 8
@ -179,9 +176,7 @@ stuff.size: (3/3)
Most common:
5: 5
3: 4"""
).strip()
)
3: 4""").strip()
def test_analyze_table_save(db_to_analyze_path):

382
tests/test_atomic.py Normal file
View file

@ -0,0 +1,382 @@
import pytest
from sqlite_utils.db import Database, _iter_complete_sql_statements
from sqlite_utils.utils import sqlite3
@pytest.mark.parametrize(
"sql,expected",
(
(
"CREATE TABLE t(id); INSERT INTO t VALUES (1)",
["CREATE TABLE t(id);", "INSERT INTO t VALUES (1)"],
),
(
"INSERT INTO t VALUES ('a;b');",
["INSERT INTO t VALUES ('a;b');"],
),
(
"-- comment;\nCREATE TABLE t(id);",
["-- comment;\nCREATE TABLE t(id);"],
),
(
"""
CREATE TRIGGER t_ai AFTER INSERT ON t
BEGIN
UPDATE t SET value = 'a;b' WHERE id = new.id;
INSERT INTO log VALUES ('x;y');
END;
""",
[
"CREATE TRIGGER t_ai AFTER INSERT ON t\n"
" BEGIN\n"
" UPDATE t SET value = 'a;b' WHERE id = new.id;\n"
" INSERT INTO log VALUES ('x;y');\n"
" END;"
],
),
),
)
def test_iter_complete_sql_statements(sql, expected):
assert list(_iter_complete_sql_statements(sql)) == expected
def test_atomic_commits(fresh_db):
with fresh_db.atomic():
fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
assert list(fresh_db["dogs"].rows) == [{"id": 1, "name": "Cleo"}]
def test_atomic_rolls_back(fresh_db):
with pytest.raises(RuntimeError):
with fresh_db.atomic():
fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
raise RuntimeError("boom")
assert not fresh_db["dogs"].exists()
def test_nested_atomic_rolls_back_to_savepoint(fresh_db):
fresh_db["dogs"].create({"id": int, "name": str}, pk="id")
with fresh_db.atomic():
fresh_db["dogs"].insert({"id": 1, "name": "Cleo"})
with pytest.raises(RuntimeError):
with fresh_db.atomic():
fresh_db["dogs"].insert({"id": 2, "name": "Pancakes"})
raise RuntimeError("boom")
fresh_db["dogs"].insert({"id": 3, "name": "Marnie"})
assert list(fresh_db["dogs"].rows) == [
{"id": 1, "name": "Cleo"},
{"id": 3, "name": "Marnie"},
]
def test_outer_atomic_rolls_back_released_savepoint(fresh_db):
with pytest.raises(RuntimeError):
with fresh_db.atomic():
fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
with fresh_db.atomic():
fresh_db["dogs"].insert({"id": 2, "name": "Pancakes"})
raise RuntimeError("boom")
assert not fresh_db["dogs"].exists()
def test_executescript_does_not_commit_open_atomic_block(fresh_db):
with pytest.raises(RuntimeError):
with fresh_db.atomic():
fresh_db.executescript("""
CREATE TABLE dogs(id INTEGER PRIMARY KEY, name TEXT);
CREATE TRIGGER dogs_ai AFTER INSERT ON dogs
BEGIN
UPDATE dogs SET name = upper(new.name) || '; updated' WHERE id = new.id;
END;
-- This comment has a semicolon;
INSERT INTO dogs VALUES (1, 'Cleo; the first');
""")
raise RuntimeError("boom")
assert not fresh_db["dogs"].exists()
def test_transform_does_not_commit_open_atomic_block(fresh_db):
fresh_db["dogs"].insert({"id": 1, "name": "Cleo", "age": "5"}, pk="id")
with pytest.raises(RuntimeError):
with fresh_db.atomic():
fresh_db["dogs"].insert({"id": 2, "name": "Pancakes", "age": "6"})
fresh_db["dogs"].transform(rename={"age": "dog_age"})
raise RuntimeError("boom")
assert (
fresh_db["dogs"].schema
== 'CREATE TABLE "dogs" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT,\n "age" TEXT\n)'
)
assert list(fresh_db["dogs"].rows) == [
{"id": 1, "name": "Cleo", "age": "5"},
]
def test_transform_parent_table_with_foreign_keys_in_atomic(fresh_db):
fresh_db.conn.execute("PRAGMA foreign_keys=ON")
fresh_db["authors"].insert({"id": 1, "name": "Tina"}, pk="id")
fresh_db["books"].insert(
{"id": 1, "title": "Book", "author_id": 1},
pk="id",
foreign_keys={"author_id"},
)
with fresh_db.atomic():
fresh_db["authors"].transform(rename={"name": "full_name"})
assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0]
assert (
fresh_db["authors"].schema
== 'CREATE TABLE "authors" (\n "id" INTEGER PRIMARY KEY,\n "full_name" TEXT\n)'
)
assert fresh_db.execute("PRAGMA foreign_key_check").fetchall() == []
def test_transform_parent_table_with_foreign_keys_rolls_back(fresh_db):
fresh_db.conn.execute("PRAGMA foreign_keys=ON")
fresh_db["authors"].insert({"id": 1, "name": "Tina"}, pk="id")
fresh_db["books"].insert(
{"id": 1, "title": "Book", "author_id": 1},
pk="id",
foreign_keys={"author_id"},
)
with pytest.raises(RuntimeError):
with fresh_db.atomic():
fresh_db["authors"].transform(rename={"name": "full_name"})
raise RuntimeError("boom")
assert (
fresh_db["authors"].schema
== 'CREATE TABLE "authors" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT\n)'
)
assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0]
assert fresh_db.execute("PRAGMA foreign_key_check").fetchall() == []
def test_transform_detects_foreign_key_check_violations(fresh_db):
fresh_db.conn.execute("PRAGMA foreign_keys=ON")
fresh_db["authors"].insert({"id": 1, "name": "Tina"}, pk="id")
fresh_db["books"].insert({"id": 1, "author_id": 2}, pk="id")
with pytest.raises(sqlite3.IntegrityError):
fresh_db["books"].transform(add_foreign_keys=(("author_id", "authors", "id"),))
assert fresh_db["books"].foreign_keys == []
assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0]
def test_atomic_inside_manual_transaction_uses_savepoint(fresh_db):
fresh_db["t"].insert({"id": 1}, pk="id")
fresh_db.execute("begin")
with fresh_db.atomic():
fresh_db["t"].insert({"id": 2}, pk="id")
# Nothing is committed until the user's own transaction commits
assert fresh_db.conn.in_transaction
fresh_db.rollback()
assert [r["id"] for r in fresh_db["t"].rows] == [1]
# And with a commit instead, the atomic block's writes persist
fresh_db.execute("begin")
with fresh_db.atomic():
fresh_db["t"].insert({"id": 3}, pk="id")
fresh_db.commit()
assert [r["id"] for r in fresh_db["t"].rows] == [1, 3]
def test_begin_commit_rollback(tmpdir):
path = str(tmpdir / "test.db")
db = Database(path)
db["t"].insert({"id": 1}, pk="id")
db.begin()
db["t"].insert({"id": 2}, pk="id")
assert db.conn.in_transaction
db.rollback()
assert not db.conn.in_transaction
assert [r["id"] for r in db["t"].rows] == [1]
db.begin()
db["t"].insert({"id": 3}, pk="id")
db.commit()
db.close()
db2 = Database(path)
assert [r["id"] for r in db2["t"].rows] == [1, 3]
db2.close()
def test_begin_inside_transaction_errors(fresh_db):
fresh_db.begin()
with pytest.raises(sqlite3.OperationalError):
fresh_db.begin()
fresh_db.rollback()
def test_commit_and_rollback_without_transaction_are_noops(fresh_db):
fresh_db.commit()
fresh_db.rollback()
assert not fresh_db.conn.in_transaction
def test_execute_write_commits_immediately(tmpdir):
path = str(tmpdir / "test.db")
db = Database(path)
db["t"].insert({"id": 1}, pk="id")
db.execute("insert into t (id) values (2)")
# No implicit transaction is left open
assert not db.conn.in_transaction
# A completely separate connection sees the row straight away
other = sqlite3.connect(path)
assert other.execute("select count(*) from t").fetchone()[0] == 2
other.close()
db.close()
def test_execute_write_respects_explicit_transaction(fresh_db):
fresh_db["t"].insert({"id": 1}, pk="id")
fresh_db.begin()
fresh_db.execute("insert into t (id) values (2)")
# Still inside the explicit transaction - not committed
assert fresh_db.conn.in_transaction
fresh_db.rollback()
assert [r["id"] for r in fresh_db["t"].rows] == [1]
def test_execute_comment_prefixed_begin_leaves_transaction_open(fresh_db):
# A BEGIN hidden behind a leading comment must not be auto-committed
# out from under the caller
fresh_db["t"].insert({"id": 1}, pk="id")
fresh_db.execute("-- start a transaction\nbegin")
assert fresh_db.conn.in_transaction
fresh_db.execute("insert into t (id) values (2)")
fresh_db.rollback()
assert [r["id"] for r in fresh_db["t"].rows] == [1]
def _sqlite_accepts_bom():
try:
sqlite3.connect(":memory:").execute("\ufeffselect 1")
return True
except sqlite3.OperationalError:
return False
@pytest.mark.parametrize("begin_sql", ["; begin", "\ufeffbegin"])
def test_execute_prefixed_begin_leaves_transaction_open(fresh_db, begin_sql):
# sqlite3 tolerates empty statements and a UTF-8 BOM before the first
# real token, so a BEGIN behind either must not be auto-committed
# out from under the caller
if begin_sql.startswith("\ufeff") and not _sqlite_accepts_bom():
pytest.skip("This SQLite version rejects a leading byte order mark")
fresh_db["t"].insert({"id": 1}, pk="id")
fresh_db.execute(begin_sql)
assert fresh_db.conn.in_transaction
fresh_db.execute("insert into t (id) values (2)")
fresh_db.rollback()
assert [r["id"] for r in fresh_db["t"].rows] == [1]
def test_execute_failed_write_rolls_back_implicit_transaction(tmpdir):
# A failed write must not leave the driver's implicit transaction open -
# that would silently disable auto-commit for every subsequent write
path = str(tmpdir / "test.db")
db = Database(path)
db["t"].insert({"id": 1}, pk="id")
with pytest.raises(sqlite3.IntegrityError):
db.execute("insert into t (id) values (1)")
assert not db.conn.in_transaction
# Subsequent writes commit as normal and survive closing the connection
db["other"].insert({"id": 2})
db.close()
db2 = Database(path)
assert db2["other"].exists()
db2.close()
def test_execute_failed_write_preserves_explicit_transaction(fresh_db):
# A failed write inside an explicit transaction must not roll back
# the caller's earlier work - only the caller decides that
fresh_db["t"].insert({"id": 1}, pk="id")
fresh_db.begin()
fresh_db.execute("insert into t (id) values (2)")
with pytest.raises(sqlite3.IntegrityError):
fresh_db.execute("insert into t (id) values (1)")
assert fresh_db.conn.in_transaction
fresh_db.commit()
assert [r["id"] for r in fresh_db["t"].rows] == [1, 2]
def test_execute_failed_write_inside_atomic_preserves_block(fresh_db):
# A caught failure inside an atomic() block must leave the block's
# transaction open so its other work still commits
fresh_db["t"].insert({"id": 1}, pk="id")
with fresh_db.atomic():
fresh_db.execute("insert into t (id) values (2)")
with pytest.raises(sqlite3.IntegrityError):
fresh_db.execute("insert into t (id) values (1)")
assert [r["id"] for r in fresh_db["t"].rows] == [1, 2]
def test_query_returning_commits_after_iteration(tmpdir):
if sqlite3.sqlite_version_info < (3, 35, 0):
import pytest as _pytest
_pytest.skip("RETURNING requires SQLite 3.35.0 or higher")
path = str(tmpdir / "test.db")
db = Database(path)
db["t"].insert({"id": 1}, pk="id")
rows = list(db.query("insert into t (id) values (2) returning id"))
assert rows == [{"id": 2}]
assert not db.conn.in_transaction
other = sqlite3.connect(path)
assert other.execute("select count(*) from t").fetchone()[0] == 2
other.close()
db.close()
TRIGGER_SQL = """
create trigger no_bad before insert on t
when new.v = 'bad'
begin
select raise(rollback, 'trigger says no');
end
"""
def test_atomic_preserves_error_from_transaction_destroying_trigger(fresh_db):
# RAISE(ROLLBACK) rolls back the whole transaction and destroys every
# savepoint - atomic()'s cleanup must not mask the IntegrityError
# with "cannot rollback - no transaction is active"
fresh_db.execute("create table t (id integer primary key, v text)")
fresh_db.execute(TRIGGER_SQL)
with pytest.raises(sqlite3.IntegrityError, match="trigger says no"):
with fresh_db.atomic():
fresh_db.execute("insert into t (v) values ('bad')")
assert not fresh_db.conn.in_transaction
def test_nested_atomic_preserves_error_from_transaction_destroying_trigger(
fresh_db,
):
# The nested savepoint branch previously raised
# "no such savepoint" from ROLLBACK TO SAVEPOINT
fresh_db.execute("create table t (id integer primary key, v text)")
fresh_db.execute(TRIGGER_SQL)
with pytest.raises(sqlite3.IntegrityError, match="trigger says no"):
with fresh_db.atomic():
with fresh_db.atomic():
fresh_db.execute("insert into t (v) values ('bad')")
assert not fresh_db.conn.in_transaction
def test_atomic_preserves_error_from_insert_or_rollback(fresh_db):
fresh_db["t"].insert({"id": 1}, pk="id")
with pytest.raises(sqlite3.IntegrityError):
with fresh_db.atomic():
fresh_db.execute("insert or rollback into t (id) values (1)")
assert not fresh_db.conn.in_transaction

269
tests/test_cli.py
View file

@ -195,6 +195,50 @@ def test_output_table(db_path, options, expected):
assert expected == result.output.strip()
@pytest.mark.parametrize(
"fmt_option", [["--fmt", "simple"], ["-t"], ["--fmt", "github"]]
)
def test_output_table_no_headers(db_path, fmt_option):
# --no-headers should omit the header row from --fmt/--table output too, not
# just from --csv/--tsv (#566). Previously the flag was silently ignored for
# tabulate formats and the column names were always printed.
db = Database(db_path)
with db.conn:
db["dogs"].insert_all(
[
{"id": 1, "name": "Cleo", "age": 4},
{"id": 2, "name": "Pancakes", "age": 2},
]
)
sql = "select id, name, age from dogs order by id"
with_headers = CliRunner().invoke(cli.cli, ["query", db_path, sql] + fmt_option)
without_headers = CliRunner().invoke(
cli.cli, ["query", db_path, sql] + fmt_option + ["--no-headers"]
)
assert with_headers.exit_code == 0
assert without_headers.exit_code == 0
# The column names appear when headers are shown, and must not appear at all
# once --no-headers is passed.
assert "name" in with_headers.output
for header in ("id", "name", "age"):
assert (
header not in without_headers.output
), f"header {header!r} leaked into --no-headers output"
# The data is still all present.
for value in ("Cleo", "Pancakes", "1", "2", "4"):
assert value in without_headers.output
# The rows command shares the same code path.
rows_no_headers = CliRunner().invoke(
cli.cli, ["rows", db_path, "dogs"] + fmt_option + ["--no-headers"]
)
assert rows_no_headers.exit_code == 0
assert "name" not in rows_no_headers.output
assert "Cleo" in rows_no_headers.output
def test_create_index(db_path):
db = Database(db_path)
assert [] == db["Gosh"].indexes
@ -738,6 +782,25 @@ def test_query_json(db_path, sql, args, expected):
assert expected == result.output.strip()
def test_query_sql_from_stdin(db_path):
# https://github.com/simonw/sqlite-utils/issues/765
db = Database(db_path)
with db.conn:
db["dogs"].insert_all(
[
{"id": 1, "age": 4, "name": "Cleo"},
{"id": 2, "age": 2, "name": "Pancakes"},
]
)
result = CliRunner().invoke(
cli.cli,
["query", db_path, "-"],
input="select name from dogs order by name",
)
assert result.exit_code == 0, result.output
assert json.loads(result.output) == [{"name": "Cleo"}, {"name": "Pancakes"}]
def test_query_json_empty(db_path):
result = CliRunner().invoke(
cli.cli,
@ -746,6 +809,26 @@ def test_query_json_empty(db_path):
assert result.output.strip() == "[]"
def test_query_json_duplicate_columns_are_deduped(db_path):
# https://github.com/simonw/sqlite-utils/issues/624
result = CliRunner().invoke(
cli.cli,
[db_path, "select 1 as id, 2 as id, 'x' as value, 'y' as value"],
)
assert result.output.strip() == (
'[{"id": 1, "id_2": 2, "value": "x", "value_2": "y"}]'
)
def test_query_csv_duplicate_columns_are_preserved(db_path):
# CSV output should keep the duplicate headers, not rename them
result = CliRunner().invoke(
cli.cli,
[db_path, "select 1 as id, 2 as id", "--csv"],
)
assert result.output.replace("\r", "").strip() == "id,id\n1,2"
def test_query_invalid_function(db_path):
result = CliRunner().invoke(
cli.cli, [db_path, "select bad()", "--functions", "def invalid_python"]
@ -967,12 +1050,9 @@ def test_query_json_with_json_cols(db_path):
result = CliRunner().invoke(
cli.cli, [db_path, "select id, name, friends from dogs"]
)
assert (
r"""
assert r"""
[{"id": 1, "name": "Cleo", "friends": "[{\"name\": \"Pancakes\"}, {\"name\": \"Bailey\"}]"}]
""".strip()
== result.output.strip()
)
""".strip() == result.output.strip()
# With --json-cols:
result = CliRunner().invoke(
cli.cli, [db_path, "select id, name, friends from dogs", "--json-cols"]
@ -986,6 +1066,34 @@ def test_query_json_with_json_cols(db_path):
assert expected == result_rows.output.strip()
def test_query_json_unicode_not_escaped_by_default(db_path):
db = Database(db_path)
with db.conn:
db["text"].insert({"id": 1, "text": "Japanese 日本語"}, pk="id")
result = CliRunner().invoke(cli.cli, [db_path, "select id, text from text"])
assert result.exit_code == 0
assert result.output.strip() == '[{"id": 1, "text": "Japanese 日本語"}]'
# Same for --nl
result = CliRunner().invoke(cli.cli, [db_path, "select id, text from text", "--nl"])
assert result.exit_code == 0
assert result.output.strip() == '{"id": 1, "text": "Japanese 日本語"}'
@pytest.mark.parametrize("command", ["query", "rows"])
def test_query_json_ascii_option(db_path, command):
db = Database(db_path)
with db.conn:
db["text"].insert({"id": 1, "text": "Japanese 日本語"}, pk="id")
if command == "query":
args = [db_path, "select id, text from text", "--ascii"]
else:
args = ["rows", db_path, "text", "--ascii"]
result = CliRunner().invoke(cli.cli, args)
assert result.exit_code == 0
expected = '[{"id": 1, "text": "Japanese ' + "\\u65e5\\u672c\\u8a9e" + '"}]'
assert result.output.strip() == expected
@pytest.mark.parametrize(
"content,is_binary",
[(b"\x00\x0fbinary", True), ("this is text", False), (1, False), (1.5, False)],
@ -1127,20 +1235,38 @@ def test_upsert(db_path, tmpdir):
]
def test_upsert_pk_required(db_path, tmpdir):
def test_upsert_pk_inferred_from_existing_table(db_path, tmpdir):
json_path = str(tmpdir / "dogs.json")
db = Database(db_path)
insert_dogs = [
{"id": 1, "name": "Cleo", "age": 4},
{"id": 2, "name": "Nixie", "age": 4},
]
write_json(json_path, insert_dogs)
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "dogs", json_path, "--pk", "id"],
catch_exceptions=False,
)
assert result.exit_code == 0, result.output
write_json(
json_path,
[
{"id": 1, "age": 5},
{"id": 2, "age": 5},
],
)
result = CliRunner().invoke(
cli.cli,
["upsert", db_path, "dogs", json_path],
catch_exceptions=False,
)
assert result.exit_code == 2
assert "Error: Missing option '--pk'" in result.output
assert result.exit_code == 0, result.output
assert list(db.query("select * from dogs order by id")) == [
{"id": 1, "name": "Cleo", "age": 5},
{"id": 2, "name": "Nixie", "age": 5},
]
def test_upsert_analyze(db_path, tmpdir):
@ -1470,6 +1596,24 @@ def test_drop_table_error():
assert result.exit_code == 0
def test_drop_table_on_view_errors():
runner = CliRunner()
with runner.isolated_filesystem():
db = Database("test.db")
db["t"].insert({"id": 1})
db.create_view("v", "select * from t")
result = runner.invoke(cli.cli, ["drop-table", "test.db", "v"])
assert result.exit_code == 1
assert 'Error: "v" is a view, not a table - use drop-view to drop it' == (
result.output.strip()
)
assert "v" in db.view_names()
# --ignore exits cleanly but must still not drop the view
result = runner.invoke(cli.cli, ["drop-table", "test.db", "v", "--ignore"])
assert result.exit_code == 0
assert "v" in db.view_names()
def test_drop_view():
runner = CliRunner()
with runner.isolated_filesystem():
@ -1488,6 +1632,23 @@ def test_drop_view():
assert "hello" not in db.view_names()
def test_drop_view_on_table_errors():
runner = CliRunner()
with runner.isolated_filesystem():
db = Database("test.db")
db["t"].insert({"id": 1})
result = runner.invoke(cli.cli, ["drop-view", "test.db", "t"])
assert result.exit_code == 1
assert 'Error: "t" is a table, not a view - use drop-table to drop it' == (
result.output.strip()
)
assert "t" in db.table_names()
# --ignore exits cleanly but must still not drop the table
result = runner.invoke(cli.cli, ["drop-view", "test.db", "t", "--ignore"])
assert result.exit_code == 0
assert "t" in db.table_names()
def test_drop_view_error():
runner = CliRunner()
with runner.isolated_filesystem():
@ -1737,6 +1898,29 @@ def test_transform(db_path, args, expected_schema):
assert schema == expected_schema
def test_transform_sql(db_path):
db = Database(db_path)
with db.conn:
db["dogs"].insert(
{"id": 1, "age": 4, "name": "Cleo"},
not_null={"age"},
defaults={"age": 1},
pk="id",
)
original_schema = db["dogs"].schema
result = CliRunner().invoke(
cli.cli, ["transform", db_path, "dogs", "--drop", "name", "--sql"]
)
assert result.exit_code == 0, result.output
assert 'CREATE TABLE "dogs_new_' in result.output
assert '"age" INTEGER NOT NULL DEFAULT' in result.output
assert 'DROP TABLE "dogs";' in result.output
assert 'ALTER TABLE "dogs_new_' in result.output
assert db["dogs"].schema == original_schema
@pytest.mark.parametrize(
"extra_args,expected_schema",
(
@ -1998,12 +2182,10 @@ def test_search_quote(tmpdir):
def test_indexes(tmpdir):
db_path = str(tmpdir / "test.db")
db = Database(db_path)
db.conn.executescript(
"""
db.conn.executescript("""
create table Gosh (c1 text, c2 text, c3 text);
create index Gosh_idx on Gosh(c2, c3 desc);
"""
)
""")
result = CliRunner().invoke(
cli.cli,
["indexes", str(db_path)],
@ -2094,16 +2276,12 @@ def test_triggers(tmpdir, extra_args, expected):
pk="id",
)
db["counter"].insert({"count": 1})
db.conn.execute(
textwrap.dedent(
"""
db.conn.execute(textwrap.dedent("""
CREATE TRIGGER blah AFTER INSERT ON articles
BEGIN
UPDATE counter SET count = count + 1;
END
"""
)
)
"""))
args = ["triggers", db_path]
if extra_args:
args.extend(extra_args)
@ -2282,18 +2460,14 @@ def test_csv_insert_bom(tmpdir):
]
@pytest.mark.parametrize("option", (None, "-d", "--detect-types"))
def test_insert_detect_types(tmpdir, option):
"""Test that type detection is now the default behavior"""
def test_insert_detect_types(tmpdir):
"""Test that type detection is the default behavior"""
db_path = str(tmpdir / "test.db")
data = "name,age,weight\nCleo,6,45.5\nDori,1,3.5"
extra = []
if option:
extra = [option]
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "creatures", "-", "--csv"] + extra,
["insert", db_path, "creatures", "-", "--csv"],
catch_exceptions=False,
input=data,
)
@ -2305,17 +2479,27 @@ def test_insert_detect_types(tmpdir, option):
]
@pytest.mark.parametrize("option", (None, "-d", "--detect-types"))
def test_upsert_detect_types(tmpdir, option):
"""Test that type detection is now the default behavior for upsert"""
@pytest.mark.parametrize("command", ("insert", "upsert"))
@pytest.mark.parametrize("option", ("-d", "--detect-types"))
def test_detect_types_flag_removed(tmpdir, command, option):
# The old no-op flag was removed in 4.0 - it should now error
db_path = str(tmpdir / "test.db")
data = "id,name,age,weight\n1,Cleo,6,45.5\n2,Dori,1,3.5"
extra = []
if option:
extra = [option]
result = CliRunner().invoke(
cli.cli,
["upsert", db_path, "creatures", "-", "--csv", "--pk", "id"] + extra,
[command, db_path, "creatures", "-", "--csv", "--pk", "id", option],
input="id,name\n1,Cleo",
)
assert result.exit_code == 2
assert "No such option" in result.output
def test_upsert_detect_types(tmpdir):
"""Test that type detection is the default behavior for upsert"""
db_path = str(tmpdir / "test.db")
data = "id,name,age,weight\n1,Cleo,6,45.5\n2,Dori,1,3.5"
result = CliRunner().invoke(
cli.cli,
["upsert", db_path, "creatures", "-", "--csv", "--pk", "id"],
catch_exceptions=False,
input=data,
)
@ -2611,3 +2795,22 @@ def test_insert_upsert_strict(tmpdir, method, strict):
assert result.exit_code == 0
db = Database(db_path)
assert db["items"].strict == strict or not db.supports_strict
def test_extract_bad_column_clean_error(db_path):
db = Database(db_path)
db["trees"].insert({"id": 1, "species": "Palm"}, pk="id")
result = CliRunner().invoke(cli.cli, ["extract", db_path, "trees", "nope"])
assert result.exit_code == 1
assert result.exception is None or isinstance(result.exception, SystemExit)
assert result.output.startswith("Error: Invalid columns")
def test_extract_view_clean_error(db_path):
db = Database(db_path)
db["trees"].insert({"id": 1, "species": "Palm"}, pk="id")
db.create_view("v", "select * from trees")
result = CliRunner().invoke(cli.cli, ["extract", db_path, "v", "species"])
assert result.exit_code == 1
assert result.exception is None or isinstance(result.exception, SystemExit)
assert result.output.startswith("Error:")

25
tests/test_cli_convert.py
View file

@ -215,6 +215,25 @@ def test_convert_multi_dryrun(test_db_and_path):
)
def test_convert_multi_dryrun_unicode_not_escaped(test_db_and_path):
db_path = test_db_and_path[1]
result = CliRunner().invoke(
cli.cli,
[
"convert",
db_path,
"example",
"dt",
"{'text': 'Japanese 日本語'}",
"--dry-run",
"--multi",
],
)
assert result.exit_code == 0
# Preview should match what jsonify_if_needed() would actually store
assert '{"text": "Japanese 日本語"}' in result.output
@pytest.mark.parametrize("drop", (True, False))
def test_convert_output_column(test_db_and_path, drop):
db, db_path = test_db_and_path
@ -371,16 +390,14 @@ def test_convert_multi_complex_column_types(fresh_db_and_path):
],
pk="id",
)
code = textwrap.dedent(
"""
code = textwrap.dedent("""
if value == 1:
return {"is_str": "", "is_float": 1.2, "is_int": None}
elif value == 2:
return {"is_float": 1, "is_int": 12}
elif value == 3:
return {"is_bytes": b"blah"}
"""
)
""")
result = CliRunner().invoke(
cli.cli,
[

291
tests/test_cli_insert.py
View file

@ -597,3 +597,294 @@ def test_insert_streaming_batch_size_1(db_path):
proc.stdin.close()
proc.wait()
assert proc.returncode == 0
def test_insert_csv_headers_only(tmpdir):
"""Test that CSV with only header row (no data) works with --detect-types (issue #702)"""
db_path = str(tmpdir / "test.db")
csv_path = str(tmpdir / "headers_only.csv")
with open(csv_path, "w") as fp:
fp.write("id,name,age\n")
# Should not crash with --detect-types (which is now the default)
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "data", csv_path, "--csv"],
catch_exceptions=False,
)
assert result.exit_code == 0
# Table should not exist since there were no data rows
db = Database(db_path)
assert not db["data"].exists()
def test_insert_into_view_errors(tmpdir):
db_path = str(tmpdir / "test.db")
db = Database(db_path)
db["t"].insert({"id": 1})
db.create_view("v", "select * from t")
db.close()
result = CliRunner().invoke(
cli.cli, ["insert", db_path, "v", "-"], input='{"id": 2}'
)
assert result.exit_code == 1
assert result.output.strip() == "Error: Table v is actually a view"
def test_insert_csv_detect_types_leaves_existing_table_alone(db_path):
# Type detection is the default for CSV/TSV inserts, but it must only
# apply to tables created by this command - transforming a pre-existing
# table would rewrite its column types and corrupt data such as
# TEXT zip codes with leading zeros
db = Database(db_path)
db["places"].insert({"name": "Boston", "zip": "01234"})
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "places", "-", "--csv"],
catch_exceptions=False,
input="name,zip\nSF,94107",
)
assert result.exit_code == 0, result.output
assert db["places"].columns_dict["zip"] is str
assert list(db["places"].rows) == [
{"name": "Boston", "zip": "01234"},
{"name": "SF", "zip": "94107"},
]
def test_insert_csv_detect_types_new_table(db_path):
# A table created by the insert still gets detected types
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "data", "-", "--csv"],
catch_exceptions=False,
input="name,age,weight\nCleo,5,12.5",
)
assert result.exit_code == 0, result.output
db = Database(db_path)
assert db["data"].columns_dict == {"name": str, "age": int, "weight": float}
@pytest.mark.parametrize(
"command,extra_args,input_text,expected_row",
(
(
"insert",
[],
"zipcode,score\n01234,9.5\n",
{"zipcode": "01234", "score": 9.5},
),
(
"upsert",
["--pk", "id"],
"id,zipcode,score\n1,01234,9.5\n",
{"id": 1, "zipcode": "01234", "score": 9.5},
),
),
)
def test_insert_upsert_csv_type_overrides_detected_types(
db_path, command, extra_args, input_text, expected_row
):
result = CliRunner().invoke(
cli.cli,
[
command,
db_path,
"places",
"-",
"--csv",
]
+ extra_args
+ [
"--type",
"zipcode",
"text",
],
catch_exceptions=False,
input=input_text,
)
assert result.exit_code == 0, result.output
db = Database(db_path)
expected_columns = {"zipcode": str, "score": float}
if command == "upsert":
expected_columns = {"id": int, **expected_columns}
assert db["places"].columns_dict == expected_columns
assert list(db["places"].rows) == [expected_row]
def test_upsert_csv_detect_types_leaves_existing_table_alone(db_path):
db = Database(db_path)
db["places"].insert({"id": 1, "name": "Boston", "zip": "01234"}, pk="id")
result = CliRunner().invoke(
cli.cli,
["upsert", db_path, "places", "-", "--csv", "--pk", "id"],
catch_exceptions=False,
input="id,name,zip\n2,SF,94107",
)
assert result.exit_code == 0, result.output
assert db["places"].columns_dict["zip"] is str
assert db["places"].get(1)["zip"] == "01234"
def test_insert_invalid_pk_clean_error(db_path):
# An invalid --pk against an existing table should be a clean CLI
# error, not a raw InvalidColumns traceback
db = Database(db_path)
db["t"].insert({"a": 1})
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "t", "-", "--pk", "badcol"],
input='{"a": 2}',
)
assert result.exit_code == 1
assert result.exception is None or isinstance(result.exception, SystemExit)
assert result.output.startswith("Error: Invalid primary key column")
# --code tests, see https://github.com/simonw/sqlite-utils/issues/684
CODE_ROWS_FUNCTION = """
def rows():
yield {"id": 1, "name": "Cleo"}
yield {"id": 2, "name": "Suna"}
"""
CODE_ROWS_ITERABLE = """
rows = [
{"id": 1, "name": "Cleo"},
{"id": 2, "name": "Suna"},
]
"""
@pytest.mark.parametrize("code", (CODE_ROWS_FUNCTION, CODE_ROWS_ITERABLE))
def test_insert_code(tmpdir, code):
db_path = str(tmpdir / "dogs.db")
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "creatures", "--code", code, "--pk", "id"],
)
assert result.exit_code == 0, result.output
db = Database(db_path)
assert db["creatures"].pks == ["id"]
assert list(db["creatures"].rows) == [
{"id": 1, "name": "Cleo"},
{"id": 2, "name": "Suna"},
]
def test_insert_code_from_file(tmpdir):
db_path = str(tmpdir / "dogs.db")
code_path = str(tmpdir / "gen.py")
with open(code_path, "w") as fp:
fp.write(CODE_ROWS_FUNCTION)
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "creatures", "--code", code_path],
)
assert result.exit_code == 0, result.output
assert list(Database(db_path)["creatures"].rows) == [
{"id": 1, "name": "Cleo"},
{"id": 2, "name": "Suna"},
]
def test_upsert_code(tmpdir):
db_path = str(tmpdir / "dogs.db")
db = Database(db_path)
db["creatures"].insert_all(
[{"id": 1, "name": "old"}, {"id": 2, "name": "Suna"}], pk="id"
)
result = CliRunner().invoke(
cli.cli,
["upsert", db_path, "creatures", "--code", CODE_ROWS_FUNCTION, "--pk", "id"],
)
assert result.exit_code == 0, result.output
assert list(db["creatures"].rows) == [
{"id": 1, "name": "Cleo"},
{"id": 2, "name": "Suna"},
]
def test_insert_code_requires_file_or_code(tmpdir):
db_path = str(tmpdir / "dogs.db")
result = CliRunner().invoke(cli.cli, ["insert", db_path, "creatures"])
assert result.exit_code == 1
assert "Provide either a FILE argument or --code" in result.output
def test_insert_code_mutually_exclusive_with_file(tmpdir):
db_path = str(tmpdir / "dogs.db")
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "creatures", "-", "--code", CODE_ROWS_FUNCTION],
input="{}",
)
assert result.exit_code == 1
assert "--code cannot be used with a FILE argument" in result.output
def test_insert_code_rejects_input_format_options(tmpdir):
db_path = str(tmpdir / "dogs.db")
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "creatures", "--code", CODE_ROWS_FUNCTION, "--csv"],
)
assert result.exit_code == 1
assert "--code cannot be used with input format options" in result.output
def test_insert_code_missing_rows(tmpdir):
db_path = str(tmpdir / "dogs.db")
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "creatures", "--code", "x = 1"],
)
assert result.exit_code == 1
assert "must define a 'rows' function or iterable" in result.output
def test_insert_code_single_dict(tmpdir):
db_path = str(tmpdir / "dogs.db")
result = CliRunner().invoke(
cli.cli,
[
"insert",
db_path,
"creatures",
"--code",
'rows = {"id": 1, "name": "Cleo"}',
"--pk",
"id",
],
)
assert result.exit_code == 0, result.output
assert list(Database(db_path)["creatures"].rows) == [{"id": 1, "name": "Cleo"}]
def test_insert_code_not_iterable(tmpdir):
db_path = str(tmpdir / "dogs.db")
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "creatures", "--code", "rows = 5"],
)
assert result.exit_code == 1
assert "must define a 'rows' function or iterable" in result.output
def test_insert_code_syntax_error(tmpdir):
db_path = str(tmpdir / "dogs.db")
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "creatures", "--code", "def rows(:"],
)
assert result.exit_code == 1
assert "Error in --code" in result.output
def test_insert_code_file_not_found(tmpdir):
db_path = str(tmpdir / "dogs.db")
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "creatures", "--code", "missing.py"],
)
assert result.exit_code == 1
assert "File not found: missing.py" in result.output

18
tests/test_cli_memory.py
View file

@ -156,6 +156,24 @@ def test_memory_csv_encoding(tmpdir, use_stdin):
}
def test_memory_csv_headers_only(tmpdir):
csv_path = str(tmpdir / "headers_only.csv")
with open(csv_path, "w") as fp:
fp.write("id,name,age\n")
result = CliRunner().invoke(
cli.cli,
["memory", csv_path, "", "--schema"],
catch_exceptions=False,
)
assert result.exit_code == 0
assert result.output.strip() == (
'CREATE VIEW "t1" AS select * from "headers_only";\n'
'CREATE VIEW "t" AS select * from "headers_only";'
)
@pytest.mark.parametrize("extra_args", ([], ["select 1"]))
def test_memory_dump(extra_args):
result = CliRunner().invoke(

507
tests/test_cli_migrate.py Normal file
View file

@ -0,0 +1,507 @@
import pathlib
from click.testing import CliRunner
import pytest
import sqlite_utils
import sqlite_utils.cli
TWO_MIGRATIONS = """
from sqlite_utils import Migrations
m = Migrations("hello")
@m()
def foo(db):
db["foo"].insert({"hello": "world"})
@m()
def bar(db):
db["bar"].insert({"hello": "world"})
"""
@pytest.fixture
def two_migrations(tmpdir):
path = pathlib.Path(tmpdir)
(path / "foo").mkdir()
migrations_py = path / "foo" / "migrations.py"
migrations_py.write_text(TWO_MIGRATIONS, "utf-8")
return path, migrations_py
@pytest.fixture
def two_sets_same_migration_name(tmpdir):
path = pathlib.Path(tmpdir)
migrations_py = path / "migrations.py"
migrations_py.write_text(
"""
from sqlite_utils import Migrations
creatures = Migrations("creatures")
@creatures()
def create_table(db):
db["creatures"].insert({"name": "Cleo"})
@creatures()
def add_weight(db):
db["creature_weights"].insert({"weight": 4.2})
sales = Migrations("sales")
@sales()
def create_table(db):
db["sales"].insert({"id": 1})
@sales()
def add_weight(db):
db["sales_weights"].insert({"weight": 10})
""",
"utf-8",
)
return path, migrations_py
@pytest.mark.parametrize("arg", ("TMPDIR", "TMPDIR/foo/migrations.py", "TMPDIR/foo/"))
def test_basic(two_migrations, arg):
path, _ = two_migrations
db_path = str(path / "test.db")
runner = CliRunner()
def _list():
list_result = runner.invoke(
sqlite_utils.cli.cli,
["migrate", db_path, "--list", arg.replace("TMPDIR", str(path))],
)
assert list_result.exit_code == 0
return list_result.output
assert _list() == (
"Migrations for: hello\n\n"
" Applied:\n\n"
" Pending:\n"
" foo\n"
" bar\n\n"
)
result = runner.invoke(
sqlite_utils.cli.cli, ["migrate", db_path, arg.replace("TMPDIR", str(path))]
)
assert result.exit_code == 0, result.output
list_output = _list()
assert "Migrations for: hello\n\n Applied:\n " in list_output
prior_to_pending = list_output.split(" Pending")[0]
assert " foo" in prior_to_pending
assert " bar" in prior_to_pending
assert " Pending:\n (none)" in list_output
db = sqlite_utils.Database(db_path)
assert db["foo"].exists()
assert db["bar"].exists()
assert db["_sqlite_migrations"].exists()
rows = list(db["_sqlite_migrations"].rows)
assert len(rows) == 2
assert rows[0]["name"] == "foo"
assert rows[1]["name"] == "bar"
def test_list_same_migration_names_in_different_sets(capsys):
applied = sqlite_utils.Migrations("applied")
@applied(name="foo")
def applied_foo(db):
db["applied"].insert({"hello": "world"})
pending = sqlite_utils.Migrations("pending")
@pending(name="foo")
def pending_foo(db):
db["pending"].insert({"hello": "world"})
db = sqlite_utils.Database(memory=True)
applied.apply(db)
sqlite_utils.cli._display_migration_list(db, [applied, pending])
output = capsys.readouterr().out
assert (
"Migrations for: pending\n\n" " Applied:\n\n" " Pending:\n" " foo\n\n"
) in output
def test_verbose(tmpdir):
path = pathlib.Path(tmpdir)
(path / "foo").mkdir()
migrations_py = path / "foo" / "migrations.py"
migrations_py.write_text(
"""
from sqlite_utils import Migrations
m = Migrations("hello")
@m()
def foo(db):
db["dogs"].insert({"id": 1, "name": "Cleo"})
""",
"utf-8",
)
db_path = str(path / "test.db")
runner = CliRunner()
result = runner.invoke(
sqlite_utils.cli.cli, ["migrate", db_path, str(migrations_py)]
)
assert result.exit_code == 0
result = runner.invoke(
sqlite_utils.cli.cli, ["migrate", db_path, str(migrations_py), "--verbose"]
)
assert result.exit_code == 0
expected = """
Schema before:
CREATE TABLE "_sqlite_migrations" (
"id" INTEGER PRIMARY KEY,
"migration_set" TEXT,
"name" TEXT,
"applied_at" TEXT
);
CREATE UNIQUE INDEX "idx__sqlite_migrations_migration_set_name"
ON "_sqlite_migrations" ("migration_set", "name");
CREATE TABLE "dogs" (
"id" INTEGER,
"name" TEXT
);
Schema after:
(unchanged)
""".strip()
assert expected in result.output
new_migration = """
@m()
def bar(db):
db["dogs"].add_column("age", int)
db["dogs"].add_column("weight", float)
db["dogs"].transform()
"""
migrations_py.write_text(migrations_py.read_text("utf-8") + new_migration)
result = runner.invoke(
sqlite_utils.cli.cli, ["migrate", db_path, str(migrations_py), "--verbose"]
)
assert result.exit_code == 0
expected_diff = """
Schema diff:
ON "_sqlite_migrations" ("migration_set", "name");
CREATE TABLE "dogs" (
"id" INTEGER,
- "name" TEXT
+ "name" TEXT,
+ "age" INTEGER,
+ "weight" REAL
);
""".strip()
assert expected_diff in result.output
def test_stop_before(two_migrations):
path, _ = two_migrations
db_path = str(path / "test.db")
result = CliRunner().invoke(
sqlite_utils.cli.cli,
[
"migrate",
db_path,
str(path / "foo" / "migrations.py"),
"--stop-before",
"bar",
],
)
assert result.exit_code == 0
db = sqlite_utils.Database(db_path)
assert db["foo"].exists()
assert not db["bar"].exists()
def test_stop_before_multiple_sets_unqualified(two_migrations):
path, _ = two_migrations
db_path = str(path / "test.db")
(path / "foo" / "migrations2.py").write_text(
"""
from sqlite_utils import Migrations
m = Migrations("hello2")
@m()
def foo(db):
db["foo"].insert({"hello": "world"})
""",
"utf-8",
)
result = CliRunner().invoke(
sqlite_utils.cli.cli,
[
"migrate",
db_path,
str(path / "foo" / "migrations.py"),
str(path / "foo" / "migrations2.py"),
"--stop-before",
"foo",
],
)
assert result.exit_code == 0, result.output
db = sqlite_utils.Database(db_path)
assert db.table_names() == ["_sqlite_migrations"]
assert list(db["_sqlite_migrations"].rows) == []
def test_stop_before_qualified_only_affects_named_set(two_sets_same_migration_name):
path, migrations_py = two_sets_same_migration_name
db_path = str(path / "test.db")
result = CliRunner().invoke(
sqlite_utils.cli.cli,
[
"migrate",
db_path,
str(migrations_py),
"--stop-before",
"creatures:add_weight",
],
)
assert result.exit_code == 0, result.output
db = sqlite_utils.Database(db_path)
assert db["creatures"].exists()
assert not db["creature_weights"].exists()
assert db["sales"].exists()
assert db["sales_weights"].exists()
def test_stop_before_multiple_qualified(two_sets_same_migration_name):
path, migrations_py = two_sets_same_migration_name
db_path = str(path / "test.db")
result = CliRunner().invoke(
sqlite_utils.cli.cli,
[
"migrate",
db_path,
str(migrations_py),
"--stop-before",
"creatures:add_weight",
"--stop-before",
"sales:add_weight",
],
)
assert result.exit_code == 0, result.output
db = sqlite_utils.Database(db_path)
assert db["creatures"].exists()
assert not db["creature_weights"].exists()
assert db["sales"].exists()
assert not db["sales_weights"].exists()
LEGACY_MIGRATIONS = """
import datetime
class _Migration:
def __init__(self, name, fn):
self.name = name
self.fn = fn
class _Applied:
def __init__(self, name, applied_at):
self.name = name
self.applied_at = applied_at
class LegacyMigrations:
# Mimics the sqlite-migrate 0.x Migrations class, in particular
# apply(db, stop_before=None) taking a single string
migrations_table = "_sqlite_migrations"
def __init__(self, name):
self.name = name
self._migrations = []
def __call__(self, fn):
self._migrations.append(_Migration(fn.__name__, fn))
return fn
def ensure_migrations_table(self, db):
db[self.migrations_table].create(
{"migration_set": str, "name": str, "applied_at": str},
pk=("migration_set", "name"),
if_not_exists=True,
)
def applied(self, db):
self.ensure_migrations_table(db)
return [
_Applied(row["name"], row["applied_at"])
for row in db[self.migrations_table].rows_where(
"migration_set = ?", [self.name]
)
]
def pending(self, db):
applied = {m.name for m in self.applied(db)}
return [m for m in self._migrations if m.name not in applied]
def apply(self, db, stop_before=None):
for migration in self.pending(db):
if migration.name == stop_before:
return
migration.fn(db)
db[self.migrations_table].insert(
{
"migration_set": self.name,
"name": migration.name,
"applied_at": str(
datetime.datetime.now(datetime.timezone.utc)
),
}
)
legacy = LegacyMigrations("legacy_set")
@legacy
def first(db):
db["first"].insert({"hello": "world"})
@legacy
def second(db):
db["second"].insert({"hello": "world"})
"""
def test_stop_before_unknown_name_errors(two_migrations):
path, _ = two_migrations
db_path = str(path / "test.db")
runner = CliRunner()
result = runner.invoke(
sqlite_utils.cli.cli,
["migrate", db_path, str(path), "--stop-before", "fooo"],
)
assert result.exit_code == 1
assert "--stop-before did not match any migrations: fooo" in result.output
# Nothing should have been applied
db = sqlite_utils.Database(db_path)
assert "foo" not in db.table_names()
assert "bar" not in db.table_names()
def test_stop_before_with_legacy_migrations_class(tmpdir):
path = pathlib.Path(tmpdir)
(path / "migrations.py").write_text(LEGACY_MIGRATIONS, "utf-8")
db_path = str(path / "test.db")
runner = CliRunner()
result = runner.invoke(
sqlite_utils.cli.cli,
["migrate", db_path, str(path), "--stop-before", "second"],
)
assert result.exit_code == 0, result.output
db = sqlite_utils.Database(db_path)
assert "first" in db.table_names()
assert "second" not in db.table_names()
def test_stop_before_multiple_values_for_legacy_set_errors(tmpdir):
path = pathlib.Path(tmpdir)
(path / "migrations.py").write_text(LEGACY_MIGRATIONS, "utf-8")
db_path = str(path / "test.db")
runner = CliRunner()
result = runner.invoke(
sqlite_utils.cli.cli,
[
"migrate",
db_path,
str(path),
"--stop-before",
"legacy_set:first",
"--stop-before",
"legacy_set:second",
],
)
assert result.exit_code == 1
assert "single --stop-before" in result.output
def test_list_does_not_create_database_file(two_migrations):
path, _ = two_migrations
db_path = path / "test.db"
runner = CliRunner()
result = runner.invoke(
sqlite_utils.cli.cli, ["migrate", str(db_path), str(path), "--list"]
)
assert result.exit_code == 0, result.output
assert "Pending:\n foo\n bar" in result.output
# Listing migrations must not create the database file
assert not db_path.exists()
def test_list_does_not_upgrade_legacy_migrations_table(two_migrations):
path, _ = two_migrations
db_path = str(path / "test.db")
db = sqlite_utils.Database(db_path)
db["_sqlite_migrations"].create(
{"migration_set": str, "name": str, "applied_at": str},
pk=("migration_set", "name"),
)
db["_sqlite_migrations"].insert(
{"migration_set": "hello", "name": "foo", "applied_at": "x"}
)
db.close()
runner = CliRunner()
result = runner.invoke(
sqlite_utils.cli.cli, ["migrate", db_path, str(path), "--list"]
)
assert result.exit_code == 0, result.output
assert "foo - x" in result.output
# --list must not perform the one-way legacy schema upgrade
db2 = sqlite_utils.Database(db_path)
assert db2["_sqlite_migrations"].pks == ["migration_set", "name"]
db2.close()
def test_stop_before_applied_migration_errors(two_migrations):
path, _ = two_migrations
db_path = str(path / "test.db")
migrations_path = str(path / "foo" / "migrations.py")
# Apply everything first
first = CliRunner().invoke(
sqlite_utils.cli.cli,
["migrate", db_path, migrations_path, "--stop-before", "bar"],
)
assert first.exit_code == 0
# foo is now applied - stopping before it is an error, and bar
# must not be applied as a side effect
result = CliRunner().invoke(
sqlite_utils.cli.cli,
["migrate", db_path, migrations_path, "--stop-before", "foo"],
)
assert result.exit_code != 0
assert "already been applied" in result.output
db = sqlite_utils.Database(db_path)
assert not db["bar"].exists()
def test_list_with_legacy_class_is_read_only(tmpdir):
# Legacy sqlite-migrate classes create the _sqlite_migrations table
# from their pending()/applied() methods - --list must roll that
# back so it stays a read-only operation as documented
path = pathlib.Path(tmpdir)
(path / "migrations.py").write_text(LEGACY_MIGRATIONS, "utf-8")
db_path = str(path / "test.db")
db = sqlite_utils.Database(db_path)
db["existing"].insert({"id": 1})
db.close()
result = CliRunner().invoke(
sqlite_utils.cli.cli, ["migrate", db_path, str(path), "--list"]
)
assert result.exit_code == 0, result.output
assert "first" in result.output
db2 = sqlite_utils.Database(db_path)
assert "_sqlite_migrations" not in db2.table_names()
db2.close()

233
tests/test_column_casing.py Normal file
View file

@ -0,0 +1,233 @@
"""
SQLite treats column names as case-insensitive. These tests exercise the
places where sqlite-utils performs Python-side lookups of column names
provided by the caller, which should match the schema case-insensitively.
https://github.com/simonw/sqlite-utils/issues/760
"""
import pytest
from sqlite_utils import Database
from sqlite_utils.db import ForeignKey
def test_insert_populates_last_pk_case_insensitively(fresh_db):
books = fresh_db["books"]
books.create({"Id": int, "Title": str}, pk="Id")
books.insert({"Id": 1, "Title": "One"}, pk="id")
assert books.last_pk == 1
def test_insert_populates_last_pk_compound_pk_case_insensitively(fresh_db):
books = fresh_db["books"]
books.create({"Author": str, "Position": int, "Title": str})
books.insert(
{"Author": "Sue", "Position": 1, "Title": "One"}, pk=("author", "position")
)
assert books.last_pk == ("Sue", 1)
@pytest.mark.parametrize("use_old_upsert", (False, True))
def test_upsert_pk_case_differs_from_schema(use_old_upsert):
db = Database(memory=True, use_old_upsert=use_old_upsert)
books = db["books"]
books.create({"Id": int, "Title": str}, pk="Id")
books.insert({"Id": 1, "Title": "One"})
books.upsert({"id": 1, "title": "Won"}, pk="id")
assert list(books.rows) == [{"Id": 1, "Title": "Won"}]
assert books.last_pk == 1
@pytest.mark.parametrize("use_old_upsert", (False, True))
def test_upsert_record_key_case_differs_from_pk(use_old_upsert):
# all_columns comes from the record keys, pk= from the caller
db = Database(memory=True, use_old_upsert=use_old_upsert)
books = db["books"]
books.create({"Id": int, "Title": str}, pk="Id")
books.upsert({"ID": 1, "Title": "One"}, pk="id")
assert list(books.rows) == [{"Id": 1, "Title": "One"}]
assert books.last_pk == 1
def test_upsert_inferred_pk_case_differs_from_record_keys(fresh_db):
# pk is inferred from the existing schema as "Id", records use "id"
books = fresh_db["books"]
books.create({"Id": int, "Title": str}, pk="Id")
books.upsert({"id": 1, "title": "One"})
assert list(books.rows) == [{"Id": 1, "Title": "One"}]
assert books.last_pk == 1
def test_upsert_list_mode_pk_case_insensitive(fresh_db):
books = fresh_db["books"]
books.create({"Id": int, "Title": str}, pk="Id")
books.upsert_all([["id", "title"], [1, "One"]], pk="Id")
assert list(books.rows) == [{"Id": 1, "Title": "One"}]
assert books.last_pk == 1
def test_lookup_pk_case_insensitive(fresh_db):
fresh_db["species"].create({"ID": int, "Name": str}, pk="ID")
fresh_db["species"].insert({"ID": 5, "Name": "Palm"})
fresh_db["species"].create_index(["Name"], unique=True)
assert fresh_db["species"].lookup({"Name": "Palm"}, pk="id") == 5
def test_lookup_does_not_create_redundant_index(fresh_db):
fresh_db["species"].create({"id": int, "Name": str}, pk="id")
fresh_db["species"].create_index(["Name"], unique=True)
fresh_db["species"].lookup({"name": "Palm"})
assert len(fresh_db["species"].indexes) == 1
def test_create_table_transform_same_columns_different_case(fresh_db):
fresh_db["t"].create({"Name": str, "Age": int})
fresh_db["t"].insert({"Name": "Cleo", "Age": 5})
fresh_db.create_table("t", {"name": str, "age": int}, transform=True)
# Schema casing is preserved - SQLite considers these the same columns
assert fresh_db["t"].columns_dict == {"Name": str, "Age": int}
assert list(fresh_db["t"].rows) == [{"Name": "Cleo", "Age": 5}]
def test_create_table_transform_case_insensitive_with_changes(fresh_db):
fresh_db["t"].create({"Name": str, "Age": int})
fresh_db.create_table("t", {"name": str, "age": str, "size": int}, transform=True)
# age changed type, size added, Name untouched
assert fresh_db["t"].columns_dict == {"Name": str, "Age": str, "size": int}
def test_transform_types_case_insensitive(fresh_db):
fresh_db["t"].create({"Name": str, "Age": str})
fresh_db["t"].transform(types={"age": int})
assert fresh_db["t"].columns_dict == {"Name": str, "Age": int}
def test_transform_rename_case_insensitive(fresh_db):
fresh_db["t"].create({"Name": str})
fresh_db["t"].transform(rename={"name": "title"})
assert fresh_db["t"].columns_dict == {"title": str}
def test_transform_drop_case_insensitive(fresh_db):
fresh_db["t"].create({"Name": str, "Age": int})
fresh_db["t"].transform(drop=["name"])
assert fresh_db["t"].columns_dict == {"Age": int}
def test_transform_not_null_and_defaults_case_insensitive(fresh_db):
fresh_db["t"].create({"Name": str, "Age": int})
fresh_db["t"].transform(not_null={"name"}, defaults={"age": 3})
columns = {c.name: c for c in fresh_db["t"].columns}
assert columns["Name"].notnull
assert fresh_db["t"].default_values == {"Age": 3}
def test_transform_pk_case_insensitive(fresh_db):
fresh_db["t"].create({"Id": int, "Name": str})
fresh_db["t"].transform(pk="id")
assert fresh_db["t"].pks == ["Id"]
assert fresh_db["t"].columns_dict == {"Id": int, "Name": str}
def test_transform_drop_foreign_keys_case_insensitive(fresh_db):
fresh_db["parent"].create({"Id": int}, pk="Id")
fresh_db["child"].create(
{"id": int, "Parent_ID": int},
pk="id",
foreign_keys=[("Parent_ID", "parent", "Id")],
)
fresh_db["child"].transform(drop_foreign_keys=["parent_id"])
assert fresh_db["child"].foreign_keys == []
def test_add_foreign_key_case_insensitive(fresh_db):
fresh_db["parent"].create({"Id": int}, pk="Id")
fresh_db["child"].create({"id": int, "Parent_ID": int}, pk="id")
fresh_db["child"].add_foreign_key("parent_id", "parent", "id")
fks = fresh_db["child"].foreign_keys
assert len(fks) == 1
# The foreign key should use the schema casing of the columns
assert fks[0].column == "Parent_ID"
assert fks[0].other_column == "Id"
def test_add_foreign_keys_case_insensitive(fresh_db):
fresh_db["parent"].create({"Id": int}, pk="Id")
fresh_db["child"].create({"id": int, "Parent_ID": int}, pk="id")
fresh_db.add_foreign_keys([("child", "parent_id", "parent", "id")])
fks = fresh_db["child"].foreign_keys
assert len(fks) == 1
assert fks[0].column == "Parent_ID"
assert fks[0].other_column == "Id"
def test_add_foreign_key_detects_existing_case_insensitively(fresh_db):
fresh_db["parent"].create({"Id": int}, pk="Id")
fresh_db["child"].create(
{"id": int, "Parent_ID": int},
pk="id",
foreign_keys=[("Parent_ID", "parent", "Id")],
)
# ignore=True should treat this as already existing, not add a duplicate
fresh_db["child"].add_foreign_key("parent_id", "parent", "id", ignore=True)
assert len(fresh_db["child"].foreign_keys) == 1
def test_add_column_fk_col_case_insensitive(fresh_db):
fresh_db["parent"].create({"Id": int}, pk="Id")
fresh_db["child"].create({"id": int}, pk="id")
fresh_db["child"].add_column("parent_id", int, fk="parent", fk_col="id")
fks = fresh_db["child"].foreign_keys
assert len(fks) == 1
assert fks[0].other_column == "Id"
def test_extract_case_insensitive(fresh_db):
fresh_db["trees"].insert({"id": 1, "Species": "Palm"}, pk="id")
fresh_db["trees"].extract("species")
assert fresh_db["trees"].columns_dict == {"id": int, "Species_id": int}
assert list(fresh_db["Species"].rows) == [{"id": 1, "Species": "Palm"}]
def test_convert_multi_case_insensitive(fresh_db):
fresh_db["t"].insert({"id": 1, "Name": "Cleo"}, pk="id")
fresh_db["t"].convert("name", lambda v: {"upper": v.upper()}, multi=True)
assert list(fresh_db["t"].rows) == [{"id": 1, "Name": "Cleo", "upper": "CLEO"}]
def test_convert_output_case_insensitive(fresh_db):
fresh_db["t"].insert({"id": 1, "Name": "Cleo", "Upper": None}, pk="id")
fresh_db["t"].convert("name", lambda v: v.upper(), output="upper")
assert list(fresh_db["t"].rows) == [{"id": 1, "Name": "Cleo", "Upper": "CLEO"}]
def test_create_table_sql_pk_case_insensitive(fresh_db):
fresh_db["t"].create({"Id": int, "Name": str}, pk="id")
# Should not have created an extra lowercase "id" column
assert fresh_db["t"].columns_dict == {"Id": int, "Name": str}
assert fresh_db["t"].pks == ["Id"]
def test_create_table_not_null_and_defaults_case_insensitive(fresh_db):
fresh_db["t"].create(
{"Name": str, "Age": int}, not_null={"name"}, defaults={"age": 1}
)
columns = {c.name: c for c in fresh_db["t"].columns}
assert columns["Name"].notnull
assert fresh_db["t"].default_values == {"Age": 1}
def test_create_table_foreign_keys_case_insensitive(fresh_db):
fresh_db["parent"].create({"Id": int}, pk="Id")
fresh_db["child"].create(
{"id": int, "Parent_ID": int},
pk="id",
foreign_keys=[("parent_id", "parent", "id")],
)
fks = fresh_db["child"].foreign_keys
assert fks == [
ForeignKey(
table="child", column="Parent_ID", other_table="parent", other_column="Id"
)
]

48
tests/test_constructor.py
View file

@ -1,6 +1,8 @@
from sqlite_utils import Database
from sqlite_utils.db import TransactionError
from sqlite_utils.utils import sqlite3
import pytest
import sys
def test_recursive_triggers():
@ -29,6 +31,24 @@ def test_sqlite_version():
assert actual == as_string
def test_database_context_manager(tmpdir):
path = str(tmpdir / "test.db")
with Database(path) as db:
db["t"].insert({"id": 1})
# Raw writes commit automatically too
db.execute("insert into t (id) values (2)")
# An explicitly opened transaction left uncommitted on purpose:
db.begin()
db.execute("insert into t (id) values (3)")
# The connection is closed...
with pytest.raises(sqlite3.ProgrammingError):
db.execute("select 1")
# ... and the open explicit transaction was rolled back, not committed
db2 = Database(path)
assert [r["id"] for r in db2["t"].rows] == [1, 2]
db2.close()
@pytest.mark.parametrize("memory", [True, False])
def test_database_close(tmpdir, memory):
if memory:
@ -70,3 +90,31 @@ def test_memory_attribute_for_existing_connection():
db = Database(conn)
assert db.memory is False
assert db.memory_name is None
@pytest.mark.skipif(
sys.version_info < (3, 12),
reason="sqlite3.connect(autocommit=) requires Python 3.12",
)
@pytest.mark.parametrize("autocommit", [True, False])
def test_autocommit_connections_are_rejected(tmpdir, autocommit):
# These connection modes break commit()/rollback() in ways that
# silently lose data, so the constructor refuses them
conn = sqlite3.connect(str(tmpdir / "test.db"), autocommit=autocommit)
with pytest.raises(TransactionError):
Database(conn)
conn.close()
@pytest.mark.skipif(
sys.version_info < (3, 12),
reason="sqlite3.LEGACY_TRANSACTION_CONTROL requires Python 3.12",
)
def test_legacy_transaction_control_connection_is_accepted(tmpdir):
conn = sqlite3.connect(
str(tmpdir / "test.db"), autocommit=sqlite3.LEGACY_TRANSACTION_CONTROL
)
db = Database(conn)
db["t"].insert({"id": 1}, pk="id")
assert [r["id"] for r in db["t"].rows] == [1]
db.close()

2
tests/test_convert.py
View file

@ -77,7 +77,7 @@ def test_convert_output(fresh_db, drop, expected):
def test_convert_output_multiple_column_error(fresh_db):
table = fresh_db["table"]
with pytest.raises(AssertionError) as excinfo:
with pytest.raises(ValueError) as excinfo:
table.convert(["title", "other"], lambda v: v, output="out")
assert "output= can only be used with a single column" in str(excinfo.value)

179
tests/test_create.py
View file

@ -3,6 +3,7 @@ from sqlite_utils.db import (
Database,
DescIndex,
AlterError,
InvalidColumns,
NoObviousTable,
OperationalError,
ForeignKey,
@ -20,7 +21,6 @@ import pathlib
import pytest
import uuid
try:
import pandas as pd # type: ignore
except ImportError:
@ -109,7 +109,7 @@ def test_create_table_with_defaults(fresh_db):
def test_create_table_with_bad_not_null(fresh_db):
with pytest.raises(AssertionError):
with pytest.raises(ValueError):
fresh_db.create_table(
"players", {"name": str, "score": int}, not_null={"mouse"}
)
@ -244,11 +244,11 @@ def test_create_table_column_order(fresh_db, use_table_factory):
# If you specify a column that doesn't point to a table, you get an error:
(("one_id", "two_id", "three_id"), NoObviousTable),
# Tuples of the wrong length get an error:
((("one_id", "one", "id", "five"), ("two_id", "two", "id")), AssertionError),
((("one_id", "one", "id", "five"), ("two_id", "two", "id")), ValueError),
# Likewise a bad column:
((("one_id", "one", "id2"),), AlterError),
# Or a list of dicts
(({"one_id": "one"},), AssertionError),
(({"one_id": "one"},), ValueError),
),
)
@pytest.mark.parametrize("use_table_factory", [True, False])
@ -701,7 +701,7 @@ def test_bulk_insert_more_than_999_values(fresh_db):
def test_error_if_more_than_999_columns(fresh_db, num_columns, should_error):
record = dict([("c{}".format(i), i) for i in range(num_columns)])
if should_error:
with pytest.raises(AssertionError):
with pytest.raises(ValueError):
fresh_db["big"].insert(record)
else:
fresh_db["big"].insert(record)
@ -926,6 +926,58 @@ def test_insert_thousands_adds_extra_columns_after_first_100_with_alter(fresh_db
assert rows == [{"i": 101, "word": None, "extra": "Should trigger ALTER"}]
@pytest.mark.parametrize("num_rows", (0, 1, 2, 3, 10))
def test_insert_all_pk_not_in_records_raises(fresh_db, num_rows):
# https://github.com/simonw/sqlite-utils/issues/732
fresh_db.conn.execute("CREATE TABLE t (a TEXT, b INT, PRIMARY KEY (a, b))")
rows = [{"a": "x{}".format(i), "b": i} for i in range(num_rows)]
with pytest.raises(InvalidColumns) as ex:
fresh_db["t"].insert_all(rows, pk="not_a_column")
assert ex.value.args == (
"Invalid primary key column ['not_a_column'] for table t with columns ['a', 'b']",
)
assert fresh_db["t"].count == 0
@pytest.mark.parametrize("num_rows", (1, 2, 3, 10))
def test_insert_all_pk_not_in_records_alter_raises(fresh_db, num_rows):
# With alter=True the check is deferred until the record keys are
# known - a pk column that is in neither the table nor the records
# still raises
fresh_db.conn.execute("CREATE TABLE t (a TEXT, b INT, PRIMARY KEY (a, b))")
rows = [{"a": "x{}".format(i), "b": i} for i in range(num_rows)]
with pytest.raises(InvalidColumns) as ex:
fresh_db["t"].insert_all(rows, pk="not_a_column", alter=True)
assert ex.value.args == (
"Invalid primary key column ['not_a_column'] for table t with columns ['a', 'b']",
)
assert fresh_db["t"].count == 0
def test_insert_pk_in_records_with_alter_adds_column(fresh_db):
# 3.x allowed insert(pk=..., alter=True) to add the pk column from the
# records - the InvalidColumns check must not fire in that case
fresh_db["t"].insert({"a": 1})
fresh_db["t"].insert({"id": 5, "a": 2}, pk="id", alter=True)
assert fresh_db["t"].columns_dict.keys() == {"a", "id"}
assert list(fresh_db.query("select * from t order by a")) == [
{"a": 1, "id": None},
{"a": 2, "id": 5},
]
def test_insert_all_invalid_pk_alter_empty_records_is_noop(fresh_db):
# With alter=True the pk check needs record keys, so an empty iterator
# returns without error - matching the 3.x no-op for empty inserts
fresh_db.conn.execute("CREATE TABLE t (a TEXT)")
fresh_db["t"].insert_all([], pk="not_a_column", alter=True)
assert fresh_db["t"].count == 0
def test_insert_ignore(fresh_db):
fresh_db["test"].insert({"id": 1, "bar": 2}, pk="id")
# Should raise an error if we try this again
@ -938,6 +990,114 @@ def test_insert_ignore(fresh_db):
assert rows == [{"id": 1, "bar": 2}]
def test_insert_ignore_reports_existing_row(fresh_db):
# An ignored insert (row already exists) should point last_rowid and
# last_pk at the existing conflicting row - see the Datasette insert API
fresh_db["docs"].insert({"id": 1, "title": "Exists"}, pk="id")
# Insert a conflicting row with ignore=True and no explicit pk=
table = fresh_db["docs"].insert({"id": 1, "title": "One"}, ignore=True)
assert table.last_rowid == 1
assert table.last_pk == 1
assert list(fresh_db["docs"].rows_where("rowid = ?", [table.last_rowid])) == [
{"id": 1, "title": "Exists"}
]
@pytest.mark.parametrize("rowid_alias", ("rowid", "_rowid_", "oid"))
@pytest.mark.parametrize("method", ("upsert", "insert_replace", "insert_ignore"))
def test_pk_rowid_alias_on_rowid_table(fresh_db, rowid_alias, method):
# rowid and its aliases are valid primary keys for a rowid table even
# though they are not listed among the table's columns - see the Datasette
# upsert API against tables without an explicit primary key
fresh_db["t"].insert({"title": "Hello"})
assert fresh_db["t"].pks == ["rowid"]
record = {rowid_alias: 1, "title": "Updated"}
if method == "upsert":
table = fresh_db["t"].upsert(record, pk=rowid_alias)
elif method == "insert_replace":
table = fresh_db["t"].insert(record, pk=rowid_alias, replace=True)
else:
table = fresh_db["t"].insert(record, pk=rowid_alias, ignore=True)
assert table.last_pk == 1
expected_title = "Hello" if method == "insert_ignore" else "Updated"
assert list(fresh_db["t"].rows) == [{"title": expected_title}]
def test_insert_ignore_reports_existing_row_compound_pk(fresh_db):
# Compound primary key variant of the ignored-insert lookup
fresh_db["t"].insert_all([{"a": 1, "b": 2, "note": "first"}], pk=("a", "b"))
table = fresh_db["t"].insert(
{"a": 1, "b": 2, "note": "second"}, pk=("a", "b"), ignore=True
)
assert table.last_pk == (1, 2)
assert list(fresh_db["t"].rows_where("rowid = ?", [table.last_rowid])) == [
{"a": 1, "b": 2, "note": "first"}
]
def test_insert_ignore_reports_existing_row_list_mode(fresh_db):
# List-based iteration variant of the ignored-insert lookup
fresh_db["t"].insert_all([["id", "title"], [1, "first"]], pk="id")
table = fresh_db["t"].insert_all(
[["id", "title"], [1, "second"]], pk="id", ignore=True
)
assert table.last_pk == 1
assert table.last_rowid == 1
assert list(fresh_db["t"].rows) == [{"id": 1, "title": "first"}]
def test_insert_ignore_hash_id_reports_pk(fresh_db):
# With hash_id the pk is the computed hash; the original record has no id
# column to look up so last_rowid is left unset
first = fresh_db["dogs"].insert({"name": "Cleo"}, hash_id="id")
table = fresh_db["dogs"].insert({"name": "Cleo"}, hash_id="id", ignore=True)
assert table.last_pk == first.last_pk
assert table.last_rowid is None
assert fresh_db["dogs"].count == 1
def test_insert_ignore_unresolvable_conflict_leaves_pk_unset(fresh_db):
# When the conflict cannot be resolved to a primary key lookup, last_pk and
# last_rowid are left unset rather than reporting a misleading value
# rowid table with a UNIQUE column and no primary key: no pk to look up
fresh_db["u"].db.execute("create table u (title text unique)")
fresh_db["u"].insert({"title": "x"})
table = fresh_db["u"].insert({"title": "x"}, ignore=True)
assert table.last_pk is None
assert table.last_rowid is None
assert fresh_db["u"].count == 1
# Conflict on a UNIQUE column other than the primary key: the pk value from
# the record does not match the existing row, so the lookup finds nothing
fresh_db["docs"].db.execute(
"create table docs (id integer primary key, email text unique)"
)
fresh_db["docs"].insert({"id": 1, "email": "a"}, pk="id")
table = fresh_db["docs"].insert({"id": 2, "email": "a"}, ignore=True)
assert table.last_pk is None
assert table.last_rowid is None
assert fresh_db["docs"].count == 1
def test_insert_ignore_with_pk_after_other_table_insert(fresh_db):
# https://github.com/simonw/sqlite-utils/issues/554
user = {"id": "abc", "name": "david"}
fresh_db["users"].insert(user, pk="id")
fresh_db["comments"].insert_all(
[
{"id": "def", "text": "ok"},
{"id": "ghi", "text": "great"},
],
)
table = fresh_db["users"].insert(user, pk="id", ignore=True)
assert table.last_pk == "abc"
assert list(fresh_db["users"].rows) == [user]
def test_insert_hash_id(fresh_db):
dogs = fresh_db["dogs"]
id = dogs.insert({"name": "Cleo", "twitter": "cleopaws"}, hash_id="id").last_pk
@ -1062,7 +1222,7 @@ def test_create_table_numpy(fresh_db):
def test_cannot_provide_both_filename_and_memory():
with pytest.raises(
AssertionError, match="Either specify a filename_or_conn or pass memory=True"
ValueError, match="Either specify a filename_or_conn or pass memory=True"
):
Database("/tmp/foo.db", memory=True)
@ -1215,7 +1375,7 @@ def test_create_if_not_exists(fresh_db):
def test_create_if_no_columns(fresh_db):
with pytest.raises(AssertionError) as error:
with pytest.raises(ValueError) as error:
fresh_db["t"].create({})
assert error.value.args[0] == "Tables must have at least one column"
@ -1383,7 +1543,10 @@ def test_bad_table_and_view_exceptions(fresh_db):
assert ex.value.args[0] == "Table v is actually a view"
with pytest.raises(NoView) as ex2:
fresh_db.view("t")
assert ex2.value.args[0] == "View t does not exist"
assert ex2.value.args[0] == "View t does not exist - t is a table"
with pytest.raises(NoView) as ex3:
fresh_db.view("missing")
assert ex3.value.args[0] == "View missing does not exist"
# Tests for issue #655: Table configuration should be stored in _defaults

2
tests/test_create_view.py
View file

@ -15,7 +15,7 @@ def test_create_view_error(fresh_db):
def test_create_view_only_arrow_one_param(fresh_db):
with pytest.raises(AssertionError):
with pytest.raises(ValueError):
fresh_db.create_view("bar", "select 1 + 2", ignore=True, replace=True)

27
tests/test_default_value.py
View file

@ -1,6 +1,5 @@
import pytest
EXAMPLES = [
("TEXT DEFAULT 'foo'", "'foo'", "'foo'"),
("TEXT DEFAULT 'foo)'", "'foo)'", "'foo)'"),
@ -22,6 +21,11 @@ EXAMPLES = [
# Strings
("TEXT DEFAULT 'CURRENT_TIMESTAMP'", "'CURRENT_TIMESTAMP'", "'CURRENT_TIMESTAMP'"),
('TEXT DEFAULT "CURRENT_TIMESTAMP"', '"CURRENT_TIMESTAMP"', '"CURRENT_TIMESTAMP"'),
# Boolean and null keyword literals must stay unquoted
("INTEGER DEFAULT TRUE", "TRUE", "TRUE"),
("INTEGER DEFAULT FALSE", "FALSE", "FALSE"),
("INTEGER DEFAULT true", "true", "true"),
("TEXT DEFAULT NULL", "NULL", "NULL"),
]
@ -32,3 +36,24 @@ def test_quote_default_value(fresh_db, column_def, initial_value, expected_value
assert expected_value == fresh_db.quote_default_value(
fresh_db["foo"].columns[0].default_value
)
def test_insert_empty_record_uses_default_values(fresh_db):
fresh_db.execute("""
CREATE TABLE has_defaults (
id INTEGER PRIMARY KEY,
name TEXT,
timestamp TEXT DEFAULT CURRENT_TIMESTAMP,
is_active INTEGER NOT NULL DEFAULT 1
)
""")
table = fresh_db["has_defaults"]
table.insert({})
rows = list(table.rows)
assert len(rows) == 1
assert rows[0]["id"] == 1
assert rows[0]["name"] is None
assert rows[0]["timestamp"] is not None
assert rows[0]["is_active"] == 1

18
tests/test_delete.py
View file

@ -1,3 +1,6 @@
import sqlite_utils
def test_delete_rowid_table(fresh_db):
table = fresh_db["table"]
table.insert({"foo": 1}).last_pk
@ -32,6 +35,21 @@ def test_delete_where_all(fresh_db):
assert table.count == 0
def test_delete_where_commits(tmpdir):
path = str(tmpdir / "test.db")
db = sqlite_utils.Database(path)
db["table"].insert_all([{"id": i} for i in range(5)], pk="id")
db["table"].delete_where("id > ?", [2])
# The connection must not be left inside an open transaction,
# otherwise subsequent atomic() blocks never commit either
assert not db.conn.in_transaction
db["table"].insert({"id": 100})
db.close()
db2 = sqlite_utils.Database(path)
assert [r["id"] for r in db2["table"].rows] == [0, 1, 2, 100]
db2.close()
def test_delete_where_analyze(fresh_db):
table = fresh_db["table"]
table.insert_all(({"id": i, "i": i} for i in range(10)), pk="id")

6
tests/test_duplicate.py
View file

@ -5,14 +5,12 @@ import pytest
def test_duplicate(fresh_db):
# Create table using native Sqlite statement:
fresh_db.execute(
"""CREATE TABLE "table1" (
fresh_db.execute("""CREATE TABLE "table1" (
"text_col" TEXT,
"real_col" REAL,
"int_col" INTEGER,
"bool_col" INTEGER,
"datetime_col" TEXT)"""
)
"datetime_col" TEXT)""")
# Insert one row of mock data:
dt = datetime.datetime.now()
data = {

120
tests/test_extract.py
View file

@ -126,9 +126,7 @@ def test_extract_rowid_table(fresh_db):
' "common_name_latin_name_id" INTEGER REFERENCES "common_name_latin_name"("id")\n'
")"
)
assert (
fresh_db.execute(
"""
assert fresh_db.execute("""
select
tree.name,
common_name_latin_name.common_name,
@ -136,10 +134,7 @@ def test_extract_rowid_table(fresh_db):
from tree
join common_name_latin_name
on tree.common_name_latin_name_id = common_name_latin_name.id
"""
).fetchall()
== [("Tree 1", "Palm", "Arecaceae")]
)
""").fetchall() == [("Tree 1", "Palm", "Arecaceae")]
def test_reuse_lookup_table(fresh_db):
@ -194,9 +189,116 @@ def test_extract_works_with_null_values(fresh_db):
)
assert list(fresh_db["listens"].rows) == [
{"id": 1, "track_title": "foo", "album_id": 1},
{"id": 2, "track_title": "baz", "album_id": 2},
{"id": 2, "track_title": "baz", "album_id": None},
]
assert list(fresh_db["albums"].rows) == [
{"id": 1, "album_title": "bar"},
{"id": 2, "album_title": None},
]
def test_extract_null_values_single_column(fresh_db):
# https://github.com/simonw/sqlite-utils/issues/186
fresh_db["species"].insert({"id": 1, "species": "Wolf"}, pk="id")
fresh_db["individuals"].insert_all(
[
{"id": 10, "name": "Terriana", "species": "Fox"},
{"id": 11, "name": "Spenidorm", "species": None},
{"id": 12, "name": "Grantheim", "species": "Wolf"},
{"id": 13, "name": "Turnutopia", "species": None},
{"id": 14, "name": "Wargal", "species": "Wolf"},
],
pk="id",
)
fresh_db["individuals"].extract("species")
# No null row should have been added to species
assert list(fresh_db["species"].rows) == [
{"id": 1, "species": "Wolf"},
{"id": 2, "species": "Fox"},
]
assert list(fresh_db["individuals"].rows) == [
{"id": 10, "name": "Terriana", "species_id": 2},
{"id": 11, "name": "Spenidorm", "species_id": None},
{"id": 12, "name": "Grantheim", "species_id": 1},
{"id": 13, "name": "Turnutopia", "species_id": None},
{"id": 14, "name": "Wargal", "species_id": 1},
]
def test_extract_null_values_multiple_columns(fresh_db):
# A row should be extracted if at least one column is not null -
# only rows where ALL extracted columns are null are left alone
fresh_db["circulation"].insert_all(
[
{"id": 1, "title": "title one", "creator": "creator one", "year": 2018},
{"id": 2, "title": "title two", "creator": None, "year": 2019},
{"id": 3, "title": None, "creator": None, "year": 2020},
{"id": 4, "title": None, "creator": None, "year": 2021},
],
pk="id",
)
fresh_db["circulation"].extract(
["title", "creator"], table="books", fk_column="book_id"
)
assert list(fresh_db["books"].rows) == [
{"id": 1, "title": "title one", "creator": "creator one"},
{"id": 2, "title": "title two", "creator": None},
]
assert list(fresh_db["circulation"].rows) == [
{"id": 1, "book_id": 1, "year": 2018},
{"id": 2, "book_id": 2, "year": 2019},
{"id": 3, "book_id": None, "year": 2020},
{"id": 4, "book_id": None, "year": 2021},
]
def test_extract_null_values_existing_lookup_table_with_null_row(fresh_db):
# Even if the lookup table already contains an all-null row, rows where
# every extracted column is null should keep a null foreign key
fresh_db["species"].insert({"id": 1, "species": None}, pk="id")
fresh_db["individuals"].insert_all(
[
{"id": 10, "name": "Terriana", "species": "Fox"},
{"id": 11, "name": "Spenidorm", "species": None},
],
pk="id",
)
fresh_db["individuals"].extract("species")
assert list(fresh_db["species"].rows) == [
{"id": 1, "species": None},
{"id": 2, "species": "Fox"},
]
assert list(fresh_db["individuals"].rows) == [
{"id": 10, "name": "Terriana", "species_id": 2},
{"id": 11, "name": "Spenidorm", "species_id": None},
]
def test_extract_repeated_into_shared_lookup_with_nulls(fresh_db):
# Unique indexes treat NULLs as distinct, so INSERT OR IGNORE alone
# cannot dedupe NULL-containing rows against the existing lookup
# table - extracting a second table into the same lookup previously
# inserted duplicate rows that nothing pointed to
fresh_db["t1"].insert_all(
[
{"id": 1, "species": None, "common": "X"},
{"id": 2, "species": "Oak", "common": "Oak"},
],
pk="id",
)
fresh_db["t2"].insert_all([{"id": 1, "species": None, "common": "X"}], pk="id")
fresh_db["t1"].extract(["species", "common"], table="lk")
fresh_db["t2"].extract(["species", "common"], table="lk")
assert fresh_db["lk"].count == 2
# Both tables point at the same lookup row
t1_fk = fresh_db.execute("select lk_id from t1 where id = 1").fetchone()[0]
t2_fk = fresh_db.execute("select lk_id from t2 where id = 1").fetchone()[0]
assert t1_fk == t2_fk
def test_extract_repeated_into_shared_lookup_no_nulls(fresh_db):
# Non-NULL rows were already deduped by the unique index - keep it so
fresh_db["t1"].insert_all([{"id": 1, "species": "Oak"}], pk="id")
fresh_db["t2"].insert_all([{"id": 1, "species": "Oak"}], pk="id")
fresh_db["t1"].extract(["species"], table="lk")
fresh_db["t2"].extract(["species"], table="lk")
assert fresh_db["lk"].count == 1

48
tests/test_extracts.py
View file

@ -67,3 +67,51 @@ def test_extracts(fresh_db, kwargs, expected_table, use_table_factory):
{"id": 2, "species_id": 1},
{"id": 3, "species_id": 2},
] == list(fresh_db["Trees"].rows)
def test_extracts_null_values(fresh_db):
# https://github.com/simonw/sqlite-utils/issues/186
# Null values should stay null, not be extracted into the lookup table
fresh_db["Trees"].insert_all(
[
{"id": 1, "species_id": "Oak"},
{"id": 2, "species_id": None},
{"id": 3, "species_id": "Palm"},
{"id": 4, "species_id": None},
],
extracts={"species_id": "Species"},
)
assert list(fresh_db["Species"].rows) == [
{"id": 1, "value": "Oak"},
{"id": 2, "value": "Palm"},
]
assert list(fresh_db["Trees"].rows) == [
{"id": 1, "species_id": 1},
{"id": 2, "species_id": None},
{"id": 3, "species_id": 2},
{"id": 4, "species_id": None},
]
def test_extracts_null_values_list_mode(fresh_db):
# Same as test_extracts_null_values but for list-based records
fresh_db["Trees"].insert_all(
[
["id", "species_id"],
[1, "Oak"],
[2, None],
[3, "Palm"],
[4, None],
],
extracts={"species_id": "Species"},
)
assert list(fresh_db["Species"].rows) == [
{"id": 1, "value": "Oak"},
{"id": 2, "value": "Palm"},
]
assert list(fresh_db["Trees"].rows) == [
{"id": 1, "species_id": 1},
{"id": 2, "species_id": None},
{"id": 3, "species_id": 2},
{"id": 4, "species_id": None},
]

691
tests/test_foreign_keys.py Normal file
View file

@ -0,0 +1,691 @@
"""Tests for compound (multi-column) foreign keys - issue #594."""
import pytest
from sqlite_utils import Database
from sqlite_utils.db import AlterError, ForeignKey
from sqlite_utils.utils import sqlite3
COMPOUND_SCHEMA = """
CREATE TABLE departments (
campus_name TEXT NOT NULL,
dept_code TEXT NOT NULL,
dept_name TEXT,
PRIMARY KEY (campus_name, dept_code)
);
CREATE TABLE courses (
course_code TEXT PRIMARY KEY,
course_name TEXT,
campus_name TEXT NOT NULL,
dept_code TEXT NOT NULL,
FOREIGN KEY (campus_name, dept_code)
REFERENCES departments(campus_name, dept_code)
);
"""
@pytest.fixture
def compound_db():
db = Database(memory=True)
db.executescript(COMPOUND_SCHEMA)
return db
def test_compound_foreign_key(compound_db):
fks = compound_db["courses"].foreign_keys
assert len(fks) == 1
fk = fks[0]
assert fk.is_compound is True
assert fk.table == "courses"
assert fk.other_table == "departments"
assert fk.columns == ("campus_name", "dept_code")
assert fk.other_columns == ("campus_name", "dept_code")
# Scalar column/other_column can't sensibly hold a compound key
assert fk.column is None
assert fk.other_column is None
def test_single_foreign_key_gets_columns_fields(fresh_db):
fresh_db["authors"].insert({"id": 1, "name": "Sally"}, pk="id")
fresh_db["books"].insert({"title": "Hedgehogs", "author_id": 1})
fresh_db["books"].add_foreign_key("author_id", "authors", "id")
fk = fresh_db["books"].foreign_keys[0]
assert fk.is_compound is False
assert fk.column == "author_id"
assert fk.other_column == "id"
assert fk.columns == ("author_id",)
assert fk.other_columns == ("id",)
def test_foreign_key_no_longer_unpacks_as_tuple(fresh_db):
# Clean break in 4.0: ForeignKey is a dataclass, not a namedtuple, so the
# old tuple unpacking and indexing patterns now fail hard.
fresh_db["authors"].insert({"id": 1, "name": "Sally"}, pk="id")
fresh_db["books"].insert({"title": "Hedgehogs", "author_id": 1})
fresh_db["books"].add_foreign_key("author_id", "authors", "id")
fk = fresh_db["books"].foreign_keys[0]
with pytest.raises(TypeError):
table, column, other_table, other_column = fk
with pytest.raises(TypeError):
fk[0]
def test_foreign_keys_are_sortable(fresh_db):
fresh_db["authors"].insert({"id": 1, "name": "Sally"}, pk="id")
fresh_db["categories"].insert({"id": 1, "name": "Wildlife"}, pk="id")
fresh_db["books"].insert({"title": "Hedgehogs", "author_id": 1, "category_id": 1})
fresh_db.add_foreign_keys(
[
("books", "author_id", "authors", "id"),
("books", "category_id", "categories", "id"),
]
)
fks = sorted(fresh_db["books"].foreign_keys)
assert fks[0].column == "author_id"
assert fks[1].column == "category_id"
def test_mixed_compound_and_single_foreign_keys_are_sortable():
# compound FKs have column=None, which must not break sorting
# against single-column FKs (None < str raises TypeError)
db = Database(memory=True)
db.executescript("""
CREATE TABLE departments (
campus_name TEXT NOT NULL,
dept_code TEXT NOT NULL,
PRIMARY KEY (campus_name, dept_code)
);
CREATE TABLE accreditations (id INTEGER PRIMARY KEY);
CREATE TABLE courses (
course_code TEXT PRIMARY KEY,
campus_name TEXT NOT NULL,
dept_code TEXT NOT NULL,
accreditation_id INTEGER REFERENCES accreditations(id),
FOREIGN KEY (campus_name, dept_code)
REFERENCES departments(campus_name, dept_code)
);
""")
fks = db["courses"].foreign_keys
assert len(fks) == 2
assert {fk.is_compound for fk in fks} == {True, False}
fks_sorted = sorted(fks)
assert fks_sorted[0].other_table == "accreditations"
assert fks_sorted[1].other_table == "departments"
@pytest.fixture
def departments_db():
db = Database(memory=True)
db.create_table(
"departments",
{"campus_name": str, "dept_code": str, "dept_name": str},
pk=("campus_name", "dept_code"),
)
return db
EXPECTED_COURSES_SCHEMA = (
'CREATE TABLE "courses" (\n'
' "course_code" TEXT PRIMARY KEY,\n'
' "campus_name" TEXT,\n'
' "dept_code" TEXT,\n'
' FOREIGN KEY ("campus_name", "dept_code") '
'REFERENCES "departments"("campus_name", "dept_code")\n'
")"
)
@pytest.mark.parametrize(
"foreign_keys",
(
[
ForeignKey(
table="courses",
column=None,
other_table="departments",
other_column=None,
columns=("campus_name", "dept_code"),
other_columns=("campus_name", "dept_code"),
is_compound=True,
)
],
[(("campus_name", "dept_code"), "departments", ("campus_name", "dept_code"))],
# Two-item form guesses the other table's primary key:
[(("campus_name", "dept_code"), "departments")],
# Lists work too, though tuples are the documented form:
[(["campus_name", "dept_code"], "departments", ["campus_name", "dept_code"])],
),
)
def test_create_table_with_compound_foreign_key(departments_db, foreign_keys):
departments_db.create_table(
"courses",
{"course_code": str, "campus_name": str, "dept_code": str},
pk="course_code",
foreign_keys=foreign_keys,
)
assert departments_db["courses"].schema == EXPECTED_COURSES_SCHEMA
fks = departments_db["courses"].foreign_keys
assert len(fks) == 1
fk = fks[0]
assert fk.is_compound is True
assert fk.columns == ("campus_name", "dept_code")
assert fk.other_table == "departments"
assert fk.other_columns == ("campus_name", "dept_code")
def test_create_table_compound_foreign_key_enforced(departments_db):
departments_db.execute("PRAGMA foreign_keys = ON")
departments_db.create_table(
"courses",
{"course_code": str, "campus_name": str, "dept_code": str},
pk="course_code",
foreign_keys=[(("campus_name", "dept_code"), "departments")],
)
departments_db["departments"].insert(
{"campus_name": "Berkeley", "dept_code": "CS", "dept_name": "Computer Science"}
)
departments_db["courses"].insert(
{"course_code": "CS101", "campus_name": "Berkeley", "dept_code": "CS"}
)
with pytest.raises(sqlite3.IntegrityError):
departments_db.execute(
"insert into courses (course_code, campus_name, dept_code) "
"values ('X1', 'Nowhere', 'NOPE')"
)
def test_create_table_compound_foreign_key_missing_other_column(departments_db):
with pytest.raises(AlterError):
departments_db.create_table(
"courses",
{"course_code": str, "campus_name": str, "dept_code": str},
pk="course_code",
foreign_keys=[
(("campus_name", "dept_code"), "departments", ("campus_name", "nope"))
],
)
def test_transform_preserves_compound_foreign_key(compound_db):
compound_db["courses"].transform(rename={"course_name": "title"})
fks = compound_db["courses"].foreign_keys
assert len(fks) == 1
fk = fks[0]
assert fk.is_compound is True
assert fk.columns == ("campus_name", "dept_code")
assert fk.other_table == "departments"
assert fk.other_columns == ("campus_name", "dept_code")
def test_transform_rename_member_column_updates_compound_foreign_key(compound_db):
compound_db["courses"].transform(rename={"campus_name": "campus"})
fks = compound_db["courses"].foreign_keys
assert len(fks) == 1
fk = fks[0]
assert fk.is_compound is True
assert fk.columns == ("campus", "dept_code")
# Referenced columns in the other table are unchanged
assert fk.other_columns == ("campus_name", "dept_code")
def test_transform_drop_member_column_drops_compound_foreign_key(compound_db):
# Matches single-column behavior: dropping the column silently
# drops the foreign key that used it
compound_db["courses"].transform(drop={"dept_code"})
assert compound_db["courses"].foreign_keys == []
assert "FOREIGN KEY" not in compound_db["courses"].schema
@pytest.mark.parametrize(
"drop_foreign_keys",
(
# A bare column name matches any foreign key it participates in:
["campus_name"],
# A tuple must match the full compound key:
[("campus_name", "dept_code")],
),
)
def test_transform_drop_compound_foreign_key(compound_db, drop_foreign_keys):
compound_db["courses"].transform(drop_foreign_keys=drop_foreign_keys)
assert compound_db["courses"].foreign_keys == []
# The columns themselves survive
assert {"campus_name", "dept_code"} <= set(
compound_db["courses"].columns_dict.keys()
)
@pytest.fixture
def courses_db(departments_db):
departments_db.create_table(
"courses",
{"course_code": str, "campus_name": str, "dept_code": str},
pk="course_code",
)
return departments_db
def test_add_compound_foreign_key(courses_db):
t = courses_db["courses"].add_foreign_key(
("campus_name", "dept_code"), "departments", ("campus_name", "dept_code")
)
# Returns self
assert t.name == "courses"
fks = courses_db["courses"].foreign_keys
assert len(fks) == 1
fk = fks[0]
assert fk.is_compound is True
assert fk.columns == ("campus_name", "dept_code")
assert fk.other_table == "departments"
assert fk.other_columns == ("campus_name", "dept_code")
def test_add_compound_foreign_key_guesses_other_columns(courses_db):
# Lists work here too, though tuples are the documented form
courses_db["courses"].add_foreign_key(["campus_name", "dept_code"], "departments")
fk = courses_db["courses"].foreign_keys[0]
assert fk.other_columns == ("campus_name", "dept_code")
def test_add_compound_foreign_key_error_if_already_exists(courses_db):
courses_db["courses"].add_foreign_key(("campus_name", "dept_code"), "departments")
with pytest.raises(AlterError) as ex:
courses_db["courses"].add_foreign_key(
("campus_name", "dept_code"), "departments"
)
assert "already exists" in ex.value.args[0]
# ignore=True should not raise
courses_db["courses"].add_foreign_key(
("campus_name", "dept_code"), "departments", ignore=True
)
def test_add_compound_foreign_key_error_if_column_missing(courses_db):
with pytest.raises(AlterError):
courses_db["courses"].add_foreign_key(("campus_name", "nope"), "departments")
def test_db_add_foreign_keys_compound(courses_db):
courses_db.add_foreign_keys(
[
(
"courses",
("campus_name", "dept_code"),
"departments",
("campus_name", "dept_code"),
)
]
)
fk = courses_db["courses"].foreign_keys[0]
assert fk.is_compound is True
assert fk.columns == ("campus_name", "dept_code")
def test_index_foreign_keys_compound_creates_composite_index(compound_db):
compound_db.index_foreign_keys()
index_columns = [i.columns for i in compound_db["courses"].indexes]
assert ["campus_name", "dept_code"] in index_columns
# No separate single-column indexes for the members
assert ["campus_name"] not in index_columns
assert ["dept_code"] not in index_columns
def test_foreign_key_captures_on_delete_and_on_update():
db = Database(memory=True)
db.executescript("""
CREATE TABLE authors (id INTEGER PRIMARY KEY);
CREATE TABLE books (
id INTEGER PRIMARY KEY,
author_id INTEGER REFERENCES authors(id)
ON DELETE CASCADE ON UPDATE RESTRICT
);
""")
fk = db["books"].foreign_keys[0]
assert fk.on_delete == "CASCADE"
assert fk.on_update == "RESTRICT"
def test_foreign_key_on_delete_defaults_to_no_action(fresh_db):
fresh_db["authors"].insert({"id": 1}, pk="id")
fresh_db["books"].insert({"id": 1, "author_id": 1}, pk="id")
fresh_db["books"].add_foreign_key("author_id", "authors", "id")
fk = fresh_db["books"].foreign_keys[0]
assert fk.on_delete == "NO ACTION"
assert fk.on_update == "NO ACTION"
def test_create_table_foreign_key_with_on_delete(fresh_db):
fresh_db["authors"].insert({"id": 1}, pk="id")
fresh_db.create_table(
"books",
{"id": int, "author_id": int},
pk="id",
foreign_keys=[
ForeignKey(
table="books",
column="author_id",
other_table="authors",
other_column="id",
on_delete="CASCADE",
)
],
)
assert "ON DELETE CASCADE" in fresh_db["books"].schema
assert fresh_db["books"].foreign_keys[0].on_delete == "CASCADE"
def test_transform_preserves_on_delete_cascade():
db = Database(memory=True)
db.executescript("""
CREATE TABLE authors (id INTEGER PRIMARY KEY);
CREATE TABLE books (
id INTEGER PRIMARY KEY,
title TEXT,
author_id INTEGER REFERENCES authors(id) ON DELETE CASCADE
);
""")
db["books"].transform(rename={"title": "book_title"})
fk = db["books"].foreign_keys[0]
assert fk.on_delete == "CASCADE"
assert fk.on_update == "NO ACTION"
assert "ON DELETE CASCADE" in db["books"].schema
def test_transform_preserves_compound_foreign_key_on_delete():
db = Database(memory=True)
db.executescript("""
CREATE TABLE departments (
campus_name TEXT NOT NULL,
dept_code TEXT NOT NULL,
PRIMARY KEY (campus_name, dept_code)
);
CREATE TABLE courses (
course_code TEXT PRIMARY KEY,
campus_name TEXT NOT NULL,
dept_code TEXT NOT NULL,
FOREIGN KEY (campus_name, dept_code)
REFERENCES departments(campus_name, dept_code) ON DELETE CASCADE
);
""")
db["courses"].transform(rename={"course_code": "code"})
fk = db["courses"].foreign_keys[0]
assert fk.is_compound is True
assert fk.on_delete == "CASCADE"
assert "ON DELETE CASCADE" in db["courses"].schema
def test_implicit_primary_key_reference_is_resolved():
# REFERENCES authors (no column) has "to" of None in the pragma -
# it should be resolved to the primary key of the other table
db = Database(memory=True)
db.executescript("""
CREATE TABLE authors (author_id INTEGER PRIMARY KEY);
CREATE TABLE books (
id INTEGER PRIMARY KEY,
author_id INTEGER REFERENCES authors
);
""")
fk = db["books"].foreign_keys[0]
assert fk.is_compound is False
assert fk.other_column == "author_id"
assert fk.other_columns == ("author_id",)
def test_implicit_compound_primary_key_reference_is_resolved():
db = Database(memory=True)
db.executescript("""
CREATE TABLE departments (
campus_name TEXT NOT NULL,
dept_code TEXT NOT NULL,
PRIMARY KEY (campus_name, dept_code)
);
CREATE TABLE courses (
course_code TEXT PRIMARY KEY,
campus_name TEXT NOT NULL,
dept_code TEXT NOT NULL,
FOREIGN KEY (campus_name, dept_code) REFERENCES departments
);
""")
fk = db["courses"].foreign_keys[0]
assert fk.is_compound is True
assert fk.other_columns == ("campus_name", "dept_code")
def test_foreign_key_normalizes_list_columns_to_tuples():
# Compound columns passed as lists are normalized to tuples, so they
# compare equal to introspected ForeignKeys
fk = ForeignKey(
table="courses",
column=None,
other_table="departments",
other_column=None,
columns=["campus_name", "dept_code"],
other_columns=["campus_name", "dept_code"],
is_compound=True,
)
assert fk.columns == ("campus_name", "dept_code")
assert fk.other_columns == ("campus_name", "dept_code")
def test_add_foreign_keys_preserves_actions(fresh_db):
# https://github.com/simonw/sqlite-utils/issues/594 review finding:
# ForeignKey objects passed to db.add_foreign_keys() were flattened
# to plain tuples, losing on_delete/on_update
fresh_db["authors"].insert({"id": 1}, pk="id")
fresh_db["books"].insert({"id": 1, "author_id": 1}, pk="id")
fresh_db.add_foreign_keys(
[ForeignKey("books", "author_id", "authors", "id", on_delete="CASCADE")]
)
fk = fresh_db["books"].foreign_keys[0]
assert fk.on_delete == "CASCADE"
assert "ON DELETE CASCADE" in fresh_db["books"].schema
def test_add_foreign_keys_preserves_actions_compound(courses_db):
courses_db.add_foreign_keys(
[
ForeignKey(
table="courses",
column=None,
other_table="departments",
other_column=None,
columns=("campus_name", "dept_code"),
other_columns=("campus_name", "dept_code"),
is_compound=True,
on_delete="CASCADE",
)
]
)
fk = courses_db["courses"].foreign_keys[0]
assert fk.is_compound is True
assert fk.on_delete == "CASCADE"
assert "ON DELETE CASCADE" in courses_db["courses"].schema
def test_add_foreign_key_on_delete_on_update(fresh_db):
fresh_db["authors"].insert({"id": 1}, pk="id")
fresh_db["books"].insert({"id": 1, "author_id": 1}, pk="id")
fresh_db["books"].add_foreign_key(
"author_id", "authors", "id", on_delete="CASCADE", on_update="RESTRICT"
)
fk = fresh_db["books"].foreign_keys[0]
assert fk.on_delete == "CASCADE"
assert fk.on_update == "RESTRICT"
assert "ON UPDATE RESTRICT ON DELETE CASCADE" in fresh_db["books"].schema
# The cascade should actually fire
fresh_db.execute("PRAGMA foreign_keys = ON")
fresh_db.execute("delete from authors where id = 1")
assert fresh_db["books"].count == 0
def test_add_compound_foreign_key_on_delete(courses_db):
courses_db["courses"].add_foreign_key(
("campus_name", "dept_code"), "departments", on_delete="SET NULL"
)
fk = courses_db["courses"].foreign_keys[0]
assert fk.is_compound is True
assert fk.on_delete == "SET NULL"
assert "ON DELETE SET NULL" in courses_db["courses"].schema
def test_implicit_compound_foreign_key_resolves_pk_declaration_order(fresh_db):
# The other table's PRIMARY KEY declares its columns in a different
# order to the table's column order. SQLite resolves the implicit
# "REFERENCES other" using PRIMARY KEY declaration order, so the
# introspected other_columns must too
fresh_db.execute("create table other (b text, a text, primary key (a, b))")
fresh_db.execute(
"create table child (x text, y text, foreign key (x, y) references other)"
)
fk = fresh_db["child"].foreign_keys[0]
assert fk.other_columns == ("a", "b")
def test_transform_implicit_compound_foreign_key_stays_valid(fresh_db):
# transform() rewrites the implicit FK with explicit columns - they
# must be in PRIMARY KEY declaration order or valid data fails the
# foreign key check with an IntegrityError
fresh_db.execute("create table other (b text, a text, primary key (a, b))")
fresh_db.execute(
"create table child (x text, y text, foreign key (x, y) references other)"
)
fresh_db.execute("PRAGMA foreign_keys = ON")
fresh_db["other"].insert({"a": "A", "b": "B"})
fresh_db["child"].insert({"x": "A", "y": "B"})
fresh_db["child"].transform(types={"x": str})
assert fresh_db["child"].foreign_keys[0].other_columns == ("a", "b")
# The constraint still points the right way around
fresh_db["child"].insert({"x": "A", "y": "B"})
with pytest.raises(sqlite3.IntegrityError):
fresh_db["child"].insert({"x": "B", "y": "A"})
def test_create_compound_foreign_key_guesses_pk_declaration_order(fresh_db):
fresh_db.execute("create table other (b text, a text, primary key (a, b))")
fresh_db["other"].insert({"a": "A", "b": "B"})
fresh_db["child"].create(
{"id": int, "x": str, "y": str},
pk="id",
foreign_keys=[(("x", "y"), "other")],
)
assert fresh_db["child"].foreign_keys[0].other_columns == ("a", "b")
fresh_db.execute("PRAGMA foreign_keys = ON")
fresh_db["child"].insert({"id": 1, "x": "A", "y": "B"})
with pytest.raises(sqlite3.IntegrityError):
fresh_db["child"].insert({"id": 2, "x": "B", "y": "A"})
def test_add_compound_foreign_key_guesses_pk_declaration_order(fresh_db):
fresh_db.execute("create table other (b text, a text, primary key (a, b))")
fresh_db["child"].insert({"id": 1, "x": "A", "y": "B"}, pk="id")
fresh_db["child"].add_foreign_key(("x", "y"), "other")
assert fresh_db["child"].foreign_keys[0].other_columns == ("a", "b")
def test_foreign_keys_are_hashable(fresh_db):
# set() over foreign_keys worked with the 3.x namedtuple and must
# keep working with the dataclass
fresh_db["p"].insert({"id": 1}, pk="id")
fresh_db["c"].insert(
{"id": 1, "pid": 1}, pk="id", foreign_keys=[("pid", "p", "id")]
)
fks = set(fresh_db["c"].foreign_keys)
assert len(fks) == 1
assert ForeignKey("c", "pid", "p", "id") in fks
# Usable as dict keys too
assert {fk: True for fk in fks}
def test_foreign_key_is_immutable():
import dataclasses
fk = ForeignKey("c", "pid", "p", "id")
with pytest.raises(dataclasses.FrozenInstanceError):
fk.table = "other"
def test_foreign_key_equality_and_hash_include_actions():
# Two foreign keys differing only in ON DELETE behavior are different
# constraints - they compare unequal and hash separately
plain = ForeignKey("c", "pid", "p", "id")
cascade = ForeignKey("c", "pid", "p", "id", on_delete="CASCADE")
assert plain != cascade
assert len({plain, cascade}) == 2
assert plain == ForeignKey("c", "pid", "p", "id")
def test_create_table_mixed_foreign_keys_list(fresh_db):
# 3.x accepted a mix of ForeignKey objects, tuples and bare column
# strings in foreign_keys= (ForeignKey was a namedtuple, so it passed
# the tuple check) - keep accepting the mix
fresh_db["authors"].insert({"id": 1}, pk="id")
fresh_db["publishers"].insert({"id": 1}, pk="id")
fresh_db["books"].create(
{"id": int, "author_id": int, "publisher_id": int},
pk="id",
foreign_keys=[
ForeignKey("books", "author_id", "authors", "id"),
("publisher_id", "publishers", "id"),
],
)
fks = {fk.column: fk.other_table for fk in fresh_db["books"].foreign_keys}
assert fks == {"author_id": "authors", "publisher_id": "publishers"}
def test_create_table_mixed_foreign_keys_with_string(fresh_db):
fresh_db["authors"].insert({"id": 1}, pk="id")
fresh_db["publishers"].insert({"id": 1}, pk="id")
fresh_db["books"].create(
{"id": int, "author_id": int, "publisher_id": int},
pk="id",
foreign_keys=[
"author_id", # bare column, table and column guessed
("publisher_id", "publishers", "id"),
],
)
fks = {fk.column: fk.other_table for fk in fresh_db["books"].foreign_keys}
assert fks == {"author_id": "authors", "publisher_id": "publishers"}
def test_add_foreign_keys_existing_with_different_actions_errors(fresh_db):
# Requesting an existing foreign key with different ON DELETE/ON UPDATE
# actions was silently skipped, dropping the requested change
fresh_db["authors"].insert({"id": 1}, pk="id")
fresh_db["books"].insert(
{"id": 1, "author_id": 1},
pk="id",
foreign_keys=[("author_id", "authors", "id")],
)
with pytest.raises(AlterError) as ex:
fresh_db.add_foreign_keys(
[ForeignKey("books", "author_id", "authors", "id", on_delete="CASCADE")]
)
assert "ON DELETE" in str(ex.value)
assert fresh_db["books"].foreign_keys[0].on_delete == "NO ACTION"
def test_add_foreign_keys_identical_existing_is_noop(fresh_db):
# An exact match, including actions, is silently skipped so repeated
# calls stay idempotent
fresh_db["authors"].insert({"id": 1}, pk="id")
fresh_db["books"].insert({"id": 1, "author_id": 1}, pk="id")
fresh_db["books"].add_foreign_key("author_id", "authors", "id", on_delete="CASCADE")
fresh_db.add_foreign_keys(
[ForeignKey("books", "author_id", "authors", "id", on_delete="CASCADE")]
)
fks = fresh_db["books"].foreign_keys
assert len(fks) == 1
assert fks[0].on_delete == "CASCADE"
def test_add_foreign_keys_compound_column_count_mismatch_errors(fresh_db):
# Previously the extra other-column was silently discarded, creating
# a single-column foreign key to just ("id")
fresh_db["departments"].insert(
{"campus": "north", "code": "cs"}, pk=("campus", "code")
)
fresh_db["courses"].insert({"id": 1, "campus": "north"}, pk="id")
with pytest.raises(ValueError) as ex:
fresh_db.add_foreign_keys(
[("courses", ("campus",), "departments", ("campus", "code"))]
)
assert "same number of columns" in str(ex.value)
assert fresh_db["courses"].foreign_keys == []

77
tests/test_fts.py
View file

@ -83,6 +83,20 @@ def test_enable_fts_escape_table_names(fresh_db):
assert [] == list(table.search("bar"))
def test_search_duplicate_columns_are_deduped(fresh_db):
# https://github.com/simonw/sqlite-utils/issues/624
table = fresh_db["t"]
table.insert_all(search_records)
table.enable_fts(["text", "country"], fts_version="FTS4")
rows = list(table.search("tanuki", columns=["text", "text"]))
assert rows == [
{
"text": "tanuki are running tricksters",
"text_2": "tanuki are running tricksters",
}
]
def test_search_limit_offset(fresh_db):
table = fresh_db["t"]
table.insert_all(search_records)
@ -336,6 +350,24 @@ def test_rebuild_fts(fresh_db):
assert len(rows2) == 2
@pytest.mark.parametrize("method", ["optimize", "rebuild_fts"])
def test_optimize_and_rebuild_fts_commit(tmpdir, method):
path = str(tmpdir / "test.db")
db = Database(path)
table = db["searchable"]
table.insert(search_records[0])
table.enable_fts(["text", "country"])
getattr(table, method)()
# The connection must not be left inside an open transaction,
# otherwise this and all subsequent writes are lost on close
assert not db.conn.in_transaction
table.insert(search_records[1])
db.close()
db2 = Database(path)
assert db2["searchable"].count == 2
db2.close()
@pytest.mark.parametrize("invalid_table", ["does_not_exist", "not_searchable"])
def test_rebuild_fts_invalid(fresh_db, invalid_table):
fresh_db["not_searchable"].insert({"foo": "bar"})
@ -420,12 +452,35 @@ def test_enable_fts_replace_does_nothing_if_args_the_same():
assert all(q[0].startswith("select ") for q in queries)
def test_enable_fts_error_message_on_views():
def test_enable_fts_replace_handles_legacy_bracket_quoted_content_table():
db = Database(memory=True)
db["books"].insert(
{
"id": 1,
"title": "Habits of Australian Marsupials",
"author": "Marlee Hawkins",
},
pk="id",
)
db.executescript("""
CREATE VIRTUAL TABLE [books_fts] USING FTS5 (
[title],
content=[books]
);
""")
db["books"].enable_fts(["title", "author"], replace=True)
assert db["books_fts"].columns_dict.keys() == {"title", "author"}
assert 'content="books"' in db["books_fts"].schema
def test_view_has_no_enable_fts():
db = Database(memory=True)
db.create_view("hello", "select 1 + 1")
with pytest.raises(NotImplementedError) as e:
db["hello"].enable_fts() # type: ignore[call-arg]
assert e.value.args[0] == "enable_fts() is supported on tables but not on views"
# Views deliberately do not have an enable_fts() method
with pytest.raises(AttributeError):
db["hello"].enable_fts() # type: ignore[union-attr]
@pytest.mark.parametrize(
@ -677,3 +732,17 @@ def test_search_quote(fresh_db):
list(table.search(query))
# No exception with quote=True
list(table.search(query, quote=True))
def test_enable_fts_cli_on_view_errors(tmpdir):
db_path = str(tmpdir / "test.db")
db = Database(db_path)
db["t"].insert({"text": "hello"})
db.create_view("v", "select * from t")
db.close()
from click.testing import CliRunner
from sqlite_utils import cli as cli_module
result = CliRunner().invoke(cli_module.cli, ["enable-fts", db_path, "v", "text"])
assert result.exit_code == 1
assert result.output.strip() == "Error: Table v is actually a view"

9
tests/test_gis.py
View file

@ -6,12 +6,6 @@ from sqlite_utils.cli import cli
from sqlite_utils.db import Database
from sqlite_utils.utils import find_spatialite, sqlite3
try:
import sqlean # type: ignore[import-not-found]
except ImportError:
sqlean = None
pytestmark = [
pytest.mark.skipif(
not find_spatialite(), reason="Could not find SpatiaLite extension"
@ -20,9 +14,6 @@ pytestmark = [
not hasattr(sqlite3.Connection, "enable_load_extension"),
reason="sqlite3.Connection missing enable_load_extension",
),
pytest.mark.skipif(
sqlean is not None, reason="sqlean.py is not compatible with SpatiaLite"
),
]

27
tests/test_introspect.py
View file

@ -109,13 +109,11 @@ def test_table_repr(fresh_db):
def test_indexes(fresh_db):
fresh_db.executescript(
"""
fresh_db.executescript("""
create table Gosh (c1 text, c2 text, c3 text);
create index Gosh_c1 on Gosh(c1);
create index Gosh_c2c3 on Gosh(c2, c3);
"""
)
""")
assert [
Index(
seq=0,
@ -130,13 +128,11 @@ def test_indexes(fresh_db):
def test_xindexes(fresh_db):
fresh_db.executescript(
"""
fresh_db.executescript("""
create table Gosh (c1 text, c2 text, c3 text);
create index Gosh_c1 on Gosh(c1);
create index Gosh_c2c3 on Gosh(c2, c3 desc);
"""
)
""")
assert fresh_db["Gosh"].xindexes == [
XIndex(
name="Gosh_c2c3",
@ -325,3 +321,18 @@ def test_table_default_values(fresh_db, value):
)
default_values = fresh_db["default_values"].default_values
assert default_values == {"value": value}
def test_pks_use_primary_key_declaration_order(fresh_db):
# PRIMARY KEY (a, b) declared against columns stored in order (b, a) -
# pks must follow the declaration order, which is what SQLite uses to
# resolve implicit foreign key references and compound pk lookups
fresh_db.execute("create table t (b text, a text, primary key (a, b))")
assert fresh_db["t"].pks == ["a", "b"]
def test_transform_preserves_compound_pk_declaration_order(fresh_db):
fresh_db.execute("create table t (a text, b text, c text, primary key (b, a))")
fresh_db["t"].transform(drop={"c"})
assert fresh_db["t"].pks == ["b", "a"]
assert 'PRIMARY KEY ("b", "a")' in fresh_db["t"].schema

24
tests/test_lookup.py
View file

@ -157,3 +157,27 @@ def test_lookup_with_extra_insert_parameters(fresh_db):
def test_lookup_new_table_strict(fresh_db, strict):
fresh_db["species"].lookup({"name": "Palm"}, strict=strict)
assert fresh_db["species"].strict == strict or not fresh_db.supports_strict
def test_lookup_null_value_idempotent(fresh_db):
# https://github.com/simonw/sqlite-utils/issues/186
# Repeated lookups of a null value should return the same row,
# not insert a duplicate row each time
species = fresh_db["species"]
first_id = species.lookup({"name": None})
second_id = species.lookup({"name": None})
assert first_id == second_id
assert list(species.rows) == [{"id": first_id, "name": None}]
def test_lookup_compound_key_with_null_idempotent(fresh_db):
species = fresh_db["species"]
palm_id = species.lookup({"name": "Palm", "type": None})
oak_id = species.lookup({"name": "Oak", "type": "Tree"})
assert palm_id == species.lookup({"name": "Palm", "type": None})
assert oak_id == species.lookup({"name": "Oak", "type": "Tree"})
assert palm_id != oak_id
assert list(species.rows) == [
{"id": palm_id, "name": "Palm", "type": None},
{"id": oak_id, "name": "Oak", "type": "Tree"},
]

4
tests/test_m2m.py
View file

@ -139,9 +139,9 @@ def test_m2m_lookup(fresh_db):
def test_m2m_requires_either_records_or_lookup(fresh_db):
people = fresh_db.table("people", pk="id").insert({"name": "Wahyu"})
with pytest.raises(AssertionError):
with pytest.raises(ValueError):
people.m2m("tags")
with pytest.raises(AssertionError):
with pytest.raises(ValueError):
people.m2m("tags", {"tag": "hello"}, lookup={"foo": "bar"})

246
tests/test_migrations.py Normal file
View file

@ -0,0 +1,246 @@
import pytest
import sqlite_utils
from sqlite_utils import Migrations
@pytest.fixture
def migrations():
migrations = Migrations("test")
@migrations()
def m001(db):
db["dogs"].insert({"name": "Cleo"})
@migrations()
def m002(db):
db["cats"].create({"name": str})
db.execute("insert into dogs (name) values ('Pancakes')")
return migrations
@pytest.fixture
def migrations_not_ordered_alphabetically():
# Names order alphabetically in the wrong direction but this
# should still be applied correctly.
migrations = Migrations("test")
@migrations()
def m002(db):
db["dogs"].insert({"name": "Cleo"})
@migrations()
def m001(db):
db["cats"].create({"name": str})
db.execute("insert into dogs (name) values ('Pancakes')")
return migrations
@pytest.fixture
def migrations2():
migrations = Migrations("test2")
@migrations()
def m001(db):
db["dogs2"].insert({"name": "Cleo"})
return migrations
def test_basic(migrations):
db = sqlite_utils.Database(memory=True)
assert db.table_names() == []
migrations.apply(db)
assert set(db.table_names()) == {"_sqlite_migrations", "dogs", "cats"}
def test_stop_before(migrations):
db = sqlite_utils.Database(memory=True)
assert db.table_names() == []
migrations.apply(db, stop_before="m002")
assert set(db.table_names()) == {"_sqlite_migrations", "dogs"}
migrations.apply(db)
assert set(db.table_names()) == {"_sqlite_migrations", "dogs", "cats"}
def test_two_migration_sets(migrations, migrations2):
db = sqlite_utils.Database(memory=True)
assert db.table_names() == []
migrations.apply(db)
migrations2.apply(db)
assert set(db.table_names()) == {"_sqlite_migrations", "dogs", "cats", "dogs2"}
def test_order_does_not_matter(migrations, migrations_not_ordered_alphabetically):
db1 = sqlite_utils.Database(memory=True)
db2 = sqlite_utils.Database(memory=True)
migrations.apply(db1)
migrations_not_ordered_alphabetically.apply(db2)
assert db1.schema == db2.schema
def test_applied_at_is_a_string(migrations):
db = sqlite_utils.Database(memory=True)
migrations.apply(db)
applied = migrations.applied(db)
assert len(applied) == 2
for migration in applied:
# applied_at is the TEXT timestamp straight from the
# _sqlite_migrations table, e.g. "2026-07-04 12:00:00.000000+00:00"
assert isinstance(migration.applied_at, str)
assert migration.applied_at.endswith("+00:00")
def test_failing_migration_rolls_back(migrations):
@migrations()
def m003(db):
db["birds"].create({"name": str})
db.execute("insert into dogs (name) values ('Dozer')")
raise ValueError("boom")
db = sqlite_utils.Database(memory=True)
with pytest.raises(ValueError):
migrations.apply(db)
# m001 and m002 committed before the failure and stay applied
assert set(db.table_names()) == {"_sqlite_migrations", "dogs", "cats"}
assert [r["name"] for r in db["dogs"].rows] == ["Cleo", "Pancakes"]
assert [m.name for m in migrations.applied(db)] == ["m001", "m002"]
# Everything m003 did was rolled back and it is still pending
assert [m.name for m in migrations.pending(db)] == ["m003"]
def test_rerun_after_failure_applies_each_migration_once():
state = {"fail": True}
migrations = Migrations("test")
@migrations()
def m001(db):
db["dogs"].insert({"name": "Cleo"})
@migrations()
def m002(db):
db["dogs"].insert({"name": "Pancakes"})
if state["fail"]:
raise ValueError("boom")
db = sqlite_utils.Database(memory=True)
with pytest.raises(ValueError):
migrations.apply(db)
state["fail"] = False
migrations.apply(db)
# m001 must not have been re-applied, m002 applied exactly once
assert [r["name"] for r in db["dogs"].rows] == ["Cleo", "Pancakes"]
assert [m.name for m in migrations.applied(db)] == ["m001", "m002"]
def test_non_transactional_migration_allows_vacuum(tmpdir):
path = str(tmpdir / "test.db")
db = sqlite_utils.Database(path)
migrations = Migrations("test")
@migrations()
def m001(db):
db["dogs"].insert({"name": "Cleo"})
@migrations(transactional=False)
def m002(db):
db.execute("VACUUM")
migrations.apply(db)
assert [m.name for m in migrations.applied(db)] == ["m001", "m002"]
db.close()
def test_apply_composes_inside_outer_transaction(migrations):
db = sqlite_utils.Database(memory=True)
with pytest.raises(ZeroDivisionError):
with db.atomic():
migrations.apply(db)
raise ZeroDivisionError
# The outer transaction rolled back, taking the migrations with it
assert db.table_names() == []
@pytest.mark.parametrize(
"create_table,pk",
(
(
{
"migration_set": str,
"name": str,
"applied_at": str,
},
"name",
),
(
{
"migration_set": str,
"name": str,
"applied_at": str,
},
("migration_set", "name"),
),
),
)
def test_upgrades_sqlite_migrations(migrations, create_table, pk):
db = sqlite_utils.Database(memory=True)
db["_sqlite_migrations"].create(create_table, pk=pk)
assert db.table_names() == ["_sqlite_migrations"]
assert db["_sqlite_migrations"].pks == ([pk] if isinstance(pk, str) else list(pk))
migrations.apply(db)
assert db["_sqlite_migrations"].pks == ["id"]
def test_pending_and_applied_are_read_only(migrations):
db = sqlite_utils.Database(memory=True)
assert [m.name for m in migrations.pending(db)] == ["m001", "m002"]
assert migrations.applied(db) == []
# Neither call should have created the tracking table
assert db.table_names() == []
def test_duplicate_migration_name_errors():
migrations = Migrations("test")
@migrations()
def m001(db):
pass
with pytest.raises(ValueError) as ex:
@migrations(name="m001")
def m001_again(db):
pass
assert "m001" in str(ex.value)
def test_stop_before_applied_migration_errors(migrations):
# Stopping before a migration that has already been applied is
# impossible to honor - previously the stop name was only checked
# against pending migrations, so everything after it was applied
db = sqlite_utils.Database(memory=True)
migrations.apply(db, stop_before="m002") # applies m001 only
with pytest.raises(ValueError) as ex:
migrations.apply(db, stop_before="m001")
assert "m001" in str(ex.value)
assert "already been applied" in str(ex.value)
# Nothing else was applied
assert not db["cats"].exists()
def test_stop_before_applied_migration_errors_before_any_apply(migrations):
# The error fires before any pending migration runs, even those that
# come before the already-applied stop target in registration order
db = sqlite_utils.Database(memory=True)
only_second = Migrations("test")
@only_second()
def m002(db):
db["cats"].create({"name": str})
only_second.apply(db) # m002 applied, m001 still pending
with pytest.raises(ValueError):
migrations.apply(db, stop_before="m002")
assert not db["dogs"].exists()

31
tests/test_plugins.py
View file

@ -2,6 +2,7 @@ from click.testing import CliRunner
import click
import importlib
import pytest
import sys
from sqlite_utils import cli, Database, hookimpl, plugins
@ -16,6 +17,36 @@ def _supports_pragma_function_list():
db.close()
def test_get_plugins_loads_setuptools_entrypoints_once(monkeypatch):
calls = []
monkeypatch.delattr(sys, "_called_from_test", raising=False)
monkeypatch.setattr(plugins, "_plugins_loaded", False)
monkeypatch.setattr(
plugins.pm,
"load_setuptools_entrypoints",
lambda group: calls.append(group) or 0,
)
plugins.get_plugins()
plugins.get_plugins()
assert calls == ["sqlite_utils"]
def test_get_plugins_does_not_load_setuptools_entrypoints_in_tests(monkeypatch):
calls = []
monkeypatch.setattr(sys, "_called_from_test", True, raising=False)
monkeypatch.setattr(plugins, "_plugins_loaded", False)
monkeypatch.setattr(
plugins.pm,
"load_setuptools_entrypoints",
lambda group: calls.append(group) or 0,
)
assert plugins.get_plugins() == []
assert calls == []
def test_register_commands():
importlib.reload(cli)
assert plugins.get_plugins() == []

286
tests/test_query.py
View file

@ -1,5 +1,8 @@
import pytest
import types
from sqlite_utils.utils import sqlite3
def test_query(fresh_db):
fresh_db["dogs"].insert_all([{"name": "Cleo"}, {"name": "Pancakes"}])
@ -8,6 +11,268 @@ def test_query(fresh_db):
assert list(results) == [{"name": "Pancakes"}, {"name": "Cleo"}]
def test_query_executes_eagerly(fresh_db):
# The SQL runs when query() is called, not when the result is iterated,
# so errors are raised at the call site
with pytest.raises(sqlite3.OperationalError):
fresh_db.query("select * from missing_table")
def test_query_rejects_statements_that_return_no_rows(fresh_db):
fresh_db["dogs"].insert({"name": "Cleo"})
with pytest.raises(ValueError) as ex:
fresh_db.query("update dogs set name = 'Cleopaws'")
assert "execute()" in str(ex.value)
# The rejected update was rolled back, and no transaction is left open
assert not fresh_db.conn.in_transaction
assert [row["name"] for row in fresh_db["dogs"].rows] == ["Cleo"]
def test_query_rejected_ddl_is_rolled_back(fresh_db):
with pytest.raises(ValueError):
fresh_db.query("create table dogs (id integer primary key)")
assert not fresh_db.conn.in_transaction
assert fresh_db.table_names() == []
def test_query_rejected_write_inside_transaction_is_rolled_back(fresh_db):
fresh_db["dogs"].insert({"name": "Cleo"})
fresh_db.begin()
fresh_db.execute("insert into dogs (name) values ('Pancakes')")
with pytest.raises(ValueError):
fresh_db.query("update dogs set name = 'Cleopaws'")
# The transaction is still open and the earlier insert is intact
assert fresh_db.conn.in_transaction
fresh_db.commit()
assert [row["name"] for row in fresh_db["dogs"].rows] == ["Cleo", "Pancakes"]
@pytest.mark.parametrize(
"sql",
[
"begin",
"commit",
"rollback",
"vacuum",
"detach database foo",
"/* comment */ commit",
"-- comment\nbegin",
"/* multi\nline */ -- and another\n vacuum",
"\t /* a */ /* b */ savepoint s1",
"; commit",
";;\n ; rollback",
"; /* comment */ vacuum",
"\ufeffbegin",
],
)
def test_query_rejects_transaction_control_and_vacuum(fresh_db, sql):
with pytest.raises(ValueError) as ex:
fresh_db.query(sql)
assert "execute()" in str(ex.value)
assert not fresh_db.conn.in_transaction
def test_query_comment_prefixed_commit_does_not_commit_transaction(fresh_db):
# A COMMIT hidden behind a leading comment must not slip past the
# keyword check - previously it committed the caller's open
# transaction before the ValueError was raised
fresh_db["dogs"].insert({"name": "Cleo"})
fresh_db.begin()
fresh_db.execute("insert into dogs (name) values ('Pancakes')")
with pytest.raises(ValueError):
fresh_db.query("/* comment */ COMMIT")
# The explicit transaction is still open and can still be rolled back
assert fresh_db.conn.in_transaction
fresh_db.rollback()
assert [row["name"] for row in fresh_db["dogs"].rows] == ["Cleo"]
@pytest.mark.parametrize("sql", ["; COMMIT", "\ufeffCOMMIT"])
def test_query_prefixed_commit_does_not_commit_transaction(fresh_db, sql):
# sqlite3 tolerates empty statements and a UTF-8 BOM before the first
# real token, so the keyword scanner must skip them too - previously
# '; COMMIT' slipped past the check and committed the caller's open
# transaction before raising OperationalError
fresh_db["dogs"].insert({"name": "Cleo"})
fresh_db.begin()
fresh_db.execute("insert into dogs (name) values ('Pancakes')")
with pytest.raises(ValueError):
fresh_db.query(sql)
# The explicit transaction is still open and can still be rolled back
assert fresh_db.conn.in_transaction
fresh_db.rollback()
assert [row["name"] for row in fresh_db["dogs"].rows] == ["Cleo"]
def test_query_error_leaves_no_transaction_open(fresh_db):
with pytest.raises(sqlite3.OperationalError):
fresh_db.query("select * from missing_table")
assert not fresh_db.conn.in_transaction
def test_query_pragma(tmpdir):
from sqlite_utils import Database
db = Database(str(tmpdir / "test.db"))
# A row-returning PRAGMA works, including one that cannot run in a transaction
assert list(db.query("pragma journal_mode = wal")) == [{"journal_mode": "wal"}]
# A PRAGMA that returns no rows raises ValueError
with pytest.raises(ValueError):
db.query("pragma user_version = 5")
db.close()
def test_query_rejected_pragma_still_takes_effect(fresh_db):
# Documented limitation: PRAGMAs run outside the savepoint guard,
# because some of them refuse to run inside a transaction - so a
# row-less PRAGMA takes effect even though it raises ValueError.
# If this test starts failing because the pragma was rolled back,
# the limitation has been fixed - update the docs in python-api.rst
# and the query() docstring to remove the carve-out
with pytest.raises(ValueError):
fresh_db.query("pragma user_version = 5")
assert fresh_db.execute("pragma user_version").fetchone()[0] == 5
def test_query_comment_prefixed_pragma(tmpdir):
from sqlite_utils import Database
db = Database(str(tmpdir / "test.db"))
# A leading comment must not stop a PRAGMA being recognized as one -
# previously it was executed inside the savepoint guard, where
# journal mode changes are refused
assert list(db.query("-- set WAL mode\npragma journal_mode = wal")) == [
{"journal_mode": "wal"}
]
db.close()
def test_query_comment_prefixed_pragma_inside_transaction(fresh_db):
fresh_db.begin()
assert list(fresh_db.query("-- check version\npragma user_version")) == [
{"user_version": 0}
]
assert fresh_db.conn.in_transaction
fresh_db.rollback()
@pytest.mark.parametrize(
"sql,expected",
[
("select 1", "SELECT"),
(" \t\n select 1", "SELECT"),
("-- comment\nbegin", "BEGIN"),
("/* one */ /* two */ pragma user_version", "PRAGMA"),
("/* multi\nline */vacuum", "VACUUM"),
("insert into t values (1)", "INSERT"),
("-- only a comment", ""),
("/* unterminated", ""),
("", ""),
(" ", ""),
("123", ""),
("; commit", "COMMIT"),
(";;\n ; rollback", "ROLLBACK"),
("; -- comment\n begin", "BEGIN"),
("\ufeffcommit", "COMMIT"),
("\ufeff ; select 1", "SELECT"),
(";", ""),
],
)
def test_first_keyword(sql, expected):
from sqlite_utils.db import _first_keyword
assert _first_keyword(sql) == expected
@pytest.mark.skipif(
sqlite3.sqlite_version_info < (3, 35, 0),
reason="RETURNING requires SQLite 3.35.0 or higher",
)
def test_query_insert_returning(fresh_db):
fresh_db["dogs"].insert({"name": "Cleo"})
rows = list(
fresh_db.query("insert into dogs (name) values ('Pancakes') returning name")
)
assert rows == [{"name": "Pancakes"}]
assert fresh_db["dogs"].count == 2
@pytest.mark.skipif(
sqlite3.sqlite_version_info < (3, 35, 0),
reason="RETURNING requires SQLite 3.35.0 or higher",
)
def test_query_insert_returning_commits_without_iteration(tmpdir):
from sqlite_utils import Database
path = str(tmpdir / "test.db")
db = Database(path)
db["dogs"].insert({"name": "Cleo"})
# Never iterate over the results
db.query("insert into dogs (name) values ('Pancakes') returning name")
assert not db.conn.in_transaction
# A completely separate connection sees the new row straight away
other = sqlite3.connect(path)
assert other.execute("select count(*) from dogs").fetchone()[0] == 2
other.close()
db.close()
@pytest.mark.skipif(
sqlite3.sqlite_version_info < (3, 35, 0),
reason="RETURNING requires SQLite 3.35.0 or higher",
)
def test_query_insert_returning_partial_iteration_still_commits(tmpdir):
from sqlite_utils import Database
path = str(tmpdir / "test.db")
db = Database(path)
db["dogs"].insert({"name": "Cleo"})
row = next(
db.query(
"insert into dogs (name) values ('Pancakes'), ('Marnie') returning name"
)
)
assert row == {"name": "Pancakes"}
assert not db.conn.in_transaction
other = sqlite3.connect(path)
assert other.execute("select count(*) from dogs").fetchone()[0] == 3
other.close()
db.close()
@pytest.mark.skipif(
sqlite3.sqlite_version_info < (3, 35, 0),
reason="RETURNING requires SQLite 3.35.0 or higher",
)
def test_query_insert_returning_respects_explicit_transaction(fresh_db):
fresh_db["dogs"].insert({"name": "Cleo"})
fresh_db.begin()
rows = list(
fresh_db.query("insert into dogs (name) values ('Pancakes') returning name")
)
assert rows == [{"name": "Pancakes"}]
# Still inside the explicit transaction - not committed
assert fresh_db.conn.in_transaction
fresh_db.rollback()
assert [row["name"] for row in fresh_db["dogs"].rows] == ["Cleo"]
def test_query_duplicate_column_names_are_deduped(fresh_db):
# https://github.com/simonw/sqlite-utils/issues/624
fresh_db["one"].insert({"id": 1, "value": "left"})
fresh_db["two"].insert({"id": 2, "value": "right"})
rows = list(
fresh_db.query("select one.id, two.id, one.value, two.value from one, two")
)
assert rows == [{"id": 1, "id_2": 2, "value": "left", "value_2": "right"}]
def test_query_deduped_column_avoids_existing_names(fresh_db):
# The renamed duplicate must not overwrite a real column called id_2
rows = list(fresh_db.query("select 1 as id, 2 as id, 3 as id_2"))
assert rows == [{"id": 1, "id_3": 2, "id_2": 3}]
def test_execute_returning_dicts(fresh_db):
# Like db.query() but returns a list, included for backwards compatibility
# see https://github.com/simonw/sqlite-utils/issues/290
@ -15,3 +280,24 @@ def test_execute_returning_dicts(fresh_db):
assert fresh_db.execute_returning_dicts("select * from test") == [
{"id": 1, "bar": 2}
]
@pytest.mark.skipif(
sqlite3.sqlite_version_info < (3, 35, 0),
reason="RETURNING requires SQLite 3.35.0 or higher",
)
def test_query_preserves_error_from_transaction_destroying_trigger(fresh_db):
# RAISE(ROLLBACK) destroys the savepoint guard - the original
# IntegrityError must propagate, not "no such savepoint"
fresh_db.execute("create table t (id integer primary key, v text)")
fresh_db.execute("""
create trigger no_bad before insert on t
when new.v = 'bad'
begin
select raise(rollback, 'trigger says no');
end
""")
with pytest.raises(sqlite3.IntegrityError, match="trigger says no"):
fresh_db.query("insert into t (id, v) values (1, 'bad') returning id")
assert not fresh_db.conn.in_transaction
assert fresh_db.execute("select count(*) from t").fetchone()[0] == 0

2
tests/test_recreate.py
View file

@ -15,7 +15,7 @@ def test_recreate_ignored_for_in_memory():
def test_recreate_not_allowed_for_connection():
conn = sqlite3.connect(":memory:")
try:
with pytest.raises(AssertionError):
with pytest.raises(ValueError):
Database(conn, recreate=True)
finally:
conn.close()

34
tests/test_rows.py
View file

@ -104,3 +104,37 @@ def test_pks_and_rows_where_compound_pk(fresh_db):
(("number", 1), {"type": "number", "number": 1, "plusone": 2}),
(("number", 2), {"type": "number", "number": 2, "plusone": 3}),
]
def test_rows_where_duplicate_select_columns_are_deduped(fresh_db):
# https://github.com/simonw/sqlite-utils/issues/624
fresh_db["t"].insert({"id": 1, "name": "Cleo"})
rows = list(fresh_db["t"].rows_where(select="id, id, name"))
assert rows == [{"id": 1, "id_2": 1, "name": "Cleo"}]
def test_pks_and_rows_where_view(fresh_db):
# pks_and_rows_where() lives on Queryable so views expose it, but
# SQLite views have no rowid. Modern SQLite (3.36+) raises an
# OperationalError from the generated SQL; older versions returned
# NULL for a view's rowid. Either way it must not fail earlier with
# an AttributeError from View lacking Table-only properties
from sqlite_utils.utils import sqlite3
fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
fresh_db.create_view("dog_names", "select name from dogs")
try:
result = list(fresh_db["dog_names"].pks_and_rows_where())
except sqlite3.OperationalError:
pass # SQLite 3.36+: no such column: rowid
else:
# Older SQLite returns NULL rowids for views
assert result == [(None, {"rowid": None, "name": "Cleo"})]
def test_pks_and_rows_where_compound_pk_declaration_order(fresh_db):
# Compound pks are returned in PRIMARY KEY declaration order
fresh_db.execute("create table t (b text, a text, primary key (a, b))")
fresh_db["t"].insert({"a": "A", "b": "B"})
pks_and_rows = list(fresh_db["t"].pks_and_rows_where())
assert pks_and_rows == [(("A", "B"), {"b": "B", "a": "A"})]

2
tests/test_sniff.py
View file

@ -6,7 +6,7 @@ import pytest
sniff_dir = pathlib.Path(__file__).parent / "sniff"
@pytest.mark.parametrize("filepath", sniff_dir.glob("example*"))
@pytest.mark.parametrize("filepath", sorted(sniff_dir.glob("example*")))
def test_sniff(tmpdir, filepath):
db_path = str(tmpdir / "test.db")
runner = CliRunner()

2
tests/test_tracer.py
View file

@ -64,7 +64,7 @@ def test_with_tracer():
" )\n"
" )",
{
"like": '%VIRTUAL TABLE%USING FTS%content="dogs"%',
"like": "%VIRTUAL TABLE%USING FTS%content=[dogs]%",
"like2": '%VIRTUAL TABLE%USING FTS%content="dogs"%',
"table": "dogs",
},

40
tests/test_transform.py
View file

@ -224,6 +224,40 @@ def test_transform_rename_pk(fresh_db):
)
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["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["dogs"]
dogs.insert({"id": 1, "name": "Cleo", "age": "5"}, pk="id")
@ -638,15 +672,13 @@ def test_transform_with_indexes_errors(fresh_db, transform_params):
def test_transform_with_unique_constraint_implicit_index(fresh_db):
dogs = fresh_db["dogs"]
# Create a table with a UNIQUE constraint on 'name', which creates an implicit index
fresh_db.execute(
"""
fresh_db.execute("""
CREATE TABLE dogs (
id INTEGER PRIMARY KEY,
name TEXT UNIQUE,
age INTEGER
);
"""
)
""")
dogs.insert({"id": 1, "name": "Cleo", "age": 5})
# Attempt to transform the table without modifying 'name'

83
tests/test_upsert.py
View file

@ -49,6 +49,89 @@ def test_upsert_error_if_no_pk(fresh_db):
table.upsert({"id": 1, "name": "Cleo"})
@pytest.mark.parametrize("use_old_upsert", (False, True))
def test_upsert_empty_record_errors(use_old_upsert):
db = Database(memory=True, use_old_upsert=use_old_upsert)
table = db["table"]
table.insert({"id": 1, "name": "Cleo"}, pk="id")
with pytest.raises(PrimaryKeyRequired):
table.upsert({}, pk="id")
with pytest.raises(PrimaryKeyRequired):
table.upsert_all([{}, {}], pk="id")
# No rows can have been inserted
assert table.count == 1
@pytest.mark.parametrize("use_old_upsert", (False, True))
def test_upsert_missing_pk_value_errors(use_old_upsert):
db = Database(memory=True, use_old_upsert=use_old_upsert)
table = db["table"]
table.insert({"id": 1, "name": "Cleo"}, pk="id")
# Records that omit the pk column entirely
with pytest.raises(PrimaryKeyRequired):
table.upsert_all([{"name": "Pancakes"}, {"name": "Marnie"}], pk="id")
# A record with an explicit None pk value can never conflict
with pytest.raises(PrimaryKeyRequired):
table.upsert({"id": None, "name": "Pancakes"}, pk="id")
assert list(table.rows) == [{"id": 1, "name": "Cleo"}]
def test_upsert_missing_compound_pk_value_errors(fresh_db):
table = fresh_db["table"]
table.insert({"a": "x", "b": "y", "v": 1}, pk=("a", "b"))
# Missing one component of the detected compound primary key
with pytest.raises(PrimaryKeyRequired):
table.upsert({"a": "x", "v": 2})
assert list(table.rows) == [{"a": "x", "b": "y", "v": 1}]
def test_upsert_error_if_existing_table_has_no_pk(fresh_db):
table = fresh_db.create_table("table", {"id": int, "name": str})
with pytest.raises(PrimaryKeyRequired):
table.upsert({"id": 1, "name": "Cleo"})
@pytest.mark.parametrize("use_old_upsert", (False, True))
def test_upsert_uses_compound_pk_from_existing_table(use_old_upsert):
# https://github.com/simonw/sqlite-utils/issues/629
db = Database(memory=True, use_old_upsert=use_old_upsert)
db.execute("""
create table summary (
Source text,
Object text,
Category text,
Count integer,
primary key (Source, Object, Category)
)
""")
table = db["summary"]
table.upsert(
{
"Source": "Client A",
"Object": "Accounts",
"Category": "All",
"Count": 3,
}
)
assert table.last_pk == ("Client A", "Accounts", "All")
table.upsert(
{
"Source": "Client A",
"Object": "Accounts",
"Category": "All",
"Count": 4,
}
)
assert list(table.rows) == [
{
"Source": "Client A",
"Object": "Accounts",
"Category": "All",
"Count": 4,
}
]
def test_upsert_with_hash_id(fresh_db):
table = fresh_db["table"]
table.upsert({"foo": "bar"}, hash_id="pk")

17
tests/test_utils.py
View file

@ -83,3 +83,20 @@ def test_maximize_csv_field_size_limit():
)
def test_flatten(input, expected):
assert utils.flatten(input) == expected
@pytest.mark.parametrize(
"input,expected",
(
([], []),
(["id", "name"], ["id", "name"]),
(["id", "id"], ["id", "id_2"]),
(["id", "id", "id"], ["id", "id_2", "id_3"]),
# A renamed duplicate must not clobber a real column called id_2
(["id", "id", "id_2"], ["id", "id_3", "id_2"]),
(["id_2", "id", "id"], ["id_2", "id", "id_3"]),
(["id", "id", "id_2", "id_2"], ["id", "id_3", "id_2", "id_2_2"]),
),
)
def test_dedupe_keys(input, expected):
assert utils.dedupe_keys(input) == expected

65
tests/test_wal.py
View file

@ -1,5 +1,6 @@
import pytest
from sqlite_utils import Database
from sqlite_utils.db import TransactionError
@pytest.fixture
@ -21,3 +22,67 @@ def test_enable_disable_wal(db_path_tmpdir):
db.disable_wal()
assert "delete" == db.journal_mode
assert "test.db-wal" not in [f.basename for f in tmpdir.listdir()]
def test_enable_wal_inside_transaction_raises(db_path_tmpdir):
db, path, tmpdir = db_path_tmpdir
db["test"].insert({"id": 1}, pk="id")
with pytest.raises(TransactionError):
with db.atomic():
db["test"].insert({"id": 2}, pk="id")
db.enable_wal()
# The atomic() block must have rolled back cleanly and the
# journal mode must be unchanged
assert db.journal_mode == "delete"
assert [r["id"] for r in db["test"].rows] == [1]
def test_disable_wal_inside_transaction_raises(db_path_tmpdir):
db, path, tmpdir = db_path_tmpdir
db.enable_wal()
db["test"].insert({"id": 1}, pk="id")
with pytest.raises(TransactionError):
with db.atomic():
db["test"].insert({"id": 2}, pk="id")
db.disable_wal()
assert db.journal_mode == "wal"
assert [r["id"] for r in db["test"].rows] == [1]
def test_ensure_autocommit_on(db_path_tmpdir):
db, path, tmpdir = db_path_tmpdir
previous_isolation_level = db.conn.isolation_level
assert previous_isolation_level is not None
with db.ensure_autocommit_on():
# isolation_level of None means driver-level autocommit mode
assert db.conn.isolation_level is None
# Restored afterwards
assert db.conn.isolation_level == previous_isolation_level
def test_enable_wal_noop_inside_transaction_is_allowed(db_path_tmpdir):
# Calling enable_wal() when WAL is already enabled is a no-op,
# so it is fine inside a transaction
db, path, tmpdir = db_path_tmpdir
db.enable_wal()
with db.atomic():
db["test"].insert({"id": 1}, pk="id")
db.enable_wal()
assert [r["id"] for r in db["test"].rows] == [1]
def test_ensure_autocommit_on_inside_transaction_raises(db_path_tmpdir):
# Setting isolation_level commits any pending transaction as a side
# effect, silently breaking the caller's rollback guarantee - so
# entering autocommit mode with a transaction open is an error
db, path, tmpdir = db_path_tmpdir
db["test"].insert({"id": 1}, pk="id")
db.begin()
db.execute("insert into test (id) values (2)")
with pytest.raises(TransactionError):
with db.ensure_autocommit_on():
pass
# The transaction is still open and can still be rolled back
assert db.conn.in_transaction
db.rollback()
assert [r["id"] for r in db["test"].rows] == [1]