Produits Entr'ouvert » Passerelle

# -*- coding: utf-8 -*-

# Generated by Django 1.11.18 on 2020-03-25 17:39

from __future__ import unicode_literals

from django.db import migrations, models

import django.db.models.deletion

import jsonfield.fields

class Migration(migrations.Migration):

    dependencies = [

        ('opengis', '0007_auto_20200324_1019'),

    ]

    operations = [

        migrations.CreateModel(

            name='FeatureCache',

            fields=[

                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),

                ('lat', models.FloatField()),

                ('lon', models.FloatField()),

                ('data', jsonfield.fields.JSONField(default=dict)),

                ('query', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='features', to='opengis.Query', verbose_name='Query')),

            ],

        ),

    ]

import datetime

from jsonfield import JSONField

from django.db import models, transaction

from django.utils import timezone

    def daily(self):

        super(OpenGIS, self).daily()

        self.update_queries()

    def update_queries(self, query_id=None):

        queries = self.query_set.all()

        if query_id:

            queries = queries.filter(pk=query_id)

        for query in queries:

            query.update_cache()

    def save(self, *args, **kwargs):

        super(OpenGIS, self).save(*args, **kwargs)

        if self.query_set.exists():

            self.add_job('update_queries')

        features = self.features.all()

        if not features.exists():

            raise APIError('Data is not synchronized yet. Retry in a few minutes.')

        data = {

            'type': 'FeatureCollection',

            'name': self.typename

        }

        data['features'] = list(features.values_list('data', flat=True))

        return data

    def update_cache(self):

        data = self.endpoint(None, self.typename, None, xml_filter=self.filter_expression)

        features = []

        for feature in data['data']:

            geometry = feature.get('geometry') or {}

            try:

                lon, lat = geometry['coordinates']

            except (KeyError, TypeError):

                self.resource.logger.warning('invalid coordinates in geometry: %s', geometry)

                continue

            features.append(FeatureCache(query=self, lat=lat, lon=lon, data=feature))

        with transaction.atomic():

            self.features.all().delete()

            FeatureCache.objects.bulk_create(features)

    def save(self, *args, **kwargs):

        super(Query, self).save(*args, **kwargs)

        self.resource.add_job('update_queries', query_id=self.pk)

class FeatureCache(models.Model):

    query = models.ForeignKey(

        to=Query,

        on_delete=models.CASCADE,

        related_name='features',

        verbose_name=_('Query'))

    lat = models.FloatField()

    lon = models.FloatField()

    data = JSONField()

from django.core.management import call_command

from passerelle.apps.opengis.models import OpenGIS, Query, FeatureCache

from passerelle.base.models import Job

from passerelle.utils import import_site

@pytest.fixture

def query(connector):

    return Query.objects.create(

        resource=connector,

        name='Test Query',

        slug='test_query',

        description='Test query.',

        typename='pvo_patrimoine_voirie.pvoparking',

        filter_expression=('<Filter><PropertyIsEqualTo><PropertyName>typeparking'

                           '</PropertyName></PropertyIsEqualTo></Filter>')

    )

def geoserver_geolocated_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_GEOLOCATED_FEATURE)

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

def test_opengis_query_cache_update(mocked_get, app, connector, query):

    mocked_get.side_effect = geoserver_geolocated_responses

    query.update_cache()

    assert mocked_get.call_args[1]['params']['FILTER'] == query.filter_expression

    assert mocked_get.call_args[1]['params']['TYPENAMES'] == query.typename

    assert FeatureCache.objects.count() == 4

    feature = FeatureCache.objects.get(lon=1914059.51, lat=4224699.2)

    assert feature.data['properties']['code_post'] == 38000

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

def test_opengis_query_q_endpoint(mocked_get, app, connector, query):

    endpoint = utils.generic_endpoint_url('opengis', 'q/test_query/', slug=connector.slug)

    assert endpoint == '/opengis/test/q/test_query/'

    mocked_get.side_effect = geoserver_geolocated_responses

    query.update_cache()

    feature = FeatureCache.objects.get(lon=1914059.51, lat=4224699.2)

    resp = app.get(endpoint)

    assert len(resp.json['features']) == FeatureCache.objects.count()

    feature_data = next(feature for feature in resp.json['features']

                        if feature['geometry']['coordinates'][0] == 1914059.51)

    assert feature_data == feature.data

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

def test_opengis_query_cache_update_change(mocked_get, app, connector, query):

    mocked_get.side_effect = geoserver_geolocated_responses

    query.update_cache()

    assert FeatureCache.objects.count() == 4

    def new_response(url, **kwargs):

        if kwargs['params'].get('request') == 'GetCapabilities':

            return utils.FakedResponse(status_code=200, content=FAKE_SERVICE_CAPABILITIES)

        return utils.FakedResponse(

            content='{"features": [{"properties": {}, "geometry": {"coordinates": [1, 1]}}]}',

            status_code=200

        )

    mocked_get.side_effect = new_response

    query.update_cache()

    assert FeatureCache.objects.count() == 1

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

def test_opengis_query_q_endpoint_cache_empty(mocked_get, app, connector, query):

    endpoint = utils.generic_endpoint_url('opengis', 'q/test_query/', slug=connector.slug)

    assert not FeatureCache.objects.exists()

    resp = app.get(endpoint)

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

    assert 'not synchronized' in resp.json['err_desc']

    assert not FeatureCache.objects.exists()

    assert mocked_get.call_count == 0

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

def test_opengis_query_cache_update_jobs(mocked_get, app, connector, query):

    mocked_get.side_effect = geoserver_geolocated_responses

    # fixtures created one query

    job = Job.objects.get(method_name='update_queries')

    assert not FeatureCache.objects.exists()

    connector.jobs()

    assert FeatureCache.objects.count() == 4

    job.refresh_from_db()

    assert job.status == 'completed'

    # modifying a query triggers an update

    query.save()

    assert Job.objects.filter(method_name='update_queries', status='registered').count() == 1

    connector.jobs()

    # modifying the connector triggers an update

    connector.save()

    assert Job.objects.filter(method_name='update_queries', status='registered').count() == 1

    connector.jobs()

    # two queries to update

    query.save()

    query.pk = None

    query.slug = query.name = 'test2'

    query.save()

    with mock.patch.object(Query, 'update_cache') as mocked:

        connector.jobs()

        assert mocked.call_count == 2

    # now only one

    query.save()

    with mock.patch.object(Query, 'update_cache') as mocked:

        connector.jobs()

        assert mocked.call_count == 1

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

def test_opengis_query_cache_update_daily(mocked_get, app, connector, query):

    mocked_get.side_effect = geoserver_geolocated_responses

    assert not FeatureCache.objects.exists()

    call_command('cron', 'daily')

    assert FeatureCache.objects.count() == 4

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

def test_opengis_query_endpoint_documentation(mocked_get, app, connector, query):

    resp = app.get(connector.get_absolute_url())

    assert query.name in resp.text

    assert query.description in resp.text

    assert '/opengis/test/q/test_query/' in resp.text

def test_opengis_export_import(query):

    assert OpenGIS.objects.count() == 1

    assert Query.objects.count() == 1

    serialization = {'resources': [query.resource.export_json()]}

    OpenGIS.objects.all().delete()

    assert OpenGIS.objects.count() == 0

    assert Query.objects.count() == 0

    import_site(serialization)

    assert OpenGIS.objects.count() == 1

    assert Query.objects.count() == 1