X-Git-Url: https://git.openstreetmap.org./nominatim.git/blobdiff_plain/340e7f7210fca8095fc07ebaf3fb0e95cd0db518..fbff4fa2187f14e1aa55f92ff4f9866a769ef808:/test/bdd/steps/http_responses.py diff --git a/test/bdd/steps/http_responses.py b/test/bdd/steps/http_responses.py index beafcd9e..aee0d7f0 100644 --- a/test/bdd/steps/http_responses.py +++ b/test/bdd/steps/http_responses.py @@ -1,12 +1,21 @@ +# SPDX-License-Identifier: GPL-2.0-only +# +# This file is part of Nominatim. (https://nominatim.org) +# +# Copyright (C) 2023 by the Nominatim developer community. +# For a full list of authors see the git log. """ Classes wrapping HTTP responses from the Nominatim API. """ -from collections import OrderedDict import re import json import xml.etree.ElementTree as ET -from check_functions import Almost +from check_functions import Almost, check_for_attributes + +OSM_TYPE = {'N' : 'node', 'W' : 'way', 'R' : 'relation', + 'n' : 'node', 'w' : 'way', 'r' : 'relation', + 'node' : 'n', 'way' : 'w', 'relation' : 'r'} def _geojson_result_to_json_result(geojson_result): result = geojson_result['properties'] @@ -60,21 +69,54 @@ class GenericResponse: else: code = m.group(2) self.header['json_func'] = m.group(1) - self.result = json.JSONDecoder(object_pairs_hook=OrderedDict).decode(code) - if isinstance(self.result, OrderedDict): - self.result = [self.result] + self.result = json.JSONDecoder().decode(code) + if isinstance(self.result, dict): + if 'error' in self.result: + self.result = [] + else: + self.result = [self.result] + def _parse_geojson(self): self._parse_json() - if 'error' in self.result[0]: + if self.result: + geojson = self.result[0] + # check for valid geojson + check_for_attributes(geojson, 'type,features') + assert geojson['type'] == 'FeatureCollection' + assert isinstance(geojson['features'], list) + self.result = [] - else: - self.result = list(map(_geojson_result_to_json_result, self.result[0]['features'])) + for result in geojson['features']: + check_for_attributes(result, 'type,properties,geometry') + assert result['type'] == 'Feature' + new = result['properties'] + check_for_attributes(new, 'geojson', 'absent') + new['geojson'] = result['geometry'] + if 'bbox' in result: + check_for_attributes(new, 'boundingbox', 'absent') + # bbox is minlon, minlat, maxlon, maxlat + # boundingbox is minlat, maxlat, minlon, maxlon + new['boundingbox'] = [result['bbox'][1], + result['bbox'][3], + result['bbox'][0], + result['bbox'][2]] + for k, v in geojson.items(): + if k not in ('type', 'features'): + check_for_attributes(new, '__' + k, 'absent') + new['__' + k] = v + self.result.append(new) + def _parse_geocodejson(self): self._parse_geojson() - if self.result is not None: - self.result = [r['geocoding'] for r in self.result] + if self.result: + for r in self.result: + assert set(r.keys()) == {'geocoding', 'geojson', '__geocoding'}, \ + f"Unexpected keys in result: {r.keys()}" + check_for_attributes(r['geocoding'], 'geojson', 'absent') + r |= r.pop('geocoding') + def assert_field(self, idx, field, value): """ Check that result row `idx` has a field `field` with value `value`. @@ -91,10 +133,33 @@ class GenericResponse: elif value.startswith("^"): assert re.fullmatch(value, self.result[idx][field]), \ BadRowValueAssert(self, idx, field, value) + elif isinstance(self.result[idx][field], dict): + assert self.result[idx][field] == eval('{' + value + '}'), \ + BadRowValueAssert(self, idx, field, value) else: assert str(self.result[idx][field]) == str(value), \ BadRowValueAssert(self, idx, field, value) + + def assert_subfield(self, idx, path, value): + assert path + + field = self.result[idx] + for p in path: + assert isinstance(field, dict) + assert p in field + field = field[p] + + if isinstance(value, float): + assert Almost(value) == float(field) + elif value.startswith("^"): + assert re.fullmatch(value, field) + elif isinstance(field, dict): + assert field, eval('{' + value + '}') + else: + assert str(field) == str(value) + + def assert_address_field(self, idx, field, value): """ Check that result rows`idx` has a field `field` with value `value` in its address. If idx is None, then all results are checked. @@ -118,7 +183,7 @@ class GenericResponse: "\nBad value for row {} field '{}' in address. Expected: {}, got: {}.\nFull address: {}"""\ .format(idx, field, value, address[field], json.dumps(address, indent=4)) - def match_row(self, row): + def match_row(self, row, context=None): """ Match the result fields against the given behave table row. """ if 'ID' in row.headings: @@ -131,12 +196,26 @@ class GenericResponse: if name == 'ID': pass elif name == 'osm': - self.assert_field(i, 'osm_type', value[0]) + assert 'osm_type' in self.result[i], \ + "Result row {} has no field 'osm_type'.\nFull row: {}"\ + .format(i, json.dumps(self.result[i], indent=4)) + assert self.result[i]['osm_type'] in (OSM_TYPE[value[0]], value[0]), \ + BadRowValueAssert(self, i, 'osm_type', value) self.assert_field(i, 'osm_id', value[1:]) + elif name == 'osm_type': + assert self.result[i]['osm_type'] in (OSM_TYPE[value[0]], value[0]), \ + BadRowValueAssert(self, i, 'osm_type', value) elif name == 'centroid': - lon, lat = value.split(' ') + if ' ' in value: + lon, lat = value.split(' ') + elif context is not None: + lon, lat = context.osm.grid_node(int(value)) + else: + raise RuntimeError("Context needed when using grid coordinates") self.assert_field(i, 'lat', float(lat)) self.assert_field(i, 'lon', float(lon)) + elif '+' in name: + self.assert_subfield(i, name.split('+'), value) else: self.assert_field(i, name, value)