From e01943271b17115fbe0e81d523126d2fb1c7c24b Mon Sep 17 00:00:00 2001 From: Amjith Ramanujam Date: Mon, 2 Sep 2019 10:21:26 -0700 Subject: [PATCH] Add tests for escaping table names. --- sqlite_utils/db.py | 15 +++++++-------- tests/test_cli.py | 14 ++++++++++++++ tests/test_enable_fts.py | 37 ++++++++++++++++++++++++++++++++++--- 3 files changed, 55 insertions(+), 11 deletions(-) diff --git a/sqlite_utils/db.py b/sqlite_utils/db.py index 138bf9f..b7837c7 100644 --- a/sqlite_utils/db.py +++ b/sqlite_utils/db.py @@ -646,7 +646,7 @@ class Table(Queryable): return self def drop(self): - self.db.conn.execute("DROP TABLE {}".format(self.name)) + self.db.conn.execute("DROP TABLE [{}]".format(self.name)) def guess_foreign_table(self, column): column = column.lower() @@ -760,21 +760,20 @@ class Table(Queryable): def detect_fts(self): "Detect if table has a corresponding FTS virtual table and return it" - rows = self.db.conn.execute( - """ + sql = """ SELECT name FROM sqlite_master WHERE rootpage = 0 AND ( - sql LIKE '%VIRTUAL TABLE%USING FTS%content="{table}"%' + sql LIKE '%VIRTUAL TABLE%USING FTS%content=%{table}%' OR ( tbl_name = "{table}" AND sql LIKE '%VIRTUAL TABLE%USING FTS%' ) ) """.format( - table=self.name - ) - ).fetchall() + table=self.name + ) + rows = self.db.conn.execute(sql).fetchall() if len(rows) == 0: return None else: @@ -1166,7 +1165,7 @@ class View(Queryable): ) def drop(self): - self.db.conn.execute("DROP VIEW {}".format(self.name)) + self.db.conn.execute("DROP VIEW [{}]".format(self.name)) def chunks(sequence, size): diff --git a/tests/test_cli.py b/tests/test_cli.py index 62fc6c2..ca91bdb 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -315,6 +315,7 @@ def test_index_foreign_keys(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 ] @@ -328,6 +329,19 @@ def test_enable_fts(db_path): assert 0 == result.exit_code assert "Gosh_fts" == Database(db_path)["Gosh"].detect_fts() + # Table names with restricted chars are handled correctly. + # colons and dots are restricted characters for table names. + Database(db_path)["http://example.com"].create({"c1": str, "c2": str, "c3": str}) + assert None == Database(db_path)["http://example.com"].detect_fts() + result = CliRunner().invoke( + cli.cli, ["enable-fts", db_path, "http://example.com", "c1", "--fts4"] + ) + assert 0 == result.exit_code + assert ( + "http://example.com_fts" == Database(db_path)["http://example.com"].detect_fts() + ) + Database(db_path)["http://example.com"].drop() + def test_enable_fts_with_triggers(db_path): Database(db_path)["Gosh"].insert_all([{"c1": "baz"}]) diff --git a/tests/test_enable_fts.py b/tests/test_enable_fts.py index d8d3193..7f58916 100644 --- a/tests/test_enable_fts.py +++ b/tests/test_enable_fts.py @@ -22,6 +22,26 @@ def test_enable_fts(fresh_db): assert [] == table.search("bar") +def test_enable_fts_escape_table_names(fresh_db): + # Table names with restricted chars are handled correctly. + # colons and dots are restricted characters for table names. + table = fresh_db["http://example.com"] + table.insert_all(search_records) + assert ["http://example.com"] == fresh_db.table_names() + table.enable_fts(["text", "country"], fts_version="FTS4") + assert [ + "http://example.com", + "http://example.com_fts", + "http://example.com_fts_segments", + "http://example.com_fts_segdir", + "http://example.com_fts_docsize", + "http://example.com_fts_stat", + ] == fresh_db.table_names() + assert [("tanuki are tricksters", "Japan", "foo")] == table.search("tanuki") + assert [("racoons are trash pandas", "USA", "bar")] == table.search("usa") + assert [] == table.search("bar") + + def test_populate_fts(fresh_db): table = fresh_db["populatable"] table.insert(search_records[0]) @@ -34,6 +54,19 @@ def test_populate_fts(fresh_db): assert [("racoons are trash pandas", "USA", "bar")] == table.search("usa") +def test_populate_fts_escape_table_names(fresh_db): + # Restricted characters such as colon and dots should be escaped. + table = fresh_db["http://example.com"] + table.insert(search_records[0]) + table.enable_fts(["text", "country"], fts_version="FTS4") + assert [] == table.search("trash pandas") + table.insert(search_records[1]) + assert [] == table.search("trash pandas") + # Now run populate_fts to make this record available + table.populate_fts(["text", "country"]) + assert [("racoons are trash pandas", "USA", "bar")] == table.search("usa") + + def test_optimize_fts(fresh_db): for fts_version in ("4", "5"): table_name = "searchable_{}".format(fts_version) @@ -53,9 +86,7 @@ def test_optimize_fts(fresh_db): def test_enable_fts_w_triggers(fresh_db): table = fresh_db["searchable"] table.insert(search_records[0]) - table.enable_fts( - ["text", "country"], fts_version="FTS4", create_triggers=True - ) + table.enable_fts(["text", "country"], fts_version="FTS4", create_triggers=True) assert [("tanuki are tricksters", "Japan", "foo")] == table.search("tanuki") table.insert(search_records[1]) # Triggers will auto-populate FTS virtual table, not need to call populate_fts()