test/python/cli/test_cmd_import.py

   1 # SPDX-License-Identifier: GPL-2.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 import command of the command-line interface wrapper.
   9 """
  10 import pytest
  11
  12 import nominatim_db.tools.database_import
  13 import nominatim_db.data.country_info
  14 import nominatim_db.tools.refresh
  15 import nominatim_db.tools.postcodes
  16 import nominatim_db.indexer.indexer
  17 import nominatim_db.db.properties
  18
  19
  20 class TestCliImportWithDb:
  21
  22     @pytest.fixture(autouse=True)
  23     def setup_cli_call(self, cli_call, temp_db, cli_tokenizer_mock):
  24         self.call_nominatim = cli_call
  25         self.tokenizer_mock = cli_tokenizer_mock
  26
  27
  28     def test_import_missing_file(self):
  29         assert self.call_nominatim('import', '--osm-file', 'sfsafegwedgw.reh.erh') == 1
  30
  31
  32     def test_import_bad_file(self):
  33         assert self.call_nominatim('import', '--osm-file', '.') == 1
  34
  35
  36     @pytest.mark.parametrize('with_updates', [True, False])
  37     def test_import_full(self, mock_func_factory, async_mock_func_factory,
  38                          with_updates, place_table, property_table):
  39         mocks = [
  40             mock_func_factory(nominatim_db.tools.database_import, 'setup_database_skeleton'),
  41             mock_func_factory(nominatim_db.data.country_info, 'setup_country_tables'),
  42             mock_func_factory(nominatim_db.tools.database_import, 'import_osm_data'),
  43             mock_func_factory(nominatim_db.tools.refresh, 'import_wikipedia_articles'),
  44             mock_func_factory(nominatim_db.tools.refresh, 'import_secondary_importance'),
  45             mock_func_factory(nominatim_db.tools.database_import, 'truncate_data_tables'),
  46             async_mock_func_factory(nominatim_db.tools.database_import, 'load_data'),
  47             mock_func_factory(nominatim_db.tools.database_import, 'create_tables'),
  48             mock_func_factory(nominatim_db.tools.database_import, 'create_table_triggers'),
  49             mock_func_factory(nominatim_db.tools.database_import, 'create_partition_tables'),
  50             async_mock_func_factory(nominatim_db.tools.database_import, 'create_search_indices'),
  51             mock_func_factory(nominatim_db.data.country_info, 'create_country_names'),
  52             mock_func_factory(nominatim_db.tools.refresh, 'load_address_levels_from_config'),
  53             mock_func_factory(nominatim_db.tools.postcodes, 'update_postcodes'),
  54             async_mock_func_factory(nominatim_db.indexer.indexer.Indexer, 'index_full'),
  55             mock_func_factory(nominatim_db.tools.refresh, 'setup_website'),
  56         ]
  57
  58         params = ['import', '--osm-file', __file__]
  59
  60         if with_updates:
  61             mocks.append(mock_func_factory(nominatim_db.tools.freeze, 'drop_update_tables'))
  62             params.append('--no-updates')
  63
  64         cf_mock = mock_func_factory(nominatim_db.tools.refresh, 'create_functions')
  65
  66
  67         assert self.call_nominatim(*params) == 0
  68         assert self.tokenizer_mock.finalize_import_called
  69
  70         assert cf_mock.called > 1
  71
  72         for mock in mocks:
  73             assert mock.called == 1, "Mock '{}' not called".format(mock.func_name)
  74
  75
  76     def test_import_continue_load_data(self, mock_func_factory, async_mock_func_factory):
  77         mocks = [
  78             mock_func_factory(nominatim_db.tools.database_import, 'truncate_data_tables'),
  79             async_mock_func_factory(nominatim_db.tools.database_import, 'load_data'),
  80             async_mock_func_factory(nominatim_db.tools.database_import, 'create_search_indices'),
  81             mock_func_factory(nominatim_db.data.country_info, 'create_country_names'),
  82             mock_func_factory(nominatim_db.tools.postcodes, 'update_postcodes'),
  83             async_mock_func_factory(nominatim_db.indexer.indexer.Indexer, 'index_full'),
  84             mock_func_factory(nominatim_db.tools.refresh, 'setup_website'),
  85             mock_func_factory(nominatim_db.db.properties, 'set_property')
  86         ]
  87
  88         assert self.call_nominatim('import', '--continue', 'load-data') == 0
  89         assert self.tokenizer_mock.finalize_import_called
  90
  91         for mock in mocks:
  92             assert mock.called == 1, "Mock '{}' not called".format(mock.func_name)
  93
  94
  95     def test_import_continue_indexing(self, mock_func_factory, async_mock_func_factory,
  96                                       placex_table, temp_db_conn):
  97         mocks = [
  98             async_mock_func_factory(nominatim_db.tools.database_import, 'create_search_indices'),
  99             mock_func_factory(nominatim_db.data.country_info, 'create_country_names'),
 100             async_mock_func_factory(nominatim_db.indexer.indexer.Indexer, 'index_full'),
 101             mock_func_factory(nominatim_db.tools.refresh, 'setup_website'),
 102             mock_func_factory(nominatim_db.db.properties, 'set_property')
 103         ]
 104
 105         assert self.call_nominatim('import', '--continue', 'indexing') == 0
 106
 107         for mock in mocks:
 108             assert mock.called == 1, "Mock '{}' not called".format(mock.func_name)
 109
 110         # Calling it again still works for the index
 111         assert self.call_nominatim('import', '--continue', 'indexing') == 0
 112
 113
 114     def test_import_continue_postprocess(self, mock_func_factory, async_mock_func_factory):
 115         mocks = [
 116             async_mock_func_factory(nominatim_db.tools.database_import, 'create_search_indices'),
 117             mock_func_factory(nominatim_db.data.country_info, 'create_country_names'),
 118             mock_func_factory(nominatim_db.tools.refresh, 'setup_website'),
 119             mock_func_factory(nominatim_db.db.properties, 'set_property')
 120         ]
 121
 122         assert self.call_nominatim('import', '--continue', 'db-postprocess') == 0
 123
 124         assert self.tokenizer_mock.finalize_import_called
 125
 126         for mock in mocks:
 127             assert mock.called == 1, "Mock '{}' not called".format(mock.func_name)