hans/Nominatim
1
0
Fork 1
mirror of https://github.com/osm-search/Nominatim.git synced 2026-02-15 02:47:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

port check-for-update function to python

Browse Source
This commit is contained in:
Sarah Hoffmann
2021-01-28 14:34:17 +01:00
parent d78f0ba804
commit 8f0885f6cb
8 changed files with 98 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
test/python/test_tools_replication.py
View File

@@ -4,8 +4,10 @@ Tests for replication functionality.
import datetime as dt
import pytest
from osmium.replication.server import OsmosisState
import nominatim.tools.replication
import nominatim.db.status as status
OSM_NODE_DATA = """\
<osm version="0.6" generator="OpenStreetMap server" copyright="OpenStreetMap and contributors" attribution="http://www.openstreetmap.org/copyright" license="http://opendatacommons.org/licenses/odbl/1-0/">
@@ -39,3 +41,37 @@ def test_init_replication_success(monkeypatch, status_table, place_row, temp_db_
expected_date = dt.datetime.fromisoformat('2006-01-27T19:09:10').replace(tzinfo=dt.timezone.utc)
assert temp_db_cursor.rowcount == 1
assert temp_db_cursor.fetchone() == [expected_date, 234, True]
def test_check_for_updates_empty_status_table(status_table, temp_db_conn):
assert nominatim.tools.replication.check_for_updates(temp_db_conn, 'https://test.io') == 254
def test_check_for_updates_seq_not_set(status_table, temp_db_conn):
status.set_status(temp_db_conn, dt.datetime.now().replace(tzinfo=dt.timezone.utc))
assert nominatim.tools.replication.check_for_updates(temp_db_conn, 'https://test.io') == 254
def test_check_for_updates_no_state(monkeypatch, status_table, temp_db_conn):
status.set_status(temp_db_conn,
dt.datetime.now().replace(tzinfo=dt.timezone.utc),
seq=345)
monkeypatch.setattr(nominatim.tools.replication.ReplicationServer,
"get_state_info", lambda self: None)
assert nominatim.tools.replication.check_for_updates(temp_db_conn, 'https://test.io') == 253
@pytest.mark.parametrize("server_sequence,result", [(344, 1), (345, 1), (346, 0)])
def test_check_for_updates_no_new_data(monkeypatch, status_table, temp_db_conn,
server_sequence, result):
date = dt.datetime.now().replace(tzinfo=dt.timezone.utc)
status.set_status(temp_db_conn, date, seq=345)
monkeypatch.setattr(nominatim.tools.replication.ReplicationServer,
"get_state_info",
lambda self: OsmosisState(server_sequence, date))
assert nominatim.tools.replication.check_for_updates(temp_db_conn, 'https://test.io') == result