test/python/test_cli_replication.py

   1 """
   2 Tests for replication command of command-line interface wrapper.
   3 """
   4 import datetime as dt
   5 import time
   6 from pathlib import Path
   7
   8 import pytest
   9
  10 import nominatim.cli
  11 import nominatim.indexer.indexer
  12 import nominatim.tools.replication
  13 from nominatim.db import status
  14
  15 from mocks import MockParamCapture
  16
  17 SRC_DIR = (Path(__file__) / '..' / '..' / '..').resolve()
  18
  19 def call_nominatim(*args):
  20     return nominatim.cli.nominatim(module_dir='build/module',
  21                                    osm2pgsql_path='build/osm2pgsql/osm2pgsql',
  22                                    phplib_dir=str(SRC_DIR / 'lib-php'),
  23                                    data_dir=str(SRC_DIR / 'data'),
  24                                    phpcgi_path='/usr/bin/php-cgi',
  25                                    sqllib_dir=str(SRC_DIR / 'lib-sql'),
  26                                    config_dir=str(SRC_DIR / 'settings'),
  27                                    cli_args=['replication'] + list(args))
  28
  29 @pytest.fixture
  30 def tokenizer_mock(monkeypatch):
  31     class DummyTokenizer:
  32         def __init__(self, *args, **kwargs):
  33             self.update_sql_functions_called = False
  34             self.finalize_import_called = False
  35
  36         def update_sql_functions(self, *args):
  37             self.update_sql_functions_called = True
  38
  39         def finalize_import(self, *args):
  40             self.finalize_import_called = True
  41
  42     tok = DummyTokenizer()
  43     monkeypatch.setattr(nominatim.tokenizer.factory, 'get_tokenizer_for_db' ,
  44                         lambda *args: tok)
  45     monkeypatch.setattr(nominatim.tokenizer.factory, 'create_tokenizer' ,
  46                         lambda *args: tok)
  47
  48     return tok
  49
  50
  51 @pytest.fixture
  52 def index_mock(monkeypatch, tokenizer_mock):
  53     mock = MockParamCapture()
  54     monkeypatch.setattr(nominatim.indexer.indexer.Indexer, 'index_boundaries', mock)
  55     monkeypatch.setattr(nominatim.indexer.indexer.Indexer, 'index_by_rank', mock)
  56
  57     return mock
  58
  59
  60 @pytest.fixture
  61 def mock_func_factory(monkeypatch):
  62     def get_mock(module, func):
  63         mock = MockParamCapture()
  64         monkeypatch.setattr(module, func, mock)
  65         return mock
  66
  67     return get_mock
  68
  69
  70 @pytest.fixture
  71 def init_status(temp_db_conn, status_table):
  72     status.set_status(temp_db_conn, date=dt.datetime.now(dt.timezone.utc), seq=1)
  73     return 1
  74
  75
  76 @pytest.fixture
  77 def update_mock(mock_func_factory, init_status, tokenizer_mock):
  78     return mock_func_factory(nominatim.tools.replication, 'update')
  79
  80 @pytest.mark.parametrize("params,func", [
  81                          (('--init', '--no-update-functions'), 'init_replication'),
  82                          (('--check-for-updates',), 'check_for_updates')
  83                          ])
  84 def test_replication_command(mock_func_factory, temp_db, params, func):
  85     func_mock = mock_func_factory(nominatim.tools.replication, func)
  86
  87     assert 0 == call_nominatim(*params)
  88     assert func_mock.called == 1
  89
  90
  91 def test_replication_update_bad_interval(monkeypatch, temp_db):
  92     monkeypatch.setenv('NOMINATIM_REPLICATION_UPDATE_INTERVAL', 'xx')
  93
  94     assert call_nominatim() == 1
  95
  96
  97 def test_replication_update_bad_interval_for_geofabrik(monkeypatch, temp_db):
  98     monkeypatch.setenv('NOMINATIM_REPLICATION_URL',
  99                        'https://download.geofabrik.de/europe/ireland-and-northern-ireland-updates')
 100
 101     assert call_nominatim() == 1
 102
 103
 104 def test_replication_update_once_no_index(update_mock):
 105     assert 0 == call_nominatim('--once', '--no-index')
 106
 107     assert str(update_mock.last_args[1]['osm2pgsql']) == 'build/osm2pgsql/osm2pgsql'
 108
 109
 110 def test_replication_update_custom_osm2pgsql(monkeypatch, update_mock):
 111     monkeypatch.setenv('NOMINATIM_OSM2PGSQL_BINARY', '/secret/osm2pgsql')
 112     assert 0 == call_nominatim('--once', '--no-index')
 113
 114     assert str(update_mock.last_args[1]['osm2pgsql']) == '/secret/osm2pgsql'
 115
 116
 117 def test_replication_update_custom_threads(update_mock):
 118     assert 0 == call_nominatim('--once', '--no-index', '--threads', '4')
 119
 120     assert update_mock.last_args[1]['threads'] == 4
 121
 122
 123 def test_replication_update_continuous(monkeypatch, init_status, index_mock):
 124     states = [nominatim.tools.replication.UpdateState.UP_TO_DATE,
 125               nominatim.tools.replication.UpdateState.UP_TO_DATE]
 126     monkeypatch.setattr(nominatim.tools.replication, 'update',
 127                         lambda *args, **kwargs: states.pop())
 128
 129     with pytest.raises(IndexError):
 130         call_nominatim()
 131
 132     assert index_mock.called == 4
 133
 134
 135 def test_replication_update_continuous_no_change(monkeypatch, init_status, index_mock):
 136     states = [nominatim.tools.replication.UpdateState.NO_CHANGES,
 137               nominatim.tools.replication.UpdateState.UP_TO_DATE]
 138     monkeypatch.setattr(nominatim.tools.replication, 'update',
 139                         lambda *args, **kwargs: states.pop())
 140
 141     sleep_mock = MockParamCapture()
 142     monkeypatch.setattr(time, 'sleep', sleep_mock)
 143
 144     with pytest.raises(IndexError):
 145         call_nominatim()
 146
 147     assert index_mock.called == 2
 148     assert sleep_mock.called == 1
 149     assert sleep_mock.last_args[0] == 60