test/bdd/test_db.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) 2026 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Collector for BDD import acceptance tests.
   9
  10 These tests check the Nominatim import chain after the osm2pgsql import.
  11 """
  12 import re
  13 from collections import defaultdict
  14
  15 import psycopg
  16 import psycopg.sql as pysql
  17
  18 import pytest
  19 from pytest_bdd import when, then, given
  20 from pytest_bdd.parsers import re as step_parse
  21
  22 from utils.place_inserter import PlaceColumn
  23 from utils.checks import check_table_content
  24 from utils.geometry_alias import ALIASES
  25
  26 from nominatim_db.config import Configuration
  27 from nominatim_db import cli
  28 from nominatim_db.tools.database_import import load_data, create_table_triggers
  29 from nominatim_db.tools.postcodes import update_postcodes
  30 from nominatim_db.tokenizer import factory as tokenizer_factory
  31 from nominatim_db.utils.asyncio_utils import asyncio_run
  32
  33
  34 def _rewrite_placeid_field(field, new_field, datatable, place_ids):
  35     try:
  36         oidx = datatable[0].index(field)
  37         datatable[0][oidx] = new_field
  38         for line in datatable[1:]:
  39             line[oidx] = None if line[oidx] == '-' else place_ids[line[oidx]]
  40     except ValueError:
  41         pass
  42
  43
  44 def _collect_place_ids(conn):
  45     pids = {}
  46     with conn.cursor() as cur:
  47         for row in cur.execute('SELECT place_id, osm_type, osm_id, class FROM placex'):
  48             pids[f"{row[1]}{row[2]}"] = row[0]
  49             pids[f"{row[1]}{row[2]}:{row[3]}"] = row[0]
  50
  51     return pids
  52
  53
  54 @pytest.fixture
  55 def row_factory(db_conn):
  56     def _insert_row(table, **data):
  57         columns = []
  58         placeholders = []
  59         values = []
  60         for k, v in data.items():
  61             columns.append(pysql.Identifier(k))
  62             if isinstance(v, tuple):
  63                 placeholders.append(pysql.SQL(v[0]))
  64                 values.append(v[1])
  65             elif isinstance(v, (pysql.Literal, pysql.SQL)):
  66                 placeholders.append(v)
  67             else:
  68                 placeholders.append(pysql.Placeholder())
  69                 values.append(v)
  70
  71         sql = pysql.SQL("INSERT INTO {table} ({columns}) VALUES({values})")\
  72                    .format(table=pysql.Identifier(table),
  73                            columns=pysql.SQL(',').join(columns),
  74                            values=pysql.SQL(',').join(placeholders))
  75
  76         db_conn.execute(sql, values)
  77         db_conn.commit()
  78
  79     return _insert_row
  80
  81
  82 @pytest.fixture
  83 def test_config_env(pytestconfig):
  84     dbname = pytestconfig.getini('nominatim_test_db')
  85
  86     config = Configuration(None).get_os_env()
  87     config['NOMINATIM_DATABASE_DSN'] = f"pgsql:dbname={dbname}"
  88     config['NOMINATIM_LANGUAGES'] = 'en,de,fr,ja'
  89     config['NOMINATIM_USE_US_TIGER_DATA'] = 'yes'
  90     if pytestconfig.option.NOMINATIM_TOKENIZER is not None:
  91         config['NOMINATIM_TOKENIZER'] = pytestconfig.option.NOMINATIM_TOKENIZER
  92
  93     return config
  94
  95
  96 @pytest.fixture
  97 def update_config(def_config):
  98     """ Prepare the database for being updatable and return the config.
  99     """
 100     cli.nominatim(['refresh', '--functions'], def_config.environ)
 101
 102     return def_config
 103
 104
 105 @given(step_parse('the (?P<named>named )?places'), target_fixture=None)
 106 def import_places(db_conn, named, datatable, node_grid):
 107     """ Insert todo rows into the place table.
 108         When 'named' is given, then a random name will be generated for all
 109         objects.
 110     """
 111     with db_conn.cursor() as cur:
 112         for row in datatable[1:]:
 113             PlaceColumn(node_grid).add_row(datatable[0], row, named is not None).db_insert(cur)
 114
 115
 116 @given(step_parse('the entrances'), target_fixture=None)
 117 def import_place_entrances(row_factory, datatable, node_grid):
 118     """ Insert todo rows into the place_entrance table.
 119     """
 120     for row in datatable[1:]:
 121         data = PlaceColumn(node_grid).add_row(datatable[0], row, False)
 122         assert data.columns['osm_type'] == 'N'
 123
 124         params = {'osm_id': data.columns['osm_id'],
 125                   'type': data.columns['type'],
 126                   'extratags': data.columns.get('extratags'),
 127                   'geometry': pysql.SQL(data.get_wkt())}
 128
 129         row_factory('place_entrance', **params)
 130
 131
 132 @given(step_parse('the interpolations'), target_fixture=None)
 133 def import_place_interpolations(row_factory, datatable, node_grid):
 134     """ Insert todo rows into the place_entrance table.
 135     """
 136     for row in datatable[1:]:
 137         data = PlaceColumn(node_grid).add_row(datatable[0], row, False)
 138         assert data.columns['osm_type'] == 'W'
 139
 140         params = {'osm_id': data.columns['osm_id'],
 141                   'type': data.columns['type'],
 142                   'address': data.columns.get('address'),
 143                   'nodes': [int(x) for x in data.columns['nodes'].split(',')],
 144                   'geometry': pysql.SQL(data.get_wkt())}
 145
 146         row_factory('place_interpolation', **params)
 147
 148
 149 @given(step_parse('the postcodes'), target_fixture=None)
 150 def import_place_postcode(db_conn, datatable, node_grid):
 151     """ Insert todo rows into the place_postcode table. If a row for the
 152         requested object already exists it is overwritten.
 153     """
 154     with db_conn.cursor() as cur:
 155         for row in datatable[1:]:
 156             data = defaultdict(lambda: None)
 157             data.update((k, v) for k, v in zip(datatable[0], row))
 158
 159             if data['centroid'].startswith('country:'):
 160                 ccode = data['centroid'][8:].upper()
 161                 data['centroid'] = 'srid=4326;POINT({} {})'.format(*ALIASES[ccode])
 162             else:
 163                 data['centroid'] = f"srid=4326;{node_grid.geometry_to_wkt(data['centroid'])}"
 164
 165             data['osm_type'] = data['osm'][0]
 166             data['osm_id'] = data['osm'][1:]
 167
 168             if 'geometry' in data:
 169                 geom = f"'srid=4326;{node_grid.geometry_to_wkt(data['geometry'])}'::geometry"
 170             else:
 171                 geom = 'null'
 172
 173             cur.execute(""" DELETE FROM place_postcode
 174                             WHERE osm_type = %(osm_type)s and osm_id = %(osm_id)s""",
 175                         data)
 176             cur.execute(f"""INSERT INTO place_postcode
 177                             (osm_type, osm_id, country_code, postcode, centroid, geometry)
 178                             VALUES (%(osm_type)s, %(osm_id)s,
 179                                     %(country)s, %(postcode)s,
 180                                     %(centroid)s, {geom})""", data)
 181     db_conn.commit()
 182
 183
 184 @given('the ways', target_fixture=None)
 185 def import_ways(row_factory, datatable):
 186     """ Import raw ways into the osm2pgsql way middle table.
 187     """
 188     id_idx = datatable[0].index('id')
 189     node_idx = datatable[0].index('nodes')
 190     for line in datatable[1:]:
 191         row_factory('planet_osm_ways',
 192                     id=line[id_idx],
 193                     nodes=[int(x) for x in line[node_idx].split(',')],
 194                     tags=psycopg.types.json.Json(
 195                         {k[5:]: v for k, v in zip(datatable[0], line)
 196                          if k.startswith("tags+")}))
 197
 198
 199 @given('the relations', target_fixture=None)
 200 def import_rels(row_factory, datatable):
 201     """ Import raw relations into the osm2pgsql relation middle table.
 202         Also populates place_associated_street for associatedStreet relations.
 203     """
 204     id_idx = datatable[0].index('id')
 205     memb_idx = datatable[0].index('members')
 206     for line in datatable[1:]:
 207         raw_tags = {k[5:]: v for k, v in zip(datatable[0], line)
 208                     if k.startswith("tags+")}
 209         members = []
 210         if line[memb_idx]:
 211             for member in line[memb_idx].split(','):
 212                 m = re.fullmatch(r'\s*([RWN])(\d+)(?::(\S+))?\s*', member)
 213                 if not m:
 214                     raise ValueError(f'Illegal member {member}.')
 215                 members.append({'ref': int(m[2]), 'role': m[3] or '', 'type': m[1]})
 216
 217         row_factory('planet_osm_rels',
 218                     id=int(line[id_idx]), tags=psycopg.types.json.Json(raw_tags),
 219                     members=psycopg.types.json.Json(members))
 220
 221         # Mirror associatedStreet data into the dedicated table.
 222         if raw_tags.get('type') == 'associatedStreet':
 223             for mem in members:
 224                 row_factory('place_associated_street',
 225                             relation_id=int(line[id_idx]),
 226                             member_type=mem['type'],
 227                             member_id=mem['ref'],
 228                             member_role=mem['role'])
 229
 230
 231 @when('importing', target_fixture='place_ids')
 232 def do_import(db_conn, def_config):
 233     """ Run a reduced version of the Nominatim import.
 234     """
 235     create_table_triggers(db_conn, def_config)
 236
 237     asyncio_run(load_data(def_config.get_libpq_dsn(), 1))
 238
 239     tokenizer = tokenizer_factory.get_tokenizer_for_db(def_config)
 240     update_postcodes(def_config.get_libpq_dsn(), None, tokenizer)
 241     cli.nominatim(['index', '-q'], def_config.environ)
 242
 243     return _collect_place_ids(db_conn)
 244
 245
 246 @when('updating places', target_fixture='place_ids')
 247 def do_update(db_conn, update_config, node_grid, datatable):
 248     """ Update the place table with the given data. Also runs all triggers
 249         related to updates and reindexes the new data.
 250     """
 251     with db_conn.cursor() as cur:
 252         for row in datatable[1:]:
 253             PlaceColumn(node_grid).add_row(datatable[0], row, False).db_insert(cur)
 254         cur.execute('SELECT flush_deleted_places()')
 255     db_conn.commit()
 256
 257     cli.nominatim(['index', '-q'], update_config.environ)
 258
 259     return _collect_place_ids(db_conn)
 260
 261
 262 @when('updating relations', target_fixture=None)
 263 def do_update_relations(db_conn, update_config, datatable):
 264     """ Update associatedStreet relations in place_associated_street.
 265         Simulates what osm2pgsql flex does: DELETE all existing rows for the
 266         relation then re-INSERT the new member list.
 267     """
 268     id_idx = datatable[0].index('id')
 269     memb_idx = datatable[0].index('members')
 270     with db_conn.cursor() as cur:
 271         for line in datatable[1:]:
 272             relation_id = int(line[id_idx])
 273             raw_tags = {k[5:]: v for k, v in zip(datatable[0], line)
 274                         if k.startswith("tags+")}
 275             # DELETE all old rows – trigger invalidates any house members.
 276             cur.execute('DELETE FROM place_associated_street WHERE relation_id = %s',
 277                         (relation_id,))
 278             # Re-INSERT the new member list when the relation is still live.
 279             if raw_tags.get('type') == 'associatedStreet' and line[memb_idx]:
 280                 for member in line[memb_idx].split(','):
 281                     m = re.fullmatch(r'\s*([RWN])(\d+)(?::(\S+))?\s*', member)
 282                     if not m:
 283                         raise ValueError(f'Illegal member {member}.')
 284                     cur.execute(
 285                         """INSERT INTO place_associated_street
 286                                (relation_id, member_type, member_id, member_role)
 287                                VALUES (%s, %s, %s, %s)""",
 288                         (relation_id, m[1], int(m[2]), m[3] or ''))
 289     db_conn.commit()
 290     cli.nominatim(['index', '-q'], update_config.environ)
 291
 292
 293 @when('updating entrances', target_fixture=None)
 294 def update_place_entrances(db_conn, datatable, node_grid):
 295     """ Update rows in the place_entrance table.
 296     """
 297     with db_conn.cursor() as cur:
 298         for row in datatable[1:]:
 299             data = PlaceColumn(node_grid).add_row(datatable[0], row, False)
 300             assert data.columns['osm_type'] == 'N'
 301
 302             cur.execute("DELETE FROM place_entrance WHERE osm_id = %s",
 303                         (data.columns['osm_id'],))
 304             cur.execute("""INSERT INTO place_entrance (osm_id, type, extratags, geometry)
 305                            VALUES (%s, %s, %s, {})""".format(data.get_wkt()),
 306                         (data.columns['osm_id'], data.columns['type'],
 307                          data.columns.get('extratags')))
 308     db_conn.commit()
 309
 310
 311 @when('updating interpolations', target_fixture=None)
 312 def update_place_interpolations(db_conn, row_factory, update_config, datatable, node_grid):
 313     """ Update rows in the place_entrance table.
 314     """
 315     for row in datatable[1:]:
 316         data = PlaceColumn(node_grid).add_row(datatable[0], row, False)
 317         assert data.columns['osm_type'] == 'W'
 318
 319         params = {'osm_id': data.columns['osm_id'],
 320                   'type': data.columns['type'],
 321                   'address': data.columns.get('address'),
 322                   'nodes': [int(x) for x in data.columns['nodes'].split(',')],
 323                   'geometry': pysql.SQL(data.get_wkt())}
 324
 325         row_factory('place_interpolation', **params)
 326
 327     db_conn.execute('SELECT flush_deleted_places()')
 328     db_conn.commit()
 329
 330     cli.nominatim(['index', '-q', '--minrank', '30'], update_config.environ)
 331
 332
 333 @when('refreshing postcodes')
 334 def do_postcode_update(update_config):
 335     """ Recompute the postcode centroids.
 336     """
 337     cli.nominatim(['refresh', '--postcodes'], update_config.environ)
 338
 339
 340 @when(step_parse(r'marking for delete (?P<otype>[NRW])(?P<oid>\d+)'),
 341       converters={'oid': int})
 342 def do_delete_place(db_conn, update_config, node_grid, otype, oid):
 343     """ Remove the given place from the database.
 344     """
 345     with db_conn.cursor() as cur:
 346         cur.execute('TRUNCATE place_to_be_deleted')
 347         cur.execute('DELETE FROM place WHERE osm_type = %s and osm_id = %s',
 348                     (otype, oid))
 349         cur.execute('DELETE FROM place_interpolation WHERE osm_id = %s',
 350                     (oid, ))
 351         cur.execute('SELECT flush_deleted_places()')
 352         if otype == 'N':
 353             cur.execute('DELETE FROM place_entrance WHERE osm_id = %s',
 354                         (oid, ))
 355         cur.execute('DELETE FROM place_postcode WHERE osm_type = %s and osm_id = %s',
 356                     (otype, oid))
 357     db_conn.commit()
 358
 359     cli.nominatim(['index', '-q'], update_config.environ)
 360
 361
 362 @then(step_parse(r'(?P<table>\w+) contains(?P<exact> exactly)?'))
 363 def then_check_table_content(db_conn, place_ids, datatable, node_grid, table, exact):
 364     _rewrite_placeid_field('object', 'place_id', datatable, place_ids)
 365     _rewrite_placeid_field('parent_place_id', 'parent_place_id', datatable, place_ids)
 366     _rewrite_placeid_field('linked_place_id', 'linked_place_id', datatable, place_ids)
 367     if table == 'place_addressline':
 368         _rewrite_placeid_field('address', 'address_place_id', datatable, place_ids)
 369
 370     for i, title in enumerate(datatable[0]):
 371         if title.startswith('addr+'):
 372             datatable[0][i] = f"address+{title[5:]}"
 373
 374     check_table_content(db_conn, table, datatable, grid=node_grid, exact=bool(exact))
 375
 376
 377 @then(step_parse(r'(DISABLED?P<table>placex?) has no entry for (?P<oid>[NRW]\d+(?::\S+)?)'))
 378 def then_check_place_missing_lines(db_conn, place_ids, table, oid):
 379     assert oid in place_ids
 380
 381     sql = pysql.SQL("""SELECT count(*) FROM {}
 382                        WHERE place_id = %s""").format(pysql.Identifier(tablename))
 383
 384     with conn.cursor(row_factory=tuple_row) as cur:
 385         assert cur.execute(sql, [place_ids[oid]]).fetchone()[0] == 0
 386
 387
 388 @then(step_parse(r'W(?P<oid>\d+) expands to interpolation'),
 389       converters={'oid': int})
 390 def then_check_interpolation_table(db_conn, node_grid, place_ids, oid, datatable):
 391     with db_conn.cursor() as cur:
 392         cur.execute('SELECT count(*) FROM location_property_osmline WHERE osm_id = %s',
 393                     [oid])
 394         assert cur.fetchone()[0] == len(datatable) - 1
 395
 396     converted = [['osm_id', 'startnumber', 'endnumber', 'linegeo!wkt']]
 397     start_idx = datatable[0].index('start') if 'start' in datatable[0] else None
 398     end_idx = datatable[0].index('end') if 'end' in datatable[0] else None
 399     geom_idx = datatable[0].index('geometry') if 'geometry' in datatable[0] else None
 400     converted = [['osm_id']]
 401     for val, col in zip((start_idx, end_idx, geom_idx),
 402                         ('startnumber', 'endnumber', 'linegeo!wkt')):
 403         if val is not None:
 404             converted[0].append(col)
 405
 406     for line in datatable[1:]:
 407         convline = [oid]
 408         for val in (start_idx, end_idx):
 409             if val is not None:
 410                 convline.append(line[val])
 411         if geom_idx is not None:
 412             convline.append(line[geom_idx])
 413         converted.append(convline)
 414
 415     _rewrite_placeid_field('parent_place_id', 'parent_place_id', converted, place_ids)
 416
 417     check_table_content(db_conn, 'location_property_osmline', converted, grid=node_grid)
 418
 419
 420 @then(step_parse(r'W(?P<oid>\d+) expands to no interpolation'),
 421       converters={'oid': int})
 422 def then_check_interpolation_table_negative(db_conn, oid):
 423     with db_conn.cursor() as cur:
 424         cur.execute("""SELECT count(*) FROM location_property_osmline
 425                        WHERE osm_id = %s and startnumber is not null""",
 426                     [oid])
 427         assert cur.fetchone()[0] == 0
 428
 429
 430 if pytest.version_tuple >= (8, 0, 0):
 431     PYTEST_BDD_SCENARIOS = ['features/db']
 432 else:
 433     from pytest_bdd import scenarios
 434     scenarios('features/db')