test/bdd/utils/checks.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2025 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Helper functions to compare expected values.
   9 """
  10 import json
  11 import re
  12 import math
  13 import itertools
  14
  15 from psycopg import sql as pysql
  16 from psycopg.rows import dict_row, tuple_row
  17 from .geometry_alias import ALIASES
  18
  19 COMPARATOR_TERMS = {
  20     'exactly': lambda exp, act: exp == act,
  21     'more than': lambda exp, act: act > exp,
  22     'less than': lambda exp, act: act < exp,
  23 }
  24
  25
  26 def _pretty(obj):
  27     return json.dumps(obj, sort_keys=True, indent=2)
  28
  29
  30 def within_box(value, expect):
  31     coord = [float(x) for x in expect.split(',')]
  32
  33     if isinstance(value, str):
  34         value = value.split(',')
  35     value = list(map(float, value))
  36
  37     if len(value) == 2:
  38         return coord[0] <= value[0] <= coord[2] \
  39                and coord[1] <= value[1] <= coord[3]
  40
  41     if len(value) == 4:
  42         return value[0] >= coord[0] and value[1] <= coord[1] \
  43                and value[2] >= coord[2] and value[3] <= coord[3]
  44
  45     raise ValueError("Not a coordinate or bbox.")
  46
  47
  48 COMPARISON_FUNCS = {
  49     None: lambda val, exp: str(val) == exp,
  50     'i': lambda val, exp: str(val).lower() == exp.lower(),
  51     'fm': lambda val, exp: re.fullmatch(exp, val) is not None,
  52     'dict': lambda val, exp: val is None if exp == '-' else (val == eval('{' + exp + '}')),
  53     'in_box': within_box
  54 }
  55
  56 OSM_TYPE = {'node': 'n', 'way': 'w', 'relation': 'r',
  57             'N': 'n', 'W': 'w', 'R': 'r'}
  58
  59
  60 class ResultAttr:
  61     """ Returns the given attribute as a string.
  62
  63         The key parameter determines how the value is formatted before
  64         returning. To refer to sub attributes, use '+' to add more keys
  65         (e.g. 'name+ref' will access obj['name']['ref']). A '!' introduces
  66         a formatting suffix. If no suffix is given, the value will be
  67         converted using the str() function.
  68
  69         Available formatters:
  70
  71         !:...   - use a formatting expression according to Python Mini Format Spec
  72         !i      - make case-insensitive comparison
  73         !fm     - consider comparison string a regular expression and match full value
  74         !wkt    - convert the expected value to a WKT string before comparing
  75         !in_box - the expected value is a comma-separated bbox description
  76     """
  77
  78     def __init__(self, obj, key, grid=None):
  79         self.grid = grid
  80         self.obj = obj
  81         if '!' in key:
  82             self.key, self.fmt = key.rsplit('!', 1)
  83         else:
  84             self.key = key
  85             self.fmt = None
  86
  87         if self.key == 'object':
  88             assert 'osm_id' in obj
  89             assert 'osm_type' in obj
  90             self.subobj = OSM_TYPE[obj['osm_type']] + str(obj['osm_id'])
  91             self.fmt = 'i'
  92         else:
  93             done = ''
  94             self.subobj = self.obj
  95             for sub in self.key.split('+'):
  96                 done += f"[{sub}]"
  97                 assert sub in self.subobj, \
  98                     f"Missing attribute {done}. Full object:\n{_pretty(self.obj)}"
  99                 self.subobj = self.subobj[sub]
 100
 101     def __eq__(self, other):
 102         if not isinstance(other, str):
 103             raise NotImplementedError()
 104
 105         # work around bad quoting by pytest-bdd
 106         other = other.replace(r'\\', '\\')
 107
 108         if self.fmt in COMPARISON_FUNCS:
 109             return COMPARISON_FUNCS[self.fmt](self.subobj, other)
 110
 111         if self.fmt.startswith(':'):
 112             return other == f"{{{self.fmt}}}".format(self.subobj)
 113
 114         if self.fmt == 'wkt':
 115             return self.compare_wkt(self.subobj, other)
 116
 117         raise RuntimeError(f"Unknown format string '{self.fmt}'.")
 118
 119     def __repr__(self):
 120         k = self.key.replace('+', '][')
 121         if self.fmt:
 122             k += '!' + self.fmt
 123         return f"result[{k}]({self.subobj})"
 124
 125     def compare_wkt(self, value, expected):
 126         """ Compare a WKT value against a compact geometry format.
 127             The function understands the following formats:
 128
 129               country:<country code>
 130                  Point geometry guaranteed to be in the given country
 131               <P>
 132                  Point geometry
 133               <P>,...,<P>
 134                  Line geometry
 135               (<P>,...,<P>)
 136                  Polygon geometry
 137
 138            <P> may either be a coordinate of the form '<x> <y>' or a single
 139            number. In the latter case it must refer to a point in
 140            a previously defined grid.
 141         """
 142         m = re.fullmatch(r'(POINT)\(([0-9. -]*)\)', value) \
 143             or re.fullmatch(r'(LINESTRING)\(([0-9,. -]*)\)', value) \
 144             or re.fullmatch(r'(POLYGON)\(\(([0-9,. -]*)\)\)', value)
 145         if not m:
 146             return False
 147
 148         converted = [list(map(float, pt.split(' ', 1)))
 149                      for pt in map(str.strip, m[2].split(','))]
 150
 151         if expected.startswith('country:'):
 152             ccode = geom[8:].upper()
 153             assert ccode in ALIASES, f"Geometry error: unknown country {ccode}"
 154             return m[1] == 'POINT' and \
 155                    all(math.isclose(p1, p2) for p1, p2 in
 156                        zip(converted[0], ALIASES[ccode]))
 157
 158         if ',' not in expected:
 159             return m[1] == 'POINT' and \
 160                    all(math.isclose(p1, p2) for p1, p2 in
 161                        zip(converted[0], self.get_point(expected)))
 162
 163         if '(' not in expected:
 164             return m[1] == 'LINESTRING' and \
 165                    all(math.isclose(p1[0], p2[0]) and math.isclose(p1[1], p2[1]) for p1, p2 in
 166                        zip(converted, (self.get_point(p) for p in expected.split(','))))
 167
 168         if m[1] != 'POLYGON':
 169             return False
 170
 171         # Polygon comparison is tricky because the polygons don't necessarily
 172         # end at the same point or have the same winding order.
 173         # Brute force all possible variants of the expected polygon
 174         exp_coords = [self.get_point(p) for p in expected[1:-1].split(',')]
 175         if exp_coords[0] != exp_coords[-1]:
 176             raise RuntimeError(f"Invalid polygon {expected}. "
 177                                "First and last point need to be the same")
 178         for line in (exp_coords[:-1], exp_coords[-1:0:-1]):
 179             for i in range(len(line)):
 180                 if all(math.isclose(p1[0], p2[0]) and math.isclose(p1[1], p2[1]) for p1, p2 in
 181                        zip(converted, line[i:] + line[:i])):
 182                     return True
 183
 184         return False
 185
 186     def get_point(self, pt):
 187         pt = pt.strip()
 188         if ' ' in pt:
 189             return list(map(float, pt.split(' ', 1)))
 190
 191         assert self.grid
 192
 193         return self.grid.get(pt)
 194
 195
 196 def check_table_content(conn, tablename, data, grid=None, exact=False):
 197     lines = set(range(1, len(data)))
 198
 199     cols = []
 200     for col in data[0]:
 201         if col == 'object':
 202             cols.extend(('osm_id', 'osm_type'))
 203         elif '!' in col:
 204             name, fmt = col.rsplit('!', 1)
 205             if fmt == 'wkt':
 206                 cols.append(f"ST_AsText({name}) as {name}")
 207             else:
 208                 cols.append(name.split('+')[0])
 209         else:
 210             cols.append(col.split('+')[0])
 211
 212     with conn.cursor(row_factory=dict_row) as cur:
 213         cur.execute(pysql.SQL(f"SELECT {','.join(cols)} FROM")
 214                     + pysql.Identifier(tablename))
 215
 216         table_content = ''
 217         for row in cur:
 218             table_content += '\n' + str(row)
 219             for i in lines:
 220                 for col, value in zip(data[0], data[i]):
 221                     if ResultAttr(row, col, grid=grid) != value:
 222                         break
 223                 else:
 224                     lines.remove(i)
 225                     break
 226             else:
 227                 assert not exact, f"Unexpected row in table {tablename}: {row}"
 228
 229         assert not lines, \
 230                "Rows not found:\n" \
 231                + '\n'.join(str(data[i]) for i in lines) \
 232                + "\nTable content:\n" \
 233                + table_content
 234
 235
 236 def check_table_has_lines(conn, tablename, osm_type, osm_id, osm_class):
 237     sql = pysql.SQL("""SELECT count(*) FROM {}
 238                        WHERE osm_type = %s and osm_id = %s""").format(pysql.Identifier(tablename))
 239     params = [osm_type, int(osm_id)]
 240     if osm_class:
 241         sql += pysql.SQL(' AND class = %s')
 242         params.append(osm_class)
 243
 244     with conn.cursor(row_factory=tuple_row) as cur:
 245         assert cur.execute(sql, params).fetchone()[0] == 0