Start work on tests for FTS backend

src/calibre/db/backend.py

@@ -928,6 +928,14 @@ class DB:
         from .fts.connect import FTS
         self.fts = FTS(self.get_connection)
 
+    def enable_fts(self, enabled=True):
+        if enabled != self.prefs['fts_enabled']:
+            self.prefs['fts_enabled'] = enabled
+            self.initialize_fts()
+            if self.fts is not None:
+                self.fts.dirty_existing()
+        return self.fts
+
     def get_connection(self):
         return self.conn
 

src/calibre/db/fts/connect.py

@@ -3,15 +3,22 @@
 # License: GPL v3 Copyright: 2022, Kovid Goyal <kovid at kovidgoyal.net>
 
 
+import builtins
 import os
-from .schema_upgrade import SchemaUpgrade
+import sys
 
+from .schema_upgrade import SchemaUpgrade
 
 # TODO: db dump+restore
 # TODO: calibre export/import
 # TODO: check library and vacuuming of fts db
 
 
+def print(*args, **kwargs):
+    kwargs['file'] = sys.__stdout__
+    builtins.print(*args, **kwargs)
+
+
 class FTS:
 
     def __init__(self, get_connection):
@@ -21,3 +28,14 @@ class FTS:
         self.dbpath = os.path.join(os.path.dirname(main_db_path), 'full-text-search.db')
         conn.execute(f'ATTACH DATABASE "{self.dbpath}" AS fts_db')
         SchemaUpgrade(conn)
+
+    def dirty_existing(self):
+        conn = self.get_connection()
+        conn.execute('''
+            INSERT OR IGNORE INTO fts_db.dirtied_formats(book, format)
+            SELECT book, format FROM main.data;
+        ''')
+
+    def all_currently_dirty(self):
+        conn = self.get_connection()
+        return conn.get('''SELECT book, format from fts_db.dirtied_formats''', all=True)

src/calibre/db/tests/fts_api.py

@@ -0,0 +1,45 @@
+#!/usr/bin/env python
+# License: GPLv3 Copyright: 2021, Kovid Goyal <kovid at kovidgoyal.net>
+
+
+import builtins
+import sys
+
+from calibre.db.tests.base import BaseTest
+
+
+def print(*args, **kwargs):
+    kwargs['file'] = sys.__stdout__
+    builtins.print(*args, **kwargs)
+
+
+class FTSAPITest(BaseTest):
+    ae = BaseTest.assertEqual
+
+    def setUp(self):
+        super().setUp()
+        from calibre_extensions.sqlite_extension import set_ui_language
+        set_ui_language('en')
+
+    def tearDown(self):
+        super().tearDown()
+        from calibre_extensions.sqlite_extension import set_ui_language
+        set_ui_language('en')
+
+    def test_fts_triggers(self):
+        cache = self.init_cache()
+        fts = cache.backend.enable_fts()
+        cd = fts.all_currently_dirty()
+        self.ae(len(cd), 3)
+        fts.dirty_existing()
+        self.ae(len(cd), 3)
+
+
+def find_tests():
+    import unittest
+    return unittest.defaultTestLoader.loadTestsFromTestCase(FTSAPITest)
+
+
+def run_tests():
+    from calibre.utils.run_tests import run_tests
+    run_tests(find_tests)