test/bdd/steps/http_responses.py

   1 """
   2 Classes wrapping HTTP responses from the Nominatim API.
   3 """
   4 from collections import OrderedDict
   5 import re
   6 import json
   7 import xml.etree.ElementTree as ET
   8
   9 from check_functions import Almost
  10
  11 def _geojson_result_to_json_result(geojson_result):
  12     result = geojson_result['properties']
  13     result['geojson'] = geojson_result['geometry']
  14     if 'bbox' in geojson_result:
  15         # bbox is  minlon, minlat, maxlon, maxlat
  16         # boundingbox is minlat, maxlat, minlon, maxlon
  17         result['boundingbox'] = [geojson_result['bbox'][1],
  18                                  geojson_result['bbox'][3],
  19                                  geojson_result['bbox'][0],
  20                                  geojson_result['bbox'][2]]
  21     return result
  22
  23 class BadRowValueAssert:
  24     """ Lazily formatted message for failures to find a field content.
  25     """
  26
  27     def __init__(self, response, idx, field, value):
  28         self.idx = idx
  29         self.field = field
  30         self.value = value
  31         self.row = response.result[idx]
  32
  33     def __str__(self):
  34         return "\nBad value for row {} field '{}'. Expected: {}, got: {}.\nFull row: {}"""\
  35                    .format(self.idx, self.field, self.value,
  36                            self.row[self.field], json.dumps(self.row, indent=4))
  37
  38
  39 class GenericResponse:
  40     """ Common base class for all API responses.
  41     """
  42     def __init__(self, page, fmt, errorcode=200):
  43         fmt = fmt.strip()
  44         if fmt == 'jsonv2':
  45             fmt = 'json'
  46
  47         self.page = page
  48         self.format = fmt
  49         self.errorcode = errorcode
  50         self.result = []
  51         self.header = dict()
  52
  53         if errorcode == 200 and fmt != 'debug':
  54             getattr(self, '_parse_' + fmt)()
  55
  56     def _parse_json(self):
  57         m = re.fullmatch(r'([\w$][^(]*)\((.*)\)', self.page)
  58         if m is None:
  59             code = self.page
  60         else:
  61             code = m.group(2)
  62             self.header['json_func'] = m.group(1)
  63         self.result = json.JSONDecoder(object_pairs_hook=OrderedDict).decode(code)
  64         if isinstance(self.result, OrderedDict):
  65             self.result = [self.result]
  66
  67     def _parse_geojson(self):
  68         self._parse_json()
  69         if 'error' in self.result[0]:
  70             self.result = []
  71         else:
  72             self.result = list(map(_geojson_result_to_json_result, self.result[0]['features']))
  73
  74     def _parse_geocodejson(self):
  75         self._parse_geojson()
  76         if self.result is not None:
  77             self.result = [r['geocoding'] for r in self.result]
  78
  79     def assert_field(self, idx, field, value):
  80         """ Check that result row `idx` has a field `field` with value `value`.
  81             Float numbers are matched approximately. When the expected value
  82             starts with a carat, regular expression matching is used.
  83         """
  84         assert field in self.result[idx], \
  85                "Result row {} has no field '{}'.\nFull row: {}"\
  86                    .format(idx, field, json.dumps(self.result[idx], indent=4))
  87
  88         if isinstance(value, float):
  89             assert Almost(value) == float(self.result[idx][field]), \
  90                    BadRowValueAssert(self, idx, field, value)
  91         elif value.startswith("^"):
  92             assert re.fullmatch(value, self.result[idx][field]), \
  93                    BadRowValueAssert(self, idx, field, value)
  94         else:
  95             assert str(self.result[idx][field]) == str(value), \
  96                    BadRowValueAssert(self, idx, field, value)
  97
  98     def assert_address_field(self, idx, field, value):
  99         """ Check that result rows`idx` has a field `field` with value `value`
 100             in its address. If idx is None, then all results are checked.
 101         """
 102         if idx is None:
 103             todo = range(len(self.result))
 104         else:
 105             todo = [int(idx)]
 106
 107         for idx in todo:
 108             assert 'address' in self.result[idx], \
 109                    "Result row {} has no field 'address'.\nFull row: {}"\
 110                        .format(idx, json.dumps(self.result[idx], indent=4))
 111
 112             address = self.result[idx]['address']
 113             assert field in address, \
 114                    "Result row {} has no field '{}' in address.\nFull address: {}"\
 115                        .format(idx, field, json.dumps(address, indent=4))
 116
 117             assert address[field] == value, \
 118                    "\nBad value for row {} field '{}' in address. Expected: {}, got: {}.\nFull address: {}"""\
 119                        .format(idx, field, value, address[field], json.dumps(address, indent=4))
 120
 121     def match_row(self, row):
 122         """ Match the result fields against the given behave table row.
 123         """
 124         if 'ID' in row.headings:
 125             todo = [int(row['ID'])]
 126         else:
 127             todo = range(len(self.result))
 128
 129         for i in todo:
 130             for name, value in zip(row.headings, row.cells):
 131                 if name == 'ID':
 132                     pass
 133                 elif name == 'osm':
 134                     self.assert_field(i, 'osm_type', value[0])
 135                     self.assert_field(i, 'osm_id', value[1:])
 136                 elif name == 'centroid':
 137                     lon, lat = value.split(' ')
 138                     self.assert_field(i, 'lat', float(lat))
 139                     self.assert_field(i, 'lon', float(lon))
 140                 else:
 141                     self.assert_field(i, name, value)
 142
 143     def property_list(self, prop):
 144         return [x[prop] for x in self.result]
 145
 146
 147 class SearchResponse(GenericResponse):
 148     """ Specialised class for search and lookup responses.
 149         Transforms the xml response in a format similar to json.
 150     """
 151
 152     def _parse_xml(self):
 153         xml_tree = ET.fromstring(self.page)
 154
 155         self.header = dict(xml_tree.attrib)
 156
 157         for child in xml_tree:
 158             assert child.tag == "place"
 159             self.result.append(dict(child.attrib))
 160
 161             address = {}
 162             for sub in child:
 163                 if sub.tag == 'extratags':
 164                     self.result[-1]['extratags'] = {}
 165                     for tag in sub:
 166                         self.result[-1]['extratags'][tag.attrib['key']] = tag.attrib['value']
 167                 elif sub.tag == 'namedetails':
 168                     self.result[-1]['namedetails'] = {}
 169                     for tag in sub:
 170                         self.result[-1]['namedetails'][tag.attrib['desc']] = tag.text
 171                 elif sub.tag == 'geokml':
 172                     self.result[-1][sub.tag] = True
 173                 else:
 174                     address[sub.tag] = sub.text
 175
 176             if address:
 177                 self.result[-1]['address'] = address
 178
 179
 180 class ReverseResponse(GenericResponse):
 181     """ Specialised class for reverse responses.
 182         Transforms the xml response in a format similar to json.
 183     """
 184
 185     def _parse_xml(self):
 186         xml_tree = ET.fromstring(self.page)
 187
 188         self.header = dict(xml_tree.attrib)
 189         self.result = []
 190
 191         for child in xml_tree:
 192             if child.tag == 'result':
 193                 assert not self.result, "More than one result in reverse result"
 194                 self.result.append(dict(child.attrib))
 195             elif child.tag == 'addressparts':
 196                 address = {}
 197                 for sub in child:
 198                     address[sub.tag] = sub.text
 199                 self.result[0]['address'] = address
 200             elif child.tag == 'extratags':
 201                 self.result[0]['extratags'] = {}
 202                 for tag in child:
 203                     self.result[0]['extratags'][tag.attrib['key']] = tag.attrib['value']
 204             elif child.tag == 'namedetails':
 205                 self.result[0]['namedetails'] = {}
 206                 for tag in child:
 207                     self.result[0]['namedetails'][tag.attrib['desc']] = tag.text
 208             elif child.tag == 'geokml':
 209                 self.result[0][child.tag] = True
 210             else:
 211                 assert child.tag == 'error', \
 212                        "Unknown XML tag {} on page: {}".format(child.tag, self.page)
 213
 214
 215 class StatusResponse(GenericResponse):
 216     """ Specialised class for status responses.
 217         Can also parse text responses.
 218     """
 219
 220     def _parse_text(self):
 221         pass