+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2024 by the Nominatim developer community.
+# For a full list of authors see the git log.
"""
Custom mocks for testing.
"""
import itertools
-import psycopg2.extras
-
-from nominatim.db import properties
-
-class MockParamCapture:
- """ Mock that records the parameters with which a function was called
- as well as the number of calls.
- """
- def __init__(self, retval=0):
- self.called = 0
- self.return_value = retval
- self.last_args = None
- self.last_kwargs = None
-
- def __call__(self, *args, **kwargs):
- self.called += 1
- self.last_args = args
- self.last_kwargs = kwargs
- return self.return_value
-
-
-class MockWordTable:
- """ A word table for testing.
- """
- def __init__(self, conn):
- self.conn = conn
- with conn.cursor() as cur:
- cur.execute("""CREATE TABLE word (word_id INTEGER,
- word_token text,
- word text,
- class text,
- type text,
- country_code varchar(2),
- search_name_count INTEGER,
- operator TEXT)""")
-
- conn.commit()
-
- def add_special(self, word_token, word, cls, typ, oper):
- with self.conn.cursor() as cur:
- cur.execute("""INSERT INTO word (word_token, word, class, type, operator)
- VALUES (%s, %s, %s, %s, %s)
- """, (word_token, word, cls, typ, oper))
- self.conn.commit()
-
-
- def add_country(self, country_code, word_token):
- with self.conn.cursor() as cur:
- cur.execute("INSERT INTO word (word_token, country_code) VALUES(%s, %s)",
- (word_token, country_code))
- self.conn.commit()
-
-
- def add_postcode(self, word_token, postcode):
- with self.conn.cursor() as cur:
- cur.execute("""INSERT INTO word (word_token, word, class, type)
- VALUES (%s, %s, 'place', 'postcode')
- """, (word_token, postcode))
- self.conn.commit()
-
-
- def count(self):
- with self.conn.cursor() as cur:
- return cur.scalar("SELECT count(*) FROM word")
-
-
- def count_special(self):
- with self.conn.cursor() as cur:
- return cur.scalar("SELECT count(*) FROM word WHERE class != 'place'")
-
-
- def get_special(self):
- with self.conn.cursor() as cur:
- cur.execute("""SELECT word_token, word, class, type, operator
- FROM word WHERE class != 'place'""")
- result = set((tuple(row) for row in cur))
- assert len(result) == cur.rowcount, "Word table has duplicates."
- return result
-
-
- def get_country(self):
- with self.conn.cursor() as cur:
- cur.execute("""SELECT country_code, word_token
- FROM word WHERE country_code is not null""")
- result = set((tuple(row) for row in cur))
- assert len(result) == cur.rowcount, "Word table has duplicates."
- return result
-
-
- def get_postcodes(self):
- with self.conn.cursor() as cur:
- cur.execute("""SELECT word FROM word
- WHERE class = 'place' and type = 'postcode'""")
- return set((row[0] for row in cur))
-
- def get_partial_words(self):
- with self.conn.cursor() as cur:
- cur.execute("""SELECT word_token, search_name_count FROM word
- WHERE class is null and country_code is null
- and not word_token like ' %'""")
- return set((tuple(row) for row in cur))
+from nominatim_db.db import properties
+# This must always point to the mock word table for the default tokenizer.
+from mock_icu_word_table import MockIcuWordTable as MockWordTable
class MockPlacexTable:
""" A placex table for testing.
admin_level smallint,
address hstore,
extratags hstore,
+ token_info jsonb,
geometry Geometry(Geometry,4326),
wikipedia TEXT,
country_code varchar(2),
def add(self, osm_type='N', osm_id=None, cls='amenity', typ='cafe', names=None,
admin_level=None, address=None, extratags=None, geom='POINT(10 4)',
- country=None):
+ country=None, housenumber=None, rank_search=30):
with self.conn.cursor() as cur:
- psycopg2.extras.register_hstore(cur)
cur.execute("""INSERT INTO placex (place_id, osm_type, osm_id, class,
type, name, admin_level, address,
+ housenumber, rank_search,
extratags, geometry, country_code)
- VALUES(nextval('seq_place'), %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)""",
+ VALUES(nextval('seq_place'), %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)""",
(osm_type, osm_id or next(self.idseq), cls, typ, names,
- admin_level, address, extratags, 'SRID=4326;' + geom,
+ admin_level, address, housenumber, rank_search,
+ extratags, 'SRID=4326;' + geom,
country))
self.conn.commit()
""" Set a property in the table to the given value.
"""
properties.set_property(self.conn, name, value)
+
+
+ def get(self, name):
+ """ Set a property in the table to the given value.
+ """
+ return properties.get_property(self.conn, name)
projects / nominatim.git / blobdiff