+# 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']
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`.
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.
"\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:
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)
projects / nominatim.git / blobdiff