osm-search/Nominatim
1
1
Fork 0
mirror of https://github.com/osm-search/Nominatim.git synced 2024-09-20 23:48:18 +03:00
Code Issues Packages Projects Releases Wiki Activity

test: use table_rows() and execute_values() where possible

Browse Source 
Some uses of scalar() could also be replaced with convenience
functions from the word table mock.
This commit is contained in:
Sarah Hoffmann 2021-05-19 10:51:10 +02:00
parent 510eb53f53
commit 65bd749918
7 changed files with 23 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
test/python/conftest.py
View File

@ -4,7 +4,6 @@ import tempfile
from pathlib import Path
import psycopg2
import psycopg2.extras
import pytest
SRC_DIR = Path(__file__) / '..' / '..' / '..'
@ -91,8 +90,7 @@ def table_factory(temp_db_cursor):
def mk_table(name, definition='id INT', content=None):
temp_db_cursor.execute('CREATE TABLE {} ({})'.format(name, definition))
if content is not None:
psycopg2.extras.execute_values(
temp_db_cursor, "INSERT INTO {} VALUES %s".format(name), content)
temp_db_cursor.execute_values("INSERT INTO {} VALUES %s".format(name), content)
return mk_table

8
test/python/test_db_status.py
View File

@ -68,7 +68,7 @@ def test_set_status_filled_table(status_table, temp_db_conn, temp_db_cursor):
date = dt.datetime.fromordinal(1000000).replace(tzinfo=dt.timezone.utc)
nominatim.db.status.set_status(temp_db_conn, date=date)
assert 1 == temp_db_cursor.scalar("SELECT count(*) FROM import_status")
assert temp_db_cursor.table_rows('import_status') == 1
date = dt.datetime.fromordinal(1000100).replace(tzinfo=dt.timezone.utc)
nominatim.db.status.set_status(temp_db_conn, date=date, seq=456, indexed=False)
@ -83,7 +83,7 @@ def test_set_status_missing_date(status_table, temp_db_conn, temp_db_cursor):
date = dt.datetime.fromordinal(1000000).replace(tzinfo=dt.timezone.utc)
nominatim.db.status.set_status(temp_db_conn, date=date)
assert 1 == temp_db_cursor.scalar("SELECT count(*) FROM import_status")
assert temp_db_cursor.table_rows('import_status') == 1
nominatim.db.status.set_status(temp_db_conn, date=None, seq=456, indexed=False)
@ -118,7 +118,7 @@ def test_set_indexed(status_table, temp_db_conn, temp_db_cursor, old_state, new_
def test_set_indexed_empty_status(status_table, temp_db_conn, temp_db_cursor):
nominatim.db.status.set_indexed(temp_db_conn, True)
assert temp_db_cursor.scalar("SELECT count(*) FROM import_status") == 0
assert temp_db_cursor.table_rows("import_status") == 0
def text_log_status(status_table, temp_db_conn):
@ -127,6 +127,6 @@ def text_log_status(status_table, temp_db_conn):
nominatim.db.status.set_status(temp_db_conn, date=date, seq=56)
nominatim.db.status.log_status(temp_db_conn, start, 'index')
assert temp_db_cursor.scalar("SELECT count(*) FROM import_osmosis_log") == 1
assert temp_db_cursor.table_rows("import_osmosis_log") == 1
assert temp_db_cursor.scalar("SELECT seq FROM import_osmosis_log") == 56
assert temp_db_cursor.scalar("SELECT date FROM import_osmosis_log") == date

4
test/python/test_tokenizer_legacy.py
View File

@ -247,11 +247,11 @@ def test_update_special_phrases_no_replace(analyzer, word_table, temp_db_cursor,
VALUES (' foo', 'foo', 'amenity', 'prison', 'in'),
(' bar', 'bar', 'highway', 'road', null)""")
assert 2 == temp_db_cursor.scalar("SELECT count(*) FROM word WHERE class != 'place'""")
assert word_table.count_special() == 2
analyzer.update_special_phrases([], False)
assert 2 == temp_db_cursor.scalar("SELECT count(*) FROM word WHERE class != 'place'""")
assert word_table.count_special() == 2
def test_update_special_phrase_modify(analyzer, word_table, make_standard_name):

19
test/python/test_tokenizer_legacy_icu.py
View File

@ -185,12 +185,12 @@ def test_update_special_phrase_delete_all(analyzer, word_table, temp_db_cursor):
VALUES (' FOO', 'foo', 'amenity', 'prison', 'in'),
(' BAR', 'bar', 'highway', 'road', null)""")
assert 2 == temp_db_cursor.scalar("SELECT count(*) FROM word WHERE class != 'place'""")
assert word_table.count_special() == 2
with analyzer() as a:
a.update_special_phrases([], True)
assert 0 == temp_db_cursor.scalar("SELECT count(*) FROM word WHERE class != 'place'""")
assert word_table.count_special() == 0
def test_update_special_phrases_no_replace(analyzer, word_table, temp_db_cursor,):
@ -198,12 +198,12 @@ def test_update_special_phrases_no_replace(analyzer, word_table, temp_db_cursor,
VALUES (' FOO', 'foo', 'amenity', 'prison', 'in'),
(' BAR', 'bar', 'highway', 'road', null)""")
assert 2 == temp_db_cursor.scalar("SELECT count(*) FROM word WHERE class != 'place'""")
assert word_table.count_special() == 2
with analyzer() as a:
a.update_special_phrases([], False)
assert 2 == temp_db_cursor.scalar("SELECT count(*) FROM word WHERE class != 'place'""")
assert word_table.count_special() == 2
def test_update_special_phrase_modify(analyzer, word_table, temp_db_cursor):
@ -211,7 +211,7 @@ def test_update_special_phrase_modify(analyzer, word_table, temp_db_cursor):
VALUES (' FOO', 'foo', 'amenity', 'prison', 'in'),
(' BAR', 'bar', 'highway', 'road', null)""")
assert 2 == temp_db_cursor.scalar("SELECT count(*) FROM word WHERE class != 'place'""")
assert word_table.count_special() == 2
with analyzer() as a:
a.update_special_phrases([
@ -220,11 +220,10 @@ def test_update_special_phrase_modify(analyzer, word_table, temp_db_cursor):
('garden', 'leisure', 'garden', 'near')
], True)
assert temp_db_cursor.row_set("""SELECT word_token, word, class, type, operator
FROM word WHERE class != 'place'""") \
== set(((' PRISON', 'prison', 'amenity', 'prison', 'in'),
(' BAR', 'bar', 'highway', 'road', None),
(' GARDEN', 'garden', 'leisure', 'garden', 'near')))
assert word_table.get_special() \
== {(' PRISON', 'prison', 'amenity', 'prison', 'in'),
(' BAR', 'bar', 'highway', 'road', None),
(' GARDEN', 'garden', 'leisure', 'garden', 'near')}
def test_process_place_names(analyzer, getorcreate_term_id):

6
test/python/test_tools_database_import.py
View File

@ -85,7 +85,7 @@ def test_import_base_data(src_dir, temp_db, temp_db_cursor):
temp_db_cursor.execute('CREATE EXTENSION postgis')
database_import.import_base_data('dbname=' + temp_db, src_dir / 'data')
assert temp_db_cursor.scalar('SELECT count(*) FROM country_name') > 0
assert temp_db_cursor.table_rows('country_name') > 0
def test_import_base_data_ignore_partitions(src_dir, temp_db, temp_db_cursor):
@ -94,8 +94,8 @@ def test_import_base_data_ignore_partitions(src_dir, temp_db, temp_db_cursor):
database_import.import_base_data('dbname=' + temp_db, src_dir / 'data',
ignore_partitions=True)
assert temp_db_cursor.scalar('SELECT count(*) FROM country_name') > 0
assert temp_db_cursor.scalar('SELECT count(*) FROM country_name WHERE partition != 0') == 0
assert temp_db_cursor.table_rows('country_name') > 0
assert temp_db_cursor.table_rows('country_name', where='partition != 0') == 0
def test_import_osm_data_simple(temp_db_cursor,osm2pgsql_options):

4
test/python/test_tools_refresh.py
View File

@ -22,5 +22,5 @@ def test_refresh_import_wikipedia(dsn, table_factory, temp_db_cursor, replace):
# use the small wikipedia file for the API testdb
assert 0 == refresh.import_wikipedia_articles(dsn, TEST_DIR / 'testdb')
assert temp_db_cursor.scalar('SELECT count(*) FROM wikipedia_article') > 0
assert temp_db_cursor.scalar('SELECT count(*) FROM wikipedia_redirect') > 0
assert temp_db_cursor.table_rows('wikipedia_article') > 0
assert temp_db_cursor.table_rows('wikipedia_redirect') > 0

4
test/python/test_tools_refresh_address_levels.py
View File

@ -11,7 +11,7 @@ from nominatim.tools.refresh import load_address_levels, load_address_levels_fro
def test_load_ranks_def_config(temp_db_conn, temp_db_cursor, def_config):
load_address_levels_from_file(temp_db_conn, Path(def_config.ADDRESS_LEVEL_CONFIG))
assert temp_db_cursor.scalar('SELECT count(*) FROM address_levels') > 0
assert temp_db_cursor.table_rows('address_levels') > 0
def test_load_ranks_from_file(temp_db_conn, temp_db_cursor, tmp_path):
test_file = tmp_path / 'test_levels.json'
@ -19,7 +19,7 @@ def test_load_ranks_from_file(temp_db_conn, temp_db_cursor, tmp_path):
load_address_levels_from_file(temp_db_conn, test_file)
assert temp_db_cursor.scalar('SELECT count(*) FROM address_levels') > 0
assert temp_db_cursor.table_rows('address_levels') > 0
def test_load_ranks_from_broken_file(temp_db_conn, tmp_path):