+# 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
from table_compare import NominatimID, DBRow
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)
-
- # XXX use tool function as soon as it is ported
- with context.db.cursor() as cur:
- with (context.nominatim.src_dir / 'lib-sql' / 'postcode_tables.sql').open('r') as fd:
- cur.execute(fd.read())
- cur.execute("""
- INSERT INTO location_postcode
- (place_id, indexed_status, country_code, postcode, geometry)
- SELECT nextval('seq_place'), 1, country_code,
- upper(trim (both ' ' from address->'postcode')) as pc,
- ST_Centroid(ST_Collect(ST_Centroid(geometry)))
- FROM placex
- WHERE address ? 'postcode' AND address->'postcode' NOT SIMILAR TO '%(,|;)%'
- AND geometry IS NOT null
- GROUP BY country_code, pc""")
-
- # Call directly as the refresh function does not include postcodes.
- indexer.LOG.setLevel(logging.ERROR)
- indexer.Indexer(context.nominatim.get_libpq_dsn(), 1).index_full(analyse=False)
+ context.nominatim.run_nominatim('import', '--continue', 'load-data',
+ '--index-noanalyse', '-q',
+ '--offline')
check_database_integrity(context)
+ # Remove the output of the input, when all was right. Otherwise it will be
+ # output when there are errors that had nothing to do with the import
+ # itself.
+ context.log_capture.buffer.clear()
+
@when("updating places")
def update_place_table(context):
""" Update the place table with the given data. Also runs all triggers
context.nominatim.reindex_placex(context.db)
check_database_integrity(context)
+ # Remove the output of the input, when all was right. Otherwise it will be
+ # output when there are errors that had nothing to do with the import
+ # itself.
+ context.log_capture.buffer.clear()
+
+
@when("updating postcodes")
def update_postcodes(context):
""" Rerun the calculation of postcodes.
context.nominatim.reindex_placex(context.db)
+ # Remove the output of the input, when all was right. Otherwise it will be
+ # output when there are errors that had nothing to do with the import
+ # itself.
+ context.log_capture.buffer.clear()
+
################################ THEN ##################################
@then("(?P<table>placex|place) contains(?P<exact> exactly)?")
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 != 'legacy':
+ 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