safety-screen/Nominatim
1
0
Fork 0
forked from hans/Nominatim
Code Pull Requests Activity

bdd: switch to auto commit mode

Browse Source 
Put the connection to the test database into auto-commit mode
and get rid of the explicit commits. Also use cursors always in
context managers and unify the two implementations that copy
data from the place table.
This commit is contained in:
Sarah Hoffmann
2021-01-05 11:22:43 +01:00
parent 58c471c627
commit 5dfa76a610
3 changed files with 289 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
test/bdd/steps/nominatim_environment.py
View File

@@ -179,6 +179,7 @@ class NominatimEnvironment:
cur.execute('CREATE DATABASE {} TEMPLATE = {}'.format(self.test_db, self.template_db))
conn.close()
context.db = self.connect_database(self.test_db)
context.db.autocommit = True
psycopg2.extras.register_hstore(context.db, globally=False)
def teardown_db(self, context):
@@ -215,3 +216,25 @@ class NominatimEnvironment:
cwd = self.build_dir
run_script(cmd, cwd=cwd, env=self.test_env)
def copy_from_place(self, db):
""" Copy data from place to the placex and location_property_osmline
tables invoking the appropriate triggers.
"""
self.run_setup_script('create-functions', 'create-partition-functions')
with db.cursor() as cur:
cur.execute("""INSERT INTO placex (osm_type, osm_id, class, type,
name, admin_level, address,
extratags, geometry)
SELECT osm_type, osm_id, class, type,
name, admin_level, address,
extratags, geometry
FROM place
WHERE not (class='place' and type='houses' and osm_type='W')""")
cur.execute("""INSERT INTO location_property_osmline (osm_id, address, linegeo)
SELECT osm_id, address, geometry
FROM place
WHERE class='place' and type='houses'
and osm_type='W'
and ST_GeometryType(geometry) = 'ST_LineString'""")

563
test/bdd/steps/steps_db_ops.py
View File

@@ -218,7 +218,7 @@ def assert_db_column(row, column, value, context):
"Row '%s': expected: %s, got: %s" % (column, value, str(row[column]))
################################ STEPS ##################################
################################ GIVEN ##################################
@given(u'the scene (?P<scene>.+)')
def set_default_scene(context, scene):
@@ -226,83 +226,72 @@ def set_default_scene(context, scene):
@given("the (?P<named>named )?places")
def add_data_to_place_table(context, named):
cur = context.db.cursor()
cur.execute('ALTER TABLE place DISABLE TRIGGER place_before_insert')
for r in context.table:
col = PlaceColumn(context, named is not None)
with context.db.cursor() as cur:
cur.execute('ALTER TABLE place DISABLE TRIGGER place_before_insert')
for r in context.table:
col = PlaceColumn(context, named is not None)
for h in r.headings:
col.add(h, r[h])
for h in r.headings:
col.add(h, r[h])
col.db_insert(cur)
cur.execute('ALTER TABLE place ENABLE TRIGGER place_before_insert')
cur.close()
context.db.commit()
col.db_insert(cur)
cur.execute('ALTER TABLE place ENABLE TRIGGER place_before_insert')
@given("the relations")
def add_data_to_planet_relations(context):
cur = context.db.cursor()
for r in context.table:
last_node = 0
last_way = 0
parts = []
if r['members']:
members = []
for m in r['members'].split(','):
mid = NominatimID(m)
if mid.typ == 'N':
parts.insert(last_node, int(mid.oid))
last_node += 1
last_way += 1
elif mid.typ == 'W':
parts.insert(last_way, int(mid.oid))
last_way += 1
else:
parts.append(int(mid.oid))
with context.db.cursor() as cur:
for r in context.table:
last_node = 0
last_way = 0
parts = []
if r['members']:
members = []
for m in r['members'].split(','):
mid = NominatimID(m)
if mid.typ == 'N':
parts.insert(last_node, int(mid.oid))
last_node += 1
last_way += 1
elif mid.typ == 'W':
parts.insert(last_way, int(mid.oid))
last_way += 1
else:
parts.append(int(mid.oid))
members.extend((mid.typ.lower() + mid.oid, mid.cls or ''))
else:
members = None
members.extend((mid.typ.lower() + mid.oid, mid.cls or ''))
else:
members = None
tags = []
for h in r.headings:
if h.startswith("tags+"):
tags.extend((h[5:], r[h]))
tags = []
for h in r.headings:
if h.startswith("tags+"):
tags.extend((h[5:], r[h]))
cur.execute("""INSERT INTO planet_osm_rels (id, way_off, rel_off, parts, members, tags)
VALUES (%s, %s, %s, %s, %s, %s)""",
(r['id'], last_node, last_way, parts, members, tags))
context.db.commit()
cur.execute("""INSERT INTO planet_osm_rels (id, way_off, rel_off, parts, members, tags)
VALUES (%s, %s, %s, %s, %s, %s)""",
(r['id'], last_node, last_way, parts, members, tags))
@given("the ways")
def add_data_to_planet_ways(context):
cur = context.db.cursor()
for r in context.table:
tags = []
for h in r.headings:
if h.startswith("tags+"):
tags.extend((h[5:], r[h]))
with context.db.cursor() as cur:
for r in context.table:
tags = []
for h in r.headings:
if h.startswith("tags+"):
tags.extend((h[5:], r[h]))
nodes = [ int(x.strip()) for x in r['nodes'].split(',') ]
nodes = [ int(x.strip()) for x in r['nodes'].split(',') ]
cur.execute("INSERT INTO planet_osm_ways (id, nodes, tags) VALUES (%s, %s, %s)",
(r['id'], nodes, tags))
context.db.commit()
cur.execute("INSERT INTO planet_osm_ways (id, nodes, tags) VALUES (%s, %s, %s)",
(r['id'], nodes, tags))
################################ WHEN ##################################
@when("importing")
def import_and_index_data_from_place_table(context):
context.nominatim.run_setup_script('create-functions', 'create-partition-functions')
cur = context.db.cursor()
cur.execute(
"""insert into placex (osm_type, osm_id, class, type, name, admin_level, address, extratags, geometry)
select osm_type, osm_id, class, type, name, admin_level, address, extratags, geometry
from place where not (class='place' and type='houses' and osm_type='W')""")
cur.execute(
"""insert into location_property_osmline (osm_id, address, linegeo)
SELECT osm_id, address, geometry from place
WHERE class='place' and type='houses' and osm_type='W'
and ST_GeometryType(geometry) = 'ST_LineString'""")
context.db.commit()
""" Import data previously set up in the place table.
"""
context.nominatim.copy_from_place(context.db)
context.nominatim.run_setup_script('calculate-postcodes', 'index', 'index-noanalyse')
check_database_integrity(context)
@@ -310,24 +299,21 @@ def import_and_index_data_from_place_table(context):
def update_place_table(context):
context.nominatim.run_setup_script(
'create-functions', 'create-partition-functions', 'enable-diff-updates')
cur = context.db.cursor()
for r in context.table:
col = PlaceColumn(context, False)
with context.db.cursor() as cur:
for r in context.table:
col = PlaceColumn(context, False)
for h in r.headings:
col.add(h, r[h])
for h in r.headings:
col.add(h, r[h])
col.db_insert(cur)
col.db_insert(cur)
context.db.commit()
while True:
context.nominatim.run_update_script('index')
while True:
context.nominatim.run_update_script('index')
cur = context.db.cursor()
cur.execute("SELECT 'a' FROM placex WHERE indexed_status != 0 LIMIT 1")
if cur.rowcount == 0:
break
cur.execute("SELECT 'a' FROM placex WHERE indexed_status != 0 LIMIT 1")
if cur.rowcount == 0:
break
check_database_integrity(context)
@@ -339,277 +325,258 @@ def update_postcodes(context):
def delete_places(context, oids):
context.nominatim.run_setup_script(
'create-functions', 'create-partition-functions', 'enable-diff-updates')
cur = context.db.cursor()
for oid in oids.split(','):
where, params = NominatimID(oid).table_select()
cur.execute("DELETE FROM place WHERE " + where, params)
context.db.commit()
with context.db.cursor() as cur:
for oid in oids.split(','):
where, params = NominatimID(oid).table_select()
cur.execute("DELETE FROM place WHERE " + where, params)
while True:
context.nominatim.run_update_script('index')
cur = context.db.cursor()
cur.execute("SELECT 'a' FROM placex WHERE indexed_status != 0 LIMIT 1")
if cur.rowcount == 0:
break
with context.db.cursor() as cur:
cur.execute("SELECT 'a' FROM placex WHERE indexed_status != 0 LIMIT 1")
if cur.rowcount == 0:
break
################################ THEN ##################################
@then("placex contains(?P<exact> exactly)?")
def check_placex_contents(context, exact):
cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
expected_content = set()
for row in context.table:
nid = NominatimID(row['object'])
where, params = nid.table_select()
cur.execute("""SELECT *, ST_AsText(geometry) as geomtxt,
ST_X(centroid) as cx, ST_Y(centroid) as cy
FROM placex where %s""" % where,
params)
assert cur.rowcount > 0, "No rows found for " + row['object']
expected_content = set()
for row in context.table:
nid = NominatimID(row['object'])
where, params = nid.table_select()
cur.execute("""SELECT *, ST_AsText(geometry) as geomtxt,
ST_X(centroid) as cx, ST_Y(centroid) as cy
FROM placex where %s""" % where,
params)
assert cur.rowcount > 0, "No rows found for " + row['object']
for res in cur:
if exact:
expected_content.add((res['osm_type'], res['osm_id'], res['class']))
for h in row.headings:
if h in ('extratags', 'address'):
if row[h] == '-':
assert res[h] is None
for res in cur:
if exact:
expected_content.add((res['osm_type'], res['osm_id'], res['class']))
for h in row.headings:
if h in ('extratags', 'address'):
if row[h] == '-':
assert res[h] is None
else:
vdict = eval('{' + row[h] + '}')
assert vdict == res[h]
elif h.startswith('name'):
name = h[5:] if h.startswith('name+') else 'name'
assert name in res['name']
assert res['name'][name] == row[h]
elif h.startswith('extratags+'):
assert res['extratags'][h[10:]] == row[h]
elif h.startswith('addr+'):
if row[h] == '-':
if res['address'] is not None:
assert h[5:] not in res['address']
else:
assert h[5:] in res['address'], "column " + h
assert res['address'][h[5:]] == row[h], "column %s" % h
elif h in ('linked_place_id', 'parent_place_id'):
compare_place_id(row[h], res[h], h, context)
else:
vdict = eval('{' + row[h] + '}')
assert vdict == res[h]
elif h.startswith('name'):
name = h[5:] if h.startswith('name+') else 'name'
assert name in res['name']
assert res['name'][name] == row[h]
elif h.startswith('extratags+'):
assert res['extratags'][h[10:]] == row[h]
elif h.startswith('addr+'):
if row[h] == '-':
if res['address'] is not None:
assert h[5:] not in res['address']
else:
assert h[5:] in res['address'], "column " + h
assert res['address'][h[5:]] == row[h], "column %s" % h
elif h in ('linked_place_id', 'parent_place_id'):
compare_place_id(row[h], res[h], h, context)
else:
assert_db_column(res, h, row[h], context)
assert_db_column(res, h, row[h], context)
if exact:
cur.execute('SELECT osm_type, osm_id, class from placex')
assert expected_content == set([(r[0], r[1], r[2]) for r in cur])
context.db.commit()
if exact:
cur.execute('SELECT osm_type, osm_id, class from placex')
assert expected_content == set([(r[0], r[1], r[2]) for r in cur])
@then("place contains(?P<exact> exactly)?")
def check_placex_contents(context, exact):
cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
expected_content = set()
for row in context.table:
nid = NominatimID(row['object'])
where, params = nid.table_select()
cur.execute("""SELECT *, ST_AsText(geometry) as geomtxt,
ST_GeometryType(geometry) as geometrytype
FROM place where %s""" % where,
params)
assert cur.rowcount > 0, "No rows found for " + row['object']
expected_content = set()
for row in context.table:
nid = NominatimID(row['object'])
where, params = nid.table_select()
cur.execute("""SELECT *, ST_AsText(geometry) as geomtxt,
ST_GeometryType(geometry) as geometrytype
FROM place where %s""" % where,
params)
assert cur.rowcount > 0, "No rows found for " + row['object']
for res in cur:
if exact:
expected_content.add((res['osm_type'], res['osm_id'], res['class']))
for h in row.headings:
msg = "%s: %s" % (row['object'], h)
if h in ('name', 'extratags', 'address'):
if row[h] == '-':
assert res[h] is None, msg
for res in cur:
if exact:
expected_content.add((res['osm_type'], res['osm_id'], res['class']))
for h in row.headings:
msg = "%s: %s" % (row['object'], h)
if h in ('name', 'extratags', 'address'):
if row[h] == '-':
assert res[h] is None, msg
else:
vdict = eval('{' + row[h] + '}')
assert vdict == res[h], msg
elif h.startswith('name+'):
assert res['name'][h[5:]] == row[h], msg
elif h.startswith('extratags+'):
assert res['extratags'][h[10:]] == row[h], msg
elif h.startswith('addr+'):
if row[h] == '-':
if res['address'] is not None:
assert h[5:] not in res['address']
else:
assert res['address'][h[5:]] == row[h], msg
elif h in ('linked_place_id', 'parent_place_id'):
compare_place_id(row[h], res[h], h, context)
else:
vdict = eval('{' + row[h] + '}')
assert vdict == res[h], msg
elif h.startswith('name+'):
assert res['name'][h[5:]] == row[h], msg
elif h.startswith('extratags+'):
assert res['extratags'][h[10:]] == row[h], msg
elif h.startswith('addr+'):
if row[h] == '-':
if res['address'] is not None:
assert h[5:] not in res['address']
else:
assert res['address'][h[5:]] == row[h], msg
elif h in ('linked_place_id', 'parent_place_id'):
compare_place_id(row[h], res[h], h, context)
else:
assert_db_column(res, h, row[h], context)
assert_db_column(res, h, row[h], context)
if exact:
cur.execute('SELECT osm_type, osm_id, class from place')
assert expected_content, set([(r[0], r[1], r[2]) for r in cur])
context.db.commit()
if exact:
cur.execute('SELECT osm_type, osm_id, class from place')
assert expected_content, set([(r[0], r[1], r[2]) for r in cur])
@then("search_name contains(?P<exclude> not)?")
def check_search_name_contents(context, exclude):
cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
for row in context.table:
pid = NominatimID(row['object']).get_place_id(cur)
cur.execute("""SELECT *, ST_X(centroid) as cx, ST_Y(centroid) as cy
FROM search_name WHERE place_id = %s""", (pid, ))
assert cur.rowcount > 0, "No rows found for " + row['object']
for row in context.table:
pid = NominatimID(row['object']).get_place_id(cur)
cur.execute("""SELECT *, ST_X(centroid) as cx, ST_Y(centroid) as cy
FROM search_name WHERE place_id = %s""", (pid, ))
assert cur.rowcount > 0, "No rows found for " + row['object']
for res in cur:
for h in row.headings:
if h in ('name_vector', 'nameaddress_vector'):
terms = [x.strip() for x in row[h].split(',') if not x.strip().startswith('#')]
words = [x.strip()[1:] for x in row[h].split(',') if x.strip().startswith('#')]
subcur = context.db.cursor()
subcur.execute(""" SELECT word_id, word_token
FROM word, (SELECT unnest(%s::TEXT[]) as term) t
WHERE word_token = make_standard_name(t.term)
and class is null and country_code is null
and operator is null
UNION
SELECT word_id, word_token
FROM word, (SELECT unnest(%s::TEXT[]) as term) t
WHERE word_token = ' ' || make_standard_name(t.term)
and class is null and country_code is null
and operator is null
""",
(terms, words))
if not exclude:
assert subcur.rowcount >= len(terms) + len(words), \
"No word entry found for " + row[h] + ". Entries found: " + str(subcur.rowcount)
for wid in subcur:
if exclude:
assert wid[0] not in res[h], "Found term for %s/%s: %s" % (pid, h, wid[1])
else:
assert wid[0] in res[h], "Missing term for %s/%s: %s" % (pid, h, wid[1])
else:
assert_db_column(res, h, row[h], context)
context.db.commit()
for res in cur:
for h in row.headings:
if h in ('name_vector', 'nameaddress_vector'):
terms = [x.strip() for x in row[h].split(',') if not x.strip().startswith('#')]
words = [x.strip()[1:] for x in row[h].split(',') if x.strip().startswith('#')]
with context.db.cursor() as subcur:
subcur.execute(""" SELECT word_id, word_token
FROM word, (SELECT unnest(%s::TEXT[]) as term) t
WHERE word_token = make_standard_name(t.term)
and class is null and country_code is null
and operator is null
UNION
SELECT word_id, word_token
FROM word, (SELECT unnest(%s::TEXT[]) as term) t
WHERE word_token = ' ' || make_standard_name(t.term)
and class is null and country_code is null
and operator is null
""",
(terms, words))
if not exclude:
assert subcur.rowcount >= len(terms) + len(words), \
"No word entry found for " + row[h] + ". Entries found: " + str(subcur.rowcount)
for wid in subcur:
if exclude:
assert wid[0] not in res[h], "Found term for %s/%s: %s" % (pid, h, wid[1])
else:
assert wid[0] in res[h], "Missing term for %s/%s: %s" % (pid, h, wid[1])
else:
assert_db_column(res, h, row[h], context)
@then("location_postcode contains exactly")
def check_location_postcode(context):
cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
cur.execute("SELECT *, ST_AsText(geometry) as geomtxt FROM location_postcode")
assert cur.rowcount == len(list(context.table)), \
"Postcode table has %d rows, expected %d rows." % (cur.rowcount, len(list(context.table)))
cur.execute("SELECT *, ST_AsText(geometry) as geomtxt FROM location_postcode")
assert cur.rowcount == len(list(context.table)), \
"Postcode table has %d rows, expected %d rows." % (cur.rowcount, len(list(context.table)))
table = list(cur)
for row in context.table:
for i in range(len(table)):
if table[i]['country_code'] != row['country'] \
or table[i]['postcode'] != row['postcode']:
continue
for h in row.headings:
if h not in ('country', 'postcode'):
assert_db_column(table[i], h, row[h], context)
table = list(cur)
for row in context.table:
for i in range(len(table)):
if table[i]['country_code'] != row['country'] \
or table[i]['postcode'] != row['postcode']:
continue
for h in row.headings:
if h not in ('country', 'postcode'):
assert_db_column(table[i], h, row[h], context)
@then("word contains(?P<exclude> not)?")
def check_word_table(context, exclude):
cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
for row in context.table:
wheres = []
values = []
for h in row.headings:
wheres.append("%s = %%s" % h)
values.append(row[h])
cur.execute("SELECT * from word WHERE %s" % ' AND '.join(wheres), values)
if exclude:
assert cur.rowcount == 0, "Row still in word table: %s" % '/'.join(values)
else:
assert cur.rowcount > 0, "Row not in word table: %s" % '/'.join(values)
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
for row in context.table:
wheres = []
values = []
for h in row.headings:
wheres.append("%s = %%s" % h)
values.append(row[h])
cur.execute("SELECT * from word WHERE %s" % ' AND '.join(wheres), values)
if exclude:
assert cur.rowcount == 0, "Row still in word table: %s" % '/'.join(values)
else:
assert cur.rowcount > 0, "Row not in word table: %s" % '/'.join(values)
@then("place_addressline contains")
def check_place_addressline(context):
cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
for row in context.table:
pid = NominatimID(row['object']).get_place_id(cur)
apid = NominatimID(row['address']).get_place_id(cur)
cur.execute(""" SELECT * FROM place_addressline
WHERE place_id = %s AND address_place_id = %s""",
(pid, apid))
assert cur.rowcount > 0, \
"No rows found for place %s and address %s" % (row['object'], row['address'])
for row in context.table:
pid = NominatimID(row['object']).get_place_id(cur)
apid = NominatimID(row['address']).get_place_id(cur)
cur.execute(""" SELECT * FROM place_addressline
WHERE place_id = %s AND address_place_id = %s""",
(pid, apid))
assert cur.rowcount > 0, \
"No rows found for place %s and address %s" % (row['object'], row['address'])
for res in cur:
for h in row.headings:
if h not in ('address', 'object'):
assert_db_column(res, h, row[h], context)
context.db.commit()
for res in cur:
for h in row.headings:
if h not in ('address', 'object'):
assert_db_column(res, h, row[h], context)
@then("place_addressline doesn't contain")
def check_place_addressline_exclude(context):
cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
for row in context.table:
pid = NominatimID(row['object']).get_place_id(cur)
apid = NominatimID(row['address']).get_place_id(cur)
cur.execute(""" SELECT * FROM place_addressline
WHERE place_id = %s AND address_place_id = %s""",
(pid, apid))
assert cur.rowcount == 0, \
"Row found for place %s and address %s" % (row['object'], row['address'])
context.db.commit()
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
for row in context.table:
pid = NominatimID(row['object']).get_place_id(cur)
apid = NominatimID(row['address']).get_place_id(cur)
cur.execute(""" SELECT * FROM place_addressline
WHERE place_id = %s AND address_place_id = %s""",
(pid, apid))
assert cur.rowcount == 0, \
"Row found for place %s and address %s" % (row['object'], row['address'])
@then("(?P<oid>\w+) expands to(?P<neg> no)? interpolation")
def check_location_property_osmline(context, oid, neg):
cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
nid = NominatimID(oid)
assert 'W' == nid.typ, "interpolation must be a way"
cur.execute("""SELECT *, ST_AsText(linegeo) as geomtxt
FROM location_property_osmline
WHERE osm_id = %s AND startnumber IS NOT NULL""",
(nid.oid, ))
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
cur.execute("""SELECT *, ST_AsText(linegeo) as geomtxt
FROM location_property_osmline
WHERE osm_id = %s AND startnumber IS NOT NULL""",
(nid.oid, ))
if neg:
assert cur.rowcount == 0
return
if neg:
assert cur.rowcount == 0
return
todo = list(range(len(list(context.table))))
for res in cur:
for i in todo:
row = context.table[i]
if (int(row['start']) == res['startnumber']
and int(row['end']) == res['endnumber']):
todo.remove(i)
break
else:
assert False, "Unexpected row %s" % (str(res))
for h in row.headings:
if h in ('start', 'end'):
continue
elif h == 'parent_place_id':
compare_place_id(row[h], res[h], h, context)
todo = list(range(len(list(context.table))))
for res in cur:
for i in todo:
row = context.table[i]
if (int(row['start']) == res['startnumber']
and int(row['end']) == res['endnumber']):
todo.remove(i)
break
else:
assert_db_column(res, h, row[h], context)
assert False, "Unexpected row %s" % (str(res))
assert not todo
for h in row.headings:
if h in ('start', 'end'):
continue
elif h == 'parent_place_id':
compare_place_id(row[h], res[h], h, context)
else:
assert_db_column(res, h, row[h], context)
assert not todo
@then("(?P<table>placex|place) has no entry for (?P<oid>.*)")
def check_placex_has_entry(context, table, oid):
cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
nid = NominatimID(oid)
where, params = nid.table_select()
cur.execute("SELECT * FROM %s where %s" % (table, where), params)
assert cur.rowcount == 0
context.db.commit()
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
nid = NominatimID(oid)
where, params = nid.table_select()
cur.execute("SELECT * FROM %s where %s" % (table, where), params)
assert cur.rowcount == 0
@then("search_name has no entry for (?P<oid>.*)")
def check_search_name_has_entry(context, oid):
cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
pid = NominatimID(oid).get_place_id(cur)
cur.execute("SELECT * FROM search_name WHERE place_id = %s", (pid, ))
assert cur.rowcount == 0
context.db.commit()
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
pid = NominatimID(oid).get_place_id(cur)
cur.execute("SELECT * FROM search_name WHERE place_id = %s", (pid, ))
assert cur.rowcount == 0

12
test/bdd/steps/steps_osm_data.py
View File

@@ -70,17 +70,7 @@ def update_from_osm_file(context):
The data is expected as attached text in OPL format.
"""
context.nominatim.run_setup_script('create-functions', 'create-partition-functions')
cur = context.db.cursor()
cur.execute("""insert into placex (osm_type, osm_id, class, type, name, admin_level, address, extratags, geometry)
select osm_type, osm_id, class, type, name, admin_level, address, extratags, geometry from place""")
cur.execute(
"""insert into location_property_osmline (osm_id, address, linegeo)
SELECT osm_id, address, geometry from place
WHERE class='place' and type='houses' and osm_type='W'
and ST_GeometryType(geometry) = 'ST_LineString'""")
context.db.commit()
context.nominatim.copy_from_place(context.db)
context.nominatim.run_setup_script('index', 'index-noanalyse')
context.nominatim.run_setup_script('create-functions', 'create-partition-functions',
'enable-diff-updates')