test/python/mock_legacy_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) 2024 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
  12 class MockLegacyWordTable:
  13     """ A word table for testing using legacy word table structure.
  14     """
  15     def __init__(self, conn):
  16         self.conn = conn
  17         with conn.cursor() as cur:
  18             cur.execute("""CREATE TABLE word (word_id INTEGER,
  19                                               word_token text,
  20                                               word text,
  21                                               class text,
  22                                               type text,
  23                                               country_code varchar(2),
  24                                               search_name_count INTEGER,
  25                                               operator TEXT)""")
  26
  27         conn.commit()
  28
  29     def add_full_word(self, word_id, word, word_token=None):
  30         with self.conn.cursor() as cur:
  31             cur.execute("""INSERT INTO word (word_id, word_token, word)
  32                            VALUES (%s, %s, %s)
  33                         """, (word_id, ' ' + (word_token or word), word))
  34         self.conn.commit()
  35
  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, word, class, type, operator)
  40                               VALUES (%s, %s, %s, %s, %s)
  41                         """, (word_token, word, cls, typ, oper))
  42         self.conn.commit()
  43
  44
  45     def add_country(self, country_code, word_token):
  46         with self.conn.cursor() as cur:
  47             cur.execute("INSERT INTO word (word_token, country_code) VALUES(%s, %s)",
  48                         (word_token, country_code))
  49         self.conn.commit()
  50
  51
  52     def add_postcode(self, word_token, postcode):
  53         with self.conn.cursor() as cur:
  54             cur.execute("""INSERT INTO word (word_token, word, class, type)
  55                               VALUES (%s, %s, 'place', 'postcode')
  56                         """, (word_token, postcode))
  57         self.conn.commit()
  58
  59
  60     def count(self):
  61         with self.conn.cursor() as cur:
  62             return cur.scalar("SELECT count(*) FROM word")
  63
  64
  65     def count_special(self):
  66         with self.conn.cursor() as cur:
  67             return cur.scalar("SELECT count(*) FROM word WHERE class != 'place'")
  68
  69
  70     def get_special(self):
  71         with self.conn.cursor() as cur:
  72             cur.execute("""SELECT word_token, word, class, type, operator
  73                            FROM word WHERE class != 'place'""")
  74             result = set((tuple(row) for row in cur))
  75             assert len(result) == cur.rowcount, "Word table has duplicates."
  76             return result
  77
  78
  79     def get_country(self):
  80         with self.conn.cursor() as cur:
  81             cur.execute("""SELECT country_code, word_token
  82                            FROM word WHERE country_code is not null""")
  83             result = set((tuple(row) for row in cur))
  84             assert len(result) == cur.rowcount, "Word table has duplicates."
  85             return result
  86
  87
  88     def get_postcodes(self):
  89         with self.conn.cursor() as cur:
  90             cur.execute("""SELECT word FROM word
  91                            WHERE class = 'place' and type = 'postcode'""")
  92             return set((row[0] for row in cur))
  93
  94     def get_partial_words(self):
  95         with self.conn.cursor() as cur:
  96             cur.execute("""SELECT word_token, search_name_count FROM word
  97                            WHERE class is null and country_code is null
  98                                  and not word_token like ' %'""")
  99             return set((tuple(row) for row in cur))
 100