]> git.openstreetmap.org Git - nominatim.git/blob - test/bdd/steps/table_compare.py
introduce external processing in indexer
[nominatim.git] / test / bdd / steps / table_compare.py
1 """
2 Functions to facilitate accessing and comparing the content of DB tables.
3 """
4 import re
5 import json
6
7 from steps.check_functions import Almost
8
9 ID_REGEX = re.compile(r"(?P<typ>[NRW])(?P<oid>\d+)(:(?P<cls>\w+))?")
10
11 class NominatimID:
12     """ Splits a unique identifier for places into its components.
13         As place_ids cannot be used for testing, we use a unique
14         identifier instead that is of the form <osmtype><osmid>[:<class>].
15     """
16
17     def __init__(self, oid):
18         self.typ = self.oid = self.cls = None
19
20         if oid is not None:
21             m = ID_REGEX.fullmatch(oid)
22             assert m is not None, \
23                    "ID '{}' not of form <osmtype><osmid>[:<class>]".format(oid)
24
25             self.typ = m.group('typ')
26             self.oid = m.group('oid')
27             self.cls = m.group('cls')
28
29     def __str__(self):
30         if self.cls is None:
31             return self.typ + self.oid
32
33         return '{self.typ}{self.oid}:{self.cls}'.format(self=self)
34
35     def query_osm_id(self, cur, query):
36         """ Run a query on cursor `cur` using osm ID, type and class. The
37             `query` string must contain exactly one placeholder '{}' where
38             the 'where' query should go.
39         """
40         where = 'osm_type = %s and osm_id = %s'
41         params = [self.typ, self. oid]
42
43         if self.cls is not None:
44             where += ' and class = %s'
45             params.append(self.cls)
46
47         cur.execute(query.format(where), params)
48
49     def row_by_place_id(self, cur, table, extra_columns=None):
50         """ Get a row by place_id from the given table using cursor `cur`.
51             extra_columns may contain a list additional elements for the select
52             part of the query.
53         """
54         pid = self.get_place_id(cur)
55         query = "SELECT {} FROM {} WHERE place_id = %s".format(
56                     ','.join(['*'] + (extra_columns or [])), table)
57         cur.execute(query, (pid, ))
58
59     def get_place_id(self, cur):
60         """ Look up the place id for the ID. Throws an assertion if the ID
61             is not unique.
62         """
63         self.query_osm_id(cur, "SELECT place_id FROM placex WHERE {}")
64         assert cur.rowcount == 1, \
65                "Place ID {!s} not unique. Found {} entries.".format(self, cur.rowcount)
66
67         return cur.fetchone()[0]
68
69
70 class DBRow:
71     """ Represents a row from a database and offers comparison functions.
72     """
73     def __init__(self, nid, db_row, context):
74         self.nid = nid
75         self.db_row = db_row
76         self.context = context
77
78     def assert_row(self, row, exclude_columns):
79         """ Check that all columns of the given behave row are contained
80             in the database row. Exclude behave rows with the names given
81             in the `exclude_columns` list.
82         """
83         for name, value in zip(row.headings, row.cells):
84             if name not in exclude_columns:
85                 assert self.contains(name, value), self.assert_msg(name, value)
86
87     def contains(self, name, expected):
88         """ Check that the DB row contains a column `name` with the given value.
89         """
90         if '+' in name:
91             column, field = name.split('+', 1)
92             return self._contains_hstore_value(column, field, expected)
93
94         if name == 'geometry':
95             return self._has_geometry(expected)
96
97         if name not in self.db_row:
98             return False
99
100         actual = self.db_row[name]
101
102         if expected == '-':
103             return actual is None
104
105         if name == 'name' and ':' not in expected:
106             return self._compare_column(actual[name], expected)
107
108         if 'place_id' in name:
109             return self._compare_place_id(actual, expected)
110
111         if name == 'centroid':
112             return self._has_centroid(expected)
113
114         return self._compare_column(actual, expected)
115
116     def _contains_hstore_value(self, column, field, expected):
117         if column == 'addr':
118             column = 'address'
119
120         if column not in self.db_row:
121             return False
122
123         if expected == '-':
124             return self.db_row[column] is None or field not in self.db_row[column]
125
126         if self.db_row[column] is None:
127             return False
128
129         return self._compare_column(self.db_row[column].get(field), expected)
130
131     def _compare_column(self, actual, expected):
132         if isinstance(actual, dict):
133             return actual == eval('{' + expected + '}')
134
135         return str(actual) == expected
136
137     def _compare_place_id(self, actual, expected):
138        if expected == '0':
139             return actual == 0
140
141        with self.context.db.cursor() as cur:
142             return NominatimID(expected).get_place_id(cur) == actual
143
144     def _has_centroid(self, expected):
145         if expected == 'in geometry':
146             with self.context.db.cursor() as cur:
147                 cur.execute("""SELECT ST_Within(ST_SetSRID(ST_Point({cx}, {cy}), 4326),
148                                         ST_SetSRID('{geomtxt}'::geometry, 4326))""".format(**self.db_row))
149                 return cur.fetchone()[0]
150
151         x, y = expected.split(' ')
152         return Almost(float(x)) == self.db_row['cx'] and Almost(float(y)) == self.db_row['cy']
153
154     def _has_geometry(self, expected):
155         geom = self.context.osm.parse_geometry(expected, self.context.scene)
156         with self.context.db.cursor() as cur:
157             cur.execute("""SELECT ST_Equals(ST_SnapToGrid({}, 0.00001, 0.00001),
158                                    ST_SnapToGrid(ST_SetSRID('{}'::geometry, 4326), 0.00001, 0.00001))""".format(
159                             geom, self.db_row['geomtxt']))
160             return cur.fetchone()[0]
161
162     def assert_msg(self, name, value):
163         """ Return a string with an informative message for a failed compare.
164         """
165         msg = "\nBad column '{}' in row '{!s}'.".format(name, self.nid)
166         actual = self._get_actual(name)
167         if actual is not None:
168             msg += " Expected: {}, got: {}.".format(value, actual)
169         else:
170             msg += " No such column."
171
172         return msg + "\nFull DB row: {}".format(json.dumps(dict(self.db_row), indent=4, default=str))
173
174     def _get_actual(self, name):
175         if '+' in name:
176             column, field = name.split('+', 1)
177             if column == 'addr':
178                 column = 'address'
179             return (self.db_row.get(column) or {}).get(field)
180
181         if name == 'geometry':
182             return self.db_row['geomtxt']
183
184         if name not in self.db_row:
185             return None
186
187         if name == 'centroid':
188             return "POINT({cx} {cy})".format(**self.db_row)
189
190         actual = self.db_row[name]
191
192         if 'place_id' in name:
193             if actual is None:
194                 return '<null>'
195
196             if actual == 0:
197                 return "place ID 0"
198
199             with self.context.db.cursor() as cur:
200                 cur.execute("""SELECT osm_type, osm_id, class
201                                FROM placex WHERE place_id = %s""",
202                             (actual, ))
203
204                 if cur.rowcount == 1:
205                     return "{0[0]}{0[1]}:{0[2]}".format(cur.fetchone())
206
207                 return "[place ID {} not found]".format(actual)
208
209         return actual