test/bdd/steps/db_ops.py

   1 import base64
   2 import random
   3 import string
   4 import re
   5 from nose.tools import * # for assert functions
   6 import psycopg2.extras
   7
   8 class PlaceColumn:
   9
  10     def __init__(self, context, force_name):
  11         self.columns = { 'admin_level' : 100}
  12         self.force_name = force_name
  13         self.context = context
  14         self.geometry = None
  15
  16     def add(self, key, value):
  17         if hasattr(self, 'set_key_' + key):
  18             getattr(self, 'set_key_' + key)(value)
  19         elif key.startswith('name+'):
  20             self.add_hstore('name', key[5:], value)
  21         elif key.startswith('extra+'):
  22             self.add_hstore('extratags', key[6:], value)
  23         else:
  24             assert_in(key, ('class', 'type', 'street', 'addr_place',
  25                             'isin', 'postcode'))
  26             self.columns[key] = value
  27
  28     def set_key_name(self, value):
  29         self.add_hstore('name', 'name', value)
  30
  31     def set_key_osm(self, value):
  32         assert_in(value[0], 'NRW')
  33         ok_(value[1:].isdigit())
  34
  35         self.columns['osm_type'] = value[0]
  36         self.columns['osm_id'] = int(value[1:])
  37
  38     def set_key_admin(self, value):
  39         self.columns['admin_level'] = int(value)
  40
  41     def set_key_housenr(self, value):
  42         self.columns['housenumber'] = value
  43
  44     def set_key_country(self, value):
  45         self.columns['country_code'] = value
  46
  47     def set_key_geometry(self, value):
  48         self.geometry = self.context.osm.parse_geometry(value, self.context.scene)
  49         assert_is_not_none(self.geometry)
  50
  51     def add_hstore(self, column, key, value):
  52         if column in self.columns:
  53             self.columns[column][key] = value
  54         else:
  55             self.columns[column] = { key : value }
  56
  57     def db_insert(self, cursor):
  58         assert_in('osm_type', self.columns)
  59         if self.force_name and 'name' not in self.columns:
  60             self.add_hstore('name', 'name', ''.join(random.choice(string.printable)
  61                                            for _ in range(int(random.random()*30))))
  62
  63         if self.columns['osm_type'] == 'N' and self.geometry is None:
  64             self.geometry = "ST_SetSRID(ST_Point(%f, %f), 4326)" % (
  65                             random.random()*360 - 180, random.random()*180 - 90)
  66
  67         query = 'INSERT INTO place (%s, geometry) values(%s, %s)' % (
  68                      ','.join(self.columns.keys()),
  69                      ','.join(['%s' for x in range(len(self.columns))]),
  70                      self.geometry)
  71         cursor.execute(query, list(self.columns.values()))
  72
  73 class NominatimID:
  74     """ Splits a unique identifier for places into its components.
  75         As place_ids cannot be used for testing, we use a unique
  76         identifier instead that is of the form <osmtype><osmid>[:<class>].
  77     """
  78
  79     id_regex = re.compile(r"(?P<tp>[NRW])(?P<id>\d+)(?P<cls>:\w+)?")
  80
  81     def __init__(self, oid):
  82         self.typ = self.oid = self.cls = None
  83
  84         if oid is not None:
  85             m = self.id_regex.fullmatch(oid)
  86             assert_is_not_none(m, "ID '%s' not of form <osmtype><osmid>[:<class>]" % oid)
  87
  88             self.typ = m.group('tp')
  89             self.oid = m.group('id')
  90             self.cls = m.group('cls')
  91
  92     def table_select(self):
  93         """ Return where clause and parameter list to select the object
  94             from a Nominatim table.
  95         """
  96         where = 'osm_type = %s and osm_id = %s'
  97         params = [self.typ, self. oid]
  98
  99         if self.cls is not None:
 100             where += ' class = %s'
 101             params.append(self.cls)
 102
 103         return where, params
 104
 105     def get_place_id(self, cur):
 106         where, params = self.table_select()
 107         cur.execute("SELECT place_id FROM placex WHERE %s" % where, params)
 108         eq_(1, cur.rowcount, "Expected exactly 1 entry in placex found %s" % cur.rowcount)
 109
 110         return cur.fetchone()[0]
 111
 112
 113 def assert_db_column(row, column, value):
 114     if column == 'object':
 115         return
 116
 117     if column.startswith('centroid'):
 118         fac = float(column[9:]) if h.startswith('centroid*') else 1.0
 119         x, y = value.split(' ')
 120         assert_almost_equal(float(x) * fac, row['cx'])
 121         assert_almost_equal(float(y) * fac, row['cy'])
 122     else:
 123         eq_(value, str(row[column]),
 124             "Row '%s': expected: %s, got: %s"
 125             % (column, value, str(row[column])))
 126
 127
 128 ################################ STEPS ##################################
 129
 130 @given(u'the scene (?P<scene>.+)')
 131 def set_default_scene(context, scene):
 132     context.scene = scene
 133
 134 @given("the (?P<named>named )?places")
 135 def add_data_to_place_table(context, named):
 136     cur = context.db.cursor()
 137     cur.execute('ALTER TABLE place DISABLE TRIGGER place_before_insert')
 138     for r in context.table:
 139         col = PlaceColumn(context, named is not None)
 140
 141         for h in r.headings:
 142             col.add(h, r[h])
 143
 144         col.db_insert(cur)
 145     cur.execute('ALTER TABLE place ENABLE TRIGGER place_before_insert')
 146     cur.close()
 147     context.db.commit()
 148
 149
 150 @when("importing")
 151 def import_and_index_data_from_place_table(context):
 152     context.nominatim.run_setup_script('create-functions', 'create-partition-functions')
 153     cur = context.db.cursor()
 154     cur.execute(
 155         """insert into placex (osm_type, osm_id, class, type, name, admin_level,
 156            housenumber, street, addr_place, isin, postcode, country_code, extratags,
 157            geometry)
 158            select * from place where not (class='place' and type='houses' and osm_type='W')""")
 159     cur.execute(
 160         """select insert_osmline (osm_id, housenumber, street, addr_place,
 161            postcode, country_code, geometry)
 162            from place where class='place' and type='houses' and osm_type='W'""")
 163     context.db.commit()
 164     context.nominatim.run_setup_script('index', 'index-noanalyse')
 165
 166
 167
 168 @then("placex contains(?P<exact> exactly)?")
 169 def check_placex_contents(context, exact):
 170     cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
 171
 172     expected_content = set()
 173     for row in context.table:
 174         nid = NominatimID(row['object'])
 175         where, params = nid.table_select()
 176         cur.execute("""SELECT *, ST_AsText(geometry) as geomtxt,
 177                        ST_X(centroid) as cx, ST_Y(centroid) as cy
 178                        FROM placex where %s""" % where,
 179                     params)
 180
 181         for res in cur:
 182             if exact:
 183                 expected_content.add((res['osm_type'], res['osm_id'], res['class']))
 184             for h in row.headings:
 185                 if h.startswith('name'):
 186                     name = h[5:] if h.startswith('name+') else 'name'
 187                     assert_in(name, res['name'])
 188                     eq_(res['name'][name], row[h])
 189                 elif h.startswith('extratags+'):
 190                     eq_(res['extratags'][h[10:]], row[h])
 191                 else:
 192                     assert_db_column(res, h, row[h])
 193
 194     if exact:
 195         cur.execute('SELECT osm_type, osm_id, class from placex')
 196         eq_(expected_content, set([(r[0], r[1], r[2]) for r in cur]))
 197
 198     context.db.commit()
 199
 200 @then("search_name contains")
 201 def check_search_name_contents(context):
 202     cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
 203
 204     for row in context.table:
 205         pid = NominatimID(row['object']).get_place_id(cur)
 206         cur.execute("""SELECT *, ST_X(centroid) as cx, ST_Y(centroid) as cy
 207                        FROM search_name WHERE place_id = %s""", (pid, ))
 208
 209         for res in cur:
 210             for h in row.headings:
 211                 if h in ('name_vector', 'nameaddress_vector'):
 212                     terms = [x.strip().replace('#', ' ') for x in row[h].split(',')]
 213                     subcur = context.db.cursor()
 214                     subcur.execute("""SELECT word_id, word_token
 215                                       FROM word, (SELECT unnest(%s) as term) t
 216                                       WHERE word_token = make_standard_name(t.term)""",
 217                                    (terms,))
 218                     ok_(subcur.rowcount >= len(terms))
 219                     for wid in subcur:
 220                         assert_in(wid[0], res[h],
 221                                   "Missing term for %s/%s: %s" % (pid, h, wid[1]))
 222                 else:
 223                     assert_db_column(res, h, row[h])
 224
 225
 226     context.db.commit()
 227
 228
 229 @then("placex has no entry for (?P<oid>.*)")
 230 def check_placex_has_entry(context, oid):
 231     cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
 232     nid = NominatimID(oid)
 233     where, params = nid.table_select()
 234     cur.execute("SELECT * FROM placex where %s" % where, params)
 235     eq_(0, cur.rowcount)
 236     context.db.commit()
 237
 238 @then("search_name has no entry for (?P<oid>.*)")
 239 def check_search_name_has_entry(context, oid):
 240     cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
 241     pid = NominatimID(oid).get_place_id(cur)
 242     cur.execute("SELECT * FROM search_name WHERE place_id = %s", (pid, ))
 243     eq_(0, cur.rowcount)
 244     context.db.commit()