test/python/tools/test_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 functionality.
   9 """
  10 import datetime as dt
  11 import time
  12
  13 import pytest
  14 from osmium.replication.server import OsmosisState
  15
  16 import nominatim.tools.replication
  17 import nominatim.db.status as status
  18 from nominatim.errors import UsageError
  19
  20 OSM_NODE_DATA = """\
  21 <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/">
  22 <node id="100" visible="true" version="1" changeset="2047" timestamp="2006-01-27T22:09:10Z" user="Foo" uid="111" lat="48.7586670" lon="8.1343060">
  23 </node>
  24 </osm>
  25 """
  26
  27 @pytest.fixture(autouse=True)
  28 def setup_status_table(status_table):
  29     pass
  30
  31 ### init replication
  32
  33 def test_init_replication_bad_base_url(monkeypatch, place_row, temp_db_conn):
  34     place_row(osm_type='N', osm_id=100)
  35
  36     monkeypatch.setattr(nominatim.db.status, "get_url", lambda u: OSM_NODE_DATA)
  37
  38     with pytest.raises(UsageError, match="Failed to reach replication service"):
  39         nominatim.tools.replication.init_replication(temp_db_conn, 'https://test.io')
  40
  41
  42 def test_init_replication_success(monkeypatch, place_row, temp_db_conn, temp_db_cursor):
  43     place_row(osm_type='N', osm_id=100)
  44
  45     monkeypatch.setattr(nominatim.db.status, "get_url", lambda u: OSM_NODE_DATA)
  46     monkeypatch.setattr(nominatim.tools.replication.ReplicationServer,
  47                         "timestamp_to_sequence",
  48                         lambda self, date: 234)
  49
  50     nominatim.tools.replication.init_replication(temp_db_conn, 'https://test.io')
  51
  52     expected_date = dt.datetime.strptime('2006-01-27T19:09:10', status.ISODATE_FORMAT)\
  53                         .replace(tzinfo=dt.timezone.utc)
  54
  55     assert temp_db_cursor.row_set("SELECT * FROM import_status") \
  56              == {(expected_date, 234, True)}
  57
  58
  59 ### checking for updates
  60
  61 def test_check_for_updates_empty_status_table(temp_db_conn):
  62     assert nominatim.tools.replication.check_for_updates(temp_db_conn, 'https://test.io') == 254
  63
  64
  65 def test_check_for_updates_seq_not_set(temp_db_conn):
  66     status.set_status(temp_db_conn, dt.datetime.now(dt.timezone.utc))
  67
  68     assert nominatim.tools.replication.check_for_updates(temp_db_conn, 'https://test.io') == 254
  69
  70
  71 def test_check_for_updates_no_state(monkeypatch, temp_db_conn):
  72     status.set_status(temp_db_conn, dt.datetime.now(dt.timezone.utc), seq=345)
  73
  74     monkeypatch.setattr(nominatim.tools.replication.ReplicationServer,
  75                         "get_state_info", lambda self: None)
  76
  77     assert nominatim.tools.replication.check_for_updates(temp_db_conn, 'https://test.io') == 253
  78
  79
  80 @pytest.mark.parametrize("server_sequence,result", [(344, 2), (345, 2), (346, 0)])
  81 def test_check_for_updates_no_new_data(monkeypatch, temp_db_conn,
  82                                        server_sequence, result):
  83     date = dt.datetime.now(dt.timezone.utc)
  84     status.set_status(temp_db_conn, date, seq=345)
  85
  86     monkeypatch.setattr(nominatim.tools.replication.ReplicationServer,
  87                         "get_state_info",
  88                         lambda self: OsmosisState(server_sequence, date))
  89
  90     assert nominatim.tools.replication.check_for_updates(temp_db_conn, 'https://test.io') == result
  91
  92
  93 ### updating
  94
  95 @pytest.fixture
  96 def update_options(tmpdir):
  97     return dict(base_url='https://test.io',
  98                 indexed_only=False,
  99                 update_interval=3600,
 100                 import_file=tmpdir / 'foo.osm',
 101                 max_diff_size=1)
 102
 103 def test_update_empty_status_table(dsn):
 104     with pytest.raises(UsageError):
 105         nominatim.tools.replication.update(dsn, {})
 106
 107
 108 def test_update_already_indexed(temp_db_conn, dsn):
 109     status.set_status(temp_db_conn, dt.datetime.now(dt.timezone.utc), seq=34, indexed=False)
 110
 111     assert nominatim.tools.replication.update(dsn, dict(indexed_only=True)) \
 112              == nominatim.tools.replication.UpdateState.MORE_PENDING
 113
 114
 115 def test_update_no_data_no_sleep(monkeypatch, temp_db_conn, dsn, update_options):
 116     date = dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=1)
 117     status.set_status(temp_db_conn, date, seq=34)
 118
 119     monkeypatch.setattr(nominatim.tools.replication.ReplicationServer,
 120                         "apply_diffs",
 121                         lambda *args, **kwargs: None)
 122
 123     sleeptime = []
 124     monkeypatch.setattr(time, 'sleep', sleeptime.append)
 125
 126     assert nominatim.tools.replication.update(dsn, update_options) \
 127              == nominatim.tools.replication.UpdateState.NO_CHANGES
 128
 129     assert not sleeptime
 130
 131
 132 def test_update_no_data_sleep(monkeypatch, temp_db_conn, dsn, update_options):
 133     date = dt.datetime.now(dt.timezone.utc) - dt.timedelta(minutes=30)
 134     status.set_status(temp_db_conn, date, seq=34)
 135
 136     monkeypatch.setattr(nominatim.tools.replication.ReplicationServer,
 137                         "apply_diffs",
 138                         lambda *args, **kwargs: None)
 139
 140     sleeptime = []
 141     monkeypatch.setattr(time, 'sleep', sleeptime.append)
 142
 143     assert nominatim.tools.replication.update(dsn, update_options) \
 144              == nominatim.tools.replication.UpdateState.NO_CHANGES
 145
 146     assert len(sleeptime) == 1
 147     assert sleeptime[0] < 3600
 148     assert sleeptime[0] > 0