mirror of
https://github.com/osm-search/Nominatim.git
synced 2026-02-15 19:07:58 +00:00
7222235579
Allows to define special functions over the arguments. Also splits CLI tests in two files as they have become too many.
128 lines
4.2 KiB
Python
128 lines
4.2 KiB
Python
"""
|
|
Tests for replication command of command-line interface wrapper.
|
|
"""
|
|
import datetime as dt
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
import nominatim.cli
|
|
import nominatim.indexer.indexer
|
|
import nominatim.tools.replication
|
|
from nominatim.db import status
|
|
|
|
from mocks import MockParamCapture
|
|
|
|
SRC_DIR = (Path(__file__) / '..' / '..' / '..').resolve()
|
|
|
|
def call_nominatim(*args):
|
|
return nominatim.cli.nominatim(module_dir='build/module',
|
|
osm2pgsql_path='build/osm2pgsql/osm2pgsql',
|
|
phplib_dir=str(SRC_DIR / 'lib-php'),
|
|
data_dir=str(SRC_DIR / 'data'),
|
|
phpcgi_path='/usr/bin/php-cgi',
|
|
sqllib_dir=str(SRC_DIR / 'lib-sql'),
|
|
config_dir=str(SRC_DIR / 'settings'),
|
|
cli_args=['replication'] + list(args))
|
|
|
|
@pytest.fixture
|
|
def index_mock(monkeypatch):
|
|
mock = MockParamCapture()
|
|
monkeypatch.setattr(nominatim.indexer.indexer.Indexer, 'index_boundaries', mock)
|
|
monkeypatch.setattr(nominatim.indexer.indexer.Indexer, 'index_by_rank', mock)
|
|
|
|
return mock
|
|
|
|
|
|
@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):
|
|
status.set_status(temp_db_conn, date=dt.datetime.now(dt.timezone.utc), seq=1)
|
|
return 1
|
|
|
|
|
|
@pytest.fixture
|
|
def update_mock(mock_func_factory, init_status):
|
|
return mock_func_factory(nominatim.tools.replication, 'update')
|
|
|
|
@pytest.mark.parametrize("params,func", [
|
|
(('--init', '--no-update-functions'), 'init_replication'),
|
|
(('--check-for-updates',), 'check_for_updates')
|
|
])
|
|
def test_replication_command(mock_func_factory, temp_db, params, func):
|
|
func_mock = mock_func_factory(nominatim.tools.replication, func)
|
|
|
|
assert 0 == call_nominatim(*params)
|
|
assert func_mock.called == 1
|
|
|
|
|
|
def test_replication_update_bad_interval(monkeypatch, temp_db):
|
|
monkeypatch.setenv('NOMINATIM_REPLICATION_UPDATE_INTERVAL', 'xx')
|
|
|
|
assert call_nominatim() == 1
|
|
|
|
|
|
def test_replication_update_bad_interval_for_geofabrik(monkeypatch, temp_db):
|
|
monkeypatch.setenv('NOMINATIM_REPLICATION_URL',
|
|
'https://download.geofabrik.de/europe/ireland-and-northern-ireland-updates')
|
|
|
|
assert call_nominatim() == 1
|
|
|
|
|
|
def test_replication_update_once_no_index(update_mock):
|
|
assert 0 == call_nominatim('--once', '--no-index')
|
|
|
|
assert str(update_mock.last_args[1]['osm2pgsql']) == 'build/osm2pgsql/osm2pgsql'
|
|
|
|
|
|
def test_replication_update_custom_osm2pgsql(monkeypatch, update_mock):
|
|
monkeypatch.setenv('NOMINATIM_OSM2PGSQL_BINARY', '/secret/osm2pgsql')
|
|
assert 0 == call_nominatim('--once', '--no-index')
|
|
|
|
assert str(update_mock.last_args[1]['osm2pgsql']) == '/secret/osm2pgsql'
|
|
|
|
|
|
def test_replication_update_custom_threads(update_mock):
|
|
assert 0 == call_nominatim('--once', '--no-index', '--threads', '4')
|
|
|
|
assert update_mock.last_args[1]['threads'] == 4
|
|
|
|
|
|
def test_replication_update_continuous(monkeypatch, init_status, 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())
|
|
|
|
with pytest.raises(IndexError):
|
|
call_nominatim()
|
|
|
|
assert index_mock.called == 4
|
|
|
|
|
|
def test_replication_update_continuous_no_change(monkeypatch, init_status, 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())
|
|
|
|
sleep_mock = MockParamCapture()
|
|
monkeypatch.setattr(time, 'sleep', sleep_mock)
|
|
|
|
with pytest.raises(IndexError):
|
|
call_nominatim()
|
|
|
|
assert index_mock.called == 2
|
|
assert sleep_mock.called == 1
|
|
assert sleep_mock.last_args[0] == 60
|