import json from unittest import mock import pytest from django.contrib.contenttypes.models import ContentType from django.core.exceptions import ValidationError import tests.utils from passerelle.apps.arcgis.models import ArcGIS, Query, SqlFormatter, validate_where from passerelle.base.models import AccessRight, ApiUser from passerelle.utils import import_site from tests.test_manager import login pytestmark = pytest.mark.django_db # from http://sampleserver1.arcgisonline.com/ArcGIS/rest/services/fold/serv/MapServer/1 STATES = '''{ "fieldAliases" : { "OBJECTID" : "OBJECTID", "STATE_NAME" : "STATE_NAME", "STATE_ABBR" : "STATE_ABBR" }, "features" : [ { "attributes" : { "STATE_NAME" : "Texas", "STATE_ABBR" : "TX", "OBJECTID" : 40 }, "geometry" : { "rings" : [ [ [-105.998886788462, 31.3939400524361], [-106.21328556164, 31.4782464373727] ] ] } }, { "geometry" : { "rings" : [ [ [-111.475425113078, 44.7021622250113], [-111.480804007084, 44.6914159859524] ] ] }, "attributes" : { "STATE_NAME" : "Montana", "STATE_ABBR" : "MT", "OBJECTID" : 2 } } ], "spatialReference" : { "wkid" : 4326 }, "fields" : [ { "alias" : "OBJECTID", "type" : "esriFieldTypeOID", "name" : "OBJECTID" }, { "type" : "esriFieldTypeString", "alias" : "STATE_NAME", "length" : 25, "name" : "STATE_NAME" }, { "length" : 2, "alias" : "STATE_ABBR", "type" : "esriFieldTypeString", "name" : "STATE_ABBR" } ], "geometryType" : "esriGeometryPolygon", "displayFieldName" : "STATE_NAME" }''' FEATURES = '''{ "features": [ { "attributes": { "nom": "Le Nom", "id": "onze", "objectid": 11 }, "geometry" : { "rings" : [ [ [-111.475425113078, 44.7021622250113], [-111.480804007084, 44.6914159859524] ] ] } }, { "attributes": { "nom": "Autre Nom", "id": "dix", "objectid": 10 }, "geometry" : { "rings" : [ [ [-111.475425113078, 44.7021622250113], [-111.480804007084, 44.6914159859524] ] ] } } ], "fields": [ { "alias": "OBJECTID", "name": "objectid", "type": "esriFieldTypeOID" }, { "alias": "quartier", "name": "nom", "type": "esriFieldTypeString" }, { "alias": "code", "length": 5, "name": "id", "type": "esriFieldTypeString" } ], "geometryType": "esriGeometryPolygon", "globalIdFieldName": "", "objectIdFieldName": "objectid", "spatialReference": { "latestWkid": 3947, "wkid": 3947 } }''' INFO = '''{ "authInfo": { "isTokenBasedSecurity": true, "tokenServicesUrl": "https://arcgis/portal/sharing/rest/generateToken" }, "currentVersion": 10.81, "fullVersion": "10.8.1", "owningSystemUrl": "https://arcgis/portal", "secureSoapUrl": null, "soapUrl": "https://arcgis/arcgis/services" }''' TOKEN = '''{ "expires": 1649761676673, "ssl": true, "token": "tok42" }''' ERRORS = [ ({"error": {"message": "crash message"}}, 'crash message'), ({"error": {"foo": "bar"}}, 'unknown ArcGIS/token error'), ({"error": "string crash"}, 'unknown ArcGIS/token error'), ] @pytest.fixture def arcgis(): return ArcGIS.objects.create(slug='test', base_url='https://arcgis.example.net/') def test_arcgis_mapservice_query(app, arcgis): endpoint = tests.utils.generic_endpoint_url('arcgis', 'mapservice-query', slug=arcgis.slug) assert endpoint == '/arcgis/test/mapservice-query' params = {'folder': 'fold', 'service': 'serv', 'layer': '1'} with mock.patch('passerelle.utils.Request.get') as requests_get: requests_get.return_value = tests.utils.FakedResponse(content=STATES, status_code=200) resp = app.get(endpoint, params=params, status=403) assert requests_get.call_count == 0 assert resp.json['err'] == 1 assert resp.json['err_class'] == 'django.core.exceptions.PermissionDenied' # open access api = ApiUser.objects.create(username='all', keytype='', key='') obj_type = ContentType.objects.get_for_model(arcgis) AccessRight.objects.create( codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=arcgis.pk ) resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 1 assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/MapServer/1/query' ) args = requests_get.call_args[1]['params'] assert args['f'] == 'json' assert args['outFields'] == '*' assert args['where'] == '1=1' assert 'data' in resp.json assert resp.json['err'] == 0 assert len(resp.json['data']) == 2 assert resp.json['data'][0]['id'] == '40' assert resp.json['data'][0]['text'] == 'Texas' assert 'geometry' not in resp.json['data'][0] assert 'metadata' not in resp.json params['full'] = 'on' resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 2 assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/MapServer/1/query' ) args = requests_get.call_args[1]['params'] assert args['f'] == 'json' assert args['outFields'] == '*' assert args['where'] == '1=1' assert 'data' in resp.json assert resp.json['err'] == 0 assert len(resp.json['data']) == 2 assert resp.json['data'][0]['id'] == '40' assert resp.json['data'][0]['text'] == 'Texas' assert resp.json['data'][0]['geometry'] assert resp.json['metadata'] params['q'] = 'Texas' resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 3 assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/MapServer/1/query' ) args = requests_get.call_args[1]['params'] assert args['text'] == 'Texas' assert 'where' not in args params['lat'] = '9.87654' params['lon'] = '1.12345' resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 4 assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/MapServer/1/query' ) args = requests_get.call_args[1]['params'] assert args['geometry'] == '1.12345,9.87654' assert args['geometryType'] == 'esriGeometryPoint' del params['lat'] # missing lat, do not search by geometry resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 5 assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/MapServer/1/query' ) args = requests_get.call_args[1]['params'] assert 'geometry' not in args assert 'geometryType' not in args params.update({'latmin': '1', 'lonmin': '2', 'latmax': '3', 'lonmax': '4'}) resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 6 assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/MapServer/1/query' ) args = requests_get.call_args[1]['params'] assert args['geometry'] == '2.0,1.0,4.0,3.0' assert args['geometryType'] == 'esriGeometryEnvelope' del params['latmin'] # incomplete box, do not search by geometry resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 7 assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/MapServer/1/query' ) args = requests_get.call_args[1]['params'] assert 'geometry' not in args assert 'geometryType' not in args # others params are directly sent to ArcGIS params['spatialRel'] = 'esriSpatialRelContains' params.update({'latmin': '1', 'lonmin': '2', 'latmax': '3', 'lonmax': '4'}) resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 8 assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/MapServer/1/query' ) args = requests_get.call_args[1]['params'] assert args['geometry'] == '2.0,1.0,4.0,3.0' assert args['geometryType'] == 'esriGeometryEnvelope' assert args['spatialRel'] == 'esriSpatialRelContains' # folder params['folder'] = 'foo/bar' resp = app.get(endpoint, params=params, status=200) assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/foo/bar/serv/MapServer/1/query' ) del params['folder'] resp = app.get(endpoint, params=params, status=200) assert requests_get.call_args[0][0] == 'https://arcgis.example.net/services/serv/MapServer/1/query' # minimal call resp = app.get(endpoint, params={'service': 'srv'}, status=200) assert requests_get.call_args[0][0] == 'https://arcgis.example.net/services/srv/MapServer/0/query' args = requests_get.call_args[1]['params'] assert args == {'f': 'json', 'inSR': '4326', 'outSR': '4326', 'outFields': '*', 'where': '1=1'} # distance resp = app.get(endpoint, params={'service': 'srv', 'distance': '100'}, status=200) assert requests_get.call_args[0][0] == 'https://arcgis.example.net/services/srv/MapServer/0/query' args = requests_get.call_args[1]['params'] assert args['distance'] == '100' assert args['units'] == 'esriSRUnit_Meter' # default unit resp = app.get( endpoint, params={'service': 'srv', 'distance': '5', 'units': 'esriSRUnit_NauticalMile'}, status=200, ) assert requests_get.call_args[0][0] == 'https://arcgis.example.net/services/srv/MapServer/0/query' args = requests_get.call_args[1]['params'] assert args['distance'] == '5' assert args['units'] == 'esriSRUnit_NauticalMile' # call errors with mock.patch('passerelle.utils.Request.get') as requests_get: requests_get.return_value = tests.utils.FakedResponse(content=STATES, status_code=200) resp = app.get(endpoint, params={}, status=400) assert requests_get.call_count == 0 assert resp.json['err'] == 1 assert resp.json['err_class'] == 'passerelle.views.WrongParameter' assert resp.json['err_desc'] == "missing parameters: 'service'." resp = app.get(endpoint, params={'service': 'src', 'lat': '0', 'lon': 'y'}, status=400) assert requests_get.call_count == 0 assert resp.json['err'] == 1 assert resp.json['err_class'] == 'passerelle.utils.jsonresponse.APIError' assert resp.json['err_desc'] == ' and must be floats' resp = app.get( endpoint, params={'service': 'src', 'latmin': '0', 'lonmin': 'y', 'latmax': '0', 'lonmax': '1'}, status=400, ) assert requests_get.call_count == 0 assert resp.json['err'] == 1 assert resp.json['err_class'] == 'passerelle.utils.jsonresponse.APIError' assert resp.json['err_desc'] == ' and must be floats' def test_arcgis_featureservice_query(app, arcgis): endpoint = tests.utils.generic_endpoint_url('arcgis', 'featureservice-query', slug=arcgis.slug) assert endpoint == '/arcgis/test/featureservice-query' params = {'folder': 'fold', 'service': 'serv', 'layer': '42'} with mock.patch('passerelle.utils.Request.get') as requests_get: requests_get.return_value = tests.utils.FakedResponse(content=FEATURES, status_code=200) resp = app.get(endpoint, params=params, status=403) assert requests_get.call_count == 0 assert resp.json['err'] == 1 assert resp.json['err_class'] == 'django.core.exceptions.PermissionDenied' # open access api = ApiUser.objects.create(username='all', keytype='', key='') obj_type = ContentType.objects.get_for_model(arcgis) AccessRight.objects.create( codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=arcgis.pk ) resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 1 assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/FeatureServer/42/query' ) args = requests_get.call_args[1]['params'] assert args['f'] == 'json' assert args['outFields'] == '*' assert 'data' in resp.json assert resp.json['err'] == 0 assert len(resp.json['data']) == 2 assert resp.json['data'][0]['id'] == '11' assert resp.json['data'][0]['text'] == '11' assert 'geometry' not in resp.json['data'][0] assert 'metadata' not in resp.json params['full'] = 'on' resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 2 assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/FeatureServer/42/query' ) args = requests_get.call_args[1]['params'] assert args['f'] == 'json' assert args['outFields'] == '*' assert 'data' in resp.json assert resp.json['err'] == 0 assert len(resp.json['data']) == 2 assert resp.json['data'][0]['id'] == '11' assert resp.json['data'][0]['text'] == '11' assert resp.json['data'][0]['geometry'] assert resp.json['metadata'] params['lat'] = '9.87654' params['lon'] = '1.12345' resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 3 assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/FeatureServer/42/query' ) args = requests_get.call_args[1]['params'] assert args['geometry'] == '1.12345,9.87654' assert args['geometryType'] == 'esriGeometryPoint' del params['lat'] # missing lat, do not search by geometry resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 4 assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/FeatureServer/42/query' ) args = requests_get.call_args[1]['params'] assert 'geometry' not in args assert 'geometryType' not in args params.update({'latmin': '1', 'lonmin': '2', 'latmax': '3', 'lonmax': '4'}) resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 5 assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/FeatureServer/42/query' ) args = requests_get.call_args[1]['params'] assert args['geometry'] == '2.0,1.0,4.0,3.0' assert args['geometryType'] == 'esriGeometryEnvelope' del params['latmin'] # incomplete box, do not search by geometry resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 6 assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/FeatureServer/42/query' ) args = requests_get.call_args[1]['params'] assert 'geometry' not in args assert 'geometryType' not in args # others params are directly sent to ArcGIS params['spatialRel'] = 'esriSpatialRelContains' params.update({'latmin': '1', 'lonmin': '2', 'latmax': '3', 'lonmax': '4'}) resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 7 assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/FeatureServer/42/query' ) args = requests_get.call_args[1]['params'] assert args['geometry'] == '2.0,1.0,4.0,3.0' assert args['geometryType'] == 'esriGeometryEnvelope' assert args['spatialRel'] == 'esriSpatialRelContains' # folder params['folder'] = 'foo/bar' resp = app.get(endpoint, params=params, status=200) assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/foo/bar/serv/FeatureServer/42/query' ) del params['folder'] resp = app.get(endpoint, params=params, status=200) assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/serv/FeatureServer/42/query' ) # minimal call resp = app.get(endpoint, params={'service': 'srv'}, status=200) assert requests_get.call_args[0][0] == 'https://arcgis.example.net/services/srv/FeatureServer/0/query' args = requests_get.call_args[1]['params'] assert args == {'f': 'json', 'inSR': '4326', 'outSR': '4326', 'outFields': '*'} # distance resp = app.get(endpoint, params={'service': 'srv', 'distance': '100'}, status=200) assert requests_get.call_args[0][0] == 'https://arcgis.example.net/services/srv/FeatureServer/0/query' args = requests_get.call_args[1]['params'] assert args['distance'] == '100' assert args['units'] == 'esriSRUnit_Meter' # default unit resp = app.get( endpoint, params={'service': 'srv', 'distance': '5', 'units': 'esriSRUnit_NauticalMile'}, status=200, ) assert requests_get.call_args[0][0] == 'https://arcgis.example.net/services/srv/FeatureServer/0/query' args = requests_get.call_args[1]['params'] assert args['distance'] == '5' assert args['units'] == 'esriSRUnit_NauticalMile' # text_fieldname for text_fieldname in ('nom', 'quartier'): # 'nom': real name, 'quartier': its alias resp = app.get( endpoint, params={ 'service': 'srv', 'text_fieldname': text_fieldname, }, status=200, ) assert 'data' in resp.json assert resp.json['err'] == 0 assert len(resp.json['data']) == 2 assert resp.json['data'][0]['id'] == '11' assert resp.json['data'][0]['text'] == 'Le Nom' assert resp.json['data'][1]['id'] == '10' assert resp.json['data'][1]['text'] == 'Autre Nom' # templates resp = app.get( endpoint, params={ 'service': 'srv', 'template': '{{ attributes.nom }}', 'id_template': '{{ attributes.id }}', }, status=200, ) assert 'data' in resp.json assert resp.json['err'] == 0 assert len(resp.json['data']) == 2 assert resp.json['data'][0]['id'] == 'onze' assert resp.json['data'][0]['text'] == 'Le Nom' assert resp.json['data'][1]['id'] == 'dix' assert resp.json['data'][1]['text'] == 'Autre Nom' # call errors with mock.patch('passerelle.utils.Request.get') as requests_get: requests_get.return_value = tests.utils.FakedResponse(content=FEATURES, status_code=200) resp = app.get(endpoint, params={}, status=400) assert requests_get.call_count == 0 assert resp.json['err'] == 1 assert resp.json['err_class'] == 'passerelle.views.WrongParameter' assert resp.json['err_desc'] == "missing parameters: 'service'." resp = app.get(endpoint, params={'service': 'src', 'lat': '0', 'lon': 'y'}, status=400) assert requests_get.call_count == 0 assert resp.json['err'] == 1 assert resp.json['err_class'] == 'passerelle.utils.jsonresponse.APIError' assert resp.json['err_desc'] == ' and must be floats' resp = app.get( endpoint, params={'service': 'src', 'latmin': '0', 'lonmin': 'y', 'latmax': '0', 'lonmax': '1'}, status=400, ) assert requests_get.call_count == 0 assert resp.json['err'] == 1 assert resp.json['err_class'] == 'passerelle.utils.jsonresponse.APIError' assert resp.json['err_desc'] == ' and must be floats' def test_arcgis_with_token_query(app, arcgis): endpoint = tests.utils.generic_endpoint_url('arcgis', 'featureservice-query', slug=arcgis.slug) assert endpoint == '/arcgis/test/featureservice-query' # open access api = ApiUser.objects.create(username='all', keytype='', key='') obj_type = ContentType.objects.get_for_model(arcgis) AccessRight.objects.create( codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=arcgis.pk ) arcgis.token_username = 'tokenuser' arcgis.token_password = 'tokenpass' arcgis.save() params = {'folder': 'fold', 'service': 'serv', 'layer': '42'} with mock.patch('passerelle.utils.Request.get') as requests_get: with mock.patch('passerelle.utils.Request.post') as requests_post: requests_get.side_effect = [ tests.utils.FakedResponse(content=INFO, status_code=200), tests.utils.FakedResponse(content=FEATURES, status_code=200), ] requests_post.side_effect = [ tests.utils.FakedResponse(content=TOKEN, status_code=200), ] resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 2 # info + featureservice-query assert requests_post.call_count == 1 # token assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/FeatureServer/42/query' ) args = requests_get.call_args[1]['params'] assert args['f'] == 'json' assert args['outFields'] == '*' assert args['token'] == 'tok42' # token added in query assert 'data' in resp.json assert resp.json['err'] == 0 assert len(resp.json['data']) == 2 assert resp.json['data'][0]['id'] == '11' assert resp.json['data'][0]['text'] == '11' # bad info/token responses requests_get.reset_mock() requests_post.reset_mock() requests_get.side_effect = [ tests.utils.FakedResponse(content='{}', status_code=200), ] resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 1 assert requests_post.call_count == 0 assert resp.json['err'] == 1 assert resp.json['err_desc'].startswith( 'ArcGIS/token responded no authInfo/tokenServicesUrl in info' ) requests_get.reset_mock() requests_get.side_effect = [ tests.utils.FakedResponse(content='CRASH', status_code=500), ] resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 1 assert requests_post.call_count == 0 assert resp.json['err'] == 1 assert resp.json['err_desc'] == 'ArcGIS/token returned status code 500' requests_get.reset_mock() requests_get.side_effect = [ tests.utils.FakedResponse(content='not json', status_code=200), ] resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 1 assert requests_post.call_count == 0 assert resp.json['err'] == 1 assert resp.json['err_desc'].startswith('ArcGIS/token returned invalid JSON content:') requests_get.reset_mock() requests_get.side_effect = [ tests.utils.FakedResponse(content='"not a dict"', status_code=200), ] resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 1 assert requests_post.call_count == 0 assert resp.json['err'] == 1 assert resp.json['err_desc'].startswith('ArcGIS/token not returned a dict:') for content, err_desc in ERRORS: requests_get.reset_mock() requests_get.side_effect = [ tests.utils.FakedResponse(content=json.dumps(content), status_code=200), ] resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 1 assert requests_post.call_count == 0 assert resp.json['err'] == 1 assert resp.json['err_desc'] == err_desc requests_get.reset_mock() requests_post.reset_mock() requests_get.side_effect = [ tests.utils.FakedResponse(content=INFO, status_code=200), ] requests_post.side_effect = [ tests.utils.FakedResponse(content="{}", status_code=200), # no token ] resp = app.get(endpoint, params=params, status=200) assert requests_get.call_count == 1 assert requests_post.call_count == 1 assert resp.json['err'] == 1 assert resp.json['err_desc'].startswith('ArcGIS/token returned no token:') @pytest.mark.parametrize( 'format_string,fail', [ ('x {é}', True), ('x {aa.bb}', True), ('x {a:s} {b:d}', False), ], ) def test_validate_where(format_string, fail): if fail: with pytest.raises(ValidationError): validate_where(format_string) else: validate_where(format_string) @pytest.mark.parametrize( 'format_string,kwargs,expected', [ ('adresse LIKE {adresse:s}', {'adresse': "AVENUE D'ANNAM"}, "adresse LIKE 'AVENUE D''ANNAM'"), ( 'adresse LIKE {adresse:s} AND population < {pop:d}', { 'adresse': "AVENUE D'ANNAM", 'pop': '34', }, "adresse LIKE 'AVENUE D''ANNAM' AND population < 34", ), ( 'adresse LIKE {adresse:s} AND population < {pop:d}', { 'adresse': "AVENUE D'ANNAM", 'pop': 'x', }, ValueError, ), ( 'ID IN {ids:l}', { 'ids': '1,3,5', }, "ID IN ('1', '3', '5')", ), ], ) def test_sql_formatter(format_string, kwargs, expected): formatter = SqlFormatter() if not isinstance(expected, type) or not issubclass(expected, Exception): assert formatter.format(format_string, **kwargs) == expected else: with pytest.raises(expected): formatter.format(format_string, **kwargs) @pytest.fixture def query(arcgis): return Query.objects.create( resource=arcgis, name='Adresses', slug='adresses', description='Rechercher une adresse', id_template='{{ attributes.ident }}', text_template='{{ attributes.address }} - {{ attributes.codepost }}', folder='fold', layer='1', service='serv', where='adress LIKE {adress:s}', ) def test_query_q_method(arcgis, query, rf): arcgis_response = { "features": [ { "attributes": { "ident": "1234", "address": "rue du calvaire", "codepost": 13200, }, 'geo': {}, }, ], 'meta': {}, } with mock.patch('passerelle.utils.Request.get') as requests_get: requests_get.return_value = tests.utils.FakedResponse( content=json.dumps(arcgis_response), status_code=200 ) assert query.q(rf.get('/', data={'adress': "AVENUE D'ANNAM"}), full=True) == { "data": [ { "attributes": {"ident": "1234", "address": "rue du calvaire", "codepost": 13200}, "geo": {}, "id": "1234", "text": "rue du calvaire - 13200", } ], "metadata": {"meta": {}}, } assert requests_get.call_count == 1 assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/MapServer/1/query' ) args = requests_get.call_args[1]['params'] assert args == { 'f': 'json', 'inSR': '4326', 'outSR': '4326', 'outFields': '*', 'where': "adress LIKE 'AVENUE D''ANNAM'", } def test_q_endpoint(arcgis, query, app): endpoint = tests.utils.generic_endpoint_url('arcgis', 'q/adresses/', slug=arcgis.slug) assert endpoint == '/arcgis/test/q/adresses/' with mock.patch('passerelle.utils.Request.get') as requests_get: requests_get.return_value = tests.utils.FakedResponse(content=STATES, status_code=200) resp = app.get(endpoint, params={}, status=403) assert requests_get.call_count == 0 assert resp.json['err'] == 1 assert resp.json['err_class'] == 'django.core.exceptions.PermissionDenied' # open access api = ApiUser.objects.create(username='all', keytype='', key='') obj_type = ContentType.objects.get_for_model(arcgis) AccessRight.objects.create( codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=arcgis.pk ) resp = app.get(endpoint, params={'adress': "AVENUE D'ANNAM"}, status=200) assert requests_get.call_count == 1 assert ( requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/MapServer/1/query' ) args = requests_get.call_args[1]['params'] assert args == { 'f': 'json', 'inSR': '4326', 'outSR': '4326', 'outFields': '*', 'where': "adress LIKE 'AVENUE D''ANNAM'", } def test_tile_endpoint(arcgis, app): assert arcgis.base_url == 'https://arcgis.example.net/' with mock.patch('passerelle.utils.Request.get') as requests_get: requests_get.return_value = tests.utils.FakedResponse(content='', status_code=200) resp = app.get('/arcgis/test/tile/layer1/13/4258/2991.png') assert requests_get.call_count == 1 assert requests_get.call_args[0][0] == ( 'https://arcgis.example.net/layer1/MapServer/export' '?dpi=96&format=png24&bboxSR=4326&imageSR=3857&transparent=true&size=256,256&f=image' '&bbox=7.119141,43.612217,7.163086,43.580391' ) assert resp.content_type == 'image/png' # test layer and folders with mock.patch('passerelle.utils.Request.get') as requests_get: requests_get.return_value = tests.utils.FakedResponse(content='', status_code=200) resp = app.get('/arcgis/test/tile/layer1/foo/bar/13/4258/2991.png') assert requests_get.call_count == 1 assert requests_get.call_args[0][0] == ( 'https://arcgis.example.net/layer1/foo/bar/MapServer/export' '?dpi=96&format=png24&bboxSR=4326&imageSR=3857&transparent=true&size=256,256&f=image' '&bbox=7.119141,43.612217,7.163086,43.580391' ) assert resp.content_type == 'image/png' # test missing trailing slash arcgis.base_url = 'https://arcgis.example.net' arcgis.save() with mock.patch('passerelle.utils.Request.get') as requests_get: requests_get.return_value = tests.utils.FakedResponse(content='', status_code=200) resp = app.get('/arcgis/test/tile/layer1/13/4258/2991.png') assert requests_get.call_count == 1 assert requests_get.call_args[0][0] == ( 'https://arcgis.example.net/layer1/MapServer/export' '?dpi=96&format=png24&bboxSR=4326&imageSR=3857&transparent=true&size=256,256&f=image' '&bbox=7.119141,43.612217,7.163086,43.580391' ) assert resp.content_type == 'image/png' def test_query_documentation(arcgis, query, app): resp = app.get(arcgis.get_absolute_url()) assert query.name in resp.text assert query.description in resp.text # additional parameter appears in endpoint documentation assert 'adress' in resp.text def test_arcgis_query_unicity(admin_user, app, arcgis): query = Query.objects.create( resource=arcgis, name='Test Query', slug='test-query', ) arcgis2 = ArcGIS.objects.create(slug='test2', base_url='https://arcgis.example.net/') Query.objects.create( resource=arcgis2, name='Foo Bar', slug='foo-bar', ) app = login(app) resp = app.get('/manage/arcgis/%s/query/new/' % arcgis.slug) resp.form['slug'] = query.slug resp.form['name'] = 'Foo Bar' resp.form['service'] = 'Foo' resp = resp.form.submit() assert resp.status_code == 200 assert Query.objects.filter(resource=arcgis).count() == 1 assert 'A query with this slug already exists' in resp.text resp.form['slug'] = 'foo-bar' resp.form['name'] = query.name resp.form['service'] = 'Foo' resp = resp.form.submit() assert Query.objects.filter(resource=arcgis).count() == 1 assert resp.status_code == 200 assert 'A query with this name already exists' in resp.text resp.form['slug'] = 'foo-bar' resp.form['name'] = 'Foo Bar' resp.form['service'] = 'Foo' resp = resp.form.submit() assert resp.status_code == 302 assert Query.objects.filter(resource=arcgis).count() == 2 new_query = Query.objects.latest('pk') assert new_query.resource == arcgis resp = app.get('/manage/arcgis/%s/query/%s/' % (arcgis.slug, new_query.pk)) resp.form['slug'] = query.slug resp.form['name'] = 'Foo Bar' resp = resp.form.submit() assert resp.status_code == 200 assert 'A query with this slug already exists' in resp.text resp.form['slug'] = 'foo-bar' resp.form['name'] = query.name resp = resp.form.submit() assert resp.status_code == 200 assert 'A query with this name already exists' in resp.text resp.form['slug'] = 'foo-bar' resp.form['name'] = 'Foo Bar' resp = resp.form.submit() assert resp.status_code == 302 def test_export_import(query): assert ArcGIS.objects.count() == 1 assert Query.objects.count() == 1 serialization = {'resources': [query.resource.export_json()]} ArcGIS.objects.all().delete() assert ArcGIS.objects.count() == 0 assert Query.objects.count() == 0 import_site(serialization) assert ArcGIS.objects.count() == 1 assert Query.objects.count() == 1