remote/sqlite-utils
1
0
Fork 0
mirror of https://github.com/simonw/sqlite-utils.git synced 2026-09-18 14:34:24 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge branch 'main' into fts-quote

Browse source
This commit is contained in:
Simon Willison 2021-08-18 11:40:46 -07:00 • committed by GitHub
commit af989af658
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
49 changed files with 5850 additions and 571 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 == [
{

536
tests/test_cli.py
View file

@ -1,6 +1,7 @@
from sqlite_utils import cli, Database
from sqlite_utils.db import Index, ForeignKey
from click.testing import CliRunner
from unittest import mock
import json
import os
import pytest
@ -24,6 +25,22 @@ def db_path(tmpdir):
return path
@pytest.mark.parametrize(
"options",
(
["-h"],
["--help"],
["insert", "-h"],
["insert", "--help"],
),
)
def test_help(options):
result = CliRunner().invoke(cli.cli, options)
assert result.exit_code == 0
assert result.output.startswith("Usage: ")
assert "-h, --help" in result.output
def test_tables(db_path):
result = CliRunner().invoke(cli.cli, ["tables", db_path])
assert '[{"table": "Gosh"},\n {"table": "Gosh2"}]' == result.output.strip()
@ -208,6 +225,17 @@ def test_create_index(db_path):
)
def test_create_index_desc(db_path):
db = Database(db_path)
assert [] == db["Gosh"].indexes
result = CliRunner().invoke(cli.cli, ["create-index", db_path, "Gosh", "--", "-c1"])
assert result.exit_code == 0
assert (
db.execute("select sql from sqlite_master where type='index'").fetchone()[0]
== "CREATE INDEX [idx_Gosh_c1]\n ON [Gosh] ([c1] desc)"
)
@pytest.mark.parametrize(
"col_name,col_type,expected_schema",
(
@ -354,6 +382,21 @@ def test_add_column_foreign_key(db_path):
assert "table 'bobcats' does not exist" in str(result.exception)
def test_suggest_alter_if_column_missing(db_path):
db = Database(db_path)
db["authors"].insert({"id": 1, "name": "Sally"}, pk="id")
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "authors", "-"],
input='{"id": 2, "name": "Barry", "age": 43}',
)
assert result.exit_code != 0
assert result.output.strip() == (
"Error: table authors has no column named age\n\n"
"Try using --alter to add additional columns"
)
def test_index_foreign_keys(db_path):
test_add_column_foreign_key(db_path)
db = Database(db_path)
@ -367,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"]
)
@ -377,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,
[
@ -476,6 +519,13 @@ def test_vacuum(db_path):
assert 0 == result.exit_code
def test_dump(db_path):
result = CliRunner().invoke(cli.cli, ["dump", db_path])
assert result.exit_code == 0
assert result.output.startswith("BEGIN TRANSACTION;")
assert result.output.strip().endswith("COMMIT;")
@pytest.mark.parametrize("tables", ([], ["Gosh"], ["Gosh2"]))
def test_optimize(db_path, tables):
db = Database(db_path)
@ -563,8 +613,8 @@ def test_insert_simple(tmpdir):
open(json_path, "w").write(json.dumps({"name": "Cleo", "age": 4}))
result = CliRunner().invoke(cli.cli, ["insert", db_path, "dogs", json_path])
assert 0 == result.exit_code
assert [{"age": 4, "name": "Cleo"}] == Database(db_path).execute_returning_dicts(
"select * from dogs"
assert [{"age": 4, "name": "Cleo"}] == list(
Database(db_path).query("select * from dogs")
)
db = Database(db_path)
assert ["dogs"] == db.table_names()
@ -579,8 +629,8 @@ def test_insert_from_stdin(tmpdir):
input=json.dumps({"name": "Cleo", "age": 4}),
)
assert 0 == result.exit_code
assert [{"age": 4, "name": "Cleo"}] == Database(db_path).execute_returning_dicts(
"select * from dogs"
assert [{"age": 4, "name": "Cleo"}] == list(
Database(db_path).query("select * from dogs")
)
@ -598,6 +648,34 @@ def test_insert_invalid_json_error(tmpdir):
)
def test_insert_json_flatten(tmpdir):
db_path = str(tmpdir / "flat.db")
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "items", "-", "--flatten"],
input=json.dumps({"nested": {"data": 4}}),
)
assert result.exit_code == 0
assert list(Database(db_path).query("select * from items")) == [{"nested_data": 4}]
def test_insert_json_flatten_nl(tmpdir):
db_path = str(tmpdir / "flat.db")
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "items", "-", "--flatten", "--nl"],
input="\n".join(
json.dumps(item)
for item in [{"nested": {"data": 4}}, {"nested": {"other": 3}}]
),
)
assert result.exit_code == 0
assert list(Database(db_path).query("select * from items")) == [
{"nested_data": 4, "nested_other": None},
{"nested_data": None, "nested_other": 3},
]
def test_insert_with_primary_key(db_path, tmpdir):
json_path = str(tmpdir / "dog.json")
open(json_path, "w").write(json.dumps({"id": 1, "name": "Cleo", "age": 4}))
@ -605,9 +683,9 @@ def test_insert_with_primary_key(db_path, tmpdir):
cli.cli, ["insert", db_path, "dogs", json_path, "--pk", "id"]
)
assert 0 == result.exit_code
assert [{"id": 1, "age": 4, "name": "Cleo"}] == Database(
db_path
).execute_returning_dicts("select * from dogs")
assert [{"id": 1, "age": 4, "name": "Cleo"}] == list(
Database(db_path).query("select * from dogs")
)
db = Database(db_path)
assert ["id"] == db["dogs"].pks
@ -621,7 +699,7 @@ def test_insert_multiple_with_primary_key(db_path, tmpdir):
)
assert 0 == result.exit_code
db = Database(db_path)
assert dogs == db.execute_returning_dicts("select * from dogs order by id")
assert dogs == list(db.query("select * from dogs order by id"))
assert ["id"] == db["dogs"].pks
@ -637,7 +715,7 @@ def test_insert_multiple_with_compound_primary_key(db_path, tmpdir):
)
assert 0 == result.exit_code
db = Database(db_path)
assert dogs == db.execute_returning_dicts("select * from dogs order by breed, id")
assert dogs == list(db.query("select * from dogs order by breed, id"))
assert {"breed", "id"} == set(db["dogs"].pks)
assert (
"CREATE TABLE [dogs] (\n"
@ -682,7 +760,7 @@ def test_insert_binary_base64(db_path):
)
assert 0 == result.exit_code, result.output
db = Database(db_path)
actual = db.execute_returning_dicts("select content from files")
actual = list(db.query("select content from files"))
assert actual == [{"content": b"hello"}]
@ -697,7 +775,7 @@ def test_insert_newline_delimited(db_path):
assert [
{"foo": "bar", "n": 1},
{"foo": "baz", "n": 2},
] == db.execute_returning_dicts("select foo, n from from_json_nl")
] == list(db.query("select foo, n from from_json_nl"))
def test_insert_ignore(db_path, tmpdir):
@ -716,9 +794,7 @@ def test_insert_ignore(db_path, tmpdir):
)
assert 0 == result.exit_code, result.output
# ... but it should actually have no effect
assert [{"id": 1, "name": "Cleo"}] == db.execute_returning_dicts(
"select * from dogs"
)
assert [{"id": 1, "name": "Cleo"}] == list(db.query("select * from dogs"))
@pytest.mark.parametrize(
@ -781,8 +857,9 @@ def test_insert_replace(db_path, tmpdir):
)
assert 0 == result.exit_code, result.output
assert 21 == db["dogs"].count
assert insert_replace_dogs == db.execute_returning_dicts(
"select * from dogs where id in (1, 2, 21) order by id"
assert (
list(db.query("select * from dogs where id in (1, 2, 21) order by id"))
== insert_replace_dogs
)
@ -797,7 +874,7 @@ def test_insert_truncate(db_path):
assert [
{"foo": "bar", "n": 1},
{"foo": "baz", "n": 2},
] == db.execute_returning_dicts("select foo, n from from_json_nl")
] == list(db.query("select foo, n from from_json_nl"))
# Truncate and insert new rows
result = CliRunner().invoke(
cli.cli,
@ -816,7 +893,7 @@ def test_insert_truncate(db_path):
assert [
{"foo": "bam", "n": 3},
{"foo": "bat", "n": 4},
] == db.execute_returning_dicts("select foo, n from from_json_nl")
] == list(db.query("select foo, n from from_json_nl"))
def test_insert_alter(db_path, tmpdir):
@ -847,7 +924,7 @@ def test_insert_alter(db_path, tmpdir):
{"foo": "bar", "n": 1, "baz": None},
{"foo": "baz", "n": 2, "baz": None},
{"foo": "bar", "baz": 5, "n": None},
] == db.execute_returning_dicts("select foo, n, baz from from_json_nl")
] == list(db.query("select foo, n, baz from from_json_nl"))
@pytest.mark.parametrize(
@ -917,7 +994,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):
@ -939,7 +1032,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"
)
),
},
}
]
@ -1023,7 +1127,7 @@ def test_query_load_extension(use_spatialite_shortcut):
# Without --load-extension:
result = CliRunner().invoke(cli.cli, [":memory:", "select spatialite_version()"])
assert result.exit_code == 1
assert "no such function: spatialite_version" in repr(result)
assert "no such function: spatialite_version" in result.output
# With --load-extension:
if use_spatialite_shortcut:
load_extension = "spatialite"
@ -1108,17 +1212,32 @@ 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},
] == db.execute_returning_dicts("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_flatten(tmpdir):
db_path = str(tmpdir / "flat.db")
db = Database(db_path)
db["upsert_me"].insert({"id": 1, "name": "Example"}, pk="id")
result = CliRunner().invoke(
cli.cli,
["upsert", db_path, "upsert_me", "-", "--flatten", "--pk", "id", "--alter"],
input=json.dumps({"id": 1, "nested": {"two": 2}}),
)
assert result.exit_code == 0
assert list(db.query("select * from upsert_me")) == [
{"id": 1, "name": "Example", "nested_two": 2}
]
def test_upsert_alter(db_path, tmpdir):
@ -1137,7 +1256,11 @@ def test_upsert_alter(db_path, tmpdir):
cli.cli, ["upsert", db_path, "dogs", json_path, "--pk", "id"]
)
assert 1 == result.exit_code
assert "no such column: age" == str(result.exception)
assert (
"Error: no such column: age\n\n"
"sql = UPDATE [dogs] SET [age] = ? WHERE [id] = ?\n"
"parameters = [5, 1]"
) == result.output.strip()
# Should succeed with --alter
result = CliRunner().invoke(
cli.cli, ["upsert", db_path, "dogs", json_path, "--pk", "id", "--alter"]
@ -1145,7 +1268,7 @@ def test_upsert_alter(db_path, tmpdir):
assert 0 == result.exit_code
assert [
{"id": 1, "name": "Cleo", "age": 5},
] == db.execute_returning_dicts("select * from dogs order by id")
] == list(db.query("select * from dogs order by id"))
@pytest.mark.parametrize(
@ -1499,7 +1622,7 @@ def test_query_update(db_path, args, expected):
cli.cli, [db_path, "update dogs set age = 5 where name = 'Cleo'"] + args
)
assert expected == result.output.strip()
assert db.execute_returning_dicts("select * from dogs") == [
assert list(db.query("select * from dogs")) == [
{"id": 1, "age": 5, "name": "Cleo"},
]
@ -1547,47 +1670,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"
")"
),
),
],
)
@ -1636,9 +1824,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"
")"
)
@ -1652,22 +1844,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)",
),
],
@ -1776,7 +1994,87 @@ def test_search(tmpdir, fts, extra_arg, expected):
assert result.output.replace("\r", "") == expected
_TRIGGERS_EXPECTED = '[{"name": "blah", "table": "articles", "sql": "CREATE TRIGGER blah AFTER INSERT ON articles\\nBEGIN\\n UPDATE counter SET count = count + 1;\\nEND"}]\n'
def test_indexes(tmpdir):
db_path = str(tmpdir / "test.db")
db = Database(db_path)
db.conn.executescript(
"""
create table Gosh (c1 text, c2 text, c3 text);
create index Gosh_idx on Gosh(c2, c3 desc);
"""
)
result = CliRunner().invoke(
cli.cli,
["indexes", str(db_path)],
catch_exceptions=False,
)
assert result.exit_code == 0
assert json.loads(result.output) == [
{
"table": "Gosh",
"index_name": "Gosh_idx",
"seqno": 0,
"cid": 1,
"name": "c2",
"desc": 0,
"coll": "BINARY",
"key": 1,
},
{
"table": "Gosh",
"index_name": "Gosh_idx",
"seqno": 1,
"cid": 2,
"name": "c3",
"desc": 1,
"coll": "BINARY",
"key": 1,
},
]
result2 = CliRunner().invoke(
cli.cli,
["indexes", str(db_path), "--aux"],
catch_exceptions=False,
)
assert result2.exit_code == 0
assert json.loads(result2.output) == [
{
"table": "Gosh",
"index_name": "Gosh_idx",
"seqno": 0,
"cid": 1,
"name": "c2",
"desc": 0,
"coll": "BINARY",
"key": 1,
},
{
"table": "Gosh",
"index_name": "Gosh_idx",
"seqno": 1,
"cid": 2,
"name": "c3",
"desc": 1,
"coll": "BINARY",
"key": 1,
},
{
"table": "Gosh",
"index_name": "Gosh_idx",
"seqno": 2,
"cid": -1,
"name": None,
"desc": 0,
"coll": "BINARY",
"key": 0,
},
]
_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(
@ -1813,6 +2111,60 @@ def test_triggers(tmpdir, extra_args, expected):
assert result.output == expected
@pytest.mark.parametrize(
"options,expected",
(
(
[],
(
"CREATE TABLE [dogs] (\n"
" [id] INTEGER,\n"
" [name] TEXT\n"
");\n"
"CREATE TABLE [chickens] (\n"
" [id] INTEGER,\n"
" [name] TEXT,\n"
" [breed] TEXT\n"
");\n"
"CREATE INDEX [idx_chickens_breed]\n"
" ON [chickens] ([breed]);\n"
),
),
(
["dogs"],
("CREATE TABLE [dogs] (\n" " [id] INTEGER,\n" " [name] TEXT\n" ")\n"),
),
(
["chickens", "dogs"],
(
"CREATE TABLE [chickens] (\n"
" [id] INTEGER,\n"
" [name] TEXT,\n"
" [breed] TEXT\n"
")\n"
"CREATE TABLE [dogs] (\n"
" [id] INTEGER,\n"
" [name] TEXT\n"
")\n"
),
),
),
)
def test_schema(tmpdir, options, expected):
db_path = str(tmpdir / "test.db")
db = Database(db_path)
db["dogs"].create({"id": int, "name": str})
db["chickens"].create({"id": int, "name": str, "breed": str})
db["chickens"].create_index(["breed"])
result = CliRunner().invoke(
cli.cli,
["schema", db_path] + options,
catch_exceptions=False,
)
assert result.exit_code == 0
assert result.output == expected
def test_long_csv_column_value(tmpdir):
db_path = str(tmpdir / "test.db")
csv_path = str(tmpdir / "test.csv")
@ -1889,3 +2241,85 @@ def test_attach(tmpdir):
{"id": 1, "text": "foo"},
{"id": 1, "text": "bar"},
]
def test_csv_insert_bom(tmpdir):
db_path = str(tmpdir / "test.db")
bom_csv_path = str(tmpdir / "bom.csv")
with open(bom_csv_path, "wb") as fp:
fp.write(b"\xef\xbb\xbfname,age\nCleo,5")
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "broken", bom_csv_path, "--encoding", "utf-8", "--csv"],
catch_exceptions=False,
)
assert result.exit_code == 0
result2 = CliRunner().invoke(
cli.cli,
["insert", db_path, "fixed", bom_csv_path, "--csv"],
catch_exceptions=False,
)
assert result2.exit_code == 0
db = Database(db_path)
tables = db.execute("select name, sql from sqlite_master").fetchall()
assert tables == [
("broken", "CREATE TABLE [broken] (\n [\ufeffname] TEXT,\n [age] TEXT\n)"),
("fixed", "CREATE TABLE [fixed] (\n [name] TEXT,\n [age] TEXT\n)"),
]
@pytest.mark.parametrize("option_or_env_var", (None, "-d", "--detect-types"))
def test_insert_detect_types(tmpdir, option_or_env_var):
db_path = str(tmpdir / "test.db")
data = "name,age,weight\nCleo,6,45.5\nDori,1,3.5"
extra = []
if option_or_env_var:
extra = [option_or_env_var]
def _test():
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "creatures", "-", "--csv"] + extra,
catch_exceptions=False,
input=data,
)
assert result.exit_code == 0
db = Database(db_path)
assert list(db["creatures"].rows) == [
{"name": "Cleo", "age": 6, "weight": 45.5},
{"name": "Dori", "age": 1, "weight": 3.5},
]
if option_or_env_var is None:
# Use environemnt variable instead of option
with mock.patch.dict(os.environ, {"SQLITE_UTILS_DETECT_TYPES": "1"}):
_test()
else:
_test()
@pytest.mark.parametrize(
"input,expected",
(
({"foo": {"bar": 1}}, {"foo_bar": 1}),
({"foo": {"bar": [1, 2, {"baz": 3}]}}, {"foo_bar": [1, 2, {"baz": 3}]}),
({"foo": {"bar": 1, "baz": {"three": 3}}}, {"foo_bar": 1, "foo_baz_three": 3}),
),
)
def test_flatten_helper(input, expected):
assert dict(cli._flatten(input)) == expected
def test_integer_overflow_error(tmpdir):
db_path = str(tmpdir / "test.db")
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "items", "-"],
input=json.dumps({"bignumber": 34223049823094832094802398430298048240}),
)
assert result.exit_code == 1
assert result.output == (
"Error: Python int too large to convert to SQLite INTEGER\n\n"
"sql = INSERT INTO [items] ([bignumber]) VALUES (?);\n"
"parameters = [34223049823094832094802398430298048240]\n"
)

517
tests/test_cli_convert.py Normal file
View file

@ -0,0 +1,517 @@
from click.testing import CliRunner
from sqlite_utils import cli
import sqlite_utils
import json
import textwrap
import pathlib
import pytest
@pytest.fixture
def test_db_and_path(fresh_db_and_path):
db, db_path = fresh_db_and_path
db["example"].insert_all(
[
{"id": 1, "dt": "5th October 2019 12:04"},
{"id": 2, "dt": "6th October 2019 00:05:06"},
{"id": 3, "dt": ""},
{"id": 4, "dt": None},
],
pk="id",
)
return db, db_path
@pytest.fixture
def fresh_db_and_path(tmpdir):
db_path = str(pathlib.Path(tmpdir) / "data.db")
db = sqlite_utils.Database(db_path)
return db, db_path
@pytest.mark.parametrize(
"code",
[
"return value.replace('October', 'Spooktober')",
# Return is optional:
"value.replace('October', 'Spooktober')",
],
)
def test_convert_single_line(test_db_and_path, code):
db, db_path = test_db_and_path
result = CliRunner().invoke(cli.cli, ["convert", db_path, "example", "dt", code])
assert 0 == result.exit_code, result.output
assert [
{"id": 1, "dt": "5th Spooktober 2019 12:04"},
{"id": 2, "dt": "6th Spooktober 2019 00:05:06"},
{"id": 3, "dt": ""},
{"id": 4, "dt": None},
] == list(db["example"].rows)
def test_convert_multiple_lines(test_db_and_path):
db, db_path = test_db_and_path
result = CliRunner().invoke(
cli.cli,
[
"convert",
db_path,
"example",
"dt",
"v = value.replace('October', 'Spooktober')\nreturn v.upper()",
],
)
assert 0 == result.exit_code, result.output
assert [
{"id": 1, "dt": "5TH SPOOKTOBER 2019 12:04"},
{"id": 2, "dt": "6TH SPOOKTOBER 2019 00:05:06"},
{"id": 3, "dt": ""},
{"id": 4, "dt": None},
] == list(db["example"].rows)
def test_convert_import(test_db_and_path):
db, db_path = test_db_and_path
result = CliRunner().invoke(
cli.cli,
[
"convert",
db_path,
"example",
"dt",
"return re.sub('O..', 'OXX', value)",
"--import",
"re",
],
)
assert 0 == result.exit_code, result.output
assert [
{"id": 1, "dt": "5th OXXober 2019 12:04"},
{"id": 2, "dt": "6th OXXober 2019 00:05:06"},
{"id": 3, "dt": ""},
{"id": 4, "dt": None},
] == list(db["example"].rows)
def test_convert_dryrun(test_db_and_path):
db, db_path = test_db_and_path
result = CliRunner().invoke(
cli.cli,
[
"convert",
db_path,
"example",
"dt",
"return re.sub('O..', 'OXX', value)",
"--import",
"re",
"--dry-run",
],
)
assert result.exit_code == 0
assert result.output.strip() == (
"5th October 2019 12:04\n"
" --- becomes:\n"
"5th OXXober 2019 12:04\n"
"\n"
"6th October 2019 00:05:06\n"
" --- becomes:\n"
"6th OXXober 2019 00:05:06\n"
"\n"
"\n"
" --- becomes:\n"
"\n"
"\n"
"None\n"
" --- becomes:\n"
"None\n\n"
"Would affect 4 rows"
)
# But it should not have actually modified the table data
assert list(db["example"].rows) == [
{"id": 1, "dt": "5th October 2019 12:04"},
{"id": 2, "dt": "6th October 2019 00:05:06"},
{"id": 3, "dt": ""},
{"id": 4, "dt": None},
]
# Test with a where clause too
result = CliRunner().invoke(
cli.cli,
[
"convert",
db_path,
"example",
"dt",
"return re.sub('O..', 'OXX', value)",
"--import",
"re",
"--dry-run",
"--where",
"id = :id",
"-p",
"id",
"4",
],
)
assert result.exit_code == 0
assert result.output.strip().split("\n")[-1] == "Would affect 1 row"
@pytest.mark.parametrize("drop", (True, False))
def test_convert_output_column(test_db_and_path, drop):
db, db_path = test_db_and_path
args = [
"convert",
db_path,
"example",
"dt",
"value.replace('October', 'Spooktober')",
"--output",
"newcol",
]
if drop:
args += ["--drop"]
result = CliRunner().invoke(cli.cli, args)
assert 0 == result.exit_code, result.output
expected = [
{
"id": 1,
"dt": "5th October 2019 12:04",
"newcol": "5th Spooktober 2019 12:04",
},
{
"id": 2,
"dt": "6th October 2019 00:05:06",
"newcol": "6th Spooktober 2019 00:05:06",
},
{"id": 3, "dt": "", "newcol": ""},
{"id": 4, "dt": None, "newcol": None},
]
if drop:
for row in expected:
del row["dt"]
assert list(db["example"].rows) == expected
@pytest.mark.parametrize(
"output_type,expected",
(
("text", [(1, "1"), (2, "2"), (3, "3"), (4, "4")]),
("float", [(1, 1.0), (2, 2.0), (3, 3.0), (4, 4.0)]),
("integer", [(1, 1), (2, 2), (3, 3), (4, 4)]),
(None, [(1, "1"), (2, "2"), (3, "3"), (4, "4")]),
),
)
def test_convert_output_column_output_type(test_db_and_path, output_type, expected):
db, db_path = test_db_and_path
args = [
"convert",
db_path,
"example",
"id",
"value",
"--output",
"new_id",
]
if output_type:
args += ["--output-type", output_type]
result = CliRunner().invoke(
cli.cli,
args,
)
assert 0 == result.exit_code, result.output
assert expected == list(db.execute("select id, new_id from example"))
@pytest.mark.parametrize(
"options,expected_error",
[
(
[
"dt",
"id",
"value.replace('October', 'Spooktober')",
"--output",
"newcol",
],
"Cannot use --output with more than one column",
),
(
[
"dt",
"value.replace('October', 'Spooktober')",
"--output",
"newcol",
"--output-type",
"invalid",
],
"Error: Invalid value for '--output-type'",
),
(
[
"value.replace('October', 'Spooktober')",
],
"Missing argument 'COLUMNS...'",
),
],
)
def test_convert_output_error(test_db_and_path, options, expected_error):
db_path = test_db_and_path[1]
result = CliRunner().invoke(
cli.cli,
[
"convert",
db_path,
"example",
]
+ options,
)
assert result.exit_code != 0
assert expected_error in result.output
@pytest.mark.parametrize("drop", (True, False))
def test_convert_multi(fresh_db_and_path, drop):
db, db_path = fresh_db_and_path
db["creatures"].insert_all(
[
{"id": 1, "name": "Simon"},
{"id": 2, "name": "Cleo"},
],
pk="id",
)
args = [
"convert",
db_path,
"creatures",
"name",
"--multi",
'{"upper": value.upper(), "lower": value.lower()}',
]
if drop:
args += ["--drop"]
result = CliRunner().invoke(cli.cli, args)
assert result.exit_code == 0, result.output
expected = [
{"id": 1, "name": "Simon", "upper": "SIMON", "lower": "simon"},
{"id": 2, "name": "Cleo", "upper": "CLEO", "lower": "cleo"},
]
if drop:
for row in expected:
del row["name"]
assert list(db["creatures"].rows) == expected
def test_convert_multi_complex_column_types(fresh_db_and_path):
db, db_path = fresh_db_and_path
db["rows"].insert_all(
[
{"id": 1},
{"id": 2},
{"id": 3},
{"id": 4},
],
pk="id",
)
code = textwrap.dedent(
"""
if value == 1:
return {"is_str": "", "is_float": 1.2, "is_int": None}
elif value == 2:
return {"is_float": 1, "is_int": 12}
elif value == 3:
return {"is_bytes": b"blah"}
"""
)
result = CliRunner().invoke(
cli.cli,
[
"convert",
db_path,
"rows",
"id",
"--multi",
code,
],
)
assert result.exit_code == 0, result.output
assert list(db["rows"].rows) == [
{"id": 1, "is_str": "", "is_float": 1.2, "is_int": None, "is_bytes": None},
{"id": 2, "is_str": None, "is_float": 1.0, "is_int": 12, "is_bytes": None},
{
"id": 3,
"is_str": None,
"is_float": None,
"is_int": None,
"is_bytes": b"blah",
},
{"id": 4, "is_str": None, "is_float": None, "is_int": None, "is_bytes": None},
]
assert db["rows"].schema == (
"CREATE TABLE [rows] (\n"
" [id] INTEGER PRIMARY KEY\n"
", [is_str] TEXT, [is_float] FLOAT, [is_int] INTEGER, [is_bytes] BLOB)"
)
@pytest.mark.parametrize("delimiter", [None, ";", "-"])
def test_recipe_jsonsplit(tmpdir, delimiter):
db_path = str(pathlib.Path(tmpdir) / "data.db")
db = sqlite_utils.Database(db_path)
db["example"].insert_all(
[
{"id": 1, "tags": (delimiter or ",").join(["foo", "bar"])},
{"id": 2, "tags": (delimiter or ",").join(["bar", "baz"])},
],
pk="id",
)
code = "r.jsonsplit(value)"
if delimiter:
code = 'recipes.jsonsplit(value, delimiter="{}")'.format(delimiter)
args = ["convert", db_path, "example", "tags", code]
result = CliRunner().invoke(cli.cli, args)
assert 0 == result.exit_code, result.output
assert list(db["example"].rows) == [
{"id": 1, "tags": '["foo", "bar"]'},
{"id": 2, "tags": '["bar", "baz"]'},
]
@pytest.mark.parametrize(
"type,expected_array",
(
(None, ["1", "2", "3"]),
("float", [1.0, 2.0, 3.0]),
("int", [1, 2, 3]),
),
)
def test_recipe_jsonsplit_type(fresh_db_and_path, type, expected_array):
db, db_path = fresh_db_and_path
db["example"].insert_all(
[
{"id": 1, "records": "1,2,3"},
],
pk="id",
)
code = "r.jsonsplit(value)"
if type:
code = "recipes.jsonsplit(value, type={})".format(type)
args = ["convert", db_path, "example", "records", code]
result = CliRunner().invoke(cli.cli, args)
assert 0 == result.exit_code, result.output
assert json.loads(db["example"].get(1)["records"]) == expected_array
@pytest.mark.parametrize("drop", (True, False))
def test_recipe_jsonsplit_output(fresh_db_and_path, drop):
db, db_path = fresh_db_and_path
db["example"].insert_all(
[
{"id": 1, "records": "1,2,3"},
],
pk="id",
)
code = "r.jsonsplit(value)"
args = ["convert", db_path, "example", "records", code, "--output", "tags"]
if drop:
args += ["--drop"]
result = CliRunner().invoke(cli.cli, args)
assert 0 == result.exit_code, result.output
expected = {
"id": 1,
"records": "1,2,3",
"tags": '["1", "2", "3"]',
}
if drop:
del expected["records"]
assert db["example"].get(1) == expected
def test_cannot_use_drop_without_multi_or_output(fresh_db_and_path):
args = ["convert", fresh_db_and_path[1], "example", "records", "value", "--drop"]
result = CliRunner().invoke(cli.cli, args)
assert result.exit_code == 1, result.output
assert "Error: --drop can only be used with --output or --multi" in result.output
def test_cannot_use_multi_with_more_than_one_column(fresh_db_and_path):
args = [
"convert",
fresh_db_and_path[1],
"example",
"records",
"othercol",
"value",
"--multi",
]
result = CliRunner().invoke(cli.cli, args)
assert result.exit_code == 1, result.output
assert "Error: Cannot use --multi with more than one column" in result.output
def test_multi_with_bad_function(test_db_and_path):
args = [
"convert",
test_db_and_path[1],
"example",
"dt",
"value.upper()",
"--multi",
]
result = CliRunner().invoke(cli.cli, args)
assert result.exit_code == 1, result.output
assert "When using --multi code must return a Python dictionary" in result.output
def test_convert_where(test_db_and_path):
db, db_path = test_db_and_path
result = CliRunner().invoke(
cli.cli,
[
"convert",
db_path,
"example",
"dt",
"str(value).upper()",
"--where",
"id = :id",
"-p",
"id",
2,
],
)
assert result.exit_code == 0, result.output
assert list(db["example"].rows) == [
{"id": 1, "dt": "5th October 2019 12:04"},
{"id": 2, "dt": "6TH OCTOBER 2019 00:05:06"},
{"id": 3, "dt": ""},
{"id": 4, "dt": None},
]
def test_convert_where_multi(fresh_db_and_path):
db, db_path = fresh_db_and_path
db["names"].insert_all(
[{"id": 1, "name": "Cleo"}, {"id": 2, "name": "Bants"}], pk="id"
)
result = CliRunner().invoke(
cli.cli,
[
"convert",
db_path,
"names",
"name",
'{"upper": value.upper()}',
"--where",
"id = :id",
"-p",
"id",
2,
"--multi",
],
)
assert 0 == result.exit_code, result.output
assert list(db["names"].rows) == [
{"id": 1, "name": "Cleo", "upper": None},
{"id": 2, "name": "Bants", "upper": "BANTS"},
]

222
tests/test_cli_memory.py Normal file
View file

@ -0,0 +1,222 @@
import json
import pytest
from click.testing import CliRunner
from sqlite_utils import Database, cli
def test_memory_basic():
result = CliRunner().invoke(cli.cli, ["memory", "select 1 + 1"])
assert result.exit_code == 0
assert result.output.strip() == '[{"1 + 1": 2}]'
@pytest.mark.parametrize("sql_from", ("test", "t", "t1"))
@pytest.mark.parametrize("use_stdin", (True, False))
def test_memory_csv(tmpdir, sql_from, use_stdin):
content = "id,name\n1,Cleo\n2,Bants"
input = None
if use_stdin:
input = content
csv_path = "-"
if sql_from == "test":
sql_from = "stdin"
else:
csv_path = str(tmpdir / "test.csv")
open(csv_path, "w").write(content)
result = CliRunner().invoke(
cli.cli,
["memory", csv_path, "select * from {}".format(sql_from), "--nl"],
input=input,
)
assert result.exit_code == 0
assert (
result.output.strip() == '{"id": 1, "name": "Cleo"}\n{"id": 2, "name": "Bants"}'
)
@pytest.mark.parametrize("use_stdin", (True, False))
def test_memory_tsv(tmpdir, use_stdin):
data = "id\tname\n1\tCleo\n2\tBants"
if use_stdin:
input = data
path = "stdin:tsv"
sql_from = "stdin"
else:
input = None
path = str(tmpdir / "chickens.tsv")
open(path, "w").write(data)
path = path + ":tsv"
sql_from = "chickens"
result = CliRunner().invoke(
cli.cli,
["memory", path, "select * from {}".format(sql_from)],
input=input,
)
assert result.exit_code == 0, result.output
assert json.loads(result.output.strip()) == [
{"id": 1, "name": "Cleo"},
{"id": 2, "name": "Bants"},
]
@pytest.mark.parametrize("use_stdin", (True, False))
def test_memory_json(tmpdir, use_stdin):
data = '[{"name": "Bants"}, {"name": "Dori", "age": 1, "nested": {"nest": 1}}]'
if use_stdin:
input = data
path = "stdin:json"
sql_from = "stdin"
else:
input = None
path = str(tmpdir / "chickens.json")
open(path, "w").write(data)
path = path + ":json"
sql_from = "chickens"
result = CliRunner().invoke(
cli.cli,
["memory", path, "select * from {}".format(sql_from)],
input=input,
)
assert result.exit_code == 0, result.output
assert json.loads(result.output.strip()) == [
{"name": "Bants", "age": None, "nested": None},
{"name": "Dori", "age": 1, "nested": '{"nest": 1}'},
]
@pytest.mark.parametrize("use_stdin", (True, False))
def test_memory_json_nl(tmpdir, use_stdin):
data = '{"name": "Bants"}\n\n{"name": "Dori"}'
if use_stdin:
input = data
path = "stdin:nl"
sql_from = "stdin"
else:
input = None
path = str(tmpdir / "chickens.json")
open(path, "w").write(data)
path = path + ":nl"
sql_from = "chickens"
result = CliRunner().invoke(
cli.cli,
["memory", path, "select * from {}".format(sql_from)],
input=input,
)
assert result.exit_code == 0, result.output
assert json.loads(result.output.strip()) == [
{"name": "Bants"},
{"name": "Dori"},
]
@pytest.mark.parametrize("use_stdin", (True, False))
def test_memory_csv_encoding(tmpdir, use_stdin):
latin1_csv = (
b"date,name,latitude,longitude\n" b"2020-03-04,S\xe3o Paulo,-23.561,-46.645\n"
)
input = None
if use_stdin:
input = latin1_csv
csv_path = "-"
sql_from = "stdin"
else:
csv_path = str(tmpdir / "test.csv")
with open(csv_path, "wb") as fp:
fp.write(latin1_csv)
sql_from = "test"
# Without --encoding should error:
assert (
CliRunner()
.invoke(
cli.cli,
["memory", csv_path, "select * from {}".format(sql_from), "--nl"],
input=input,
)
.exit_code
== 1
)
# With --encoding should work:
result = CliRunner().invoke(
cli.cli,
["memory", "-", "select * from stdin", "--encoding", "latin-1", "--nl"],
input=latin1_csv,
)
assert result.exit_code == 0, result.output
assert json.loads(result.output.strip()) == {
"date": "2020-03-04",
"name": "São Paulo",
"latitude": -23.561,
"longitude": -46.645,
}
@pytest.mark.parametrize("extra_args", ([], ["select 1"]))
def test_memory_dump(extra_args):
result = CliRunner().invoke(
cli.cli,
["memory", "-"] + extra_args + ["--dump"],
input="id,name\n1,Cleo\n2,Bants",
)
assert result.exit_code == 0
assert result.output.strip() == (
"BEGIN TRANSACTION;\n"
'CREATE TABLE "stdin" (\n'
" [id] INTEGER,\n"
" [name] TEXT\n"
");\n"
"INSERT INTO \"stdin\" VALUES(1,'Cleo');\n"
"INSERT INTO \"stdin\" VALUES(2,'Bants');\n"
"CREATE VIEW t1 AS select * from [stdin];\n"
"CREATE VIEW t AS select * from [stdin];\n"
"COMMIT;"
)
@pytest.mark.parametrize("extra_args", ([], ["select 1"]))
def test_memory_schema(extra_args):
result = CliRunner().invoke(
cli.cli,
["memory", "-"] + extra_args + ["--schema"],
input="id,name\n1,Cleo\n2,Bants",
)
assert result.exit_code == 0
assert result.output.strip() == (
'CREATE TABLE "stdin" (\n'
" [id] INTEGER,\n"
" [name] TEXT\n"
");\n"
"CREATE VIEW t1 AS select * from [stdin];\n"
"CREATE VIEW t AS select * from [stdin];"
)
@pytest.mark.parametrize("extra_args", ([], ["select 1"]))
def test_memory_save(tmpdir, extra_args):
save_to = str(tmpdir / "save.db")
result = CliRunner().invoke(
cli.cli,
["memory", "-"] + extra_args + ["--save", save_to],
input="id,name\n1,Cleo\n2,Bants",
)
assert result.exit_code == 0
db = Database(save_to)
assert list(db["stdin"].rows) == [
{"id": 1, "name": "Cleo"},
{"id": 2, "name": "Bants"},
]
@pytest.mark.parametrize("option", ("-n", "--no-detect-types"))
def test_memory_no_detect_types(option):
result = CliRunner().invoke(
cli.cli,
["memory", "-", "select * from stdin"] + [option],
input="id,name,weight\n1,Cleo,45.5\n2,Bants,3.5",
)
assert result.exit_code == 0, result.output
assert json.loads(result.output.strip()) == [
{"id": "1", "name": "Cleo", "weight": "45.5"},
{"id": "2", "name": "Bants", "weight": "3.5"},
]

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(?)"})

117
tests/test_convert.py Normal file
View file

@ -0,0 +1,117 @@
from sqlite_utils.db import BadMultiValues
import pytest
@pytest.mark.parametrize(
"columns,fn,expected",
(
(
"title",
lambda value: value.upper(),
{"title": "MIXED CASE", "abstract": "Abstract"},
),
(
["title", "abstract"],
lambda value: value.upper(),
{"title": "MIXED CASE", "abstract": "ABSTRACT"},
),
),
)
def test_convert(fresh_db, columns, fn, expected):
table = fresh_db["table"]
table.insert({"title": "Mixed Case", "abstract": "Abstract"})
table.convert(columns, fn)
assert list(table.rows) == [expected]
@pytest.mark.parametrize(
"where,where_args", (("id > 1", None), ("id > :id", {"id": 1}), ("id > ?", [1]))
)
def test_convert_where(fresh_db, where, where_args):
table = fresh_db["table"]
table.insert_all(
[
{"id": 1, "title": "One"},
{"id": 2, "title": "Two"},
],
pk="id",
)
table.convert(
"title", lambda value: value.upper(), where=where, where_args=where_args
)
assert list(table.rows) == [{"id": 1, "title": "One"}, {"id": 2, "title": "TWO"}]
@pytest.mark.parametrize(
"drop,expected",
(
(False, {"title": "Mixed Case", "other": "MIXED CASE"}),
(True, {"other": "MIXED CASE"}),
),
)
def test_convert_output(fresh_db, drop, expected):
table = fresh_db["table"]
table.insert({"title": "Mixed Case"})
table.convert("title", lambda v: v.upper(), output="other", drop=drop)
assert list(table.rows) == [expected]
def test_convert_output_multiple_column_error(fresh_db):
table = fresh_db["table"]
with pytest.raises(AssertionError) 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)
@pytest.mark.parametrize(
"type,expected",
(
(int, {"other": 123}),
(float, {"other": 123.0}),
),
)
def test_convert_output_type(fresh_db, type, expected):
table = fresh_db["table"]
table.insert({"number": "123"})
table.convert("number", lambda v: v, output="other", output_type=type, drop=True)
assert list(table.rows) == [expected]
def test_convert_multi(fresh_db):
table = fresh_db["table"]
table.insert({"title": "Mixed Case"})
table.convert(
"title", lambda v: {"upper": v.upper(), "lower": v.lower()}, multi=True
)
assert list(table.rows) == [
{"title": "Mixed Case", "upper": "MIXED CASE", "lower": "mixed case"}
]
def test_convert_multi_where(fresh_db):
table = fresh_db["table"]
table.insert_all(
[
{"id": 1, "title": "One"},
{"id": 2, "title": "Two"},
],
pk="id",
)
table.convert(
"title",
lambda v: {"upper": v.upper(), "lower": v.lower()},
multi=True,
where="id > ?",
where_args=[1],
)
assert list(table.rows) == [
{"id": 1, "lower": None, "title": "One", "upper": None},
{"id": 2, "lower": "two", "title": "Two", "upper": "TWO"},
]
def test_convert_multi_exception(fresh_db):
table = fresh_db["table"]
table.insert({"title": "Mixed Case"})
with pytest.raises(BadMultiValues):
table.convert("title", lambda v: v.upper(), multi=True)

48
tests/test_create.py
View file

@ -1,7 +1,7 @@
from sqlite_utils.db import (
Index,
Database,
ForeignKey,
DescIndex,
AlterError,
NoObviousTable,
ForeignKey,
@ -147,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
@ -514,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)
@ -739,6 +739,19 @@ def test_create_index_if_not_exists(fresh_db):
dogs.create_index(["name"], if_not_exists=True)
def test_create_index_desc(fresh_db):
dogs = fresh_db["dogs"]
dogs.insert({"name": "Cleo", "twitter": "cleopaws", "age": 3, "is good dog": True})
assert [] == dogs.indexes
dogs.create_index([DescIndex("age"), "name"])
sql = fresh_db.execute(
"select sql from sqlite_master where name='idx_dogs_age_name'"
).fetchone()[0]
assert sql == (
"CREATE INDEX [idx_dogs_age_name]\n" " ON [dogs] ([age] desc, [name])"
)
@pytest.mark.parametrize(
"data_structure",
(
@ -748,7 +761,7 @@ def test_create_index_if_not_exists(fresh_db):
{"dictionary": {"nested": "complex"}},
collections.OrderedDict(
[
("key1", {"nested": "complex"}),
("key1", {"nested": ["cømplex"]}),
("key2", "foo"),
]
),
@ -762,6 +775,14 @@ def test_insert_dictionaries_and_lists_as_json(fresh_db, data_structure):
assert data_structure == json.loads(row[1])
def test_insert_list_nested_unicode(fresh_db):
fresh_db["test"].insert(
{"id": 1, "data": {"key1": {"nested": ["cømplex"]}}}, pk="id"
)
row = fresh_db.execute("select id, data from test").fetchone()
assert row[1] == '{"key1": {"nested": ["cømplex"]}}'
def test_insert_uuid(fresh_db):
uuid4 = uuid.uuid4()
fresh_db["test"].insert({"uuid": uuid4})
@ -805,8 +826,8 @@ def test_insert_thousands_adds_extra_columns_after_first_100_with_alter(fresh_db
+ [{"i": 101, "extra": "Should trigger ALTER"}],
alter=True,
)
rows = fresh_db.execute_returning_dicts("select * from test where i = 101")
assert [{"i": 101, "word": None, "extra": "Should trigger ALTER"}] == rows
rows = list(fresh_db.query("select * from test where i = 101"))
assert rows == [{"i": 101, "word": None, "extra": "Should trigger ALTER"}]
def test_insert_ignore(fresh_db):
@ -817,8 +838,8 @@ def test_insert_ignore(fresh_db):
# Using ignore=True should cause our insert to be silently ignored
fresh_db["test"].insert({"id": 1, "bar": 3}, pk="id", ignore=True)
# Only one row, and it should be bar=2, not bar=3
rows = fresh_db.execute_returning_dicts("select * from test")
assert [{"id": 1, "bar": 2}] == rows
rows = list(fresh_db.query("select * from test"))
assert rows == [{"id": 1, "bar": 2}]
def test_insert_hash_id(fresh_db):
@ -848,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 [
@ -969,6 +988,13 @@ def test_insert_all_empty_list(fresh_db):
assert 1 == fresh_db["t"].count
def test_insert_all_single_column(fresh_db):
table = fresh_db["table"]
table.insert_all([{"name": "Cleo"}], pk="name")
assert [{"name": "Cleo"}] == list(table.rows)
assert table.pks == ["name"]
def test_create_with_a_null_column(fresh_db):
record = {"name": "Name", "description": None}
fresh_db["t"].insert(record)

33
tests/test_docs.py
View file

@ -1,10 +1,12 @@
from sqlite_utils import cli
from click.testing import CliRunner
from sqlite_utils import cli, recipes
from pathlib import Path
import pytest
import re
docs_path = Path(__file__).parent.parent / "docs"
commands_re = re.compile(r"(?:\$ | )sqlite-utils (\S+) ")
recipes_re = re.compile(r"r\.(\w+)\(")
@pytest.fixture(scope="session")
@ -17,11 +19,36 @@ def documented_commands():
}
@pytest.fixture(scope="session")
def documented_recipes():
rst = (docs_path / "cli.rst").read_text()
return set(recipes_re.findall(rst))
@pytest.mark.parametrize("command", cli.cli.commands.keys())
def test_commands_are_documented(documented_commands, command):
assert command in documented_commands
@pytest.mark.parametrize("command", cli.cli.commands.values())
def test_commands_have_docstrings(command):
assert command.__doc__, "{} is missing a docstring".format(command)
def test_commands_have_help(command):
assert command.help, "{} is missing its help".format(command)
def test_convert_help():
result = CliRunner().invoke(cli.cli, ["convert", "--help"])
assert result.exit_code == 0
for expected in (
"r.jsonsplit(value, ",
"r.parsedate(value, ",
"r.parsedatetime(value, ",
):
assert expected in result.output
@pytest.mark.parametrize(
"recipe",
[n for n in dir(recipes) if not n.startswith("_") and n not in ("json", "parser")],
)
def test_recipes_are_documented(documented_recipes, recipe):
assert recipe in documented_recipes

29
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"}]
@ -109,7 +132,7 @@ def test_uses_counts_after_enable_counts(counts_db_path):
assert db["foo"].count == 1
assert logged == [
("select name from sqlite_master where type = 'view'", None),
("select count(*) from [foo]", None),
("select count(*) from [foo]", []),
]
logged.clear()
assert not db.use_counts_table

17
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
@ -126,12 +126,25 @@ def test_extract_rowid_table(fresh_db):
fresh_db["tree"].extract(["common_name", "latin_name"])
assert fresh_db["tree"].schema == (
'CREATE TABLE "tree" (\n'
" [rowid] INTEGER PRIMARY KEY,\n"
" [name] TEXT,\n"
" [common_name_latin_name_id] INTEGER,\n"
" FOREIGN KEY([common_name_latin_name_id]) REFERENCES [common_name_latin_name]([id])\n"
")"
)
assert (
fresh_db.execute(
"""
select
tree.name,
common_name_latin_name.common_name,
common_name_latin_name.latin_name
from tree
join common_name_latin_name
on tree.common_name_latin_name_id = common_name_latin_name.id
"""
).fetchall()
== [("Tree 1", "Palm", "Arecaceae")]
)
def test_reuse_lookup_table(fresh_db):

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):

9
tests/test_insert_files.py
View file

@ -2,9 +2,11 @@ from sqlite_utils import cli, Database
from click.testing import CliRunner
import os
import pathlib
import pytest
def test_insert_files():
@pytest.mark.parametrize("silent", (False, True))
def test_insert_files(silent):
runner = CliRunner()
with runner.isolated_filesystem():
tmpdir = pathlib.Path(".")
@ -34,7 +36,10 @@ def test_insert_files():
cols += ["-c", "{}:{}".format(coltype, coltype)]
result = runner.invoke(
cli.cli,
["insert-files", db_path, "files", str(tmpdir)] + cols + ["--pk", "path"],
["insert-files", db_path, "files", str(tmpdir)]
+ cols
+ ["--pk", "path"]
+ (["--silent"] if silent else []),
catch_exceptions=False,
)
assert result.exit_code == 0, result.stdout

74
tests/test_introspect.py
View file

@ -1,4 +1,4 @@
from sqlite_utils.db import Index, View, Database
from sqlite_utils.db import Index, View, Database, XIndex, XIndexColumn
import pytest
@ -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):
@ -52,7 +52,14 @@ def test_views(fresh_db):
def test_count(existing_db):
assert 3 == existing_db["foo"].count
assert existing_db["foo"].count == 3
assert existing_db["foo"].count_where() == 3
assert existing_db["foo"].execute_count() == 3
def test_count_where(existing_db):
assert existing_db["foo"].count_where("text != ?", ["two"]) == 2
assert existing_db["foo"].count_where("text != :t", {"t": "two"}) == 2
def test_columns(existing_db):
@ -62,8 +69,12 @@ def test_columns(existing_db):
]
def test_schema(existing_db):
assert "CREATE TABLE foo (text TEXT)" == existing_db["foo"].schema
def test_table_schema(existing_db):
assert existing_db["foo"].schema == "CREATE TABLE foo (text TEXT)"
def test_database_schema(existing_db):
assert existing_db.schema == "CREATE TABLE foo (text TEXT);"
def test_table_repr(fresh_db):
@ -93,6 +104,33 @@ def test_indexes(fresh_db):
] == fresh_db["Gosh"].indexes
def test_xindexes(fresh_db):
fresh_db.executescript(
"""
create table Gosh (c1 text, c2 text, c3 text);
create index Gosh_c1 on Gosh(c1);
create index Gosh_c2c3 on Gosh(c2, c3 desc);
"""
)
assert fresh_db["Gosh"].xindexes == [
XIndex(
name="Gosh_c2c3",
columns=[
XIndexColumn(seqno=0, cid=1, name="c2", desc=0, coll="BINARY", key=1),
XIndexColumn(seqno=1, cid=2, name="c3", desc=1, coll="BINARY", key=1),
XIndexColumn(seqno=2, cid=-1, name=None, desc=0, coll="BINARY", key=0),
],
),
XIndex(
name="Gosh_c1",
columns=[
XIndexColumn(seqno=0, cid=0, name="c1", desc=0, coll="BINARY", key=1),
XIndexColumn(seqno=1, cid=-1, name=None, desc=0, coll="BINARY", key=0),
],
),
]
@pytest.mark.parametrize(
"column,expected_table_guess",
(
@ -144,9 +182,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 == []
@ -206,3 +256,11 @@ def test_virtual_table_using(sql, expected_name, expected_using):
db = Database(memory=True)
db.execute(sql)
assert db[expected_name].virtual_table_using == expected_using
def test_use_rowid():
db = Database(memory=True)
db["rowid_table"].insert({"name": "Cleo"})
db["regular_table"].insert({"id": 1, "name": "Cleo"}, pk="id")
assert db["rowid_table"].use_rowid
assert not db["regular_table"].use_rowid

17
tests/test_query.py Normal file
View file

@ -0,0 +1,17 @@
import types
def test_query(fresh_db):
fresh_db["dogs"].insert_all([{"name": "Cleo"}, {"name": "Pancakes"}])
results = fresh_db.query("select * from dogs order by name desc")
assert isinstance(results, types.GeneratorType)
assert list(results) == [{"name": "Pancakes"}, {"name": "Cleo"}]
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
fresh_db["test"].insert({"id": 1, "bar": 2}, pk="id")
assert fresh_db.execute_returning_dicts("select * from test") == [
{"id": 1, "bar": 2}
]

108
tests/test_recipes.py Normal file
View file

@ -0,0 +1,108 @@
from sqlite_utils import recipes
import json
import pytest
@pytest.fixture
def dates_db(fresh_db):
fresh_db["example"].insert_all(
[
{"id": 1, "dt": "5th October 2019 12:04"},
{"id": 2, "dt": "6th October 2019 00:05:06"},
{"id": 3, "dt": ""},
{"id": 4, "dt": None},
],
pk="id",
)
return fresh_db
def test_parsedate(dates_db):
dates_db["example"].convert("dt", recipes.parsedate)
assert list(dates_db["example"].rows) == [
{"id": 1, "dt": "2019-10-05"},
{"id": 2, "dt": "2019-10-06"},
{"id": 3, "dt": ""},
{"id": 4, "dt": None},
]
def test_parsedatetime(dates_db):
dates_db["example"].convert("dt", recipes.parsedatetime)
assert list(dates_db["example"].rows) == [
{"id": 1, "dt": "2019-10-05T12:04:00"},
{"id": 2, "dt": "2019-10-06T00:05:06"},
{"id": 3, "dt": ""},
{"id": 4, "dt": None},
]
@pytest.mark.parametrize(
"recipe,kwargs,expected",
(
("parsedate", {}, "2005-03-04"),
("parsedate", {"dayfirst": True}, "2005-04-03"),
("parsedatetime", {}, "2005-03-04T00:00:00"),
("parsedatetime", {"dayfirst": True}, "2005-04-03T00:00:00"),
),
)
def test_dayfirst_yearfirst(fresh_db, recipe, kwargs, expected):
fresh_db["example"].insert_all(
[
{"id": 1, "dt": "03/04/05"},
],
pk="id",
)
fresh_db["example"].convert(
"dt", lambda value: getattr(recipes, recipe)(value, **kwargs)
)
assert list(fresh_db["example"].rows) == [
{"id": 1, "dt": expected},
]
@pytest.mark.parametrize("delimiter", [None, ";", "-"])
def test_jsonsplit(fresh_db, delimiter):
fresh_db["example"].insert_all(
[
{"id": 1, "tags": (delimiter or ",").join(["foo", "bar"])},
{"id": 2, "tags": (delimiter or ",").join(["bar", "baz"])},
],
pk="id",
)
fn = recipes.jsonsplit
if delimiter is not None:
def fn(value):
return recipes.jsonsplit(value, delimiter=delimiter)
fresh_db["example"].convert("tags", fn)
assert list(fresh_db["example"].rows) == [
{"id": 1, "tags": '["foo", "bar"]'},
{"id": 2, "tags": '["bar", "baz"]'},
]
@pytest.mark.parametrize(
"type,expected",
(
(None, ["1", "2", "3"]),
(float, [1.0, 2.0, 3.0]),
(int, [1, 2, 3]),
),
)
def test_jsonsplit_type(fresh_db, type, expected):
fresh_db["example"].insert_all(
[
{"id": 1, "records": "1,2,3"},
],
pk="id",
)
fn = recipes.jsonsplit
if type is not None:
def fn(value):
return recipes.jsonsplit(value, type=type)
fresh_db["example"].convert("records", fn)
assert json.loads(fresh_db["example"].get(1)["records"]) == expected

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(

5
tests/test_register_function.py
View file

@ -1,3 +1,4 @@
# flake8: noqa
import pytest
import sys
from unittest.mock import MagicMock
@ -55,14 +56,14 @@ def test_register_function_replace(fresh_db):
# This will fail to replace the function:
@fresh_db.register_function()
def one():
def one(): # noqa
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
return "two"
assert "two" == fresh_db.execute("select one()").fetchone()[0]

2
tests/test_rows.py
View file

@ -1,4 +1,3 @@
from sqlite_utils.db import Index, View
import pytest
@ -13,6 +12,7 @@ def test_rows(existing_db):
[
("name = ?", ["Pancakes"], {2}),
("age > ?", [3], {1}),
("age > :age", {"age": 3}, {1}),
("name is not null", [], {1, 2}),
("is_good = ?", [True], {1, 2}),
],

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"},
),
]

84
tests/test_transform.py
View file

@ -89,9 +89,14 @@ import pytest
],
)
@pytest.mark.parametrize("use_pragma_foreign_keys", [False, True])
def test_transform_sql(fresh_db, params, expected_sql, use_pragma_foreign_keys):
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")
@ -111,7 +116,80 @@ def test_transform_sql(fresh_db, params, expected_sql, use_pragma_foreign_keys):
assert ("PRAGMA foreign_keys=1;", None) not in captured
def test_transform_sql_rowid_to_id(fresh_db):
@pytest.mark.parametrize(
"params,expected_sql",
[
# Identity transform - nothing changes
(
{},
[
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER,\n [name] TEXT,\n [age] TEXT\n);",
"INSERT INTO [dogs_new_suffix] ([id], [name], [age])\n SELECT [id], [name], [age] FROM [dogs];",
"DROP TABLE [dogs];",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];",
],
),
# Change column type
(
{"types": {"age": int}},
[
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER,\n [name] TEXT,\n [age] INTEGER\n);",
"INSERT INTO [dogs_new_suffix] ([id], [name], [age])\n SELECT [id], [name], [age] FROM [dogs];",
"DROP TABLE [dogs];",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];",
],
),
# Rename a column
(
{"rename": {"age": "dog_age"}},
[
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER,\n [name] TEXT,\n [dog_age] TEXT\n);",
"INSERT INTO [dogs_new_suffix] ([id], [name], [dog_age])\n SELECT [id], [name], [age] FROM [dogs];",
"DROP TABLE [dogs];",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];",
],
),
# Make ID a primary key
(
{"pk": "id"},
[
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [age] TEXT\n);",
"INSERT INTO [dogs_new_suffix] ([id], [name], [age])\n SELECT [id], [name], [age] FROM [dogs];",
"DROP TABLE [dogs];",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];",
],
),
],
)
@pytest.mark.parametrize("use_pragma_foreign_keys", [False, True])
def test_transform_sql_table_with_no_primary_key(
fresh_db, params, expected_sql, use_pragma_foreign_keys
):
captured = []
def tracer(sql, params):
return captured.append((sql, params))
dogs = fresh_db["dogs"]
if use_pragma_foreign_keys:
fresh_db.conn.execute("PRAGMA foreign_keys=ON")
dogs.insert({"id": 1, "name": "Cleo", "age": "5"})
sql = dogs.transform_sql(**{**params, **{"tmp_suffix": "suffix"}})
assert sql == expected_sql
# Check that .transform() runs without exceptions:
with fresh_db.tracer(tracer):
dogs.transform(**params)
# If use_pragma_foreign_keys, check that we did the right thing
if use_pragma_foreign_keys:
assert ("PRAGMA foreign_keys=0;", None) in captured
assert captured[-2] == ("PRAGMA foreign_key_check;", None)
assert captured[-1] == ("PRAGMA foreign_keys=1;", None)
else:
assert ("PRAGMA foreign_keys=0;", None) not in captured
assert ("PRAGMA foreign_keys=1;", None) not in captured
def test_transform_sql_with_no_primary_key_to_primary_key_of_id(fresh_db):
dogs = fresh_db["dogs"]
dogs.insert({"id": 1, "name": "Cleo", "age": "5"})
assert (

9
tests/test_upsert.py
View file

@ -21,6 +21,13 @@ def test_upsert_all(fresh_db):
assert table.last_pk is None
def test_upsert_all_single_column(fresh_db):
table = fresh_db["table"]
table.upsert_all([{"name": "Cleo"}], pk="name")
assert [{"name": "Cleo"}] == list(table.rows)
assert table.pks == ["name"]
def test_upsert_error_if_no_pk(fresh_db):
table = fresh_db["table"]
with pytest.raises(PrimaryKeyRequired):
@ -47,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