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