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, OsmType, Field, check_for_attributes
  15
  16
  17 class GenericResponse:
  18     """ Common base class for all API responses.
  19     """
  20     def __init__(self, page, fmt, errorcode=200):
  21         fmt = fmt.strip()
  22         if fmt == 'jsonv2':
  23             fmt = 'json'
  24
  25         self.page = page
  26         self.format = fmt
  27         self.errorcode = errorcode
  28         self.result = []
  29         self.header = dict()
  30
  31         if errorcode == 200 and fmt != 'debug':
  32             getattr(self, '_parse_' + fmt)()
  33
  34     def _parse_json(self):
  35         m = re.fullmatch(r'([\w$][^(]*)\((.*)\)', self.page)
  36         if m is None:
  37             code = self.page
  38         else:
  39             code = m.group(2)
  40             self.header['json_func'] = m.group(1)
  41         self.result = json.JSONDecoder().decode(code)
  42         if isinstance(self.result, dict):
  43             if 'error' in self.result:
  44                 self.result = []
  45             else:
  46                 self.result = [self.result]
  47
  48
  49     def _parse_geojson(self):
  50         self._parse_json()
  51         if self.result:
  52             geojson = self.result[0]
  53             # check for valid geojson
  54             check_for_attributes(geojson, 'type,features')
  55             assert geojson['type'] == 'FeatureCollection'
  56             assert isinstance(geojson['features'], list)
  57
  58             self.result = []
  59             for result in geojson['features']:
  60                 check_for_attributes(result, 'type,properties,geometry')
  61                 assert result['type'] == 'Feature'
  62                 new = result['properties']
  63                 check_for_attributes(new, 'geojson', 'absent')
  64                 new['geojson'] = result['geometry']
  65                 if 'bbox' in result:
  66                     check_for_attributes(new, 'boundingbox', 'absent')
  67                     # bbox is  minlon, minlat, maxlon, maxlat
  68                     # boundingbox is minlat, maxlat, minlon, maxlon
  69                     new['boundingbox'] = [result['bbox'][1],
  70                                           result['bbox'][3],
  71                                           result['bbox'][0],
  72                                           result['bbox'][2]]
  73                 for k, v in geojson.items():
  74                     if k not in ('type', 'features'):
  75                         check_for_attributes(new, '__' + k, 'absent')
  76                         new['__' + k] = v
  77                 self.result.append(new)
  78
  79
  80     def _parse_geocodejson(self):
  81         self._parse_geojson()
  82         if self.result:
  83             for r in self.result:
  84                 assert set(r.keys()) == {'geocoding', 'geojson', '__geocoding'}, \
  85                        f"Unexpected keys in result: {r.keys()}"
  86                 check_for_attributes(r['geocoding'], 'geojson', 'absent')
  87                 r |= r.pop('geocoding')
  88
  89
  90     def assert_subfield(self, idx, path, value):
  91         assert path
  92
  93         field = self.result[idx]
  94         for p in path:
  95             assert isinstance(field, dict)
  96             assert p in field
  97             field = field[p]
  98
  99         if isinstance(value, float):
 100             assert Almost(value) == float(field)
 101         elif value.startswith("^"):
 102             assert re.fullmatch(value, field)
 103         elif isinstance(field, dict):
 104             assert field, eval('{' + value + '}')
 105         else:
 106             assert str(field) == str(value)
 107
 108
 109     def assert_address_field(self, idx, field, value):
 110         """ Check that result rows`idx` has a field `field` with value `value`
 111             in its address. If idx is None, then all results are checked.
 112         """
 113         if idx is None:
 114             todo = range(len(self.result))
 115         else:
 116             todo = [int(idx)]
 117
 118         for idx in todo:
 119             self.check_row(idx, 'address' in self.result[idx], "No field 'address'")
 120
 121             address = self.result[idx]['address']
 122             self.check_row_field(idx, field, value, base=address)
 123
 124
 125     def match_row(self, row, context=None):
 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                     self.check_row_field(i, 'osm_type', OsmType(value[0]))
 139                     self.check_row_field(i, 'osm_id', Field(value[1:]))
 140                 elif name == 'osm_type':
 141                     self.check_row_field(i, 'osm_type', OsmType(value[0]))
 142                 elif name == 'centroid':
 143                     if ' ' in value:
 144                         lon, lat = value.split(' ')
 145                     elif context is not None:
 146                         lon, lat = context.osm.grid_node(int(value))
 147                     else:
 148                         raise RuntimeError("Context needed when using grid coordinates")
 149                     self.check_row_field(i, 'lat', Field(float(lat)), base=subdict)
 150                     self.check_row_field(i, 'lon', Field(float(lon)), base=subdict)
 151                 elif '+' in name:
 152                     self.assert_subfield(i, name.split('+'), value)
 153                 else:
 154                     self.check_row_field(i, name, Field(value))
 155
 156
 157     def property_list(self, prop):
 158         return [x[prop] for x in self.result]
 159
 160
 161     def check_row(self, idx, check, msg):
 162         """ Assert for the condition 'check' and print 'msg' on fail together
 163             with the contents of the failing result.
 164         """
 165         class _RowError:
 166             def __init__(self, row):
 167                 self.row = row
 168
 169             def __str__(self):
 170                 return f"{msg}. Full row {idx}:\n" \
 171                        + json.dumps(self.row, indent=4, ensure_ascii=False)
 172
 173         assert check, _RowError(self.result[idx])
 174
 175
 176     def check_row_field(self, idx, field, expected, base=None):
 177         """ Check field 'field' of result 'idx' for the expected value
 178             and print a meaningful error if the condition fails.
 179             When 'base' is set to a dictionary, then the field is checked
 180             in that base. The error message will still report the contents
 181             of the full result.
 182         """
 183         if base is None:
 184             base = self.result[idx]
 185
 186         self.check_row(idx, field in base, f"No field '{field}'")
 187         value = base[field]
 188
 189         self.check_row(idx, expected == value,
 190                        f"\nBad value for field '{field}'. Expected: {expected}, got: {value}")
 191
 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