test/python/tools/test_postcodes.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2026 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Tests for functions to maintain the artificial postcode table.
   9 """
  10 import subprocess
  11 import json
  12
  13 import pytest
  14
  15 from psycopg.rows import tuple_row
  16
  17 from nominatim_db.tools import postcodes
  18 from nominatim_db.data import country_info
  19
  20 import dummy_tokenizer
  21
  22
  23 @pytest.fixture
  24 def insert_implicit_postcode(placex_row, place_postcode_row, country_row):
  25     """ Insert data into the placex and place table
  26         which can then be used to compute one postcode.
  27     """
  28     def _insert_implicit_postcode(osm_id, country, geometry, postcode, in_placex=False):
  29         country_row(country=country, names={"name": country})
  30
  31         if in_placex:
  32             placex_row(osm_id=osm_id, country=country, geom=geometry,
  33                        centroid=geometry,
  34                        address={'postcode': postcode})
  35         else:
  36             place_postcode_row(osm_id=osm_id, centroid=geometry,
  37                                country=country, postcode=postcode)
  38     return _insert_implicit_postcode
  39
  40
  41 @pytest.fixture
  42 def insert_postcode_area(place_postcode_row, country_row):
  43     """ Insert an area around a centroid to the postcode table.
  44     """
  45     def _do(osm_id, country, postcode, x, y):
  46         country_row(country=country, names={"name": country})
  47
  48         x1, x2, y1, y2 = x - 0.001, x + 0.001, y - 0.001, y + 0.001
  49         place_postcode_row(osm_type='R', osm_id=osm_id, postcode=postcode, country=country,
  50                            centroid=f"POINT({x} {y})",
  51                            geom=f"POLYGON(({x1} {y1}, {x1} {y2}, {x2} {y2}, {x2} {y1}, {x1} {y1}))")
  52     return _do
  53
  54
  55 @pytest.fixture
  56 def postcode_update(dsn, temp_db_conn):
  57     tokenizer = dummy_tokenizer.DummyTokenizer(None)
  58
  59     def _do(data_path=None):
  60         with temp_db_conn.cursor() as cur:
  61             cur.execute("""CREATE TRIGGER location_postcodes_before_update
  62                             BEFORE UPDATE ON location_postcodes
  63                             FOR EACH ROW EXECUTE PROCEDURE postcodes_update()""")
  64             cur.execute("""CREATE TRIGGER location_postcodes_before_delete
  65                             BEFORE DELETE ON location_postcodes
  66                             FOR EACH ROW EXECUTE PROCEDURE postcodes_delete()""")
  67             cur.execute("""CREATE TRIGGER location_postcodes_before_insert
  68                             BEFORE INSERT ON location_postcodes
  69                             FOR EACH ROW EXECUTE PROCEDURE postcodes_insert()""")
  70         temp_db_conn.commit()
  71         postcodes.update_postcodes(dsn, data_path, tokenizer)
  72     return _do
  73
  74
  75 class TestPostcodes:
  76     @pytest.fixture(autouse=True)
  77     def setup(self, def_config, postcode_table, placex_table, place_postcode_table,
  78               load_sql, temp_db_conn):
  79         self.conn = temp_db_conn
  80         country_info.setup_country_config(def_config)
  81         load_sql('functions/postcode_triggers.sql')
  82
  83         temp_db_conn.execute("""
  84             CREATE OR REPLACE FUNCTION get_country_code(place geometry)
  85             RETURNS TEXT AS $$
  86               SELECT NULL
  87             $$ LANGUAGE sql;
  88
  89             CREATE OR REPLACE FUNCTION expand_by_meters(geom GEOMETRY, meters FLOAT)
  90             RETURNS GEOMETRY AS $$
  91               SELECT ST_Envelope(ST_Buffer(geom::geography, meters, 1)::geometry)
  92             $$ LANGUAGE sql;
  93         """)
  94
  95     @property
  96     def row_set(self):
  97         with self.conn.cursor(row_factory=tuple_row) as cur:
  98             cur.execute("""SELECT osm_id, country_code, postcode,
  99                                   ST_X(centroid), ST_Y(centroid)
 100                            FROM location_postcodes""")
 101             return {r for r in cur}
 102
 103     def test_postcodes_empty(self, postcode_update):
 104         postcode_update()
 105
 106         assert not self.row_set
 107
 108     @pytest.mark.parametrize('in_placex', [True, False])
 109     def test_postcodes_add_new_point(self, postcode_update, postcode_row,
 110                                      insert_implicit_postcode, in_placex):
 111         insert_implicit_postcode(1, 'xx', 'POINT(10 12)', '9486', in_placex)
 112         postcode_row('yy', '9486', 99, 34)
 113
 114         postcode_update()
 115
 116         assert self.row_set == {(None, 'xx', '9486', 10, 12), }
 117
 118     def test_postcodes_add_new_area(self, postcode_update, insert_postcode_area):
 119         insert_postcode_area(345, 'de', '10445', 23.5, 46.2)
 120
 121         postcode_update()
 122
 123         assert self.row_set == {(345, 'de', '10445', 23.5, 46.2)}
 124
 125     @pytest.mark.parametrize('in_placex', [True, False])
 126     def test_postcodes_add_area_and_point(self, postcode_update, insert_postcode_area,
 127                                           insert_implicit_postcode, in_placex):
 128         insert_implicit_postcode(1, 'xx', 'POINT(10 12)', '10445', in_placex)
 129         insert_postcode_area(345, 'xx', '10445', 23.5, 46.2)
 130
 131         postcode_update()
 132
 133         assert self.row_set == {(345, 'xx', '10445', 23.5, 46.2)}
 134
 135     @pytest.mark.parametrize('in_placex', [True, False])
 136     def test_postcodes_add_point_within_area(self, postcode_update, insert_postcode_area,
 137                                              insert_implicit_postcode, in_placex):
 138         insert_implicit_postcode(1, 'xx', 'POINT(23.5 46.2)', '10446', in_placex)
 139         insert_postcode_area(345, 'xx', '10445', 23.5, 46.2)
 140
 141         postcode_update()
 142
 143         assert self.row_set == {(345, 'xx', '10445', 23.5, 46.2)}
 144
 145     @pytest.mark.parametrize('coords', [(99, 34), (10, 34), (99, 12),
 146                                         (9, 34), (9, 11), (23, 11)])
 147     def test_postcodes_replace_coordinates(self, postcode_update, postcode_row, tmp_path,
 148                                            insert_implicit_postcode, coords):
 149         insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
 150         postcode_row('xx', 'AB 4511', *coords)
 151
 152         postcode_update(tmp_path)
 153
 154         assert self.row_set == {(None, 'xx', 'AB 4511', 10, 12)}
 155
 156     def test_postcodes_replace_coordinates_close(self, postcode_update, postcode_row,
 157                                                  insert_implicit_postcode):
 158         insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
 159         postcode_row('xx', 'AB 4511', 10, 11.99999999)
 160
 161         postcode_update()
 162
 163         assert self.row_set == {(None, 'xx', 'AB 4511', 10, 11.99999999)}
 164
 165     def test_postcodes_remove_point(self, postcode_update, postcode_row,
 166                                     insert_implicit_postcode):
 167         insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
 168         postcode_row('xx', 'badname', 10, 12)
 169
 170         postcode_update()
 171
 172         assert self.row_set == {(None, 'xx', 'AB 4511', 10, 12)}
 173
 174     def test_postcodes_ignore_empty_country(self, postcode_update, insert_implicit_postcode):
 175         insert_implicit_postcode(1, None, 'POINT(10 12)', 'AB 4511')
 176         postcode_update()
 177         assert not self.row_set
 178
 179     def test_postcodes_remove_all(self, postcode_update, postcode_row, place_postcode_table):
 180         postcode_row('ch', '5613', 10, 12)
 181         postcode_update()
 182
 183         assert not self.row_set
 184
 185     def test_postcodes_multi_country(self, postcode_update,
 186                                      insert_implicit_postcode):
 187         insert_implicit_postcode(1, 'de', 'POINT(10 12)', '54451')
 188         insert_implicit_postcode(2, 'cc', 'POINT(100 56)', 'DD23 T')
 189         insert_implicit_postcode(3, 'de', 'POINT(10.3 11.0)', '54452')
 190         insert_implicit_postcode(4, 'cc', 'POINT(10.3 11.0)', '54452')
 191
 192         postcode_update()
 193
 194         assert self.row_set == {(None, 'de', '54451', 10, 12),
 195                                 (None, 'de', '54452', 10.3, 11.0),
 196                                 (None, 'cc', '54452', 10.3, 11.0),
 197                                 (None, 'cc', 'DD23 T', 100, 56)}
 198
 199     @pytest.mark.parametrize("gzipped", [True, False])
 200     def test_postcodes_extern_jsonl(self, postcode_update, tmp_path,
 201                                     insert_implicit_postcode, gzipped):
 202         insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
 203
 204         extfile = tmp_path / 'xx_postcodes_geometry.jsonl'
 205         extfile.write_text(
 206             json.dumps({'properties': {'postcode': 'CD 4511'},
 207                         # Centroid : 0.5, 0.5
 208                         'geometry': {'type': 'Polygon', 'coordinates': [[
 209                             [0, 0], [0, 1], [1, 1], [1, 0], [0, 0]]]}}) + '\n' +
 210             json.dumps({'properties': {'postcode': '822114', 'lat': 0.1, 'lon': 0.2},
 211                         'geometry': {'type': 'Polygon', 'coordinates': [[
 212                             [0, 0], [0, 1], [1, 1], [1, 0], [0, 0]]]}}) + '\n', encoding='utf-8')
 213
 214         if gzipped:
 215             subprocess.run(['gzip', str(extfile)])
 216             assert not extfile.is_file()
 217
 218         postcode_update(tmp_path)
 219
 220         assert self.row_set == {(None, 'xx', 'AB 4511', 10, 12),
 221                                 (None, 'xx', 'CD 4511', 0.5, 0.5),
 222                                 (None, 'xx', '822114', 0.2, 0.1)}
 223
 224     def test_postcodes_extern_jsonl_invalid_data(self, postcode_update, tmp_path,
 225                                                  insert_implicit_postcode):
 226         insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
 227
 228         extfile = tmp_path / 'xx_postcodes_geometry.jsonl'
 229         extfile.write_text(
 230             # Invalid JSON
 231             '{"geometry": {"type": "Polygon"}, "properties": {"postcode": "BAD"}\n'
 232             # Missing geometry
 233             + json.dumps({'properties': {'postcode': 'NOGEOM'}}) + '\n'
 234             # Invalid geometry type
 235             + json.dumps({'geometry': {'type': 'Point', 'coordinates': [0, 0]},
 236                           'properties': {'postcode': 'BADGEOM'}}) + '\n'
 237             # Missing postcode
 238             + json.dumps({'properties': {'lat': 0, 'lon': 0},
 239                           'geometry': {'type': 'Polygon', 'coordinates': [[
 240                               [0, 0], [0, 1], [1, 1], [1, 0], [0, 0]]]}}) + '\n'
 241             # Valid one
 242             + json.dumps({'geometry': {'type': 'Polygon', 'coordinates': [[
 243                 [0, 0], [0, 1], [1, 1], [1, 0], [0, 0]]]},
 244                 'properties': {'postcode': 'GOOD'}}) + '\n', encoding='utf-8')
 245
 246         postcode_update(tmp_path)
 247
 248         assert self.row_set == {(None, 'xx', 'AB 4511', 10, 12),
 249                                 (None, 'xx', 'GOOD', 0.5, 0.5)}
 250
 251     def test_postcodes_extern_jsonl_overwrite_past_import(self, postcode_update, tmp_path,
 252                                                           postcode_row, country_row):
 253         country_row(country="xx", names={"name": "xx"})
 254         postcode_row('xx', '822114', 83.8, 24.1, True)  # area from past geometry import
 255         postcode_row('xx', '110000', 77.2, 28.6, True)
 256
 257         extfile = tmp_path / 'xx_postcodes_geometry.jsonl'
 258         extfile.write_text(
 259             # overwrites past postcode
 260             json.dumps({'properties': {'postcode': '822114'},
 261                         'geometry': {'type': 'Polygon', 'coordinates': [[
 262                             [0, 0], [0, 1], [1, 1], [1, 0], [0, 0]]]}}) + '\n', encoding='utf-8')
 263
 264         postcode_update(tmp_path)
 265
 266         # pc 110000 no longer exist in import file, thus deleted
 267         assert self.row_set == {(None, 'xx', '822114', 0.5, 0.5)}
 268
 269     @pytest.mark.parametrize("gzipped", [True, False])
 270     def test_postcodes_extern_csv(self, postcode_update, tmp_path,
 271                                   insert_implicit_postcode, gzipped):
 272         insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
 273
 274         extfile = tmp_path / 'xx_postcodes.csv'
 275         extfile.write_text("postcode,lat,lon\nAB 4511,-4,-1\nCD 4511,-5, -10", encoding='utf-8')
 276
 277         if gzipped:
 278             subprocess.run(['gzip', str(extfile)])
 279             assert not extfile.is_file()
 280
 281         postcode_update(tmp_path)
 282
 283         assert self.row_set == {(None, 'xx', 'AB 4511', 10, 12),
 284                                 (None, 'xx', 'CD 4511', -10, -5)}
 285
 286     def test_postcodes_extern_csv_bad_column(self, postcode_update, tmp_path,
 287                                              insert_implicit_postcode):
 288         insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
 289
 290         extfile = tmp_path / 'xx_postcodes.csv'
 291         extfile.write_text("postode,lat,lon\nAB 4511,-4,-1\nCD 4511,-5, -10", encoding='utf-8')
 292
 293         postcode_update(tmp_path)
 294
 295         assert self.row_set == {(None, 'xx', 'AB 4511', 10, 12)}
 296
 297     def test_postcodes_extern_csv_bad_number(self, postcode_update, insert_implicit_postcode,
 298                                              tmp_path):
 299         insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
 300
 301         extfile = tmp_path / 'xx_postcodes.csv'
 302         extfile.write_text(
 303             "postcode,lat,lon\nXX 4511,-4,NaN\nCD 4511,-5, -10\n34,200,0", encoding='utf-8')
 304
 305         postcode_update(tmp_path)
 306
 307         assert self.row_set == {(None, 'xx', 'AB 4511', 10, 12),
 308                                 (None, 'xx', 'CD 4511', -10, -5)}
 309
 310     def test_postcodes_import_precedence(self, postcode_update, tmp_path,
 311                                          insert_implicit_postcode, insert_postcode_area):
 312         # osm area precedes all external imoprts
 313         insert_postcode_area(3, 'xx', '110000', 77.2, 28.6)
 314         # guessed postcode area from osm points, should be overwritten by geometry import
 315         insert_implicit_postcode(1, 'xx', 'POINT(10 12)', '822114')
 316         insert_implicit_postcode(2, 'xx', 'POINT(0 12)', '822114')
 317
 318         josnlfile = tmp_path / 'xx_postcodes_geometry.jsonl'
 319         josnlfile.write_text(
 320             # ignored, osm area takes precedence over geometry import
 321             json.dumps({'properties': {'postcode': '110000', 'lat': 0.1, 'lon': 0.2},
 322                         'geometry': {'type': 'Polygon', 'coordinates': [[
 323                             [0, 0], [0, 1], [1, 1], [1, 0], [0, 0]]]}}) + '\n' +
 324             # geometry import takes precedence over guessed postcode from osm points
 325             json.dumps({'properties': {'postcode': '822114'},
 326                         'geometry': {'type': 'Polygon', 'coordinates': [[
 327                             [0, 0], [0, 10], [10, 10], [10, 0], [0, 0]]]}}) + '\n' +
 328             # geometry import takes precedence over csv import
 329             json.dumps({'properties': {'postcode': '822115'},
 330                         'geometry': {'type': 'Polygon', 'coordinates': [[
 331                             [0, 0], [0, 2], [2, 2], [2, 0], [0, 0]]]}}) + '\n', encoding='utf-8')
 332
 333         csvfile = tmp_path / 'xx_postcodes.csv'
 334         csvfile.write_text(
 335             "postcode,lat,lon\n"
 336             "822115,0.2,0.2\n"  # ignored, geometry import takes precedence over csv import
 337             "822116,-5,-10",  # added
 338             encoding='utf-8')
 339
 340         postcode_update(tmp_path)
 341
 342         assert self.row_set == {(None, 'xx', '822114', 5, 5),
 343                                 (None, 'xx', '822115', 1, 1),
 344                                 (3, 'xx', '110000', 77.2, 28.6),
 345                                 (None, 'xx', '822116', -10, -5)}
 346
 347     def test_no_placex_entry(self, postcode_update, temp_db_cursor, place_postcode_row):
 348         # Rewrite the get_country_code function to verify its execution.
 349         temp_db_cursor.execute("""
 350             CREATE OR REPLACE FUNCTION get_country_code(place geometry) RETURNS TEXT AS $$
 351               SELECT 'yy' $$ LANGUAGE sql""")
 352         place_postcode_row(centroid='POINT(10 12)', postcode='AB 4511')
 353         postcode_update()
 354
 355         assert self.row_set == {(None, 'yy', 'AB 4511', 10, 12)}
 356
 357     def test_discard_badly_formatted_postcodes(self, postcode_update, place_postcode_row):
 358         place_postcode_row(centroid='POINT(10 12)', country='fr', postcode='AB 4511')
 359         postcode_update()
 360
 361         assert not self.row_set
 362
 363
 364 def test_can_compute(dsn, table_factory):
 365     assert not postcodes.can_compute(dsn)
 366     table_factory('place_postcode')
 367     assert postcodes.can_compute(dsn)