X-Git-Url: https://git.openstreetmap.org./nominatim.git/blobdiff_plain/b90e719da595b6760b39b7cd64ee29447de2d5e8..c41f2fed2133668dc3179813261d39d3ff69cbdd:/test/python/tools/test_database_import.py?ds=inline diff --git a/test/python/tools/test_database_import.py b/test/python/tools/test_database_import.py index eda88903..68d19a07 100644 --- a/test/python/tools/test_database_import.py +++ b/test/python/tools/test_database_import.py @@ -1,3 +1,9 @@ +# SPDX-License-Identifier: GPL-2.0-only +# +# This file is part of Nominatim. (https://nominatim.org) +# +# Copyright (C) 2022 by the Nominatim developer community. +# For a full list of authors see the git log. """ Tests for functions to import a new database. """ @@ -60,6 +66,11 @@ class TestDatabaseSetup: database_import.setup_database_skeleton(f'dbname={self.DBNAME}') + def test_create_db_explicit_ro_user(self): + database_import.setup_database_skeleton(f'dbname={self.DBNAME}', + rouser='postgres') + + def test_create_db_missing_ro_user(self): with pytest.raises(UsageError, match='Missing read-only user.'): database_import.setup_database_skeleton(f'dbname={self.DBNAME}', @@ -78,13 +89,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,15 +112,26 @@ 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): table_factory('place') - with pytest.raises(UsageError, match='No data.*'): + with pytest.raises(UsageError, match='No data imported'): database_import.import_osm_data(Path('file.pbf'), osm2pgsql_options) +def test_import_osm_data_simple_ignore_no_data(table_factory, osm2pgsql_options): + table_factory('place') + + database_import.import_osm_data(Path('file.pbf'), osm2pgsql_options, + ignore_errors=True) + + def test_import_osm_data_drop(table_factory, temp_db_conn, tmp_path, osm2pgsql_options): table_factory('place', content=((1, ), )) table_factory('planet_osm_nodes') @@ -117,19 +147,26 @@ 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): - tables = ('placex', 'place_addressline', 'location_area', +@pytest.mark.parametrize("with_search", (True, False)) +def test_truncate_database_tables(temp_db_conn, temp_db_cursor, table_factory, with_search): + tables = ['placex', 'place_addressline', 'location_area', 'location_area_country', 'location_property_tiger', 'location_property_osmline', - 'location_postcode', 'search_name', 'location_road_23') + 'location_postcode', 'location_road_23'] + if with_search: + tables.append('search_name') + for table in tables: table_factory(table, content=((1, ), (2, ), (3, ))) assert temp_db_cursor.table_rows(table) == 3 @@ -142,11 +179,11 @@ def test_truncate_database_tables(temp_db_conn, temp_db_cursor, table_factory): @pytest.mark.parametrize("threads", (1, 5)) def test_load_data(dsn, place_row, placex_table, osmline_table, - word_table, temp_db_cursor, threads): + temp_db_cursor, threads): for func in ('precompute_words', 'getorcreate_housenumber_id', 'make_standard_name'): - temp_db_cursor.execute("""CREATE FUNCTION {} (src TEXT) + temp_db_cursor.execute(f"""CREATE FUNCTION {func} (src TEXT) RETURNS TEXT AS $$ SELECT 'a'::TEXT $$ LANGUAGE SQL - """.format(func)) + """) for oid in range(100, 130): place_row(osm_id=oid) place_row(osm_type='W', osm_id=342, cls='place', typ='houses', @@ -156,3 +193,59 @@ def test_load_data(dsn, place_row, placex_table, osmline_table, assert temp_db_cursor.table_rows('placex') == 30 assert temp_db_cursor.table_rows('location_property_osmline') == 1 + + +class TestSetupSQL: + + @pytest.fixture(autouse=True) + def init_env(self, temp_db, tmp_path, def_config, sql_preprocessor_cfg): + def_config.lib_dir.sql = tmp_path / 'sql' + def_config.lib_dir.sql.mkdir() + + self.config = def_config + + + def write_sql(self, fname, content): + (self.config.lib_dir.sql / fname).write_text(content) + + + @pytest.mark.parametrize("reverse", [True, False]) + def test_create_tables(self, temp_db_conn, temp_db_cursor, reverse): + self.write_sql('tables.sql', + """CREATE FUNCTION test() RETURNS bool + AS $$ SELECT {{db.reverse_only}} $$ LANGUAGE SQL""") + + database_import.create_tables(temp_db_conn, self.config, reverse) + + temp_db_cursor.scalar('SELECT test()') == reverse + + + def test_create_table_triggers(self, temp_db_conn, temp_db_cursor): + self.write_sql('table-triggers.sql', + """CREATE FUNCTION test() RETURNS TEXT + AS $$ SELECT 'a'::text $$ LANGUAGE SQL""") + + database_import.create_table_triggers(temp_db_conn, self.config) + + temp_db_cursor.scalar('SELECT test()') == 'a' + + + def test_create_partition_tables(self, temp_db_conn, temp_db_cursor): + self.write_sql('partition-tables.src.sql', + """CREATE FUNCTION test() RETURNS TEXT + AS $$ SELECT 'b'::text $$ LANGUAGE SQL""") + + database_import.create_partition_tables(temp_db_conn, self.config) + + temp_db_cursor.scalar('SELECT test()') == 'b' + + + @pytest.mark.parametrize("drop", [True, False]) + def test_create_search_indices(self, temp_db_conn, temp_db_cursor, drop): + self.write_sql('indices.sql', + """CREATE FUNCTION test() RETURNS bool + AS $$ SELECT {{drop}} $$ LANGUAGE SQL""") + + database_import.create_search_indices(temp_db_conn, self.config, drop) + + temp_db_cursor.scalar('SELECT test()') == drop