test/python/test_tools_tiger_data.py

   1 """
   2 Test for tiger data function
   3 """
   4 from pathlib import Path
   5
   6 import pytest
   7 import tarfile
   8
   9 from nominatim.tools import tiger_data, database_import
  10
  11
  12 @pytest.mark.parametrize("threads", (1, 5))
  13 def test_add_tiger_data(def_config, tmp_path, sql_preprocessor,
  14                         temp_db_cursor, threads, temp_db_with_extensions):
  15     temp_db_cursor.execute('CREATE TABLE place (id INT)')
  16     sqlfile = tmp_path / '1010.sql'
  17     sqlfile.write_text("""INSERT INTO place values (1);
  18                           INSERT INTO non_existant_table values (1);""")
  19     tiger_data.add_tiger_data(str(tmp_path), def_config, threads)
  20
  21     assert temp_db_cursor.table_rows('place') == 1
  22
  23
  24 @pytest.mark.parametrize("threads", (1, 5))
  25 def test_add_tiger_data_bad_file(def_config, tmp_path, sql_preprocessor,
  26                                  temp_db_cursor, threads, temp_db_with_extensions):
  27     temp_db_cursor.execute('CREATE TABLE place (id INT)')
  28     sqlfile = tmp_path / '1010.txt'
  29     sqlfile.write_text("""Random text""")
  30     tiger_data.add_tiger_data(str(tmp_path), def_config, threads)
  31
  32     assert temp_db_cursor.table_rows('place') == 0
  33
  34
  35 @pytest.mark.parametrize("threads", (1, 5))
  36 def test_add_tiger_data_tarfile(def_config, tmp_path, temp_db_cursor,
  37                                 threads, temp_db_with_extensions, sql_preprocessor):
  38     temp_db_cursor.execute('CREATE TABLE place (id INT)')
  39     sqlfile = tmp_path / '1010.sql'
  40     sqlfile.write_text("""INSERT INTO place values (1);
  41                           INSERT INTO non_existant_table values (1);""")
  42     tar = tarfile.open(str(tmp_path / 'sample.tar.gz'), "w:gz")
  43     tar.add(sqlfile)
  44     tar.close()
  45     tiger_data.add_tiger_data(str(tmp_path / 'sample.tar.gz'), def_config, threads)
  46
  47     assert temp_db_cursor.table_rows('place') == 1
  48
  49
  50 @pytest.mark.parametrize("threads", (1, 5))
  51 def test_add_tiger_data_bad_tarfile(def_config, tmp_path, temp_db_cursor, threads,
  52                                     temp_db_with_extensions, sql_preprocessor):
  53     temp_db_cursor.execute('CREATE TABLE place (id INT)')
  54     sqlfile = tmp_path / '1010.txt'
  55     sqlfile.write_text("""Random text""")
  56     tar = tarfile.open(str(tmp_path / 'sample.tar.gz'), "w:gz")
  57     tar.add(sqlfile)
  58     tar.close()
  59     tiger_data.add_tiger_data(str(tmp_path / 'sample.tar.gz'), def_config, threads)
  60
  61     assert temp_db_cursor.table_rows('place') == 0