import io
import re
import logging
-from tidylib import tidy_document
import xml.etree.ElementTree as ET
import subprocess
from urllib.parse import urlencode
pass
elif h == 'osm':
assert_equal(res['osm_type'], row[h][0])
- assert_equal(res['osm_id'], row[h][1:])
+ assert_equal(res['osm_id'], int(row[h][1:]))
elif h == 'centroid':
x, y = row[h].split(' ')
assert_almost_equal(float(y), float(res['lat']))
self.result = geojson_results_to_json_results(self.result)
def parse_geocodejson(self):
- return self.parse_geojson()
-
- def parse_html(self):
- content, errors = tidy_document(self.page,
- options={'char-encoding' : 'utf8'})
- #eq_(len(errors), 0 , "Errors found in HTML document:\n%s" % errors)
-
- b = content.find('nominatim_results =')
- e = content.find('</script>')
- content = content[b:e]
- b = content.find('[')
- e = content.rfind(']')
-
- self.result = json.JSONDecoder(object_pairs_hook=OrderedDict).decode(content[b:e+1])
+ self.parse_geojson()
+ if self.result is not None:
+ self.result = [r['geocoding'] for r in self.result]
def parse_xml(self):
et = ET.fromstring(self.page)
if errorcode == 200:
getattr(self, 'parse_' + fmt)()
- def parse_html(self):
- content, errors = tidy_document(self.page,
- options={'char-encoding' : 'utf8'})
- #eq_(len(errors), 0 , "Errors found in HTML document:\n%s" % errors)
-
- b = content.find('nominatim_results =')
- e = content.find('</script>')
- content = content[b:e]
- b = content.find('[')
- e = content.rfind(']')
-
- self.result = json.JSONDecoder(object_pairs_hook=OrderedDict).decode(content[b:e+1])
-
def parse_json(self):
m = re.fullmatch(r'([\w$][^(]*)\((.*)\)', self.page)
if m is None:
self.result = geojson_results_to_json_results(self.result[0])
def parse_geocodejson(self):
- return self.parse_geojson()
+ self.parse_geojson()
+ if self.result is not None:
+ self.result = [r['geocoding'] for r in self.result]
def parse_xml(self):
et = ET.fromstring(self.page)
if errorcode == 200:
getattr(self, 'parse_' + fmt)()
- def parse_html(self):
- content, errors = tidy_document(self.page,
- options={'char-encoding' : 'utf8'})
- self.result = {}
-
def parse_json(self):
self.result = [json.JSONDecoder(object_pairs_hook=OrderedDict).decode(self.page)]
"""
cmd = ['/usr/bin/env', 'php']
cmd.append(os.path.join(context.nominatim.build_dir, 'utils', 'query.php'))
- cmd.extend(['--search', query])
+ if query:
+ cmd.extend(['--search', query])
# add more parameters in table form
if context.table:
for h in context.table.headings:
(outp, err) = proc.communicate()
assert_equals (0, proc.returncode, "query.php failed with message: %s\noutput: %s" % (err, outp))
+ logger.debug("run_nominatim_script: %s\n%s\n" % (cmd, outp.decode('utf-8').replace('\\n', '\n')))
context.response = SearchResponse(outp.decode('utf-8'), 'json')
outp, status = send_api_query('search', params, fmt, context)
- if fmt is None:
- outfmt = 'html'
- elif fmt == 'jsonv2 ':
+ if fmt is None or fmt == 'jsonv2 ':
outfmt = 'json'
else:
outfmt = fmt.strip()
outp, status = send_api_query('details', params, fmt, context)
if fmt is None:
- outfmt = 'html'
+ outfmt = 'json'
else:
outfmt = fmt.strip()
if fmt == 'json ':
outfmt = 'json'
+ elif fmt == 'jsonv2 ':
+ outfmt = 'json'
elif fmt == 'geojson ':
outfmt = 'geojson'
+ elif fmt == 'geocodejson ':
+ outfmt = 'geocodejson'
else:
outfmt = 'xml'
context.execute_steps("Then a HTTP 200 is returned")
eq_(context.response.format, fmt)
+@then(u'a (?P<fmt>\w+) user error is returned')
+def check_page_error(context, fmt):
+ context.execute_steps("Then a HTTP 400 is returned")
+ eq_(context.response.format, fmt)
+
+ if fmt == 'xml':
+ assert_is_not_none(re.search(r'<error>.+</error>', context.response.page, re.DOTALL))
+ else:
+ assert_is_not_none(re.search(r'({"error":)', context.response.page, re.DOTALL))
+
@then(u'result header contains')
def check_header_attr(context):
for line in context.table:
else:
assert_in(attr, addr_parts)
-@then(u'address of result (?P<lid>\d+) is')
-def check_address(context, lid):
+@then(u'address of result (?P<lid>\d+) (?P<complete>is|contains)')
+def check_address(context, lid, complete):
context.execute_steps("then more than %s results are returned" % lid)
addr_parts = dict(context.response.result[int(lid)]['address'])
"Bad address value for %s" % line['type'])
del addr_parts[line['type']]
- eq_(0, len(addr_parts), "Additional address parts found: %s" % str(addr_parts))
+ if complete == 'is':
+ eq_(0, len(addr_parts), "Additional address parts found: %s" % str(addr_parts))
@then(u'result (?P<lid>\d+ )?has bounding box in (?P<coords>[\d,.-]+)')
def step_impl(context, lid, coords):
assert_greater_equal(bbox[2], coord[2])
assert_less_equal(bbox[3], coord[3])
+@then(u'result (?P<lid>\d+ )?has centroid in (?P<coords>[\d,.-]+)')
+def step_impl(context, lid, coords):
+ if lid is None:
+ context.execute_steps("then at least 1 result is returned")
+ bboxes = zip(context.response.property_list('lat'),
+ context.response.property_list('lon'))
+ else:
+ context.execute_steps("then more than %sresults are returned" % lid)
+ res = context.response.result[int(lid)]
+ bboxes = [ (res['lat'], res['lon']) ]
+
+ coord = [ float(x) for x in coords.split(',') ]
+
+ for lat, lon in bboxes:
+ lat = float(lat)
+ lon = float(lon)
+ assert_greater_equal(lat, coord[0])
+ assert_less_equal(lat, coord[1])
+ assert_greater_equal(lon, coord[2])
+ assert_less_equal(lon, coord[3])
+
@then(u'there are(?P<neg> no)? duplicates')
def check_for_duplicates(context, neg):
context.execute_steps("then at least 1 result is returned")
projects / nominatim.git / blobdiff