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