X-Git-Url: https://git.openstreetmap.org./nominatim.git/blobdiff_plain/c4f22a42eba499ed1854c5967b2e3932fe9b3896..734dff037405469ada9a226be9c98be583823226:/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 f5d89e4f..e8b4390f 100644 --- a/test/python/mock_icu_word_table.py +++ b/test/python/mock_icu_word_table.py @@ -1,13 +1,14 @@ -# SPDX-License-Identifier: GPL-2.0-only +# SPDX-License-Identifier: GPL-3.0-or-later # # This file is part of Nominatim. (https://nominatim.org) # -# Copyright (C) 2022 by the Nominatim developer community. +# 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 MockIcuWordTable: """ A word table for testing using legacy word table structure. @@ -35,9 +36,9 @@ class MockIcuWordTable: 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() @@ -58,14 +59,34 @@ class MockIcuWordTable: self.conn.commit() - def count(self): + def add_housenumber(self, word_id, word_tokens, word=None): 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): - 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):