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

properly close connection in test

Browse Source
This commit is contained in:
Sarah Hoffmann
2015-01-09 23:16:53 +01:00
parent 6d0ab44800
commit 14d17a3ad4
1 changed files with 46 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
tests/steps/db_results.py
View File

@@ -22,7 +22,7 @@ logger = logging.getLogger(__name__)
@step(u'table placex contains as names for (N|R|W)(\d+)') @step(u'table placex contains as names for (N|R|W)(\d+)')
def check_placex_names(step, osmtyp, osmid): def check_placex_names(step, osmtyp, osmid):
""" Check for the exact content of the name hstaore in placex. """ Check for the exact content of the name hstore in placex.
""" """
cur = world.conn.cursor(cursor_factory=psycopg2.extras.DictCursor) cur = world.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
cur.execute('SELECT name FROM placex where osm_type = %s and osm_id =%s', (osmtyp, int(osmid))) cur.execute('SELECT name FROM placex where osm_type = %s and osm_id =%s', (osmtyp, int(osmid)))
@@ -43,47 +43,55 @@ def check_placex_content(step, tablename):
given columns are tested. If there is more than one given columns are tested. If there is more than one
line for an OSM object, they must match in these columns. line for an OSM object, they must match in these columns.
""" """
cur = world.conn.cursor(cursor_factory=psycopg2.extras.DictCursor) try:
for line in step.hashes: cur = world.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
osmtype, osmid, cls = world.split_id(line['object']) for line in step.hashes:
q = 'SELECT *' osmtype, osmid, cls = world.split_id(line['object'])
if tablename == 'placex': q = 'SELECT *'
q = q + ", ST_X(centroid) as clat, ST_Y(centroid) as clon" if tablename == 'placex':
q = q + ", ST_GeometryType(geometry) as geometrytype" q = q + ", ST_X(centroid) as clat, ST_Y(centroid) as clon"
q = q + ' FROM %s where osm_type = %%s and osm_id = %%s' % (tablename,) q = q + ", ST_GeometryType(geometry) as geometrytype"
if cls is None: q = q + ' FROM %s where osm_type = %%s and osm_id = %%s' % (tablename,)
params = (osmtype, osmid) if cls is None:
else: params = (osmtype, osmid)
q = q + ' and class = %s' else:
params = (osmtype, osmid, cls) q = q + ' and class = %s'
cur.execute(q, params) params = (osmtype, osmid, cls)
assert(cur.rowcount > 0) cur.execute(q, params)
for res in cur: assert(cur.rowcount > 0)
for k,v in line.iteritems(): for res in cur:
if not k == 'object': for k,v in line.iteritems():
assert_in(k, res) if not k == 'object':
if type(res[k]) is dict: assert_in(k, res)
val = world.make_hash(v) if type(res[k]) is dict:
assert_equals(res[k], val) val = world.make_hash(v)
elif k in ('parent_place_id', 'linked_place_id'): assert_equals(res[k], val)
pid = world.get_placeid(v) elif k in ('parent_place_id', 'linked_place_id'):
assert_equals(pid, res[k], "Results for '%s'/'%s' differ: '%s' != '%s'" % (line['object'], k, pid, res[k])) pid = world.get_placeid(v)
elif k == 'centroid': assert_equals(pid, res[k], "Results for '%s'/'%s' differ: '%s' != '%s'" % (line['object'], k, pid, res[k]))
world.match_geometry((res['clat'], res['clon']), v) elif k == 'centroid':
else: world.match_geometry((res['clat'], res['clon']), v)
assert_equals(str(res[k]), v, "Results for '%s'/'%s' differ: '%s' != '%s'" % (line['object'], k, str(res[k]), v)) else:
assert_equals(str(res[k]), v, "Results for '%s'/'%s' differ: '%s' != '%s'" % (line['object'], k, str(res[k]), v))
finally:
cur.close()
world.conn.commit()
@step(u'table (placex?) has no entry for (N|R|W)(\d+)(:\w+)?') @step(u'table (placex?) has no entry for (N|R|W)(\d+)(:\w+)?')
def check_placex_missing(step, tablename, osmtyp, osmid, placeclass): def check_placex_missing(step, tablename, osmtyp, osmid, placeclass):
cur = world.conn.cursor() cur = world.conn.cursor()
q = 'SELECT count(*) FROM %s where osm_type = %%s and osm_id = %%s' % (tablename, ) try:
args = [osmtyp, int(osmid)] q = 'SELECT count(*) FROM %s where osm_type = %%s and osm_id = %%s' % (tablename, )
if placeclass is not None: args = [osmtyp, int(osmid)]
q = q + ' and class = %s' if placeclass is not None:
args.append(placeclass[1:]) q = q + ' and class = %s'
cur.execute(q, args) args.append(placeclass[1:])
numres = cur.fetchone()[0] cur.execute(q, args)
assert_equals (numres, 0) numres = cur.fetchone()[0]
assert_equals (numres, 0)
finally:
cur.close()
world.conn.commit()
@step(u'search_name table contains$') @step(u'search_name table contains$')
def check_search_name_content(step): def check_search_name_content(step):