test/bdd/steps/queries.py

   1 """ Steps that run search queries.
   2
   3     Queries may either be run directly via PHP using the query script
   4     or via the HTTP interface.
   5 """
   6
   7 import json
   8 import os
   9 import subprocess
  10 from collections import OrderedDict
  11 from nose.tools import * # for assert functions
  12
  13 class SearchResponse(object):
  14
  15     def __init__(self, page, fmt='json', errorcode=200):
  16         self.page = page
  17         self.format = fmt
  18         self.errorcode = errorcode
  19         getattr(self, 'parse_' + fmt)()
  20
  21     def parse_json(self):
  22         self.result = json.JSONDecoder(object_pairs_hook=OrderedDict).decode(self.page)
  23
  24     def match_row(self, row):
  25         if 'ID' in row.headings:
  26             todo = [int(row['ID'])]
  27         else:
  28             todo = range(len(self.result))
  29
  30         for i in todo:
  31             res = self.result[i]
  32             for h in row.headings:
  33                 if h == 'ID':
  34                     pass
  35                 elif h == 'osm':
  36                     assert_equal(res['osm_type'], row[h][0])
  37                     assert_equal(res['osm_id'], row[h][1:])
  38                 elif h == 'centroid':
  39                     x, y = row[h].split(' ')
  40                     assert_almost_equal(float(y), float(res['lat']))
  41                     assert_almost_equal(float(x), float(res['lon']))
  42                 else:
  43                     assert_in(h, res)
  44                     assert_equal(str(res[h]), str(row[h]))
  45
  46
  47 @when(u'searching for "(?P<query>.*)"(?P<dups> with dups)?')
  48 def query_cmd(context, query, dups):
  49     """ Query directly via PHP script.
  50     """
  51     cmd = [os.path.join(context.nominatim.build_dir, 'utils', 'query.php'),
  52            '--search', query]
  53     # add more parameters in table form
  54     if context.table:
  55         for h in context.table.headings:
  56             value = context.table[0][h].strip()
  57             if value:
  58                 cmd.extend(('--' + h, value))
  59
  60     if dups:
  61         cmd.extend(('--dedupe', '0'))
  62
  63     proc = subprocess.Popen(cmd, cwd=context.nominatim.build_dir,
  64                             stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  65     (outp, err) = proc.communicate()
  66
  67     assert_equals (0, proc.returncode, "query.php failed with message: %s" % err)
  68
  69     context.response = SearchResponse(outp.decode('utf-8'), 'json')