1 # SPDX-License-Identifier: GPL-2.0-only
 
   3 # This file is part of Nominatim. (https://nominatim.org)
 
   5 # Copyright (C) 2022 by the Nominatim developer community.
 
   6 # For a full list of authors see the git log.
 
   8 Legacy word table for testing with functions to prefil and test contents
 
  12 class MockLegacyWordTable:
 
  13     """ A word table for testing using legacy word table structure.
 
  15     def __init__(self, conn):
 
  17         with conn.cursor() as cur:
 
  18             cur.execute("""CREATE TABLE word (word_id INTEGER,
 
  23                                               country_code varchar(2),
 
  24                                               search_name_count INTEGER,
 
  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)
 
  33                         """, (word_id, ' ' + (word_token or word), word))
 
  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))
 
  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))
 
  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))
 
  61         with self.conn.cursor() as cur:
 
  62             return cur.scalar("SELECT count(*) FROM word")
 
  65     def count_special(self):
 
  66         with self.conn.cursor() as cur:
 
  67             return cur.scalar("SELECT count(*) FROM word WHERE class != 'place'")
 
  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."
 
  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."
 
  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))
 
  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))