1 # SPDX-License-Identifier: GPL-2.0-only
3 # This file is part of Nominatim. (https://nominatim.org)
5 # Copyright (C) 2022 by the Nominatim developer community.
6 # For a full list of authors see the git log.
8 Tests for migration functions
11 import psycopg2.extras
13 from nominatim.tools import migration
14 from nominatim.errors import UsageError
15 import nominatim.version
17 from mock_legacy_word_table import MockLegacyWordTable
21 def update_sql_functions(self, config):
26 def postprocess_mock(monkeypatch):
27 monkeypatch.setattr(migration.refresh, 'create_functions', lambda *args: args)
28 monkeypatch.setattr(migration.tokenizer_factory, 'get_tokenizer_for_db',
29 lambda *args: DummyTokenizer())
32 def legacy_word_table(temp_db_conn):
33 return MockLegacyWordTable(temp_db_conn)
36 def test_no_migration_old_versions(temp_db_with_extensions, table_factory, def_config):
37 table_factory('country_name', 'name HSTORE, country_code TEXT')
39 with pytest.raises(UsageError, match='Migration not possible'):
40 migration.migrate(def_config, {})
43 def test_set_up_migration_for_36(temp_db_with_extensions, temp_db_cursor,
44 table_factory, def_config, monkeypatch,
46 psycopg2.extras.register_hstore(temp_db_cursor)
47 # don't actually run any migration, except the property table creation
48 monkeypatch.setattr(migration, '_MIGRATION_FUNCTIONS',
49 [((3, 5, 0, 99), migration.add_nominatim_property_table)])
50 # Use a r/o user name that always exists
51 monkeypatch.setenv('NOMINATIM_DATABASE_WEBUSER', 'postgres')
53 table_factory('country_name', 'name HSTORE, country_code TEXT',
54 (({str(x): 'a' for x in range(200)}, 'gb'),))
56 assert not temp_db_cursor.table_exists('nominatim_properties')
58 assert migration.migrate(def_config, {}) == 0
60 assert temp_db_cursor.table_exists('nominatim_properties')
62 assert 1 == temp_db_cursor.scalar(""" SELECT count(*) FROM nominatim_properties
63 WHERE property = 'database_version'""")
66 def test_already_at_version(def_config, property_table):
68 property_table.set('database_version',
69 '{0[0]}.{0[1]}.{0[2]}-{0[3]}'.format(nominatim.version.NOMINATIM_VERSION))
71 assert migration.migrate(def_config, {}) == 0
74 def test_no_migrations_necessary(def_config, temp_db_cursor, property_table,
76 oldversion = [x for x in nominatim.version.NOMINATIM_VERSION]
78 property_table.set('database_version',
79 '{0[0]}.{0[1]}.{0[2]}-{0[3]}'.format(oldversion))
82 monkeypatch.setattr(migration, '_MIGRATION_FUNCTIONS',
83 [(tuple(oldversion), lambda **attr: True)])
85 assert migration.migrate(def_config, {}) == 0
88 def test_run_single_migration(def_config, temp_db_cursor, property_table,
89 monkeypatch, postprocess_mock):
90 oldversion = [x for x in nominatim.version.NOMINATIM_VERSION]
92 property_table.set('database_version',
93 '{0[0]}.{0[1]}.{0[2]}-{0[3]}'.format(oldversion))
95 done = {'old': False, 'new': False}
97 """ Dummy migration"""
100 def _old_migration(**_):
101 """ Dummy migration"""
105 monkeypatch.setattr(migration, '_MIGRATION_FUNCTIONS',
106 [(tuple(oldversion), _old_migration),
107 (nominatim.version.NOMINATIM_VERSION, _migration)])
109 assert migration.migrate(def_config, {}) == 0
112 assert not done['old']
113 assert property_table.get('database_version') == \
114 '{0[0]}.{0[1]}.{0[2]}-{0[3]}'.format(nominatim.version.NOMINATIM_VERSION)
117 ###### Tests for specific migrations
119 # Each migration should come with two tests:
120 # 1. Test that migration from old to new state works as expected.
121 # 2. Test that the migration can be rerun on the new state without side effects.
124 @pytest.mark.parametrize('in_attr', ('', 'with time zone'))
125 def test_import_status_timestamp_change(temp_db_conn, temp_db_cursor,
126 table_factory, in_attr):
127 table_factory('import_status',
128 f"""lastimportdate timestamp {in_attr},
132 migration.import_status_timestamp_change(temp_db_conn)
133 temp_db_conn.commit()
135 assert temp_db_cursor.scalar("""SELECT data_type FROM information_schema.columns
136 WHERE table_name = 'import_status'
137 and column_name = 'lastimportdate'""")\
138 == 'timestamp with time zone'
141 def test_add_nominatim_property_table(temp_db_conn, temp_db_cursor,
142 def_config, monkeypatch):
143 # Use a r/o user name that always exists
144 monkeypatch.setenv('NOMINATIM_DATABASE_WEBUSER', 'postgres')
146 assert not temp_db_cursor.table_exists('nominatim_properties')
148 migration.add_nominatim_property_table(temp_db_conn, def_config)
149 temp_db_conn.commit()
151 assert temp_db_cursor.table_exists('nominatim_properties')
154 def test_add_nominatim_property_table_repeat(temp_db_conn, temp_db_cursor,
155 def_config, property_table):
156 assert temp_db_cursor.table_exists('nominatim_properties')
158 migration.add_nominatim_property_table(temp_db_conn, def_config)
159 temp_db_conn.commit()
161 assert temp_db_cursor.table_exists('nominatim_properties')
164 def test_change_housenumber_transliteration(temp_db_conn, temp_db_cursor,
165 legacy_word_table, placex_table):
166 placex_table.add(housenumber='3A')
168 temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION make_standard_name(name TEXT)
169 RETURNS TEXT AS $$ SELECT lower(name) $$ LANGUAGE SQL """)
170 temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION getorcreate_housenumber_id(lookup_word TEXT)
171 RETURNS INTEGER AS $$ SELECT 4325 $$ LANGUAGE SQL """)
173 migration.change_housenumber_transliteration(temp_db_conn)
174 temp_db_conn.commit()
176 assert temp_db_cursor.scalar('SELECT housenumber from placex') == '3a'
178 migration.change_housenumber_transliteration(temp_db_conn)
179 temp_db_conn.commit()
181 assert temp_db_cursor.scalar('SELECT housenumber from placex') == '3a'
184 def test_switch_placenode_geometry_index(temp_db_conn, temp_db_cursor, placex_table):
185 temp_db_cursor.execute("""CREATE INDEX idx_placex_adminname
186 ON placex (place_id)""")
188 migration.switch_placenode_geometry_index(temp_db_conn)
189 temp_db_conn.commit()
191 assert temp_db_cursor.index_exists('placex', 'idx_placex_geometry_placenode')
192 assert not temp_db_cursor.index_exists('placex', 'idx_placex_adminname')
195 def test_switch_placenode_geometry_index_repeat(temp_db_conn, temp_db_cursor, placex_table):
196 temp_db_cursor.execute("""CREATE INDEX idx_placex_geometry_placenode
197 ON placex (place_id)""")
199 migration.switch_placenode_geometry_index(temp_db_conn)
200 temp_db_conn.commit()
202 assert temp_db_cursor.index_exists('placex', 'idx_placex_geometry_placenode')
203 assert not temp_db_cursor.index_exists('placex', 'idx_placex_adminname')
204 assert temp_db_cursor.scalar("""SELECT indexdef from pg_indexes
205 WHERE tablename = 'placex'
206 and indexname = 'idx_placex_geometry_placenode'
207 """).endswith('(place_id)')
210 def test_install_legacy_tokenizer(temp_db_conn, temp_db_cursor, project_env,
211 property_table, table_factory, monkeypatch,
213 table_factory('placex', 'place_id BIGINT')
214 table_factory('location_property_osmline', 'place_id BIGINT')
216 # Setting up the tokenizer is problematic
218 def migrate_database(self, config):
221 monkeypatch.setattr(migration.tokenizer_factory, 'create_tokenizer',
222 lambda cfg, **kwargs: MiniTokenizer())
224 migration.install_legacy_tokenizer(temp_db_conn, project_env)
225 temp_db_conn.commit()
229 def test_install_legacy_tokenizer_repeat(temp_db_conn, temp_db_cursor,
230 def_config, property_table):
232 property_table.set('tokenizer', 'dummy')
233 migration.install_legacy_tokenizer(temp_db_conn, def_config)
234 temp_db_conn.commit()
237 def test_create_tiger_housenumber_index(temp_db_conn, temp_db_cursor, table_factory):
238 table_factory('location_property_tiger',
239 'parent_place_id BIGINT, startnumber INT, endnumber INT')
241 migration.create_tiger_housenumber_index(temp_db_conn)
242 temp_db_conn.commit()
244 if temp_db_conn.server_version_tuple() >= (11, 0, 0):
245 assert temp_db_cursor.index_exists('location_property_tiger',
246 'idx_location_property_tiger_housenumber_migrated')
248 migration.create_tiger_housenumber_index(temp_db_conn)
249 temp_db_conn.commit()