test/python/test_tools_database_import.py

   1 """
   2 Tests for functions to import a new database.
   3 """
   4 import pytest
   5 import psycopg2
   6 import sys
   7 from pathlib import Path
   8
   9 from nominatim.tools import database_import
  10 from nominatim.errors import UsageError
  11
  12 @pytest.fixture
  13 def nonexistant_db():
  14     dbname = 'test_nominatim_python_unittest'
  15
  16     conn = psycopg2.connect(database='postgres')
  17
  18     conn.set_isolation_level(0)
  19     with conn.cursor() as cur:
  20         cur.execute('DROP DATABASE IF EXISTS {}'.format(dbname))
  21
  22     yield dbname
  23
  24     with conn.cursor() as cur:
  25         cur.execute('DROP DATABASE IF EXISTS {}'.format(dbname))
  26
  27
  28 def test_create_db_success(nonexistant_db):
  29     database_import.create_db('dbname=' + nonexistant_db, rouser='www-data')
  30
  31     conn = psycopg2.connect(database=nonexistant_db)
  32     conn.close()
  33
  34
  35 def test_create_db_already_exists(temp_db):
  36     with pytest.raises(UsageError):
  37         database_import.create_db('dbname=' + temp_db)
  38
  39
  40 def test_create_db_unsupported_version(nonexistant_db, monkeypatch):
  41     monkeypatch.setattr(database_import, 'POSTGRESQL_REQUIRED_VERSION', (100, 4))
  42
  43     with pytest.raises(UsageError, match='PostgreSQL server is too old.'):
  44         database_import.create_db('dbname=' + nonexistant_db)
  45
  46
  47 def test_create_db_missing_ro_user(nonexistant_db):
  48     with pytest.raises(UsageError, match='Missing read-only user.'):
  49         database_import.create_db('dbname=' + nonexistant_db, rouser='sdfwkjkjgdugu2;jgsafkljas;')
  50
  51
  52 def test_setup_extensions(temp_db_conn, temp_db_cursor):
  53     database_import.setup_extensions(temp_db_conn)
  54
  55     temp_db_cursor.execute('CREATE TABLE t (h HSTORE, geom GEOMETRY(Geometry, 4326))')
  56
  57
  58 def test_setup_extensions_old_postgis(temp_db_conn, monkeypatch):
  59     monkeypatch.setattr(database_import, 'POSTGIS_REQUIRED_VERSION', (50, 50))
  60
  61     with pytest.raises(UsageError, match='PostGIS version is too old.'):
  62         database_import.setup_extensions(temp_db_conn)
  63
  64
  65 def test_install_module(tmp_path):
  66     src_dir = tmp_path / 'source'
  67     src_dir.mkdir()
  68     (src_dir / 'nominatim.so').write_text('TEST nomiantim.so')
  69
  70     project_dir = tmp_path / 'project'
  71     project_dir.mkdir()
  72
  73     database_import.install_module(src_dir, project_dir, '')
  74
  75     outfile = project_dir / 'module' / 'nominatim.so'
  76
  77     assert outfile.exists()
  78     assert outfile.read_text() == 'TEST nomiantim.so'
  79     assert outfile.stat().st_mode == 33261
  80
  81
  82 def test_import_base_data(src_dir, temp_db, temp_db_cursor):
  83     temp_db_cursor.execute('CREATE EXTENSION hstore')
  84     temp_db_cursor.execute('CREATE EXTENSION postgis')
  85     database_import.import_base_data('dbname=' + temp_db, src_dir / 'data')
  86
  87     assert temp_db_cursor.scalar('SELECT count(*) FROM country_name') > 0
  88
  89
  90 def test_import_base_data_ignore_partitions(src_dir, temp_db, temp_db_cursor):
  91     temp_db_cursor.execute('CREATE EXTENSION hstore')
  92     temp_db_cursor.execute('CREATE EXTENSION postgis')
  93     database_import.import_base_data('dbname=' + temp_db, src_dir / 'data',
  94                                      ignore_partitions=True)
  95
  96     assert temp_db_cursor.scalar('SELECT count(*) FROM country_name') > 0
  97     assert temp_db_cursor.scalar('SELECT count(*) FROM country_name WHERE partition != 0') == 0
  98
  99
 100 def test_import_osm_data_simple(temp_db_cursor,osm2pgsql_options):
 101     temp_db_cursor.execute('CREATE TABLE place (id INT)')
 102     temp_db_cursor.execute('INSERT INTO place values (1)')
 103
 104     database_import.import_osm_data('file.pdf', osm2pgsql_options)
 105
 106
 107 def test_import_osm_data_simple_no_data(temp_db_cursor,osm2pgsql_options):
 108     temp_db_cursor.execute('CREATE TABLE place (id INT)')
 109
 110     with pytest.raises(UsageError, match='No data.*'):
 111         database_import.import_osm_data('file.pdf', osm2pgsql_options)
 112
 113
 114 def test_import_osm_data_drop(temp_db_conn, temp_db_cursor, tmp_path, osm2pgsql_options):
 115     temp_db_cursor.execute('CREATE TABLE place (id INT)')
 116     temp_db_cursor.execute('CREATE TABLE planet_osm_nodes (id INT)')
 117     temp_db_cursor.execute('INSERT INTO place values (1)')
 118
 119     flatfile = tmp_path / 'flatfile'
 120     flatfile.write_text('touch')
 121
 122     osm2pgsql_options['flatnode_file'] = str(flatfile.resolve())
 123
 124     database_import.import_osm_data('file.pdf', osm2pgsql_options, drop=True)
 125
 126     assert not flatfile.exists()
 127     assert not temp_db_conn.table_exists('planet_osm_nodes')
 128
 129
 130 def test_import_osm_data_default_cache(temp_db_cursor,osm2pgsql_options):
 131     temp_db_cursor.execute('CREATE TABLE place (id INT)')
 132     temp_db_cursor.execute('INSERT INTO place values (1)')
 133
 134     osm2pgsql_options['osm2pgsql_cache'] = 0
 135
 136     database_import.import_osm_data(Path(__file__), osm2pgsql_options)