test/python/test_tools_postcodes.py

   1 """
   2 Tests for functions to maintain the artificial postcode table.
   3 """
   4 import subprocess
   5
   6 import pytest
   7
   8 from nominatim.tools import postcodes
   9 import dummy_tokenizer
  10
  11 class MockPostcodeTable:
  12     """ A location_postcode table for testing.
  13     """
  14     def __init__(self, conn):
  15         self.conn = conn
  16         with conn.cursor() as cur:
  17             cur.execute("""CREATE TABLE location_postcode (
  18                                place_id BIGINT,
  19                                parent_place_id BIGINT,
  20                                rank_search SMALLINT,
  21                                rank_address SMALLINT,
  22                                indexed_status SMALLINT,
  23                                indexed_date TIMESTAMP,
  24                                country_code varchar(2),
  25                                postcode TEXT,
  26                                geometry GEOMETRY(Geometry, 4326))""")
  27             cur.execute("""CREATE OR REPLACE FUNCTION token_normalized_postcode(postcode TEXT)
  28                            RETURNS TEXT AS $$ BEGIN RETURN postcode; END; $$ LANGUAGE plpgsql;
  29
  30                            CREATE OR REPLACE FUNCTION get_country_code(place geometry)
  31                            RETURNS TEXT AS $$ BEGIN
  32                            RETURN null;
  33                            END; $$ LANGUAGE plpgsql;
  34                         """)
  35         conn.commit()
  36
  37     def add(self, country, postcode, x, y):
  38         with self.conn.cursor() as cur:
  39             cur.execute("""INSERT INTO location_postcode (place_id, indexed_status,
  40                                                           country_code, postcode,
  41                                                           geometry)
  42                            VALUES (nextval('seq_place'), 1, %s, %s,
  43                                    'SRID=4326;POINT(%s %s)')""",
  44                         (country, postcode, x, y))
  45         self.conn.commit()
  46
  47
  48     @property
  49     def row_set(self):
  50         with self.conn.cursor() as cur:
  51             cur.execute("""SELECT country_code, postcode,
  52                                   ST_X(geometry), ST_Y(geometry)
  53                            FROM location_postcode""")
  54             return set((tuple(row) for row in cur))
  55
  56
  57 @pytest.fixture
  58 def tokenizer():
  59     return dummy_tokenizer.DummyTokenizer(None, None)
  60
  61 @pytest.fixture
  62 def postcode_table(temp_db_conn, placex_table, word_table):
  63     return MockPostcodeTable(temp_db_conn)
  64
  65
  66 def test_postcodes_empty(dsn, postcode_table, place_table,
  67                          tmp_path, tokenizer):
  68     postcodes.update_postcodes(dsn, tmp_path, tokenizer)
  69
  70     assert not postcode_table.row_set
  71
  72
  73 def test_postcodes_add_new(dsn, postcode_table, tmp_path,
  74                            insert_implicit_postcode, tokenizer):
  75     insert_implicit_postcode(1, 'xx', 'POINT(10 12)', dict(postcode='9486'))
  76     postcode_table.add('yy', '9486', 99, 34)
  77
  78     postcodes.update_postcodes(dsn, tmp_path, tokenizer)
  79
  80     assert postcode_table.row_set == {('xx', '9486', 10, 12), }
  81
  82
  83 def test_postcodes_replace_coordinates(dsn, postcode_table, tmp_path,
  84                                        insert_implicit_postcode, tokenizer):
  85     insert_implicit_postcode(1, 'xx', 'POINT(10 12)', dict(postcode='AB 4511'))
  86     postcode_table.add('xx', 'AB 4511', 99, 34)
  87
  88     postcodes.update_postcodes(dsn, tmp_path, tokenizer)
  89
  90     assert postcode_table.row_set == {('xx', 'AB 4511', 10, 12)}
  91
  92
  93 def test_postcodes_replace_coordinates_close(dsn, postcode_table, tmp_path,
  94                                              insert_implicit_postcode, tokenizer):
  95     insert_implicit_postcode(1, 'xx', 'POINT(10 12)', dict(postcode='AB 4511'))
  96     postcode_table.add('xx', 'AB 4511', 10, 11.99999)
  97
  98     postcodes.update_postcodes(dsn, tmp_path, tokenizer)
  99
 100     assert postcode_table.row_set == {('xx', 'AB 4511', 10, 11.99999)}
 101
 102
 103 def test_postcodes_remove(dsn, postcode_table, tmp_path,
 104                           insert_implicit_postcode, tokenizer):
 105     insert_implicit_postcode(1, 'xx', 'POINT(10 12)', dict(postcode='AB 4511'))
 106     postcode_table.add('xx', 'badname', 10, 12)
 107
 108     postcodes.update_postcodes(dsn, tmp_path, tokenizer)
 109
 110     assert postcode_table.row_set == {('xx', 'AB 4511', 10, 12)}
 111
 112
 113 def test_postcodes_ignore_empty_country(dsn, postcode_table, tmp_path,
 114                                         insert_implicit_postcode, tokenizer):
 115     insert_implicit_postcode(1, None, 'POINT(10 12)', dict(postcode='AB 4511'))
 116     postcodes.update_postcodes(dsn, tmp_path, tokenizer)
 117     print(postcode_table.row_set)
 118     assert not postcode_table.row_set
 119
 120
 121 def test_postcodes_remove_all(dsn, postcode_table, place_table,
 122                               tmp_path, tokenizer):
 123     postcode_table.add('ch', '5613', 10, 12)
 124     postcodes.update_postcodes(dsn, tmp_path, tokenizer)
 125
 126     assert not postcode_table.row_set
 127
 128
 129 def test_postcodes_multi_country(dsn, postcode_table, tmp_path,
 130                                  insert_implicit_postcode, tokenizer):
 131     insert_implicit_postcode(1, 'de', 'POINT(10 12)', dict(postcode='54451'))
 132     insert_implicit_postcode(2, 'cc', 'POINT(100 56)', dict(postcode='DD23 T'))
 133     insert_implicit_postcode(3, 'de', 'POINT(10.3 11.0)', dict(postcode='54452'))
 134     insert_implicit_postcode(4, 'cc', 'POINT(10.3 11.0)', dict(postcode='54452'))
 135
 136     postcodes.update_postcodes(dsn, tmp_path, tokenizer)
 137
 138     assert postcode_table.row_set == {('de', '54451', 10, 12),
 139                                       ('de', '54452', 10.3, 11.0),
 140                                       ('cc', '54452', 10.3, 11.0),
 141                                       ('cc', 'DD23 T', 100, 56)}
 142
 143
 144 @pytest.mark.parametrize("gzipped", [True, False])
 145 def test_postcodes_extern(dsn, postcode_table, tmp_path,
 146                           insert_implicit_postcode, tokenizer, gzipped):
 147     insert_implicit_postcode(1, 'xx', 'POINT(10 12)', dict(postcode='AB 4511'))
 148
 149     extfile = tmp_path / 'xx_postcodes.csv'
 150     extfile.write_text("postcode,lat,lon\nAB 4511,-4,-1\nCD 4511,-5, -10")
 151
 152     if gzipped:
 153         subprocess.run(['gzip', str(extfile)])
 154         assert not extfile.is_file()
 155
 156     postcodes.update_postcodes(dsn, tmp_path, tokenizer)
 157
 158     assert postcode_table.row_set == {('xx', 'AB 4511', 10, 12),
 159                                       ('xx', 'CD 4511', -10, -5)}
 160
 161
 162 def test_postcodes_extern_bad_column(dsn, postcode_table, tmp_path,
 163                                      insert_implicit_postcode, tokenizer):
 164     insert_implicit_postcode(1, 'xx', 'POINT(10 12)', dict(postcode='AB 4511'))
 165
 166     extfile = tmp_path / 'xx_postcodes.csv'
 167     extfile.write_text("postode,lat,lon\nAB 4511,-4,-1\nCD 4511,-5, -10")
 168
 169     postcodes.update_postcodes(dsn, tmp_path, tokenizer)
 170
 171     assert postcode_table.row_set == {('xx', 'AB 4511', 10, 12)}
 172
 173
 174 def test_postcodes_extern_bad_number(dsn, insert_implicit_postcode,
 175                                      postcode_table, tmp_path, tokenizer):
 176     insert_implicit_postcode(1, 'xx', 'POINT(10 12)', dict(postcode='AB 4511'))
 177
 178     extfile = tmp_path / 'xx_postcodes.csv'
 179     extfile.write_text("postcode,lat,lon\nXX 4511,-4,NaN\nCD 4511,-5, -10\n34,200,0")
 180
 181     postcodes.update_postcodes(dsn, tmp_path, tokenizer)
 182
 183     assert postcode_table.row_set == {('xx', 'AB 4511', 10, 12),
 184                                       ('xx', 'CD 4511', -10, -5)}
 185
 186 def test_can_compute(dsn, table_factory):
 187     assert not postcodes.can_compute(dsn)
 188     table_factory('place')
 189     assert postcodes.can_compute(dsn)
 190
 191 def test_no_placex_entry(dsn, tmp_path, temp_db_cursor, place_row, postcode_table, tokenizer):
 192     #Rewrite the get_country_code function to verify its execution.
 193     temp_db_cursor.execute("""
 194         CREATE OR REPLACE FUNCTION get_country_code(place geometry)
 195         RETURNS TEXT AS $$ BEGIN
 196         RETURN 'fr';
 197         END; $$ LANGUAGE plpgsql;
 198     """)
 199     place_row(geom='SRID=4326;POINT(10 12)', address=dict(postcode='AB 4511'))
 200     postcodes.update_postcodes(dsn, tmp_path, tokenizer)
 201
 202     assert postcode_table.row_set == {('fr', 'AB 4511', 10, 12)}
 203
 204 @pytest.fixture
 205 def insert_implicit_postcode(placex_table, place_row):
 206     """
 207         Inserts data into the placex and place table
 208         which can then be used to compute one postcode.
 209     """
 210     def _insert_implicit_postcode(osm_id, country, geometry, address):
 211         placex_table.add(osm_id=osm_id, country=country, geom=geometry)
 212         place_row(osm_id=osm_id, geom='SRID=4326;'+geometry, address=address)
 213
 214     return _insert_implicit_postcode