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) 2025 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 asyncio
  13 import re
  14 from pathlib import Path
  15
  16 import psycopg
  17
  18 import pytest
  19 from pytest_bdd import scenarios, 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
  25 from nominatim_db.config import Configuration
  26 from nominatim_db import cli
  27 from nominatim_db.tools.database_import import load_data, create_table_triggers
  28 from nominatim_db.tools.postcodes import update_postcodes
  29 from nominatim_db.tokenizer import factory as tokenizer_factory
  30
  31
  32 def _rewrite_placeid_field(field, new_field, datatable, place_ids):
  33     try:
  34         oidx = datatable[0].index(field)
  35         datatable[0][oidx] = new_field
  36         for line in datatable[1:]:
  37             line[oidx] = None if line[oidx] == '-' else place_ids[line[oidx]]
  38     except ValueError:
  39         pass
  40
  41
  42 def _collect_place_ids(conn):
  43     pids = {}
  44     with conn.cursor() as cur:
  45         for row in cur.execute('SELECT place_id, osm_type, osm_id, class FROM placex'):
  46             pids[f"{row[1]}{row[2]}"] = row[0]
  47             pids[f"{row[1]}{row[2]}:{row[3]}"] = row[0]
  48
  49     return pids
  50
  51
  52 @pytest.fixture
  53 def test_config_env(pytestconfig):
  54     dbname = pytestconfig.getini('nominatim_test_db')
  55
  56     config = Configuration(None).get_os_env()
  57     config['NOMINATIM_DATABASE_DSN'] = f"pgsql:dbname={dbname}"
  58     config['NOMINATIM_LANGUAGES'] = 'en,de,fr,ja'
  59     config['NOMINATIM_USE_US_TIGER_DATA'] = 'yes'
  60     if pytestconfig.option.NOMINATIM_TOKENIZER is not None:
  61         config['NOMINATIM_TOKENIZER'] = pytestconfig.option.NOMINATIM_TOKENIZER
  62
  63     return config
  64
  65
  66 @pytest.fixture
  67 def update_config(def_config):
  68     """ Prepare the database for being updatable and return the config.
  69     """
  70     cli.nominatim(['refresh', '--functions'], def_config.environ)
  71
  72     return def_config
  73
  74
  75 @given(step_parse('the (?P<named>named )?places'), target_fixture=None)
  76 def import_places(db_conn, named, datatable, node_grid):
  77     """ Insert todo rows into the place table.
  78         When 'named' is given, then a random name will be generated for all
  79         objects.
  80     """
  81     with db_conn.cursor() as cur:
  82         for row in datatable[1:]:
  83             PlaceColumn(node_grid).add_row(datatable[0], row, named is not None).db_insert(cur)
  84
  85
  86 @given('the ways', target_fixture=None)
  87 def import_ways(db_conn, datatable):
  88     """ Import raw ways into the osm2pgsql way middle table.
  89     """
  90     with db_conn.cursor() as cur:
  91         id_idx = datatable[0].index('id')
  92         node_idx = datatable[0].index('nodes')
  93         for line in datatable[1:]:
  94             tags = psycopg.types.json.Json(
  95                 {k[5:]: v for k, v in zip(datatable[0], line)
  96                  if k.startswith("tags+")})
  97             nodes = [int(x) for x in line[node_idx].split(',')]
  98
  99             cur.execute("INSERT INTO planet_osm_ways (id, nodes, tags) VALUES (%s, %s, %s)",
 100                         (line[id_idx], nodes, tags))
 101
 102
 103 @given('the relations', target_fixture=None)
 104 def import_rels(db_conn, datatable):
 105     """ Import raw relations into the osm2pgsql relation middle table.
 106     """
 107     with db_conn.cursor() as cur:
 108         id_idx = datatable[0].index('id')
 109         memb_idx = datatable[0].index('members')
 110         for line in datatable[1:]:
 111             tags = psycopg.types.json.Json(
 112                 {k[5:]: v for k, v in zip(datatable[0], line)
 113                  if k.startswith("tags+")})
 114             members = []
 115             if line[memb_idx]:
 116                 for member in line[memb_idx].split(','):
 117                     m = re.fullmatch(r'\s*([RWN])(\d+)(?::(\S+))?\s*', member)
 118                     if not m:
 119                         raise ValueError(f'Illegal member {member}.')
 120                     members.append({'ref': int(m[2]), 'role': m[3] or '', 'type': m[1]})
 121
 122             cur.execute('INSERT INTO planet_osm_rels (id, tags, members) VALUES (%s, %s, %s)',
 123                         (int(line[id_idx]), tags, psycopg.types.json.Json(members)))
 124
 125
 126 @when('importing', target_fixture='place_ids')
 127 def do_import(db_conn, def_config):
 128     """ Run a reduced version of the Nominatim import.
 129     """
 130     create_table_triggers(db_conn, def_config)
 131     asyncio.run(load_data(def_config.get_libpq_dsn(), 1))
 132     tokenizer = tokenizer_factory.get_tokenizer_for_db(def_config)
 133     update_postcodes(def_config.get_libpq_dsn(), Path('/xxxx'), tokenizer)
 134     cli.nominatim(['index', '-q'], def_config.environ)
 135
 136     return _collect_place_ids(db_conn)
 137
 138
 139 @when('updating places', target_fixture='place_ids')
 140 def do_update(db_conn, update_config, node_grid, datatable):
 141     """ Update the place table with the given data. Also runs all triggers
 142         related to updates and reindexes the new data.
 143     """
 144     with db_conn.cursor() as cur:
 145         for row in datatable[1:]:
 146             PlaceColumn(node_grid).add_row(datatable[0], row, False).db_insert(cur)
 147         cur.execute('SELECT flush_deleted_places()')
 148     db_conn.commit()
 149
 150     cli.nominatim(['index', '-q'], update_config.environ)
 151
 152     return _collect_place_ids(db_conn)
 153
 154
 155 @when('updating postcodes')
 156 def do_postcode_update(update_config):
 157     """ Recompute the postcode centroids.
 158     """
 159     cli.nominatim(['refresh', '--postcodes'], update_config.environ)
 160
 161
 162 @when(step_parse(r'marking for delete (?P<otype>[NRW])(?P<oid>\d+)'),
 163       converters={'oid': int})
 164 def do_delete_place(db_conn, update_config, node_grid, otype, oid):
 165     """ Remove the given place from the database.
 166     """
 167     with db_conn.cursor() as cur:
 168         cur.execute('TRUNCATE place_to_be_deleted')
 169         cur.execute('DELETE FROM place WHERE osm_type = %s and osm_id = %s',
 170                     (otype, oid))
 171         cur.execute('SELECT flush_deleted_places()')
 172     db_conn.commit()
 173
 174     cli.nominatim(['index', '-q'], update_config.environ)
 175
 176
 177 @then(step_parse(r'(?P<table>\w+) contains(?P<exact> exactly)?'))
 178 def then_check_table_content(db_conn, place_ids, datatable, node_grid, table, exact):
 179     _rewrite_placeid_field('object', 'place_id', datatable, place_ids)
 180     _rewrite_placeid_field('parent_place_id', 'parent_place_id', datatable, place_ids)
 181     _rewrite_placeid_field('linked_place_id', 'linked_place_id', datatable, place_ids)
 182     if table == 'place_addressline':
 183         _rewrite_placeid_field('address', 'address_place_id', datatable, place_ids)
 184
 185     for i, title in enumerate(datatable[0]):
 186         if title.startswith('addr+'):
 187             datatable[0][i] = f"address+{title[5:]}"
 188
 189     check_table_content(db_conn, table, datatable, grid=node_grid, exact=bool(exact))
 190
 191
 192 @then(step_parse(r'(DISABLED?P<table>placex?) has no entry for (?P<oid>[NRW]\d+(?::\S+)?)'))
 193 def then_check_place_missing_lines(db_conn, place_ids, table, oid):
 194     assert oid in place_ids
 195
 196     sql = pysql.SQL("""SELECT count(*) FROM {}
 197                        WHERE place_id = %s""").format(pysql.Identifier(tablename))
 198
 199     with conn.cursor(row_factory=tuple_row) as cur:
 200         assert cur.execute(sql, [place_ids[oid]]).fetchone()[0] == 0
 201
 202
 203 @then(step_parse(r'W(?P<oid>\d+) expands to interpolation'),
 204       converters={'oid': int})
 205 def then_check_interpolation_table(db_conn, node_grid, place_ids, oid, datatable):
 206     with db_conn.cursor() as cur:
 207         cur.execute('SELECT count(*) FROM location_property_osmline WHERE osm_id = %s',
 208                     [oid])
 209         assert cur.fetchone()[0] == len(datatable) - 1
 210
 211     converted = [['osm_id', 'startnumber', 'endnumber', 'linegeo!wkt']]
 212     start_idx = datatable[0].index('start') if 'start' in datatable[0] else None
 213     end_idx = datatable[0].index('end') if 'end' in datatable[0] else None
 214     geom_idx = datatable[0].index('geometry') if 'geometry' in datatable[0] else None
 215     converted = [['osm_id']]
 216     for val, col in zip((start_idx, end_idx, geom_idx),
 217                         ('startnumber', 'endnumber', 'linegeo!wkt')):
 218         if val is not None:
 219             converted[0].append(col)
 220
 221     for line in datatable[1:]:
 222         convline = [oid]
 223         for val in (start_idx, end_idx):
 224             if val is not None:
 225                 convline.append(line[val])
 226         if geom_idx is not None:
 227             convline.append(line[geom_idx])
 228         converted.append(convline)
 229
 230     _rewrite_placeid_field('parent_place_id', 'parent_place_id', converted, place_ids)
 231
 232     check_table_content(db_conn, 'location_property_osmline', converted, grid=node_grid)
 233
 234
 235 @then(step_parse(r'W(?P<oid>\d+) expands to no interpolation'),
 236       converters={'oid': int})
 237 def then_check_interpolation_table_negative(db_conn, oid):
 238     with db_conn.cursor() as cur:
 239         cur.execute("""SELECT count(*) FROM location_property_osmline
 240                        WHERE osm_id = %s and startnumber is not null""",
 241                     [oid])
 242         assert cur.fetchone()[0] == 0
 243
 244
 245 scenarios('features/db')