mirror of
https://github.com/simonw/sqlite-utils.git
synced 2026-07-23 01:14:31 +02:00
c710ade644
Also made it so .duplicate() raises new NoTable exception rather than raising an AssertionError if the source table does not exist.
2154 lines
66 KiB
Python
2154 lines
66 KiB
Python
from sqlite_utils import cli, Database
|
|
from sqlite_utils.db import Index, ForeignKey
|
|
from click.testing import CliRunner
|
|
import subprocess
|
|
import sys
|
|
from unittest import mock
|
|
import json
|
|
import os
|
|
import pytest
|
|
import textwrap
|
|
|
|
from .utils import collapse_whitespace
|
|
|
|
|
|
@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()
|
|
|
|
|
|
def test_views(db_path):
|
|
Database(db_path).create_view("hello", "select sqlite_version()")
|
|
result = CliRunner().invoke(cli.cli, ["views", db_path, "--table", "--schema"])
|
|
assert (
|
|
"view schema\n"
|
|
"------ --------------------------------------------\n"
|
|
"hello CREATE VIEW hello AS select sqlite_version()"
|
|
) == result.output.strip()
|
|
|
|
|
|
def test_tables_fts4(db_path):
|
|
Database(db_path)["Gosh"].enable_fts(["c2"], fts_version="FTS4")
|
|
result = CliRunner().invoke(cli.cli, ["tables", "--fts4", db_path])
|
|
assert '[{"table": "Gosh_fts"}]' == result.output.strip()
|
|
|
|
|
|
def test_tables_fts5(db_path):
|
|
Database(db_path)["Gosh"].enable_fts(["c2"], fts_version="FTS5")
|
|
result = CliRunner().invoke(cli.cli, ["tables", "--fts5", db_path])
|
|
assert '[{"table": "Gosh_fts"}]' == result.output.strip()
|
|
|
|
|
|
def test_tables_counts_and_columns(db_path):
|
|
db = Database(db_path)
|
|
with db.conn:
|
|
db["lots"].insert_all([{"id": i, "age": i + 1} for i in range(30)])
|
|
result = CliRunner().invoke(cli.cli, ["tables", "--counts", "--columns", db_path])
|
|
assert (
|
|
'[{"table": "Gosh", "count": 0, "columns": ["c1", "c2", "c3"]},\n'
|
|
' {"table": "Gosh2", "count": 0, "columns": ["c1", "c2", "c3"]},\n'
|
|
' {"table": "lots", "count": 30, "columns": ["id", "age"]}]'
|
|
) == result.output.strip()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"format,expected",
|
|
[
|
|
(
|
|
"--csv",
|
|
(
|
|
"table,count,columns\n"
|
|
'Gosh,0,"c1\n'
|
|
"c2\n"
|
|
'c3"\n'
|
|
'Gosh2,0,"c1\n'
|
|
"c2\n"
|
|
'c3"\n'
|
|
'lots,30,"id\n'
|
|
'age"'
|
|
),
|
|
),
|
|
(
|
|
"--tsv",
|
|
"table\tcount\tcolumns\nGosh\t0\t['c1', 'c2', 'c3']\nGosh2\t0\t['c1', 'c2', 'c3']\nlots\t30\t['id', 'age']",
|
|
),
|
|
],
|
|
)
|
|
def test_tables_counts_and_columns_csv(db_path, format, expected):
|
|
db = Database(db_path)
|
|
with db.conn:
|
|
db["lots"].insert_all([{"id": i, "age": i + 1} for i in range(30)])
|
|
result = CliRunner().invoke(
|
|
cli.cli, ["tables", "--counts", "--columns", format, db_path]
|
|
)
|
|
assert result.output.strip().replace("\r", "") == expected
|
|
|
|
|
|
def test_tables_schema(db_path):
|
|
db = Database(db_path)
|
|
with db.conn:
|
|
db["lots"].insert_all([{"id": i, "age": i + 1} for i in range(30)])
|
|
result = CliRunner().invoke(cli.cli, ["tables", "--schema", db_path])
|
|
assert (
|
|
'[{"table": "Gosh", "schema": "CREATE TABLE Gosh (c1 text, c2 text, c3 text)"},\n'
|
|
' {"table": "Gosh2", "schema": "CREATE TABLE Gosh2 (c1 text, c2 text, c3 text)"},\n'
|
|
' {"table": "lots", "schema": "CREATE TABLE [lots] (\\n [id] INTEGER,\\n [age] INTEGER\\n)"}]'
|
|
) == result.output.strip()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"options,expected",
|
|
[
|
|
(
|
|
["--fmt", "simple"],
|
|
(
|
|
"c1 c2 c3\n"
|
|
"----- ----- ----------\n"
|
|
"verb0 noun0 adjective0\n"
|
|
"verb1 noun1 adjective1\n"
|
|
"verb2 noun2 adjective2\n"
|
|
"verb3 noun3 adjective3"
|
|
),
|
|
),
|
|
(
|
|
["-t"],
|
|
(
|
|
"c1 c2 c3\n"
|
|
"----- ----- ----------\n"
|
|
"verb0 noun0 adjective0\n"
|
|
"verb1 noun1 adjective1\n"
|
|
"verb2 noun2 adjective2\n"
|
|
"verb3 noun3 adjective3"
|
|
),
|
|
),
|
|
(
|
|
["--fmt", "rst"],
|
|
(
|
|
"===== ===== ==========\n"
|
|
"c1 c2 c3\n"
|
|
"===== ===== ==========\n"
|
|
"verb0 noun0 adjective0\n"
|
|
"verb1 noun1 adjective1\n"
|
|
"verb2 noun2 adjective2\n"
|
|
"verb3 noun3 adjective3\n"
|
|
"===== ===== =========="
|
|
),
|
|
),
|
|
],
|
|
)
|
|
def test_output_table(db_path, options, expected):
|
|
db = Database(db_path)
|
|
with db.conn:
|
|
db["rows"].insert_all(
|
|
[
|
|
{
|
|
"c1": "verb{}".format(i),
|
|
"c2": "noun{}".format(i),
|
|
"c3": "adjective{}".format(i),
|
|
}
|
|
for i in range(4)
|
|
]
|
|
)
|
|
result = CliRunner().invoke(cli.cli, ["rows", db_path, "rows"] + options)
|
|
assert 0 == result.exit_code
|
|
assert expected == result.output.strip()
|
|
|
|
|
|
def test_create_index(db_path):
|
|
db = Database(db_path)
|
|
assert [] == db["Gosh"].indexes
|
|
result = CliRunner().invoke(cli.cli, ["create-index", db_path, "Gosh", "c1"])
|
|
assert 0 == result.exit_code
|
|
assert [
|
|
Index(
|
|
seq=0, name="idx_Gosh_c1", unique=0, origin="c", partial=0, columns=["c1"]
|
|
)
|
|
] == db["Gosh"].indexes
|
|
# Try with a custom name
|
|
result = CliRunner().invoke(
|
|
cli.cli, ["create-index", db_path, "Gosh", "c2", "--name", "blah"]
|
|
)
|
|
assert 0 == result.exit_code
|
|
assert [
|
|
Index(seq=0, name="blah", unique=0, origin="c", partial=0, columns=["c2"]),
|
|
Index(
|
|
seq=1, name="idx_Gosh_c1", unique=0, origin="c", partial=0, columns=["c1"]
|
|
),
|
|
] == db["Gosh"].indexes
|
|
# Try a two-column unique index
|
|
create_index_unique_args = [
|
|
"create-index",
|
|
db_path,
|
|
"Gosh2",
|
|
"c1",
|
|
"c2",
|
|
"--unique",
|
|
]
|
|
result = CliRunner().invoke(cli.cli, create_index_unique_args)
|
|
assert 0 == result.exit_code
|
|
assert [
|
|
Index(
|
|
seq=0,
|
|
name="idx_Gosh2_c1_c2",
|
|
unique=1,
|
|
origin="c",
|
|
partial=0,
|
|
columns=["c1", "c2"],
|
|
)
|
|
] == db["Gosh2"].indexes
|
|
# Trying to create the same index should fail
|
|
assert 0 != CliRunner().invoke(cli.cli, create_index_unique_args).exit_code
|
|
# ... unless we use --if-not-exists
|
|
assert (
|
|
0
|
|
== CliRunner()
|
|
.invoke(cli.cli, create_index_unique_args + ["--if-not-exists"])
|
|
.exit_code
|
|
)
|
|
|
|
|
|
def test_create_index_analyze(db_path):
|
|
db = Database(db_path)
|
|
assert "sqlite_stat1" not in db.table_names()
|
|
assert [] == db["Gosh"].indexes
|
|
result = CliRunner().invoke(
|
|
cli.cli, ["create-index", db_path, "Gosh", "c1", "--analyze"]
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "sqlite_stat1" in db.table_names()
|
|
|
|
|
|
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",
|
|
(
|
|
("text", "TEXT", "CREATE TABLE [dogs] ( [name] TEXT , [text] TEXT)"),
|
|
(
|
|
"integer",
|
|
"INTEGER",
|
|
"CREATE TABLE [dogs] ( [name] TEXT , [integer] INTEGER)",
|
|
),
|
|
("float", "FLOAT", "CREATE TABLE [dogs] ( [name] TEXT , [float] FLOAT)"),
|
|
("blob", "blob", "CREATE TABLE [dogs] ( [name] TEXT , [blob] BLOB)"),
|
|
("default", None, "CREATE TABLE [dogs] ( [name] TEXT , [default] TEXT)"),
|
|
),
|
|
)
|
|
def test_add_column(db_path, col_name, col_type, expected_schema):
|
|
db = Database(db_path)
|
|
db.create_table("dogs", {"name": str})
|
|
assert "CREATE TABLE [dogs] ( [name] TEXT )" == collapse_whitespace(
|
|
db["dogs"].schema
|
|
)
|
|
args = ["add-column", db_path, "dogs", col_name]
|
|
if col_type is not None:
|
|
args.append(col_type)
|
|
assert 0 == CliRunner().invoke(cli.cli, args).exit_code
|
|
assert expected_schema == collapse_whitespace(db["dogs"].schema)
|
|
|
|
|
|
def test_add_column_not_null_default(db_path):
|
|
db = Database(db_path)
|
|
db.create_table("dogs", {"name": str})
|
|
assert "CREATE TABLE [dogs] ( [name] TEXT )" == collapse_whitespace(
|
|
db["dogs"].schema
|
|
)
|
|
args = [
|
|
"add-column",
|
|
db_path,
|
|
"dogs",
|
|
"nickname",
|
|
"--not-null-default",
|
|
"dogs'dawg",
|
|
]
|
|
assert 0 == CliRunner().invoke(cli.cli, args).exit_code
|
|
assert (
|
|
"CREATE TABLE [dogs] ( [name] TEXT , [nickname] TEXT NOT NULL DEFAULT 'dogs''dawg')"
|
|
== collapse_whitespace(db["dogs"].schema)
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"args,assert_message",
|
|
(
|
|
(
|
|
["books", "author_id", "authors", "id"],
|
|
"Explicit other_table and other_column",
|
|
),
|
|
(["books", "author_id", "authors"], "Explicit other_table, guess other_column"),
|
|
(["books", "author_id"], "Automatically guess other_table and other_column"),
|
|
),
|
|
)
|
|
def test_add_foreign_key(db_path, args, assert_message):
|
|
db = Database(db_path)
|
|
db["authors"].insert_all(
|
|
[{"id": 1, "name": "Sally"}, {"id": 2, "name": "Asheesh"}], pk="id"
|
|
)
|
|
db["books"].insert_all(
|
|
[
|
|
{"title": "Hedgehogs of the world", "author_id": 1},
|
|
{"title": "How to train your wolf", "author_id": 2},
|
|
]
|
|
)
|
|
assert (
|
|
0 == CliRunner().invoke(cli.cli, ["add-foreign-key", db_path] + args).exit_code
|
|
), assert_message
|
|
assert [
|
|
ForeignKey(
|
|
table="books", column="author_id", other_table="authors", other_column="id"
|
|
)
|
|
] == db["books"].foreign_keys
|
|
|
|
# Error if we try to add it twice:
|
|
result = CliRunner().invoke(
|
|
cli.cli, ["add-foreign-key", db_path, "books", "author_id", "authors", "id"]
|
|
)
|
|
assert 0 != result.exit_code
|
|
assert (
|
|
"Error: Foreign key already exists for author_id => authors.id"
|
|
== result.output.strip()
|
|
)
|
|
|
|
# No error if we add it twice with --ignore
|
|
result = CliRunner().invoke(
|
|
cli.cli,
|
|
["add-foreign-key", db_path, "books", "author_id", "authors", "id", "--ignore"],
|
|
)
|
|
assert 0 == result.exit_code
|
|
|
|
# Error if we try against an invalid column
|
|
result = CliRunner().invoke(
|
|
cli.cli, ["add-foreign-key", db_path, "books", "author_id", "authors", "bad"]
|
|
)
|
|
assert 0 != result.exit_code
|
|
assert "Error: No such column: authors.bad" == result.output.strip()
|
|
|
|
|
|
def test_add_column_foreign_key(db_path):
|
|
db = Database(db_path)
|
|
db["authors"].insert({"id": 1, "name": "Sally"}, pk="id")
|
|
db["books"].insert({"title": "Hedgehogs of the world"})
|
|
# Add an author_id foreign key column to the books table
|
|
result = CliRunner().invoke(
|
|
cli.cli, ["add-column", db_path, "books", "author_id", "--fk", "authors"]
|
|
)
|
|
assert 0 == result.exit_code, result.output
|
|
assert (
|
|
"CREATE TABLE [books] ( [title] TEXT , [author_id] INTEGER, FOREIGN KEY([author_id]) REFERENCES [authors]([id]) )"
|
|
== collapse_whitespace(db["books"].schema)
|
|
)
|
|
# Try it again with a custom --fk-col
|
|
result = CliRunner().invoke(
|
|
cli.cli,
|
|
[
|
|
"add-column",
|
|
db_path,
|
|
"books",
|
|
"author_name_ref",
|
|
"--fk",
|
|
"authors",
|
|
"--fk-col",
|
|
"name",
|
|
],
|
|
)
|
|
assert 0 == result.exit_code, result.output
|
|
assert (
|
|
"CREATE TABLE [books] ( [title] TEXT , [author_id] INTEGER, [author_name_ref] TEXT, "
|
|
"FOREIGN KEY([author_id]) REFERENCES [authors]([id]), "
|
|
"FOREIGN KEY([author_name_ref]) REFERENCES [authors]([name]) )"
|
|
== collapse_whitespace(db["books"].schema)
|
|
)
|
|
# Throw an error if the --fk table does not exist
|
|
result = CliRunner().invoke(
|
|
cli.cli, ["add-column", db_path, "books", "author_id", "--fk", "bobcats"]
|
|
)
|
|
assert 0 != result.exit_code
|
|
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)
|
|
assert [] == db["books"].indexes
|
|
result = CliRunner().invoke(cli.cli, ["index-foreign-keys", db_path])
|
|
assert 0 == result.exit_code
|
|
assert [["author_id"], ["author_name_ref"]] == [
|
|
i.columns for i in db["books"].indexes
|
|
]
|
|
|
|
|
|
def test_enable_fts(db_path):
|
|
db = Database(db_path)
|
|
assert db["Gosh"].detect_fts() is None
|
|
result = CliRunner().invoke(
|
|
cli.cli, ["enable-fts", db_path, "Gosh", "c1", "--fts4"]
|
|
)
|
|
assert 0 == result.exit_code
|
|
assert "Gosh_fts" == db["Gosh"].detect_fts()
|
|
|
|
# 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 db["http://example.com"].detect_fts() is None
|
|
result = CliRunner().invoke(
|
|
cli.cli,
|
|
[
|
|
"enable-fts",
|
|
db_path,
|
|
"http://example.com",
|
|
"c1",
|
|
"--fts4",
|
|
"--tokenize",
|
|
"porter",
|
|
],
|
|
)
|
|
assert 0 == result.exit_code
|
|
assert "http://example.com_fts" == db["http://example.com"].detect_fts()
|
|
# Check tokenize was set to porter
|
|
assert (
|
|
"CREATE VIRTUAL TABLE [http://example.com_fts] USING FTS4 (\n"
|
|
" [c1],\n"
|
|
" tokenize='porter',\n"
|
|
" content=[http://example.com]"
|
|
"\n)"
|
|
) == db["http://example.com_fts"].schema
|
|
db["http://example.com"].drop()
|
|
|
|
|
|
def test_enable_fts_with_triggers(db_path):
|
|
Database(db_path)["Gosh"].insert_all([{"c1": "baz"}])
|
|
exit_code = (
|
|
CliRunner()
|
|
.invoke(
|
|
cli.cli,
|
|
["enable-fts", db_path, "Gosh", "c1", "--fts4", "--create-triggers"],
|
|
)
|
|
.exit_code
|
|
)
|
|
assert 0 == exit_code
|
|
|
|
def search(q):
|
|
return (
|
|
Database(db_path)
|
|
.execute("select c1 from Gosh_fts where c1 match ?", [q])
|
|
.fetchall()
|
|
)
|
|
|
|
assert [("baz",)] == search("baz")
|
|
Database(db_path)["Gosh"].insert_all([{"c1": "martha"}])
|
|
assert [("martha",)] == search("martha")
|
|
|
|
|
|
def test_populate_fts(db_path):
|
|
Database(db_path)["Gosh"].insert_all([{"c1": "baz"}])
|
|
exit_code = (
|
|
CliRunner()
|
|
.invoke(cli.cli, ["enable-fts", db_path, "Gosh", "c1", "--fts4"])
|
|
.exit_code
|
|
)
|
|
assert 0 == exit_code
|
|
|
|
def search(q):
|
|
return (
|
|
Database(db_path)
|
|
.execute("select c1 from Gosh_fts where c1 match ?", [q])
|
|
.fetchall()
|
|
)
|
|
|
|
assert [("baz",)] == search("baz")
|
|
Database(db_path)["Gosh"].insert_all([{"c1": "martha"}])
|
|
assert [] == search("martha")
|
|
exit_code = (
|
|
CliRunner().invoke(cli.cli, ["populate-fts", db_path, "Gosh", "c1"]).exit_code
|
|
)
|
|
assert 0 == exit_code
|
|
assert [("martha",)] == search("martha")
|
|
|
|
|
|
def test_disable_fts(db_path):
|
|
db = Database(db_path)
|
|
assert {"Gosh", "Gosh2"} == set(db.table_names())
|
|
db["Gosh"].enable_fts(["c1"], create_triggers=True)
|
|
assert {
|
|
"Gosh_fts",
|
|
"Gosh_fts_idx",
|
|
"Gosh_fts_data",
|
|
"Gosh2",
|
|
"Gosh_fts_config",
|
|
"Gosh",
|
|
"Gosh_fts_docsize",
|
|
} == set(db.table_names())
|
|
exit_code = CliRunner().invoke(cli.cli, ["disable-fts", db_path, "Gosh"]).exit_code
|
|
assert 0 == exit_code
|
|
assert {"Gosh", "Gosh2"} == set(db.table_names())
|
|
|
|
|
|
def test_vacuum(db_path):
|
|
result = CliRunner().invoke(cli.cli, ["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)
|
|
with db.conn:
|
|
for table in ("Gosh", "Gosh2"):
|
|
db[table].insert_all(
|
|
[
|
|
{
|
|
"c1": "verb{}".format(i),
|
|
"c2": "noun{}".format(i),
|
|
"c3": "adjective{}".format(i),
|
|
}
|
|
for i in range(10000)
|
|
]
|
|
)
|
|
db["Gosh"].enable_fts(["c1", "c2", "c3"], fts_version="FTS4")
|
|
db["Gosh2"].enable_fts(["c1", "c2", "c3"], fts_version="FTS5")
|
|
size_before_optimize = os.stat(db_path).st_size
|
|
result = CliRunner().invoke(cli.cli, ["optimize", db_path] + tables)
|
|
assert 0 == result.exit_code
|
|
size_after_optimize = os.stat(db_path).st_size
|
|
# Weirdest thing: tests started failing because size after
|
|
# ended up larger than size before in some cases. I think
|
|
# it's OK to tolerate that happening, though it's very strange.
|
|
assert size_after_optimize <= (size_before_optimize + 10000)
|
|
# Soundness check that --no-vacuum doesn't throw errors:
|
|
result = CliRunner().invoke(cli.cli, ["optimize", "--no-vacuum", db_path])
|
|
assert 0 == result.exit_code
|
|
|
|
|
|
def test_rebuild_fts_fixes_docsize_error(db_path):
|
|
db = Database(db_path, recursive_triggers=False)
|
|
records = [
|
|
{
|
|
"c1": "verb{}".format(i),
|
|
"c2": "noun{}".format(i),
|
|
"c3": "adjective{}".format(i),
|
|
}
|
|
for i in range(10000)
|
|
]
|
|
with db.conn:
|
|
db["fts5_table"].insert_all(records, pk="c1")
|
|
db["fts5_table"].enable_fts(
|
|
["c1", "c2", "c3"], fts_version="FTS5", create_triggers=True
|
|
)
|
|
# Search should work
|
|
assert list(db["fts5_table"].search("verb1"))
|
|
# Replicate docsize error from this issue for FTS5
|
|
# https://github.com/simonw/sqlite-utils/issues/149
|
|
assert db["fts5_table_fts_docsize"].count == 10000
|
|
db["fts5_table"].insert_all(records, replace=True)
|
|
assert db["fts5_table"].count == 10000
|
|
assert db["fts5_table_fts_docsize"].count == 20000
|
|
# Running rebuild-fts should fix this
|
|
result = CliRunner().invoke(cli.cli, ["rebuild-fts", db_path, "fts5_table"])
|
|
assert 0 == result.exit_code
|
|
assert db["fts5_table_fts_docsize"].count == 10000
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"format,expected",
|
|
[
|
|
("--csv", "id,name,age\n1,Cleo,4\n2,Pancakes,2\n"),
|
|
("--tsv", "id\tname\tage\n1\tCleo\t4\n2\tPancakes\t2\n"),
|
|
],
|
|
)
|
|
def test_query_csv(db_path, format, expected):
|
|
db = Database(db_path)
|
|
with db.conn:
|
|
db["dogs"].insert_all(
|
|
[
|
|
{"id": 1, "age": 4, "name": "Cleo"},
|
|
{"id": 2, "age": 2, "name": "Pancakes"},
|
|
]
|
|
)
|
|
result = CliRunner().invoke(
|
|
cli.cli, [db_path, "select id, name, age from dogs", format]
|
|
)
|
|
assert 0 == result.exit_code
|
|
assert result.output.replace("\r", "") == expected
|
|
# Test the no-headers option:
|
|
result = CliRunner().invoke(
|
|
cli.cli, [db_path, "select id, name, age from dogs", "--no-headers", format]
|
|
)
|
|
expected_rest = "\n".join(expected.split("\n")[1:]).strip()
|
|
assert result.output.strip().replace("\r", "") == expected_rest
|
|
|
|
|
|
_all_query = "select id, name, age from dogs"
|
|
_one_query = "select id, name, age from dogs where id = 1"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"sql,args,expected",
|
|
[
|
|
(
|
|
_all_query,
|
|
[],
|
|
'[{"id": 1, "name": "Cleo", "age": 4},\n {"id": 2, "name": "Pancakes", "age": 2}]',
|
|
),
|
|
(
|
|
_all_query,
|
|
["--nl"],
|
|
'{"id": 1, "name": "Cleo", "age": 4}\n{"id": 2, "name": "Pancakes", "age": 2}',
|
|
),
|
|
(_all_query, ["--arrays"], '[[1, "Cleo", 4],\n [2, "Pancakes", 2]]'),
|
|
(_all_query, ["--arrays", "--nl"], '[1, "Cleo", 4]\n[2, "Pancakes", 2]'),
|
|
(_one_query, [], '[{"id": 1, "name": "Cleo", "age": 4}]'),
|
|
(_one_query, ["--nl"], '{"id": 1, "name": "Cleo", "age": 4}'),
|
|
(_one_query, ["--arrays"], '[[1, "Cleo", 4]]'),
|
|
(_one_query, ["--arrays", "--nl"], '[1, "Cleo", 4]'),
|
|
],
|
|
)
|
|
def test_query_json(db_path, sql, args, expected):
|
|
db = Database(db_path)
|
|
with db.conn:
|
|
db["dogs"].insert_all(
|
|
[
|
|
{"id": 1, "age": 4, "name": "Cleo"},
|
|
{"id": 2, "age": 2, "name": "Pancakes"},
|
|
]
|
|
)
|
|
result = CliRunner().invoke(cli.cli, [db_path, sql] + args)
|
|
assert expected == result.output.strip()
|
|
|
|
|
|
def test_query_json_empty(db_path):
|
|
result = CliRunner().invoke(
|
|
cli.cli, [db_path, "select * from sqlite_master where 0"]
|
|
)
|
|
assert 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\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):
|
|
db = Database(db_path)
|
|
with db.conn:
|
|
db["files"].insert(
|
|
{
|
|
"name": "lorem.txt",
|
|
"sz": 16984,
|
|
"data": LOREM_IPSUM_COMPRESSED,
|
|
},
|
|
pk="name",
|
|
)
|
|
result = CliRunner().invoke(cli.cli, [db_path, "select name, sz, data from files"])
|
|
assert result.exit_code == 0, str(result)
|
|
assert json.loads(result.output.strip()) == [
|
|
{
|
|
"name": "lorem.txt",
|
|
"sz": 16984,
|
|
"data": {
|
|
"$base64": True,
|
|
"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"
|
|
)
|
|
),
|
|
},
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"sql,params,expected",
|
|
[
|
|
("select 1 + 1 as out", {"p": "2"}, 2),
|
|
("select 1 + :p as out", {"p": "2"}, 3),
|
|
(
|
|
"select :hello as out",
|
|
{"hello": """This"has'many'quote"s"""},
|
|
"""This"has'many'quote"s""",
|
|
),
|
|
],
|
|
)
|
|
def test_query_params(db_path, sql, params, expected):
|
|
extra_args = []
|
|
for key, value in params.items():
|
|
extra_args.extend(["-p", key, value])
|
|
result = CliRunner().invoke(cli.cli, [db_path, sql] + extra_args)
|
|
assert result.exit_code == 0, str(result)
|
|
assert json.loads(result.output.strip()) == [{"out": expected}]
|
|
|
|
|
|
def test_query_json_with_json_cols(db_path):
|
|
db = Database(db_path)
|
|
with db.conn:
|
|
db["dogs"].insert(
|
|
{
|
|
"id": 1,
|
|
"name": "Cleo",
|
|
"friends": [{"name": "Pancakes"}, {"name": "Bailey"}],
|
|
}
|
|
)
|
|
result = CliRunner().invoke(
|
|
cli.cli, [db_path, "select id, name, friends from dogs"]
|
|
)
|
|
assert (
|
|
r"""
|
|
[{"id": 1, "name": "Cleo", "friends": "[{\"name\": \"Pancakes\"}, {\"name\": \"Bailey\"}]"}]
|
|
""".strip()
|
|
== result.output.strip()
|
|
)
|
|
# With --json-cols:
|
|
result = CliRunner().invoke(
|
|
cli.cli, [db_path, "select id, name, friends from dogs", "--json-cols"]
|
|
)
|
|
expected = r"""
|
|
[{"id": 1, "name": "Cleo", "friends": [{"name": "Pancakes"}, {"name": "Bailey"}]}]
|
|
""".strip()
|
|
assert expected == result.output.strip()
|
|
# Test rows command too
|
|
result_rows = CliRunner().invoke(cli.cli, ["rows", db_path, "dogs", "--json-cols"])
|
|
assert expected == result_rows.output.strip()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"content,is_binary",
|
|
[(b"\x00\x0Fbinary", True), ("this is text", False), (1, False), (1.5, False)],
|
|
)
|
|
def test_query_raw(db_path, content, is_binary):
|
|
Database(db_path)["files"].insert({"content": content})
|
|
result = CliRunner().invoke(
|
|
cli.cli, [db_path, "select content from files", "--raw"]
|
|
)
|
|
if is_binary:
|
|
assert result.stdout_bytes == content
|
|
else:
|
|
assert result.output == str(content)
|
|
|
|
|
|
def test_query_memory_does_not_create_file(tmpdir):
|
|
owd = os.getcwd()
|
|
try:
|
|
os.chdir(tmpdir)
|
|
# This should create a foo.db file
|
|
CliRunner().invoke(cli.cli, ["foo.db", "select sqlite_version()"])
|
|
# This should NOT create a file
|
|
result = CliRunner().invoke(cli.cli, [":memory:", "select sqlite_version()"])
|
|
assert ["sqlite_version()"] == list(json.loads(result.output)[0].keys())
|
|
finally:
|
|
os.chdir(owd)
|
|
assert ["foo.db"] == os.listdir(tmpdir)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"args,expected",
|
|
[
|
|
(
|
|
[],
|
|
'[{"id": 1, "name": "Cleo", "age": 4},\n {"id": 2, "name": "Pancakes", "age": 2}]',
|
|
),
|
|
(
|
|
["--nl"],
|
|
'{"id": 1, "name": "Cleo", "age": 4}\n{"id": 2, "name": "Pancakes", "age": 2}',
|
|
),
|
|
(["--arrays"], '[[1, "Cleo", 4],\n [2, "Pancakes", 2]]'),
|
|
(["--arrays", "--nl"], '[1, "Cleo", 4]\n[2, "Pancakes", 2]'),
|
|
(
|
|
["--nl", "-c", "age", "-c", "name"],
|
|
'{"age": 4, "name": "Cleo"}\n{"age": 2, "name": "Pancakes"}',
|
|
),
|
|
# --limit and --offset
|
|
(
|
|
["-c", "name", "--limit", "1"],
|
|
'[{"name": "Cleo"}]',
|
|
),
|
|
(
|
|
["-c", "name", "--limit", "1", "--offset", "1"],
|
|
'[{"name": "Pancakes"}]',
|
|
),
|
|
# --where
|
|
(
|
|
["-c", "name", "--where", "id = 1"],
|
|
'[{"name": "Cleo"}]',
|
|
),
|
|
(
|
|
["-c", "name", "--where", "id = :id", "-p", "id", "1"],
|
|
'[{"name": "Cleo"}]',
|
|
),
|
|
(
|
|
["-c", "name", "--where", "id = :id", "--param", "id", "1"],
|
|
'[{"name": "Cleo"}]',
|
|
),
|
|
],
|
|
)
|
|
def test_rows(db_path, args, expected):
|
|
db = Database(db_path)
|
|
with db.conn:
|
|
db["dogs"].insert_all(
|
|
[
|
|
{"id": 1, "age": 4, "name": "Cleo"},
|
|
{"id": 2, "age": 2, "name": "Pancakes"},
|
|
],
|
|
column_order=("id", "name", "age"),
|
|
)
|
|
result = CliRunner().invoke(cli.cli, ["rows", db_path, "dogs"] + args)
|
|
assert expected == result.output.strip()
|
|
|
|
|
|
def test_upsert(db_path, tmpdir):
|
|
json_path = str(tmpdir / "dogs.json")
|
|
db = Database(db_path)
|
|
insert_dogs = [
|
|
{"id": 1, "name": "Cleo", "age": 4},
|
|
{"id": 2, "name": "Nixie", "age": 4},
|
|
]
|
|
open(json_path, "w").write(json.dumps(insert_dogs))
|
|
result = CliRunner().invoke(
|
|
cli.cli,
|
|
["insert", db_path, "dogs", json_path, "--pk", "id"],
|
|
catch_exceptions=False,
|
|
)
|
|
assert 0 == result.exit_code, result.output
|
|
assert 2 == db["dogs"].count
|
|
# Now run the upsert to update just their ages
|
|
upsert_dogs = [
|
|
{"id": 1, "age": 5},
|
|
{"id": 2, "age": 5},
|
|
]
|
|
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 list(db.query("select * from dogs order by id")) == [
|
|
{"id": 1, "name": "Cleo", "age": 5},
|
|
{"id": 2, "name": "Nixie", "age": 5},
|
|
]
|
|
|
|
|
|
def test_upsert_pk_required(db_path, tmpdir):
|
|
json_path = str(tmpdir / "dogs.json")
|
|
insert_dogs = [
|
|
{"id": 1, "name": "Cleo", "age": 4},
|
|
{"id": 2, "name": "Nixie", "age": 4},
|
|
]
|
|
open(json_path, "w").write(json.dumps(insert_dogs))
|
|
result = CliRunner().invoke(
|
|
cli.cli,
|
|
["upsert", db_path, "dogs", json_path],
|
|
catch_exceptions=False,
|
|
)
|
|
assert result.exit_code == 2
|
|
assert "Error: Missing option '--pk'" in result.output
|
|
|
|
|
|
def test_upsert_analyze(db_path, tmpdir):
|
|
db = Database(db_path)
|
|
db["rows"].insert({"id": 1, "foo": "x", "n": 3}, pk="id")
|
|
db["rows"].create_index(["n"])
|
|
assert "sqlite_stat1" not in db.table_names()
|
|
result = CliRunner().invoke(
|
|
cli.cli,
|
|
["upsert", db_path, "rows", "-", "--nl", "--analyze", "--pk", "id"],
|
|
input='{"id": 2, "foo": "bar", "n": 1}',
|
|
)
|
|
assert 0 == result.exit_code, result.output
|
|
assert "sqlite_stat1" in db.table_names()
|
|
|
|
|
|
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):
|
|
json_path = str(tmpdir / "dogs.json")
|
|
db = Database(db_path)
|
|
insert_dogs = [{"id": 1, "name": "Cleo"}]
|
|
open(json_path, "w").write(json.dumps(insert_dogs))
|
|
result = CliRunner().invoke(
|
|
cli.cli, ["insert", db_path, "dogs", json_path, "--pk", "id"]
|
|
)
|
|
assert 0 == result.exit_code, result.output
|
|
# Should fail with error code if no --alter
|
|
upsert_dogs = [{"id": 1, "age": 5}]
|
|
open(json_path, "w").write(json.dumps(upsert_dogs))
|
|
result = CliRunner().invoke(
|
|
cli.cli, ["upsert", db_path, "dogs", json_path, "--pk", "id"]
|
|
)
|
|
assert 1 == result.exit_code
|
|
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"]
|
|
)
|
|
assert 0 == result.exit_code
|
|
assert [
|
|
{"id": 1, "name": "Cleo", "age": 5},
|
|
] == list(db.query("select * from dogs order by id"))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"args,schema",
|
|
[
|
|
# No primary key
|
|
(
|
|
[
|
|
"name",
|
|
"text",
|
|
"age",
|
|
"integer",
|
|
],
|
|
("CREATE TABLE [t] (\n [name] TEXT,\n [age] INTEGER\n)"),
|
|
),
|
|
# All types:
|
|
(
|
|
[
|
|
"id",
|
|
"integer",
|
|
"name",
|
|
"text",
|
|
"age",
|
|
"integer",
|
|
"weight",
|
|
"float",
|
|
"thumbnail",
|
|
"blob",
|
|
"--pk",
|
|
"id",
|
|
],
|
|
(
|
|
"CREATE TABLE [t] (\n"
|
|
" [id] INTEGER PRIMARY KEY,\n"
|
|
" [name] TEXT,\n"
|
|
" [age] INTEGER,\n"
|
|
" [weight] FLOAT,\n"
|
|
" [thumbnail] BLOB\n"
|
|
")"
|
|
),
|
|
),
|
|
# Not null:
|
|
(
|
|
["name", "text", "--not-null", "name"],
|
|
("CREATE TABLE [t] (\n" " [name] TEXT NOT NULL\n" ")"),
|
|
),
|
|
# Default:
|
|
(
|
|
["age", "integer", "--default", "age", "3"],
|
|
("CREATE TABLE [t] (\n" " [age] INTEGER DEFAULT '3'\n" ")"),
|
|
),
|
|
],
|
|
)
|
|
def test_create_table(args, schema):
|
|
runner = CliRunner()
|
|
with runner.isolated_filesystem():
|
|
result = runner.invoke(
|
|
cli.cli,
|
|
[
|
|
"create-table",
|
|
"test.db",
|
|
"t",
|
|
]
|
|
+ args,
|
|
catch_exceptions=False,
|
|
)
|
|
assert 0 == result.exit_code
|
|
db = Database("test.db")
|
|
assert schema == db["t"].schema
|
|
|
|
|
|
def test_create_table_foreign_key():
|
|
runner = CliRunner()
|
|
creates = (
|
|
["authors", "id", "integer", "name", "text", "--pk", "id"],
|
|
[
|
|
"books",
|
|
"id",
|
|
"integer",
|
|
"title",
|
|
"text",
|
|
"author_id",
|
|
"integer",
|
|
"--pk",
|
|
"id",
|
|
"--fk",
|
|
"author_id",
|
|
"authors",
|
|
"id",
|
|
],
|
|
)
|
|
with runner.isolated_filesystem():
|
|
for args in creates:
|
|
result = runner.invoke(
|
|
cli.cli, ["create-table", "books.db"] + args, catch_exceptions=False
|
|
)
|
|
assert 0 == result.exit_code
|
|
db = Database("books.db")
|
|
assert (
|
|
"CREATE TABLE [authors] (\n"
|
|
" [id] INTEGER PRIMARY KEY,\n"
|
|
" [name] TEXT\n"
|
|
")"
|
|
) == db["authors"].schema
|
|
assert (
|
|
"CREATE TABLE [books] (\n"
|
|
" [id] INTEGER PRIMARY KEY,\n"
|
|
" [title] TEXT,\n"
|
|
" [author_id] INTEGER REFERENCES [authors]([id])\n"
|
|
")"
|
|
) == db["books"].schema
|
|
|
|
|
|
def test_create_table_error_if_table_exists():
|
|
runner = CliRunner()
|
|
with runner.isolated_filesystem():
|
|
db = Database("test.db")
|
|
db["dogs"].insert({"name": "Cleo"})
|
|
result = runner.invoke(
|
|
cli.cli, ["create-table", "test.db", "dogs", "id", "integer"]
|
|
)
|
|
assert 1 == result.exit_code
|
|
assert (
|
|
'Error: Table "dogs" already exists. Use --replace to delete and replace it.'
|
|
== result.output.strip()
|
|
)
|
|
|
|
|
|
def test_create_table_ignore():
|
|
runner = CliRunner()
|
|
with runner.isolated_filesystem():
|
|
db = Database("test.db")
|
|
db["dogs"].insert({"name": "Cleo"})
|
|
result = runner.invoke(
|
|
cli.cli, ["create-table", "test.db", "dogs", "id", "integer", "--ignore"]
|
|
)
|
|
assert 0 == result.exit_code
|
|
assert "CREATE TABLE [dogs] (\n [name] TEXT\n)" == db["dogs"].schema
|
|
|
|
|
|
def test_create_table_replace():
|
|
runner = CliRunner()
|
|
with runner.isolated_filesystem():
|
|
db = Database("test.db")
|
|
db["dogs"].insert({"name": "Cleo"})
|
|
result = runner.invoke(
|
|
cli.cli, ["create-table", "test.db", "dogs", "id", "integer", "--replace"]
|
|
)
|
|
assert 0 == result.exit_code
|
|
assert "CREATE TABLE [dogs] (\n [id] INTEGER\n)" == db["dogs"].schema
|
|
|
|
|
|
def test_create_view():
|
|
runner = CliRunner()
|
|
with runner.isolated_filesystem():
|
|
db = Database("test.db")
|
|
result = runner.invoke(
|
|
cli.cli, ["create-view", "test.db", "version", "select sqlite_version()"]
|
|
)
|
|
assert 0 == result.exit_code
|
|
assert "CREATE VIEW version AS select sqlite_version()" == db["version"].schema
|
|
|
|
|
|
def test_create_view_error_if_view_exists():
|
|
runner = CliRunner()
|
|
with runner.isolated_filesystem():
|
|
db = Database("test.db")
|
|
db.create_view("version", "select sqlite_version() + 1")
|
|
result = runner.invoke(
|
|
cli.cli, ["create-view", "test.db", "version", "select sqlite_version()"]
|
|
)
|
|
assert 1 == result.exit_code
|
|
assert (
|
|
'Error: View "version" already exists. Use --replace to delete and replace it.'
|
|
== result.output.strip()
|
|
)
|
|
|
|
|
|
def test_create_view_ignore():
|
|
runner = CliRunner()
|
|
with runner.isolated_filesystem():
|
|
db = Database("test.db")
|
|
db.create_view("version", "select sqlite_version() + 1")
|
|
result = runner.invoke(
|
|
cli.cli,
|
|
[
|
|
"create-view",
|
|
"test.db",
|
|
"version",
|
|
"select sqlite_version()",
|
|
"--ignore",
|
|
],
|
|
)
|
|
assert 0 == result.exit_code
|
|
assert (
|
|
"CREATE VIEW version AS select sqlite_version() + 1" == db["version"].schema
|
|
)
|
|
|
|
|
|
def test_create_view_replace():
|
|
runner = CliRunner()
|
|
with runner.isolated_filesystem():
|
|
db = Database("test.db")
|
|
db.create_view("version", "select sqlite_version() + 1")
|
|
result = runner.invoke(
|
|
cli.cli,
|
|
[
|
|
"create-view",
|
|
"test.db",
|
|
"version",
|
|
"select sqlite_version()",
|
|
"--replace",
|
|
],
|
|
)
|
|
assert 0 == result.exit_code
|
|
assert "CREATE VIEW version AS select sqlite_version()" == db["version"].schema
|
|
|
|
|
|
def test_drop_table():
|
|
runner = CliRunner()
|
|
with runner.isolated_filesystem():
|
|
db = Database("test.db")
|
|
db["t"].create({"pk": int}, pk="pk")
|
|
assert "t" in db.table_names()
|
|
result = runner.invoke(
|
|
cli.cli,
|
|
[
|
|
"drop-table",
|
|
"test.db",
|
|
"t",
|
|
],
|
|
)
|
|
assert 0 == result.exit_code
|
|
assert "t" not in db.table_names()
|
|
|
|
|
|
def test_drop_table_error():
|
|
runner = CliRunner()
|
|
with runner.isolated_filesystem():
|
|
db = Database("test.db")
|
|
db["t"].create({"pk": int}, pk="pk")
|
|
result = runner.invoke(
|
|
cli.cli,
|
|
[
|
|
"drop-table",
|
|
"test.db",
|
|
"t2",
|
|
],
|
|
)
|
|
assert 1 == result.exit_code
|
|
assert 'Error: Table "t2" does not exist' == result.output.strip()
|
|
# Using --ignore suppresses that error
|
|
result = runner.invoke(
|
|
cli.cli,
|
|
["drop-table", "test.db", "t2", "--ignore"],
|
|
)
|
|
assert 0 == result.exit_code
|
|
|
|
|
|
def test_drop_view():
|
|
runner = CliRunner()
|
|
with runner.isolated_filesystem():
|
|
db = Database("test.db")
|
|
db.create_view("hello", "select 1")
|
|
assert "hello" in db.view_names()
|
|
result = runner.invoke(
|
|
cli.cli,
|
|
[
|
|
"drop-view",
|
|
"test.db",
|
|
"hello",
|
|
],
|
|
)
|
|
assert 0 == result.exit_code
|
|
assert "hello" not in db.view_names()
|
|
|
|
|
|
def test_drop_view_error():
|
|
runner = CliRunner()
|
|
with runner.isolated_filesystem():
|
|
db = Database("test.db")
|
|
db["t"].create({"pk": int}, pk="pk")
|
|
result = runner.invoke(
|
|
cli.cli,
|
|
[
|
|
"drop-view",
|
|
"test.db",
|
|
"t2",
|
|
],
|
|
)
|
|
assert 1 == result.exit_code
|
|
assert 'Error: View "t2" does not exist' == result.output.strip()
|
|
# Using --ignore suppresses that error
|
|
result = runner.invoke(
|
|
cli.cli,
|
|
["drop-view", "test.db", "t2", "--ignore"],
|
|
)
|
|
assert 0 == result.exit_code
|
|
|
|
|
|
def test_enable_wal():
|
|
runner = CliRunner()
|
|
dbs = ["test.db", "test2.db"]
|
|
with runner.isolated_filesystem():
|
|
for dbname in dbs:
|
|
db = Database(dbname)
|
|
db["t"].create({"pk": int}, pk="pk")
|
|
assert db.journal_mode == "delete"
|
|
result = runner.invoke(cli.cli, ["enable-wal"] + dbs)
|
|
assert 0 == result.exit_code
|
|
for dbname in dbs:
|
|
db = Database(dbname)
|
|
assert db.journal_mode == "wal"
|
|
|
|
|
|
def test_disable_wal():
|
|
runner = CliRunner()
|
|
dbs = ["test.db", "test2.db"]
|
|
with runner.isolated_filesystem():
|
|
for dbname in dbs:
|
|
db = Database(dbname)
|
|
db["t"].create({"pk": int}, pk="pk")
|
|
db.enable_wal()
|
|
assert db.journal_mode == "wal"
|
|
result = runner.invoke(cli.cli, ["disable-wal"] + dbs)
|
|
assert 0 == result.exit_code
|
|
for dbname in dbs:
|
|
db = Database(dbname)
|
|
assert db.journal_mode == "delete"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"args,expected",
|
|
[
|
|
(
|
|
[],
|
|
'[{"rows_affected": 1}]',
|
|
),
|
|
(["-t"], "rows_affected\n---------------\n 1"),
|
|
],
|
|
)
|
|
def test_query_update(db_path, args, expected):
|
|
db = Database(db_path)
|
|
with db.conn:
|
|
db["dogs"].insert_all(
|
|
[
|
|
{"id": 1, "age": 4, "name": "Cleo"},
|
|
]
|
|
)
|
|
result = CliRunner().invoke(
|
|
cli.cli, [db_path, "update dogs set age = 5 where name = 'Cleo'"] + args
|
|
)
|
|
assert expected == result.output.strip()
|
|
assert list(db.query("select * from dogs")) == [
|
|
{"id": 1, "age": 5, "name": "Cleo"},
|
|
]
|
|
|
|
|
|
def test_add_foreign_keys(db_path):
|
|
db = Database(db_path)
|
|
db["countries"].insert({"id": 7, "name": "Panama"}, pk="id")
|
|
db["authors"].insert({"id": 3, "name": "Matilda", "country_id": 7}, pk="id")
|
|
db["books"].insert({"id": 2, "title": "Wolf anatomy", "author_id": 3}, pk="id")
|
|
assert db["authors"].foreign_keys == []
|
|
assert db["books"].foreign_keys == []
|
|
result = CliRunner().invoke(
|
|
cli.cli,
|
|
[
|
|
"add-foreign-keys",
|
|
db_path,
|
|
"authors",
|
|
"country_id",
|
|
"countries",
|
|
"id",
|
|
"books",
|
|
"author_id",
|
|
"authors",
|
|
"id",
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert db["authors"].foreign_keys == [
|
|
ForeignKey(
|
|
table="authors",
|
|
column="country_id",
|
|
other_table="countries",
|
|
other_column="id",
|
|
)
|
|
]
|
|
assert db["books"].foreign_keys == [
|
|
ForeignKey(
|
|
table="books", column="author_id", other_table="authors", other_column="id"
|
|
)
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"args,expected_schema",
|
|
[
|
|
(
|
|
[],
|
|
(
|
|
'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"
|
|
")"
|
|
),
|
|
),
|
|
(
|
|
["--drop", "age"],
|
|
(
|
|
'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"
|
|
")"
|
|
),
|
|
),
|
|
(
|
|
["--not-null", "name"],
|
|
(
|
|
'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"
|
|
")"
|
|
),
|
|
),
|
|
(
|
|
["--pk", "name"],
|
|
(
|
|
'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"
|
|
")"
|
|
),
|
|
),
|
|
(
|
|
["--default", "name", "Turnip"],
|
|
(
|
|
'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"
|
|
")"
|
|
),
|
|
),
|
|
(
|
|
["-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"
|
|
")"
|
|
),
|
|
),
|
|
],
|
|
)
|
|
def test_transform(db_path, args, expected_schema):
|
|
db = Database(db_path)
|
|
with db.conn:
|
|
db["dogs"].insert(
|
|
{"id": 1, "age": 4, "name": "Cleo"},
|
|
not_null={"age"},
|
|
defaults={"age": 1},
|
|
pk="id",
|
|
)
|
|
result = CliRunner().invoke(cli.cli, ["transform", db_path, "dogs"] + args)
|
|
print(result.output)
|
|
assert result.exit_code == 0
|
|
schema = db["dogs"].schema
|
|
assert schema == expected_schema
|
|
|
|
|
|
def test_transform_drop_foreign_key(db_path):
|
|
db = Database(db_path)
|
|
with db.conn:
|
|
# Create table with three foreign keys so we can drop two of them
|
|
db["country"].insert({"id": 1, "name": "France"}, pk="id")
|
|
db["city"].insert({"id": 24, "name": "Paris"}, pk="id")
|
|
db["places"].insert(
|
|
{
|
|
"id": 32,
|
|
"name": "Caveau de la Huchette",
|
|
"country": 1,
|
|
"city": 24,
|
|
},
|
|
foreign_keys=("country", "city"),
|
|
pk="id",
|
|
)
|
|
result = CliRunner().invoke(
|
|
cli.cli,
|
|
[
|
|
"transform",
|
|
db_path,
|
|
"places",
|
|
"--drop-foreign-key",
|
|
"country",
|
|
],
|
|
)
|
|
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"
|
|
")"
|
|
)
|
|
|
|
|
|
_common_other_schema = (
|
|
"CREATE TABLE [species] (\n [id] INTEGER PRIMARY KEY,\n [species] TEXT\n)"
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"args,expected_table_schema,expected_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"
|
|
")"
|
|
),
|
|
_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 [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"
|
|
")"
|
|
),
|
|
_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 [species] (\n [id] INTEGER PRIMARY KEY,\n [species] TEXT\n)",
|
|
),
|
|
],
|
|
)
|
|
def test_extract(db_path, args, expected_table_schema, expected_other_schema):
|
|
db = Database(db_path)
|
|
with db.conn:
|
|
db["trees"].insert(
|
|
{"id": 1, "address": "4 Park Ave", "species": "Palm"},
|
|
pk="id",
|
|
)
|
|
result = CliRunner().invoke(
|
|
cli.cli, ["extract", db_path, "trees", "species"] + args
|
|
)
|
|
print(result.output)
|
|
assert result.exit_code == 0
|
|
schema = db["trees"].schema
|
|
assert schema == expected_table_schema
|
|
other_schema = [t for t in db.tables if t.name not in ("trees", "Gosh", "Gosh2")][
|
|
0
|
|
].schema
|
|
assert other_schema == expected_other_schema
|
|
|
|
|
|
def test_insert_encoding(tmpdir):
|
|
db_path = str(tmpdir / "test.db")
|
|
latin1_csv = (
|
|
b"date,name,latitude,longitude\n"
|
|
b"2020-01-01,Barra da Lagoa,-27.574,-48.422\n"
|
|
b"2020-03-04,S\xe3o Paulo,-23.561,-46.645\n"
|
|
b"2020-04-05,Salta,-24.793:-65.408"
|
|
)
|
|
assert latin1_csv.decode("latin-1").split("\n")[2].split(",")[1] == "São Paulo"
|
|
csv_path = str(tmpdir / "test.csv")
|
|
open(csv_path, "wb").write(latin1_csv)
|
|
# First attempt should error:
|
|
bad_result = CliRunner().invoke(
|
|
cli.cli,
|
|
["insert", db_path, "places", csv_path, "--csv"],
|
|
catch_exceptions=False,
|
|
)
|
|
assert bad_result.exit_code == 1
|
|
assert (
|
|
"The input you provided uses a character encoding other than utf-8"
|
|
in bad_result.output
|
|
)
|
|
# Using --encoding=latin-1 should work
|
|
good_result = CliRunner().invoke(
|
|
cli.cli,
|
|
["insert", db_path, "places", csv_path, "--encoding", "latin-1", "--csv"],
|
|
catch_exceptions=False,
|
|
)
|
|
assert good_result.exit_code == 0
|
|
db = Database(db_path)
|
|
assert list(db["places"].rows) == [
|
|
{
|
|
"date": "2020-01-01",
|
|
"name": "Barra da Lagoa",
|
|
"latitude": "-27.574",
|
|
"longitude": "-48.422",
|
|
},
|
|
{
|
|
"date": "2020-03-04",
|
|
"name": "São Paulo",
|
|
"latitude": "-23.561",
|
|
"longitude": "-46.645",
|
|
},
|
|
{
|
|
"date": "2020-04-05",
|
|
"name": "Salta",
|
|
"latitude": "-24.793:-65.408",
|
|
"longitude": None,
|
|
},
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("fts", ["FTS4", "FTS5"])
|
|
@pytest.mark.parametrize(
|
|
"extra_arg,expected",
|
|
[
|
|
(
|
|
None,
|
|
'[{"rowid": 2, "id": 2, "title": "Title the second"}]\n',
|
|
),
|
|
("--csv", "rowid,id,title\n2,2,Title the second\n"),
|
|
],
|
|
)
|
|
def test_search(tmpdir, fts, extra_arg, expected):
|
|
db_path = str(tmpdir / "test.db")
|
|
db = Database(db_path)
|
|
db["articles"].insert_all(
|
|
[
|
|
{"id": 1, "title": "Title the first"},
|
|
{"id": 2, "title": "Title the second"},
|
|
{"id": 3, "title": "Title the third"},
|
|
],
|
|
pk="id",
|
|
)
|
|
db["articles"].enable_fts(["title"], fts_version=fts)
|
|
result = CliRunner().invoke(
|
|
cli.cli,
|
|
["search", db_path, "articles", "second"] + ([extra_arg] if extra_arg else []),
|
|
catch_exceptions=False,
|
|
)
|
|
assert result.exit_code == 0
|
|
assert result.output.replace("\r", "") == expected
|
|
|
|
|
|
def test_search_quote(tmpdir):
|
|
db_path = str(tmpdir / "test.db")
|
|
db = Database(db_path)
|
|
db["creatures"].insert({"name": "dog."}).enable_fts(["name"])
|
|
# Without --quote should return an error
|
|
error_result = CliRunner().invoke(cli.cli, ["search", db_path, "creatures", 'dog"'])
|
|
assert error_result.exit_code == 1
|
|
assert error_result.output == (
|
|
"Error: unterminated string\n\n"
|
|
"Try running this again with the --quote option\n"
|
|
)
|
|
# With --quote it should work
|
|
result = CliRunner().invoke(
|
|
cli.cli, ["search", db_path, "creatures", 'dog"', "--quote"]
|
|
)
|
|
assert result.exit_code == 0
|
|
assert result.output.strip() == '[{"rowid": 1, "name": "dog."}]'
|
|
|
|
|
|
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(
|
|
"extra_args,expected",
|
|
[
|
|
([], _TRIGGERS_EXPECTED),
|
|
(["articles"], _TRIGGERS_EXPECTED),
|
|
(["counter"], "[]\n"),
|
|
],
|
|
)
|
|
def test_triggers(tmpdir, extra_args, expected):
|
|
db_path = str(tmpdir / "test.db")
|
|
db = Database(db_path)
|
|
db["articles"].insert(
|
|
{"id": 1, "title": "Title the first"},
|
|
pk="id",
|
|
)
|
|
db["counter"].insert({"count": 1})
|
|
db.conn.execute(
|
|
textwrap.dedent(
|
|
"""
|
|
CREATE TRIGGER blah AFTER INSERT ON articles
|
|
BEGIN
|
|
UPDATE counter SET count = count + 1;
|
|
END
|
|
"""
|
|
)
|
|
)
|
|
args = ["triggers", db_path]
|
|
if extra_args:
|
|
args.extend(extra_args)
|
|
result = CliRunner().invoke(
|
|
cli.cli,
|
|
args,
|
|
catch_exceptions=False,
|
|
)
|
|
assert result.exit_code == 0
|
|
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")
|
|
csv_file = open(csv_path, "w")
|
|
long_string = "a" * 131073
|
|
csv_file.write("id,text\n")
|
|
csv_file.write("1,{}\n".format(long_string))
|
|
csv_file.close()
|
|
result = CliRunner().invoke(
|
|
cli.cli,
|
|
["insert", db_path, "bigtable", csv_path, "--csv"],
|
|
catch_exceptions=False,
|
|
)
|
|
assert result.exit_code == 0
|
|
db = Database(db_path)
|
|
rows = list(db["bigtable"].rows)
|
|
assert len(rows) == 1
|
|
assert rows[0]["text"] == long_string
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"args,tsv",
|
|
(
|
|
(["--csv", "--no-headers"], False),
|
|
(["--no-headers"], False),
|
|
(["--tsv", "--no-headers"], True),
|
|
),
|
|
)
|
|
def test_import_no_headers(tmpdir, args, tsv):
|
|
db_path = str(tmpdir / "test.db")
|
|
csv_path = str(tmpdir / "test.csv")
|
|
csv_file = open(csv_path, "w")
|
|
sep = "\t" if tsv else ","
|
|
csv_file.write("Cleo{sep}Dog{sep}5\n".format(sep=sep))
|
|
csv_file.write("Tracy{sep}Spider{sep}7\n".format(sep=sep))
|
|
csv_file.close()
|
|
result = CliRunner().invoke(
|
|
cli.cli,
|
|
["insert", db_path, "creatures", csv_path] + args,
|
|
catch_exceptions=False,
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
db = Database(db_path)
|
|
schema = db["creatures"].schema
|
|
assert schema == (
|
|
"CREATE TABLE [creatures] (\n"
|
|
" [untitled_1] TEXT,\n"
|
|
" [untitled_2] TEXT,\n"
|
|
" [untitled_3] TEXT\n"
|
|
")"
|
|
)
|
|
rows = list(db["creatures"].rows)
|
|
assert rows == [
|
|
{"untitled_1": "Cleo", "untitled_2": "Dog", "untitled_3": "5"},
|
|
{"untitled_1": "Tracy", "untitled_2": "Spider", "untitled_3": "7"},
|
|
]
|
|
|
|
|
|
def test_attach(tmpdir):
|
|
foo_path = str(tmpdir / "foo.db")
|
|
bar_path = str(tmpdir / "bar.db")
|
|
db = Database(foo_path)
|
|
with db.conn:
|
|
db["foo"].insert({"id": 1, "text": "foo"})
|
|
db2 = Database(bar_path)
|
|
with db2.conn:
|
|
db2["bar"].insert({"id": 1, "text": "bar"})
|
|
db.attach("bar", bar_path)
|
|
sql = "select * from foo union all select * from bar.bar"
|
|
result = CliRunner().invoke(
|
|
cli.cli,
|
|
[foo_path, "--attach", "bar", bar_path, sql],
|
|
catch_exceptions=False,
|
|
)
|
|
assert json.loads(result.output) == [
|
|
{"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 environment variable instead of option
|
|
with mock.patch.dict(os.environ, {"SQLITE_UTILS_DETECT_TYPES": "1"}):
|
|
_test()
|
|
else:
|
|
_test()
|
|
|
|
|
|
@pytest.mark.parametrize("option", ("-d", "--detect-types"))
|
|
def test_upsert_detect_types(tmpdir, option):
|
|
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"] + [option],
|
|
catch_exceptions=False,
|
|
input=data,
|
|
)
|
|
assert result.exit_code == 0
|
|
db = Database(db_path)
|
|
assert list(db["creatures"].rows) == [
|
|
{"id": 1, "name": "Cleo", "age": 6, "weight": 45.5},
|
|
{"id": 2, "name": "Dori", "age": 1, "weight": 3.5},
|
|
]
|
|
|
|
|
|
@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"
|
|
)
|
|
|
|
|
|
def test_python_dash_m():
|
|
"Tool can be run using python -m sqlite_utils"
|
|
result = subprocess.run(
|
|
[sys.executable, "-m", "sqlite_utils", "--help"], stdout=subprocess.PIPE
|
|
)
|
|
assert result.returncode == 0
|
|
assert b"Commands for interacting with a SQLite database" in result.stdout
|
|
|
|
|
|
@pytest.mark.parametrize("enable_wal", (False, True))
|
|
def test_create_database(tmpdir, enable_wal):
|
|
db_path = tmpdir / "test.db"
|
|
assert not db_path.exists()
|
|
args = ["create-database", str(db_path)]
|
|
if enable_wal:
|
|
args.append("--enable-wal")
|
|
result = CliRunner().invoke(cli.cli, args)
|
|
assert result.exit_code == 0, result.output
|
|
assert db_path.exists()
|
|
assert db_path.read_binary()[:16] == b"SQLite format 3\x00"
|
|
db = Database(str(db_path))
|
|
if enable_wal:
|
|
assert db.journal_mode == "wal"
|
|
else:
|
|
assert db.journal_mode == "delete"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"options,expected",
|
|
(
|
|
(
|
|
[],
|
|
[
|
|
{"tbl": "two_indexes", "idx": "idx_two_indexes_species", "stat": "1 1"},
|
|
{"tbl": "two_indexes", "idx": "idx_two_indexes_name", "stat": "1 1"},
|
|
{"tbl": "one_index", "idx": "idx_one_index_name", "stat": "1 1"},
|
|
],
|
|
),
|
|
(
|
|
["one_index"],
|
|
[
|
|
{"tbl": "one_index", "idx": "idx_one_index_name", "stat": "1 1"},
|
|
],
|
|
),
|
|
(
|
|
["idx_two_indexes_name"],
|
|
[
|
|
{"tbl": "two_indexes", "idx": "idx_two_indexes_name", "stat": "1 1"},
|
|
],
|
|
),
|
|
),
|
|
)
|
|
def test_analyze(tmpdir, options, expected):
|
|
db_path = str(tmpdir / "test.db")
|
|
db = Database(db_path)
|
|
db["one_index"].insert({"id": 1, "name": "Cleo"}, pk="id")
|
|
db["one_index"].create_index(["name"])
|
|
db["two_indexes"].insert({"id": 1, "name": "Cleo", "species": "dog"}, pk="id")
|
|
db["two_indexes"].create_index(["name"])
|
|
db["two_indexes"].create_index(["species"])
|
|
result = CliRunner().invoke(cli.cli, ["analyze", db_path] + options)
|
|
assert result.exit_code == 0
|
|
assert list(db["sqlite_stat1"].rows) == expected
|
|
|
|
|
|
def test_duplicate_table(tmpdir):
|
|
db_path = str(tmpdir / "test.db")
|
|
db = Database(db_path)
|
|
db["one"].insert({"id": 1, "name": "Cleo"}, pk="id")
|
|
# First try a non-existent table
|
|
result_error = CliRunner().invoke(
|
|
cli.cli,
|
|
["duplicate", db_path, "missing", "two"],
|
|
catch_exceptions=False,
|
|
)
|
|
assert result_error.exit_code == 1
|
|
assert result_error.output == 'Error: Table "missing" does not exist\n'
|
|
# Now try for a table that exists
|
|
result = CliRunner().invoke(
|
|
cli.cli,
|
|
["duplicate", db_path, "one", "two"],
|
|
catch_exceptions=False,
|
|
)
|
|
assert result.exit_code == 0
|
|
assert db["one"].columns_dict == db["two"].columns_dict
|
|
assert list(db["one"].rows) == list(db["two"].rows)
|