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) 2025 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 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     def _parse_geojson(self):
  49         self._parse_json()
  50         if self.result:
  51             geojson = self.result[0]
  52             # check for valid geojson
  53             check_for_attributes(geojson, 'type,features')
  54             assert geojson['type'] == 'FeatureCollection'
  55             assert isinstance(geojson['features'], list)
  56
  57             self.result = []
  58             for result in geojson['features']:
  59                 check_for_attributes(result, 'type,properties,geometry')
  60                 assert result['type'] == 'Feature'
  61                 new = result['properties']
  62                 check_for_attributes(new, 'geojson', 'absent')
  63                 new['geojson'] = result['geometry']
  64                 if 'bbox' in result:
  65                     check_for_attributes(new, 'boundingbox', 'absent')
  66                     # bbox is  minlon, minlat, maxlon, maxlat
  67                     # boundingbox is minlat, maxlat, minlon, maxlon
  68                     new['boundingbox'] = [result['bbox'][1],
  69                                           result['bbox'][3],
  70                                           result['bbox'][0],
  71                                           result['bbox'][2]]
  72                 for k, v in geojson.items():
  73                     if k not in ('type', 'features'):
  74                         check_for_attributes(new, '__' + k, 'absent')
  75                         new['__' + k] = v
  76                 self.result.append(new)
  77
  78     def _parse_geocodejson(self):
  79         self._parse_geojson()
  80         if self.result:
  81             for r in self.result:
  82                 assert set(r.keys()) == {'geocoding', 'geojson', '__geocoding'}, \
  83                        f"Unexpected keys in result: {r.keys()}"
  84                 check_for_attributes(r['geocoding'], 'geojson', 'absent')
  85                 inner = r.pop('geocoding')
  86                 r.update(inner)
  87
  88     def assert_address_field(self, idx, field, value):
  89         """ Check that result rows`idx` has a field `field` with value `value`
  90             in its address. If idx is None, then all results are checked.
  91         """
  92         if idx is None:
  93             todo = range(len(self.result))
  94         else:
  95             todo = [int(idx)]
  96
  97         for idx in todo:
  98             self.check_row(idx, 'address' in self.result[idx], "No field 'address'")
  99
 100             address = self.result[idx]['address']
 101             self.check_row_field(idx, field, value, base=address)
 102
 103     def match_row(self, row, context=None, field=None):
 104         """ Match the result fields against the given behave table row.
 105         """
 106         if 'ID' in row.headings:
 107             todo = [int(row['ID'])]
 108         else:
 109             todo = range(len(self.result))
 110
 111         for i in todo:
 112             subdict = self.result[i]
 113             if field is not None:
 114                 for key in field.split('.'):
 115                     self.check_row(i, key in subdict, f"Missing subfield {key}")
 116                     subdict = subdict[key]
 117                     self.check_row(i, isinstance(subdict, dict),
 118                                    f"Subfield {key} not a dict")
 119
 120             for name, value in zip(row.headings, row.cells):
 121                 if name == 'ID':
 122                     pass
 123                 elif name == 'osm':
 124                     self.check_row_field(i, 'osm_type', OsmType(value[0]), base=subdict)
 125                     self.check_row_field(i, 'osm_id', Field(value[1:]), base=subdict)
 126                 elif name == 'centroid':
 127                     if ' ' in value:
 128                         lon, lat = value.split(' ')
 129                     elif context is not None:
 130                         lon, lat = context.osm.grid_node(int(value))
 131                     else:
 132                         raise RuntimeError("Context needed when using grid coordinates")
 133                     self.check_row_field(i, 'lat', Field(float(lat), abs_tol=1e-07), base=subdict)
 134                     self.check_row_field(i, 'lon', Field(float(lon), abs_tol=1e-07), base=subdict)
 135                 else:
 136                     self.check_row_field(i, name, Field(value), base=subdict)
 137
 138     def check_row(self, idx, check, msg):
 139         """ Assert for the condition 'check' and print 'msg' on fail together
 140             with the contents of the failing result.
 141         """
 142         class _RowError:
 143             def __init__(self, row):
 144                 self.row = row
 145
 146             def __str__(self):
 147                 return f"{msg}. Full row {idx}:\n" \
 148                        + json.dumps(self.row, indent=4, ensure_ascii=False)
 149
 150         assert check, _RowError(self.result[idx])
 151
 152     def check_row_field(self, idx, field, expected, base=None):
 153         """ Check field 'field' of result 'idx' for the expected value
 154             and print a meaningful error if the condition fails.
 155             When 'base' is set to a dictionary, then the field is checked
 156             in that base. The error message will still report the contents
 157             of the full result.
 158         """
 159         if base is None:
 160             base = self.result[idx]
 161
 162         self.check_row(idx, field in base, f"No field '{field}'")
 163         value = base[field]
 164
 165         self.check_row(idx, expected == value,
 166                        f"\nBad value for field '{field}'. Expected: {expected}, got: {value}")
 167
 168
 169 class SearchResponse(GenericResponse):
 170     """ Specialised class for search and lookup responses.
 171         Transforms the xml response in a format similar to json.
 172     """
 173
 174     def _parse_xml(self):
 175         xml_tree = ET.fromstring(self.page)
 176
 177         self.header = dict(xml_tree.attrib)
 178
 179         for child in xml_tree:
 180             assert child.tag == "place"
 181             self.result.append(dict(child.attrib))
 182
 183             address = {}
 184             for sub in child:
 185                 if sub.tag == 'extratags':
 186                     self.result[-1]['extratags'] = {}
 187                     for tag in sub:
 188                         self.result[-1]['extratags'][tag.attrib['key']] = tag.attrib['value']
 189                 elif sub.tag == 'namedetails':
 190                     self.result[-1]['namedetails'] = {}
 191                     for tag in sub:
 192                         self.result[-1]['namedetails'][tag.attrib['desc']] = tag.text
 193                 elif sub.tag == 'geokml':
 194                     self.result[-1][sub.tag] = True
 195                 else:
 196                     address[sub.tag] = sub.text
 197
 198             if address:
 199                 self.result[-1]['address'] = address
 200
 201
 202 class ReverseResponse(GenericResponse):
 203     """ Specialised class for reverse responses.
 204         Transforms the xml response in a format similar to json.
 205     """
 206
 207     def _parse_xml(self):
 208         xml_tree = ET.fromstring(self.page)
 209
 210         self.header = dict(xml_tree.attrib)
 211         self.result = []
 212
 213         for child in xml_tree:
 214             if child.tag == 'result':
 215                 assert not self.result, "More than one result in reverse result"
 216                 self.result.append(dict(child.attrib))
 217                 check_for_attributes(self.result[0], 'display_name', 'absent')
 218                 self.result[0]['display_name'] = child.text
 219             elif child.tag == 'addressparts':
 220                 assert 'address' not in self.result[0], "More than one address in result"
 221                 address = {}
 222                 for sub in child:
 223                     assert len(sub) == 0, f"Address element '{sub.tag}' has subelements"
 224                     address[sub.tag] = sub.text
 225                 self.result[0]['address'] = address
 226             elif child.tag == 'extratags':
 227                 assert 'extratags' not in self.result[0], "More than one extratags in result"
 228                 self.result[0]['extratags'] = {}
 229                 for tag in child:
 230                     assert len(tag) == 0, f"Extratags element '{tag.attrib['key']}' has subelements"
 231                     self.result[0]['extratags'][tag.attrib['key']] = tag.attrib['value']
 232             elif child.tag == 'namedetails':
 233                 assert 'namedetails' not in self.result[0], "More than one namedetails in result"
 234                 self.result[0]['namedetails'] = {}
 235                 for tag in child:
 236                     assert len(tag) == 0, \
 237                         f"Namedetails element '{tag.attrib['desc']}' has subelements"
 238                     self.result[0]['namedetails'][tag.attrib['desc']] = tag.text
 239             elif child.tag == 'geokml':
 240                 assert 'geokml' not in self.result[0], "More than one geokml in result"
 241                 self.result[0]['geokml'] = ET.tostring(child, encoding='unicode')
 242             else:
 243                 assert child.tag == 'error', \
 244                        f"Unknown XML tag {child.tag} on page: {self.page}"
 245
 246
 247 class StatusResponse(GenericResponse):
 248     """ Specialised class for status responses.
 249         Can also parse text responses.
 250     """
 251
 252     def _parse_text(self):
 253         pass