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 pytest
  14 from pytest_bdd import scenarios, when, given
  15
  16 from nominatim_db import cli
  17 from nominatim_db.tools.exec_utils import run_osm2pgsql
  18 from nominatim_db.tools.database_import import load_data, create_table_triggers
  19 from nominatim_db.tools.replication import run_osm2pgsql_updates
  20
  21
  22 @pytest.fixture
  23 def osm2pgsql_options(def_config):
  24     return dict(osm2pgsql='osm2pgsql',
  25                 osm2pgsql_cache=50,
  26                 osm2pgsql_style=str(def_config.get_import_style_file()),
  27                 osm2pgsql_style_path=def_config.lib_dir.lua,
  28                 threads=1,
  29                 dsn=def_config.get_libpq_dsn(),
  30                 flatnode_file='',
  31                 tablespaces=dict(slim_data='', slim_index='',
  32                                  main_data='', main_index=''),
  33                 append=False)
  34
  35
  36 @pytest.fixture
  37 def opl_writer(tmp_path, node_grid):
  38     nr = [0]
  39
  40     def _write(data):
  41         fname = tmp_path / f"test_osm_{nr[0]}.opl"
  42         nr[0] += 1
  43         with fname.open('wt') as fd:
  44             for line in data.split('\n'):
  45                 if line.startswith('n') and ' x' not in line:
  46                     coord = node_grid.get(line[1:].split(' ')[0]) \
  47                             or (random.uniform(-180, 180), random.uniform(-90, 90))
  48                     line = f"{line} x{coord[0]:.7f} y{coord[1]:.7f}"
  49                 fd.write(line)
  50                 fd.write('\n')
  51         return fname
  52
  53     return _write
  54
  55
  56 @given('the lua style file', target_fixture='osm2pgsql_options')
  57 def set_lua_style_file(osm2pgsql_options, docstring, tmp_path):
  58     style = tmp_path / 'custom.lua'
  59     style.write_text(docstring)
  60     osm2pgsql_options['osm2pgsql_style'] = str(style)
  61
  62     return osm2pgsql_options
  63
  64
  65 @when('loading osm data')
  66 def load_from_osm_file(db, osm2pgsql_options, opl_writer, docstring):
  67     """ Load the given data into a freshly created test database using osm2pgsql.
  68         No further indexing is done.
  69
  70         The data is expected as attached text in OPL format.
  71     """
  72     osm2pgsql_options['import_file'] = opl_writer(docstring.replace(r'//', r'/'))
  73     osm2pgsql_options['append'] = False
  74     run_osm2pgsql(osm2pgsql_options)
  75
  76
  77 @when('updating osm data')
  78 def update_from_osm_file(db_conn, def_config, osm2pgsql_options, opl_writer, docstring):
  79     """ Update a database previously populated with 'loading osm data'.
  80         Needs to run indexing on the existing data first to yield the correct
  81         result.
  82
  83         The data is expected as attached text in OPL format.
  84     """
  85     create_table_triggers(db_conn, def_config)
  86     asyncio.run(load_data(def_config.get_libpq_dsn(), 1))
  87     cli.nominatim(['index'], def_config.environ)
  88     cli.nominatim(['refresh', '--functions'], def_config.environ)
  89
  90     osm2pgsql_options['import_file'] = opl_writer(docstring.replace(r'//', r'/'))
  91     run_osm2pgsql_updates(db_conn, osm2pgsql_options)
  92
  93
  94 scenarios('features/osm2pgsql')