Merge pull request #3487 from lonvia/port-to-psycopg3

[nominatim.git] / test / python / mock_legacy_word_table.py

diff --git a/test/python/mock_legacy_word_table.py b/test/python/mock_legacy_word_table.py
index 8baf3adc003d191f6f7741c01eb4459e0834692e..d3f81a4db3948bcb95d267676e236d392987bf71 100644 (file)
--- a/test/python/mock_legacy_word_table.py
+++ b/test/python/mock_legacy_word_table.py
@@ -1,7 +1,14 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2024 by the Nominatim developer community.
+# For a full list of authors see the git log.
 """
 Legacy word table for testing with functions to prefil and test contents
 of the table.
 """
+from nominatim_db.db.connection import execute_scalar
 
 class MockLegacyWordTable:
     """ A word table for testing using legacy word table structure.
@@ -20,6 +27,14 @@ class MockLegacyWordTable:
 
         conn.commit()
 
+    def add_full_word(self, word_id, word, word_token=None):
+        with self.conn.cursor() as cur:
+            cur.execute("""INSERT INTO word (word_id, word_token, word)
+                           VALUES (%s, %s, %s)
+                        """, (word_id, ' ' + (word_token or word), word))
+        self.conn.commit()
+
+
     def add_special(self, word_token, word, cls, typ, oper):
         with self.conn.cursor() as cur:
             cur.execute("""INSERT INTO word (word_token, word, class, type, operator)
@@ -44,18 +59,16 @@ class MockLegacyWordTable:
 
 
     def count(self):
-        with self.conn.cursor() as cur:
-            return cur.scalar("SELECT count(*) FROM word")
+        return execute_scalar(self.conn, "SELECT count(*) FROM word")
 
 
     def count_special(self):
-        with self.conn.cursor() as cur:
-            return cur.scalar("SELECT count(*) FROM word WHERE class != 'place'")
+        return execute_scalar(self.conn, "SELECT count(*) FROM word WHERE class != 'place'")
 
 
     def get_special(self):
         with self.conn.cursor() as cur:
-            cur.execute("""SELECT word_token, word, class, type, operator
+            cur.execute("""SELECT word_token, word, class as cls, type, operator
                            FROM word WHERE class != 'place'""")
             result = set((tuple(row) for row in cur))
             assert len(result) == cur.rowcount, "Word table has duplicates."