enable flake for bdd test code

[nominatim.git] / test / bdd / steps / steps_db_ops.py

diff --git a/test/bdd/steps/steps_db_ops.py b/test/bdd/steps/steps_db_ops.py
index 078e29fe7f8e1bd20befbb02686b95d44e6e7005..8b62cbc6b3d6bd06baaf001a7daeadce97c22e05 100644 (file)
--- a/test/bdd/steps/steps_db_ops.py
+++ b/test/bdd/steps/steps_db_ops.py
@@ -1,623 +1,464 @@
-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)
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2025 by the Nominatim developer community.
+# For a full list of authors see the git log.
+from itertools import chain

 
 

-    def set_key_street(self, value):
-        if value:
-            self.add_hstore('address', 'street', value)
+import psycopg
+from psycopg import sql as pysql

 
 

-    def set_key_addr_place(self, value):
-        if value:
-            self.add_hstore('address', 'place', value)
+from place_inserter import PlaceColumn
+from table_compare import NominatimID, DBRow

 
 

-    def set_key_country(self, value):
-        if value:
-            self.add_hstore('address', 'country', value)
+from nominatim_db.tokenizer import factory as tokenizer_factory

 
 

-    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 }
-
-    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))))
-
-        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)
-
-            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 LazyFmt(object):
-
-    def __init__(self, fmtstr, *args):
-        self.fmt = fmtstr
-        self.args = args
-
-    def __str__(self):
-        return self.fmt % self.args
-
-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, \
-            LazyFmt("Bad place id in column %s. Expected: 0, got: %s.",
-                    column, PlaceObjName(result, context.db))
-    elif expected == '-':
-        assert result is None, \
-                LazyFmt("bad place id in column %s: %s.",
-                        column, PlaceObjName(result, context.db))
-    else:
-        assert NominatimID(expected).get_place_id(context.db.cursor()) == result, \
-            LazyFmt("Bad place id in column %s. Expected: %s, got: %s.",
-                    column, expected, PlaceObjName(result, context.db))

 
 def check_database_integrity(context):
     """ Check some generic constraints on the tables.
     """
 
 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"
-
-
-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)
+    with context.db.cursor(row_factory=psycopg.rows.tuple_row) 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"

 
 

-    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]
+        # word table must not have empty word_tokens
+        cur.execute("SELECT count(*) FROM word WHERE word_token = ''")
+        assert cur.fetchone()[0] == 0, "Empty word tokens found in word table"

 
 

-        if self.cls is not None:
-            where += ' and class = %s'
-            params.append(self.cls)
+# GIVEN ##################################

 
 

-        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]))
-
-
-################################ STEPS ##################################
-
-@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):
 
 @given("the (?P<named>named )?places")
 def add_data_to_place_table(context, named):

-    cur = context.db.cursor()
-    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])
+    """ 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 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')

 
 

-        col.db_insert(cur)
-    cur.execute('ALTER TABLE place ENABLE TRIGGER place_before_insert')
-    cur.close()
-    context.db.commit()

 
 @given("the relations")
 def add_data_to_planet_relations(context):
 
 @given("the relations")
 def add_data_to_planet_relations(context):

-    cur = context.db.cursor()
-    for r in context.table:
-        last_node = 0
-        last_way = 0
-        parts = []
-        if r['members']:
-            members = []
-            for m in r['members'].split(','):
-                mid = NominatimID(m)
-                if mid.typ == 'N':
-                    parts.insert(last_node, int(mid.oid))
-                    last_node += 1
-                    last_way += 1
-                elif mid.typ == 'W':
-                    parts.insert(last_way, int(mid.oid))
-                    last_way += 1
+    """ 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:
+        cur.execute("SELECT value FROM osm2pgsql_properties WHERE property = 'db_format'")
+        row = cur.fetchone()
+        if row is None or row['value'] == '1':
+            for r in context.table:
+                last_node = 0
+                last_way = 0
+                parts = []
+                if r['members']:
+                    members = []
+                    for m in r['members'].split(','):
+                        mid = NominatimID(m)
+                        if mid.typ == 'N':
+                            parts.insert(last_node, int(mid.oid))
+                            last_node += 1
+                            last_way += 1
+                        elif mid.typ == 'W':
+                            parts.insert(last_way, int(mid.oid))
+                            last_way += 1
+                        else:
+                            parts.append(int(mid.oid))
+
+                        members.extend((mid.typ.lower() + mid.oid, mid.cls or ''))

                 else:
                 else:

-                    parts.append(int(mid.oid))
+                    members = None
+
+                tags = chain.from_iterable([(h[5:], r[h]) for h in r.headings
+                                            if h.startswith("tags+")])

 
 

-                members.extend((mid.typ.lower() + mid.oid, mid.cls or ''))
+                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, list(tags)))

         else:
         else:

-            members = None
+            for r in context.table:
+                if r['members']:
+                    members = []
+                    for m in r['members'].split(','):
+                        mid = NominatimID(m)
+                        members.append({'ref': mid.oid, 'role': mid.cls or '', 'type': mid.typ})
+                else:
+                    members = []
+
+                tags = {h[5:]: r[h] for h in r.headings if h.startswith("tags+")}

 
 

-        tags = []
-        for h in r.headings:
-            if h.startswith("tags+"):
-                tags.extend((h[5:], r[h]))
+                cur.execute("""INSERT INTO planet_osm_rels (id, tags, members)
+                               VALUES (%s, %s, %s)""",
+                            (r['id'], psycopg.types.json.Json(tags),
+                             psycopg.types.json.Json(members)))

 
 

-        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))
-    context.db.commit()

 
 @given("the ways")
 def add_data_to_planet_ways(context):
 
 @given("the ways")
 def add_data_to_planet_ways(context):

-    cur = context.db.cursor()
-    for r in context.table:
-        tags = []
-        for h in r.headings:
-            if h.startswith("tags+"):
-                tags.extend((h[5:], r[h]))
+    """ 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:
+        cur.execute("SELECT value FROM osm2pgsql_properties WHERE property = 'db_format'")
+        row = cur.fetchone()
+        json_tags = row is not None and row['value'] != '1'
+        for r in context.table:
+            if json_tags:
+                tags = psycopg.types.json.Json({h[5:]: r[h] for h in r.headings
+                                                if h.startswith("tags+")})
+            else:
+                tags = list(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))

 
 

-        nodes = [ int(x.strip()) for x in r['nodes'].split(',') ]
+# WHEN ##################################

 
 

-        cur.execute("INSERT INTO planet_osm_ways (id, nodes, tags) VALUES (%s, %s, %s)",
-                    (r['id'], nodes, tags))
-    context.db.commit()

 
 @when("importing")
 def import_and_index_data_from_place_table(context):
 
 @when("importing")
 def import_and_index_data_from_place_table(context):

-    context.nominatim.run_setup_script('create-functions', 'create-partition-functions')
-    cur = context.db.cursor()
-    cur.execute(
-        """insert into placex (osm_type, osm_id, class, type, name, admin_level, address, extratags, geometry)
-           select              osm_type, osm_id, class, type, name, admin_level, address, extratags, geometry
-           from place where not (class='place' and type='houses' and osm_type='W')""")
-    cur.execute(
-            """insert into location_property_osmline (osm_id, address, linegeo)
-             SELECT osm_id, address, geometry from place
-              WHERE class='place' and type='houses' and osm_type='W'
-                    and ST_GeometryType(geometry) = 'ST_LineString'""")
-    context.db.commit()
-    context.nominatim.run_setup_script('calculate-postcodes', 'index', 'index-noanalyse')
-    check_database_integrity(context)
-
-@when("updating places")
-def update_place_table(context):
-    context.nominatim.run_setup_script(
-        'create-functions', 'create-partition-functions', 'enable-diff-updates')
-    cur = context.db.cursor()
-    for r in context.table:
-        col = PlaceColumn(context, False)
+    """ Import data previously set up in the place table.
+    """
+    context.nominatim.run_nominatim('import', '--continue', 'load-data',
+                                              '--index-noanalyse', '-q',
+                                              '--offline')

 
 

-        for h in r.headings:
-            col.add(h, r[h])
+    check_database_integrity(context)

 
 

-        col.db_insert(cur)
+    # 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()

 
 

-    context.db.commit()

 
 

-    while True:
-        context.nominatim.run_update_script('index')
+@when("updating places")
+def update_place_table(context):
+    """ 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 row in context.table:
+            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)

 
 

-        cur = context.db.cursor()
-        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):
 
 @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):
 
 @when("marking for delete (?P<oids>.*)")
 def delete_places(context, oids):

-    context.nominatim.run_setup_script(
-        'create-functions', 'create-partition-functions', 'enable-diff-updates')
-    cur = context.db.cursor()
-    for oid in oids.split(','):
-        where, params = NominatimID(oid).table_select()
-        cur.execute("DELETE FROM place WHERE " + where, params)
-    context.db.commit()
-
-    while True:
-        context.nominatim.run_update_script('index')
-
-        cur = context.db.cursor()
-        cur.execute("SELECT 'a' FROM placex WHERE indexed_status != 0 LIMIT 1")
-        if cur.rowcount == 0:
-            break
-
-@then("placex contains(?P<exact> exactly)?")
-def check_placex_contents(context, exact):
-    cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
-
-    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)
-        assert cur.rowcount > 0, "No rows found for " + row['object']
+    """ 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:
+        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)?")
+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 arbitrary. 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() as cur:
+        expected_content = set()
+        for row in context.table:
+            nid = NominatimID(row['object'])
+            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']))
+
+                DBRow(nid, res, context).assert_row(row, ['object'])
+
+        if exact:
+            cur.execute(pysql.SQL('SELECT osm_type, osm_id, class from')
+                        + pysql.Identifier(table))
+            actual = set([(r['osm_type'], r['osm_id'], r['class']) for r in cur])
+            assert expected_content == actual, \
+                   f"Missing entries: {expected_content - actual}\n" \
+                   f"Not expected in table: {actual - expected_content}"

 
 

-        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)

 
 

-    if exact:
-        cur.execute('SELECT osm_type, osm_id, class from placex')
-        assert expected_content == set([(r[0], r[1], r[2]) for r in cur])
+@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() 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)

 
 

-    context.db.commit()

 
 

-@then("place contains(?P<exact> exactly)?")
-def check_placex_contents(context, exact):
-    cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
+@then("search_name contains(?P<exclude> not)?")
+def check_search_name_contents(context, exclude):
+    """ 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 arbitrary. 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() 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)
+
+                            if not exclude:
+                                assert len(tokens) >= len(items), \
+                                    f"No word entry found for {value}. Entries found: {len(tokens)}"
+                            for word, token, wid in tokens:
+                                if exclude:
+                                    assert wid not in res[name], \
+                                        "Found term for {}/{}: {}".format(nid, name, wid)
+                                else:
+                                    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)

 
 

-    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("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() as cur:
+        NominatimID(oid).row_by_place_id(cur, 'search_name')

 
 

-    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])
+        assert cur.rowcount == 0, \
+               "Found {} entries for ID {}".format(cur.rowcount, oid)

 
 

-    context.db.commit()

 
 

-@then("search_name contains(?P<exclude> not)?")
-def check_search_name_contents(context, exclude):
-    cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
+@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 arbitrary. 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() as cur:
+        cur.execute("SELECT *, ST_AsText(geometry) as geomtxt FROM location_postcode")
+        assert cur.rowcount == len(list(context.table)), \
+            "Postcode table has {cur.rowcount} rows, expected {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)
+
+        for row in context.table:
+            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(',')]

 
 

-    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']
+    plist.sort()

 
 

-        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('#')]
-                    subcur = context.db.cursor()
-                    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:
-                        if exclude:
-                            assert wid[0] not in res[h], "Found term for %s/%s: %s" % (pid, h, wid[1])
-                        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)
+    with context.db.cursor() as cur:
+        cur.execute("SELECT word FROM word WHERE type = 'P' and word = any(%s)",
+                    (plist,))

 
 

+        found = [row['word'] for row in cur]
+        assert len(found) == len(set(found)), f"Duplicate rows for postcodes: {found}"

 
 

-    context.db.commit()
+    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("location_postcode contains exactly")
-def check_location_postcode(context):
-    cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
-
-    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)))
-
-    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):
-    cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
-
-    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)

 
 @then("place_addressline contains")
 def check_place_addressline(context):
 
 @then("place_addressline contains")
 def check_place_addressline(context):

-    cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
+    """ 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 arbitrary. The rows are identified via the 'object' column,
+        representing the addressee and the 'address' column, representing the
+        address item.
+    """
+    with context.db.cursor() as cur:
+        for row in context.table:
+            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""",
+                        (pid, apid))
+            assert cur.rowcount > 0, \
+                f"No rows found for place {row['object']} and address {row['address']}."

 
 

-    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, \
-                    "No rows found for place %s and address %s" % (row['object'], row['address'])
+            for res in cur:
+                DBRow(nid, res, context).assert_row(row, ('address', 'object'))

 
 

-        for res in cur:
-            for h in row.headings:
-                if h not in ('address', 'object'):
-                    assert_db_column(res, h, row[h], context)
-
-    context.db.commit()

 
 @then("place_addressline doesn't contain")
 def check_place_addressline_exclude(context):
 
 @then("place_addressline doesn't contain")
 def check_place_addressline_exclude(context):

-    cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
-
-    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'])
+    """ Check that the place_addressline doesn't contain any entries for the
+        given addressee/address item pairs.
+    """
+    with context.db.cursor() as cur:
+        for row in context.table:
+            pid = NominatimID(row['object']).get_place_id(cur)
+            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, \
+                    f"Row found for place {row['object']} and address {row['address']}."
+
+
+@then(r"W(?P<oid>\d+) expands to(?P<neg> no)? interpolation")
+def check_location_property_osmline(context, oid, neg):
+    """ Check that the given way is present in the interpolation table.
+    """
+    with context.db.cursor() as cur:
+        cur.execute("""SELECT *, ST_AsText(linegeo) as geomtxt
+                       FROM location_property_osmline
+                       WHERE osm_id = %s AND startnumber IS NOT NULL""",
+                    (oid, ))

 
 

-    context.db.commit()
+        if neg:
+            assert cur.rowcount == 0, "Interpolation found for way {}.".format(oid)
+            return

 
 

-@then("(?P<oid>\w+) expands to(?P<neg> no)? interpolation")
-def check_location_property_osmline(context, oid, neg):
-    cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
-    nid = NominatimID(oid)
-
-    assert 'W' == nid.typ, "interpolation must be a way"
-
-    cur.execute("""SELECT *, ST_AsText(linegeo) as geomtxt
-                   FROM location_property_osmline
-                   WHERE osm_id = %s AND startnumber IS NOT NULL""",
-                (nid.oid, ))
-
-    if neg:
-        assert cur.rowcount == 0
-        return
-
-    todo = list(range(len(list(context.table))))
-    for res in cur:
-        for i in todo:
-            row = context.table[i]
-            if (int(row['start']) == res['startnumber']
-                and int(row['end']) == res['endnumber']):
-                todo.remove(i)
-                break
-        else:
-            assert False, "Unexpected row %s" % (str(res))
+        todo = list(range(len(list(context.table))))
+        for res in cur:
+            for i in todo:
+                row = context.table[i]
+                if (int(row['start']) == res['startnumber']
+                        and int(row['end']) == res['endnumber']):
+                    todo.remove(i)
+                    break
+            else:
+                assert False, "Unexpected row " + str(res)
+
+            DBRow(oid, res, context).assert_row(row, ('start', 'end'))

 
 

-        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)
+        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_osmline_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 arbitrary. 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() as cur:
+        expected_content = set()
+        for row in context.table:
+            if ':' in row['object']:
+                nid, start = row['object'].split(':', 2)
+                start = int(start)

             else:
             else:

-                assert_db_column(res, h, row[h], context)
+                nid, start = row['object'], None

 
 

-    assert not todo
+            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']), )

 
 

-@then("(?P<table>placex|place) has no entry for (?P<oid>.*)")
-def check_placex_has_entry(context, table, oid):
-    cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
-    nid = NominatimID(oid)
-    where, params = nid.table_select()
-    cur.execute("SELECT * FROM %s where %s" % (table, where), params)
-    assert cur.rowcount == 0
-    context.db.commit()
+            cur.execute(query, params)
+            assert cur.rowcount > 0, "No rows found for " + row['object']

 
 

-@then("search_name has no entry for (?P<oid>.*)")
-def check_search_name_has_entry(context, oid):
-    cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
-    pid = NominatimID(oid).get_place_id(cur)
-    cur.execute("SELECT * FROM search_name WHERE place_id = %s", (pid, ))
-    assert cur.rowcount == 0
-    context.db.commit()
+            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['osm_id'], r['startnumber']) for r in cur])
+            assert expected_content == actual, \
+                   f"Missing entries: {expected_content - actual}\n" \
+                   f"Not expected in table: {actual - expected_content}"