test/python/test_indexing.py

   1 """
   2 Tests for running the indexing.
   3 """
   4 import itertools
   5 import psycopg2
   6 import pytest
   7
   8 from nominatim.indexer.indexer import Indexer
   9
  10 class IndexerTestDB:
  11
  12     def __init__(self, name):
  13         self.name = name
  14         self.conn = None
  15         self.placex_id = itertools.count(100000)
  16         self.osmline_id = itertools.count(500000)
  17
  18     def setup(self):
  19         with psycopg2.connect(database='postgres') as conn:
  20             conn.set_isolation_level(0)
  21             with conn.cursor() as cur:
  22                 cur.execute('DROP DATABASE IF EXISTS {}'.format(self.name))
  23                 cur.execute('CREATE DATABASE {}'.format(self.name))
  24         self.conn = psycopg2.connect(database=self.name)
  25         self.conn.set_isolation_level(0)
  26         with self.conn.cursor() as cur:
  27             cur.execute("""CREATE TABLE placex (place_id BIGINT,
  28                                                 class TEXT,
  29                                                 type TEXT,
  30                                                 rank_address SMALLINT,
  31                                                 rank_search SMALLINT,
  32                                                 indexed_status SMALLINT,
  33                                                 indexed_date TIMESTAMP,
  34                                                 partition SMALLINT,
  35                                                 admin_level SMALLINT,
  36                                                 geometry_sector INTEGER)""")
  37             cur.execute("""CREATE TABLE location_property_osmline (
  38                                place_id BIGINT,
  39                                indexed_status SMALLINT,
  40                                indexed_date TIMESTAMP,
  41                                geometry_sector INTEGER)""")
  42             cur.execute("""CREATE OR REPLACE FUNCTION date_update() RETURNS TRIGGER
  43                            AS $$
  44                            BEGIN
  45                              IF NEW.indexed_status = 0 and OLD.indexed_status != 0 THEN
  46                                NEW.indexed_date = now();
  47                              END IF;
  48                              RETURN NEW;
  49                            END; $$ LANGUAGE plpgsql;""")
  50             cur.execute("""CREATE TRIGGER placex_update BEFORE UPDATE ON placex
  51                            FOR EACH ROW EXECUTE PROCEDURE date_update()""")
  52             cur.execute("""CREATE TRIGGER osmline_update BEFORE UPDATE ON location_property_osmline
  53                            FOR EACH ROW EXECUTE PROCEDURE date_update()""")
  54
  55
  56     def drop(self):
  57         if self.conn:
  58             self.conn.close()
  59             self.conn = None
  60         with psycopg2.connect(database='postgres') as conn:
  61             conn.set_isolation_level(0)
  62             with conn.cursor() as cur:
  63                 cur.execute('DROP DATABASE IF EXISTS {}'.format(self.name))
  64
  65     def scalar(self, query):
  66         with self.conn.cursor() as cur:
  67             cur.execute(query)
  68             return cur.fetchone()[0]
  69
  70     def add_place(self, cls='place', typ='locality',
  71                   rank_search=30, rank_address=30, sector=20):
  72         next_id = next(self.placex_id)
  73         with self.conn.cursor() as cur:
  74             cur.execute("""INSERT INTO placex
  75                               (place_id, class, type, rank_search, rank_address,
  76                                indexed_status, geometry_sector)
  77                               VALUES (%s, %s, %s, %s, %s, 1, %s)""",
  78                         (next_id, cls, typ, rank_search, rank_address, sector))
  79         return next_id
  80
  81     def add_admin(self, **kwargs):
  82         kwargs['cls'] = 'boundary'
  83         kwargs['typ'] = 'administrative'
  84         return self.add_place(**kwargs)
  85
  86     def add_osmline(self, sector=20):
  87         next_id = next(self.osmline_id)
  88         with self.conn.cursor() as cur:
  89             cur.execute("""INSERT INTO location_property_osmline
  90                               (place_id, indexed_status, geometry_sector)
  91                               VALUES (%s, 1, %s)""",
  92                         (next_id, sector))
  93         return next_id
  94
  95     def placex_unindexed(self):
  96         return self.scalar('SELECT count(*) from placex where indexed_status > 0')
  97
  98     def osmline_unindexed(self):
  99         return self.scalar('SELECT count(*) from location_property_osmline where indexed_status > 0')
 100
 101
 102 @pytest.fixture
 103 def test_db():
 104     db = IndexerTestDB('test_nominatim_python_unittest')
 105     db.setup()
 106     yield db
 107     db.drop()
 108
 109
 110 @pytest.mark.parametrize("threads", [1, 15])
 111 def test_index_full(test_db, threads):
 112     for rank in range(31):
 113         test_db.add_place(rank_address=rank, rank_search=rank)
 114     test_db.add_osmline()
 115
 116     assert 31 == test_db.placex_unindexed()
 117     assert 1 == test_db.osmline_unindexed()
 118
 119     idx = Indexer('dbname=test_nominatim_python_unittest', threads)
 120     idx.index_by_rank(0, 30)
 121
 122     assert 0 == test_db.placex_unindexed()
 123     assert 0 == test_db.osmline_unindexed()
 124
 125     assert 0 == test_db.scalar("""SELECT count(*) from placex
 126                                WHERE indexed_status = 0 and indexed_date is null""")
 127     # ranks come in order of rank address
 128     assert 0 == test_db.scalar("""
 129         SELECT count(*) FROM placex p WHERE rank_address > 0
 130           AND indexed_date >= (SELECT min(indexed_date) FROM placex o
 131                                WHERE p.rank_address < o.rank_address)""")
 132     # placex rank < 30 objects come before interpolations
 133     assert 0 == test_db.scalar(
 134         """SELECT count(*) FROM placex WHERE rank_address < 30
 135              AND indexed_date > (SELECT min(indexed_date) FROM location_property_osmline)""")
 136     # placex rank = 30 objects come after interpolations
 137     assert 0 == test_db.scalar(
 138         """SELECT count(*) FROM placex WHERE rank_address = 30
 139              AND indexed_date < (SELECT max(indexed_date) FROM location_property_osmline)""")
 140     # rank 0 comes after rank 29 and before rank 30
 141     assert 0 == test_db.scalar(
 142         """SELECT count(*) FROM placex WHERE rank_address < 30
 143              AND indexed_date > (SELECT min(indexed_date) FROM placex WHERE rank_address = 0)""")
 144     assert 0 == test_db.scalar(
 145         """SELECT count(*) FROM placex WHERE rank_address = 30
 146              AND indexed_date < (SELECT max(indexed_date) FROM placex WHERE rank_address = 0)""")
 147
 148
 149 @pytest.mark.parametrize("threads", [1, 15])
 150 def test_index_partial_without_30(test_db, threads):
 151     for rank in range(31):
 152         test_db.add_place(rank_address=rank, rank_search=rank)
 153     test_db.add_osmline()
 154
 155     assert 31 == test_db.placex_unindexed()
 156     assert 1 == test_db.osmline_unindexed()
 157
 158     idx = Indexer('dbname=test_nominatim_python_unittest', threads)
 159     idx.index_by_rank(4, 15)
 160
 161     assert 19 == test_db.placex_unindexed()
 162     assert 1 == test_db.osmline_unindexed()
 163
 164     assert 0 == test_db.scalar("""
 165                     SELECT count(*) FROM placex
 166                       WHERE indexed_status = 0 AND not rank_address between 4 and 15""")
 167
 168
 169 @pytest.mark.parametrize("threads", [1, 15])
 170 def test_index_partial_with_30(test_db, threads):
 171     for rank in range(31):
 172         test_db.add_place(rank_address=rank, rank_search=rank)
 173     test_db.add_osmline()
 174
 175     assert 31 == test_db.placex_unindexed()
 176     assert 1 == test_db.osmline_unindexed()
 177
 178     idx = Indexer('dbname=test_nominatim_python_unittest', threads)
 179     idx.index_by_rank(28, 30)
 180
 181     assert 27 == test_db.placex_unindexed()
 182     assert 0 == test_db.osmline_unindexed()
 183
 184     assert 0 == test_db.scalar("""
 185                     SELECT count(*) FROM placex
 186                       WHERE indexed_status = 0 AND rank_address between 1 and 27""")
 187
 188 @pytest.mark.parametrize("threads", [1, 15])
 189 def test_index_boundaries(test_db, threads):
 190     for rank in range(4, 10):
 191         test_db.add_admin(rank_address=rank, rank_search=rank)
 192     for rank in range(31):
 193         test_db.add_place(rank_address=rank, rank_search=rank)
 194     test_db.add_osmline()
 195
 196     assert 37 == test_db.placex_unindexed()
 197     assert 1 == test_db.osmline_unindexed()
 198
 199     idx = Indexer('dbname=test_nominatim_python_unittest', threads)
 200     idx.index_boundaries(0, 30)
 201
 202     assert 31 == test_db.placex_unindexed()
 203     assert 1 == test_db.osmline_unindexed()
 204
 205     assert 0 == test_db.scalar("""
 206                     SELECT count(*) FROM placex
 207                       WHERE indexed_status = 0 AND class != 'boundary'""")