safety-screen/Nominatim
1
0
Fork 0
forked from hans/Nominatim
Code Pull Requests Activity

generalize fixtures for cli tests

Browse Source
This commit is contained in:
Sarah Hoffmann
2021-11-30 14:07:39 +01:00
parent b2df8e478a
commit 37afa2180b
4 changed files with 57 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
test/python/cli/test_cmd_replication.py
View File

@@ -11,8 +11,6 @@ import nominatim.indexer.indexer
import nominatim.tools.replication
from nominatim.db import status
from mocks import MockParamCapture
@pytest.fixture
def tokenizer_mock(monkeypatch):
class DummyTokenizer:
@@ -35,15 +33,6 @@ def tokenizer_mock(monkeypatch):
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):
@@ -51,11 +40,8 @@ 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
@@ -69,6 +55,16 @@ class TestCliReplication:
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')
@@ -107,11 +103,9 @@ class TestCliReplication:
@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
@@ -122,11 +116,9 @@ class TestCliReplication:
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()
@@ -134,14 +126,12 @@ class TestCliReplication:
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()