test/python/mock_icu_word_table.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2022 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 MockIcuWordTable:
  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 NOT NULL,
  20                                               type text NOT NULL,
  21                                               word text,
  22                                               info jsonb)""")
  23
  24         conn.commit()
  25
  26     def add_full_word(self, word_id, word, word_token=None):
  27         with self.conn.cursor() as cur:
  28             cur.execute("""INSERT INTO word (word_id, word_token, type, word, info)
  29                            VALUES(%s, %s, 'W', %s, '{}'::jsonb)""",
  30                         (word_id, word or word_token, word))
  31         self.conn.commit()
  32
  33
  34     def add_special(self, word_token, word, cls, typ, oper):
  35         with self.conn.cursor() as cur:
  36             cur.execute("""INSERT INTO word (word_token, type, word, info)
  37                               VALUES (%s, 'S', %s,
  38                                       json_build_object('class', %s,
  39                                                         'type', %s,
  40                                                         'op', %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, type, word)
  48                            VALUES(%s, 'C', %s)""",
  49                         (word_token, country_code))
  50         self.conn.commit()
  51
  52
  53     def add_postcode(self, word_token, postcode):
  54         with self.conn.cursor() as cur:
  55             cur.execute("""INSERT INTO word (word_token, type, word)
  56                               VALUES (%s, 'P', %s)
  57                         """, (word_token, postcode))
  58         self.conn.commit()
  59
  60
  61     def add_housenumber(self, word_id, word_token):
  62         with self.conn.cursor() as cur:
  63             cur.execute("""INSERT INTO word (word_id, word_token, type)
  64                               VALUES (%s, %s, 'H')
  65                         """, (word_id, word_token))
  66         self.conn.commit()
  67
  68
  69     def count(self):
  70         with self.conn.cursor() as cur:
  71             return cur.scalar("SELECT count(*) FROM word")
  72
  73
  74     def count_special(self):
  75         with self.conn.cursor() as cur:
  76             return cur.scalar("SELECT count(*) FROM word WHERE type = 'S'")
  77
  78
  79     def count_housenumbers(self):
  80         with self.conn.cursor() as cur:
  81             return cur.scalar("SELECT count(*) FROM word WHERE type = 'H'")
  82
  83
  84     def get_special(self):
  85         with self.conn.cursor() as cur:
  86             cur.execute("SELECT word_token, info, word FROM word WHERE type = 'S'")
  87             result = set(((row[0], row[2], row[1]['class'],
  88                            row[1]['type'], row[1]['op']) for row in cur))
  89             assert len(result) == cur.rowcount, "Word table has duplicates."
  90             return result
  91
  92
  93     def get_country(self):
  94         with self.conn.cursor() as cur:
  95             cur.execute("SELECT word, word_token FROM word WHERE type = 'C'")
  96             result = set((tuple(row) for row in cur))
  97             assert len(result) == cur.rowcount, "Word table has duplicates."
  98             return result
  99
 100
 101     def get_postcodes(self):
 102         with self.conn.cursor() as cur:
 103             cur.execute("SELECT word FROM word WHERE type = 'P'")
 104             return set((row[0] for row in cur))
 105
 106
 107     def get_partial_words(self):
 108         with self.conn.cursor() as cur:
 109             cur.execute("SELECT word_token, info FROM word WHERE type ='w'")
 110             return set(((row[0], row[1]['count']) for row in cur))
 111