publish 5.0.0.post7

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

diff --git a/test/python/mock_icu_word_table.py b/test/python/mock_icu_word_table.py
index cde5e770990dd9655e9708d611ae4d070b5675ed..b26025a05a285cd6be0aa4fba2acdb8c27c7ba8e 100644 (file)
--- a/test/python/mock_icu_word_table.py
+++ b/test/python/mock_icu_word_table.py
@@ -1,7 +1,15 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2025 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 MockIcuWordTable:
     """ A word table for testing using legacy word table structure.
 
 class MockIcuWordTable:
     """ A word table for testing using legacy word table structure.


@@ -17,17 +25,23 @@ class MockIcuWordTable:
 
         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, type, word, info)
+                           VALUES(%s, %s, 'W', %s, '{}'::jsonb)""",
+                        (word_id, word or word_token, 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, type, word, info)
                               VALUES (%s, 'S', %s,
     def add_special(self, word_token, word, cls, typ, oper):
         with self.conn.cursor() as cur:
             cur.execute("""INSERT INTO word (word_token, type, word, info)
                               VALUES (%s, 'S', %s,

-                                      json_build_object('class', %s,
-                                                        'type', %s,
-                                                        'op', %s))
+                                      json_build_object('class', %s::text,
+                                                        'type', %s::text,
+                                                        'op', %s::text))

                         """, (word_token, word, cls, typ, oper))
         self.conn.commit()
 
                         """, (word_token, word, cls, typ, oper))
         self.conn.commit()
 

-

     def add_country(self, country_code, word_token):
         with self.conn.cursor() as cur:
             cur.execute("""INSERT INTO word (word_token, type, word)
     def add_country(self, country_code, word_token):
         with self.conn.cursor() as cur:
             cur.execute("""INSERT INTO word (word_token, type, word)


@@ -35,7 +49,6 @@ class MockIcuWordTable:
                         (word_token, country_code))
         self.conn.commit()
 
                         (word_token, country_code))
         self.conn.commit()
 

-

     def add_postcode(self, word_token, postcode):
         with self.conn.cursor() as cur:
             cur.execute("""INSERT INTO word (word_token, type, word)
     def add_postcode(self, word_token, postcode):
         with self.conn.cursor() as cur:
             cur.execute("""INSERT INTO word (word_token, type, word)


@@ -43,16 +56,32 @@ class MockIcuWordTable:
                         """, (word_token, postcode))
         self.conn.commit()
 
                         """, (word_token, postcode))
         self.conn.commit()
 

-
-    def count(self):
+    def add_housenumber(self, word_id, word_tokens, word=None):

         with self.conn.cursor() as cur:
         with self.conn.cursor() as cur:

-            return cur.scalar("SELECT count(*) FROM word")
+            if isinstance(word_tokens, str):
+                # old style without analyser
+                cur.execute("""INSERT INTO word (word_id, word_token, type)
+                                  VALUES (%s, %s, 'H')
+                            """, (word_id, word_tokens))
+            else:
+                if word is None:
+                    word = word_tokens[0]
+                for token in word_tokens:
+                    cur.execute("""INSERT INTO word (word_id, word_token, type, word, info)
+                                      VALUES (%s, %s, 'H', %s,
+                                              jsonb_build_object('lookup', %s::text))
+                                """, (word_id, token, word, word_tokens[0]))

 
 

+        self.conn.commit()
+
+    def count(self):
+        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 type = 'S'")
+        return execute_scalar(self.conn, "SELECT count(*) FROM word WHERE type = 'S'")

 
 

+    def count_housenumbers(self):
+        return execute_scalar(self.conn, "SELECT count(*) FROM word WHERE type = 'H'")

 
     def get_special(self):
         with self.conn.cursor() as cur:
 
     def get_special(self):
         with self.conn.cursor() as cur:


@@ -62,7 +91,6 @@ class MockIcuWordTable:
             assert len(result) == cur.rowcount, "Word table has duplicates."
             return result
 
             assert len(result) == cur.rowcount, "Word table has duplicates."
             return result
 

-

     def get_country(self):
         with self.conn.cursor() as cur:
             cur.execute("SELECT word, word_token FROM word WHERE type = 'C'")
     def get_country(self):
         with self.conn.cursor() as cur:
             cur.execute("SELECT word, word_token FROM word WHERE type = 'C'")


@@ -70,15 +98,12 @@ class MockIcuWordTable:
             assert len(result) == cur.rowcount, "Word table has duplicates."
             return result
 
             assert len(result) == cur.rowcount, "Word table has duplicates."
             return result
 

-

     def get_postcodes(self):
         with self.conn.cursor() as cur:
             cur.execute("SELECT word FROM word WHERE type = 'P'")
             return set((row[0] for row in cur))
 
     def get_postcodes(self):
         with self.conn.cursor() as cur:
             cur.execute("SELECT word FROM word WHERE type = 'P'")
             return set((row[0] for row in cur))
 

-

     def get_partial_words(self):
         with self.conn.cursor() as cur:
             cur.execute("SELECT word_token, info FROM word WHERE type ='w'")
             return set(((row[0], row[1]['count']) for row in cur))
     def get_partial_words(self):
         with self.conn.cursor() as cur:
             cur.execute("SELECT word_token, info FROM word WHERE type ='w'")
             return set(((row[0], row[1]['count']) for row in cur))

-