test/python/mock_icu_word_table.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2025 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Legacy word table for testing with functions to prefil and test contents
   9 of the table.
  10 """
  11 from nominatim_db.db.connection import execute_scalar
  12
  13 from psycopg.types.json import Jsonb
  14
  15
  16 class MockIcuWordTable:
  17     """ A word table for testing using legacy word table structure.
  18     """
  19     def __init__(self, conn):
  20         self.conn = conn
  21         with conn.cursor() as cur:
  22             cur.execute("""CREATE TABLE word (word_id INTEGER,
  23                                               word_token text NOT NULL,
  24                                               type text NOT NULL,
  25                                               word text,
  26                                               info jsonb)""")
  27
  28         conn.commit()
  29
  30     def add_full_word(self, word_id, word, word_token=None):
  31         with self.conn.cursor() as cur:
  32             cur.execute("""INSERT INTO word (word_id, word_token, type, word, info)
  33                            VALUES(%s, %s, 'W', %s, '{}'::jsonb)""",
  34                         (word_id, word or word_token, word))
  35         self.conn.commit()
  36
  37     def add_special(self, word_token, word, cls, typ, oper):
  38         with self.conn.cursor() as cur:
  39             cur.execute("""INSERT INTO word (word_token, type, word, info)
  40                               VALUES (%s, 'S', %s,
  41                                       json_build_object('class', %s::text,
  42                                                         'type', %s::text,
  43                                                         'op', %s::text))
  44                         """, (word_token, word, cls, typ, oper))
  45         self.conn.commit()
  46
  47     def add_country(self, country_code, word_token, lookup):
  48         with self.conn.cursor() as cur:
  49             cur.execute("""INSERT INTO word (word_token, type, word, info)
  50                            VALUES(%s, 'C', %s, %s)""",
  51                         (word_token, lookup, Jsonb({'cc': country_code})))
  52         self.conn.commit()
  53
  54     def add_postcode(self, word_token, postcode):
  55         with self.conn.cursor() as cur:
  56             cur.execute("""INSERT INTO word (word_token, type, word)
  57                               VALUES (%s, 'P', %s)
  58                         """, (word_token, postcode))
  59         self.conn.commit()
  60
  61     def add_housenumber(self, word_id, word_tokens, word=None):
  62         with self.conn.cursor() as cur:
  63             if isinstance(word_tokens, str):
  64                 # old style without analyser
  65                 cur.execute("""INSERT INTO word (word_id, word_token, type)
  66                                   VALUES (%s, %s, 'H')
  67                             """, (word_id, word_tokens))
  68             else:
  69                 if word is None:
  70                     word = word_tokens[0]
  71                 for token in word_tokens:
  72                     cur.execute("""INSERT INTO word (word_id, word_token, type, word, info)
  73                                       VALUES (%s, %s, 'H', %s,
  74                                               jsonb_build_object('lookup', %s::text))
  75                                 """, (word_id, token, word, word_tokens[0]))
  76
  77         self.conn.commit()
  78
  79     def count(self):
  80         return execute_scalar(self.conn, "SELECT count(*) FROM word")
  81
  82     def count_special(self):
  83         return execute_scalar(self.conn, "SELECT count(*) FROM word WHERE type = 'S'")
  84
  85     def count_housenumbers(self):
  86         return execute_scalar(self.conn, "SELECT count(*) FROM word WHERE type = 'H'")
  87
  88     def get_special(self):
  89         with self.conn.cursor() as cur:
  90             cur.execute("SELECT word_token, info, word FROM word WHERE type = 'S'")
  91             result = set(((row[0], row[2], row[1]['class'],
  92                            row[1]['type'], row[1]['op']) for row in cur))
  93             assert len(result) == cur.rowcount, "Word table has duplicates."
  94             return result
  95
  96     def get_country(self):
  97         with self.conn.cursor() as cur:
  98             cur.execute("SELECT info->>'cc', word_token, word FROM word WHERE type = 'C'")
  99             result = set((tuple(row) for row in cur))
 100             assert len(result) == cur.rowcount, "Word table has duplicates."
 101             return result
 102
 103     def get_postcodes(self):
 104         with self.conn.cursor() as cur:
 105             cur.execute("SELECT word FROM word WHERE type = 'P'")
 106             return set((row[0] for row in cur))
 107
 108     def get_partial_words(self):
 109         with self.conn.cursor() as cur:
 110             cur.execute("SELECT word_token, info FROM word WHERE type ='w'")
 111             return set(((row[0], row[1]['count']) for row in cur))