test/bdd/test_osm2pgsql.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 osm2pgsql import style tests.
   9 """
  10 import asyncio
  11 import random
  12
  13 import psycopg
  14
  15 import pytest
  16 from pytest_bdd.parsers import re as step_parse
  17 from pytest_bdd import scenarios, when, given, then
  18
  19 from nominatim_db import cli
  20 from nominatim_db.config import Configuration
  21 from nominatim_db.tools.exec_utils import run_osm2pgsql
  22 from nominatim_db.tools.database_import import load_data, create_table_triggers
  23 from nominatim_db.tools.replication import run_osm2pgsql_updates
  24
  25 from utils.db import DBManager
  26 from utils.checks import check_table_content, check_table_has_lines
  27
  28
  29 @pytest.fixture
  30 def def_config(pytestconfig):
  31     dbname = pytestconfig.getini('nominatim_test_db')
  32
  33     return Configuration(None,
  34                          environ={'NOMINATIM_DATABASE_DSN': f"pgsql:dbname={dbname}"})
  35
  36
  37 @pytest.fixture
  38 def db(template_db, pytestconfig):
  39     """ Set up an empty database for use with osm2pgsql.
  40     """
  41     dbm = DBManager(purge=pytestconfig.option.NOMINATIM_PURGE)
  42
  43     dbname = pytestconfig.getini('nominatim_test_db')
  44
  45     dbm.create_db_from_template(dbname, template_db)
  46
  47     yield dbname
  48
  49     if not pytestconfig.option.NOMINATIM_KEEP_DB:
  50         dbm.drop_db(dbname)
  51
  52
  53 @pytest.fixture
  54 def db_conn(def_config):
  55     with psycopg.connect(def_config.get_libpq_dsn()) as conn:
  56         info = psycopg.types.TypeInfo.fetch(conn, "hstore")
  57         psycopg.types.hstore.register_hstore(info, conn)
  58         yield conn
  59
  60
  61 @pytest.fixture
  62 def osm2pgsql_options(def_config):
  63     return dict(osm2pgsql='osm2pgsql',
  64                 osm2pgsql_cache=50,
  65                 osm2pgsql_style=str(def_config.get_import_style_file()),
  66                 osm2pgsql_style_path=def_config.lib_dir.lua,
  67                 threads=1,
  68                 dsn=def_config.get_libpq_dsn(),
  69                 flatnode_file='',
  70                 tablespaces=dict(slim_data='', slim_index='',
  71                                  main_data='', main_index=''),
  72                 append=False)
  73
  74
  75 @pytest.fixture
  76 def opl_writer(tmp_path, node_grid):
  77     nr = [0]
  78
  79     def _write(data):
  80         fname = tmp_path / f"test_osm_{nr[0]}.opl"
  81         nr[0] += 1
  82         with fname.open('wt') as fd:
  83             for line in data.split('\n'):
  84                 if line.startswith('n') and ' x' not in line:
  85                     coord = node_grid.get(line[1:].split(' ')[0]) \
  86                             or (random.uniform(-180, 180), random.uniform(-90, 90))
  87                     line = f"{line} x{coord[0]:.7f} y{coord[1]:.7f}"
  88                 fd.write(line)
  89                 fd.write('\n')
  90         return fname
  91
  92     return _write
  93
  94
  95 @given('the lua style file', target_fixture='osm2pgsql_options')
  96 def set_lua_style_file(osm2pgsql_options, docstring, tmp_path):
  97     style = tmp_path / 'custom.lua'
  98     style.write_text(docstring)
  99     osm2pgsql_options['osm2pgsql_style'] = str(style)
 100
 101     return osm2pgsql_options
 102
 103
 104 @when('loading osm data')
 105 def load_from_osm_file(db, osm2pgsql_options, opl_writer, docstring):
 106     """ Load the given data into a freshly created test database using osm2pgsql.
 107         No further indexing is done.
 108
 109         The data is expected as attached text in OPL format.
 110     """
 111     osm2pgsql_options['import_file'] = opl_writer(docstring.replace(r'//', r'/'))
 112     osm2pgsql_options['append'] = False
 113     run_osm2pgsql(osm2pgsql_options)
 114
 115
 116 @when('updating osm data')
 117 def update_from_osm_file(db_conn, def_config, osm2pgsql_options, opl_writer, docstring):
 118     """ Update a database previously populated with 'loading osm data'.
 119         Needs to run indexing on the existing data first to yield the correct
 120         result.
 121
 122         The data is expected as attached text in OPL format.
 123     """
 124     create_table_triggers(db_conn, def_config)
 125     asyncio.run(load_data(def_config.get_libpq_dsn(), 1))
 126     cli.nominatim(['index'], def_config.environ)
 127     cli.nominatim(['refresh', '--functions'], def_config.environ)
 128
 129     osm2pgsql_options['import_file'] = opl_writer(docstring.replace(r'//', r'/'))
 130     run_osm2pgsql_updates(db_conn, osm2pgsql_options)
 131
 132
 133 @when('indexing')
 134 def do_index(def_config):
 135     """ Run Nominatim's indexing step.
 136     """
 137     cli.nominatim(['index'], def_config.environ)
 138
 139
 140 @then(step_parse(r'(?P<table>\w+) contains(?P<exact> exactly)?'))
 141 def check_place_content(db_conn, datatable, node_grid, table, exact):
 142     check_table_content(db_conn, table, datatable, grid=node_grid, exact=bool(exact))
 143
 144
 145 @then(step_parse('(?P<table>placex?) has no entry for '
 146                  r'(?P<osm_type>[NRW])(?P<osm_id>\d+)(?::(?P<osm_class>\S+))?'),
 147       converters={'osm_id': int})
 148 def check_place_missing_lines(db_conn, table, osm_type, osm_id, osm_class):
 149     check_table_has_lines(db_conn, table, osm_type, osm_id, osm_class)
 150
 151
 152 scenarios('features/osm2pgsql')