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