test/python/test_tools_replication.py

   1 """
   2 Tests for replication functionality.
   3 """
   4 import datetime as dt
   5
   6 import pytest
   7
   8 import nominatim.tools.replication
   9
  10 OSM_NODE_DATA = """\
  11 <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/">
  12 <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">
  13 </node>
  14 </osm>
  15 """
  16
  17
  18 def test_init_replication_bad_base_url(monkeypatch, status_table, place_row, temp_db_conn, temp_db_cursor):
  19     place_row(osm_type='N', osm_id=100)
  20
  21     monkeypatch.setattr(nominatim.db.status, "get_url", lambda u : OSM_NODE_DATA)
  22
  23     with pytest.raises(RuntimeError, match="Failed to reach replication service"):
  24         nominatim.tools.replication.init_replication(temp_db_conn, 'https://test.io')
  25
  26
  27 def test_init_replication_success(monkeypatch, status_table, place_row, temp_db_conn, temp_db_cursor):
  28     place_row(osm_type='N', osm_id=100)
  29
  30     monkeypatch.setattr(nominatim.db.status, "get_url", lambda u : OSM_NODE_DATA)
  31     monkeypatch.setattr(nominatim.tools.replication.ReplicationServer,
  32                         "timestamp_to_sequence",
  33                         lambda self, date: 234)
  34
  35     nominatim.tools.replication.init_replication(temp_db_conn, 'https://test.io')
  36
  37     temp_db_cursor.execute("SELECT * FROM import_status")
  38
  39     expected_date = dt.datetime.fromisoformat('2006-01-27T19:09:10').replace(tzinfo=dt.timezone.utc)
  40     assert temp_db_cursor.rowcount == 1
  41     assert temp_db_cursor.fetchone() == [expected_date, 234, True]