+# 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
def check_database_integrity(context):
""" Check some generic constraints on the tables.
"""
- # place_addressline should not have duplicate (place_id, address_place_id)
- cur = context.db.cursor()
- cur.execute("""SELECT count(*) FROM
- (SELECT place_id, address_place_id, count(*) as c
- FROM place_addressline GROUP BY place_id, address_place_id) x
- WHERE c > 1""")
- assert cur.fetchone()[0] == 0, "Duplicates found in place_addressline"
+ with context.db.cursor() as cur:
+ # place_addressline should not have duplicate (place_id, address_place_id)
+ cur.execute("""SELECT count(*) FROM
+ (SELECT place_id, address_place_id, count(*) as c
+ FROM place_addressline GROUP BY place_id, address_place_id) x
+ WHERE c > 1""")
+ assert cur.fetchone()[0] == 0, "Duplicates found in place_addressline"
+
+ # word table must not have empty word_tokens
+ if context.nominatim.tokenizer != 'legacy':
+ cur.execute("SELECT count(*) FROM word WHERE word_token = ''")
+ assert cur.fetchone()[0] == 0, "Empty word tokens found in word table"
+
################################ GIVEN ##################################
def import_and_index_data_from_place_table(context):
""" Import data previously set up in the place table.
"""
- nctx = context.nominatim
-
- tokenizer = tokenizer_factory.create_tokenizer(nctx.get_test_config())
- 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(), tokenizer, 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.run_nominatim('refresh', '--functions')
with context.db.cursor() as cur:
for row in context.table:
- PlaceColumn(context).add_row(row, False).db_insert(cur)
+ col = PlaceColumn(context).add_row(row, False)
+ col.db_delete(cur)
+ col.db_insert(cur)
+ cur.execute('SELECT flush_deleted_places()')
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.run_nominatim('refresh', '--functions')
with context.db.cursor() as cur:
+ cur.execute('TRUNCATE place_to_be_deleted')
for oid in oids.split(','):
NominatimID(oid).query_osm_id(cur, 'DELETE FROM place WHERE {}')
+ cur.execute('SELECT flush_deleted_places()')
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)?")
if exact:
cur.execute('SELECT osm_type, osm_id, class from {}'.format(table))
- assert expected_content == set([(r[0], r[1], r[2]) for r in cur])
+ actual = set([(r[0], r[1], r[2]) for r in cur])
+ assert expected_content == actual, \
+ f"Missing entries: {expected_content - actual}\n" \
+ f"Not expected in table: {actual - expected_content}"
@then("(?P<table>placex|place) has no entry for (?P<oid>.*)")
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'))
plist.sort()
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
- if nctx.tokenizer == 'legacy_icu':
+ if nctx.tokenizer != 'legacy':
cur.execute("SELECT word FROM word WHERE type = 'P' and word = any(%s)",
(plist,))
else:
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):
DBRow(oid, res, context).assert_row(row, ('start', 'end'))
- assert not todo
+ assert not todo, f"Unmatched lines in table: {list(context.table[i] for i in todo)}"
+@then("location_property_osmline contains(?P<exact> exactly)?")
+def check_place_contents(context, exact):
+ """ Check contents of the interpolation 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 the 'object' column which must
+ have an identifier of the form '<osm id>[:<startnumber>]'. When multiple
+ rows match (for example because 'startnumber' was left out and there are
+ multiple entries for the given OSM object) then all must match. All
+ expected rows are expected to be present with at least one database row.
+ When 'exactly' is given, there must not be additional rows in the database.
+ """
+ with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
+ expected_content = set()
+ for row in context.table:
+ if ':' in row['object']:
+ nid, start = row['object'].split(':', 2)
+ start = int(start)
+ else:
+ nid, start = row['object'], None
+
+ query = """SELECT *, ST_AsText(linegeo) as geomtxt,
+ ST_GeometryType(linegeo) as geometrytype
+ FROM location_property_osmline WHERE osm_id=%s"""
+
+ if ':' in row['object']:
+ query += ' and startnumber = %s'
+ params = [int(val) for val in row['object'].split(':', 2)]
+ else:
+ params = (int(row['object']), )
+
+ cur.execute(query, params)
+ assert cur.rowcount > 0, "No rows found for " + row['object']
+
+ for res in cur:
+ if exact:
+ expected_content.add((res['osm_id'], res['startnumber']))
+
+ DBRow(nid, res, context).assert_row(row, ['object'])
+
+ if exact:
+ cur.execute('SELECT osm_id, startnumber from location_property_osmline')
+ actual = set([(r[0], r[1]) for r in cur])
+ assert expected_content == actual, \
+ f"Missing entries: {expected_content - actual}\n" \
+ f"Not expected in table: {actual - expected_content}"
projects / nominatim.git / blobdiff