passerelle/tests/test_opengis.py

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>
    <ows:OperationsMetadata>
        <ows:Operation name="GetFeature">
            <ows:Parameter name="outputFormat">
                <ows:AllowedValues>
                    <ows:Value>application/gml+xml; version=3.2</ows:Value>
                    <ows:Value>application/json; subtype=geojson</ows:Value>
                </ows:AllowedValues>
            </ows:Parameter>
        </ows:Operation>
    </ows:OperationsMetadata>
</wfs:WFS_Capabilities>'''

FAKE_SERVICE_CAPABILITIES_V1_0_0 = '''<?xml version="1.0" encoding="UTF-8"?>
<wfs:WFS_Capabilities version="1.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>1.0.0</ows:ServiceTypeVersion><ows:Fees/>
        <ows:AccessConstraints/>
    </ows:ServiceIdentification>
    <ows:OperationsMetadata>
        <ows:Operation name="GetFeature">
            <ows:Parameter name="outputFormat">
                <ows:AllowedValues>
                    <ows:Value>application/gml+xml; version=3.2</ows:Value>
                    <ows:Value>application/json; subtype=geojson</ows:Value>
                </ows:AllowedValues>
            </ows:Parameter>
        </ows:Operation>
    </ows:OperationsMetadata>
</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 &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": {
        "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':
        assert kwargs['params'].get('service')
        return utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)
    return utils.FakedResponse(status_code=200, content=FAKE_FEATURES_JSON)

def geoserver_responses_v1_0_0(url, **kwargs):
    if kwargs['params'].get('request') == 'GetCapabilities':
        assert kwargs['params'].get('service')
        return utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES_V1_0_0)
    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.261214,532579.160633,5747876.194333'
    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.796790,4.784040,45.796990,4.784240'
    assert mocked_get.call_args[1]['params']['CRS'] == 'EPSG:4326'


@mock.patch('passerelle.utils.Request.get')
@pytest.mark.parametrize('lat,lon', [
    ('bad-value', '4.784140'),
    ('45.796890', 'bad-value'),
])
def test_feature_info_bad_request(mocked_get, app, connector, lat, lon):
    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': lat, 'lon': lon})
    assert resp.json['err'] == 1
    assert resp.json['err_desc'] == 'Bad coordinates format'


@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=b'\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.671580,5741338.068556,539951.167806,5741949.564782'
    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.756026,4.844971,45.759859,4.850464'
    assert resp.content == b'\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 'json' in mocked_get.call_args[1]['params']['outputFormat']
    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 'json' in mocked_get.call_args[1]['params']['outputFormat']
    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 'json' in mocked_get.call_args[1]['params']['outputFormat']
    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']


@pytest.mark.parametrize("server_responses, version, typename_label", [
    (geoserver_responses_v1_0_0, '1.0.0', 'TYPENAME'),
    (geoserver_responses, '2.0.0', 'TYPENAMES')])
@mock.patch('passerelle.utils.Request.get')
def test_typename_parameter_upgrade(mocked_get, server_responses, version, typename_label, app, connector):
    endpoint = utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)
    assert endpoint == '/opengis/test/features'
    mocked_get.side_effect = server_responses
    resp = app.get(endpoint, params={'type_names': '...', 'property_name': '...'})
    assert mocked_get.call_args[1]['params']['request'] == 'GetFeature'
    assert mocked_get.call_args[1]['params']['version'] == version
    assert typename_label in mocked_get.call_args[1]['params'].keys()


@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.486036 4224640.457791),45,meters)')
    assert resp.json['lon'] == '5.724077'
    assert resp.json['lat'] == '45.189397'
    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'