test/bdd/steps/http_responses.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2023 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Classes wrapping HTTP responses from the Nominatim API.
   9 """
  10 import re
  11 import json
  12 import xml.etree.ElementTree as ET
  13
  14 from check_functions import Almost, check_for_attributes
  15
  16 OSM_TYPE = {'N' : 'node', 'W' : 'way', 'R' : 'relation',
  17             'n' : 'node', 'w' : 'way', 'r' : 'relation',
  18             'node' : 'n', 'way' : 'w', 'relation' : 'r'}
  19
  20 def _geojson_result_to_json_result(geojson_result):
  21     result = geojson_result['properties']
  22     result['geojson'] = geojson_result['geometry']
  23     if 'bbox' in geojson_result:
  24         # bbox is  minlon, minlat, maxlon, maxlat
  25         # boundingbox is minlat, maxlat, minlon, maxlon
  26         result['boundingbox'] = [geojson_result['bbox'][1],
  27                                  geojson_result['bbox'][3],
  28                                  geojson_result['bbox'][0],
  29                                  geojson_result['bbox'][2]]
  30     return result
  31
  32 class BadRowValueAssert:
  33     """ Lazily formatted message for failures to find a field content.
  34     """
  35
  36     def __init__(self, response, idx, field, value):
  37         self.idx = idx
  38         self.field = field
  39         self.value = value
  40         self.row = response.result[idx]
  41
  42     def __str__(self):
  43         return "\nBad value for row {} field '{}'. Expected: {}, got: {}.\nFull row: {}"""\
  44                    .format(self.idx, self.field, self.value,
  45                            self.row[self.field], json.dumps(self.row, indent=4))
  46
  47
  48 class GenericResponse:
  49     """ Common base class for all API responses.
  50     """
  51     def __init__(self, page, fmt, errorcode=200):
  52         fmt = fmt.strip()
  53         if fmt == 'jsonv2':
  54             fmt = 'json'
  55
  56         self.page = page
  57         self.format = fmt
  58         self.errorcode = errorcode
  59         self.result = []
  60         self.header = dict()
  61
  62         if errorcode == 200 and fmt != 'debug':
  63             getattr(self, '_parse_' + fmt)()
  64
  65     def _parse_json(self):
  66         m = re.fullmatch(r'([\w$][^(]*)\((.*)\)', self.page)
  67         if m is None:
  68             code = self.page
  69         else:
  70             code = m.group(2)
  71             self.header['json_func'] = m.group(1)
  72         self.result = json.JSONDecoder().decode(code)
  73         if isinstance(self.result, dict):
  74             if 'error' in self.result:
  75                 self.result = []
  76             else:
  77                 self.result = [self.result]
  78
  79
  80     def _parse_geojson(self):
  81         self._parse_json()
  82         if self.result:
  83             geojson = self.result[0]
  84             # check for valid geojson
  85             check_for_attributes(geojson, 'type,features')
  86             assert geojson['type'] == 'FeatureCollection'
  87             assert isinstance(geojson['features'], list)
  88
  89             self.result = []
  90             for result in geojson['features']:
  91                 check_for_attributes(result, 'type,properties,geometry')
  92                 assert result['type'] == 'Feature'
  93                 new = result['properties']
  94                 check_for_attributes(new, 'geojson', 'absent')
  95                 new['geojson'] = result['geometry']
  96                 if 'bbox' in result:
  97                     check_for_attributes(new, 'boundingbox', 'absent')
  98                     # bbox is  minlon, minlat, maxlon, maxlat
  99                     # boundingbox is minlat, maxlat, minlon, maxlon
 100                     new['boundingbox'] = [result['bbox'][1],
 101                                           result['bbox'][3],
 102                                           result['bbox'][0],
 103                                           result['bbox'][2]]
 104                 for k, v in geojson.items():
 105                     if k not in ('type', 'features'):
 106                         check_for_attributes(new, '__' + k, 'absent')
 107                         new['__' + k] = v
 108                 self.result.append(new)
 109
 110
 111     def _parse_geocodejson(self):
 112         self._parse_geojson()
 113         if self.result:
 114             for r in self.result:
 115                 assert set(r.keys()) == {'geocoding', 'geojson', '__geocoding'}, \
 116                        f"Unexpected keys in result: {r.keys()}"
 117                 check_for_attributes(r['geocoding'], 'geojson', 'absent')
 118                 r |= r.pop('geocoding')
 119
 120
 121     def assert_field(self, idx, field, value):
 122         """ Check that result row `idx` has a field `field` with value `value`.
 123             Float numbers are matched approximately. When the expected value
 124             starts with a carat, regular expression matching is used.
 125         """
 126         assert field in self.result[idx], \
 127                "Result row {} has no field '{}'.\nFull row: {}"\
 128                    .format(idx, field, json.dumps(self.result[idx], indent=4))
 129
 130         if isinstance(value, float):
 131             assert Almost(value) == float(self.result[idx][field]), \
 132                    BadRowValueAssert(self, idx, field, value)
 133         elif value.startswith("^"):
 134             assert re.fullmatch(value, self.result[idx][field]), \
 135                    BadRowValueAssert(self, idx, field, value)
 136         elif isinstance(self.result[idx][field], dict):
 137             assert self.result[idx][field] == eval('{' + value + '}'), \
 138                    BadRowValueAssert(self, idx, field, value)
 139         else:
 140             assert str(self.result[idx][field]) == str(value), \
 141                    BadRowValueAssert(self, idx, field, value)
 142
 143
 144     def assert_subfield(self, idx, path, value):
 145         assert path
 146
 147         field = self.result[idx]
 148         for p in path:
 149             assert isinstance(field, dict)
 150             assert p in field
 151             field = field[p]
 152
 153         if isinstance(value, float):
 154             assert Almost(value) == float(field)
 155         elif value.startswith("^"):
 156             assert re.fullmatch(value, field)
 157         elif isinstance(field, dict):
 158             assert field, eval('{' + value + '}')
 159         else:
 160             assert str(field) == str(value)
 161
 162
 163     def assert_address_field(self, idx, field, value):
 164         """ Check that result rows`idx` has a field `field` with value `value`
 165             in its address. If idx is None, then all results are checked.
 166         """
 167         if idx is None:
 168             todo = range(len(self.result))
 169         else:
 170             todo = [int(idx)]
 171
 172         for idx in todo:
 173             assert 'address' in self.result[idx], \
 174                    "Result row {} has no field 'address'.\nFull row: {}"\
 175                        .format(idx, json.dumps(self.result[idx], indent=4))
 176
 177             address = self.result[idx]['address']
 178             assert field in address, \
 179                    "Result row {} has no field '{}' in address.\nFull address: {}"\
 180                        .format(idx, field, json.dumps(address, indent=4))
 181
 182             assert address[field] == value, \
 183                    "\nBad value for row {} field '{}' in address. Expected: {}, got: {}.\nFull address: {}"""\
 184                        .format(idx, field, value, address[field], json.dumps(address, indent=4))
 185
 186     def match_row(self, row, context=None):
 187         """ Match the result fields against the given behave table row.
 188         """
 189         if 'ID' in row.headings:
 190             todo = [int(row['ID'])]
 191         else:
 192             todo = range(len(self.result))
 193
 194         for i in todo:
 195             for name, value in zip(row.headings, row.cells):
 196                 if name == 'ID':
 197                     pass
 198                 elif name == 'osm':
 199                     assert 'osm_type' in self.result[i], \
 200                            "Result row {} has no field 'osm_type'.\nFull row: {}"\
 201                                .format(i, json.dumps(self.result[i], indent=4))
 202                     assert self.result[i]['osm_type'] in (OSM_TYPE[value[0]], value[0]), \
 203                            BadRowValueAssert(self, i, 'osm_type', value)
 204                     self.assert_field(i, 'osm_id', value[1:])
 205                 elif name == 'osm_type':
 206                     assert self.result[i]['osm_type'] in (OSM_TYPE[value[0]], value[0]), \
 207                            BadRowValueAssert(self, i, 'osm_type', value)
 208                 elif name == 'centroid':
 209                     if ' ' in value:
 210                         lon, lat = value.split(' ')
 211                     elif context is not None:
 212                         lon, lat = context.osm.grid_node(int(value))
 213                     else:
 214                         raise RuntimeError("Context needed when using grid coordinates")
 215                     self.assert_field(i, 'lat', float(lat))
 216                     self.assert_field(i, 'lon', float(lon))
 217                 elif '+' in name:
 218                     self.assert_subfield(i, name.split('+'), value)
 219                 else:
 220                     self.assert_field(i, name, value)
 221
 222     def property_list(self, prop):
 223         return [x[prop] for x in self.result]
 224
 225
 226 class SearchResponse(GenericResponse):
 227     """ Specialised class for search and lookup responses.
 228         Transforms the xml response in a format similar to json.
 229     """
 230
 231     def _parse_xml(self):
 232         xml_tree = ET.fromstring(self.page)
 233
 234         self.header = dict(xml_tree.attrib)
 235
 236         for child in xml_tree:
 237             assert child.tag == "place"
 238             self.result.append(dict(child.attrib))
 239
 240             address = {}
 241             for sub in child:
 242                 if sub.tag == 'extratags':
 243                     self.result[-1]['extratags'] = {}
 244                     for tag in sub:
 245                         self.result[-1]['extratags'][tag.attrib['key']] = tag.attrib['value']
 246                 elif sub.tag == 'namedetails':
 247                     self.result[-1]['namedetails'] = {}
 248                     for tag in sub:
 249                         self.result[-1]['namedetails'][tag.attrib['desc']] = tag.text
 250                 elif sub.tag == 'geokml':
 251                     self.result[-1][sub.tag] = True
 252                 else:
 253                     address[sub.tag] = sub.text
 254
 255             if address:
 256                 self.result[-1]['address'] = address
 257
 258
 259 class ReverseResponse(GenericResponse):
 260     """ Specialised class for reverse responses.
 261         Transforms the xml response in a format similar to json.
 262     """
 263
 264     def _parse_xml(self):
 265         xml_tree = ET.fromstring(self.page)
 266
 267         self.header = dict(xml_tree.attrib)
 268         self.result = []
 269
 270         for child in xml_tree:
 271             if child.tag == 'result':
 272                 assert not self.result, "More than one result in reverse result"
 273                 self.result.append(dict(child.attrib))
 274             elif child.tag == 'addressparts':
 275                 address = {}
 276                 for sub in child:
 277                     address[sub.tag] = sub.text
 278                 self.result[0]['address'] = address
 279             elif child.tag == 'extratags':
 280                 self.result[0]['extratags'] = {}
 281                 for tag in child:
 282                     self.result[0]['extratags'][tag.attrib['key']] = tag.attrib['value']
 283             elif child.tag == 'namedetails':
 284                 self.result[0]['namedetails'] = {}
 285                 for tag in child:
 286                     self.result[0]['namedetails'][tag.attrib['desc']] = tag.text
 287             elif child.tag == 'geokml':
 288                 self.result[0][child.tag] = True
 289             else:
 290                 assert child.tag == 'error', \
 291                        "Unknown XML tag {} on page: {}".format(child.tag, self.page)
 292
 293
 294 class StatusResponse(GenericResponse):
 295     """ Specialised class for status responses.
 296         Can also parse text responses.
 297     """
 298
 299     def _parse_text(self):
 300         pass