Merge remote-tracking branch 'upstream/master'

[nominatim.git] / test / bdd / steps / db_ops.py

diff --git a/test/bdd/steps/db_ops.py b/test/bdd/steps/db_ops.py
index b50f1d59baebb1c1556b0a60a8edae3f046bc31e..9604fd1089549946dfec258785d04afc7137412f 100644 (file)
--- a/test/bdd/steps/db_ops.py
+++ b/test/bdd/steps/db_ops.py
@@ -110,6 +110,9 @@ class PlaceObjName(object):
         if self.pid is None:
             return "<null>"
 
         if self.pid is None:
             return "<null>"
 

+        if self.pid == 0:
+            return "place ID 0"
+

         cur = self.conn.cursor()
         cur.execute("""SELECT osm_type, osm_id, class
                        FROM placex WHERE place_id = %s""",
         cur = self.conn.cursor()
         cur.execute("""SELECT osm_type, osm_id, class
                        FROM placex WHERE place_id = %s""",


@@ -197,10 +200,18 @@ def assert_db_column(row, column, value, context):
         return
 
     if column.startswith('centroid'):
         return
 
     if column.startswith('centroid'):

-        fac = float(column[9:]) if column.startswith('centroid*') else 1.0
-        x, y = value.split(' ')
-        assert_almost_equal(float(x) * fac, row['cx'], "Bad x coordinate")
-        assert_almost_equal(float(y) * fac, row['cy'], "Bad y coordinate")
+        if value == 'in geometry':
+            query = """SELECT ST_Within(ST_SetSRID(ST_Point({}, {}), 4326),
+                                        ST_SetSRID('{}'::geometry, 4326))""".format(
+                      row['cx'], row['cy'], row['geomtxt'])
+            cur = context.db.cursor()
+            cur.execute(query)
+            eq_(cur.fetchone()[0], True, "(Row %s failed: %s)" % (column, query))
+        else:
+            fac = float(column[9:]) if column.startswith('centroid*') else 1.0
+            x, y = value.split(' ')
+            assert_almost_equal(float(x) * fac, row['cx'], msg="Bad x coordinate")
+            assert_almost_equal(float(y) * fac, row['cy'], msg="Bad y coordinate")

     elif column == 'geometry':
         geom = context.osm.parse_geometry(value, context.scene)
         cur = context.db.cursor()
     elif column == 'geometry':
         geom = context.osm.parse_geometry(value, context.scene)
         cur = context.db.cursor()


@@ -329,6 +340,10 @@ def update_place_table(context):
 
     check_database_integrity(context)
 
 
     check_database_integrity(context)
 

+@when("updating postcodes")
+def update_postcodes(context):
+    context.nominatim.run_update_script('calculate-postcodes')
+

 @when("marking for delete (?P<oids>.*)")
 def delete_places(context, oids):
     context.nominatim.run_setup_script(
 @when("marking for delete (?P<oids>.*)")
 def delete_places(context, oids):
     context.nominatim.run_setup_script(


@@ -477,6 +492,43 @@ def check_search_name_contents(context, exclude):
 
     context.db.commit()
 
 
     context.db.commit()
 

+@then("location_postcode contains exactly")
+def check_location_postcode(context):
+    cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)
+
+    cur.execute("SELECT *, ST_AsText(geometry) as geomtxt FROM location_postcode")
+    eq_(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)
+
+@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:
+            eq_(0, cur.rowcount,
+                "Row still in word table: %s" % '/'.join(values))
+        else:
+            assert_greater(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)
 @then("place_addressline contains")
 def check_place_addressline(context):
     cur = context.db.cursor(cursor_factory=psycopg2.extras.DictCursor)