Produits Entr'ouvert » Passerelle

import xml.etree.ElementTree as ET

from HTMLParser import HTMLParser

import pyproj

from django.core.cache import cache

    projection = models.CharField(_('GIS projection'), choices=PROJECTIONS,

                                  default='EPSG:3857', max_length=16)

    search_radius = models.IntegerField(_('Radius for point search'), default=5)

    class Meta:

        verbose_name = _('OpenGIS')

    def check_status(self):

        if self.wms_service_url:

            response.raise_for_status()

        if self.wfs_service_url:

            response.raise_for_status()

    @endpoint(perm='can_access', description='Get features',

        params = {

            'VERSION': self.get_wfs_service_version(),

            'SERVICE': 'WFS',

        response = self.requests.get(self.wfs_service_url, params=params)

        data = []

        try:

        except ValueError:

            self.handle_opengis_error(response)

            # if handle_opengis_error did not raise an error, we raise a generic one

            'SRS': self.projection,

            'BBOX': bbox,

        }

    @endpoint(perm='can_access', description=_('Get feature info'))

    def reverse(self, request, lat, lon, **kwargs):

            for attribute, properties in self.attributes_mapping:

                for field in properties:

                    if closest_feature['properties'].get(field):

                        break

            return result

        raise APIError('Unable to geocode')

import utils

FAKE_FEATURE_INFO = '''<?xml version="1.0" encoding="UTF-8"?>

  <cad_cadastre.cadparcelle_layer>

    <gml:name>Parcelle cadastrale (Plan cadastral informatise du Grand Lyon)</gml:name>

    <cad_cadastre.cadparcelle_feature>

</msGMLOutput>'''

FAKE_SERVICE_CAPABILITIES = '''<?xml version="1.0" encoding="UTF-8"?>

    <ows:ServiceIdentification>

        <ows:Title/><ows:Abstract/>

        <ows:ServiceType>WFS</ows:ServiceType><ows:ServiceTypeVersion>2.0.0</ows:ServiceTypeVersion><ows:Fees/>

    "type": "FeatureCollection"

}'''

FAKE_GEOLOCATED_FEATURE = '''{

    "crs": {

        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)

    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'

    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)

    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['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)

    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

    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)

    params = {'type_names': 'ref_metro_limites_communales',

              'property_name': 'nom', 'cql_filter': 'nom=\'Fontaine\'',

              'filter_property_name': 'nom'}

    assert mocked_get.call_args[1]['params']['CQL_FILTER'] == 'nom=\'Fontaine\''

    params['q'] = 'bens'

    assert mocked_get.call_args[1]['params']['CQL_FILTER'] == 'nom=\'Fontaine\' AND nom LIKE \'%bens%\''

    params['case-insensitive'] = True

    assert mocked_get.call_args[1]['params']['CQL_FILTER'] == 'nom=\'Fontaine\' AND nom ILIKE \'%bens%\''

    params.pop('cql_filter')

    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 result['err_desc'] == 'OpenGIS Error: NoApplicableCode'

    assert 'Could not parse' in result['data']['text']

@mock.patch('passerelle.utils.Request.get')

    endpoint = utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)

    assert endpoint == '/opengis/test/features'

    mocked_get.side_effect = geoserver_responses_errors_unparsable

    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

    endpoint = utils.generic_endpoint_url('opengis', 'reverse', slug=connector.slug)

    assert endpoint == '/opengis/test/reverse'

    mocked_get.return_value = utils.FakedResponse(content=FAKE_GEOLOCATED_FEATURE, status_code=200)

    assert resp.json['lon'] == '5.72407744145'

    assert resp.json['lat'] == '45.1893972656'

    assert resp.json['address']['house_number'] == '4'

    connector.search_radius = 10

    connector.save()

    mocked_get.return_value = utils.FakedResponse(content='{"features": []}', status_code=200)

    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)

    assert resp.json['err'] == 1

    assert resp.json['err_desc'] == 'Webservice returned status code 404'