Merge pull request #3492 from lonvia/drop-waste-disposal

[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.
 """
 """
 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.
 
 class MockLegacyWordTable:
     """ A word table for testing using legacy word table structure.


@@ -20,6 +27,14 @@ class MockLegacyWordTable:
 
         conn.commit()
 
 
         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)
     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):
 
 
     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):
 
 
     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:
 
 
     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."
                            FROM word WHERE class != 'place'""")
             result = set((tuple(row) for row in cur))
             assert len(result) == cur.rowcount, "Word table has duplicates."