test/python/cli/test_cmd_replication.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2022 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Tests for replication command of command-line interface wrapper.
   9 """
  10 import datetime as dt
  11 import time
  12
  13 import pytest
  14
  15 import nominatim.cli
  16 import nominatim.indexer.indexer
  17 import nominatim.tools.replication
  18 import nominatim.tools.refresh
  19 from nominatim.db import status
  20
  21 @pytest.fixture
  22 def tokenizer_mock(monkeypatch):
  23     class DummyTokenizer:
  24         def __init__(self, *args, **kwargs):
  25             self.update_sql_functions_called = False
  26             self.finalize_import_called = False
  27
  28         def update_sql_functions(self, *args):
  29             self.update_sql_functions_called = True
  30
  31         def finalize_import(self, *args):
  32             self.finalize_import_called = True
  33
  34     tok = DummyTokenizer()
  35     monkeypatch.setattr(nominatim.tokenizer.factory, 'get_tokenizer_for_db',
  36                         lambda *args: tok)
  37     monkeypatch.setattr(nominatim.tokenizer.factory, 'create_tokenizer',
  38                         lambda *args: tok)
  39
  40     return tok
  41
  42
  43
  44 @pytest.fixture
  45 def init_status(temp_db_conn, status_table):
  46     status.set_status(temp_db_conn, date=dt.datetime.now(dt.timezone.utc), seq=1)
  47
  48
  49 @pytest.fixture
  50 def index_mock(mock_func_factory, tokenizer_mock, init_status):
  51     return mock_func_factory(nominatim.indexer.indexer.Indexer, 'index_full')
  52
  53
  54 @pytest.fixture
  55 def update_mock(mock_func_factory, init_status, tokenizer_mock):
  56     return mock_func_factory(nominatim.tools.replication, 'update')
  57
  58
  59 class TestCliReplication:
  60
  61     @pytest.fixture(autouse=True)
  62     def setup_cli_call(self, cli_call, temp_db):
  63         self.call_nominatim = lambda *args: cli_call('replication', *args)
  64
  65
  66     @pytest.fixture(autouse=True)
  67     def setup_update_function(self, monkeypatch):
  68         def _mock_updates(states):
  69             monkeypatch.setattr(nominatim.tools.replication, 'update',
  70                             lambda *args, **kwargs: states.pop())
  71
  72         self.update_states = _mock_updates
  73
  74
  75     @pytest.mark.parametrize("params,func", [
  76                              (('--init',), 'init_replication'),
  77                              (('--init', '--no-update-functions'), 'init_replication'),
  78                              (('--check-for-updates',), 'check_for_updates')
  79                              ])
  80     def test_replication_command(self, mock_func_factory, params, func):
  81         func_mock = mock_func_factory(nominatim.tools.replication, func)
  82
  83         if params == ('--init',):
  84             umock = mock_func_factory(nominatim.tools.refresh, 'create_functions')
  85
  86         assert self.call_nominatim(*params) == 0
  87         assert func_mock.called == 1
  88         if params == ('--init',):
  89             assert umock.called == 1
  90
  91
  92     def test_replication_update_bad_interval(self, monkeypatch):
  93         monkeypatch.setenv('NOMINATIM_REPLICATION_UPDATE_INTERVAL', 'xx')
  94
  95         assert self.call_nominatim() == 1
  96
  97
  98     def test_replication_update_bad_interval_for_geofabrik(self, monkeypatch):
  99         monkeypatch.setenv('NOMINATIM_REPLICATION_URL',
 100                            'https://download.geofabrik.de/europe/italy-updates')
 101
 102         assert self.call_nominatim() == 1
 103
 104
 105     def test_replication_update_continuous_no_index(self):
 106         assert self.call_nominatim('--no-index') == 1
 107
 108     def test_replication_update_once_no_index(self, update_mock):
 109         assert self.call_nominatim('--once', '--no-index') == 0
 110
 111         assert str(update_mock.last_args[1]['osm2pgsql']).endswith('OSM2PGSQL NOT AVAILABLE')
 112
 113
 114     def test_replication_update_custom_osm2pgsql(self, monkeypatch, update_mock):
 115         monkeypatch.setenv('NOMINATIM_OSM2PGSQL_BINARY', '/secret/osm2pgsql')
 116         assert self.call_nominatim('--once', '--no-index') == 0
 117
 118         assert str(update_mock.last_args[1]['osm2pgsql']) == '/secret/osm2pgsql'
 119
 120
 121     @pytest.mark.parametrize("update_interval", [60, 3600])
 122     def test_replication_catchup(self, placex_table, monkeypatch, index_mock, update_interval):
 123         monkeypatch.setenv('NOMINATIM_REPLICATION_UPDATE_INTERVAL', str(update_interval))
 124         self.update_states([nominatim.tools.replication.UpdateState.NO_CHANGES])
 125
 126         assert self.call_nominatim('--catch-up') == 0
 127
 128
 129     def test_replication_update_custom_threads(self, update_mock):
 130         assert self.call_nominatim('--once', '--no-index', '--threads', '4') == 0
 131
 132         assert update_mock.last_args[1]['threads'] == 4
 133
 134
 135     def test_replication_update_continuous(self, index_mock):
 136         self.update_states([nominatim.tools.replication.UpdateState.UP_TO_DATE,
 137                             nominatim.tools.replication.UpdateState.UP_TO_DATE])
 138
 139         with pytest.raises(IndexError):
 140             self.call_nominatim()
 141
 142         assert index_mock.called == 2
 143
 144
 145     def test_replication_update_continuous_no_change(self, mock_func_factory,
 146                                                      index_mock):
 147         self.update_states([nominatim.tools.replication.UpdateState.NO_CHANGES,
 148                             nominatim.tools.replication.UpdateState.UP_TO_DATE])
 149
 150         sleep_mock = mock_func_factory(time, 'sleep')
 151
 152         with pytest.raises(IndexError):
 153             self.call_nominatim()
 154
 155         assert index_mock.called == 1
 156         assert sleep_mock.called == 1
 157         assert sleep_mock.last_args[0] == 60