Add tests for escaping table names.

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

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

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