test/bdd/steps/steps_api_queries.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   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 """ Steps that run queries against the API.
   8 """
   9 from pathlib import Path
  10 import re
  11 import logging
  12 import asyncio
  13 import xml.etree.ElementTree as ET
  14
  15 from http_responses import GenericResponse, SearchResponse, ReverseResponse, StatusResponse
  16 from check_functions import Bbox, check_for_attributes
  17 from table_compare import NominatimID
  18
  19 LOG = logging.getLogger(__name__)
  20
  21
  22 def make_todo_list(context, result_id):
  23     if result_id is None:
  24         context.execute_steps("then at least 1 result is returned")
  25         return range(len(context.response.result))
  26
  27     context.execute_steps(f"then more than {result_id}results are returned")
  28     return (int(result_id.strip()), )
  29
  30
  31 def compare(operator, op1, op2):
  32     if operator == 'less than':
  33         return op1 < op2
  34     elif operator == 'more than':
  35         return op1 > op2
  36     elif operator == 'exactly':
  37         return op1 == op2
  38     elif operator == 'at least':
  39         return op1 >= op2
  40     elif operator == 'at most':
  41         return op1 <= op2
  42     else:
  43         raise ValueError(f"Unknown operator '{operator}'")
  44
  45
  46 def send_api_query(endpoint, params, fmt, context):
  47     if fmt is not None:
  48         if fmt.strip() == 'debug':
  49             params['debug'] = '1'
  50         else:
  51             params['format'] = fmt.strip()
  52
  53     if context.table:
  54         if context.table.headings[0] == 'param':
  55             for line in context.table:
  56                 params[line['param']] = line['value']
  57         else:
  58             for h in context.table.headings:
  59                 params[h] = context.table[0][h]
  60
  61     return asyncio.run(context.nominatim.api_engine(endpoint, params,
  62                                                     Path(context.nominatim.website_dir.name),
  63                                                     context.nominatim.test_env,
  64                                                     getattr(context, 'http_headers', {})))
  65
  66
  67 @given('the HTTP header')
  68 def add_http_header(context):
  69     if not hasattr(context, 'http_headers'):
  70         context.http_headers = {}
  71
  72     for h in context.table.headings:
  73         context.http_headers[h] = context.table[0][h]
  74
  75
  76 @when(r'sending (?P<fmt>\S+ )?search query "(?P<query>.*)"(?P<addr> with address)?')
  77 def website_search_request(context, fmt, query, addr):
  78     params = {}
  79     if query:
  80         params['q'] = query
  81     if addr is not None:
  82         params['addressdetails'] = '1'
  83
  84     outp, status = send_api_query('search', params, fmt, context)
  85
  86     context.response = SearchResponse(outp, fmt or 'json', status)
  87
  88
  89 @when(r'sending v1/reverse at (?P<lat>[\d.-]*),(?P<lon>[\d.-]*)(?: with format (?P<fmt>.+))?')
  90 def api_endpoint_v1_reverse(context, lat, lon, fmt):
  91     params = {}
  92     if lat is not None:
  93         params['lat'] = lat
  94     if lon is not None:
  95         params['lon'] = lon
  96     if fmt is None:
  97         fmt = 'jsonv2'
  98     elif fmt == "''":
  99         fmt = None
 100
 101     outp, status = send_api_query('reverse', params, fmt, context)
 102     context.response = ReverseResponse(outp, fmt or 'xml', status)
 103
 104
 105 @when(r'sending v1/reverse N(?P<nodeid>\d+)(?: with format (?P<fmt>.+))?')
 106 def api_endpoint_v1_reverse_from_node(context, nodeid, fmt):
 107     params = {}
 108     params['lon'], params['lat'] = (f'{c:f}' for c in context.osm.grid_node(int(nodeid)))
 109
 110     outp, status = send_api_query('reverse', params, fmt, context)
 111     context.response = ReverseResponse(outp, fmt or 'xml', status)
 112
 113
 114 @when(r'sending (?P<fmt>\S+ )?details query for (?P<query>.*)')
 115 def website_details_request(context, fmt, query):
 116     params = {}
 117     if query[0] in 'NWR':
 118         nid = NominatimID(query)
 119         params['osmtype'] = nid.typ
 120         params['osmid'] = nid.oid
 121         if nid.cls:
 122             params['class'] = nid.cls
 123     else:
 124         params['place_id'] = query
 125     outp, status = send_api_query('details', params, fmt, context)
 126
 127     context.response = GenericResponse(outp, fmt or 'json', status)
 128
 129
 130 @when(r'sending (?P<fmt>\S+ )?lookup query for (?P<query>.*)')
 131 def website_lookup_request(context, fmt, query):
 132     params = {'osm_ids': query}
 133     outp, status = send_api_query('lookup', params, fmt, context)
 134
 135     context.response = SearchResponse(outp, fmt or 'xml', status)
 136
 137
 138 @when(r'sending (?P<fmt>\S+ )?status query')
 139 def website_status_request(context, fmt):
 140     params = {}
 141     outp, status = send_api_query('status', params, fmt, context)
 142
 143     context.response = StatusResponse(outp, fmt or 'text', status)
 144
 145
 146 @step(r'(?P<operator>less than|more than|exactly|at least|at most) '
 147       r'(?P<number>\d+) results? (?:is|are) returned')
 148 def validate_result_number(context, operator, number):
 149     context.execute_steps("Then a HTTP 200 is returned")
 150     numres = len(context.response.result)
 151     assert compare(operator, numres, int(number)), \
 152            f"Bad number of results: expected {operator} {number}, got {numres}."
 153
 154
 155 @then(r'a HTTP (?P<status>\d+) is returned')
 156 def check_http_return_status(context, status):
 157     assert context.response.errorcode == int(status), \
 158            f"Return HTTP status is {context.response.errorcode}."\
 159            f" Full response:\n{context.response.page}"
 160
 161
 162 @then(r'the page contents equals "(?P<text>.+)"')
 163 def check_page_content_equals(context, text):
 164     assert context.response.page == text
 165
 166
 167 @then(r'the result is valid (?P<fmt>\w+)')
 168 def step_impl(context, fmt):
 169     context.execute_steps("Then a HTTP 200 is returned")
 170     if fmt.strip() == 'html':
 171         try:
 172             tree = ET.fromstring(context.response.page)
 173         except Exception as ex:
 174             assert False, f"Could not parse page: {ex}\n{context.response.page}"
 175
 176         assert tree.tag == 'html'
 177         body = tree.find('./body')
 178         assert body is not None
 179         assert body.find('.//script') is None
 180     else:
 181         assert context.response.format == fmt
 182
 183
 184 @then(r'a (?P<fmt>\w+) user error is returned')
 185 def check_page_error(context, fmt):
 186     context.execute_steps("Then a HTTP 400 is returned")
 187     assert context.response.format == fmt
 188
 189     if fmt == 'xml':
 190         assert re.search(r'<error>.+</error>', context.response.page, re.DOTALL) is not None
 191     else:
 192         assert re.search(r'({"error":)', context.response.page, re.DOTALL) is not None
 193
 194
 195 @then('result header contains')
 196 def check_header_attr(context):
 197     context.execute_steps("Then a HTTP 200 is returned")
 198     for line in context.table:
 199         assert line['attr'] in context.response.header, \
 200                f"Field '{line['attr']}' missing in header. " \
 201                f"Full header:\n{context.response.header}"
 202         value = context.response.header[line['attr']]
 203         assert re.fullmatch(line['value'], value) is not None, \
 204                f"Attribute '{line['attr']}': expected: '{line['value']}', got '{value}'"
 205
 206
 207 @then('result header has (?P<neg>not )?attributes (?P<attrs>.*)')
 208 def check_header_no_attr(context, neg, attrs):
 209     check_for_attributes(context.response.header, attrs,
 210                          'absent' if neg else 'present')
 211
 212
 213 @then(r'results contain(?: in field (?P<field>.*))?')
 214 def results_contain_in_field(context, field):
 215     context.execute_steps("then at least 1 result is returned")
 216
 217     for line in context.table:
 218         context.response.match_row(line, context=context, field=field)
 219
 220
 221 @then(r'result (?P<lid>\d+ )?has (?P<neg>not )?attributes (?P<attrs>.*)')
 222 def validate_attributes(context, lid, neg, attrs):
 223     for i in make_todo_list(context, lid):
 224         check_for_attributes(context.response.result[i], attrs,
 225                              'absent' if neg else 'present')
 226
 227
 228 @then(u'result addresses contain')
 229 def result_addresses_contain(context):
 230     context.execute_steps("then at least 1 result is returned")
 231
 232     for line in context.table:
 233         idx = int(line['ID']) if 'ID' in line.headings else None
 234
 235         for name, value in zip(line.headings, line.cells):
 236             if name != 'ID':
 237                 context.response.assert_address_field(idx, name, value)
 238
 239
 240 @then(r'address of result (?P<lid>\d+) has(?P<neg> no)? types (?P<attrs>.*)')
 241 def check_address_has_types(context, lid, neg, attrs):
 242     context.execute_steps(f"then more than {lid} results are returned")
 243
 244     addr_parts = context.response.result[int(lid)]['address']
 245
 246     for attr in attrs.split(','):
 247         if neg:
 248             assert attr not in addr_parts
 249         else:
 250             assert attr in addr_parts
 251
 252
 253 @then(r'address of result (?P<lid>\d+) (?P<complete>is|contains)')
 254 def check_address(context, lid, complete):
 255     context.execute_steps(f"then more than {lid} results are returned")
 256
 257     lid = int(lid)
 258     addr_parts = dict(context.response.result[lid]['address'])
 259
 260     for line in context.table:
 261         context.response.assert_address_field(lid, line['type'], line['value'])
 262         del addr_parts[line['type']]
 263
 264     if complete == 'is':
 265         assert len(addr_parts) == 0, f"Additional address parts found: {addr_parts!s}"
 266
 267
 268 @then(r'result (?P<lid>\d+ )?has bounding box in (?P<coords>[\d,.-]+)')
 269 def check_bounding_box_in_area(context, lid, coords):
 270     expected = Bbox(coords)
 271
 272     for idx in make_todo_list(context, lid):
 273         res = context.response.result[idx]
 274         check_for_attributes(res, 'boundingbox')
 275         context.response.check_row(idx, res['boundingbox'] in expected,
 276                                    f"Bbox is not contained in {expected}")
 277
 278
 279 @then(r'result (?P<lid>\d+ )?has centroid in (?P<coords>[\d,.-]+)')
 280 def check_centroid_in_area(context, lid, coords):
 281     expected = Bbox(coords)
 282
 283     for idx in make_todo_list(context, lid):
 284         res = context.response.result[idx]
 285         check_for_attributes(res, 'lat,lon')
 286         context.response.check_row(idx, (res['lon'], res['lat']) in expected,
 287                                    f"Centroid is not inside {expected}")
 288
 289
 290 @then('there are(?P<neg> no)? duplicates')
 291 def check_for_duplicates(context, neg):
 292     context.execute_steps("then at least 1 result is returned")
 293
 294     resarr = set()
 295     has_dupe = False
 296
 297     for res in context.response.result:
 298         dup = (res['osm_type'], res['class'], res['type'], res['display_name'])
 299         if dup in resarr:
 300             has_dupe = True
 301             break
 302         resarr.add(dup)
 303
 304     if neg:
 305         assert not has_dupe, f"Found duplicate for {dup}"
 306     else:
 307         assert has_dupe, "No duplicates found"