test/python/tools/test_add_osm_data.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 Tests for functions to add additional data to the database.
   9 """
  10 from pathlib import Path
  11
  12 import pytest
  13
  14 from nominatim_db.tools import add_osm_data
  15
  16
  17 class CaptureGetUrl:
  18
  19     def __init__(self, monkeypatch):
  20         self.url = None
  21         monkeypatch.setattr(add_osm_data, 'get_url', self)
  22
  23     def __call__(self, url):
  24         self.url = url
  25         return '<xml></xml>'
  26
  27
  28 @pytest.fixture(autouse=True)
  29 def setup_delete_postprocessing(temp_db_cursor):
  30     temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION flush_deleted_places()
  31                               RETURNS INTEGER AS $$ SELECT 1 $$ LANGUAGE SQL""")
  32
  33
  34 def test_import_osm_file_simple(dsn, table_factory, osm2pgsql_options, capfd):
  35
  36     assert add_osm_data.add_data_from_file(dsn, Path('change.osm'), osm2pgsql_options) == 0
  37     captured = capfd.readouterr()
  38
  39     assert '--append' in captured.out
  40     assert '--output flex' in captured.out
  41     assert f'--style {osm2pgsql_options["osm2pgsql_style"]}' in captured.out
  42     assert f'--number-processes {osm2pgsql_options["threads"]}' in captured.out
  43     assert f'--cache {osm2pgsql_options["osm2pgsql_cache"]}' in captured.out
  44     assert 'change.osm' in captured.out
  45
  46
  47 @pytest.mark.parametrize("osm_type", ['node', 'way', 'relation'])
  48 @pytest.mark.parametrize("main_api,url", [(True, 'https://www.openstreetmap.org/api'),
  49                                           (False, 'https://overpass-api.de/api/interpreter?')])
  50 def test_import_osm_object_main_api(dsn, osm2pgsql_options, monkeypatch,
  51                                     capfd, osm_type, main_api, url):
  52     get_url_mock = CaptureGetUrl(monkeypatch)
  53
  54     add_osm_data.add_osm_object(dsn, osm_type, 4536, main_api, osm2pgsql_options)
  55     captured = capfd.readouterr()
  56
  57     assert get_url_mock.url.startswith(url)
  58
  59     assert '--append' in captured.out
  60     assert '--output flex' in captured.out
  61     assert f'--style {osm2pgsql_options["osm2pgsql_style"]}' in captured.out
  62     assert f'--number-processes {osm2pgsql_options["threads"]}' in captured.out
  63     assert f'--cache {osm2pgsql_options["osm2pgsql_cache"]}' in captured.out
  64     assert captured.out.endswith(' -\n')