Documentation for tracer mechanism

This commit is contained in:
Simon Willison 2020-09-07 14:54:07 -07:00
commit f12251092c
12 changed files with 68 additions and 22 deletions

View file

@ -10,7 +10,7 @@ def fresh_db():
@pytest.fixture
def existing_db():
database = Database(memory=True)
database.conn.executescript(
database.executescript(
"""
CREATE TABLE foo (text TEXT);
INSERT INTO foo (text) values ("one");

View file

@ -396,7 +396,7 @@ def test_enable_fts_with_triggers(db_path):
def search(q):
return (
Database(db_path)
.conn.execute("select c1 from Gosh_fts where c1 match ?", [q])
.execute("select c1 from Gosh_fts where c1 match ?", [q])
.fetchall()
)
@ -417,7 +417,7 @@ def test_populate_fts(db_path):
def search(q):
return (
Database(db_path)
.conn.execute("select c1 from Gosh_fts where c1 match ?", [q])
.execute("select c1 from Gosh_fts where c1 match ?", [q])
.fetchall()
)

View file

@ -41,5 +41,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.conn.execute("create table foo (col {})".format(column_def))
fresh_db.execute("create table foo (col {})".format(column_def))
assert {"col": expected_type} == fresh_db["foo"].columns_dict

View file

@ -4,9 +4,9 @@ import pytest
def test_recursive_triggers():
db = Database(memory=True)
assert db.conn.execute("PRAGMA recursive_triggers").fetchone()[0]
assert db.execute("PRAGMA recursive_triggers").fetchone()[0]
def test_recursive_triggers_off():
db = Database(memory=True, recursive_triggers=False)
assert not db.conn.execute("PRAGMA recursive_triggers").fetchone()[0]
assert not db.execute("PRAGMA recursive_triggers").fetchone()[0]

View file

@ -680,7 +680,7 @@ def test_create_index_if_not_exists(fresh_db):
)
def test_insert_dictionaries_and_lists_as_json(fresh_db, data_structure):
fresh_db["test"].insert({"id": 1, "data": data_structure}, pk="id")
row = fresh_db.conn.execute("select id, data from test").fetchone()
row = fresh_db.execute("select id, data from test").fetchone()
assert row[0] == 1
assert data_structure == json.loads(row[1])

View file

@ -4,7 +4,7 @@ from sqlite_utils.utils import OperationalError
def test_create_view(fresh_db):
fresh_db.create_view("bar", "select 1 + 1")
rows = fresh_db.conn.execute("select * from bar").fetchall()
rows = fresh_db.execute("select * from bar").fetchall()
assert [(2,)] == rows
@ -23,7 +23,7 @@ def test_create_view_ignore(fresh_db):
fresh_db.create_view("bar", "select 1 + 1").create_view(
"bar", "select 1 + 2", ignore=True
)
rows = fresh_db.conn.execute("select * from bar").fetchall()
rows = fresh_db.execute("select * from bar").fetchall()
assert [(2,)] == rows
@ -31,13 +31,13 @@ def test_create_view_replace(fresh_db):
fresh_db.create_view("bar", "select 1 + 1").create_view(
"bar", "select 1 + 2", replace=True
)
rows = fresh_db.conn.execute("select * from bar").fetchall()
rows = fresh_db.execute("select * from bar").fetchall()
assert [(3,)] == rows
def test_create_view_replace_with_same_does_nothing(fresh_db):
fresh_db.create_view("bar", "select 1 + 1")
initial_version = fresh_db.conn.execute("PRAGMA schema_version").fetchone()[0]
initial_version = fresh_db.execute("PRAGMA schema_version").fetchone()[0]
fresh_db.create_view("bar", "select 1 + 1", replace=True)
after_version = fresh_db.conn.execute("PRAGMA schema_version").fetchone()[0]
after_version = fresh_db.execute("PRAGMA schema_version").fetchone()[0]
assert after_version == initial_version

View file

@ -160,7 +160,7 @@ def test_disable_fts(fresh_db, create_triggers):
expected_triggers = set()
assert expected_triggers == set(
r[0]
for r in fresh_db.conn.execute(
for r in fresh_db.execute(
"select name from sqlite_master where type = 'trigger'"
).fetchall()
)
@ -168,7 +168,7 @@ def test_disable_fts(fresh_db, create_triggers):
table.disable_fts()
assert (
0
== fresh_db.conn.execute(
== fresh_db.execute(
"select count(*) from sqlite_master where type = 'trigger'"
).fetchone()[0]
)

View file

@ -73,7 +73,7 @@ def test_table_repr(fresh_db):
def test_indexes(fresh_db):
fresh_db.conn.executescript(
fresh_db.executescript(
"""
create table Gosh (c1 text, c2 text, c3 text);
create index Gosh_c1 on Gosh(c1);

View file

@ -11,6 +11,7 @@ def test_tracer():
db["dogs"].enable_fts(["name"])
db["dogs"].search("Cleopaws")
assert collected == [
("PRAGMA recursive_triggers=on;", None),
("select name from sqlite_master where type = 'view'", None),
("select name from sqlite_master where type = 'table'", None),
("CREATE TABLE [dogs] (\n [name] TEXT\n);\n ", None),