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
  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     def _parse_geojson(self):
  80         self._parse_json()
  81         if self.result:
  82             self.result = list(map(_geojson_result_to_json_result, self.result[0]['features']))
  83
  84     def _parse_geocodejson(self):
  85         self._parse_geojson()
  86         if self.result is not None:
  87             self.result = [r['geocoding'] for r in self.result]
  88
  89     def assert_field(self, idx, field, value):
  90         """ Check that result row `idx` has a field `field` with value `value`.
  91             Float numbers are matched approximately. When the expected value
  92             starts with a carat, regular expression matching is used.
  93         """
  94         assert field in self.result[idx], \
  95                "Result row {} has no field '{}'.\nFull row: {}"\
  96                    .format(idx, field, json.dumps(self.result[idx], indent=4))
  97
  98         if isinstance(value, float):
  99             assert Almost(value) == float(self.result[idx][field]), \
 100                    BadRowValueAssert(self, idx, field, value)
 101         elif value.startswith("^"):
 102             assert re.fullmatch(value, self.result[idx][field]), \
 103                    BadRowValueAssert(self, idx, field, value)
 104         elif isinstance(self.result[idx][field], dict):
 105             assert self.result[idx][field] == eval('{' + value + '}'), \
 106                    BadRowValueAssert(self, idx, field, value)
 107         else:
 108             assert str(self.result[idx][field]) == str(value), \
 109                    BadRowValueAssert(self, idx, field, value)
 110
 111
 112     def assert_subfield(self, idx, path, value):
 113         assert path
 114
 115         field = self.result[idx]
 116         for p in path:
 117             assert isinstance(field, dict)
 118             assert p in field
 119             field = field[p]
 120
 121         if isinstance(value, float):
 122             assert Almost(value) == float(field)
 123         elif value.startswith("^"):
 124             assert re.fullmatch(value, field)
 125         elif isinstance(field, dict):
 126             assert field, eval('{' + value + '}')
 127         else:
 128             assert str(field) == str(value)
 129
 130
 131     def assert_address_field(self, idx, field, value):
 132         """ Check that result rows`idx` has a field `field` with value `value`
 133             in its address. If idx is None, then all results are checked.
 134         """
 135         if idx is None:
 136             todo = range(len(self.result))
 137         else:
 138             todo = [int(idx)]
 139
 140         for idx in todo:
 141             assert 'address' in self.result[idx], \
 142                    "Result row {} has no field 'address'.\nFull row: {}"\
 143                        .format(idx, json.dumps(self.result[idx], indent=4))
 144
 145             address = self.result[idx]['address']
 146             assert field in address, \
 147                    "Result row {} has no field '{}' in address.\nFull address: {}"\
 148                        .format(idx, field, json.dumps(address, indent=4))
 149
 150             assert address[field] == value, \
 151                    "\nBad value for row {} field '{}' in address. Expected: {}, got: {}.\nFull address: {}"""\
 152                        .format(idx, field, value, address[field], json.dumps(address, indent=4))
 153
 154     def match_row(self, row, context=None):
 155         """ Match the result fields against the given behave table row.
 156         """
 157         if 'ID' in row.headings:
 158             todo = [int(row['ID'])]
 159         else:
 160             todo = range(len(self.result))
 161
 162         for i in todo:
 163             for name, value in zip(row.headings, row.cells):
 164                 if name == 'ID':
 165                     pass
 166                 elif name == 'osm':
 167                     assert 'osm_type' in self.result[i], \
 168                            "Result row {} has no field 'osm_type'.\nFull row: {}"\
 169                                .format(i, json.dumps(self.result[i], indent=4))
 170                     assert self.result[i]['osm_type'] in (OSM_TYPE[value[0]], value[0]), \
 171                            BadRowValueAssert(self, i, 'osm_type', value)
 172                     self.assert_field(i, 'osm_id', value[1:])
 173                 elif name == 'osm_type':
 174                     assert self.result[i]['osm_type'] in (OSM_TYPE[value[0]], value[0]), \
 175                            BadRowValueAssert(self, i, 'osm_type', value)
 176                 elif name == 'centroid':
 177                     if ' ' in value:
 178                         lon, lat = value.split(' ')
 179                     elif context is not None:
 180                         lon, lat = context.osm.grid_node(int(value))
 181                     else:
 182                         raise RuntimeError("Context needed when using grid coordinates")
 183                     self.assert_field(i, 'lat', float(lat))
 184                     self.assert_field(i, 'lon', float(lon))
 185                 elif '+' in name:
 186                     self.assert_subfield(i, name.split('+'), value)
 187                 else:
 188                     self.assert_field(i, name, value)
 189
 190     def property_list(self, prop):
 191         return [x[prop] for x in self.result]
 192
 193
 194 class SearchResponse(GenericResponse):
 195     """ Specialised class for search and lookup responses.
 196         Transforms the xml response in a format similar to json.
 197     """
 198
 199     def _parse_xml(self):
 200         xml_tree = ET.fromstring(self.page)
 201
 202         self.header = dict(xml_tree.attrib)
 203
 204         for child in xml_tree:
 205             assert child.tag == "place"
 206             self.result.append(dict(child.attrib))
 207
 208             address = {}
 209             for sub in child:
 210                 if sub.tag == 'extratags':
 211                     self.result[-1]['extratags'] = {}
 212                     for tag in sub:
 213                         self.result[-1]['extratags'][tag.attrib['key']] = tag.attrib['value']
 214                 elif sub.tag == 'namedetails':
 215                     self.result[-1]['namedetails'] = {}
 216                     for tag in sub:
 217                         self.result[-1]['namedetails'][tag.attrib['desc']] = tag.text
 218                 elif sub.tag == 'geokml':
 219                     self.result[-1][sub.tag] = True
 220                 else:
 221                     address[sub.tag] = sub.text
 222
 223             if address:
 224                 self.result[-1]['address'] = address
 225
 226
 227 class ReverseResponse(GenericResponse):
 228     """ Specialised class for reverse responses.
 229         Transforms the xml response in a format similar to json.
 230     """
 231
 232     def _parse_xml(self):
 233         xml_tree = ET.fromstring(self.page)
 234
 235         self.header = dict(xml_tree.attrib)
 236         self.result = []
 237
 238         for child in xml_tree:
 239             if child.tag == 'result':
 240                 assert not self.result, "More than one result in reverse result"
 241                 self.result.append(dict(child.attrib))
 242             elif child.tag == 'addressparts':
 243                 address = {}
 244                 for sub in child:
 245                     address[sub.tag] = sub.text
 246                 self.result[0]['address'] = address
 247             elif child.tag == 'extratags':
 248                 self.result[0]['extratags'] = {}
 249                 for tag in child:
 250                     self.result[0]['extratags'][tag.attrib['key']] = tag.attrib['value']
 251             elif child.tag == 'namedetails':
 252                 self.result[0]['namedetails'] = {}
 253                 for tag in child:
 254                     self.result[0]['namedetails'][tag.attrib['desc']] = tag.text
 255             elif child.tag == 'geokml':
 256                 self.result[0][child.tag] = True
 257             else:
 258                 assert child.tag == 'error', \
 259                        "Unknown XML tag {} on page: {}".format(child.tag, self.page)
 260
 261
 262 class StatusResponse(GenericResponse):
 263     """ Specialised class for status responses.
 264         Can also parse text responses.
 265     """
 266
 267     def _parse_text(self):
 268         pass