import nominatim.cli
+class MockParamCapture:
+ """ Mock that records the parameters with which a function was called
+ as well as the number of calls.
+ """
+ def __init__(self, retval=0):
+ self.called = 0
+ self.return_value = retval
+ self.last_args = None
+ self.last_kwargs = None
+
+ def __call__(self, *args, **kwargs):
+ self.called += 1
+ self.last_args = args
+ self.last_kwargs = kwargs
+ return self.return_value
+
+
+
@pytest.fixture
def cli_call(src_dir):
""" Call the nominatim main function with the correct paths set.
return _call_nominatim
+
+@pytest.fixture
+def mock_run_legacy(monkeypatch):
+ mock = MockParamCapture()
+ monkeypatch.setattr(nominatim.cli, 'run_legacy_script', mock)
+ return mock
+
+
+@pytest.fixture
+def mock_func_factory(monkeypatch):
+ def get_mock(module, func):
+ mock = MockParamCapture()
+ mock.func_name = func
+ monkeypatch.setattr(module, func, mock)
+ return mock
+
+ return get_mock
import nominatim.tools.replication
from nominatim.db import status
-from mocks import MockParamCapture
-
@pytest.fixture
def tokenizer_mock(monkeypatch):
class DummyTokenizer:
return tok
-@pytest.fixture
-def mock_func_factory(monkeypatch):
- def get_mock(module, func):
- mock = MockParamCapture()
- monkeypatch.setattr(module, func, mock)
- return mock
-
- return get_mock
-
@pytest.fixture
def init_status(temp_db_conn, status_table):
@pytest.fixture
-def index_mock(monkeypatch, tokenizer_mock, init_status):
- mock = MockParamCapture()
- monkeypatch.setattr(nominatim.indexer.indexer.Indexer, 'index_full', mock)
-
- return mock
+def index_mock(mock_func_factory, tokenizer_mock, init_status):
+ return mock_func_factory(nominatim.indexer.indexer.Indexer, 'index_full')
@pytest.fixture
def setup_cli_call(self, cli_call, temp_db):
self.call_nominatim = lambda *args: cli_call('replication', *args)
+
+ @pytest.fixture(autouse=True)
+ def setup_update_function(self, monkeypatch):
+ def _mock_updates(states):
+ monkeypatch.setattr(nominatim.tools.replication, 'update',
+ lambda *args, **kwargs: states.pop())
+
+ self.update_states = _mock_updates
+
+
@pytest.mark.parametrize("params,func", [
(('--init', '--no-update-functions'), 'init_replication'),
(('--check-for-updates',), 'check_for_updates')
@pytest.mark.parametrize("update_interval", [60, 3600])
- def test_replication_catchup(self, monkeypatch, index_mock, update_interval, placex_table):
+ def test_replication_catchup(self, placex_table, monkeypatch, index_mock, update_interval):
monkeypatch.setenv('NOMINATIM_REPLICATION_UPDATE_INTERVAL', str(update_interval))
- states = [nominatim.tools.replication.UpdateState.NO_CHANGES]
- monkeypatch.setattr(nominatim.tools.replication, 'update',
- lambda *args, **kwargs: states.pop())
+ self.update_states([nominatim.tools.replication.UpdateState.NO_CHANGES])
assert self.call_nominatim('--catch-up') == 0
assert update_mock.last_args[1]['threads'] == 4
- def test_replication_update_continuous(self, monkeypatch, index_mock):
- states = [nominatim.tools.replication.UpdateState.UP_TO_DATE,
- nominatim.tools.replication.UpdateState.UP_TO_DATE]
- monkeypatch.setattr(nominatim.tools.replication, 'update',
- lambda *args, **kwargs: states.pop())
+ def test_replication_update_continuous(self, index_mock):
+ self.update_states([nominatim.tools.replication.UpdateState.UP_TO_DATE,
+ nominatim.tools.replication.UpdateState.UP_TO_DATE])
with pytest.raises(IndexError):
self.call_nominatim()
assert index_mock.called == 2
- def test_replication_update_continuous_no_change(self, monkeypatch, index_mock):
- states = [nominatim.tools.replication.UpdateState.NO_CHANGES,
- nominatim.tools.replication.UpdateState.UP_TO_DATE]
- monkeypatch.setattr(nominatim.tools.replication, 'update',
- lambda *args, **kwargs: states.pop())
+ def test_replication_update_continuous_no_change(self, mock_func_factory,
+ index_mock):
+ self.update_states([nominatim.tools.replication.UpdateState.NO_CHANGES,
+ nominatim.tools.replication.UpdateState.UP_TO_DATE])
- sleep_mock = MockParamCapture()
- monkeypatch.setattr(time, 'sleep', sleep_mock)
+ sleep_mock = mock_func_factory(time, 'sleep')
with pytest.raises(IndexError):
self.call_nominatim()