mirror of
https://github.com/osm-search/Nominatim.git
synced 2026-02-16 05:18:00 +00:00
adapt unit tests to new postcode algorithms
This commit is contained in:
Sarah Hoffmann
@@ -109,14 +109,16 @@ class APITester:
|
||||
def add_postcode(self, **kw):
|
||||
self.add_data('postcode',
|
||||
{'place_id': kw.get('place_id', 1000),
|
||||
'osm_id': kw.get('osm_id'),
|
||||
'parent_place_id': kw.get('parent_place_id'),
|
||||
'country_code': kw.get('country_code'),
|
||||
'postcode': kw.get('postcode'),
|
||||
'rank_search': kw.get('rank_search', 20),
|
||||
'rank_address': kw.get('rank_address', 22),
|
||||
'rank_search': kw.get('rank_search', 21),
|
||||
'indexed_date': kw.get('indexed_date',
|
||||
dt.datetime(2022, 12, 7, 14, 14, 46, 0)),
|
||||
'geometry': kw.get('geometry', 'POINT(23 34)')})
|
||||
'centroid': kw.get('centroid', 'POINT(23 34)'),
|
||||
'geometry': kw.get('geometry', 'POLYGON((22.99 33.99, 22.99 34.01, '
|
||||
'23.01 34, 22.99 33.99))')})
|
||||
|
||||
def add_country(self, country_code, geometry):
|
||||
self.add_data('country_grid',
|
||||
|
||||
@@ -16,6 +16,11 @@ from nominatim_api.search.db_search_fields import WeightedStrings, FieldLookup,
|
||||
FieldRanking, RankedTokens
|
||||
|
||||
|
||||
def poly_around(x, y, diff=0.01):
|
||||
return f"POLYGON(({x - diff} {y - diff}, {x + diff} {y - diff},"\
|
||||
f" {x + diff} {y + diff}, {x - diff} {y + diff}, {x - diff} {y - diff}))"
|
||||
|
||||
|
||||
def run_search(apiobj, frontend, global_penalty, pcs, pc_penalties=None,
|
||||
ccodes=[], lookup=[], ranking=[], details=SearchDetails()):
|
||||
if pc_penalties is None:
|
||||
@@ -58,19 +63,19 @@ def test_postcode_with_country(apiobj, frontend):
|
||||
|
||||
assert len(results) == 1
|
||||
assert results[0].place_id == 101
|
||||
assert results[0].osm_object is None
|
||||
|
||||
|
||||
def test_postcode_area(apiobj, frontend):
|
||||
apiobj.add_postcode(place_id=100, country_code='ch', postcode='12345')
|
||||
apiobj.add_placex(place_id=200, country_code='ch', postcode='12345',
|
||||
osm_type='R', osm_id=34, class_='boundary', type='postal_code',
|
||||
geometry='POLYGON((0 0, 1 0, 1 1, 0 1, 0 0))')
|
||||
apiobj.add_postcode(place_id=200, country_code='ch', postcode='12345',
|
||||
osm_id=34,
|
||||
centroid='POINT(0.5 0.5)', geometry=poly_around(0.5, 0.5))
|
||||
|
||||
results = run_search(apiobj, frontend, 0.3, ['12345'], [0.0])
|
||||
|
||||
assert len(results) == 1
|
||||
assert results[0].place_id == 200
|
||||
assert results[0].bbox.area == 1
|
||||
assert results[0].osm_object == ('R', 34)
|
||||
|
||||
|
||||
class TestPostcodeSearchWithAddress:
|
||||
@@ -79,10 +84,10 @@ class TestPostcodeSearchWithAddress:
|
||||
def fill_database(self, apiobj):
|
||||
apiobj.add_postcode(place_id=100, country_code='ch',
|
||||
parent_place_id=1000, postcode='12345',
|
||||
geometry='POINT(17 5)')
|
||||
centroid='POINT(17 5)', geometry=poly_around(17, 5))
|
||||
apiobj.add_postcode(place_id=101, country_code='pl',
|
||||
parent_place_id=2000, postcode='12345',
|
||||
geometry='POINT(-45 7)')
|
||||
centroid='POINT(-45 7)', geometry=poly_around(-45, 7))
|
||||
apiobj.add_placex(place_id=1000, class_='place', type='village',
|
||||
rank_search=22, rank_address=22,
|
||||
country_code='ch')
|
||||
|
||||
@@ -489,9 +489,10 @@ def test_lookup_in_postcode(apiobj, frontend):
|
||||
parent_place_id=152,
|
||||
postcode='34 425',
|
||||
country_code='gb',
|
||||
rank_search=20, rank_address=22,
|
||||
rank_search=20,
|
||||
indexed_date=import_date,
|
||||
geometry='POINT(-9.45 5.6)')
|
||||
centroid='POINT(-9.45 5.6)',
|
||||
geometry='POLYGON((-9.5 5.5, -9.5 5.7, -9.4 5.6, -9.5 5.5))')
|
||||
|
||||
api = frontend(apiobj, options={'details'})
|
||||
result = api.details(napi.PlaceID(554))
|
||||
@@ -517,7 +518,7 @@ def test_lookup_in_postcode(apiobj, frontend):
|
||||
assert result.wikipedia is None
|
||||
|
||||
assert result.rank_search == 20
|
||||
assert result.rank_address == 22
|
||||
assert result.rank_address == 5
|
||||
assert result.importance is None
|
||||
|
||||
assert result.country_code == 'gb'
|
||||
@@ -529,15 +530,17 @@ def test_lookup_in_postcode(apiobj, frontend):
|
||||
assert result.name_keywords is None
|
||||
assert result.address_keywords is None
|
||||
|
||||
assert result.geometry == {'type': 'ST_Point'}
|
||||
assert result.geometry == {'type': 'ST_Polygon'}
|
||||
|
||||
|
||||
def test_lookup_postcode_with_address_details(apiobj, frontend):
|
||||
apiobj.add_postcode(place_id=9000,
|
||||
@pytest.mark.parametrize('lookup', [napi.PlaceID(9000),
|
||||
napi.OsmID('R', 12)])
|
||||
def test_lookup_postcode_with_address_details(apiobj, frontend, lookup):
|
||||
apiobj.add_postcode(place_id=9000, osm_id=12,
|
||||
parent_place_id=332,
|
||||
postcode='34 425',
|
||||
country_code='gb',
|
||||
rank_search=25, rank_address=25)
|
||||
rank_search=25)
|
||||
apiobj.add_placex(place_id=332, osm_type='N', osm_id=3333,
|
||||
class_='place', type='suburb', name='Smallplace',
|
||||
country_code='gb', admin_level=13,
|
||||
@@ -549,15 +552,15 @@ def test_lookup_postcode_with_address_details(apiobj, frontend):
|
||||
rank_search=17, rank_address=16)
|
||||
|
||||
api = frontend(apiobj, options={'details'})
|
||||
result = api.details(napi.PlaceID(9000), address_details=True)
|
||||
result = api.details(lookup, address_details=True)
|
||||
napi.Locales().localize_results([result])
|
||||
|
||||
assert result.address_rows == [
|
||||
napi.AddressLine(place_id=9000, osm_object=None,
|
||||
category=('place', 'postcode'),
|
||||
napi.AddressLine(place_id=9000, osm_object=('R', 12),
|
||||
category=('boundary', 'postal_code'),
|
||||
names={'ref': '34 425'}, extratags={},
|
||||
admin_level=15, fromarea=True, isaddress=True,
|
||||
rank_address=25, distance=0.0,
|
||||
rank_address=5, distance=0.0,
|
||||
local_name='34 425'),
|
||||
napi.AddressLine(place_id=332, osm_object=('N', 3333),
|
||||
category=('place', 'suburb'),
|
||||
|
||||
@@ -41,7 +41,8 @@ class TestRefresh:
|
||||
assert self.call_nominatim('refresh', '--word-tokens') == 0
|
||||
assert self.tokenizer_mock.update_word_tokens_called
|
||||
|
||||
def test_refresh_postcodes(self, async_mock_func_factory, mock_func_factory, place_table):
|
||||
def test_refresh_postcodes(self, async_mock_func_factory, mock_func_factory,
|
||||
place_postcode_table):
|
||||
func_mock = mock_func_factory(nominatim_db.tools.postcodes, 'update_postcodes')
|
||||
idx_mock = async_mock_func_factory(nominatim_db.indexer.indexer.Indexer, 'index_postcodes')
|
||||
|
||||
|
||||
@@ -26,6 +26,13 @@ import mocks
|
||||
from cursor import CursorForTesting
|
||||
|
||||
|
||||
def _with_srid(geom, default=None):
|
||||
if geom is None:
|
||||
return None if default is None else f"SRID=4326;{default}"
|
||||
|
||||
return f"SRID=4326;{geom}"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def src_dir():
|
||||
return SRC_DIR
|
||||
@@ -179,6 +186,36 @@ def place_row(place_table, temp_db_cursor):
|
||||
return _insert
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def place_postcode_table(temp_db_with_extensions, table_factory):
|
||||
""" Create an empty version of the place_postcode table.
|
||||
"""
|
||||
table_factory('place_postcode',
|
||||
"""osm_type char(1) NOT NULL,
|
||||
osm_id bigint NOT NULL,
|
||||
postcode text NOT NULL,
|
||||
country_code text,
|
||||
centroid Geometry(Point, 4326) NOT NULL,
|
||||
geometry Geometry(Geometry, 4326)""")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def place_postcode_row(place_postcode_table, temp_db_cursor):
|
||||
""" A factory for rows in the place table. The table is created as a
|
||||
prerequisite to the fixture.
|
||||
"""
|
||||
idseq = itertools.count(5001)
|
||||
def _insert(osm_type='N', osm_id=None, postcode=None, country=None,
|
||||
centroid=None, geom=None):
|
||||
temp_db_cursor.execute("INSERT INTO place_postcode VALUES (%s, %s, %s, %s, %s, %s)",
|
||||
(osm_type, osm_id or next(idseq),
|
||||
postcode, country,
|
||||
_with_srid(centroid, 'POINT(12.0 4.0)'),
|
||||
_with_srid(geom)))
|
||||
|
||||
return _insert
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def placex_table(temp_db_with_extensions, temp_db_conn):
|
||||
""" Create an empty version of the place table.
|
||||
|
||||
@@ -48,7 +48,7 @@ class IndexerTestDB:
|
||||
indexed_status SMALLINT,
|
||||
indexed_date TIMESTAMP,
|
||||
geometry_sector INTEGER)""")
|
||||
cur.execute("""CREATE TABLE location_postcode (
|
||||
cur.execute("""CREATE TABLE location_postcodes (
|
||||
place_id BIGINT,
|
||||
indexed_status SMALLINT,
|
||||
indexed_date TIMESTAMP,
|
||||
@@ -94,7 +94,7 @@ class IndexerTestDB:
|
||||
$$ LANGUAGE plpgsql STABLE;
|
||||
""")
|
||||
|
||||
for table in ('placex', 'location_property_osmline', 'location_postcode'):
|
||||
for table in ('placex', 'location_property_osmline', 'location_postcodes'):
|
||||
cur.execute("""CREATE TRIGGER {0}_update BEFORE UPDATE ON {0}
|
||||
FOR EACH ROW EXECUTE PROCEDURE date_update()
|
||||
""".format(table))
|
||||
@@ -132,7 +132,7 @@ class IndexerTestDB:
|
||||
def add_postcode(self, country, postcode):
|
||||
next_id = next(self.postcode_id)
|
||||
with self.conn.cursor() as cur:
|
||||
cur.execute("""INSERT INTO location_postcode
|
||||
cur.execute("""INSERT INTO location_postcodes
|
||||
(place_id, indexed_status, country_code, postcode)
|
||||
VALUES (%s, 1, %s, %s)""",
|
||||
(next_id, country, postcode))
|
||||
@@ -268,7 +268,7 @@ async def test_index_postcodes(test_db, threads, test_tokenizer):
|
||||
idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, threads)
|
||||
await idx.index_postcodes()
|
||||
|
||||
assert test_db.scalar("""SELECT count(*) FROM location_postcode
|
||||
assert test_db.scalar("""SELECT count(*) FROM location_postcodes
|
||||
WHERE indexed_status != 0""") == 0
|
||||
|
||||
|
||||
@@ -288,5 +288,5 @@ async def test_index_full(test_db, analyse, test_tokenizer):
|
||||
|
||||
assert test_db.placex_unindexed() == 0
|
||||
assert test_db.osmline_unindexed() == 0
|
||||
assert test_db.scalar("""SELECT count(*) FROM location_postcode
|
||||
assert test_db.scalar("""SELECT count(*) FROM location_postcodes
|
||||
WHERE indexed_status != 0""") == 0
|
||||
|
||||
@@ -50,18 +50,18 @@ class MockPlacexTable:
|
||||
|
||||
def add(self, osm_type='N', osm_id=None, cls='amenity', typ='cafe', names=None,
|
||||
admin_level=None, address=None, extratags=None, geom='POINT(10 4)',
|
||||
country=None, housenumber=None, rank_search=30):
|
||||
country=None, housenumber=None, rank_search=30, centroid=None):
|
||||
with self.conn.cursor() as cur:
|
||||
cur.execute("""INSERT INTO placex (place_id, osm_type, osm_id, class,
|
||||
type, name, admin_level, address,
|
||||
housenumber, rank_search,
|
||||
extratags, geometry, country_code)
|
||||
extratags, centroid, geometry, country_code)
|
||||
VALUES(nextval('seq_place'), %s, %s, %s, %s, %s, %s,
|
||||
%s, %s, %s, %s, %s, %s)
|
||||
%s, %s, %s, %s, %s, %s, %s)
|
||||
RETURNING place_id""",
|
||||
(osm_type, osm_id or next(self.idseq), cls, typ, names,
|
||||
admin_level, address, housenumber, rank_search,
|
||||
extratags, 'SRID=4326;' + geom,
|
||||
extratags, centroid, 'SRID=4326;' + geom,
|
||||
country))
|
||||
place_id = cur.fetchone()[0]
|
||||
self.conn.commit()
|
||||
|
||||
@@ -150,7 +150,7 @@ def test_truncate_database_tables(temp_db_conn, temp_db_cursor, table_factory, w
|
||||
tables = ['placex', 'place_addressline', 'location_area',
|
||||
'location_area_country',
|
||||
'location_property_tiger', 'location_property_osmline',
|
||||
'location_postcode', 'location_road_23']
|
||||
'location_postcodes', 'location_road_23']
|
||||
if with_search:
|
||||
tables.append('search_name')
|
||||
|
||||
|
||||
@@ -11,7 +11,7 @@ from nominatim_db.tools import freeze
|
||||
|
||||
NOMINATIM_RUNTIME_TABLES = [
|
||||
'country_name', 'country_osm_grid',
|
||||
'location_postcode', 'location_property_osmline', 'location_property_tiger',
|
||||
'location_postcodes', 'location_property_osmline', 'location_property_tiger',
|
||||
'placex', 'place_addressline',
|
||||
'search_name',
|
||||
'word'
|
||||
|
||||
@@ -13,25 +13,29 @@ import pytest
|
||||
|
||||
from nominatim_db.tools import postcodes
|
||||
from nominatim_db.data import country_info
|
||||
from nominatim_db.db.sql_preprocessor import SQLPreprocessor
|
||||
|
||||
import dummy_tokenizer
|
||||
|
||||
|
||||
class MockPostcodeTable:
|
||||
""" A location_postcode table for testing.
|
||||
""" A location_postcodes table for testing.
|
||||
"""
|
||||
def __init__(self, conn):
|
||||
def __init__(self, conn, config):
|
||||
self.conn = conn
|
||||
SQLPreprocessor(conn, config).run_sql_file(conn, 'functions/postcode_triggers.sql')
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("""CREATE TABLE location_postcode (
|
||||
cur.execute("""CREATE TABLE location_postcodes (
|
||||
place_id BIGINT,
|
||||
osm_id BIGINT,
|
||||
parent_place_id BIGINT,
|
||||
rank_search SMALLINT,
|
||||
rank_address SMALLINT,
|
||||
indexed_status SMALLINT,
|
||||
indexed_date TIMESTAMP,
|
||||
country_code varchar(2),
|
||||
postcode TEXT,
|
||||
geometry GEOMETRY(Geometry, 4326))""")
|
||||
geometry GEOMETRY(Geometry, 4326),
|
||||
centroid GEOMETRY(Point, 4326))""")
|
||||
cur.execute("""CREATE OR REPLACE FUNCTION token_normalized_postcode(postcode TEXT)
|
||||
RETURNS TEXT AS $$ BEGIN RETURN postcode; END; $$ LANGUAGE plpgsql;
|
||||
|
||||
@@ -40,131 +44,205 @@ class MockPostcodeTable:
|
||||
RETURN null;
|
||||
END; $$ LANGUAGE plpgsql;
|
||||
""")
|
||||
cur.execute("""CREATE OR REPLACE FUNCTION expand_by_meters(geom GEOMETRY, meters FLOAT)
|
||||
RETURNS GEOMETRY AS $$
|
||||
SELECT ST_Envelope(ST_Buffer(geom::geography, meters, 1)::geometry)
|
||||
$$ LANGUAGE sql;""")
|
||||
|
||||
conn.commit()
|
||||
|
||||
def add(self, country, postcode, x, y):
|
||||
with self.conn.cursor() as cur:
|
||||
cur.execute("""INSERT INTO location_postcode (place_id, indexed_status,
|
||||
country_code, postcode,
|
||||
geometry)
|
||||
VALUES (nextval('seq_place'), 1, %s, %s,
|
||||
ST_SetSRID(ST_MakePoint(%s, %s), 4326))""",
|
||||
(country, postcode, x, y))
|
||||
cur.execute(
|
||||
"""INSERT INTO location_postcodes
|
||||
(place_id, indexed_status, country_code, postcode, centroid, geometry)
|
||||
VALUES (nextval('seq_place'), 1, %(cc)s, %(pc)s,
|
||||
ST_SetSRID(ST_MakePoint(%(x)s, %(y)s), 4326),
|
||||
ST_Expand(ST_SetSRID(ST_MakePoint(%(x)s, %(y)s), 4326), 0.005))""",
|
||||
{'cc': country, 'pc': postcode, 'x': x, 'y': y})
|
||||
|
||||
self.conn.commit()
|
||||
|
||||
@property
|
||||
def row_set(self):
|
||||
with self.conn.cursor() as cur:
|
||||
cur.execute("""SELECT country_code, postcode,
|
||||
ST_X(geometry), ST_Y(geometry)
|
||||
FROM location_postcode""")
|
||||
cur.execute("""SELECT osm_id, country_code, postcode,
|
||||
ST_X(centroid), ST_Y(centroid)
|
||||
FROM location_postcodes""")
|
||||
return set((tuple(row) for row in cur))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tokenizer():
|
||||
return dummy_tokenizer.DummyTokenizer(None)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def postcode_table(def_config, temp_db_conn, placex_table):
|
||||
def postcode_table(def_config, temp_db_conn, placex_table, table_factory):
|
||||
country_info.setup_country_config(def_config)
|
||||
return MockPostcodeTable(temp_db_conn)
|
||||
table_factory('country_name', 'partition INT', ((0, ), (1, ), (2, )))
|
||||
return MockPostcodeTable(temp_db_conn, def_config)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def insert_implicit_postcode(placex_table, place_row):
|
||||
"""
|
||||
Inserts data into the placex and place table
|
||||
def insert_implicit_postcode(placex_table, place_postcode_row):
|
||||
""" Insert data into the placex and place table
|
||||
which can then be used to compute one postcode.
|
||||
"""
|
||||
def _insert_implicit_postcode(osm_id, country, geometry, address):
|
||||
placex_table.add(osm_id=osm_id, country=country, geom=geometry)
|
||||
place_row(osm_id=osm_id, geom='SRID=4326;'+geometry, address=address)
|
||||
def _insert_implicit_postcode(osm_id, country, geometry, postcode, in_placex=False):
|
||||
if in_placex:
|
||||
placex_table.add(osm_id=osm_id, country=country, geom=geometry,
|
||||
centroid=f'SRID=4326;{geometry}',
|
||||
address={'postcode': postcode})
|
||||
else:
|
||||
place_postcode_row(osm_id=osm_id, centroid=geometry,
|
||||
country=country, postcode=postcode)
|
||||
|
||||
return _insert_implicit_postcode
|
||||
|
||||
|
||||
def test_postcodes_empty(dsn, postcode_table, place_table, tokenizer):
|
||||
postcodes.update_postcodes(dsn, None, tokenizer)
|
||||
@pytest.fixture
|
||||
def insert_postcode_area(place_postcode_row):
|
||||
""" Insert an area around a centroid to the postcode table.
|
||||
"""
|
||||
def _do(osm_id, country, postcode, x, y):
|
||||
x1, x2, y1, y2 = x - 0.001, x + 0.001, y - 0.001, y + 0.001
|
||||
place_postcode_row(osm_type='R', osm_id=osm_id, postcode=postcode, country=country,
|
||||
centroid=f"POINT({x} {y})",
|
||||
geom=f"POLYGON(({x1} {y1}, {x1} {y2}, {x2} {y2}, {x2} {y1}, {x1} {y1}))")
|
||||
|
||||
return _do
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def postcode_update(dsn, temp_db_conn):
|
||||
tokenizer = dummy_tokenizer.DummyTokenizer(None)
|
||||
|
||||
def _do(data_path=None):
|
||||
with temp_db_conn.cursor() as cur:
|
||||
cur.execute("""CREATE TRIGGER location_postcodes_before_update
|
||||
BEFORE UPDATE ON location_postcodes
|
||||
FOR EACH ROW EXECUTE PROCEDURE postcodes_update()""")
|
||||
cur.execute("""CREATE TRIGGER location_postcodes_before_delete
|
||||
BEFORE DELETE ON location_postcodes
|
||||
FOR EACH ROW EXECUTE PROCEDURE postcodes_delete()""")
|
||||
cur.execute("""CREATE TRIGGER location_postcodes_before_insert
|
||||
BEFORE INSERT ON location_postcodes
|
||||
FOR EACH ROW EXECUTE PROCEDURE postcodes_insert()""")
|
||||
temp_db_conn.commit()
|
||||
|
||||
postcodes.update_postcodes(dsn, data_path, tokenizer)
|
||||
|
||||
return _do
|
||||
|
||||
|
||||
def test_postcodes_empty(postcode_update, postcode_table, place_postcode_table):
|
||||
postcode_update()
|
||||
|
||||
assert not postcode_table.row_set
|
||||
|
||||
|
||||
def test_postcodes_add_new(dsn, postcode_table, insert_implicit_postcode, tokenizer):
|
||||
insert_implicit_postcode(1, 'xx', 'POINT(10 12)', dict(postcode='9486'))
|
||||
@pytest.mark.parametrize('in_placex', [True, False])
|
||||
def test_postcodes_add_new_point(postcode_update, postcode_table,
|
||||
insert_implicit_postcode, in_placex):
|
||||
insert_implicit_postcode(1, 'xx', 'POINT(10 12)', '9486', in_placex)
|
||||
postcode_table.add('yy', '9486', 99, 34)
|
||||
|
||||
postcodes.update_postcodes(dsn, None, tokenizer)
|
||||
postcode_update()
|
||||
|
||||
assert postcode_table.row_set == {('xx', '9486', 10, 12), }
|
||||
assert postcode_table.row_set == {(None, 'xx', '9486', 10, 12), }
|
||||
|
||||
|
||||
def test_postcodes_add_new_area(postcode_update, insert_postcode_area, postcode_table):
|
||||
insert_postcode_area(345, 'de', '10445', 23.5, 46.2)
|
||||
|
||||
postcode_update()
|
||||
|
||||
assert postcode_table.row_set == {(345, 'de', '10445', 23.5, 46.2)}
|
||||
|
||||
|
||||
@pytest.mark.parametrize('in_placex', [True, False])
|
||||
def test_postcodes_add_area_and_point(postcode_update, insert_postcode_area,
|
||||
insert_implicit_postcode, postcode_table, in_placex):
|
||||
insert_implicit_postcode(1, 'xx', 'POINT(10 12)', '10445', in_placex)
|
||||
insert_postcode_area(345, 'xx', '10445', 23.5, 46.2)
|
||||
|
||||
postcode_update()
|
||||
|
||||
assert postcode_table.row_set == {(345, 'xx', '10445', 23.5, 46.2)}
|
||||
|
||||
|
||||
@pytest.mark.parametrize('in_placex', [True, False])
|
||||
def test_postcodes_add_point_within_area(postcode_update, insert_postcode_area,
|
||||
insert_implicit_postcode, postcode_table, in_placex):
|
||||
insert_implicit_postcode(1, 'xx', 'POINT(23.5 46.2)', '10446', in_placex)
|
||||
insert_postcode_area(345, 'xx', '10445', 23.5, 46.2)
|
||||
|
||||
postcode_update()
|
||||
|
||||
assert postcode_table.row_set == {(345, 'xx', '10445', 23.5, 46.2)}
|
||||
|
||||
|
||||
@pytest.mark.parametrize('coords', [(99, 34), (10, 34), (99, 12),
|
||||
(9, 34), (9, 11), (23, 11)])
|
||||
def test_postcodes_replace_coordinates(dsn, postcode_table, tmp_path,
|
||||
insert_implicit_postcode, tokenizer, coords):
|
||||
insert_implicit_postcode(1, 'xx', 'POINT(10 12)', dict(postcode='AB 4511'))
|
||||
def test_postcodes_replace_coordinates(postcode_update, postcode_table, tmp_path,
|
||||
insert_implicit_postcode, coords):
|
||||
insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
|
||||
postcode_table.add('xx', 'AB 4511', *coords)
|
||||
|
||||
postcodes.update_postcodes(dsn, tmp_path, tokenizer)
|
||||
postcode_update(tmp_path)
|
||||
|
||||
assert postcode_table.row_set == {('xx', 'AB 4511', 10, 12)}
|
||||
assert postcode_table.row_set == {(None, 'xx', 'AB 4511', 10, 12)}
|
||||
|
||||
|
||||
def test_postcodes_replace_coordinates_close(dsn, postcode_table,
|
||||
insert_implicit_postcode, tokenizer):
|
||||
insert_implicit_postcode(1, 'xx', 'POINT(10 12)', dict(postcode='AB 4511'))
|
||||
def test_postcodes_replace_coordinates_close(postcode_update, postcode_table,
|
||||
insert_implicit_postcode):
|
||||
insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
|
||||
postcode_table.add('xx', 'AB 4511', 10, 11.99999999)
|
||||
|
||||
postcodes.update_postcodes(dsn, None, tokenizer)
|
||||
postcode_update()
|
||||
|
||||
assert postcode_table.row_set == {('xx', 'AB 4511', 10, 11.99999999)}
|
||||
assert postcode_table.row_set == {(None, 'xx', 'AB 4511', 10, 11.99999999)}
|
||||
|
||||
|
||||
def test_postcodes_remove(dsn, postcode_table,
|
||||
insert_implicit_postcode, tokenizer):
|
||||
insert_implicit_postcode(1, 'xx', 'POINT(10 12)', dict(postcode='AB 4511'))
|
||||
def test_postcodes_remove_point(postcode_update, postcode_table,
|
||||
insert_implicit_postcode):
|
||||
insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
|
||||
postcode_table.add('xx', 'badname', 10, 12)
|
||||
|
||||
postcodes.update_postcodes(dsn, None, tokenizer)
|
||||
postcode_update()
|
||||
|
||||
assert postcode_table.row_set == {('xx', 'AB 4511', 10, 12)}
|
||||
assert postcode_table.row_set == {(None, 'xx', 'AB 4511', 10, 12)}
|
||||
|
||||
|
||||
def test_postcodes_ignore_empty_country(dsn, postcode_table,
|
||||
insert_implicit_postcode, tokenizer):
|
||||
insert_implicit_postcode(1, None, 'POINT(10 12)', dict(postcode='AB 4511'))
|
||||
postcodes.update_postcodes(dsn, None, tokenizer)
|
||||
def test_postcodes_ignore_empty_country(postcode_update, postcode_table,
|
||||
insert_implicit_postcode):
|
||||
insert_implicit_postcode(1, None, 'POINT(10 12)', 'AB 4511')
|
||||
postcode_update()
|
||||
assert not postcode_table.row_set
|
||||
|
||||
|
||||
def test_postcodes_remove_all(dsn, postcode_table, place_table, tokenizer):
|
||||
def test_postcodes_remove_all(postcode_update, postcode_table, place_postcode_table):
|
||||
postcode_table.add('ch', '5613', 10, 12)
|
||||
postcodes.update_postcodes(dsn, None, tokenizer)
|
||||
postcode_update()
|
||||
|
||||
assert not postcode_table.row_set
|
||||
|
||||
|
||||
def test_postcodes_multi_country(dsn, postcode_table,
|
||||
insert_implicit_postcode, tokenizer):
|
||||
insert_implicit_postcode(1, 'de', 'POINT(10 12)', dict(postcode='54451'))
|
||||
insert_implicit_postcode(2, 'cc', 'POINT(100 56)', dict(postcode='DD23 T'))
|
||||
insert_implicit_postcode(3, 'de', 'POINT(10.3 11.0)', dict(postcode='54452'))
|
||||
insert_implicit_postcode(4, 'cc', 'POINT(10.3 11.0)', dict(postcode='54452'))
|
||||
def test_postcodes_multi_country(postcode_update, postcode_table,
|
||||
insert_implicit_postcode):
|
||||
insert_implicit_postcode(1, 'de', 'POINT(10 12)', '54451')
|
||||
insert_implicit_postcode(2, 'cc', 'POINT(100 56)', 'DD23 T')
|
||||
insert_implicit_postcode(3, 'de', 'POINT(10.3 11.0)', '54452')
|
||||
insert_implicit_postcode(4, 'cc', 'POINT(10.3 11.0)', '54452')
|
||||
|
||||
postcodes.update_postcodes(dsn, None, tokenizer)
|
||||
postcode_update()
|
||||
|
||||
assert postcode_table.row_set == {('de', '54451', 10, 12),
|
||||
('de', '54452', 10.3, 11.0),
|
||||
('cc', '54452', 10.3, 11.0),
|
||||
('cc', 'DD23 T', 100, 56)}
|
||||
assert postcode_table.row_set == {(None, 'de', '54451', 10, 12),
|
||||
(None, 'de', '54452', 10.3, 11.0),
|
||||
(None, 'cc', '54452', 10.3, 11.0),
|
||||
(None, 'cc', 'DD23 T', 100, 56)}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("gzipped", [True, False])
|
||||
def test_postcodes_extern(dsn, postcode_table, tmp_path,
|
||||
insert_implicit_postcode, tokenizer, gzipped):
|
||||
insert_implicit_postcode(1, 'xx', 'POINT(10 12)', dict(postcode='AB 4511'))
|
||||
def test_postcodes_extern(postcode_update, postcode_table, tmp_path,
|
||||
insert_implicit_postcode, gzipped):
|
||||
insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
|
||||
|
||||
extfile = tmp_path / 'xx_postcodes.csv'
|
||||
extfile.write_text("postcode,lat,lon\nAB 4511,-4,-1\nCD 4511,-5, -10")
|
||||
@@ -173,44 +251,44 @@ def test_postcodes_extern(dsn, postcode_table, tmp_path,
|
||||
subprocess.run(['gzip', str(extfile)])
|
||||
assert not extfile.is_file()
|
||||
|
||||
postcodes.update_postcodes(dsn, tmp_path, tokenizer)
|
||||
postcode_update(tmp_path)
|
||||
|
||||
assert postcode_table.row_set == {('xx', 'AB 4511', 10, 12),
|
||||
('xx', 'CD 4511', -10, -5)}
|
||||
assert postcode_table.row_set == {(None, 'xx', 'AB 4511', 10, 12),
|
||||
(None, 'xx', 'CD 4511', -10, -5)}
|
||||
|
||||
|
||||
def test_postcodes_extern_bad_column(dsn, postcode_table, tmp_path,
|
||||
insert_implicit_postcode, tokenizer):
|
||||
insert_implicit_postcode(1, 'xx', 'POINT(10 12)', dict(postcode='AB 4511'))
|
||||
def test_postcodes_extern_bad_column(postcode_update, postcode_table, tmp_path,
|
||||
insert_implicit_postcode):
|
||||
insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
|
||||
|
||||
extfile = tmp_path / 'xx_postcodes.csv'
|
||||
extfile.write_text("postode,lat,lon\nAB 4511,-4,-1\nCD 4511,-5, -10")
|
||||
|
||||
postcodes.update_postcodes(dsn, tmp_path, tokenizer)
|
||||
postcode_update(tmp_path)
|
||||
|
||||
assert postcode_table.row_set == {('xx', 'AB 4511', 10, 12)}
|
||||
assert postcode_table.row_set == {(None, 'xx', 'AB 4511', 10, 12)}
|
||||
|
||||
|
||||
def test_postcodes_extern_bad_number(dsn, insert_implicit_postcode,
|
||||
postcode_table, tmp_path, tokenizer):
|
||||
insert_implicit_postcode(1, 'xx', 'POINT(10 12)', dict(postcode='AB 4511'))
|
||||
def test_postcodes_extern_bad_number(postcode_update, insert_implicit_postcode,
|
||||
postcode_table, tmp_path):
|
||||
insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
|
||||
|
||||
extfile = tmp_path / 'xx_postcodes.csv'
|
||||
extfile.write_text("postcode,lat,lon\nXX 4511,-4,NaN\nCD 4511,-5, -10\n34,200,0")
|
||||
|
||||
postcodes.update_postcodes(dsn, tmp_path, tokenizer)
|
||||
postcode_update(tmp_path)
|
||||
|
||||
assert postcode_table.row_set == {('xx', 'AB 4511', 10, 12),
|
||||
('xx', 'CD 4511', -10, -5)}
|
||||
assert postcode_table.row_set == {(None, 'xx', 'AB 4511', 10, 12),
|
||||
(None, 'xx', 'CD 4511', -10, -5)}
|
||||
|
||||
|
||||
def test_can_compute(dsn, table_factory):
|
||||
assert not postcodes.can_compute(dsn)
|
||||
table_factory('place')
|
||||
table_factory('place_postcode')
|
||||
assert postcodes.can_compute(dsn)
|
||||
|
||||
|
||||
def test_no_placex_entry(dsn, temp_db_cursor, place_row, postcode_table, tokenizer):
|
||||
def test_no_placex_entry(postcode_update, temp_db_cursor, place_postcode_row, postcode_table):
|
||||
# Rewrite the get_country_code function to verify its execution.
|
||||
temp_db_cursor.execute("""
|
||||
CREATE OR REPLACE FUNCTION get_country_code(place geometry)
|
||||
@@ -218,22 +296,14 @@ def test_no_placex_entry(dsn, temp_db_cursor, place_row, postcode_table, tokeniz
|
||||
RETURN 'yy';
|
||||
END; $$ LANGUAGE plpgsql;
|
||||
""")
|
||||
place_row(geom='SRID=4326;POINT(10 12)', address=dict(postcode='AB 4511'))
|
||||
postcodes.update_postcodes(dsn, None, tokenizer)
|
||||
place_postcode_row(centroid='POINT(10 12)', postcode='AB 4511')
|
||||
postcode_update()
|
||||
|
||||
assert postcode_table.row_set == {('yy', 'AB 4511', 10, 12)}
|
||||
assert postcode_table.row_set == {(None, 'yy', 'AB 4511', 10, 12)}
|
||||
|
||||
|
||||
def test_discard_badly_formatted_postcodes(dsn, temp_db_cursor, place_row,
|
||||
postcode_table, tokenizer):
|
||||
# Rewrite the get_country_code function to verify its execution.
|
||||
temp_db_cursor.execute("""
|
||||
CREATE OR REPLACE FUNCTION get_country_code(place geometry)
|
||||
RETURNS TEXT AS $$ BEGIN
|
||||
RETURN 'fr';
|
||||
END; $$ LANGUAGE plpgsql;
|
||||
""")
|
||||
place_row(geom='SRID=4326;POINT(10 12)', address=dict(postcode='AB 4511'))
|
||||
postcodes.update_postcodes(dsn, None, tokenizer)
|
||||
def test_discard_badly_formatted_postcodes(postcode_update, place_postcode_row, postcode_table):
|
||||
place_postcode_row(centroid='POINT(10 12)', country='fr', postcode='AB 4511')
|
||||
postcode_update()
|
||||
|
||||
assert not postcode_table.row_set
|
||||
|
||||
Reference in New Issue
Block a user