entrouvert
/
passerelle
15
0
Fork 0
Code Issues Pull Requests 16 Releases Activity

opengis: PEP8ness, code style (#30458)

This commit is contained in:
Benjamin Dauvergne 2019-02-08 07:27:10 +01:00
parent ddffadf025
commit 204c214026
2 changed files with 131 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
passerelle/apps/opengis/models.py
View File

@ -18,6 +18,8 @@ import math
import xml.etree.ElementTree as ET
from HTMLParser import HTMLParser
import six
import pyproj
from django.core.cache import cache
@ -59,12 +61,13 @@ class OpenGIS(BaseResource):
projection = models.CharField(_('GIS projection'), choices=PROJECTIONS,
default='EPSG:3857', max_length=16)
search_radius = models.IntegerField(_('Radius for point search'), default=5)
attributes_mapping = (('road', ('road', 'road_name', 'street', 'street_name', 'voie', 'nom_voie', 'rue')),
('city', ('city', 'city_name', 'town', 'town_name', 'commune', 'nom_commune', 'ville', 'nom_ville')),
('house_number', ('house_number', 'number', 'numero', 'numero_voie', 'numero_rue')),
('postcode', ('postcode', 'postalCode', 'zipcode', 'codepostal', 'cp', 'code_postal', 'code_post')),
('country', ('country', 'country_name', 'pays', 'nom_pays'))
)
attributes_mapping = (
('road', ('road', 'road_name', 'street', 'street_name', 'voie', 'nom_voie', 'rue')),
('city', ('city', 'city_name', 'town', 'town_name', 'commune', 'nom_commune', 'ville', 'nom_ville')),
('house_number', ('house_number', 'number', 'numero', 'numero_voie', 'numero_rue')),
('postcode', ('postcode', 'postalCode', 'zipcode', 'codepostal', 'cp', 'code_postal', 'code_post')),
('country', ('country', 'country_name', 'pays', 'nom_pays'))
)
class Meta:
verbose_name = _('OpenGIS')
@ -92,23 +95,51 @@ class OpenGIS(BaseResource):
def check_status(self):
if self.wms_service_url:
response = self.requests.get(self.wms_service_url,
params={'service': 'WMS', 'request': 'GetCapabilities'})
response = self.requests.get(
self.wms_service_url,
params={
'service': 'WMS',
'request': 'GetCapabilities'
})
response.raise_for_status()
if self.wfs_service_url:
response = self.requests.get(self.wfs_service_url,
params={'service': 'WFS', 'request': 'GetCapabilities'})
response = self.requests.get(
self.wfs_service_url,
params={
'service': 'WFS',
'request': 'GetCapabilities'
})
response.raise_for_status()
@endpoint(perm='can_access', description='Get features',
parameters={'type_names': {'description': _('Type of feature to query'), 'example_value':'feature'},
'property_name': {'description': _('Property to list'), 'example_value':'nom_commune'},
'cql_filter': {'description': _('CQL filter applied to the query'), 'example_value':'commune=\'Paris\''},
'filter_property_name': {'description': _('Property by which to filter'), 'example_value':'voie'},
'q': {'description': _('Filter value'), 'example_value':'rue du chateau'},
'case-insensitive': {'description': _('Enables case-insensitive search'), 'example_value':'true'}
})
def features(self, request, type_names, property_name, cql_filter=None, filter_property_name=None, q=None, **kwargs):
parameters={
'type_names': {
'description': _('Type of feature to query'),
'example_value': 'feature'
},
'property_name': {
'description': _('Property to list'),
'example_value': 'nom_commune'
},
'cql_filter': {
'description': _('CQL filter applied to the query'),
'example_value': 'commune=\'Paris\''
},
'filter_property_name': {
'description': _('Property by which to filter'),
'example_value': 'voie'
},
'q': {
'description': _('Filter value'),
'example_value': 'rue du chateau'
},
'case-insensitive': {
'description': _('Enables case-insensitive search'),
'example_value': 'true'
}
})
def features(self, request, type_names, property_name, cql_filter=None,
filter_property_name=None, q=None, **kwargs):
params = {
'VERSION': self.get_wfs_service_version(),
'SERVICE': 'WFS',
@ -128,7 +159,7 @@ class OpenGIS(BaseResource):
response = self.requests.get(self.wfs_service_url, params=params)
data = []
try:
json_result = response.json()
response.json()
except ValueError:
self.handle_opengis_error(response)
# if handle_opengis_error did not raise an error, we raise a generic one
@ -252,9 +283,11 @@ class OpenGIS(BaseResource):
'SRS': self.projection,
'BBOX': bbox,
}
return HttpResponse(
self.requests.get(self.wms_service_url, params=params, cache_duration=300).content,
content_type='image/png')
response = self.requests.get(
self.wms_service_url,
params=params,
cache_duration=300)
return HttpResponse(response.content, content_type='image/png')
@endpoint(perm='can_access', description=_('Get feature info'))
def reverse(self, request, lat, lon, **kwargs):
@ -299,7 +332,7 @@ class OpenGIS(BaseResource):
for attribute, properties in self.attributes_mapping:
for field in properties:
if closest_feature['properties'].get(field):
result['address'][attribute] = unicode(closest_feature['properties'][field])
result['address'][attribute] = six.text_type(closest_feature['properties'][field])
break
return result
raise APIError('Unable to geocode')

92
tests/test_opengis.py
View File

@ -6,7 +6,10 @@ from passerelle.apps.opengis.models import OpenGIS
import utils
FAKE_FEATURE_INFO = '''<?xml version="1.0" encoding="UTF-8"?>
<msGMLOutput xmlns:gml="http://www.opengis.net/gml" xmlns:xlink="http://www.w3.org/1999/xlink" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<msGMLOutput
xmlns:gml="http://www.opengis.net/gml"
xmlns:xlink="http://www.w3.org/1999/xlink"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<cad_cadastre.cadparcelle_layer>
<gml:name>Parcelle cadastrale (Plan cadastral informatise du Grand Lyon)</gml:name>
<cad_cadastre.cadparcelle_feature>
@ -28,7 +31,9 @@ FAKE_FEATURE_INFO = '''<?xml version="1.0" encoding="UTF-8"?>
</msGMLOutput>'''
FAKE_SERVICE_CAPABILITIES = '''<?xml version="1.0" encoding="UTF-8"?>
<wfs:WFS_Capabilities version="2.0.0" xmlns:wfs="http://www.opengis.net/wfs/2.0" xmlns:ows="http://www.opengis.net/ows/1.1">
<wfs:WFS_Capabilities version="2.0.0"
xmlns:wfs="http://www.opengis.net/wfs/2.0"
xmlns:ows="http://www.opengis.net/ows/1.1">
<ows:ServiceIdentification>
<ows:Title/><ows:Abstract/>
<ows:ServiceType>WFS</ows:ServiceType><ows:ServiceTypeVersion>2.0.0</ows:ServiceTypeVersion><ows:Fees/>
@ -99,7 +104,28 @@ FAKE_FEATURES_JSON = '''
"type": "FeatureCollection"
}'''
FAKE_ERROR = u'<ows:ExceptionReport xmlns:xs="http://www.w3.org/2001/XMLSchema" xmlns:ows="http://www.opengis.net/ows/1.1" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" version="2.0.0" xsi:schemaLocation="http://www.opengis.net/ows/1.1 https://sigmetropole.lametro.fr/geoserver/schemas/ows/1.1.0/owsAll.xsd">\n <ows:Exception exceptionCode="NoApplicableCode">\n <ows:ExceptionText>Could not parse CQL filter list.\nEncountered &amp;quot;BIS&amp;quot; at line 1, column 129.\nWas expecting one of:\n &amp;lt;EOF&amp;gt; \n &amp;quot;and&amp;quot; ...\n &amp;quot;or&amp;quot; ...\n &amp;quot;;&amp;quot; ...\n &amp;quot;/&amp;quot; ...\n &amp;quot;*&amp;quot; ...\n &amp;quot;+&amp;quot; ...\n &amp;quot;-&amp;quot; ...\n Parsing : strEqualsIgnoreCase(nom_commune, &amp;apos;Grenoble&amp;apos;) = true AND strEqualsIgnoreCase(nom_voie, &amp;apos;rue albert recoura&amp;apos;) = true AND numero=8 BIS.</ows:ExceptionText>\n </ows:Exception>\n</ows:ExceptionReport>\n'
FAKE_ERROR = u'''<ows:ExceptionReport
xmlns:xs="http://www.w3.org/2001/XMLSchema"
xmlns:ows="http://www.opengis.net/ows/1.1"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
version="2.0.0"
xsi:schemaLocation="http://www.opengis.net/ows/1.1 https://sigmetropole.lametro.fr/geoserver/schemas/ows/1.1.0/owsAll.xsd">
<ows:Exception exceptionCode="NoApplicableCode">
<ows:ExceptionText>Could not parse CQL filter list.
Encountered &amp;quot;BIS&amp;quot; at line 1, column 129.
Was expecting one of:
&amp;lt;EOF&amp;gt;
&amp;quot;and&amp;quot; ...
&amp;quot;or&amp;quot; ...
&amp;quot;;&amp;quot; ...
&amp;quot;/&amp;quot; ...
&amp;quot;*&amp;quot; ...
&amp;quot;+&amp;quot; ...
&amp;quot;-&amp;quot; ...
Parsing : strEqualsIgnoreCase(nom_commune, &amp;apos;Grenoble&amp;apos;) = true AND strEqualsIgnoreCase(nom_voie, &amp;apos;rue albert recoura&amp;apos;) = true AND numero=8 BIS.</ows:ExceptionText>
</ows:Exception>
</ows:ExceptionReport>
'''
FAKE_GEOLOCATED_FEATURE = '''{
"crs": {
@ -195,21 +221,25 @@ def connector(db):
wms_service_url='http://example.net/wms',
wfs_service_url='http://example.net/wfs'))
def geoserver_responses(url, **kwargs):
if kwargs['params'].get('request') == 'GetCapabilities':
return utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)
return utils.FakedResponse(status_code=200, content=FAKE_FEATURES_JSON)
def geoserver_responses_errors(url, **kwargs):
if kwargs['params'].get('request') == 'GetCapabilities':
return utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)
return utils.FakedResponse(status_code=200, content=FAKE_ERROR)
def geoserver_responses_errors_unparsable(url, **kwargs):
if kwargs['params'].get('request') == 'GetCapabilities':
return utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)
return utils.FakedResponse(status_code=200, content=FAKE_ERROR[:10])
@mock.patch('passerelle.utils.Request.get')
def test_feature_info(mocked_get, app, connector):
endpoint = utils.generic_endpoint_url('opengis', 'feature_info', slug=connector.slug)
@ -218,13 +248,18 @@ def test_feature_info(mocked_get, app, connector):
resp = app.get(endpoint, params={'lat': '45.796890', 'lon': '4.784140'})
assert mocked_get.call_args[1]['params']['BBOX'] == '5747860.22776,532568.028684,5748179.56467,532790.667665'
assert mocked_get.call_args[1]['params']['CRS'] == 'EPSG:3857'
assert resp.json['data']['cad_cadastrecadparcelle_layer']['cad_cadastrecadparcelle_feature']['natureproprietaire'] == 'Particulier'
assert (resp.json['data']
['cad_cadastrecadparcelle_layer']
['cad_cadastrecadparcelle_feature']
['natureproprietaire']
== 'Particulier')
connector.projection = 'EPSG:4326'
connector.save()
resp = app.get(endpoint, params={'lat': '45.796890', 'lon': '4.784140'})
assert mocked_get.call_args[1]['params']['BBOX'] == '45.796890,4.784140,45.79889,4.78614'
assert mocked_get.call_args[1]['params']['CRS'] == 'EPSG:4326'
@mock.patch('passerelle.utils.Request.get')
def test_tile(mocked_get, app, connector):
endpoint = utils.generic_endpoint_url('opengis', 'tile', slug=connector.slug)
@ -249,11 +284,11 @@ def test_get_feature_with_no_wfs_url(mocked_get, app, connector):
assert endpoint == '/opengis/test/features'
mocked_get.side_effect = geoserver_responses
resp = app.get(endpoint, params={'type_names': 'ref_metro_limites_communales', 'property_name': 'nom'})
print resp.content
assert resp.json['data'] == None
assert resp.json['data'] is None
assert resp.json['err'] == 1
assert resp.json['err_desc'] == 'no wfs URL declared'
@mock.patch('passerelle.utils.Request.get')
def test_get_feature(mocked_get, app, connector):
endpoint = utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
@ -270,14 +305,20 @@ def test_get_feature(mocked_get, app, connector):
for item in resp.json['data']:
assert 'text' in item
@mock.patch('passerelle.utils.Request.get')
def test_get_filtered_feature(mocked_get, app, connector):
endpoint = utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
mocked_get.side_effect = geoserver_responses
resp = app.get(endpoint, params={'type_names': 'ref_metro_limites_communales',
'property_name': 'nom', 'cql_filter': 'nom=\'Fontaine\''})
app.get(endpoint,
params={
'type_names': 'ref_metro_limites_communales',
'property_name': 'nom',
'cql_filter': 'nom=\'Fontaine\''
})
assert mocked_get.call_args[1]['params']['CQL_FILTER'] == 'nom=\'Fontaine\''
@mock.patch('passerelle.utils.Request.get')
def test_get_filtered_by_property_feature(mocked_get, app, connector):
endpoint = utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
@ -285,18 +326,19 @@ def test_get_filtered_by_property_feature(mocked_get, app, connector):
params = {'type_names': 'ref_metro_limites_communales',
'property_name': 'nom', 'cql_filter': 'nom=\'Fontaine\'',
'filter_property_name': 'nom'}
resp = app.get(endpoint, params=params)
app.get(endpoint, params=params)
assert mocked_get.call_args[1]['params']['CQL_FILTER'] == 'nom=\'Fontaine\''
params['q'] = 'bens'
resp = app.get(endpoint, params=params)
app.get(endpoint, params=params)
assert mocked_get.call_args[1]['params']['CQL_FILTER'] == 'nom=\'Fontaine\' AND nom LIKE \'%bens%\''
params['case-insensitive'] = True
resp = app.get(endpoint, params=params)
app.get(endpoint, params=params)
assert mocked_get.call_args[1]['params']['CQL_FILTER'] == 'nom=\'Fontaine\' AND nom ILIKE \'%bens%\''
params.pop('cql_filter')
resp = app.get(endpoint, params=params)
app.get(endpoint, params=params)
assert 'CQL_FILTER' not in mocked_get.call_args[1]['params']
@mock.patch('passerelle.utils.Request.get')
def test_get_feature_error(mocked_get, app, connector):
endpoint = utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
@ -317,8 +359,9 @@ def test_get_feature_error(mocked_get, app, connector):
assert result['err_desc'] == 'OpenGIS Error: NoApplicableCode'
assert 'Could not parse' in result['data']['text']
@mock.patch('passerelle.utils.Request.get')
def test_get_feature_error(mocked_get, app, connector):
def test_get_feature_error2(mocked_get, app, connector):
endpoint = utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
assert endpoint == '/opengis/test/features'
mocked_get.side_effect = geoserver_responses_errors_unparsable
@ -337,6 +380,7 @@ def test_get_feature_error(mocked_get, app, connector):
assert result['err_desc'] == 'OpenGIS Error: unparsable error'
assert '<ows:' in result['data']['content']
@mock.patch('passerelle.utils.Request.get')
def test_reverse_geocoding(mocked_get, app, connector):
connector.search_radius = 45
@ -345,8 +389,13 @@ def test_reverse_geocoding(mocked_get, app, connector):
endpoint = utils.generic_endpoint_url('opengis', 'reverse', slug=connector.slug)
assert endpoint == '/opengis/test/reverse'
mocked_get.return_value = utils.FakedResponse(content=FAKE_GEOLOCATED_FEATURE, status_code=200)
resp = app.get(endpoint, params={'lat':'45.1893469606986', 'lon': '5.72462060798'})
assert mocked_get.call_args[1]['params']['CQL_FILTER'] == 'DWITHIN(the_geom,Point(1914061.48604 4224640.45779),45,meters)'
resp = app.get(endpoint,
params={
'lat': '45.1893469606986',
'lon': '5.72462060798'
})
assert (mocked_get.call_args[1]['params']['CQL_FILTER']
== 'DWITHIN(the_geom,Point(1914061.48604 4224640.45779),45,meters)')
assert resp.json['lon'] == '5.72407744145'
assert resp.json['lat'] == '45.1893972656'
assert resp.json['address']['house_number'] == '4'
@ -358,12 +407,21 @@ def test_reverse_geocoding(mocked_get, app, connector):
connector.search_radius = 10
connector.save()
mocked_get.return_value = utils.FakedResponse(content='{"features": []}', status_code=200)
resp = app.get(endpoint, params={'lat':'45.183784', 'lon': '5.714885'})
resp = app.get(
endpoint,
params={
'lat': '45.183784',
'lon': '5.714885'
})
assert mocked_get.call_args[1]['params']['CQL_FILTER'] == 'DWITHIN(the_geom,Point(5.714885 45.183784),10,meters)'
assert resp.json['err'] == 1
assert resp.json['err_desc'] == 'Unable to geocode'
mocked_get.return_value = utils.FakedResponse(status_code=404, content='{}', ok=False)
resp = app.get(endpoint, params={'lat':'45.183784', 'lon': '5.714885'})
resp = app.get(endpoint,
params={
'lat': '45.183784',
'lon': '5.714885'
})
assert resp.json['err'] == 1
assert resp.json['err_desc'] == 'Webservice returned status code 404'