test/bdd/steps/steps_db_ops.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2025 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 from itertools import chain
   8
   9 import psycopg
  10 from psycopg import sql as pysql
  11
  12 from place_inserter import PlaceColumn
  13 from table_compare import NominatimID, DBRow
  14
  15 from nominatim_db.tokenizer import factory as tokenizer_factory
  16
  17
  18 def check_database_integrity(context):
  19     """ Check some generic constraints on the tables.
  20     """
  21     with context.db.cursor(row_factory=psycopg.rows.tuple_row) as cur:
  22         # place_addressline should not have duplicate (place_id, address_place_id)
  23         cur.execute("""SELECT count(*) FROM
  24                         (SELECT place_id, address_place_id, count(*) as c
  25                          FROM place_addressline GROUP BY place_id, address_place_id) x
  26                        WHERE c > 1""")
  27         assert cur.fetchone()[0] == 0, "Duplicates found in place_addressline"
  28
  29         # word table must not have empty word_tokens
  30         cur.execute("SELECT count(*) FROM word WHERE word_token = ''")
  31         assert cur.fetchone()[0] == 0, "Empty word tokens found in word table"
  32
  33 # GIVEN ##################################
  34
  35
  36 @given("the (?P<named>named )?places")
  37 def add_data_to_place_table(context, named):
  38     """ Add entries into the place table. 'named places' makes sure that
  39         the entries get a random name when none is explicitly given.
  40     """
  41     with context.db.cursor() as cur:
  42         cur.execute('ALTER TABLE place DISABLE TRIGGER place_before_insert')
  43         for row in context.table:
  44             PlaceColumn(context).add_row(row, named is not None).db_insert(cur)
  45         cur.execute('ALTER TABLE place ENABLE TRIGGER place_before_insert')
  46
  47
  48 @given("the relations")
  49 def add_data_to_planet_relations(context):
  50     """ Add entries into the osm2pgsql relation middle table. This is needed
  51         for tests on data that looks up members.
  52     """
  53     with context.db.cursor() as cur:
  54         cur.execute("SELECT value FROM osm2pgsql_properties WHERE property = 'db_format'")
  55         row = cur.fetchone()
  56         if row is None or row['value'] == '1':
  57             for r in context.table:
  58                 last_node = 0
  59                 last_way = 0
  60                 parts = []
  61                 if r['members']:
  62                     members = []
  63                     for m in r['members'].split(','):
  64                         mid = NominatimID(m)
  65                         if mid.typ == 'N':
  66                             parts.insert(last_node, int(mid.oid))
  67                             last_node += 1
  68                             last_way += 1
  69                         elif mid.typ == 'W':
  70                             parts.insert(last_way, int(mid.oid))
  71                             last_way += 1
  72                         else:
  73                             parts.append(int(mid.oid))
  74
  75                         members.extend((mid.typ.lower() + mid.oid, mid.cls or ''))
  76                 else:
  77                     members = None
  78
  79                 tags = chain.from_iterable([(h[5:], r[h]) for h in r.headings
  80                                             if h.startswith("tags+")])
  81
  82                 cur.execute("""INSERT INTO planet_osm_rels (id, way_off, rel_off,
  83                                                             parts, members, tags)
  84                                VALUES (%s, %s, %s, %s, %s, %s)""",
  85                             (r['id'], last_node, last_way, parts, members, list(tags)))
  86         else:
  87             for r in context.table:
  88                 if r['members']:
  89                     members = []
  90                     for m in r['members'].split(','):
  91                         mid = NominatimID(m)
  92                         members.append({'ref': mid.oid, 'role': mid.cls or '', 'type': mid.typ})
  93                 else:
  94                     members = []
  95
  96                 tags = {h[5:]: r[h] for h in r.headings if h.startswith("tags+")}
  97
  98                 cur.execute("""INSERT INTO planet_osm_rels (id, tags, members)
  99                                VALUES (%s, %s, %s)""",
 100                             (r['id'], psycopg.types.json.Json(tags),
 101                              psycopg.types.json.Json(members)))
 102
 103
 104 @given("the ways")
 105 def add_data_to_planet_ways(context):
 106     """ Add entries into the osm2pgsql way middle table. This is necessary for
 107         tests on that that looks up node ids in this table.
 108     """
 109     with context.db.cursor() as cur:
 110         cur.execute("SELECT value FROM osm2pgsql_properties WHERE property = 'db_format'")
 111         row = cur.fetchone()
 112         json_tags = row is not None and row['value'] != '1'
 113         for r in context.table:
 114             if json_tags:
 115                 tags = psycopg.types.json.Json({h[5:]: r[h] for h in r.headings
 116                                                 if h.startswith("tags+")})
 117             else:
 118                 tags = list(chain.from_iterable([(h[5:], r[h])
 119                                                  for h in r.headings if h.startswith("tags+")]))
 120             nodes = [int(x.strip()) for x in r['nodes'].split(',')]
 121
 122             cur.execute("INSERT INTO planet_osm_ways (id, nodes, tags) VALUES (%s, %s, %s)",
 123                         (r['id'], nodes, tags))
 124
 125 # WHEN ##################################
 126
 127
 128 @when("importing")
 129 def import_and_index_data_from_place_table(context):
 130     """ Import data previously set up in the place table.
 131     """
 132     context.nominatim.run_nominatim('import', '--continue', 'load-data',
 133                                               '--index-noanalyse', '-q',
 134                                               '--offline')
 135
 136     check_database_integrity(context)
 137
 138     # Remove the output of the input, when all was right. Otherwise it will be
 139     # output when there are errors that had nothing to do with the import
 140     # itself.
 141     context.log_capture.buffer.clear()
 142
 143
 144 @when("updating places")
 145 def update_place_table(context):
 146     """ Update the place table with the given data. Also runs all triggers
 147         related to updates and reindexes the new data.
 148     """
 149     context.nominatim.run_nominatim('refresh', '--functions')
 150     with context.db.cursor() as cur:
 151         for row in context.table:
 152             col = PlaceColumn(context).add_row(row, False)
 153             col.db_delete(cur)
 154             col.db_insert(cur)
 155         cur.execute('SELECT flush_deleted_places()')
 156
 157     context.nominatim.reindex_placex(context.db)
 158     check_database_integrity(context)
 159
 160     # Remove the output of the input, when all was right. Otherwise it will be
 161     # output when there are errors that had nothing to do with the import
 162     # itself.
 163     context.log_capture.buffer.clear()
 164
 165
 166 @when("updating postcodes")
 167 def update_postcodes(context):
 168     """ Rerun the calculation of postcodes.
 169     """
 170     context.nominatim.run_nominatim('refresh', '--postcodes')
 171
 172
 173 @when("marking for delete (?P<oids>.*)")
 174 def delete_places(context, oids):
 175     """ Remove entries from the place table. Multiple ids may be given
 176         separated by commas. Also runs all triggers
 177         related to updates and reindexes the new data.
 178     """
 179     context.nominatim.run_nominatim('refresh', '--functions')
 180     with context.db.cursor() as cur:
 181         cur.execute('TRUNCATE place_to_be_deleted')
 182         for oid in oids.split(','):
 183             NominatimID(oid).query_osm_id(cur, 'DELETE FROM place WHERE {}')
 184         cur.execute('SELECT flush_deleted_places()')
 185
 186     context.nominatim.reindex_placex(context.db)
 187
 188     # Remove the output of the input, when all was right. Otherwise it will be
 189     # output when there are errors that had nothing to do with the import
 190     # itself.
 191     context.log_capture.buffer.clear()
 192
 193 # THEN ##################################
 194
 195
 196 @then("(?P<table>placex|place) contains(?P<exact> exactly)?")
 197 def check_place_contents(context, table, exact):
 198     """ Check contents of place/placex tables. Each row represents a table row
 199         and all data must match. Data not present in the expected table, may
 200         be arbitrary. The rows are identified via the 'object' column which must
 201         have an identifier of the form '<NRW><osm id>[:<class>]'. When multiple
 202         rows match (for example because 'class' was left out and there are
 203         multiple entries for the given OSM object) then all must match. All
 204         expected rows are expected to be present with at least one database row.
 205         When 'exactly' is given, there must not be additional rows in the database.
 206     """
 207     with context.db.cursor() as cur:
 208         expected_content = set()
 209         for row in context.table:
 210             nid = NominatimID(row['object'])
 211             query = """SELECT *, ST_AsText(geometry) as geomtxt,
 212                               ST_GeometryType(geometry) as geometrytype """
 213             if table == 'placex':
 214                 query += ' ,ST_X(centroid) as cx, ST_Y(centroid) as cy'
 215             query += " FROM %s WHERE {}" % (table, )
 216             nid.query_osm_id(cur, query)
 217             assert cur.rowcount > 0, "No rows found for " + row['object']
 218
 219             for res in cur:
 220                 if exact:
 221                     expected_content.add((res['osm_type'], res['osm_id'], res['class']))
 222
 223                 DBRow(nid, res, context).assert_row(row, ['object'])
 224
 225         if exact:
 226             cur.execute(pysql.SQL('SELECT osm_type, osm_id, class from')
 227                         + pysql.Identifier(table))
 228             actual = set([(r['osm_type'], r['osm_id'], r['class']) for r in cur])
 229             assert expected_content == actual, \
 230                    f"Missing entries: {expected_content - actual}\n" \
 231                    f"Not expected in table: {actual - expected_content}"
 232
 233
 234 @then("(?P<table>placex|place) has no entry for (?P<oid>.*)")
 235 def check_place_has_entry(context, table, oid):
 236     """ Ensure that no database row for the given object exists. The ID
 237         must be of the form '<NRW><osm id>[:<class>]'.
 238     """
 239     with context.db.cursor() as cur:
 240         NominatimID(oid).query_osm_id(cur, "SELECT * FROM %s where {}" % table)
 241         assert cur.rowcount == 0, \
 242                "Found {} entries for ID {}".format(cur.rowcount, oid)
 243
 244
 245 @then("search_name contains(?P<exclude> not)?")
 246 def check_search_name_contents(context, exclude):
 247     """ Check contents of place/placex tables. Each row represents a table row
 248         and all data must match. Data not present in the expected table, may
 249         be arbitrary. The rows are identified via the 'object' column which must
 250         have an identifier of the form '<NRW><osm id>[:<class>]'. All
 251         expected rows are expected to be present with at least one database row.
 252     """
 253     tokenizer = tokenizer_factory.get_tokenizer_for_db(context.nominatim.get_test_config())
 254
 255     with tokenizer.name_analyzer() as analyzer:
 256         with context.db.cursor() as cur:
 257             for row in context.table:
 258                 nid = NominatimID(row['object'])
 259                 nid.row_by_place_id(cur, 'search_name',
 260                                     ['ST_X(centroid) as cx', 'ST_Y(centroid) as cy'])
 261                 assert cur.rowcount > 0, "No rows found for " + row['object']
 262
 263                 for res in cur:
 264                     db_row = DBRow(nid, res, context)
 265                     for name, value in zip(row.headings, row.cells):
 266                         if name in ('name_vector', 'nameaddress_vector'):
 267                             items = [x.strip() for x in value.split(',')]
 268                             tokens = analyzer.get_word_token_info(items)
 269
 270                             if not exclude:
 271                                 assert len(tokens) >= len(items), \
 272                                     f"No word entry found for {value}. Entries found: {len(tokens)}"
 273                             for word, token, wid in tokens:
 274                                 if exclude:
 275                                     assert wid not in res[name], \
 276                                         "Found term for {}/{}: {}".format(nid, name, wid)
 277                                 else:
 278                                     assert wid in res[name], \
 279                                         "Missing term for {}/{}: {}".format(nid, name, wid)
 280                         elif name != 'object':
 281                             assert db_row.contains(name, value), db_row.assert_msg(name, value)
 282
 283
 284 @then("search_name has no entry for (?P<oid>.*)")
 285 def check_search_name_has_entry(context, oid):
 286     """ Check that there is noentry in the search_name table for the given
 287         objects. IDs are in format '<NRW><osm id>[:<class>]'.
 288     """
 289     with context.db.cursor() as cur:
 290         NominatimID(oid).row_by_place_id(cur, 'search_name')
 291
 292         assert cur.rowcount == 0, \
 293                "Found {} entries for ID {}".format(cur.rowcount, oid)
 294
 295
 296 @then("location_postcode contains exactly")
 297 def check_location_postcode(context):
 298     """ Check full contents for location_postcode table. Each row represents a table row
 299         and all data must match. Data not present in the expected table, may
 300         be arbitrary. The rows are identified via 'country' and 'postcode' columns.
 301         All rows must be present as excepted and there must not be additional
 302         rows.
 303     """
 304     with context.db.cursor() as cur:
 305         cur.execute("SELECT *, ST_AsText(geometry) as geomtxt FROM location_postcode")
 306         assert cur.rowcount == len(list(context.table)), \
 307             "Postcode table has {cur.rowcount} rows, expected {len(list(context.table))}."
 308
 309         results = {}
 310         for row in cur:
 311             key = (row['country_code'], row['postcode'])
 312             assert key not in results, "Postcode table has duplicate entry: {}".format(row)
 313             results[key] = DBRow((row['country_code'], row['postcode']), row, context)
 314
 315         for row in context.table:
 316             db_row = results.get((row['country'], row['postcode']))
 317             assert db_row is not None, \
 318                 f"Missing row for country '{row['country']}' postcode '{row['postcode']}'."
 319
 320             db_row.assert_row(row, ('country', 'postcode'))
 321
 322
 323 @then("there are(?P<exclude> no)? word tokens for postcodes (?P<postcodes>.*)")
 324 def check_word_table_for_postcodes(context, exclude, postcodes):
 325     """ Check that the tokenizer produces postcode tokens for the given
 326         postcodes. The postcodes are a comma-separated list of postcodes.
 327         Whitespace matters.
 328     """
 329     nctx = context.nominatim
 330     tokenizer = tokenizer_factory.get_tokenizer_for_db(nctx.get_test_config())
 331     with tokenizer.name_analyzer() as ana:
 332         plist = [ana.normalize_postcode(p) for p in postcodes.split(',')]
 333
 334     plist.sort()
 335
 336     with context.db.cursor() as cur:
 337         cur.execute("SELECT word FROM word WHERE type = 'P' and word = any(%s)",
 338                     (plist,))
 339
 340         found = [row['word'] for row in cur]
 341         assert len(found) == len(set(found)), f"Duplicate rows for postcodes: {found}"
 342
 343     if exclude:
 344         assert len(found) == 0, f"Unexpected postcodes: {found}"
 345     else:
 346         assert set(found) == set(plist), \
 347             f"Missing postcodes {set(plist) - set(found)}. Found: {found}"
 348
 349
 350 @then("place_addressline contains")
 351 def check_place_addressline(context):
 352     """ Check the contents of the place_addressline table. Each row represents
 353         a table row and all data must match. Data not present in the expected
 354         table, may be arbitrary. The rows are identified via the 'object' column,
 355         representing the addressee and the 'address' column, representing the
 356         address item.
 357     """
 358     with context.db.cursor() as cur:
 359         for row in context.table:
 360             nid = NominatimID(row['object'])
 361             pid = nid.get_place_id(cur)
 362             apid = NominatimID(row['address']).get_place_id(cur)
 363             cur.execute(""" SELECT * FROM place_addressline
 364                             WHERE place_id = %s AND address_place_id = %s""",
 365                         (pid, apid))
 366             assert cur.rowcount > 0, \
 367                 f"No rows found for place {row['object']} and address {row['address']}."
 368
 369             for res in cur:
 370                 DBRow(nid, res, context).assert_row(row, ('address', 'object'))
 371
 372
 373 @then("place_addressline doesn't contain")
 374 def check_place_addressline_exclude(context):
 375     """ Check that the place_addressline doesn't contain any entries for the
 376         given addressee/address item pairs.
 377     """
 378     with context.db.cursor() as cur:
 379         for row in context.table:
 380             pid = NominatimID(row['object']).get_place_id(cur)
 381             apid = NominatimID(row['address']).get_place_id(cur, allow_empty=True)
 382             if apid is not None:
 383                 cur.execute(""" SELECT * FROM place_addressline
 384                                 WHERE place_id = %s AND address_place_id = %s""",
 385                             (pid, apid))
 386                 assert cur.rowcount == 0, \
 387                     f"Row found for place {row['object']} and address {row['address']}."
 388
 389
 390 @then(r"W(?P<oid>\d+) expands to(?P<neg> no)? interpolation")
 391 def check_location_property_osmline(context, oid, neg):
 392     """ Check that the given way is present in the interpolation table.
 393     """
 394     with context.db.cursor() as cur:
 395         cur.execute("""SELECT *, ST_AsText(linegeo) as geomtxt
 396                        FROM location_property_osmline
 397                        WHERE osm_id = %s AND startnumber IS NOT NULL""",
 398                     (oid, ))
 399
 400         if neg:
 401             assert cur.rowcount == 0, "Interpolation found for way {}.".format(oid)
 402             return
 403
 404         todo = list(range(len(list(context.table))))
 405         for res in cur:
 406             for i in todo:
 407                 row = context.table[i]
 408                 if (int(row['start']) == res['startnumber']
 409                         and int(row['end']) == res['endnumber']):
 410                     todo.remove(i)
 411                     break
 412             else:
 413                 assert False, "Unexpected row " + str(res)
 414
 415             DBRow(oid, res, context).assert_row(row, ('start', 'end'))
 416
 417         assert not todo, f"Unmatched lines in table: {list(context.table[i] for i in todo)}"
 418
 419
 420 @then("location_property_osmline contains(?P<exact> exactly)?")
 421 def check_osmline_contents(context, exact):
 422     """ Check contents of the interpolation table. Each row represents a table row
 423         and all data must match. Data not present in the expected table, may
 424         be arbitrary. The rows are identified via the 'object' column which must
 425         have an identifier of the form '<osm id>[:<startnumber>]'. When multiple
 426         rows match (for example because 'startnumber' was left out and there are
 427         multiple entries for the given OSM object) then all must match. All
 428         expected rows are expected to be present with at least one database row.
 429         When 'exactly' is given, there must not be additional rows in the database.
 430     """
 431     with context.db.cursor() as cur:
 432         expected_content = set()
 433         for row in context.table:
 434             if ':' in row['object']:
 435                 nid, start = row['object'].split(':', 2)
 436                 start = int(start)
 437             else:
 438                 nid, start = row['object'], None
 439
 440             query = """SELECT *, ST_AsText(linegeo) as geomtxt,
 441                               ST_GeometryType(linegeo) as geometrytype
 442                        FROM location_property_osmline WHERE osm_id=%s"""
 443
 444             if ':' in row['object']:
 445                 query += ' and startnumber = %s'
 446                 params = [int(val) for val in row['object'].split(':', 2)]
 447             else:
 448                 params = (int(row['object']), )
 449
 450             cur.execute(query, params)
 451             assert cur.rowcount > 0, "No rows found for " + row['object']
 452
 453             for res in cur:
 454                 if exact:
 455                     expected_content.add((res['osm_id'], res['startnumber']))
 456
 457                 DBRow(nid, res, context).assert_row(row, ['object'])
 458
 459         if exact:
 460             cur.execute('SELECT osm_id, startnumber from location_property_osmline')
 461             actual = set([(r['osm_id'], r['startnumber']) for r in cur])
 462             assert expected_content == actual, \
 463                    f"Missing entries: {expected_content - actual}\n" \
 464                    f"Not expected in table: {actual - expected_content}"