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

Now complies with flake8, refs #291

Browse source
This commit is contained in:
Simon Willison 2021-06-22 18:22:08 -07:00
commit 90e211e3e2
20 changed files with 252 additions and 67 deletions
Download patch file Download diff file Expand all files Collapse all files

5
tests/test_analyze_tables.py
View file

@ -1,10 +1,8 @@
from sqlite_utils.db import Database, ForeignKey, ColumnDetails
from sqlite_utils.db import Database, ColumnDetails
from sqlite_utils import cli
from sqlite_utils.utils import OperationalError
from click.testing import CliRunner
import pytest
import sqlite3
import textwrap
@pytest.fixture
@ -132,6 +130,7 @@ def test_analyze_table_save(db_to_analyze_path):
result = CliRunner().invoke(
cli.cli, ["analyze-tables", db_to_analyze_path, "--save"]
)
assert result.exit_code == 0
rows = list(Database(db_to_analyze_path)["_analyze_tables_"].rows)
assert rows == [
{

181
tests/test_cli.py
View file

@ -410,7 +410,7 @@ def test_index_foreign_keys(db_path):
def test_enable_fts(db_path):
db = Database(db_path)
assert None == db["Gosh"].detect_fts()
assert db["Gosh"].detect_fts() is None
result = CliRunner().invoke(
cli.cli, ["enable-fts", db_path, "Gosh", "c1", "--fts4"]
)
@ -420,7 +420,7 @@ def test_enable_fts(db_path):
# Table names with restricted chars are handled correctly.
# colons and dots are restricted characters for table names.
db["http://example.com"].create({"c1": str, "c2": str, "c3": str})
assert None == db["http://example.com"].detect_fts()
assert db["http://example.com"].detect_fts() is None
result = CliRunner().invoke(
cli.cli,
[
@ -966,7 +966,23 @@ def test_query_json(db_path, sql, args, expected):
assert expected == result.output.strip()
LOREM_IPSUM_COMPRESSED = b"x\x9c\xed\xd1\xcdq\x03!\x0c\x05\xe0\xbb\xabP\x01\x1eW\x91\xdc|M\x01\n\xc8\x8ef\xf83H\x1e\x97\x1f\x91M\x8e\xe9\xe0\xdd\x96\x05\x84\xf4\xbek\x9fRI\xc7\xf2J\xb9\x97>i\xa9\x11W\xb13\xa5\xde\x96$\x13\xf3I\x9cu\xe8J\xda\xee$EcsI\x8e\x0b$\xea\xab\xf6L&u\xc4emI\xb3foFnT\xf83\xca\x93\xd8QZ\xa8\xf2\xbd1q\xd1\x87\xf3\x85>\x8c\xa4i\x8d\xdaTu\x7f<c\xc9\xf5L\x0f\xd7E\xad/\x9b\x9eI^2\x93\x1a\x9b\xf6F^\n\xd7\xd4\x8f\xca\xfb\x90.\xdd/\xfd\x94\xd4\x11\x87I8\x1a\xaf\xd1S?\x06\x88\xa7\xecBo\xbb$\xbb\t\xe9\xf4\xe8\xe4\x98U\x1bM\x19S\xbe\xa4e\x991x\xfcx\xf6\xe2#\x9e\x93h'&%YK(i)\x7f\t\xc5@N7\xbf+\x1b\xb5\xdd\x10\r\x9e\xb1\xf0y\xa1\xf7W\x92a\xe2;\xc6\xc8\xa0\xa7\xc4\x92\xe2\\\xf2\xa1\x99m\xdf\x88)\xc6\xec\x9a\xa5\xed\x14wR\xf1h\xf22x\xcfM\xfdv\xd3\xa4LY\x96\xcc\xbd[{\xd9m\xf0\x0eH#\x8e\xf5\x9b\xab\xd7\xcb\xe9t\x05\x1f\xf8\xc0\x07>\xf0\x81\x0f|\xe0\x03\x1f\xf8\xc0\x07>\xf0\x81\x0f|\xe0\x03\x1f\xf8\xc0\x07>\xf0\x81\x0f|\xe0\x03\x1f\xf8\xc0\x07>\xf0\x81\x0f|\xe0\x03\x1f\xf8\xc0\x07>\xf0\x81\x0f|\xe0\x03\x1f\xf8\xc0\x07>\xf0\x81\x0f|\xe0\x03\x1f\xf8\xc0\x07>\xf0\x81\x0f|\xe0\x03\x1f\xf8\xc0\x07>\xf0\x81\x0f|\xe0\xfb\x8f\xef\x1b\x9b\x06\x83}"
LOREM_IPSUM_COMPRESSED = (
b"x\x9c\xed\xd1\xcdq\x03!\x0c\x05\xe0\xbb\xabP\x01\x1eW\x91\xdc|M\x01\n\xc8\x8e"
b"f\xf83H\x1e\x97\x1f\x91M\x8e\xe9\xe0\xdd\x96\x05\x84\xf4\xbek\x9fRI\xc7\xf2J"
b"\xb9\x97>i\xa9\x11W\xb13\xa5\xde\x96$\x13\xf3I\x9cu\xe8J\xda\xee$EcsI\x8e\x0b"
b"$\xea\xab\xf6L&u\xc4emI\xb3foFnT\xf83\xca\x93\xd8QZ\xa8\xf2\xbd1q\xd1\x87\xf3"
b"\x85>\x8c\xa4i\x8d\xdaTu\x7f<c\xc9\xf5L\x0f\xd7E\xad/\x9b\x9eI^2\x93\x1a\x9b"
b"\xf6F^\n\xd7\xd4\x8f\xca\xfb\x90.\xdd/\xfd\x94\xd4\x11\x87I8\x1a\xaf\xd1S?\x06"
b"\x88\xa7\xecBo\xbb$\xbb\t\xe9\xf4\xe8\xe4\x98U\x1bM\x19S\xbe\xa4e\x991x\xfc"
b"x\xf6\xe2#\x9e\x93h'&%YK(i)\x7f\t\xc5@N7\xbf+\x1b\xb5\xdd\x10\r\x9e\xb1\xf0"
b"y\xa1\xf7W\x92a\xe2;\xc6\xc8\xa0\xa7\xc4\x92\xe2\\\xf2\xa1\x99m\xdf\x88)\xc6"
b"\xec\x9a\xa5\xed\x14wR\xf1h\xf22x\xcfM\xfdv\xd3\xa4LY\x96\xcc\xbd[{\xd9m\xf0"
b"\x0eH#\x8e\xf5\x9b\xab\xd7\xcb\xe9t\x05\x1f\xf8\xc0\x07>\xf0\x81\x0f|\xe0\x03"
b"\x1f\xf8\xc0\x07>\xf0\x81\x0f|\xe0\x03\x1f\xf8\xc0\x07>\xf0\x81\x0f|\xe0\x03"
b"\x1f\xf8\xc0\x07>\xf0\x81\x0f|\xe0\x03\x1f\xf8\xc0\x07>\xf0\x81\x0f|\xe0\x03"
b"\x1f\xf8\xc0\x07>\xf0\x81\x0f|\xe0\x03\x1f\xf8\xc0\x07>\xf0\x81\x0f|\xe0\x03"
b"\x1f\xf8\xc0\x07>\xf0\x81\x0f|\xe0\xfb\x8f\xef\x1b\x9b\x06\x83}"
)
def test_query_json_binary(db_path):
@ -988,7 +1004,18 @@ def test_query_json_binary(db_path):
"sz": 16984,
"data": {
"$base64": True,
"encoded": "eJzt0c1xAyEMBeC7q1ABHleR3HxNAQrIjmb4M0gelx+RTY7p4N2WBYT0vmufUknH8kq5lz5pqRFXsTOl3pYkE/NJnHXoStruJEVjc0mOCyTqq/ZMJnXEZW1Js2ZvRm5U+DPKk9hRWqjyvTFx0YfzhT6MpGmN2lR1fzxjyfVMD9dFrS+bnkleMpMam/ZGXgrX1I/K+5Au3S/9lNQRh0k4Gq/RUz8GiKfsQm+7JLsJ6fTo5JhVG00ZU76kZZkxePx49uIjnpNoJyYlWUsoaSl/CcVATje/Kxu13RANnrHweaH3V5Jh4jvGyKCnxJLiXPKhmW3fiCnG7Jql7RR3UvFo8jJ4z039dtOkTFmWzL1be9lt8A5II471m6vXy+l0BR/4wAc+8IEPfOADH/jABz7wgQ984AMf+MAHPvCBD3zgAx/4wAc+8IEPfOADH/jABz7wgQ984AMf+MAHPvCBD3zgAx/4wAc+8IEPfOADH/jABz7wgQ984PuP7xubBoN9",
"encoded": (
(
"eJzt0c1xAyEMBeC7q1ABHleR3HxNAQrIjmb4M0gelx+RTY7p4N2WBYT0vmufUknH"
"8kq5lz5pqRFXsTOl3pYkE/NJnHXoStruJEVjc0mOCyTqq/ZMJnXEZW1Js2ZvRm5U+"
"DPKk9hRWqjyvTFx0YfzhT6MpGmN2lR1fzxjyfVMD9dFrS+bnkleMpMam/ZGXgrX1I"
"/K+5Au3S/9lNQRh0k4Gq/RUz8GiKfsQm+7JLsJ6fTo5JhVG00ZU76kZZkxePx49uI"
"jnpNoJyYlWUsoaSl/CcVATje/Kxu13RANnrHweaH3V5Jh4jvGyKCnxJLiXPKhmW3f"
"iCnG7Jql7RR3UvFo8jJ4z039dtOkTFmWzL1be9lt8A5II471m6vXy+l0BR/4wAc+8"
"IEPfOADH/jABz7wgQ984AMf+MAHPvCBD3zgAx/4wAc+8IEPfOADH/jABz7wgQ984A"
"Mf+MAHPvCBD3zgAx/4wAc+8IEPfOADH/jABz7wgQ984PuP7xubBoN9"
)
),
},
}
]
@ -1157,17 +1184,17 @@ def test_upsert(db_path, tmpdir):
{"id": 1, "age": 5},
{"id": 2, "age": 5},
]
open(json_path, "w").write(json.dumps(insert_dogs))
open(json_path, "w").write(json.dumps(upsert_dogs))
result = CliRunner().invoke(
cli.cli,
["upsert", db_path, "dogs", json_path, "--pk", "id"],
catch_exceptions=False,
)
assert 0 == result.exit_code, result.output
assert [
{"id": 1, "name": "Cleo", "age": 4},
{"id": 2, "name": "Nixie", "age": 4},
] == list(db.query("select * from dogs order by id"))
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_alter(db_path, tmpdir):
@ -1596,47 +1623,112 @@ def test_add_foreign_keys(db_path):
[
(
[],
"CREATE TABLE \"dogs\" (\n [id] INTEGER PRIMARY KEY,\n [age] INTEGER NOT NULL DEFAULT '1',\n [name] TEXT\n)",
(
'CREATE TABLE "dogs" (\n'
" [id] INTEGER PRIMARY KEY,\n"
" [age] INTEGER NOT NULL DEFAULT '1',\n"
" [name] TEXT\n"
")"
),
),
(
["--type", "age", "text"],
"CREATE TABLE \"dogs\" (\n [id] INTEGER PRIMARY KEY,\n [age] TEXT NOT NULL DEFAULT '1',\n [name] TEXT\n)",
(
'CREATE TABLE "dogs" (\n'
" [id] INTEGER PRIMARY KEY,\n"
" [age] TEXT NOT NULL DEFAULT '1',\n"
" [name] TEXT\n"
")"
),
),
(
["--drop", "age"],
'CREATE TABLE "dogs" (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT\n)',
(
'CREATE TABLE "dogs" (\n'
" [id] INTEGER PRIMARY KEY,\n"
" [name] TEXT\n"
")"
),
),
(
["--rename", "age", "age2", "--rename", "id", "pk"],
"CREATE TABLE \"dogs\" (\n [pk] INTEGER PRIMARY KEY,\n [age2] INTEGER NOT NULL DEFAULT '1',\n [name] TEXT\n)",
(
'CREATE TABLE "dogs" (\n'
" [pk] INTEGER PRIMARY KEY,\n"
" [age2] INTEGER NOT NULL DEFAULT '1',\n"
" [name] TEXT\n"
")"
),
),
(
["--not-null", "name"],
"CREATE TABLE \"dogs\" (\n [id] INTEGER PRIMARY KEY,\n [age] INTEGER NOT NULL DEFAULT '1',\n [name] TEXT NOT NULL\n)",
(
'CREATE TABLE "dogs" (\n'
" [id] INTEGER PRIMARY KEY,\n"
" [age] INTEGER NOT NULL DEFAULT '1',\n"
" [name] TEXT NOT NULL\n"
")"
),
),
(
["--not-null-false", "age"],
"CREATE TABLE \"dogs\" (\n [id] INTEGER PRIMARY KEY,\n [age] INTEGER DEFAULT '1',\n [name] TEXT\n)",
(
'CREATE TABLE "dogs" (\n'
" [id] INTEGER PRIMARY KEY,\n"
" [age] INTEGER DEFAULT '1',\n"
" [name] TEXT\n"
")"
),
),
(
["--pk", "name"],
"CREATE TABLE \"dogs\" (\n [id] INTEGER,\n [age] INTEGER NOT NULL DEFAULT '1',\n [name] TEXT PRIMARY KEY\n)",
(
'CREATE TABLE "dogs" (\n'
" [id] INTEGER,\n"
" [age] INTEGER NOT NULL DEFAULT '1',\n"
" [name] TEXT PRIMARY KEY\n"
")"
),
),
(
["--pk-none"],
"CREATE TABLE \"dogs\" (\n [id] INTEGER,\n [age] INTEGER NOT NULL DEFAULT '1',\n [name] TEXT\n)",
(
'CREATE TABLE "dogs" (\n'
" [id] INTEGER,\n"
" [age] INTEGER NOT NULL DEFAULT '1',\n"
" [name] TEXT\n"
")"
),
),
(
["--default", "name", "Turnip"],
"CREATE TABLE \"dogs\" (\n [id] INTEGER PRIMARY KEY,\n [age] INTEGER NOT NULL DEFAULT '1',\n [name] TEXT DEFAULT 'Turnip'\n)",
(
'CREATE TABLE "dogs" (\n'
" [id] INTEGER PRIMARY KEY,\n"
" [age] INTEGER NOT NULL DEFAULT '1',\n"
" [name] TEXT DEFAULT 'Turnip'\n"
")"
),
),
(
["--default-none", "age"],
'CREATE TABLE "dogs" (\n [id] INTEGER PRIMARY KEY,\n [age] INTEGER NOT NULL,\n [name] TEXT\n)',
(
'CREATE TABLE "dogs" (\n'
" [id] INTEGER PRIMARY KEY,\n"
" [age] INTEGER NOT NULL,\n"
" [name] TEXT\n"
")"
),
),
(
["-o", "name", "--column-order", "age", "-o", "id"],
"CREATE TABLE \"dogs\" (\n [name] TEXT,\n [age] INTEGER NOT NULL DEFAULT '1',\n [id] INTEGER PRIMARY KEY\n)",
(
'CREATE TABLE "dogs" (\n'
" [name] TEXT,\n"
" [age] INTEGER NOT NULL DEFAULT '1',\n"
" [id] INTEGER PRIMARY KEY\n"
")"
),
),
],
)
@ -1685,9 +1777,13 @@ def test_transform_drop_foreign_key(db_path):
print(result.output)
assert result.exit_code == 0
schema = db["places"].schema
assert (
schema
== 'CREATE TABLE "places" (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [country] INTEGER,\n [city] INTEGER REFERENCES [city]([id])\n)'
assert schema == (
'CREATE TABLE "places" (\n'
" [id] INTEGER PRIMARY KEY,\n"
" [name] TEXT,\n"
" [country] INTEGER,\n"
" [city] INTEGER REFERENCES [city]([id])\n"
")"
)
@ -1701,22 +1797,48 @@ _common_other_schema = (
[
(
[],
'CREATE TABLE "trees" (\n [id] INTEGER PRIMARY KEY,\n [address] TEXT,\n [species_id] INTEGER,\n FOREIGN KEY([species_id]) REFERENCES [species]([id])\n)',
(
'CREATE TABLE "trees" (\n'
" [id] INTEGER PRIMARY KEY,\n"
" [address] TEXT,\n"
" [species_id] INTEGER,\n"
" FOREIGN KEY([species_id]) REFERENCES [species]([id])\n"
")"
),
_common_other_schema,
),
(
["--table", "custom_table"],
'CREATE TABLE "trees" (\n [id] INTEGER PRIMARY KEY,\n [address] TEXT,\n [custom_table_id] INTEGER,\n FOREIGN KEY([custom_table_id]) REFERENCES [custom_table]([id])\n)',
(
'CREATE TABLE "trees" (\n'
" [id] INTEGER PRIMARY KEY,\n"
" [address] TEXT,\n"
" [custom_table_id] INTEGER,\n"
" FOREIGN KEY([custom_table_id]) REFERENCES [custom_table]([id])\n"
")"
),
"CREATE TABLE [custom_table] (\n [id] INTEGER PRIMARY KEY,\n [species] TEXT\n)",
),
(
["--fk-column", "custom_fk"],
'CREATE TABLE "trees" (\n [id] INTEGER PRIMARY KEY,\n [address] TEXT,\n [custom_fk] INTEGER,\n FOREIGN KEY([custom_fk]) REFERENCES [species]([id])\n)',
(
'CREATE TABLE "trees" (\n'
" [id] INTEGER PRIMARY KEY,\n"
" [address] TEXT,\n"
" [custom_fk] INTEGER,\n"
" FOREIGN KEY([custom_fk]) REFERENCES [species]([id])\n"
")"
),
_common_other_schema,
),
(
["--rename", "name", "name2"],
'CREATE TABLE "trees" (\n [id] INTEGER PRIMARY KEY,\n [address] TEXT,\n [species_id] INTEGER,\n FOREIGN KEY([species_id]) REFERENCES [species]([id])\n)',
'CREATE TABLE "trees" (\n'
" [id] INTEGER PRIMARY KEY,\n"
" [address] TEXT,\n"
" [species_id] INTEGER,\n"
" FOREIGN KEY([species_id]) REFERENCES [species]([id])\n"
")",
"CREATE TABLE [species] (\n [id] INTEGER PRIMARY KEY,\n [species] TEXT\n)",
),
],
@ -1902,7 +2024,10 @@ def test_indexes(tmpdir):
]
_TRIGGERS_EXPECTED = '[{"name": "blah", "table": "articles", "sql": "CREATE TRIGGER blah AFTER INSERT ON articles\\nBEGIN\\n UPDATE counter SET count = count + 1;\\nEND"}]\n'
_TRIGGERS_EXPECTED = (
'[{"name": "blah", "table": "articles", "sql": "CREATE TRIGGER blah '
'AFTER INSERT ON articles\\nBEGIN\\n UPDATE counter SET count = count + 1;\\nEND"}]\n'
)
@pytest.mark.parametrize(

4
tests/test_cli_memory.py
View file

@ -52,7 +52,7 @@ def test_memory_tsv(tmpdir, use_stdin):
result = CliRunner().invoke(
cli.cli,
["memory", path, "select * from {}".format(sql_from)],
input=data,
input=input,
)
assert result.exit_code == 0, result.output
assert json.loads(result.output.strip()) == [
@ -102,7 +102,7 @@ def test_memory_json_nl(tmpdir, use_stdin):
result = CliRunner().invoke(
cli.cli,
["memory", path, "select * from {}".format(sql_from)],
input=data,
input=input,
)
assert result.exit_code == 0, result.output
assert json.loads(result.output.strip()) == [

1
tests/test_constructor.py
View file

@ -1,5 +1,4 @@
from sqlite_utils import Database
import pytest
def test_recursive_triggers():

3
tests/test_conversions.py
View file

@ -1,6 +1,3 @@
import pytest
def test_insert_conversion(fresh_db):
table = fresh_db["table"]
table.insert({"foo": "bar"}, conversions={"foo": "upper(?)"})

9
tests/test_create.py
View file

@ -2,7 +2,6 @@ from sqlite_utils.db import (
Index,
Database,
DescIndex,
ForeignKey,
AlterError,
NoObviousTable,
ForeignKey,
@ -148,8 +147,8 @@ def test_create_table_with_not_null(fresh_db):
)
def test_create_table_from_example(fresh_db, example, expected_columns):
people_table = fresh_db["people"]
assert None == people_table.last_rowid
assert None == people_table.last_pk
assert people_table.last_rowid is None
assert people_table.last_pk is None
people_table.insert(example)
assert 1 == people_table.last_rowid
assert 1 == people_table.last_pk
@ -515,7 +514,7 @@ def test_insert_row_alter_table(
def test_insert_row_alter_table_invalid_column_characters(fresh_db):
table = fresh_db["table"]
rowid = table.insert({"foo": "bar"}).last_pk
table.insert({"foo": "bar"}).last_pk
with pytest.raises(AssertionError):
table.insert({"foo": "baz", "new_col[abc]": 1.2}, alter=True)
@ -870,8 +869,6 @@ def test_works_with_pathlib_path(tmpdir):
@pytest.mark.skipif(pd is None, reason="pandas and numpy are not installed")
def test_create_table_numpy(fresh_db):
import numpy as np
df = pd.DataFrame({"col 1": range(3), "col 2": range(3)})
fresh_db["pandas"].insert_all(df.to_dict(orient="records"))
assert [

27
tests/test_enable_counts.py
View file

@ -14,8 +14,31 @@ def test_enable_counts_specific_table(fresh_db):
# Now enable counts
foo.enable_counts()
assert foo.triggers_dict == {
"foo_counts_insert": "CREATE TRIGGER [foo_counts_insert] AFTER INSERT ON [foo]\nBEGIN\n INSERT OR REPLACE INTO [_counts]\n VALUES (\n 'foo',\n COALESCE(\n (SELECT count FROM [_counts] WHERE [table] = 'foo'),\n 0\n ) + 1\n );\nEND",
"foo_counts_delete": "CREATE TRIGGER [foo_counts_delete] AFTER DELETE ON [foo]\nBEGIN\n INSERT OR REPLACE INTO [_counts]\n VALUES (\n 'foo',\n COALESCE(\n (SELECT count FROM [_counts] WHERE [table] = 'foo'),\n 0\n ) - 1\n );\nEND",
"foo_counts_insert": (
"CREATE TRIGGER [foo_counts_insert] AFTER INSERT ON [foo]\n"
"BEGIN\n"
" INSERT OR REPLACE INTO [_counts]\n"
" VALUES (\n 'foo',\n"
" COALESCE(\n"
" (SELECT count FROM [_counts] WHERE [table] = 'foo'),\n"
" 0\n"
" ) + 1\n"
" );\n"
"END"
),
"foo_counts_delete": (
"CREATE TRIGGER [foo_counts_delete] AFTER DELETE ON [foo]\n"
"BEGIN\n"
" INSERT OR REPLACE INTO [_counts]\n"
" VALUES (\n"
" 'foo',\n"
" COALESCE(\n"
" (SELECT count FROM [_counts] WHERE [table] = 'foo'),\n"
" 0\n"
" ) - 1\n"
" );\n"
"END"
),
}
assert fresh_db.table_names() == ["foo", "_counts"]
assert list(fresh_db["_counts"].rows) == [{"count": 10, "table": "foo"}]

2
tests/test_extract.py
View file

@ -1,4 +1,4 @@
from sqlite_utils.db import Index, InvalidColumns
from sqlite_utils.db import InvalidColumns
import itertools
import pytest

2
tests/test_extracts.py
View file

@ -1,4 +1,4 @@
from sqlite_utils.db import Index, ForeignKey
from sqlite_utils.db import Index
import pytest

1
tests/test_hypothesis.py
View file

@ -2,6 +2,7 @@ from hypothesis import given
import hypothesis.strategies as st
import sqlite_utils
# SQLite integers are -(2^63) to 2^63 - 1
@given(st.integers(-9223372036854775808, 9223372036854775807))
def test_roundtrip_integers(integer):

20
tests/test_introspect.py
View file

@ -33,7 +33,7 @@ def test_detect_fts(existing_db):
assert "woo_fts" == existing_db["woo_fts"].detect_fts()
assert "woo2_fts" == existing_db["woo2"].detect_fts()
assert "woo2_fts" == existing_db["woo2_fts"].detect_fts()
assert None == existing_db["foo"].detect_fts()
assert existing_db["foo"].detect_fts() is None
def test_tables(existing_db):
@ -175,9 +175,21 @@ def test_triggers_and_triggers_dict(fresh_db):
(t.name, t.table) for t in fresh_db["authors"].triggers
}
expected_triggers = {
"authors_ai": "CREATE TRIGGER [authors_ai] AFTER INSERT ON [authors] BEGIN\n INSERT INTO [authors_fts] (rowid, [name], [famous_works]) VALUES (new.rowid, new.[name], new.[famous_works]);\nEND",
"authors_ad": "CREATE TRIGGER [authors_ad] AFTER DELETE ON [authors] BEGIN\n INSERT INTO [authors_fts] ([authors_fts], rowid, [name], [famous_works]) VALUES('delete', old.rowid, old.[name], old.[famous_works]);\nEND",
"authors_au": "CREATE TRIGGER [authors_au] AFTER UPDATE ON [authors] BEGIN\n INSERT INTO [authors_fts] ([authors_fts], rowid, [name], [famous_works]) VALUES('delete', old.rowid, old.[name], old.[famous_works]);\n INSERT INTO [authors_fts] (rowid, [name], [famous_works]) VALUES (new.rowid, new.[name], new.[famous_works]);\nEND",
"authors_ai": (
"CREATE TRIGGER [authors_ai] AFTER INSERT ON [authors] BEGIN\n"
" INSERT INTO [authors_fts] (rowid, [name], [famous_works]) VALUES (new.rowid, new.[name], new.[famous_works]);\n"
"END"
),
"authors_ad": (
"CREATE TRIGGER [authors_ad] AFTER DELETE ON [authors] BEGIN\n"
" INSERT INTO [authors_fts] ([authors_fts], rowid, [name], [famous_works]) VALUES('delete', old.rowid, old.[name], old.[famous_works]);\n"
"END"
),
"authors_au": (
"CREATE TRIGGER [authors_au] AFTER UPDATE ON [authors] BEGIN\n"
" INSERT INTO [authors_fts] ([authors_fts], rowid, [name], [famous_works]) VALUES('delete', old.rowid, old.[name], old.[famous_works]);\n"
" INSERT INTO [authors_fts] (rowid, [name], [famous_works]) VALUES (new.rowid, new.[name], new.[famous_works]);\nEND"
),
}
assert authors.triggers_dict == expected_triggers
assert fresh_db["other"].triggers == []

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:")
with pytest.raises(AssertionError):
db = Database(conn, recreate=True)
Database(conn, recreate=True)
@pytest.mark.parametrize(

4
tests/test_register_function.py
View file

@ -55,14 +55,14 @@ def test_register_function_replace(fresh_db):
# This will fail to replace the function:
@fresh_db.register_function()
def one():
def one(): # noqa: F811
return "two"
assert "one" == fresh_db.execute("select one()").fetchone()[0]
# This will replace it
@fresh_db.register_function(replace=True)
def one():
def one(): # noqa: F811
return "two"
assert "two" == fresh_db.execute("select one()").fetchone()[0]

1
tests/test_rows.py
View file

@ -1,4 +1,3 @@
from sqlite_utils.db import Index, View
import pytest

35
tests/test_tracer.py
View file

@ -1,4 +1,3 @@
import pytest
from sqlite_utils import Database
@ -32,7 +31,9 @@ def test_tracer():
def test_with_tracer():
collected = []
tracer = lambda sql, params: collected.append((sql, params))
def tracer(sql, params):
return collected.append((sql, params))
db = Database(memory=True)
@ -48,13 +49,39 @@ def test_with_tracer():
assert collected == [
("select name from sqlite_master where type = 'view'", None),
(
"SELECT name FROM sqlite_master\n WHERE rootpage = 0\n AND (\n sql LIKE '%VIRTUAL TABLE%USING FTS%content=%dogs%'\n OR (\n tbl_name = \"dogs\"\n AND sql LIKE '%VIRTUAL TABLE%USING FTS%'\n )\n )",
(
"SELECT name FROM sqlite_master\n"
" WHERE rootpage = 0\n"
" AND (\n"
" sql LIKE '%VIRTUAL TABLE%USING FTS%content=%dogs%'\n"
" OR (\n"
' tbl_name = "dogs"\n'
" AND sql LIKE '%VIRTUAL TABLE%USING FTS%'\n"
" )\n"
" )"
),
None,
),
("select name from sqlite_master where type = 'view'", None),
("select sql from sqlite_master where name = ?", ("dogs_fts",)),
(
"with original as (\n select\n rowid,\n *\n from [dogs]\n)\nselect\n [original].*\nfrom\n [original]\n join [dogs_fts] on [original].rowid = [dogs_fts].rowid\nwhere\n [dogs_fts] match :query\norder by\n [dogs_fts].rank",
(
"with original as (\n"
" select\n"
" rowid,\n"
" *\n"
" from [dogs]\n"
")\n"
"select\n"
" [original].*\n"
"from\n"
" [original]\n"
" join [dogs_fts] on [original].rowid = [dogs_fts].rowid\n"
"where\n"
" [dogs_fts] match :query\n"
"order by\n"
" [dogs_fts].rank"
),
{"query": "Cleopaws"},
),
]

10
tests/test_transform.py
View file

@ -93,7 +93,10 @@ def test_transform_sql_table_with_primary_key(
fresh_db, params, expected_sql, use_pragma_foreign_keys
):
captured = []
tracer = lambda sql, params: captured.append((sql, params))
def tracer(sql, params):
return captured.append((sql, params))
dogs = fresh_db["dogs"]
if use_pragma_foreign_keys:
fresh_db.conn.execute("PRAGMA foreign_keys=ON")
@ -163,7 +166,10 @@ def test_transform_sql_table_with_no_primary_key(
fresh_db, params, expected_sql, use_pragma_foreign_keys
):
captured = []
tracer = lambda sql, params: captured.append((sql, params))
def tracer(sql, params):
return captured.append((sql, params))
dogs = fresh_db["dogs"]
if use_pragma_foreign_keys:
fresh_db.conn.execute("PRAGMA foreign_keys=ON")

2
tests/test_upsert.py
View file

@ -54,7 +54,7 @@ def test_upsert_compound_primary_key(fresh_db):
],
pk=("species", "id"),
)
assert None == table.last_pk
assert table.last_pk is None
table.upsert({"species": "dog", "id": 1, "age": 5}, pk=("species", "id"))
assert ("dog", 1) == table.last_pk
assert [

1
tests/test_wal.py
View file

@ -1,6 +1,5 @@
import pytest
from sqlite_utils import Database
import sqlite3
@pytest.fixture