add tests for adding additional data

Also adds checks that parameters for osm2pgsql are set
as expected.

test/python/tools/test_add_osm_data.py

@@ -0,0 +1,52 @@
+"""
+Tests for functions to add additional data to the database.
+"""
+from pathlib import Path
+
+import pytest
+
+from nominatim.tools import add_osm_data
+
+class CaptureGetUrl:
+
+    def __init__(self, monkeypatch):
+        self.url = None
+        monkeypatch.setattr(add_osm_data, 'get_url', self)
+
+    def __call__(self, url):
+        self.url = url
+        return '<xml></xml>'
+
+
+def test_import_osm_file_simple(table_factory, osm2pgsql_options, capfd):
+    table_factory('place', content=((1, ), ))
+
+    assert add_osm_data.add_data_from_file(Path('change.osm'), osm2pgsql_options) == 0
+    captured = capfd.readouterr()
+
+    assert '--append' in captured.out
+    assert '--output gazetteer' in captured.out
+    assert f'--style {osm2pgsql_options["osm2pgsql_style"]}' in captured.out
+    assert f'--number-processes {osm2pgsql_options["threads"]}' in captured.out
+    assert f'--cache {osm2pgsql_options["osm2pgsql_cache"]}' in captured.out
+    assert 'change.osm' in captured.out
+
+
+@pytest.mark.parametrize("osm_type", ['node', 'way', 'relation'])
+@pytest.mark.parametrize("main_api,url", [(True, 'https://www.openstreetmap.org/api'),
+                                          (False, 'https://overpass-api.de/api/interpreter?')])
+def test_import_osm_object_main_api(osm2pgsql_options, monkeypatch, capfd,
+                                    osm_type, main_api, url):
+    get_url_mock = CaptureGetUrl(monkeypatch)
+
+    add_osm_data.add_osm_object(osm_type, 4536, main_api, osm2pgsql_options)
+    captured = capfd.readouterr()
+
+    assert get_url_mock.url.startswith(url)
+
+    assert '--append' in captured.out
+    assert '--output gazetteer' in captured.out
+    assert f'--style {osm2pgsql_options["osm2pgsql_style"]}' in captured.out
+    assert f'--number-processes {osm2pgsql_options["threads"]}' in captured.out
+    assert f'--cache {osm2pgsql_options["osm2pgsql_cache"]}' in captured.out
+    assert captured.out.endswith(' -\n')

test/python/tools/test_database_import.py

@@ -78,13 +78,21 @@ def test_setup_skeleton_already_exists(temp_db):
         database_import.setup_database_skeleton(f'dbname={temp_db}')
 
 
-def test_import_osm_data_simple(table_factory, osm2pgsql_options):
+def test_import_osm_data_simple(table_factory, osm2pgsql_options, capfd):
     table_factory('place', content=((1, ), ))
 
     database_import.import_osm_data(Path('file.pbf'), osm2pgsql_options)
+    captured = capfd.readouterr()
+
+    assert '--create' in captured.out
+    assert '--output gazetteer' in captured.out
+    assert f'--style {osm2pgsql_options["osm2pgsql_style"]}' in captured.out
+    assert f'--number-processes {osm2pgsql_options["threads"]}' in captured.out
+    assert f'--cache {osm2pgsql_options["osm2pgsql_cache"]}' in captured.out
+    assert 'file.pbf' in captured.out
 
 
-def test_import_osm_data_multifile(table_factory, tmp_path, osm2pgsql_options):
+def test_import_osm_data_multifile(table_factory, tmp_path, osm2pgsql_options, capfd):
     table_factory('place', content=((1, ), ))
     osm2pgsql_options['osm2pgsql_cache'] = 0
 
@@ -93,6 +101,10 @@ def test_import_osm_data_multifile(table_factory, tmp_path, osm2pgsql_options):
         f.write_text('test')
 
     database_import.import_osm_data(files, osm2pgsql_options)
+    captured = capfd.readouterr()
+
+    assert 'file1.osm' in captured.out
+    assert 'file2.osm' in captured.out
 
 
 def test_import_osm_data_simple_no_data(table_factory, osm2pgsql_options):
@@ -117,12 +129,15 @@ def test_import_osm_data_drop(table_factory, temp_db_conn, tmp_path, osm2pgsql_o
     assert not temp_db_conn.table_exists('planet_osm_nodes')
 
 
-def test_import_osm_data_default_cache(table_factory, osm2pgsql_options):
+def test_import_osm_data_default_cache(table_factory, osm2pgsql_options, capfd):
     table_factory('place', content=((1, ), ))
 
     osm2pgsql_options['osm2pgsql_cache'] = 0
 
     database_import.import_osm_data(Path(__file__), osm2pgsql_options)
+    captured = capfd.readouterr()
+
+    assert f'--cache {osm2pgsql_options["osm2pgsql_cache"]}' in captured.out
 
 
 def test_truncate_database_tables(temp_db_conn, temp_db_cursor, table_factory):