523 lines
19 KiB
Python
523 lines
19 KiB
Python
# -*- coding: utf-8 -*-
|
|
from __future__ import unicode_literals
|
|
|
|
import json
|
|
|
|
import mock
|
|
import pytest
|
|
import utils
|
|
from django.contrib.auth.models import User
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.core.exceptions import ValidationError
|
|
from test_manager import login
|
|
|
|
from passerelle.apps.arcgis.models import ArcGIS, Query, SqlFormatter, validate_where
|
|
from passerelle.base.models import AccessRight, ApiUser
|
|
from passerelle.utils import import_site
|
|
|
|
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"
|
|
}'''
|
|
|
|
|
|
@pytest.fixture
|
|
def arcgis():
|
|
return ArcGIS.objects.create(slug='test', base_url='https://arcgis.example.net/')
|
|
|
|
|
|
def test_arcgis_mapservice_query(app, arcgis):
|
|
endpoint = 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 = 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 = 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'] == '<lon> and <lat> 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'] == '<lonmin> <latmin> <lonmax> and <latmax> must be floats'
|
|
|
|
|
|
@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,
|
|
),
|
|
],
|
|
)
|
|
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 = 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 = 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 = 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 = 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 = 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 = 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 '<span class="param-name">adress</span>' 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
|