test/bdd/conftest.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 """
   8 Fixtures for BDD test steps
   9 """
  10 import sys
  11 import json
  12 from pathlib import Path
  13
  14 # always test against the source
  15 SRC_DIR = (Path(__file__) / '..' / '..' / '..').resolve()
  16 sys.path.insert(0, str(SRC_DIR / 'src'))
  17
  18 import pytest
  19 from pytest_bdd.parsers import re as step_parse
  20 from pytest_bdd import given, when, then
  21
  22 pytest.register_assert_rewrite('utils')
  23
  24 from utils.api_runner import APIRunner
  25 from utils.api_result import APIResult
  26 from utils.checks import ResultAttr, COMPARATOR_TERMS
  27 from utils.geometry_alias import ALIASES
  28 from utils.grid import Grid
  29 from utils.db import DBManager
  30
  31 from nominatim_db.config import Configuration
  32
  33 def _strlist(inp):
  34     return [s.strip() for s in inp.split(',')]
  35
  36
  37 def _pretty_json(inp):
  38     return json.dumps(inp, indent=2)
  39
  40
  41 def pytest_addoption(parser, pluginmanager):
  42     parser.addoption('--nominatim-purge', dest='NOMINATIM_PURGE', action='store_true',
  43                      help='Force recreation of test databases from scratch.')
  44     parser.addoption('--nominatim-keep-db', dest='NOMINATIM_KEEP_DB', action='store_true',
  45                      help='Do not drop the database after tests are finished.')
  46     parser.addoption('--nominatim-api-engine', dest='NOMINATIM_API_ENGINE',
  47                      default='falcon',
  48                      help='Chose the API engine to use when sending requests.')
  49     parser.addoption('--nominatim-tokenizer', dest='NOMINATIM_TOKENIZER',
  50                      metavar='TOKENIZER',
  51                      help='Use the specified tokenizer for importing data into '
  52                           'a Nominatim database.')
  53
  54     parser.addini('nominatim_test_db', default='test_nominatim',
  55                   help='Name of the database used for running a single test.')
  56     parser.addini('nominatim_api_test_db', default='test_api_nominatim',
  57                   help='Name of the database for storing API test data.')
  58     parser.addini('nominatim_template_db', default='test_template_nominatim',
  59                   help='Name of database used as a template for test databases.')
  60
  61
  62 @pytest.fixture
  63 def datatable():
  64     """ Default fixture for datatables, so that their presence can be optional.
  65     """
  66     return None
  67
  68
  69 @pytest.fixture
  70 def node_grid():
  71     """ Default fixture for node grids. Nothing set.
  72     """
  73     return Grid([[]], None, None)
  74
  75
  76 @pytest.fixture(scope='session')
  77 def template_db(pytestconfig):
  78     """ Create a template database containing the extensions and base data
  79         needed by Nominatim. Using the template instead of doing the full
  80         setup can speed up the tests.
  81
  82         The template database will only be created if it does not exist yet
  83         or a purge has been explicitly requested.
  84     """
  85     dbm = DBManager(purge=pytestconfig.option.NOMINATIM_PURGE)
  86
  87     template_db = pytestconfig.getini('nominatim_template_db')
  88
  89     template_config = Configuration(
  90         None, environ={'NOMINATIM_DATABASE_DSN': f"pgsql:dbname={template_db}"})
  91
  92     dbm.setup_template_db(template_config)
  93
  94     return template_db
  95
  96
  97
  98 @when(step_parse(r'reverse geocoding (?P<lat>[\d.-]*),(?P<lon>[\d.-]*)'),
  99       target_fixture='nominatim_result')
 100 def reverse_geocode_via_api(test_config_env, pytestconfig, datatable, lat, lon):
 101     runner = APIRunner(test_config_env, pytestconfig.option.NOMINATIM_API_ENGINE)
 102     api_response = runner.run_step('reverse',
 103                                    {'lat': float(lat), 'lon': float(lon)},
 104                                    datatable, 'jsonv2', {})
 105
 106     assert api_response.status == 200
 107     assert api_response.headers['content-type'] == 'application/json; charset=utf-8'
 108
 109     result = APIResult('json', 'reverse', api_response.body)
 110     assert result.is_simple()
 111
 112     return result
 113
 114
 115 @when(step_parse(r'geocoding(?: "(?P<query>.*)")?'),
 116       target_fixture='nominatim_result')
 117 def forward_geocode_via_api(test_config_env, pytestconfig, datatable, query):
 118     runner = APIRunner(test_config_env, pytestconfig.option.NOMINATIM_API_ENGINE)
 119
 120     params = {'addressdetails': '1'}
 121     if query:
 122         params['q'] = query
 123
 124     api_response = runner.run_step('search', params, datatable, 'jsonv2', {})
 125
 126     assert api_response.status == 200
 127     assert api_response.headers['content-type'] == 'application/json; charset=utf-8'
 128
 129     result = APIResult('json', 'search', api_response.body)
 130     assert not result.is_simple()
 131
 132     return result
 133
 134
 135 @then(step_parse(r'(?P<op>[a-z ]+) (?P<num>\d+) results? (?:are|is) returned'),
 136       converters={'num': int})
 137 def check_number_of_results(nominatim_result, op, num):
 138     assert not nominatim_result.is_simple()
 139     assert COMPARATOR_TERMS[op](num, len(nominatim_result))
 140
 141
 142 @then(step_parse('the result metadata contains'))
 143 def check_metadata_for_fields(nominatim_result, datatable):
 144     if datatable[0] == ['param', 'value']:
 145         pairs = datatable[1:]
 146     else:
 147         pairs = zip(datatable[0], datatable[1])
 148
 149     for k, v in pairs:
 150         assert ResultAttr(nominatim_result.meta, k) == v
 151
 152
 153 @then(step_parse('the result metadata has no attributes (?P<attributes>.*)'),
 154       converters={'attributes': _strlist})
 155 def check_metadata_for_field_presence(nominatim_result, attributes):
 156     assert all(a not in nominatim_result.meta for a in attributes), \
 157         f"Unexpectedly have one of the attributes '{attributes}' in\n" \
 158         f"{_pretty_json(nominatim_result.meta)}"
 159
 160
 161 @then(step_parse(r'the result contains(?: in field (?P<field>\S+))?'))
 162 def check_result_for_fields(nominatim_result, datatable, field):
 163     assert nominatim_result.is_simple()
 164
 165     if datatable[0] == ['param', 'value']:
 166         pairs = datatable[1:]
 167     else:
 168         pairs = zip(datatable[0], datatable[1])
 169
 170     prefix = field + '+' if field else ''
 171
 172     for k, v in pairs:
 173         assert ResultAttr(nominatim_result.result, prefix + k) == v
 174
 175
 176 @then(step_parse('the result has attributes (?P<attributes>.*)'),
 177       converters={'attributes': _strlist})
 178 def check_result_for_field_presence(nominatim_result, attributes):
 179     assert nominatim_result.is_simple()
 180     assert all(a in nominatim_result.result for a in attributes)
 181
 182
 183 @then(step_parse('the result has no attributes (?P<attributes>.*)'),
 184       converters={'attributes': _strlist})
 185 def check_result_for_field_absence(nominatim_result, attributes):
 186     assert nominatim_result.is_simple()
 187     assert all(a not in nominatim_result.result for a in attributes)
 188
 189
 190 @then(step_parse('the result set contains(?P<exact> exactly)?'))
 191 def check_result_list_match(nominatim_result, datatable, exact):
 192     assert not nominatim_result.is_simple()
 193
 194     result_set = set(range(len(nominatim_result.result)))
 195
 196     for row in datatable[1:]:
 197         for idx in result_set:
 198             for key, value in zip(datatable[0], row):
 199                 if ResultAttr(nominatim_result.result[idx], key) != value:
 200                     break
 201             else:
 202                 # found a match
 203                 result_set.remove(idx)
 204                 break
 205         else:
 206             assert False, f"Missing data row {row}. Full response:\n{nominatim_result}"
 207
 208     if exact:
 209         assert not [nominatim_result.result[i] for i in result_set]
 210
 211
 212 @then(step_parse('all results have attributes (?P<attributes>.*)'),
 213       converters={'attributes': _strlist})
 214 def check_all_results_for_field_presence(nominatim_result, attributes):
 215     assert not nominatim_result.is_simple()
 216     for res in nominatim_result.result:
 217         assert all(a in res for a in attributes), \
 218             f"Missing one of the attributes '{attributes}' in\n{_pretty_json(res)}"
 219
 220
 221 @then(step_parse('all results have no attributes (?P<attributes>.*)'),
 222       converters={'attributes': _strlist})
 223 def check_all_result_for_field_absence(nominatim_result, attributes):
 224     assert not nominatim_result.is_simple()
 225     for res in nominatim_result.result:
 226         assert all(a not in res for a in attributes), \
 227             f"Unexpectedly have one of the attributes '{attributes}' in\n{_pretty_json(res)}"
 228
 229
 230 @then(step_parse(r'all results contain(?: in field (?P<field>\S+))?'))
 231 def check_all_results_contain(nominatim_result, datatable, field):
 232     assert not nominatim_result.is_simple()
 233
 234     if datatable[0] == ['param', 'value']:
 235         pairs = datatable[1:]
 236     else:
 237         pairs = zip(datatable[0], datatable[1])
 238
 239     prefix = field + '+' if field else ''
 240
 241     for k, v in pairs:
 242         for r in nominatim_result.result:
 243             assert ResultAttr(r, prefix + k) == v
 244
 245
 246 @then(step_parse(r'result (?P<num>\d+) contains(?: in field (?P<field>\S+))?'),
 247       converters={'num': int})
 248 def check_specific_result_for_fields(nominatim_result, datatable, num, field):
 249     assert not nominatim_result.is_simple()
 250     assert len(nominatim_result) >= num + 1
 251
 252     if datatable[0] == ['param', 'value']:
 253         pairs = datatable[1:]
 254     else:
 255         pairs = zip(datatable[0], datatable[1])
 256
 257     prefix = field + '+' if field else ''
 258
 259     for k, v in pairs:
 260         assert ResultAttr(nominatim_result.result[num], prefix + k) == v
 261
 262
 263 @given(step_parse(r'the (?P<step>[0-9.]+ )?grid(?: with origin (?P<origin>.*))?'),
 264        target_fixture='node_grid')
 265 def set_node_grid(datatable, step, origin):
 266     if step is not None:
 267         step = float(step)
 268
 269     if origin:
 270         if ',' in origin:
 271             coords = origin.split(',')
 272             if len(coords) != 2:
 273                 raise RuntimeError('Grid origin expects origin with x,y coordinates.')
 274             origin = list(map(float, coords))
 275         elif origin in ALIASES:
 276             origin = ALIASES[origin]
 277         else:
 278             raise RuntimeError('Grid origin must be either coordinate or alias.')
 279
 280     return Grid(datatable, step, origin)