From 345709f166fe177d11df2cc6a1645323d8029053 Mon Sep 17 00:00:00 2001 From: Benjamin Dauvergne Date: Fri, 8 Feb 2019 07:27:10 +0100 Subject: [PATCH 03/13] opengis: PEP8ness, code style (#30458) --- passerelle/apps/opengis/models.py | 79 ++++++++++++++++++-------- tests/test_opengis.py | 92 +++++++++++++++++++++++++------ 2 files changed, 131 insertions(+), 40 deletions(-) diff --git a/passerelle/apps/opengis/models.py b/passerelle/apps/opengis/models.py index 078578a..a930398 100644 --- a/passerelle/apps/opengis/models.py +++ b/passerelle/apps/opengis/models.py @@ -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') diff --git a/tests/test_opengis.py b/tests/test_opengis.py index 13fb80d..4f42e33 100644 --- a/tests/test_opengis.py +++ b/tests/test_opengis.py @@ -6,7 +6,10 @@ from passerelle.apps.opengis.models import OpenGIS import utils FAKE_FEATURE_INFO = ''' - + Parcelle cadastrale (Plan cadastral informatise du Grand Lyon) @@ -28,7 +31,9 @@ FAKE_FEATURE_INFO = ''' ''' FAKE_SERVICE_CAPABILITIES = ''' - + WFS2.0.0 @@ -99,7 +104,28 @@ FAKE_FEATURES_JSON = ''' "type": "FeatureCollection" }''' -FAKE_ERROR = u'\n \n Could not parse CQL filter list.\nEncountered "BIS" at line 1, column 129.\nWas expecting one of:\n <EOF> \n "and" ...\n "or" ...\n ";" ...\n "/" ...\n "*" ...\n "+" ...\n "-" ...\n Parsing : strEqualsIgnoreCase(nom_commune, 'Grenoble') = true AND strEqualsIgnoreCase(nom_voie, 'rue albert recoura') = true AND numero=8 BIS.\n \n\n' +FAKE_ERROR = u''' + + Could not parse CQL filter list. +Encountered "BIS" at line 1, column 129. +Was expecting one of: + <EOF> + "and" ... + "or" ... + ";" ... + "/" ... + "*" ... + "+" ... + "-" ... + Parsing : strEqualsIgnoreCase(nom_commune, 'Grenoble') = true AND strEqualsIgnoreCase(nom_voie, 'rue albert recoura') = true AND numero=8 BIS. + + +''' 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 '