test/python/tools/test_tiger_data.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 Test for tiger data function
   9 """
  10 import tarfile
  11 from textwrap import dedent
  12
  13 import pytest
  14 import pytest_asyncio  # noqa: F401
  15
  16 from nominatim_db.tools import tiger_data
  17 from nominatim_db.errors import UsageError
  18
  19
  20 @pytest.fixture
  21 def csv_factory(tmp_path):
  22     def _mk_file(fname, hnr_from=1, hnr_to=9, interpol='odd', street='Main St',
  23                  city='Newtown', state='AL', postcode='12345',
  24                  geometry='LINESTRING(-86.466995 32.428956,-86.466923 32.428933)'):
  25         (tmp_path / (fname + '.csv')).write_text(dedent("""\
  26         from;to;interpolation;street;city;state;postcode;geometry
  27         {};{};{};{};{};{};{};{}
  28         """.format(hnr_from, hnr_to, interpol, street, city, state,
  29                    postcode, geometry)), encoding='utf-8')
  30
  31     return _mk_file
  32
  33
  34 class TestTiger:
  35
  36     @pytest.fixture(autouse=True)
  37     def setup(self, temp_db_conn, placex_row, load_sql):
  38         load_sql('tables/search_name.sql', create_reverse_only=False)
  39         load_sql('tables/tiger.sql')
  40
  41         # fake parent roads
  42         for x in range(-870, -863):
  43             for y in range(323, 328):
  44                 placex_row(rank_search=26, rank_address=26,
  45                            geom=f"LINESTRING({x/10 - 0.1} {y/10}, {x/10 + 0.1} {y/10})")
  46
  47         temp_db_conn.execute("""
  48             CREATE OR REPLACE FUNCTION get_partition(cc VARCHAR(10)) RETURNS INTEGER AS $$
  49               SELECT 0;
  50             $$ LANGUAGE sql;
  51             CREATE OR REPLACE FUNCTION token_matches_street(i JSONB, s INT[]) RETURNS BOOLEAN AS $$
  52              SELECT false
  53             $$ LANGUAGE SQL IMMUTABLE STRICT PARALLEL SAFE;
  54         """)
  55
  56     @pytest.mark.parametrize("threads", (1, 5))
  57     @pytest.mark.asyncio
  58     async def test_add_tiger_data_database_frozen(self, def_config, src_dir, temp_db_cursor,
  59                                                   tokenizer_mock, threads):
  60         await tiger_data.add_tiger_data(str(src_dir / 'test' / 'testdb' / 'tiger'),
  61                                         def_config, threads, tokenizer_mock())
  62
  63         assert temp_db_cursor.table_rows('location_property_tiger') == 6209
  64
  65     @pytest.mark.asyncio
  66     async def test_add_tiger_data_reverse_only(self, def_config, src_dir, temp_db_cursor,
  67                                                tokenizer_mock):
  68         temp_db_cursor.execute("DROP TABLE search_name")
  69
  70         with pytest.raises(UsageError,
  71                            match="Cannot perform tiger import: required tables are missing. "
  72                            "See https://github.com/osm-search/Nominatim/issues/2463 for details."):
  73             await tiger_data.add_tiger_data(str(src_dir / 'test' / 'testdb' / 'tiger'),
  74                                             def_config, 1, tokenizer_mock())
  75
  76         assert temp_db_cursor.table_rows('location_property_tiger') == 0
  77
  78     @pytest.mark.asyncio
  79     async def test_add_tiger_data_no_files(self, def_config, temp_db_cursor, tokenizer_mock,
  80                                            tmp_path):
  81         await tiger_data.add_tiger_data(str(tmp_path), def_config, 1, tokenizer_mock())
  82
  83         assert temp_db_cursor.table_rows('location_property_tiger') == 0
  84
  85     @pytest.mark.asyncio
  86     async def test_add_tiger_data_bad_file(self, def_config, temp_db_cursor, tokenizer_mock,
  87                                            tmp_path):
  88         sqlfile = tmp_path / '1010.csv'
  89         sqlfile.write_text('Random text', encoding='utf-8')
  90
  91         await tiger_data.add_tiger_data(str(tmp_path), def_config, 1, tokenizer_mock())
  92
  93         assert temp_db_cursor.table_rows('location_property_tiger') == 0
  94
  95     @pytest.mark.asyncio
  96     async def test_add_tiger_data_hnr_nan(self, def_config, temp_db_cursor, tokenizer_mock,
  97                                           csv_factory, tmp_path):
  98         csv_factory('file1', hnr_to=99)
  99         csv_factory('file2', hnr_from='L12')
 100         csv_factory('file3', hnr_to='12.4')
 101
 102         await tiger_data.add_tiger_data(str(tmp_path), def_config, 1, tokenizer_mock())
 103
 104         rows = temp_db_cursor.row_set("""
 105             SELECT startnumber, endnumber FROM location_property_tiger""")
 106
 107         assert rows == {(1, 99)}
 108
 109     @pytest.mark.parametrize("threads", (1, 5))
 110     @pytest.mark.asyncio
 111     async def test_add_tiger_data_tarfile(self, def_config, temp_db_cursor, tokenizer_mock,
 112                                           tmp_path, src_dir, threads):
 113         tar = tarfile.open(str(tmp_path / 'sample.tar.gz'), "w:gz")
 114         tar.add(str(src_dir / 'test' / 'testdb' / 'tiger' / '01001.csv'))
 115         tar.close()
 116
 117         await tiger_data.add_tiger_data(str(tmp_path / 'sample.tar.gz'), def_config, threads,
 118                                         tokenizer_mock())
 119
 120         assert temp_db_cursor.table_rows('location_property_tiger') == 6209
 121
 122     @pytest.mark.asyncio
 123     async def test_add_tiger_data_bad_tarfile(self, def_config, tokenizer_mock, tmp_path):
 124         tarfile = tmp_path / 'sample.tar.gz'
 125         tarfile.write_text("""Random text""", encoding='utf-8')
 126
 127         with pytest.raises(UsageError):
 128             await tiger_data.add_tiger_data(str(tarfile), def_config, 1, tokenizer_mock())
 129
 130     @pytest.mark.asyncio
 131     async def test_add_tiger_data_empty_tarfile(self, def_config, temp_db_cursor, tokenizer_mock,
 132                                                 tmp_path):
 133         tar = tarfile.open(str(tmp_path / 'sample.tar.gz'), "w:gz")
 134         tar.add(__file__)
 135         tar.close()
 136
 137         await tiger_data.add_tiger_data(str(tmp_path / 'sample.tar.gz'), def_config, 1,
 138                                         tokenizer_mock())
 139
 140         assert temp_db_cursor.table_rows('location_property_tiger') == 0