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