-# SPDX-License-Identifier: GPL-2.0-only
+# SPDX-License-Identifier: GPL-3.0-or-later
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2022 by the Nominatim developer community.
+# Copyright (C) 2024 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for migration functions
"""
import pytest
-import psycopg2.extras
-from nominatim.tools import migration
-from nominatim.errors import UsageError
-import nominatim.version
+from nominatim_db.tools import migration
+from nominatim_db.errors import UsageError
+from nominatim_db.db.connection import server_version_tuple
+import nominatim_db.version
class DummyTokenizer:
def test_set_up_migration_for_36(temp_db_with_extensions, temp_db_cursor,
table_factory, def_config, monkeypatch,
postprocess_mock):
- psycopg2.extras.register_hstore(temp_db_cursor)
# don't actually run any migration, except the property table creation
monkeypatch.setattr(migration, '_MIGRATION_FUNCTIONS',
[((3, 5, 0, 99), migration.add_nominatim_property_table)])
WHERE property = 'database_version'""")
-def test_already_at_version(def_config, property_table):
+def test_already_at_version(temp_db_with_extensions, def_config, property_table):
property_table.set('database_version',
- '{0[0]}.{0[1]}.{0[2]}-{0[3]}'.format(nominatim.version.NOMINATIM_VERSION))
+ str(nominatim_db.version.NOMINATIM_VERSION))
assert migration.migrate(def_config, {}) == 0
-def test_no_migrations_necessary(def_config, temp_db_cursor, property_table,
- monkeypatch):
- oldversion = [x for x in nominatim.version.NOMINATIM_VERSION]
+def test_run_single_migration(temp_db_with_extensions, def_config, temp_db_cursor,
+ property_table, monkeypatch, postprocess_mock):
+ oldversion = [x for x in nominatim_db.version.NOMINATIM_VERSION]
oldversion[0] -= 1
property_table.set('database_version',
- '{0[0]}.{0[1]}.{0[2]}-{0[3]}'.format(oldversion))
-
- oldversion[0] = 0
- monkeypatch.setattr(migration, '_MIGRATION_FUNCTIONS',
- [(tuple(oldversion), lambda **attr: True)])
-
- assert migration.migrate(def_config, {}) == 0
-
-
-def test_run_single_migration(def_config, temp_db_cursor, property_table,
- monkeypatch, postprocess_mock):
- oldversion = [x for x in nominatim.version.NOMINATIM_VERSION]
- oldversion[0] -= 1
- property_table.set('database_version',
- '{0[0]}.{0[1]}.{0[2]}-{0[3]}'.format(oldversion))
+ str(nominatim_db.version.NominatimVersion(*oldversion)))
done = {'old': False, 'new': False}
def _migration(**_):
oldversion[0] = 0
monkeypatch.setattr(migration, '_MIGRATION_FUNCTIONS',
[(tuple(oldversion), _old_migration),
- (nominatim.version.NOMINATIM_VERSION, _migration)])
+ (nominatim_db.version.NOMINATIM_VERSION, _migration)])
assert migration.migrate(def_config, {}) == 0
assert done['new']
assert not done['old']
- assert property_table.get('database_version') == \
- '{0[0]}.{0[1]}.{0[2]}-{0[3]}'.format(nominatim.version.NOMINATIM_VERSION)
+ assert property_table.get('database_version') == str(nominatim_db.version.NOMINATIM_VERSION)
###### Tests for specific migrations
assert temp_db_cursor.table_exists('nominatim_properties')
-def test_change_housenumber_transliteration(temp_db_conn, temp_db_cursor,
- word_table, placex_table):
- placex_table.add(housenumber='3A')
-
- temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION make_standard_name(name TEXT)
- RETURNS TEXT AS $$ SELECT lower(name) $$ LANGUAGE SQL """)
- temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION getorcreate_housenumber_id(lookup_word TEXT)
- RETURNS INTEGER AS $$ SELECT 4325 $$ LANGUAGE SQL """)
-
- migration.change_housenumber_transliteration(temp_db_conn)
- temp_db_conn.commit()
-
- assert temp_db_cursor.scalar('SELECT housenumber from placex') == '3a'
-
- migration.change_housenumber_transliteration(temp_db_conn)
- temp_db_conn.commit()
-
- assert temp_db_cursor.scalar('SELECT housenumber from placex') == '3a'
-
-
def test_switch_placenode_geometry_index(temp_db_conn, temp_db_cursor, placex_table):
temp_db_cursor.execute("""CREATE INDEX idx_placex_adminname
ON placex (place_id)""")
migration.create_tiger_housenumber_index(temp_db_conn)
temp_db_conn.commit()
- if temp_db_conn.server_version_tuple() >= (11, 0, 0):
+ if server_version_tuple(temp_db_conn) >= (11, 0, 0):
assert temp_db_cursor.index_exists('location_property_tiger',
'idx_location_property_tiger_housenumber_migrated')