mirror of
https://github.com/simonw/sqlite-utils.git
synced 2026-09-17 14:04:11 +02:00
Merge branch 'main' into fix-transform-keyword-literal-defaults
This commit is contained in:
Simon Willison
commit
750686c6d1
44 changed files with 3440 additions and 376 deletions
|
|
@ -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():
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
import pytest
|
||||
|
||||
from sqlite_utils.db import _iter_complete_sql_statements
|
||||
from sqlite_utils.db import Database, _iter_complete_sql_statements
|
||||
from sqlite_utils.utils import sqlite3
|
||||
|
||||
|
||||
|
|
@ -172,3 +172,104 @@ def test_transform_detects_foreign_key_check_violations(fresh_db):
|
|||
|
||||
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 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()
|
||||
|
|
|
|||
|
|
@ -746,6 +746,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"]
|
||||
|
|
@ -1467,6 +1487,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():
|
||||
|
|
@ -1485,6 +1523,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():
|
||||
|
|
@ -2296,18 +2351,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,
|
||||
)
|
||||
|
|
@ -2319,17 +2370,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,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -615,3 +615,16 @@ def test_insert_csv_headers_only(tmpdir):
|
|||
# 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"
|
||||
|
|
|
|||
|
|
@ -301,3 +301,165 @@ def test_stop_before_multiple_qualified(two_sets_same_migration_name):
|
|||
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()
|
||||
|
|
|
|||
233
tests/test_column_casing.py
Normal file
233
tests/test_column_casing.py
Normal 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"
|
||||
)
|
||||
]
|
||||
|
|
@ -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:
|
||||
|
|
@ -39,3 +59,31 @@ def test_database_close(tmpdir, memory):
|
|||
db.close()
|
||||
with pytest.raises(sqlite3.ProgrammingError):
|
||||
db.execute("select 1 + 1")
|
||||
|
||||
|
||||
@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()
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -108,7 +108,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"}
|
||||
)
|
||||
|
|
@ -243,11 +243,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])
|
||||
|
|
@ -700,7 +700,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)
|
||||
|
|
@ -1061,7 +1061,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)
|
||||
|
||||
|
|
@ -1214,7 +1214,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"
|
||||
|
||||
|
|
@ -1382,7 +1382,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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
526
tests/test_foreign_keys.py
Normal file
526
tests/test_foreign_keys.py
Normal file
|
|
@ -0,0 +1,526 @@
|
|||
"""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
|
||||
|
|
@ -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"})
|
||||
|
|
@ -443,12 +475,12 @@ def test_enable_fts_replace_handles_legacy_bracket_quoted_content_table():
|
|||
assert 'content="books"' in db["books_fts"].schema
|
||||
|
||||
|
||||
def test_enable_fts_error_message_on_views():
|
||||
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(
|
||||
|
|
@ -700,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"
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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"})
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ def migrations():
|
|||
@migrations()
|
||||
def m002(db):
|
||||
db["cats"].create({"name": str})
|
||||
db.query("insert into dogs (name) values ('Pancakes')")
|
||||
db.execute("insert into dogs (name) values ('Pancakes')")
|
||||
|
||||
return migrations
|
||||
|
||||
|
|
@ -32,7 +32,7 @@ def migrations_not_ordered_alphabetically():
|
|||
@migrations()
|
||||
def m001(db):
|
||||
db["cats"].create({"name": str})
|
||||
db.query("insert into dogs (name) values ('Pancakes')")
|
||||
db.execute("insert into dogs (name) values ('Pancakes')")
|
||||
|
||||
return migrations
|
||||
|
||||
|
|
@ -80,6 +80,88 @@ def test_order_does_not_matter(migrations, migrations_not_ordered_alphabetically
|
|||
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",
|
||||
(
|
||||
|
|
@ -108,3 +190,27 @@ def test_upgrades_sqlite_migrations(migrations, create_table, pk):
|
|||
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)
|
||||
|
|
|
|||
|
|
@ -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,229 @@ 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",
|
||||
],
|
||||
)
|
||||
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"]
|
||||
|
||||
|
||||
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_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", ""),
|
||||
],
|
||||
)
|
||||
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,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()
|
||||
|
|
|
|||
|
|
@ -104,3 +104,10 @@ 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"}]
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -49,6 +49,42 @@ 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):
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import pytest
|
||||
from sqlite_utils import Database
|
||||
from sqlite_utils.db import TransactionError
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -21,3 +22,50 @@ 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]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue