test/bdd/steps/steps_api_queries.py

   1 """ Steps that run queries against the API.
   2
   3     Queries may either be run directly via PHP using the query script
   4     or via the HTTP interface using php-cgi.
   5 """
   6 import json
   7 import os
   8 import re
   9 import logging
  10 from urllib.parse import urlencode
  11
  12 from utils import run_script
  13 from http_responses import GenericResponse, SearchResponse, ReverseResponse, StatusResponse
  14
  15 LOG = logging.getLogger(__name__)
  16
  17 BASE_SERVER_ENV = {
  18     'HTTP_HOST' : 'localhost',
  19     'HTTP_USER_AGENT' : 'Mozilla/5.0 (X11; Linux x86_64; rv:51.0) Gecko/20100101 Firefox/51.0',
  20     'HTTP_ACCEPT' : 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
  21     'HTTP_ACCEPT_ENCODING' : 'gzip, deflate',
  22     'HTTP_CONNECTION' : 'keep-alive',
  23     'SERVER_SIGNATURE' : '<address>Nominatim BDD Tests</address>',
  24     'SERVER_SOFTWARE' : 'Nominatim test',
  25     'SERVER_NAME' : 'localhost',
  26     'SERVER_ADDR' : '127.0.1.1',
  27     'SERVER_PORT' : '80',
  28     'REMOTE_ADDR' : '127.0.0.1',
  29     'DOCUMENT_ROOT' : '/var/www',
  30     'REQUEST_SCHEME' : 'http',
  31     'CONTEXT_PREFIX' : '/',
  32     'SERVER_ADMIN' : 'webmaster@localhost',
  33     'REMOTE_PORT' : '49319',
  34     'GATEWAY_INTERFACE' : 'CGI/1.1',
  35     'SERVER_PROTOCOL' : 'HTTP/1.1',
  36     'REQUEST_METHOD' : 'GET',
  37     'REDIRECT_STATUS' : 'CGI'
  38 }
  39
  40
  41 def compare(operator, op1, op2):
  42     if operator == 'less than':
  43         return op1 < op2
  44     elif operator == 'more than':
  45         return op1 > op2
  46     elif operator == 'exactly':
  47         return op1 == op2
  48     elif operator == 'at least':
  49         return op1 >= op2
  50     elif operator == 'at most':
  51         return op1 <= op2
  52     else:
  53         raise Exception("unknown operator '%s'" % operator)
  54
  55
  56 @when(u'searching for "(?P<query>.*)"(?P<dups> with dups)?')
  57 def query_cmd(context, query, dups):
  58     """ Query directly via PHP script.
  59     """
  60     cmd = ['/usr/bin/env', 'php']
  61     cmd.append(os.path.join(context.nominatim.build_dir, 'utils', 'query.php'))
  62     if query:
  63         cmd.extend(['--search', query])
  64     # add more parameters in table form
  65     if context.table:
  66         for h in context.table.headings:
  67             value = context.table[0][h].strip()
  68             if value:
  69                 cmd.extend(('--' + h, value))
  70
  71     if dups:
  72         cmd.extend(('--dedupe', '0'))
  73
  74     outp, err = run_script(cmd, cwd=context.nominatim.build_dir)
  75
  76     context.response = SearchResponse(outp, 'json')
  77
  78 def send_api_query(endpoint, params, fmt, context):
  79     if fmt is not None:
  80         params['format'] = fmt.strip()
  81     if context.table:
  82         if context.table.headings[0] == 'param':
  83             for line in context.table:
  84                 params[line['param']] = line['value']
  85         else:
  86             for h in context.table.headings:
  87                 params[h] = context.table[0][h]
  88
  89     env = dict(BASE_SERVER_ENV)
  90     env['QUERY_STRING'] = urlencode(params)
  91
  92     env['SCRIPT_NAME'] = '/%s.php' % endpoint
  93     env['REQUEST_URI'] = '%s?%s' % (env['SCRIPT_NAME'], env['QUERY_STRING'])
  94     env['CONTEXT_DOCUMENT_ROOT'] = os.path.join(context.nominatim.website_dir.name, 'website')
  95     env['SCRIPT_FILENAME'] = os.path.join(env['CONTEXT_DOCUMENT_ROOT'],
  96                                           '%s.php' % endpoint)
  97
  98     LOG.debug("Environment:" + json.dumps(env, sort_keys=True, indent=2))
  99
 100     if hasattr(context, 'http_headers'):
 101         env.update(context.http_headers)
 102
 103     cmd = ['/usr/bin/env', 'php-cgi', '-f']
 104     if context.nominatim.code_coverage_path:
 105         env['COV_SCRIPT_FILENAME'] = env['SCRIPT_FILENAME']
 106         env['COV_PHP_DIR'] = os.path.join(context.nominatim.src_dir, "lib")
 107         env['COV_TEST_NAME'] = '%s:%s' % (context.scenario.filename, context.scenario.line)
 108         env['SCRIPT_FILENAME'] = \
 109                 os.path.join(os.path.split(__file__)[0], 'cgi-with-coverage.php')
 110         cmd.append(env['SCRIPT_FILENAME'])
 111         env['PHP_CODE_COVERAGE_FILE'] = context.nominatim.next_code_coverage_file()
 112     else:
 113         cmd.append(env['SCRIPT_FILENAME'])
 114
 115     for k,v in params.items():
 116         cmd.append("%s=%s" % (k, v))
 117
 118     outp, err = run_script(cmd, cwd=context.nominatim.website_dir.name, env=env)
 119
 120     assert len(err) == 0, "Unexpected PHP error: %s" % (err)
 121
 122     if outp.startswith('Status: '):
 123         status = int(outp[8:11])
 124     else:
 125         status = 200
 126
 127     content_start = outp.find('\r\n\r\n')
 128
 129     return outp[content_start + 4:], status
 130
 131 @given(u'the HTTP header')
 132 def add_http_header(context):
 133     if not hasattr(context, 'http_headers'):
 134         context.http_headers = {}
 135
 136     for h in context.table.headings:
 137         envvar = 'HTTP_' + h.upper().replace('-', '_')
 138         context.http_headers[envvar] = context.table[0][h]
 139
 140
 141 @when(u'sending (?P<fmt>\S+ )?search query "(?P<query>.*)"(?P<addr> with address)?')
 142 def website_search_request(context, fmt, query, addr):
 143     params = {}
 144     if query:
 145         params['q'] = query
 146     if addr is not None:
 147         params['addressdetails'] = '1'
 148
 149     outp, status = send_api_query('search', params, fmt, context)
 150
 151     if fmt is None or fmt == 'jsonv2 ':
 152         outfmt = 'json'
 153     else:
 154         outfmt = fmt.strip()
 155
 156     context.response = SearchResponse(outp, outfmt, status)
 157
 158 @when(u'sending (?P<fmt>\S+ )?reverse coordinates (?P<lat>.+)?,(?P<lon>.+)?')
 159 def website_reverse_request(context, fmt, lat, lon):
 160     params = {}
 161     if lat is not None:
 162         params['lat'] = lat
 163     if lon is not None:
 164         params['lon'] = lon
 165
 166     outp, status = send_api_query('reverse', params, fmt, context)
 167
 168     if fmt is None:
 169         outfmt = 'xml'
 170     elif fmt == 'jsonv2 ':
 171         outfmt = 'json'
 172     else:
 173         outfmt = fmt.strip()
 174
 175     context.response = ReverseResponse(outp, outfmt, status)
 176
 177 @when(u'sending (?P<fmt>\S+ )?details query for (?P<query>.*)')
 178 def website_details_request(context, fmt, query):
 179     params = {}
 180     if query[0] in 'NWR':
 181         params['osmtype'] = query[0]
 182         params['osmid'] = query[1:]
 183     else:
 184         params['place_id'] = query
 185     outp, status = send_api_query('details', params, fmt, context)
 186
 187     if fmt is None:
 188         outfmt = 'json'
 189     else:
 190         outfmt = fmt.strip()
 191
 192     context.response = GenericResponse(outp, outfmt, status)
 193
 194 @when(u'sending (?P<fmt>\S+ )?lookup query for (?P<query>.*)')
 195 def website_lookup_request(context, fmt, query):
 196     params = { 'osm_ids' : query }
 197     outp, status = send_api_query('lookup', params, fmt, context)
 198
 199     if fmt == 'json ':
 200         outfmt = 'json'
 201     elif fmt == 'jsonv2 ':
 202         outfmt = 'json'
 203     elif fmt == 'geojson ':
 204         outfmt = 'geojson'
 205     elif fmt == 'geocodejson ':
 206         outfmt = 'geocodejson'
 207     else:
 208         outfmt = 'xml'
 209
 210     context.response = SearchResponse(outp, outfmt, status)
 211
 212 @when(u'sending (?P<fmt>\S+ )?status query')
 213 def website_status_request(context, fmt):
 214     params = {}
 215     outp, status = send_api_query('status', params, fmt, context)
 216
 217     if fmt is None:
 218         outfmt = 'text'
 219     else:
 220         outfmt = fmt.strip()
 221
 222     context.response = StatusResponse(outp, outfmt, status)
 223
 224 @step(u'(?P<operator>less than|more than|exactly|at least|at most) (?P<number>\d+) results? (?:is|are) returned')
 225 def validate_result_number(context, operator, number):
 226     assert context.response.errorcode == 200
 227     numres = len(context.response.result)
 228     assert compare(operator, numres, int(number)), \
 229         "Bad number of results: expected %s %s, got %d." % (operator, number, numres)
 230
 231 @then(u'a HTTP (?P<status>\d+) is returned')
 232 def check_http_return_status(context, status):
 233     assert context.response.errorcode == int(status)
 234
 235 @then(u'the page contents equals "(?P<text>.+)"')
 236 def check_page_content_equals(context, text):
 237     assert context.response.page == text
 238
 239 @then(u'the result is valid (?P<fmt>\w+)')
 240 def step_impl(context, fmt):
 241     context.execute_steps("Then a HTTP 200 is returned")
 242     assert context.response.format == fmt
 243
 244 @then(u'a (?P<fmt>\w+) user error is returned')
 245 def check_page_error(context, fmt):
 246     context.execute_steps("Then a HTTP 400 is returned")
 247     assert context.response.format == fmt
 248
 249     if fmt == 'xml':
 250         assert re.search(r'<error>.+</error>', context.response.page, re.DOTALL) is not None
 251     else:
 252         assert re.search(r'({"error":)', context.response.page, re.DOTALL) is not None
 253
 254 @then(u'result header contains')
 255 def check_header_attr(context):
 256     for line in context.table:
 257         assert re.fullmatch(line['value'], context.response.header[line['attr']]) is not None, \
 258                "attribute '%s': expected: '%s', got '%s'" % (
 259                     line['attr'], line['value'],
 260                     context.response.header[line['attr']])
 261
 262 @then(u'result header has (?P<neg>not )?attributes (?P<attrs>.*)')
 263 def check_header_no_attr(context, neg, attrs):
 264     for attr in attrs.split(','):
 265         if neg:
 266             assert attr not in context.response.header
 267         else:
 268             assert attr in context.response.header
 269
 270 @then(u'results contain')
 271 def step_impl(context):
 272     context.execute_steps("then at least 1 result is returned")
 273
 274     for line in context.table:
 275         context.response.match_row(line)
 276
 277 @then(u'result (?P<lid>\d+ )?has (?P<neg>not )?attributes (?P<attrs>.*)')
 278 def validate_attributes(context, lid, neg, attrs):
 279     if lid is None:
 280         idx = range(len(context.response.result))
 281         context.execute_steps("then at least 1 result is returned")
 282     else:
 283         idx = [int(lid.strip())]
 284         context.execute_steps("then more than %sresults are returned" % lid)
 285
 286     for i in idx:
 287         for attr in attrs.split(','):
 288             if neg:
 289                 assert attr not in context.response.result[i]
 290             else:
 291                 assert attr in context.response.result[i]
 292
 293 @then(u'result addresses contain')
 294 def step_impl(context):
 295     context.execute_steps("then at least 1 result is returned")
 296
 297     if 'ID' not in context.table.headings:
 298         addr_parts = context.response.property_list('address')
 299
 300     for line in context.table:
 301         if 'ID' in context.table.headings:
 302             addr_parts = [dict(context.response.result[int(line['ID'])]['address'])]
 303
 304         for h in context.table.headings:
 305             if h != 'ID':
 306                 for p in addr_parts:
 307                     assert h in p
 308                     assert p[h] == line[h], "Bad address value for %s" % h
 309
 310 @then(u'address of result (?P<lid>\d+) has(?P<neg> no)? types (?P<attrs>.*)')
 311 def check_address(context, lid, neg, attrs):
 312     context.execute_steps("then more than %s results are returned" % lid)
 313
 314     addr_parts = context.response.result[int(lid)]['address']
 315
 316     for attr in attrs.split(','):
 317         if neg:
 318             assert attr not in addr_parts
 319         else:
 320             assert attr in addr_parts
 321
 322 @then(u'address of result (?P<lid>\d+) (?P<complete>is|contains)')
 323 def check_address(context, lid, complete):
 324     context.execute_steps("then more than %s results are returned" % lid)
 325
 326     addr_parts = dict(context.response.result[int(lid)]['address'])
 327
 328     for line in context.table:
 329         assert line['type'] in addr_parts
 330         assert addr_parts[line['type']] == line['value'], \
 331                      "Bad address value for %s" % line['type']
 332         del addr_parts[line['type']]
 333
 334     if complete == 'is':
 335         assert len(addr_parts) == 0, "Additional address parts found: %s" % str(addr_parts)
 336
 337 @then(u'result (?P<lid>\d+ )?has bounding box in (?P<coords>[\d,.-]+)')
 338 def step_impl(context, lid, coords):
 339     if lid is None:
 340         context.execute_steps("then at least 1 result is returned")
 341         bboxes = context.response.property_list('boundingbox')
 342     else:
 343         context.execute_steps("then more than %sresults are returned" % lid)
 344         bboxes = [ context.response.result[int(lid)]['boundingbox']]
 345
 346     coord = [ float(x) for x in coords.split(',') ]
 347
 348     for bbox in bboxes:
 349         if isinstance(bbox, str):
 350             bbox = bbox.split(',')
 351         bbox = [ float(x) for x in bbox ]
 352
 353         assert bbox[0] >= coord[0]
 354         assert bbox[1] <= coord[1]
 355         assert bbox[2] >= coord[2]
 356         assert bbox[3] <= coord[3]
 357
 358 @then(u'result (?P<lid>\d+ )?has centroid in (?P<coords>[\d,.-]+)')
 359 def step_impl(context, lid, coords):
 360     if lid is None:
 361         context.execute_steps("then at least 1 result is returned")
 362         bboxes = zip(context.response.property_list('lat'),
 363                      context.response.property_list('lon'))
 364     else:
 365         context.execute_steps("then more than %sresults are returned" % lid)
 366         res = context.response.result[int(lid)]
 367         bboxes = [ (res['lat'], res['lon']) ]
 368
 369     coord = [ float(x) for x in coords.split(',') ]
 370
 371     for lat, lon in bboxes:
 372         lat = float(lat)
 373         lon = float(lon)
 374         assert lat >= coord[0]
 375         assert lat <= coord[1]
 376         assert lon >= coord[2]
 377         assert lon <= coord[3]
 378
 379 @then(u'there are(?P<neg> no)? duplicates')
 380 def check_for_duplicates(context, neg):
 381     context.execute_steps("then at least 1 result is returned")
 382
 383     resarr = set()
 384     has_dupe = False
 385
 386     for res in context.response.result:
 387         dup = (res['osm_type'], res['class'], res['type'], res['display_name'])
 388         if dup in resarr:
 389             has_dupe = True
 390             break
 391         resarr.add(dup)
 392
 393     if neg:
 394         assert not has_dupe, "Found duplicate for %s" % (dup, )
 395     else:
 396         assert has_dupe, "No duplicates found"