bdd: add osm2pgsql tests for updating interpolations

test/bdd/steps/steps_db_ops.py

@@ -380,4 +380,49 @@ def check_location_property_osmline(context, oid, neg):
 
         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}"