]> git.openstreetmap.org Git - nominatim.git/blob - test/python/mock_icu_word_table.py
Merge pull request #2981 from lonvia/add-point-wkb-decoder
[nominatim.git] / 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_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, jsonb_build_object('lookup', %s))
74                                 """, (word_id, token, word, word_tokens[0]))
75
76         self.conn.commit()
77
78
79     def count(self):
80         with self.conn.cursor() as cur:
81             return cur.scalar("SELECT count(*) FROM word")
82
83
84     def count_special(self):
85         with self.conn.cursor() as cur:
86             return cur.scalar("SELECT count(*) FROM word WHERE type = 'S'")
87
88
89     def count_housenumbers(self):
90         with self.conn.cursor() as cur:
91             return cur.scalar("SELECT count(*) FROM word WHERE type = 'H'")
92
93
94     def get_special(self):
95         with self.conn.cursor() as cur:
96             cur.execute("SELECT word_token, info, word FROM word WHERE type = 'S'")
97             result = set(((row[0], row[2], row[1]['class'],
98                            row[1]['type'], row[1]['op']) for row in cur))
99             assert len(result) == cur.rowcount, "Word table has duplicates."
100             return result
101
102
103     def get_country(self):
104         with self.conn.cursor() as cur:
105             cur.execute("SELECT word, word_token FROM word WHERE type = 'C'")
106             result = set((tuple(row) for row in cur))
107             assert len(result) == cur.rowcount, "Word table has duplicates."
108             return result
109
110
111     def get_postcodes(self):
112         with self.conn.cursor() as cur:
113             cur.execute("SELECT word FROM word WHERE type = 'P'")
114             return set((row[0] for row in cur))
115
116
117     def get_partial_words(self):
118         with self.conn.cursor() as cur:
119             cur.execute("SELECT word_token, info FROM word WHERE type ='w'")
120             return set(((row[0], row[1]['count']) for row in cur))
121