+# SPDX-License-Identifier: GPL-2.0-only
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2022 by the Nominatim developer community.
+# For a full list of authors see the git log.
+import logging
from itertools import chain
import psycopg2.extras
from place_inserter import PlaceColumn
from table_compare import NominatimID, DBRow
-from nominatim.indexer.indexer import Indexer
+from nominatim.indexer import indexer
+from nominatim.tokenizer import factory as tokenizer_factory
def check_database_integrity(context):
""" Check some generic constraints on the tables.
def import_and_index_data_from_place_table(context):
""" Import data previously set up in the place table.
"""
- context.nominatim.copy_from_place(context.db)
- context.nominatim.run_setup_script('calculate-postcodes')
-
- # Call directly as the refresh function does not include postcodes.
- indexer = Indexer(context.nominatim.test_env['NOMINATIM_DATABASE_DSN'][6:], 1)
- indexer.index_full(analyse=False)
+ context.nominatim.run_nominatim('import', '--continue', 'load-data',
+ '--index-noanalyse', '-q')
check_database_integrity(context)
have an identifier of the form '<NRW><osm id>[:<class>]'. All
expected rows are expected to be present with at least one database row.
"""
- with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
- for row in context.table:
- nid = NominatimID(row['object'])
- nid.row_by_place_id(cur, 'search_name',
- ['ST_X(centroid) as cx', 'ST_Y(centroid) as cy'])
- assert cur.rowcount > 0, "No rows found for " + row['object']
+ tokenizer = tokenizer_factory.get_tokenizer_for_db(context.nominatim.get_test_config())
+
+ with tokenizer.name_analyzer() as analyzer:
+ with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
+ for row in context.table:
+ nid = NominatimID(row['object'])
+ nid.row_by_place_id(cur, 'search_name',
+ ['ST_X(centroid) as cx', 'ST_Y(centroid) as cy'])
+ assert cur.rowcount > 0, "No rows found for " + row['object']
+
+ for res in cur:
+ db_row = DBRow(nid, res, context)
+ for name, value in zip(row.headings, row.cells):
+ if name in ('name_vector', 'nameaddress_vector'):
+ items = [x.strip() for x in value.split(',')]
+ tokens = analyzer.get_word_token_info(items)
- for res in cur:
- db_row = DBRow(nid, res, context)
- for name, value in zip(row.headings, row.cells):
- if name in ('name_vector', 'nameaddress_vector'):
- items = [x.strip() for x in value.split(',')]
- with context.db.cursor() as subcur:
- subcur.execute(""" SELECT word_id, word_token
- FROM word, (SELECT unnest(%s::TEXT[]) as term) t
- WHERE word_token = make_standard_name(t.term)
- and class is null and country_code is null
- and operator is null
- UNION
- SELECT word_id, word_token
- FROM word, (SELECT unnest(%s::TEXT[]) as term) t
- WHERE word_token = ' ' || make_standard_name(t.term)
- and class is null and country_code is null
- and operator is null
- """,
- (list(filter(lambda x: not x.startswith('#'), items)),
- list(filter(lambda x: x.startswith('#'), items))))
if not exclude:
- assert subcur.rowcount >= len(items), \
- "No word entry found for {}. Entries found: {!s}".format(value, subcur.rowcount)
- for wid in subcur:
- present = wid[0] in res[name]
+ assert len(tokens) >= len(items), \
+ "No word entry found for {}. Entries found: {!s}".format(value, len(tokens))
+ for word, token, wid in tokens:
if exclude:
- assert not present, "Found term for {}/{}: {}".format(row['object'], name, wid[1])
+ assert wid not in res[name], \
+ "Found term for {}/{}: {}".format(nid, name, wid)
else:
- assert present, "Missing term for {}/{}: {}".fromat(row['object'], name, wid[1])
- elif name != 'object':
- assert db_row.contains(name, value), db_row.assert_msg(name, value)
+ assert wid in res[name], \
+ "Missing term for {}/{}: {}".format(nid, name, wid)
+ elif name != 'object':
+ assert db_row.contains(name, value), db_row.assert_msg(name, value)
@then("search_name has no entry for (?P<oid>.*)")
def check_search_name_has_entry(context, oid):
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
cur.execute("SELECT *, ST_AsText(geometry) as geomtxt FROM location_postcode")
assert cur.rowcount == len(list(context.table)), \
- "Postcode table has {} rows, expected {}.".foramt(cur.rowcount, len(list(context.table)))
+ "Postcode table has {} rows, expected {}.".format(cur.rowcount, len(list(context.table)))
results = {}
for row in cur:
for row in context.table:
db_row = results.get((row['country'],row['postcode']))
assert db_row is not None, \
- "Missing row for country '{r['country']}' postcode '{r['postcode']}'.".format(r=row)
+ f"Missing row for country '{row['country']}' postcode '{row['postcode']}'."
db_row.assert_row(row, ('country', 'postcode'))
-@then("word contains(?P<exclude> not)?")
-def check_word_table(context, exclude):
- """ Check the contents of the word table. Each row represents a table row
- and all data must match. Data not present in the expected table, may
- be arbitry. The rows are identified via all given columns.
+@then("there are(?P<exclude> no)? word tokens for postcodes (?P<postcodes>.*)")
+def check_word_table_for_postcodes(context, exclude, postcodes):
+ """ Check that the tokenizer produces postcode tokens for the given
+ postcodes. The postcodes are a comma-separated list of postcodes.
+ Whitespace matters.
"""
+ nctx = context.nominatim
+ tokenizer = tokenizer_factory.get_tokenizer_for_db(nctx.get_test_config())
+ with tokenizer.name_analyzer() as ana:
+ plist = [ana.normalize_postcode(p) for p in postcodes.split(',')]
+
+ plist.sort()
+
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
- for row in context.table:
- wheres = ' AND '.join(["{} = %s".format(h) for h in row.headings])
- cur.execute("SELECT * from word WHERE " + wheres, list(row.cells))
- if exclude:
- assert cur.rowcount == 0, "Row still in word table: %s" % '/'.join(values)
- else:
- assert cur.rowcount > 0, "Row not in word table: %s" % '/'.join(values)
+ if nctx.tokenizer == 'icu':
+ cur.execute("SELECT word FROM word WHERE type = 'P' and word = any(%s)",
+ (plist,))
+ else:
+ cur.execute("""SELECT word FROM word WHERE word = any(%s)
+ and class = 'place' and type = 'postcode'""",
+ (plist,))
+
+ found = [row[0] for row in cur]
+ assert len(found) == len(set(found)), f"Duplicate rows for postcodes: {found}"
+
+ if exclude:
+ assert len(found) == 0, f"Unexpected postcodes: {found}"
+ else:
+ assert set(found) == set(plist), \
+ f"Missing postcodes {set(plist) - set(found)}. Found: {found}"
@then("place_addressline contains")
def check_place_addressline(context):
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
for row in context.table:
pid = NominatimID(row['object']).get_place_id(cur)
- apid = NominatimID(row['address']).get_place_id(cur)
- cur.execute(""" SELECT * FROM place_addressline
- WHERE place_id = %s AND address_place_id = %s""",
- (pid, apid))
- assert cur.rowcount == 0, \
- "Row found for place %s and address %s" % (row['object'], row['address'])
+ apid = NominatimID(row['address']).get_place_id(cur, allow_empty=True)
+ if apid is not None:
+ cur.execute(""" SELECT * FROM place_addressline
+ WHERE place_id = %s AND address_place_id = %s""",
+ (pid, apid))
+ assert cur.rowcount == 0, \
+ "Row found for place %s and address %s" % (row['object'], row['address'])
@then("W(?P<oid>\d+) expands to(?P<neg> no)? interpolation")
def check_location_property_osmline(context, oid, neg):
projects / nominatim.git / blobdiff