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

Fixes for Ruff>=0.16.0 (#814)

Browse source 
* Automated upgrades by Ruff

    uvx --with 'ruff>=0.16.0' ruff check . --fix --unsafe-fixes

* Fix remaining Ruff errors with GPT-5.6 Sol high

https://gist.github.com/simonw/6da7906a9fea6e90da131c21a9055199

* Fix flake E501 long lines
* New Protocol for migrations to make ty happy
This commit is contained in:
Simon Willison 2026-07-25 14:53:12 -07:00 • committed by GitHub
commit 69a1c0d960
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
57 changed files with 967 additions and 1042 deletions
Download patch file Download diff file Expand all files Collapse all files

5
tests/conftest.py
View file

@ -1,6 +1,7 @@
import pytest
from sqlite_utils import Database
from sqlite_utils.utils import sqlite3
import pytest
CREATE_TABLES = """
create table Gosh (c1 text, c2 text, c3 text);
@ -55,7 +56,7 @@ def close_all_databases():
for db in databases:
try:
db.close()
except Exception:
except sqlite3.Error:
pass

10
tests/test_analyze_tables.py
View file

@ -1,9 +1,11 @@
from sqlite_utils.db import Database, ColumnDetails
from sqlite_utils import cli
from click.testing import CliRunner
import pytest
import sqlite3
import pytest
from click.testing import CliRunner
from sqlite_utils import cli
from sqlite_utils.db import ColumnDetails, Database
@pytest.fixture
def db_to_analyze(fresh_db):

81
tests/test_atomic.py
View file

@ -28,11 +28,13 @@ from sqlite_utils.utils import sqlite3
END;
""",
[
"CREATE TRIGGER t_ai AFTER INSERT ON t\n"
" BEGIN\n"
" UPDATE t SET value = 'a;b' WHERE id = new.id;\n"
" INSERT INTO log VALUES ('x;y');\n"
" END;"
(
"CREATE TRIGGER t_ai AFTER INSERT ON t\n"
" BEGIN\n"
" UPDATE t SET value = 'a;b' WHERE id = new.id;\n"
" INSERT INTO log VALUES ('x;y');\n"
" END;"
)
],
),
),
@ -49,10 +51,9 @@ def test_atomic_commits(fresh_db):
def test_atomic_rolls_back(fresh_db):
with pytest.raises(RuntimeError):
with fresh_db.atomic():
fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
raise RuntimeError("boom")
with pytest.raises(RuntimeError), fresh_db.atomic():
fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
raise RuntimeError("boom")
assert not fresh_db["dogs"].exists()
@ -62,10 +63,9 @@ def test_nested_atomic_rolls_back_to_savepoint(fresh_db):
with fresh_db.atomic():
fresh_db["dogs"].insert({"id": 1, "name": "Cleo"})
with pytest.raises(RuntimeError):
with fresh_db.atomic():
fresh_db["dogs"].insert({"id": 2, "name": "Pancakes"})
raise RuntimeError("boom")
with pytest.raises(RuntimeError), fresh_db.atomic():
fresh_db["dogs"].insert({"id": 2, "name": "Pancakes"})
raise RuntimeError("boom")
fresh_db["dogs"].insert({"id": 3, "name": "Marnie"})
assert list(fresh_db["dogs"].rows) == [
@ -75,20 +75,18 @@ def test_nested_atomic_rolls_back_to_savepoint(fresh_db):
def test_outer_atomic_rolls_back_released_savepoint(fresh_db):
with pytest.raises(RuntimeError):
with pytest.raises(RuntimeError), fresh_db.atomic():
fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
with fresh_db.atomic():
fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
with fresh_db.atomic():
fresh_db["dogs"].insert({"id": 2, "name": "Pancakes"})
raise RuntimeError("boom")
fresh_db["dogs"].insert({"id": 2, "name": "Pancakes"})
raise RuntimeError("boom")
assert not fresh_db["dogs"].exists()
def test_executescript_does_not_commit_open_atomic_block(fresh_db):
with pytest.raises(RuntimeError):
with fresh_db.atomic():
fresh_db.executescript("""
with pytest.raises(RuntimeError), fresh_db.atomic():
fresh_db.executescript("""
CREATE TABLE dogs(id INTEGER PRIMARY KEY, name TEXT);
CREATE TRIGGER dogs_ai AFTER INSERT ON dogs
BEGIN
@ -97,7 +95,7 @@ def test_executescript_does_not_commit_open_atomic_block(fresh_db):
-- This comment has a semicolon;
INSERT INTO dogs VALUES (1, 'Cleo; the first');
""")
raise RuntimeError("boom")
raise RuntimeError("boom")
assert not fresh_db["dogs"].exists()
@ -105,11 +103,10 @@ def test_executescript_does_not_commit_open_atomic_block(fresh_db):
def test_transform_does_not_commit_open_atomic_block(fresh_db):
fresh_db["dogs"].insert({"id": 1, "name": "Cleo", "age": "5"}, pk="id")
with pytest.raises(RuntimeError):
with fresh_db.atomic():
fresh_db["dogs"].insert({"id": 2, "name": "Pancakes", "age": "6"})
fresh_db["dogs"].transform(rename={"age": "dog_age"})
raise RuntimeError("boom")
with pytest.raises(RuntimeError), fresh_db.atomic():
fresh_db["dogs"].insert({"id": 2, "name": "Pancakes", "age": "6"})
fresh_db["dogs"].transform(rename={"age": "dog_age"})
raise RuntimeError("boom")
assert (
fresh_db["dogs"].schema
@ -149,10 +146,9 @@ def test_transform_parent_table_with_foreign_keys_rolls_back(fresh_db):
foreign_keys={"author_id"},
)
with pytest.raises(RuntimeError):
with fresh_db.atomic():
fresh_db["authors"].transform(rename={"name": "full_name"})
raise RuntimeError("boom")
with pytest.raises(RuntimeError), fresh_db.atomic():
fresh_db["authors"].transform(rename={"name": "full_name"})
raise RuntimeError("boom")
assert (
fresh_db["authors"].schema
@ -354,9 +350,11 @@ def test_atomic_preserves_error_from_transaction_destroying_trigger(fresh_db):
# with "cannot rollback - no transaction is active"
fresh_db.execute("create table t (id integer primary key, v text)")
fresh_db.execute(TRIGGER_SQL)
with pytest.raises(sqlite3.IntegrityError, match="trigger says no"):
with fresh_db.atomic():
fresh_db.execute("insert into t (v) values ('bad')")
with (
pytest.raises(sqlite3.IntegrityError, match="trigger says no"),
fresh_db.atomic(),
):
fresh_db.execute("insert into t (v) values ('bad')")
assert not fresh_db.conn.in_transaction
@ -367,16 +365,17 @@ def test_nested_atomic_preserves_error_from_transaction_destroying_trigger(
# "no such savepoint" from ROLLBACK TO SAVEPOINT
fresh_db.execute("create table t (id integer primary key, v text)")
fresh_db.execute(TRIGGER_SQL)
with pytest.raises(sqlite3.IntegrityError, match="trigger says no"):
with fresh_db.atomic():
with fresh_db.atomic():
fresh_db.execute("insert into t (v) values ('bad')")
with (
pytest.raises(sqlite3.IntegrityError, match="trigger says no"),
fresh_db.atomic(),
fresh_db.atomic(),
):
fresh_db.execute("insert into t (v) values ('bad')")
assert not fresh_db.conn.in_transaction
def test_atomic_preserves_error_from_insert_or_rollback(fresh_db):
fresh_db["t"].insert({"id": 1}, pk="id")
with pytest.raises(sqlite3.IntegrityError):
with fresh_db.atomic():
fresh_db.execute("insert or rollback into t (id) values (1)")
with pytest.raises(sqlite3.IntegrityError), fresh_db.atomic():
fresh_db.execute("insert or rollback into t (id) values (1)")
assert not fresh_db.conn.in_transaction

88
tests/test_cli.py
View file

@ -1,14 +1,16 @@
from sqlite_utils import cli, Database
from sqlite_utils.db import Index, ForeignKey
from click.testing import CliRunner
from pathlib import Path
import subprocess
import sqlite3
import sys
import json
import os
import pytest
import sqlite3
import subprocess
import sys
import textwrap
from pathlib import Path
import pytest
from click.testing import CliRunner
from sqlite_utils import Database, cli
from sqlite_utils.db import ForeignKey, Index
def write_json(file_path, data):
@ -21,7 +23,7 @@ def _supports_pragma_function_list():
try:
db.execute("select * from pragma_function_list()")
return True
except Exception:
except sqlite3.DatabaseError:
return False
finally:
db.close()
@ -184,9 +186,9 @@ def test_output_table(db_path, options, expected):
db["rows"].insert_all(
[
{
"c1": "verb{}".format(i),
"c2": "noun{}".format(i),
"c3": "adjective{}".format(i),
"c1": f"verb{i}",
"c2": f"noun{i}",
"c3": f"adjective{i}",
}
for i in range(4)
]
@ -678,9 +680,9 @@ def test_optimize(db_path, tables):
db[table].insert_all(
[
{
"c1": "verb{}".format(i),
"c2": "noun{}".format(i),
"c3": "adjective{}".format(i),
"c1": f"verb{i}",
"c2": f"noun{i}",
"c3": f"adjective{i}",
}
for i in range(10000)
]
@ -704,9 +706,9 @@ 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),
"c1": f"verb{i}",
"c2": f"noun{i}",
"c3": f"adjective{i}",
}
for i in range(10000)
]
@ -1019,16 +1021,14 @@ def test_query_json_binary(db_path):
"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"
)
"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"
),
},
}
@ -2114,11 +2114,13 @@ _common_other_schema = (
),
(
["--rename", "name", "name2"],
'CREATE TABLE "trees" (\n'
' "id" INTEGER PRIMARY KEY,\n'
' "address" TEXT,\n'
' "species_id" INTEGER REFERENCES "species"("id")\n'
")",
(
'CREATE TABLE "trees" (\n'
' "id" INTEGER PRIMARY KEY,\n'
' "address" TEXT,\n'
' "species_id" INTEGER REFERENCES "species"("id")\n'
")"
),
'CREATE TABLE "species" (\n "id" INTEGER PRIMARY KEY,\n "species" TEXT\n)',
),
],
@ -2137,9 +2139,9 @@ def test_extract(db_path, args, expected_table_schema, expected_other_schema):
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
other_schema = next(
t for t in db.tables if t.name not in ("trees", "Gosh", "Gosh2")
).schema
assert other_schema == expected_other_schema
@ -2431,7 +2433,7 @@ def test_long_csv_column_value(tmpdir):
with open(csv_path, "w") as csv_file:
long_string = "a" * 131073
csv_file.write("id,text\n")
csv_file.write("1,{}\n".format(long_string))
csv_file.write(f"1,{long_string}\n")
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "bigtable", csv_path, "--csv"],
@ -2457,8 +2459,8 @@ def test_import_no_headers(tmpdir, args, tsv):
csv_path = str(tmpdir / "test.csv")
with open(csv_path, "w") as csv_file:
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.write(f"Cleo{sep}Dog{sep}5\n")
csv_file.write(f"Tracy{sep}Spider{sep}7\n")
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "creatures", csv_path] + args + ["--no-detect-types"],
@ -2690,7 +2692,9 @@ def test_integer_overflow_error(tmpdir):
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
[sys.executable, "-m", "sqlite_utils", "--help"],
stdout=subprocess.PIPE,
check=False,
)
assert result.returncode == 0
assert b"Commands for interacting with a SQLite database" in result.stdout
@ -2830,14 +2834,14 @@ def test_load_extension(entrypoint, should_pass, should_fail):
for func in should_pass:
result = CliRunner().invoke(
cli.cli,
["memory", "select {}()".format(func), "--load-extension", ext],
["memory", f"select {func}()", "--load-extension", ext],
catch_exceptions=False,
)
assert result.exit_code == 0
for func in should_fail:
result = CliRunner().invoke(
cli.cli,
["memory", "select {}()".format(func), "--load-extension", ext],
["memory", f"select {func}()", "--load-extension", ext],
catch_exceptions=False,
)
assert result.exit_code == 1

8
tests/test_cli_bulk.py
View file

@ -1,11 +1,13 @@
from click.testing import CliRunner
from sqlite_utils import cli, Database
import pathlib
import pytest
import subprocess
import sys
import time
import pytest
from click.testing import CliRunner
from sqlite_utils import Database, cli
@pytest.fixture
def test_db_and_path(tmpdir):

16
tests/test_cli_convert.py
View file

@ -1,10 +1,12 @@
from click.testing import CliRunner
from sqlite_utils import cli
import sqlite_utils
import json
import textwrap
import pathlib
import textwrap
import pytest
from click.testing import CliRunner
import sqlite_utils
from sqlite_utils import cli
@pytest.fixture
@ -50,7 +52,7 @@ def test_convert_code(fresh_db_and_path, code):
cli.cli, ["convert", db_path, "t", "text", code], catch_exceptions=False
)
assert result.exit_code == 0, result.output
value = list(db["t"].rows)[0]["text"]
value = next(iter(db["t"].rows))["text"]
assert value == "Spooktober"
@ -442,7 +444,7 @@ def test_recipe_jsonsplit(tmpdir, delimiter):
)
code = "r.jsonsplit(value)"
if delimiter:
code = 'recipes.jsonsplit(value, delimiter="{}")'.format(delimiter)
code = f'recipes.jsonsplit(value, delimiter="{delimiter}")'
args = ["convert", db_path, "example", "tags", code]
result = CliRunner().invoke(cli.cli, args)
assert result.exit_code == 0, result.output
@ -470,7 +472,7 @@ def test_recipe_jsonsplit_type(fresh_db_and_path, type, expected_array):
)
code = "r.jsonsplit(value)"
if type:
code = "recipes.jsonsplit(value, type={})".format(type)
code = f"recipes.jsonsplit(value, type={type})"
args = ["convert", db_path, "example", "records", code]
result = CliRunner().invoke(cli.cli, args)
assert result.exit_code == 0, result.output

17
tests/test_cli_insert.py
View file

@ -1,11 +1,13 @@
from sqlite_utils import cli, Database
from click.testing import CliRunner
import json
import pytest
import subprocess
import sys
import time
import pytest
from click.testing import CliRunner
from sqlite_utils import Database, cli
def test_insert_simple(tmpdir):
json_path = str(tmpdir / "dog.json")
@ -99,7 +101,7 @@ def test_insert_with_primary_keys(db_path, tmpdir, args, expected_pks):
def test_insert_multiple_with_primary_key(db_path, tmpdir):
json_path = str(tmpdir / "dogs.json")
dogs = [{"id": i, "name": "Cleo {}".format(i), "age": i + 3} for i in range(1, 21)]
dogs = [{"id": i, "name": f"Cleo {i}", "age": i + 3} for i in range(1, 21)]
with open(json_path, "w") as fp:
fp.write(json.dumps(dogs))
result = CliRunner().invoke(
@ -114,7 +116,7 @@ def test_insert_multiple_with_primary_key(db_path, tmpdir):
def test_insert_multiple_with_compound_primary_key(db_path, tmpdir):
json_path = str(tmpdir / "dogs.json")
dogs = [
{"breed": "mixed", "id": i, "name": "Cleo {}".format(i), "age": i + 3}
{"breed": "mixed", "id": i, "name": f"Cleo {i}", "age": i + 3}
for i in range(1, 21)
]
with open(json_path, "w") as fp:
@ -140,8 +142,7 @@ def test_insert_multiple_with_compound_primary_key(db_path, tmpdir):
def test_insert_not_null_default(db_path, tmpdir):
json_path = str(tmpdir / "dogs.json")
dogs = [
{"id": i, "name": "Cleo {}".format(i), "age": i + 3, "score": 10}
for i in range(1, 21)
{"id": i, "name": f"Cleo {i}", "age": i + 3, "score": 10} for i in range(1, 21)
]
with open(json_path, "w") as fp:
fp.write(json.dumps(dogs))
@ -587,7 +588,7 @@ def test_insert_streaming_batch_size_1(db_path):
return
tries += 1
if tries > 10:
assert False, "Expected {}, got {}".format(expected, rows)
assert False, f"Expected {expected}, got {rows}"
time.sleep(tries * 0.1)
try_until([{"name": "Azi"}])

13
tests/test_cli_memory.py
View file

@ -1,5 +1,6 @@
import click
import json
import click
import pytest
from click.testing import CliRunner
@ -28,7 +29,7 @@ def test_memory_csv(tmpdir, sql_from, use_stdin):
fp.write(content)
result = CliRunner().invoke(
cli.cli,
["memory", csv_path, "select * from {}".format(sql_from), "--nl"],
["memory", csv_path, f"select * from {sql_from}", "--nl"],
input=input,
)
assert result.exit_code == 0
@ -53,7 +54,7 @@ def test_memory_tsv(tmpdir, use_stdin):
sql_from = "chickens"
result = CliRunner().invoke(
cli.cli,
["memory", path, "select * from {}".format(sql_from)],
["memory", path, f"select * from {sql_from}"],
input=input,
)
assert result.exit_code == 0, result.output
@ -79,7 +80,7 @@ def test_memory_json(tmpdir, use_stdin):
sql_from = "chickens"
result = CliRunner().invoke(
cli.cli,
["memory", path, "select * from {}".format(sql_from)],
["memory", path, f"select * from {sql_from}"],
input=input,
)
assert result.exit_code == 0, result.output
@ -105,7 +106,7 @@ def test_memory_json_nl(tmpdir, use_stdin):
sql_from = "chickens"
result = CliRunner().invoke(
cli.cli,
["memory", path, "select * from {}".format(sql_from)],
["memory", path, f"select * from {sql_from}"],
input=input,
)
assert result.exit_code == 0, result.output
@ -135,7 +136,7 @@ def test_memory_csv_encoding(tmpdir, use_stdin):
CliRunner()
.invoke(
cli.cli,
["memory", csv_path, "select * from {}".format(sql_from), "--nl"],
["memory", csv_path, f"select * from {sql_from}", "--nl"],
input=input,
)
.exit_code

3
tests/test_cli_migrate.py
View file

@ -1,7 +1,8 @@
import pathlib
from click.testing import CliRunner
import pytest
from click.testing import CliRunner
import sqlite_utils
import sqlite_utils.cli

3
tests/test_column_affinity.py
View file

@ -1,4 +1,5 @@
import pytest
from sqlite_utils.utils import column_affinity
EXAMPLES = [
@ -41,5 +42,5 @@ def test_column_affinity(column_def, expected_type):
@pytest.mark.parametrize("column_def,expected_type", EXAMPLES)
def test_columns_dict(fresh_db, column_def, expected_type):
fresh_db.execute("create table foo (col {})".format(column_def))
fresh_db.execute(f"create table foo (col {column_def})")
assert {"col": expected_type} == fresh_db["foo"].columns_dict

6
tests/test_constructor.py
View file

@ -1,8 +1,10 @@
import sys
import pytest
from sqlite_utils import Database
from sqlite_utils.db import TransactionError
from sqlite_utils.utils import sqlite3
import pytest
import sys
def test_recursive_triggers():

3
tests/test_convert.py
View file

@ -1,6 +1,7 @@
from sqlite_utils.db import BadMultiValues
import pytest
from sqlite_utils.db import BadMultiValues
@pytest.mark.parametrize(
"columns,fn,expected",

68
tests/test_create.py
View file

@ -1,26 +1,28 @@
from sqlite_utils.db import (
Index,
Database,
DescIndex,
AlterError,
InvalidColumns,
NoObviousTable,
OperationalError,
ForeignKey,
Table,
View,
NoTable,
NoView,
)
from sqlite_utils.utils import hash_record, sqlite3
import collections
import datetime
import decimal
import json
import pathlib
import pytest
import uuid
import pytest
from sqlite_utils.db import (
AlterError,
Database,
DescIndex,
ForeignKey,
Index,
InvalidColumns,
NoObviousTable,
NoTable,
NoView,
OperationalError,
Table,
View,
)
from sqlite_utils.utils import hash_record, sqlite3
try:
import pandas as pd # type: ignore
except ImportError:
@ -699,7 +701,7 @@ def test_bulk_insert_more_than_999_values(fresh_db):
"num_columns,should_error", ((900, False), (999, False), (1000, True))
)
def test_error_if_more_than_999_columns(fresh_db, num_columns, should_error):
record = dict([("c{}".format(i), i) for i in range(num_columns)])
record = {f"c{i}": i for i in range(num_columns)}
if should_error:
with pytest.raises(ValueError):
fresh_db["big"].insert(record)
@ -718,17 +720,9 @@ def test_columns_not_in_first_record_should_not_cause_batch_to_be_too_large(fres
records = [
{"c0": "first record"}, # one column in first record -> batch size = 999
# fill out the batch with 99 records with enough columns to exceed THRESHOLD
*[
dict([("c{}".format(i), j) for i in range(extra_columns)])
for j in range(batch_size - 1)
],
*[{f"c{i}": j for i in range(extra_columns)} for j in range(batch_size - 1)],
]
try:
fresh_db["too_many_columns"].insert_all(
records, alter=True, batch_size=batch_size
)
except sqlite3.OperationalError:
raise
fresh_db["too_many_columns"].insert_all(records, alter=True, batch_size=batch_size)
@pytest.mark.parametrize(
@ -910,7 +904,7 @@ def test_insert_list_nested_unicode(fresh_db):
def test_insert_uuid(fresh_db):
uuid4 = uuid.uuid4()
fresh_db["test"].insert({"uuid": uuid4})
row = list(fresh_db["test"].rows)[0]
row = next(iter(fresh_db["test"].rows))
assert {"uuid"} == row.keys()
assert isinstance(row["uuid"], str)
assert row["uuid"] == str(uuid4)
@ -918,16 +912,14 @@ def test_insert_uuid(fresh_db):
def test_insert_memoryview(fresh_db):
fresh_db["test"].insert({"data": memoryview(b"hello")})
row = list(fresh_db["test"].rows)[0]
row = next(iter(fresh_db["test"].rows))
assert {"data"} == row.keys()
assert isinstance(row["data"], bytes)
assert row["data"] == b"hello"
def test_insert_thousands_using_generator(fresh_db):
fresh_db["test"].insert_all(
{"i": i, "word": "word_{}".format(i)} for i in range(10000)
)
fresh_db["test"].insert_all({"i": i, "word": f"word_{i}"} for i in range(10000))
assert [{"name": "i", "type": "INTEGER"}, {"name": "word", "type": "TEXT"}] == [
{"name": col.name, "type": col.type} for col in fresh_db["test"].columns
]
@ -938,7 +930,7 @@ def test_insert_thousands_raises_exception_with_extra_columns_after_first_100(fr
# https://github.com/simonw/sqlite-utils/issues/139
with pytest.raises(Exception, match="table test has no column named extra"):
fresh_db["test"].insert_all(
[{"i": i, "word": "word_{}".format(i)} for i in range(100)]
[{"i": i, "word": f"word_{i}"} for i in range(100)]
+ [{"i": 101, "extra": "This extra column should cause an exception"}],
)
@ -946,7 +938,7 @@ def test_insert_thousands_raises_exception_with_extra_columns_after_first_100(fr
def test_insert_thousands_adds_extra_columns_after_first_100_with_alter(fresh_db):
# https://github.com/simonw/sqlite-utils/issues/139
fresh_db["test"].insert_all(
[{"i": i, "word": "word_{}".format(i)} for i in range(100)]
[{"i": i, "word": f"word_{i}"} for i in range(100)]
+ [{"i": 101, "extra": "Should trigger ALTER"}],
alter=True,
)
@ -958,7 +950,7 @@ def test_insert_thousands_adds_extra_columns_after_first_100_with_alter(fresh_db
def test_insert_all_pk_not_in_records_raises(fresh_db, num_rows):
# https://github.com/simonw/sqlite-utils/issues/732
fresh_db.conn.execute("CREATE TABLE t (a TEXT, b INT, PRIMARY KEY (a, b))")
rows = [{"a": "x{}".format(i), "b": i} for i in range(num_rows)]
rows = [{"a": f"x{i}", "b": i} for i in range(num_rows)]
with pytest.raises(InvalidColumns) as ex:
fresh_db["t"].insert_all(rows, pk="not_a_column")
@ -975,7 +967,7 @@ def test_insert_all_pk_not_in_records_alter_raises(fresh_db, num_rows):
# known - a pk column that is in neither the table nor the records
# still raises
fresh_db.conn.execute("CREATE TABLE t (a TEXT, b INT, PRIMARY KEY (a, b))")
rows = [{"a": "x{}".format(i), "b": i} for i in range(num_rows)]
rows = [{"a": f"x{i}", "b": i} for i in range(num_rows)]
with pytest.raises(InvalidColumns) as ex:
fresh_db["t"].insert_all(rows, pk="not_a_column", alter=True)
@ -1146,7 +1138,7 @@ def test_insert_hash_id_columns(fresh_db, use_table_factory):
insert_kwargs = {}
else:
dogs = fresh_db["dogs"]
insert_kwargs = dict(hash_id_columns=("name", "twitter"))
insert_kwargs = {"hash_id_columns": ("name", "twitter")}
id = dogs.insert(
{"name": "Cleo", "twitter": "cleopaws", "age": 5},
@ -1654,7 +1646,7 @@ def test_upsert_uses_pk_from_prior_insert_655(fresh_db):
# Upsert should work without specifying pk again
table.upsert({"id": 1, "name": "Alice Updated"})
assert table.count == 1
assert list(table.rows)[0]["name"] == "Alice Updated"
assert next(iter(table.rows))["name"] == "Alice Updated"
def test_upsert_all_uses_pk_from_prior_insert_655(fresh_db):

1
tests/test_create_view.py
View file

@ -1,4 +1,5 @@
import pytest
from sqlite_utils.utils import OperationalError

2
tests/test_default_value.py
View file

@ -31,7 +31,7 @@ EXAMPLES = [
@pytest.mark.parametrize("column_def,initial_value,expected_value", EXAMPLES)
def test_quote_default_value(fresh_db, column_def, initial_value, expected_value):
fresh_db.execute("create table foo (col {})".format(column_def))
fresh_db.execute(f"create table foo (col {column_def})")
assert initial_value == fresh_db["foo"].columns[0].default_value
assert expected_value == fresh_db.quote_default_value(
fresh_db["foo"].columns[0].default_value

2
tests/test_delete.py
View file

@ -3,7 +3,7 @@ import sqlite_utils
def test_delete_rowid_table(fresh_db):
table = fresh_db["table"]
table.insert({"foo": 1}).last_pk
table.insert({"foo": 1})
rowid = table.insert({"foo": 2}).last_pk
table.delete(rowid)
assert [{"foo": 1}] == list(table.rows)

12
tests/test_docs.py
View file

@ -1,8 +1,10 @@
from click.testing import CliRunner
from sqlite_utils import cli, recipes
from pathlib import Path
import pytest
import re
from pathlib import Path
import pytest
from click.testing import CliRunner
from sqlite_utils import cli, recipes
docs_path = Path(__file__).parent.parent / "docs"
commands_re = re.compile(r"(?:\$ | )sqlite-utils (\S+)")
@ -34,7 +36,7 @@ def test_commands_are_documented(documented_commands, command):
@pytest.mark.parametrize("command", cli.cli.commands.values())
def test_commands_have_help(command):
assert command.help, "{} is missing its help".format(command)
assert command.help, f"{command} is missing its help"
def test_convert_help():

6
tests/test_duplicate.py
View file

@ -1,7 +1,9 @@
from sqlite_utils.db import NoTable
import datetime
import pytest
from sqlite_utils.db import NoTable
def test_duplicate(fresh_db):
# Create table using native Sqlite statement:
@ -12,7 +14,7 @@ def test_duplicate(fresh_db):
"bool_col" INTEGER,
"datetime_col" TEXT)""")
# Insert one row of mock data:
dt = datetime.datetime.now()
dt = datetime.datetime.now(datetime.timezone.utc)
data = {
"text_col": "Cleo",
"real_col": 3.14,

10
tests/test_enable_counts.py
View file

@ -1,14 +1,14 @@
from sqlite_utils import Database
from sqlite_utils import cli
from click.testing import CliRunner
import pytest
from click.testing import CliRunner
from sqlite_utils import Database, cli
def test_enable_counts_specific_table(fresh_db):
foo = fresh_db["foo"]
assert fresh_db.table_names() == []
for i in range(10):
foo.insert({"name": "item {}".format(i)})
foo.insert({"name": f"item {i}"})
assert fresh_db.table_names() == ["foo"]
assert foo.count == 10
# Now enable counts
@ -44,7 +44,7 @@ def test_enable_counts_specific_table(fresh_db):
assert list(fresh_db["_counts"].rows) == [{"count": 10, "table": "foo"}]
# Add some items to test the triggers
for i in range(5):
foo.insert({"name": "item {}".format(10 + i)})
foo.insert({"name": f"item {10 + i}"})
assert foo.count == 15
assert list(fresh_db["_counts"].rows) == [{"count": 15, "table": "foo"}]
# Delete some items

15
tests/test_extract.py
View file

@ -1,19 +1,21 @@
from sqlite_utils.db import InvalidColumns
import itertools
import pytest
from sqlite_utils.db import InvalidColumns
@pytest.mark.parametrize("table", [None, "Species"])
@pytest.mark.parametrize("fk_column", [None, "species"])
def test_extract_single_column(fresh_db, table, fk_column):
expected_table = table or "species"
expected_fk = fk_column or "{}_id".format(expected_table)
expected_fk = fk_column or f"{expected_table}_id"
iter_species = itertools.cycle(["Palm", "Spruce", "Mangrove", "Oak"])
fresh_db["tree"].insert_all(
(
{
"id": i,
"name": "Tree {}".format(i),
"name": f"Tree {i}",
"species": next(iter_species),
"end": 1,
}
@ -26,13 +28,12 @@ def test_extract_single_column(fresh_db, table, fk_column):
'CREATE TABLE "tree" (\n'
' "id" INTEGER PRIMARY KEY,\n'
' "name" TEXT,\n'
' "{}" INTEGER REFERENCES "{}"("id"),\n'.format(expected_fk, expected_table)
f' "{expected_fk}" INTEGER REFERENCES "{expected_table}"("id"),\n'
+ ' "end" INTEGER\n'
+ ")"
)
assert fresh_db[expected_table].schema == (
'CREATE TABLE "{}" (\n'.format(expected_table)
+ ' "id" INTEGER PRIMARY KEY,\n'
f'CREATE TABLE "{expected_table}" (\n' + ' "id" INTEGER PRIMARY KEY,\n'
' "species" TEXT\n'
")"
)
@ -57,7 +58,7 @@ def test_extract_multiple_columns_with_rename(fresh_db):
(
{
"id": i,
"name": "Tree {}".format(i),
"name": f"Tree {i}",
"common_name": next(iter_common),
"latin_name": next(iter_latin),
}

19
tests/test_extracts.py
View file

@ -1,13 +1,14 @@
from sqlite_utils.db import Index
import pytest
from sqlite_utils.db import Index
@pytest.mark.parametrize(
"kwargs,expected_table",
[
(dict(extracts={"species_id": "Species"}), "Species"),
(dict(extracts=["species_id"]), "species_id"),
(dict(extracts=("species_id",)), "species_id"),
({"extracts": {"species_id": "Species"}}, "Species"),
({"extracts": ["species_id"]}, "species_id"),
({"extracts": ("species_id",)}, "species_id"),
],
)
@pytest.mark.parametrize("use_table_factory", [True, False])
@ -30,15 +31,11 @@ def test_extracts(fresh_db, kwargs, expected_table, use_table_factory):
# Should now have two tables: Trees and Species
assert {expected_table, "Trees"} == set(fresh_db.table_names())
assert (
'CREATE TABLE "{}" (\n "id" INTEGER PRIMARY KEY,\n "value" TEXT\n)'.format(
expected_table
)
f'CREATE TABLE "{expected_table}" (\n "id" INTEGER PRIMARY KEY,\n "value" TEXT\n)'
== fresh_db[expected_table].schema
)
assert (
'CREATE TABLE "Trees" (\n "id" INTEGER,\n "species_id" INTEGER REFERENCES "{}"("id")\n)'.format(
expected_table
)
f'CREATE TABLE "Trees" (\n "id" INTEGER,\n "species_id" INTEGER REFERENCES "{expected_table}"("id")\n)'
== fresh_db["Trees"].schema
)
# Should have a foreign key reference
@ -51,7 +48,7 @@ def test_extracts(fresh_db, kwargs, expected_table, use_table_factory):
assert [
Index(
seq=0,
name="idx_{}_value".format(expected_table),
name=f"idx_{expected_table}_value",
unique=1,
origin="c",
partial=0,

3
tests/test_foreign_keys.py
View file

@ -1,6 +1,7 @@
"""Tests for compound (multi-column) foreign keys - issue #594."""
import pytest
from sqlite_utils import Database
from sqlite_utils.db import AlterError, ForeignKey
from sqlite_utils.utils import sqlite3
@ -64,7 +65,7 @@ def test_foreign_key_no_longer_unpacks_as_tuple(fresh_db):
fresh_db["books"].add_foreign_key("author_id", "authors", "id")
fk = fresh_db["books"].foreign_keys[0]
with pytest.raises(TypeError):
table, column, other_table, other_column = fk
_table, _column, _other_table, _other_column = fk
with pytest.raises(TypeError):
fk[0]

26
tests/test_fts.py
View file

@ -1,7 +1,9 @@
from unittest.mock import ANY
import pytest
from sqlite_utils import Database
from sqlite_utils.utils import sqlite3
from unittest.mock import ANY
search_records = [
{
@ -103,9 +105,10 @@ def test_search_limit_offset(fresh_db):
table.enable_fts(["text", "country"], fts_version="FTS4")
assert len(list(table.search("are"))) == 2
assert len(list(table.search("are", limit=1))) == 1
assert list(table.search("are", limit=1, order_by="rowid"))[0]["rowid"] == 1
assert next(iter(table.search("are", limit=1, order_by="rowid")))["rowid"] == 1
assert (
list(table.search("are", limit=1, offset=1, order_by="rowid"))[0]["rowid"] == 2
next(iter(table.search("are", limit=1, offset=1, order_by="rowid")))["rowid"]
== 2
)
@ -223,20 +226,20 @@ def test_populate_fts_escape_table_names(fresh_db):
@pytest.mark.parametrize("fts_version", ("4", "5"))
def test_fts_tokenize(fresh_db, fts_version):
table_name = "searchable_{}".format(fts_version)
table_name = f"searchable_{fts_version}"
table = fresh_db[table_name]
table.insert_all(search_records)
# Test without porter stemming
table.enable_fts(
["text", "country"],
fts_version="FTS{}".format(fts_version),
fts_version=f"FTS{fts_version}",
)
assert [] == list(table.search("bite"))
# Test WITH stemming
table.disable_fts()
table.enable_fts(
["text", "country"],
fts_version="FTS{}".format(fts_version),
fts_version=f"FTS{fts_version}",
tokenize="porter",
)
rows = list(table.search("bite", order_by="rowid"))
@ -251,10 +254,10 @@ def test_fts_tokenize(fresh_db, fts_version):
def test_optimize_fts(fresh_db):
for fts_version in ("4", "5"):
table_name = "searchable_{}".format(fts_version)
table_name = f"searchable_{fts_version}"
table = fresh_db[table_name]
table.insert_all(search_records)
table.enable_fts(["text", "country"], fts_version="FTS{}".format(fts_version))
table.enable_fts(["text", "country"], fts_version=f"FTS{fts_version}")
# You can call optimize successfully against the tables OR their _fts equivalents:
for table_name in (
"searchable_4",
@ -310,12 +313,12 @@ def test_disable_fts(fresh_db, create_triggers):
expected_triggers = {"searchable_ai", "searchable_ad", "searchable_au"}
else:
expected_triggers = set()
assert expected_triggers == set(
assert expected_triggers == {
r[0]
for r in fresh_db.execute(
"select name from sqlite_master where type = 'trigger'"
).fetchall()
)
}
# Now run .disable_fts() and confirm it worked
table.disable_fts()
assert (
@ -424,7 +427,7 @@ def test_enable_fts_replace(kwargs):
db["books"].enable_fts(**kwargs, replace=True)
# Check that the new configuration is correct
if should_have_changed_columns:
assert db["books_fts"].columns_dict.keys() == set(["title"])
assert db["books_fts"].columns_dict.keys() == {"title"}
if "create_triggers" in kwargs:
assert db["books"].triggers
if "fts_version" in kwargs:
@ -741,6 +744,7 @@ def test_enable_fts_cli_on_view_errors(tmpdir):
db.create_view("v", "select * from t")
db.close()
from click.testing import CliRunner
from sqlite_utils import cli as cli_module
result = CliRunner().invoke(cli_module.cli, ["enable-fts", db_path, "v", "text"])

1
tests/test_get.py
View file

@ -1,4 +1,5 @@
import pytest
from sqlite_utils.db import NotFoundError

5
tests/test_gis.py
View file

@ -1,7 +1,8 @@
import json
import pytest
import pytest
from click.testing import CliRunner
from sqlite_utils.cli import cli
from sqlite_utils.db import Database
from sqlite_utils.utils import find_spatialite, sqlite3
@ -104,7 +105,7 @@ def test_query_load_extension(use_spatialite_shortcut):
[
":memory:",
"select spatialite_version()",
"--load-extension={}".format(load_extension),
f"--load-extension={load_extension}",
],
)
assert result.exit_code == 0, result.stdout

3
tests/test_hypothesis.py
View file

@ -1,5 +1,6 @@
from hypothesis import given
import hypothesis.strategies as st
from hypothesis import given
import sqlite_utils

14
tests/test_insert_files.py
View file

@ -1,10 +1,12 @@
from sqlite_utils import cli, Database
from click.testing import CliRunner
import os
import pathlib
import pytest
import sys
import pytest
from click.testing import CliRunner
from sqlite_utils import Database, cli
@pytest.mark.parametrize("silent", (False, True))
@pytest.mark.parametrize(
@ -44,7 +46,7 @@ def test_insert_files(silent, pk_args, expected_pks):
)
cols = []
for coltype in coltypes:
cols += ["-c", "{}:{}".format(coltype, coltype)]
cols += ["-c", f"{coltype}:{coltype}"]
result = runner.invoke(
cli.cli,
["insert-files", db_path, "files", str(tmpdir)]
@ -142,7 +144,7 @@ def test_insert_files_stdin(use_text, encoding, input, expected):
)
assert result.exit_code == 0, result.stdout
db = Database(db_path)
row = list(db["files"].rows)[0]
row = next(iter(db["files"].rows))
key = "content"
if use_text:
key = "content_text"
@ -167,5 +169,5 @@ def test_insert_files_bad_text_encoding_error():
)
assert result.exit_code == 1, result.output
assert result.output.strip().startswith(
"Error: Could not read file '{}' as text".format(str(latin.resolve()))
f"Error: Could not read file '{latin.resolve()!s}' as text"
)

7
tests/test_introspect.py
View file

@ -1,6 +1,7 @@
from sqlite_utils.db import Index, View, Database, XIndex, XIndexColumn
import pytest
from sqlite_utils.db import Database, Index, View, XIndex, XIndexColumn
def _check_supports_strict():
"""Check if SQLite supports strict tables without leaking the database."""
@ -57,8 +58,8 @@ def test_detect_fts_similar_tables(fresh_db, reverse_order):
fresh_db[table2].insert({"title": "Hello"}).enable_fts(
["title"], fts_version="FTS4"
)
assert fresh_db[table1].detect_fts() == "{}_fts".format(table1)
assert fresh_db[table2].detect_fts() == "{}_fts".format(table2)
assert fresh_db[table1].detect_fts() == f"{table1}_fts"
assert fresh_db[table2].detect_fts() == f"{table2}_fts"
def test_tables(existing_db):

1
tests/test_list_mode.py
View file

@ -3,6 +3,7 @@ Tests for list-based iteration in insert_all and upsert_all
"""
import pytest
from sqlite_utils import Database

3
tests/test_lookup.py
View file

@ -1,6 +1,7 @@
from sqlite_utils.db import Index
import pytest
from sqlite_utils.db import Index
def test_lookup_new_table(fresh_db):
species = fresh_db["species"]

6
tests/test_m2m.py
View file

@ -1,6 +1,7 @@
from sqlite_utils.db import ForeignKey, NoObviousTable
import pytest
from sqlite_utils.db import ForeignKey, NoObviousTable
def test_insert_m2m_single(fresh_db):
dogs = fresh_db["dogs"]
@ -65,8 +66,7 @@ def test_insert_m2m_iterable(fresh_db):
iterable_records = ({"id": 1, "name": "Phineas"}, {"id": 2, "name": "Ferb"})
def iterable():
for record in iterable_records:
yield record
yield from iterable_records
platypuses = fresh_db["platypuses"]
platypuses.insert({"id": 1, "name": "Perry"}, pk="id").m2m(

8
tests/test_migrations.py
View file

@ -1,4 +1,5 @@
import pytest
import sqlite_utils
from sqlite_utils import Migrations
@ -154,10 +155,9 @@ def test_non_transactional_migration_allows_vacuum(tmpdir):
def test_apply_composes_inside_outer_transaction(migrations):
db = sqlite_utils.Database(memory=True)
with pytest.raises(ZeroDivisionError):
with db.atomic():
migrations.apply(db)
raise ZeroDivisionError
with pytest.raises(ZeroDivisionError), db.atomic():
migrations.apply(db)
raise ZeroDivisionError
# The outer transaction rolled back, taking the migrations with it
assert db.table_names() == []

13
tests/test_plugins.py
View file

@ -1,9 +1,12 @@
from click.testing import CliRunner
import click
import importlib
import pytest
import sqlite3
import sys
from sqlite_utils import cli, Database, hookimpl, plugins
import click
import pytest
from click.testing import CliRunner
from sqlite_utils import Database, cli, hookimpl, plugins
def _supports_pragma_function_list():
@ -11,7 +14,7 @@ def _supports_pragma_function_list():
try:
db.execute("select * from pragma_function_list()")
return True
except Exception:
except sqlite3.DatabaseError:
return False
finally:
db.close()

3
tests/test_query.py
View file

@ -1,6 +1,7 @@
import pytest
import types
import pytest
from sqlite_utils.utils import sqlite3

6
tests/test_recipes.py
View file

@ -1,7 +1,9 @@
import json
import pytest
from sqlite_utils import recipes
from sqlite_utils.utils import sqlite3
import json
import pytest
@pytest.fixture

6
tests/test_recreate.py
View file

@ -1,8 +1,10 @@
from sqlite_utils import Database
import sqlite3
import pathlib
import sqlite3
import pytest
from sqlite_utils import Database
def test_recreate_ignored_for_in_memory():
# None of these should raise an exception:

6
tests/test_rows_from_file.py
View file

@ -1,7 +1,9 @@
from sqlite_utils.utils import rows_from_file, Format, RowError
from io import BytesIO, StringIO
import pytest
from sqlite_utils.utils import Format, RowError, rows_from_file
@pytest.mark.parametrize(
"input,expected_format",
@ -29,7 +31,7 @@ def test_rows_from_file_detect_format(input, expected_format):
)
def test_rows_from_file_extra_fields_strategies(ignore_extras, extras_key, expected):
try:
rows, format = rows_from_file(
rows, _format = rows_from_file(
BytesIO(b"id,name\r\n1,Cleo,oops"),
format=Format.CSV,
ignore_extras=ignore_extras,

6
tests/test_sniff.py
View file

@ -1,7 +1,9 @@
from sqlite_utils import cli, Database
from click.testing import CliRunner
import pathlib
import pytest
from click.testing import CliRunner
from sqlite_utils import Database, cli
sniff_dir = pathlib.Path(__file__).parent / "sniff"

4
tests/test_suggest_column_types.py
View file

@ -1,5 +1,7 @@
import pytest
from collections import OrderedDict
import pytest
from sqlite_utils.utils import suggest_column_types

54
tests/test_tracer.py
View file

@ -53,16 +53,18 @@ def test_with_tracer():
assert len(collected) == 4
assert collected == [
(
"SELECT name FROM sqlite_master\n"
" WHERE rootpage = 0\n"
" AND (\n"
" sql LIKE :like\n"
" OR sql LIKE :like2\n"
" OR (\n"
" tbl_name = :table\n"
" AND sql LIKE '%VIRTUAL TABLE%USING FTS%'\n"
" )\n"
" )",
(
"SELECT name FROM sqlite_master\n"
" WHERE rootpage = 0\n"
" AND (\n"
" sql LIKE :like\n"
" OR sql LIKE :like2\n"
" OR (\n"
" tbl_name = :table\n"
" AND sql LIKE '%VIRTUAL TABLE%USING FTS%'\n"
" )\n"
" )"
),
{
"like": "%VIRTUAL TABLE%USING FTS%content=[dogs]%",
"like2": '%VIRTUAL TABLE%USING FTS%content="dogs"%',
@ -72,21 +74,23 @@ def test_with_tracer():
("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'
")\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',
(
'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"},
),
]

33
tests/test_transform.py
View file

@ -1,8 +1,9 @@
import sqlite3
import pytest
from sqlite_utils.db import ForeignKey, TransactionError, TransformError
from sqlite_utils.utils import OperationalError
import pytest
@pytest.mark.parametrize(
@ -113,7 +114,7 @@ def test_transform_sql_table_with_primary_key(
if use_pragma_foreign_keys:
fresh_db.conn.execute("PRAGMA foreign_keys=ON")
dogs.insert({"id": 1, "name": "Cleo", "age": "5"}, pk="id")
sql = dogs.transform_sql(**{**params, **{"tmp_suffix": "suffix"}})
sql = dogs.transform_sql(**{**params, "tmp_suffix": "suffix"})
assert sql == expected_sql
# Check that .transform() runs without exceptions:
with fresh_db.tracer(tracer):
@ -186,7 +187,7 @@ def test_transform_sql_table_with_no_primary_key(
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"}})
sql = dogs.transform_sql(**{**params, "tmp_suffix": "suffix"})
assert sql == expected_sql
# Check that .transform() runs without exceptions:
with fresh_db.tracer(tracer):
@ -476,23 +477,22 @@ def test_transform_in_transaction_refuses_destructive_on_delete(fresh_db, on_del
# keys inside an open transaction would fire those actions when the old
# table is dropped - transform() should refuse instead
fresh_db.conn.execute("PRAGMA foreign_keys=ON")
fresh_db.executescript("""
fresh_db.executescript(f"""
CREATE TABLE authors (id INTEGER PRIMARY KEY, name TEXT);
CREATE TABLE books (
id INTEGER PRIMARY KEY,
title TEXT,
author_id INTEGER REFERENCES authors(id) ON DELETE {}
author_id INTEGER REFERENCES authors(id) ON DELETE {on_delete}
);
""".format(on_delete))
""")
fresh_db["authors"].insert({"id": 1, "name": "Ursula K. Le Guin"})
fresh_db["books"].insert({"id": 1, "title": "The Dispossessed", "author_id": 1})
previous_schema = fresh_db["authors"].schema
with fresh_db.atomic():
with pytest.raises(TransactionError) as excinfo:
fresh_db["authors"].transform(rename={"name": "author_name"})
with fresh_db.atomic(), pytest.raises(TransactionError) as excinfo:
fresh_db["authors"].transform(rename={"name": "author_name"})
message = str(excinfo.value)
assert "books" in message
assert "ON DELETE {}".format(on_delete.upper()) in message
assert f"ON DELETE {on_delete.upper()}" in message
# Nothing should have changed
assert fresh_db["authors"].schema == previous_schema
assert list(fresh_db["books"].rows) == [
@ -518,9 +518,8 @@ def test_transform_in_transaction_refuses_self_referential_cascade(fresh_db):
{"id": 2, "name": "Science Fiction", "parent_id": 1},
]
)
with fresh_db.atomic():
with pytest.raises(TransactionError) as excinfo:
fresh_db["categories"].transform(rename={"name": "title"})
with fresh_db.atomic(), pytest.raises(TransactionError) as excinfo:
fresh_db["categories"].transform(rename={"name": "title"})
assert "categories" in str(excinfo.value)
assert fresh_db["categories"].count == 2
@ -715,15 +714,15 @@ def test_transform_preserves_rowids(fresh_db, table_type):
# Now delete and insert a row to mix up the `rowid` sequence
fresh_db["places"].delete_where("id = ?", ["2"])
fresh_db["places"].insert({"id": "4", "name": "London", "country": "UK"})
previous_rows = list(
previous_rows = [
tuple(row) for row in fresh_db.execute("select rowid, id, name from places")
)
]
# Transform it
fresh_db["places"].transform(column_order=("country", "name"))
# Should be the same
next_rows = list(
next_rows = [
tuple(row) for row in fresh_db.execute("select rowid, id, name from places")
)
]
assert previous_rows == next_rows

2
tests/test_update.py
View file

@ -43,7 +43,7 @@ def test_update_compound_pk_table(fresh_db):
)
def test_update_invalid_pk(fresh_db, pk, update_pk):
table = fresh_db["table"]
table.insert({"id1": 5, "id2": 3, "v": 1}, pk=pk).last_pk
table.insert({"id1": 5, "id2": 3, "v": 1}, pk=pk)
with pytest.raises(NotFoundError):
table.update(update_pk, {"v": 2})

5
tests/test_upsert.py
View file

@ -1,7 +1,8 @@
from sqlite_utils.db import PrimaryKeyRequired
from sqlite_utils import Database
import pytest
from sqlite_utils import Database
from sqlite_utils.db import PrimaryKeyRequired
@pytest.mark.parametrize("use_old_upsert", (False, True))
def test_upsert(use_old_upsert):

6
tests/test_utils.py
View file

@ -1,8 +1,10 @@
from sqlite_utils import utils
import csv
import io
import pytest
from sqlite_utils import utils
@pytest.mark.parametrize(
"input,expected,should_be_is",
@ -57,7 +59,7 @@ def test_maximize_csv_field_size_limit():
# Reset to default in case other tests have changed it
csv.field_size_limit(utils.ORIGINAL_CSV_FIELD_SIZE_LIMIT)
long_value = "a" * 131073
long_csv = "id,text\n1,{}".format(long_value)
long_csv = f"id,text\n1,{long_value}"
fp = io.BytesIO(long_csv.encode("utf-8"))
# Using rows_from_file should error
with pytest.raises(csv.Error):

32
tests/test_wal.py
View file

@ -1,4 +1,5 @@
import pytest
from sqlite_utils import Database
from sqlite_utils.db import TransactionError
@ -11,7 +12,7 @@ def db_path_tmpdir(tmpdir):
def test_enable_disable_wal(db_path_tmpdir):
db, path, tmpdir = db_path_tmpdir
db, _path, tmpdir = db_path_tmpdir
assert len(tmpdir.listdir()) == 1
assert "delete" == db.journal_mode
assert "test.db-wal" not in [f.basename for f in tmpdir.listdir()]
@ -25,12 +26,11 @@ def test_enable_disable_wal(db_path_tmpdir):
def test_enable_wal_inside_transaction_raises(db_path_tmpdir):
db, path, tmpdir = db_path_tmpdir
db, _path, _tmpdir = db_path_tmpdir
db["test"].insert({"id": 1}, pk="id")
with pytest.raises(TransactionError):
with db.atomic():
db["test"].insert({"id": 2}, pk="id")
db.enable_wal()
with pytest.raises(TransactionError), db.atomic():
db["test"].insert({"id": 2}, pk="id")
db.enable_wal()
# The atomic() block must have rolled back cleanly and the
# journal mode must be unchanged
assert db.journal_mode == "delete"
@ -38,19 +38,18 @@ def test_enable_wal_inside_transaction_raises(db_path_tmpdir):
def test_disable_wal_inside_transaction_raises(db_path_tmpdir):
db, path, tmpdir = db_path_tmpdir
db, _path, _tmpdir = db_path_tmpdir
db.enable_wal()
db["test"].insert({"id": 1}, pk="id")
with pytest.raises(TransactionError):
with db.atomic():
db["test"].insert({"id": 2}, pk="id")
db.disable_wal()
with pytest.raises(TransactionError), db.atomic():
db["test"].insert({"id": 2}, pk="id")
db.disable_wal()
assert db.journal_mode == "wal"
assert [r["id"] for r in db["test"].rows] == [1]
def test_ensure_autocommit_on(db_path_tmpdir):
db, path, tmpdir = db_path_tmpdir
db, _path, _tmpdir = db_path_tmpdir
previous_isolation_level = db.conn.isolation_level
assert previous_isolation_level is not None
with db.ensure_autocommit_on():
@ -63,7 +62,7 @@ def test_ensure_autocommit_on(db_path_tmpdir):
def test_enable_wal_noop_inside_transaction_is_allowed(db_path_tmpdir):
# Calling enable_wal() when WAL is already enabled is a no-op,
# so it is fine inside a transaction
db, path, tmpdir = db_path_tmpdir
db, _path, _tmpdir = db_path_tmpdir
db.enable_wal()
with db.atomic():
db["test"].insert({"id": 1}, pk="id")
@ -75,13 +74,12 @@ def test_ensure_autocommit_on_inside_transaction_raises(db_path_tmpdir):
# Setting isolation_level commits any pending transaction as a side
# effect, silently breaking the caller's rollback guarantee - so
# entering autocommit mode with a transaction open is an error
db, path, tmpdir = db_path_tmpdir
db, _path, _tmpdir = db_path_tmpdir
db["test"].insert({"id": 1}, pk="id")
db.begin()
db.execute("insert into test (id) values (2)")
with pytest.raises(TransactionError):
with db.ensure_autocommit_on():
pass
with pytest.raises(TransactionError), db.ensure_autocommit_on():
pass
# The transaction is still open and can still be rolled back
assert db.conn.in_transaction
db.rollback()