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