-import base64
-import random
-import string
-import re
-import psycopg2.extras
-
-from check_functions import Almost
-
-class PlaceColumn:
-
- def __init__(self, context, force_name):
- self.columns = { 'admin_level' : 15}
- self.force_name = force_name
- self.context = context
- self.geometry = None
-
- def add(self, key, value):
- if hasattr(self, 'set_key_' + key):
- getattr(self, 'set_key_' + key)(value)
- elif key.startswith('name+'):
- self.add_hstore('name', key[5:], value)
- elif key.startswith('extra+'):
- self.add_hstore('extratags', key[6:], value)
- elif key.startswith('addr+'):
- self.add_hstore('address', key[5:], value)
- elif key in ('name', 'address', 'extratags'):
- self.columns[key] = eval('{' + value + '}')
- else:
- assert key in ('class', 'type')
- self.columns[key] = None if value == '' else value
-
- def set_key_name(self, value):
- self.add_hstore('name', 'name', value)
-
- def set_key_osm(self, value):
- assert value[0] in 'NRW'
- assert value[1:].isdigit()
-
- self.columns['osm_type'] = value[0]
- self.columns['osm_id'] = int(value[1:])
-
- def set_key_admin(self, value):
- self.columns['admin_level'] = int(value)
-
- def set_key_housenr(self, value):
- if value:
- self.add_hstore('address', 'housenumber', value)
-
- def set_key_postcode(self, value):
- if value:
- self.add_hstore('address', 'postcode', value)
-
- def set_key_street(self, value):
- if value:
- self.add_hstore('address', 'street', value)
-
- def set_key_addr_place(self, value):
- if value:
- self.add_hstore('address', 'place', value)
-
- def set_key_country(self, value):
- if value:
- self.add_hstore('address', 'country', value)
-
- def set_key_geometry(self, value):
- self.geometry = self.context.osm.parse_geometry(value, self.context.scene)
- assert self.geometry is not None
-
- def add_hstore(self, column, key, value):
- if column in self.columns:
- self.columns[column][key] = value
- else:
- self.columns[column] = { key : value }
+# 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 db_insert(self, cursor):
- assert 'osm_type' in self.columns
- if self.force_name and 'name' not in self.columns:
- self.add_hstore('name', 'name', ''.join(random.choice(string.printable)
- for _ in range(int(random.random()*30))))
+import psycopg2.extras
- if self.columns['osm_type'] == 'N' and self.geometry is None:
- pt = self.context.osm.grid_node(self.columns['osm_id'])
- if pt is None:
- pt = (random.random()*360 - 180, random.random()*180 - 90)
+from place_inserter import PlaceColumn
+from table_compare import NominatimID, DBRow
- self.geometry = "ST_SetSRID(ST_Point(%f, %f), 4326)" % pt
- else:
- assert self.geometry is not None, "Geometry missing"
- query = 'INSERT INTO place (%s, geometry) values(%s, %s)' % (
- ','.join(self.columns.keys()),
- ','.join(['%s' for x in range(len(self.columns))]),
- self.geometry)
- cursor.execute(query, list(self.columns.values()))
-
-
-class PlaceObjName(object):
-
- def __init__(self, placeid, conn):
- self.pid = placeid
- self.conn = conn
-
- def __str__(self):
- if self.pid is None:
- return "<null>"
-
- if self.pid == 0:
- return "place ID 0"
-
- cur = self.conn.cursor()
- cur.execute("""SELECT osm_type, osm_id, class
- FROM placex WHERE place_id = %s""",
- (self.pid, ))
- assert cur.rowcount == 1, "No entry found for place id %s" % self.pid
-
- return "%s%s:%s" % cur.fetchone()
-
-def compare_place_id(expected, result, column, context):
- if expected == '0':
- assert result == 0, \
- "Bad place id in column {}. Expected: 0, got: {!s}.".format(
- column, PlaceObjName(result, context.db))
- elif expected == '-':
- assert result is None, \
- "Bad place id in column {}: {!s}.".format(
- column, PlaceObjName(result, context.db))
- else:
- assert NominatimID(expected).get_place_id(context.db.cursor()) == result, \
- "Bad place id in column {}. Expected: {}, got: {!s}.".format(
- column, expected, PlaceObjName(result, context.db))
+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.
assert cur.fetchone()[0] == 0, "Duplicates found in place_addressline"
-class NominatimID:
- """ Splits a unique identifier for places into its components.
- As place_ids cannot be used for testing, we use a unique
- identifier instead that is of the form <osmtype><osmid>[:<class>].
- """
-
- id_regex = re.compile(r"(?P<tp>[NRW])(?P<id>\d+)(:(?P<cls>\w+))?")
-
- def __init__(self, oid):
- self.typ = self.oid = self.cls = None
-
- if oid is not None:
- m = self.id_regex.fullmatch(oid)
- assert m is not None, "ID '%s' not of form <osmtype><osmid>[:<class>]" % oid
-
- self.typ = m.group('tp')
- self.oid = m.group('id')
- self.cls = m.group('cls')
-
- def __str__(self):
- if self.cls is None:
- return self.typ + self.oid
-
- return '%s%d:%s' % (self.typ, self.oid, self.cls)
-
- def table_select(self):
- """ Return where clause and parameter list to select the object
- from a Nominatim table.
- """
- where = 'osm_type = %s and osm_id = %s'
- params = [self.typ, self. oid]
-
- if self.cls is not None:
- where += ' and class = %s'
- params.append(self.cls)
-
- return where, params
-
- def get_place_id(self, cur):
- where, params = self.table_select()
- cur.execute("SELECT place_id FROM placex WHERE %s" % where, params)
- assert cur.rowcount == 1, \
- "Expected exactly 1 entry in placex for %s found %s" % (str(self), cur.rowcount)
-
- return cur.fetchone()[0]
-
-
-def assert_db_column(row, column, value, context):
- if column == 'object':
- return
-
- if column.startswith('centroid'):
- if value == 'in geometry':
- query = """SELECT ST_Within(ST_SetSRID(ST_Point({}, {}), 4326),
- ST_SetSRID('{}'::geometry, 4326))""".format(
- row['cx'], row['cy'], row['geomtxt'])
- cur = context.db.cursor()
- cur.execute(query)
- assert cur.fetchone()[0], "(Row %s failed: %s)" % (column, query)
- else:
- fac = float(column[9:]) if column.startswith('centroid*') else 1.0
- x, y = value.split(' ')
- assert Almost(float(x) * fac) == row['cx'], "Bad x coordinate"
- assert Almost(float(y) * fac) == row['cy'], "Bad y coordinate"
- elif column == 'geometry':
- geom = context.osm.parse_geometry(value, context.scene)
- cur = context.db.cursor()
- query = "SELECT ST_Equals(ST_SnapToGrid(%s, 0.00001, 0.00001), ST_SnapToGrid(ST_SetSRID('%s'::geometry, 4326), 0.00001, 0.00001))" % (
- geom, row['geomtxt'],)
- cur.execute(query)
- assert cur.fetchone()[0], "(Row %s failed: %s)" % (column, query)
- elif value == '-':
- assert row[column] is None, "Row %s" % column
- else:
- assert value == str(row[column]), \
- "Row '%s': expected: %s, got: %s" % (column, value, str(row[column]))
-
-
################################ GIVEN ##################################
-@given(u'the scene (?P<scene>.+)')
-def set_default_scene(context, scene):
- context.scene = scene
-
@given("the (?P<named>named )?places")
def add_data_to_place_table(context, named):
+ """ Add entries into the place table. 'named places' makes sure that
+ the entries get a random name when none is explicitly given.
+ """
with context.db.cursor() as cur:
cur.execute('ALTER TABLE place DISABLE TRIGGER place_before_insert')
- for r in context.table:
- col = PlaceColumn(context, named is not None)
-
- for h in r.headings:
- col.add(h, r[h])
-
- col.db_insert(cur)
+ for row in context.table:
+ PlaceColumn(context).add_row(row, named is not None).db_insert(cur)
cur.execute('ALTER TABLE place ENABLE TRIGGER place_before_insert')
@given("the relations")
def add_data_to_planet_relations(context):
+ """ Add entries into the osm2pgsql relation middle table. This is needed
+ for tests on data that looks up members.
+ """
with context.db.cursor() as cur:
for r in context.table:
last_node = 0
else:
members = None
- tags = []
- for h in r.headings:
- if h.startswith("tags+"):
- tags.extend((h[5:], r[h]))
+ tags = chain.from_iterable([(h[5:], r[h]) for h in r.headings if h.startswith("tags+")])
cur.execute("""INSERT INTO planet_osm_rels (id, way_off, rel_off, parts, members, tags)
VALUES (%s, %s, %s, %s, %s, %s)""",
- (r['id'], last_node, last_way, parts, members, tags))
+ (r['id'], last_node, last_way, parts, members, list(tags)))
@given("the ways")
def add_data_to_planet_ways(context):
+ """ Add entries into the osm2pgsql way middle table. This is necessary for
+ tests on that that looks up node ids in this table.
+ """
with context.db.cursor() as cur:
for r in context.table:
- tags = []
- for h in r.headings:
- if h.startswith("tags+"):
- tags.extend((h[5:], r[h]))
-
+ tags = chain.from_iterable([(h[5:], r[h]) for h in r.headings if h.startswith("tags+")])
nodes = [ int(x.strip()) for x in r['nodes'].split(',') ]
cur.execute("INSERT INTO planet_osm_ways (id, nodes, tags) VALUES (%s, %s, %s)",
- (r['id'], nodes, tags))
+ (r['id'], nodes, list(tags)))
################################ WHEN ##################################
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', 'index', 'index-noanalyse')
+ 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):
- context.nominatim.run_setup_script(
- 'create-functions', 'create-partition-functions', 'enable-diff-updates')
+ """ Update the place table with the given data. Also runs all triggers
+ related to updates and reindexes the new data.
+ """
+ context.nominatim.run_nominatim('refresh', '--functions')
with context.db.cursor() as cur:
- for r in context.table:
- col = PlaceColumn(context, False)
-
- for h in r.headings:
- col.add(h, r[h])
-
- col.db_insert(cur)
+ for row in context.table:
+ PlaceColumn(context).add_row(row, False).db_insert(cur)
- while True:
- context.nominatim.run_update_script('index')
+ context.nominatim.reindex_placex(context.db)
+ check_database_integrity(context)
- cur.execute("SELECT 'a' FROM placex WHERE indexed_status != 0 LIMIT 1")
- if cur.rowcount == 0:
- break
+ # 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()
- check_database_integrity(context)
@when("updating postcodes")
def update_postcodes(context):
- context.nominatim.run_update_script('calculate-postcodes')
+ """ Rerun the calculation of postcodes.
+ """
+ context.nominatim.run_nominatim('refresh', '--postcodes')
@when("marking for delete (?P<oids>.*)")
def delete_places(context, oids):
- context.nominatim.run_setup_script(
- 'create-functions', 'create-partition-functions', 'enable-diff-updates')
+ """ Remove entries from the place table. Multiple ids may be given
+ separated by commas. Also runs all triggers
+ related to updates and reindexes the new data.
+ """
+ context.nominatim.run_nominatim('refresh', '--functions')
with context.db.cursor() as cur:
for oid in oids.split(','):
- where, params = NominatimID(oid).table_select()
- cur.execute("DELETE FROM place WHERE " + where, params)
+ NominatimID(oid).query_osm_id(cur, 'DELETE FROM place WHERE {}')
- while True:
- context.nominatim.run_update_script('index')
+ context.nominatim.reindex_placex(context.db)
- with context.db.cursor() as cur:
- cur.execute("SELECT 'a' FROM placex WHERE indexed_status != 0 LIMIT 1")
- if cur.rowcount == 0:
- break
+ # 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("placex contains(?P<exact> exactly)?")
-def check_placex_contents(context, exact):
+@then("(?P<table>placex|place) contains(?P<exact> exactly)?")
+def check_place_contents(context, table, exact):
+ """ Check contents of place/placex tables. 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 '<NRW><osm id>[:<class>]'. When multiple
+ rows match (for example because 'class' 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:
nid = NominatimID(row['object'])
- where, params = nid.table_select()
- cur.execute("""SELECT *, ST_AsText(geometry) as geomtxt,
- ST_X(centroid) as cx, ST_Y(centroid) as cy
- FROM placex where %s""" % where,
- params)
+ query = 'SELECT *, ST_AsText(geometry) as geomtxt, ST_GeometryType(geometry) as geometrytype'
+ if table == 'placex':
+ query += ' ,ST_X(centroid) as cx, ST_Y(centroid) as cy'
+ query += " FROM %s WHERE {}" % (table, )
+ nid.query_osm_id(cur, query)
assert cur.rowcount > 0, "No rows found for " + row['object']
for res in cur:
if exact:
expected_content.add((res['osm_type'], res['osm_id'], res['class']))
- for h in row.headings:
- if h in ('extratags', 'address'):
- if row[h] == '-':
- assert res[h] is None
- else:
- vdict = eval('{' + row[h] + '}')
- assert vdict == res[h]
- elif h.startswith('name'):
- name = h[5:] if h.startswith('name+') else 'name'
- assert name in res['name']
- assert res['name'][name] == row[h]
- elif h.startswith('extratags+'):
- assert res['extratags'][h[10:]] == row[h]
- elif h.startswith('addr+'):
- if row[h] == '-':
- if res['address'] is not None:
- assert h[5:] not in res['address']
- else:
- assert h[5:] in res['address'], "column " + h
- assert res['address'][h[5:]] == row[h], "column %s" % h
- elif h in ('linked_place_id', 'parent_place_id'):
- compare_place_id(row[h], res[h], h, context)
- else:
- assert_db_column(res, h, row[h], context)
+
+ DBRow(nid, res, context).assert_row(row, ['object'])
if exact:
- cur.execute('SELECT osm_type, osm_id, class from placex')
+ 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])
-@then("place contains(?P<exact> exactly)?")
-def check_placex_contents(context, exact):
- with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
- expected_content = set()
- for row in context.table:
- nid = NominatimID(row['object'])
- where, params = nid.table_select()
- cur.execute("""SELECT *, ST_AsText(geometry) as geomtxt,
- ST_GeometryType(geometry) as geometrytype
- FROM place where %s""" % where,
- params)
- assert cur.rowcount > 0, "No rows found for " + row['object']
- for res in cur:
- if exact:
- expected_content.add((res['osm_type'], res['osm_id'], res['class']))
- for h in row.headings:
- msg = "%s: %s" % (row['object'], h)
- if h in ('name', 'extratags', 'address'):
- if row[h] == '-':
- assert res[h] is None, msg
- else:
- vdict = eval('{' + row[h] + '}')
- assert vdict == res[h], msg
- elif h.startswith('name+'):
- assert res['name'][h[5:]] == row[h], msg
- elif h.startswith('extratags+'):
- assert res['extratags'][h[10:]] == row[h], msg
- elif h.startswith('addr+'):
- if row[h] == '-':
- if res['address'] is not None:
- assert h[5:] not in res['address']
- else:
- assert res['address'][h[5:]] == row[h], msg
- elif h in ('linked_place_id', 'parent_place_id'):
- compare_place_id(row[h], res[h], h, context)
- else:
- assert_db_column(res, h, row[h], context)
+@then("(?P<table>placex|place) has no entry for (?P<oid>.*)")
+def check_place_has_entry(context, table, oid):
+ """ Ensure that no database row for the given object exists. The ID
+ must be of the form '<NRW><osm id>[:<class>]'.
+ """
+ with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
+ NominatimID(oid).query_osm_id(cur, "SELECT * FROM %s where {}" % table)
+ assert cur.rowcount == 0, \
+ "Found {} entries for ID {}".format(cur.rowcount, oid)
- if exact:
- cur.execute('SELECT osm_type, osm_id, class from place')
- assert expected_content, set([(r[0], r[1], r[2]) for r in cur])
@then("search_name contains(?P<exclude> not)?")
def check_search_name_contents(context, exclude):
- with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
- for row in context.table:
- pid = NominatimID(row['object']).get_place_id(cur)
- cur.execute("""SELECT *, ST_X(centroid) as cx, ST_Y(centroid) as cy
- FROM search_name WHERE place_id = %s""", (pid, ))
- assert cur.rowcount > 0, "No rows found for " + row['object']
+ """ Check contents of place/placex tables. 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 '<NRW><osm id>[:<class>]'. All
+ expected rows are expected to be present with at least one database row.
+ """
+ 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:
- for h in row.headings:
- if h in ('name_vector', 'nameaddress_vector'):
- terms = [x.strip() for x in row[h].split(',') if not x.strip().startswith('#')]
- words = [x.strip()[1:] for x in row[h].split(',') if x.strip().startswith('#')]
- 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
- """,
- (terms, words))
if not exclude:
- assert subcur.rowcount >= len(terms) + len(words), \
- "No word entry found for " + row[h] + ". Entries found: " + str(subcur.rowcount)
- for wid in subcur:
+ 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 wid[0] not in res[h], "Found term for %s/%s: %s" % (pid, h, wid[1])
+ assert wid not in res[name], \
+ "Found term for {}/{}: {}".format(nid, name, wid)
else:
- assert wid[0] in res[h], "Missing term for %s/%s: %s" % (pid, h, wid[1])
- else:
- assert_db_column(res, h, row[h], context)
+ 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):
+ """ Check that there is noentry in the search_name table for the given
+ objects. IDs are in format '<NRW><osm id>[:<class>]'.
+ """
+ with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
+ NominatimID(oid).row_by_place_id(cur, 'search_name')
+
+ assert cur.rowcount == 0, \
+ "Found {} entries for ID {}".format(cur.rowcount, oid)
@then("location_postcode contains exactly")
def check_location_postcode(context):
+ """ Check full contents for location_postcode 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 'country' and 'postcode' columns.
+ All rows must be present as excepted and there must not be additional
+ rows.
+ """
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 %d rows, expected %d rows." % (cur.rowcount, len(list(context.table)))
+ "Postcode table has {} rows, expected {}.".format(cur.rowcount, len(list(context.table)))
+
+ results = {}
+ for row in cur:
+ key = (row['country_code'], row['postcode'])
+ assert key not in results, "Postcode table has duplicate entry: {}".format(row)
+ results[key] = DBRow((row['country_code'],row['postcode']), row, context)
- table = list(cur)
for row in context.table:
- for i in range(len(table)):
- if table[i]['country_code'] != row['country'] \
- or table[i]['postcode'] != row['postcode']:
- continue
- for h in row.headings:
- if h not in ('country', 'postcode'):
- assert_db_column(table[i], h, row[h], context)
-
-@then("word contains(?P<exclude> not)?")
-def check_word_table(context, exclude):
+ db_row = results.get((row['country'],row['postcode']))
+ assert db_row is not None, \
+ f"Missing row for country '{row['country']}' postcode '{row['postcode']}'."
+
+ db_row.assert_row(row, ('country', 'postcode'))
+
+@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 = []
- values = []
- for h in row.headings:
- wheres.append("%s = %%s" % h)
- values.append(row[h])
- cur.execute("SELECT * from word WHERE %s" % ' AND '.join(wheres), values)
- 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):
+ """ Check the contents of the place_addressline 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,
+ representing the addressee and the 'address' column, representing the
+ address item.
+ """
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
for row in context.table:
- pid = NominatimID(row['object']).get_place_id(cur)
+ nid = NominatimID(row['object'])
+ pid = nid.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""",
"No rows found for place %s and address %s" % (row['object'], row['address'])
for res in cur:
- for h in row.headings:
- if h not in ('address', 'object'):
- assert_db_column(res, h, row[h], context)
+ DBRow(nid, res, context).assert_row(row, ('address', 'object'))
@then("place_addressline doesn't contain")
def check_place_addressline_exclude(context):
+ """ Check that the place_addressline doesn't contain any entries for the
+ given addressee/address item pairs.
+ """
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'])
-
-@then("(?P<oid>\w+) expands to(?P<neg> no)? interpolation")
+ 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):
- nid = NominatimID(oid)
-
- assert 'W' == nid.typ, "interpolation must be a way"
-
+ """ Check that the given way is present in the interpolation table.
+ """
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
cur.execute("""SELECT *, ST_AsText(linegeo) as geomtxt
FROM location_property_osmline
WHERE osm_id = %s AND startnumber IS NOT NULL""",
- (nid.oid, ))
+ (oid, ))
if neg:
- assert cur.rowcount == 0
+ assert cur.rowcount == 0, "Interpolation found for way {}.".format(oid)
return
todo = list(range(len(list(context.table))))
todo.remove(i)
break
else:
- assert False, "Unexpected row %s" % (str(res))
+ assert False, "Unexpected row " + str(res)
- for h in row.headings:
- if h in ('start', 'end'):
- continue
- elif h == 'parent_place_id':
- compare_place_id(row[h], res[h], h, context)
- else:
- assert_db_column(res, h, row[h], context)
+ DBRow(oid, res, context).assert_row(row, ('start', 'end'))
assert not todo
-@then("(?P<table>placex|place) has no entry for (?P<oid>.*)")
-def check_placex_has_entry(context, table, oid):
- with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
- nid = NominatimID(oid)
- where, params = nid.table_select()
- cur.execute("SELECT * FROM %s where %s" % (table, where), params)
- assert cur.rowcount == 0
-
-@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:
- pid = NominatimID(oid).get_place_id(cur)
- cur.execute("SELECT * FROM search_name WHERE place_id = %s", (pid, ))
- assert cur.rowcount == 0
projects / nominatim.git / blobdiff