434 lines
17 KiB
Python
434 lines
17 KiB
Python
import mock
|
|
import pytest
|
|
|
|
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">
|
|
<cad_cadastre.cadparcelle_layer>
|
|
<gml:name>Parcelle cadastrale (Plan cadastral informatise du Grand Lyon)</gml:name>
|
|
<cad_cadastre.cadparcelle_feature>
|
|
<gml:boundedBy>
|
|
<gml:Box srsName="EPSG:4171">
|
|
<gml:coordinates>4.784140,45.796890 4.784834,45.797365</gml:coordinates>
|
|
</gml:Box>
|
|
</gml:boundedBy>
|
|
<identifiant>69040BD309</identifiant>
|
|
<codedgi>040000BD0309</codedgi>
|
|
<numero>309</numero>
|
|
<surfacecadastrale>2406</surfacecadastrale>
|
|
<natureproprietaire>Particulier</natureproprietaire>
|
|
<indice>Parcelle figuree au plan</indice>
|
|
<arpentage>Arpentee</arpentage>
|
|
<gid>75404</gid>
|
|
</cad_cadastre.cadparcelle_feature>
|
|
</cad_cadastre.cadparcelle_layer>
|
|
</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">
|
|
<ows:ServiceIdentification>
|
|
<ows:Title/><ows:Abstract/>
|
|
<ows:ServiceType>WFS</ows:ServiceType><ows:ServiceTypeVersion>2.0.0</ows:ServiceTypeVersion><ows:Fees/>
|
|
<ows:AccessConstraints/>
|
|
</ows:ServiceIdentification>
|
|
</wfs:WFS_Capabilities>'''
|
|
|
|
FAKE_FEATURES_JSON = '''
|
|
{
|
|
"features": [
|
|
{
|
|
"geometry": null,
|
|
"id": "ref_metro_limites_communales.fid--204aa923_15ffdce8d91_-27be",
|
|
"properties": {
|
|
"nom": "Bri\u00e9-et-Angonnes"
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": null,
|
|
"id": "ref_metro_limites_communales.fid--204aa923_15ffdce8d91_-27bd",
|
|
"properties": {
|
|
"nom": "Champagnier"
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": null,
|
|
"id": "ref_metro_limites_communales.fid--204aa923_15ffdce8d91_-27bb",
|
|
"properties": {
|
|
"nom": "Claix"
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": null,
|
|
"id": "ref_metro_limites_communales.fid--204aa923_15ffdce8d91_-27ba",
|
|
"properties": {
|
|
"nom": "Corenc"
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": null,
|
|
"id": "ref_metro_limites_communales.fid--204aa923_15ffdce8d91_-27b9",
|
|
"properties": {
|
|
"nom": "\u00c9chirolles"
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": null,
|
|
"id": "ref_metro_limites_communales.fid--204aa923_15ffdce8d91_-27b8",
|
|
"properties": {
|
|
"nom": "Eybens"
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": null,
|
|
"id": "ref_metro_limites_communales.fid--204aa923_15ffdce8d91_-27b7",
|
|
"properties": {
|
|
"nom": "Fontaine"
|
|
},
|
|
"type": "Feature"
|
|
}
|
|
],
|
|
"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">
|
|
<ows:Exception exceptionCode="NoApplicableCode">
|
|
<ows:ExceptionText>Could not parse CQL filter list.
|
|
Encountered &quot;BIS&quot; at line 1, column 129.
|
|
Was expecting one of:
|
|
&lt;EOF&gt;
|
|
&quot;and&quot; ...
|
|
&quot;or&quot; ...
|
|
&quot;;&quot; ...
|
|
&quot;/&quot; ...
|
|
&quot;*&quot; ...
|
|
&quot;+&quot; ...
|
|
&quot;-&quot; ...
|
|
Parsing : strEqualsIgnoreCase(nom_commune, &apos;Grenoble&apos;) = true AND strEqualsIgnoreCase(nom_voie, &apos;rue albert recoura&apos;) = true AND numero=8 BIS.</ows:ExceptionText>
|
|
</ows:Exception>
|
|
</ows:ExceptionReport>
|
|
'''
|
|
|
|
FAKE_GEOLOCATED_FEATURE = '''{
|
|
"crs": {
|
|
"properties": {
|
|
"name": "urn:ogc:def:crs:EPSG::3945"
|
|
},
|
|
"type": "name"
|
|
},
|
|
"features": [
|
|
{
|
|
"geometry": {
|
|
"coordinates": [
|
|
1914059.51,
|
|
4224699.2
|
|
],
|
|
"type": "Point"
|
|
},
|
|
"geometry_name": "the_geom",
|
|
"properties": {
|
|
"code_insee": 38185,
|
|
"code_post": 38000,
|
|
"nom_afnor": "BOULEVARD EDOUARD REY",
|
|
"nom_commune": "Grenoble",
|
|
"nom_voie": "boulevard \u00e9douard rey",
|
|
"numero": 17
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": {
|
|
"coordinates": [
|
|
1914042.47,
|
|
4224665.2
|
|
],
|
|
"type": "Point"
|
|
},
|
|
"geometry_name": "the_geom",
|
|
"properties": {
|
|
"code_insee": 38185,
|
|
"code_post": 38000,
|
|
"nom_commune": "Grenoble",
|
|
"nom_voie": "place victor hugo",
|
|
"numero": 2
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": {
|
|
"coordinates": [
|
|
1914035.7,
|
|
4224700.42
|
|
],
|
|
"type": "Point"
|
|
},
|
|
"geometry_name": "the_geom",
|
|
"properties": {
|
|
"code_insee": 38185,
|
|
"code_post": 38000,
|
|
"nom_commune": "Grenoble",
|
|
"nom_voie": "boulevard \u00e9douard rey",
|
|
"numero": 28
|
|
},
|
|
"type": "Feature"
|
|
},
|
|
{
|
|
"geometry": {
|
|
"coordinates": [
|
|
1914018.64,
|
|
4224644.61
|
|
],
|
|
"type": "Point"
|
|
},
|
|
"geometry_name": "the_geom",
|
|
"properties": {
|
|
"code_insee": 38185,
|
|
"code_post": 38000,
|
|
"nom_commune": "Grenoble",
|
|
"nom_voie": "place victor hugo",
|
|
"numero": 4
|
|
},
|
|
"type": "Feature"
|
|
}
|
|
],
|
|
"totalFeatures": 4,
|
|
"type": "FeatureCollection"
|
|
}'''
|
|
|
|
|
|
@pytest.fixture
|
|
def connector(db):
|
|
return utils.setup_access_rights(OpenGIS.objects.create(
|
|
slug='test',
|
|
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)
|
|
assert endpoint == '/opengis/test/feature_info'
|
|
mocked_get.return_value = utils.FakedResponse(content=FAKE_FEATURE_INFO, status_code=200)
|
|
resp = app.get(endpoint, params={'lat': '45.796890', 'lon': '4.784140'})
|
|
assert mocked_get.call_args[1]['params']['BBOX'] == '532556.896735,5747844.26121,532579.160633,5747876.19433'
|
|
assert mocked_get.call_args[1]['params']['CRS'] == 'EPSG:3857'
|
|
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.79679,4.78404,45.79699,4.78424'
|
|
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)
|
|
assert endpoint == '/opengis/test/tile'
|
|
mocked_get.return_value = utils.FakedResponse(content='\x89PNG\r\n\x1a\n\x00\x00...', status_code=200)
|
|
resp = app.get(endpoint + '/16/33650/23378.png')
|
|
assert mocked_get.call_args[1]['params']['CRS'] == 'EPSG:3857'
|
|
assert mocked_get.call_args[1]['params']['BBOX'] == '539339.67158,5741338.06856,539951.167806,5741949.56478'
|
|
connector.projection = 'EPSG:4326'
|
|
connector.save()
|
|
resp = app.get(endpoint + '/16/33650/23378.png')
|
|
assert mocked_get.call_args[1]['params']['CRS'] == 'EPSG:4326'
|
|
assert mocked_get.call_args[1]['params']['BBOX'] == '45.7560261559,4.84497070312,45.7598586879,4.85046386719'
|
|
assert resp.content == '\x89PNG\r\n\x1a\n\x00\x00...'
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_get_feature_with_no_wfs_url(mocked_get, app, connector):
|
|
connector.wfs_service_url = ''
|
|
connector.save()
|
|
endpoint = utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
|
|
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'})
|
|
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)
|
|
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'})
|
|
assert mocked_get.call_args[1]['params']['REQUEST'] == 'GetFeature'
|
|
assert mocked_get.call_args[1]['params']['PROPERTYNAME'] == 'nom'
|
|
assert mocked_get.call_args[1]['params']['TYPENAMES'] == 'ref_metro_limites_communales'
|
|
assert mocked_get.call_args[1]['params']['OUTPUTFORMAT'] == 'json'
|
|
assert mocked_get.call_args[1]['params']['SERVICE'] == 'WFS'
|
|
assert mocked_get.call_args[1]['params']['VERSION'] == connector.get_wfs_service_version()
|
|
assert len(resp.json['data']) == 7
|
|
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
|
|
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)
|
|
mocked_get.side_effect = geoserver_responses
|
|
params = {'type_names': 'ref_metro_limites_communales',
|
|
'property_name': 'nom', 'cql_filter': 'nom=\'Fontaine\'',
|
|
'filter_property_name': 'nom'}
|
|
app.get(endpoint, params=params)
|
|
assert mocked_get.call_args[1]['params']['CQL_FILTER'] == 'nom=\'Fontaine\''
|
|
params['q'] = 'bens'
|
|
app.get(endpoint, params=params)
|
|
assert mocked_get.call_args[1]['params']['CQL_FILTER'] == 'nom=\'Fontaine\' AND nom LIKE \'%bens%\''
|
|
params['case-insensitive'] = True
|
|
app.get(endpoint, params=params)
|
|
assert mocked_get.call_args[1]['params']['CQL_FILTER'] == 'nom=\'Fontaine\' AND nom ILIKE \'%bens%\''
|
|
params.pop('cql_filter')
|
|
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)
|
|
assert endpoint == '/opengis/test/features'
|
|
mocked_get.side_effect = geoserver_responses_errors
|
|
resp = app.get(endpoint, params={
|
|
'type_names': 'ref_metro_limites_communales',
|
|
'property_name': 'nom'
|
|
})
|
|
assert mocked_get.call_args[1]['params']['REQUEST'] == 'GetFeature'
|
|
assert mocked_get.call_args[1]['params']['PROPERTYNAME'] == 'nom'
|
|
assert mocked_get.call_args[1]['params']['TYPENAMES'] == 'ref_metro_limites_communales'
|
|
assert mocked_get.call_args[1]['params']['OUTPUTFORMAT'] == 'json'
|
|
assert mocked_get.call_args[1]['params']['SERVICE'] == 'WFS'
|
|
assert mocked_get.call_args[1]['params']['VERSION'] == connector.get_wfs_service_version()
|
|
result = resp.json
|
|
assert result['err'] == 1
|
|
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_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
|
|
resp = app.get(endpoint, params={
|
|
'type_names': 'ref_metro_limites_communales',
|
|
'property_name': 'nom'
|
|
})
|
|
assert mocked_get.call_args[1]['params']['REQUEST'] == 'GetFeature'
|
|
assert mocked_get.call_args[1]['params']['PROPERTYNAME'] == 'nom'
|
|
assert mocked_get.call_args[1]['params']['TYPENAMES'] == 'ref_metro_limites_communales'
|
|
assert mocked_get.call_args[1]['params']['OUTPUTFORMAT'] == 'json'
|
|
assert mocked_get.call_args[1]['params']['SERVICE'] == 'WFS'
|
|
assert mocked_get.call_args[1]['params']['VERSION'] == connector.get_wfs_service_version()
|
|
result = resp.json
|
|
assert result['err'] == 1
|
|
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
|
|
connector.projection = 'EPSG:3945'
|
|
connector.save()
|
|
endpoint = utils.generic_endpoint_url('opengis', 'reverse', slug=connector.slug)
|
|
assert endpoint == '/opengis/test/reverse'
|
|
|
|
def side_effect(url, **kwargs):
|
|
if kwargs['params'].get('request') == 'GetCapabilities':
|
|
return utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)
|
|
return mock.DEFAULT
|
|
mocked_get.side_effect = side_effect
|
|
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)')
|
|
assert resp.json['lon'] == '5.72407744145'
|
|
assert resp.json['lat'] == '45.1893972656'
|
|
assert resp.json['address']['house_number'] == '4'
|
|
assert resp.json['address']['road'] == 'place victor hugo'
|
|
assert resp.json['address']['postcode'] == '38000'
|
|
assert resp.json['address']['city'] == 'Grenoble'
|
|
|
|
connector.projection = 'EPSG:4326'
|
|
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'
|
|
})
|
|
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'
|
|
})
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'Webservice returned status code 404'
|