Produits Entr'ouvert » Passerelle

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

# Generated by Django 1.11.18 on 2020-01-30 11:34

from __future__ import unicode_literals

from django.db import migrations, models

import jsonfield.fields

class Migration(migrations.Migration):

    dependencies = [

        ('base_adresse', '0015_auto_20191206_1244'),

    ]

    operations = [

        migrations.CreateModel(

            name='AddressCacheModel',

            fields=[

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

                ('api_id', models.CharField(max_length=30, unique=True)),

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

                ('timestamp', models.DateTimeField(auto_now=True)),

            ],

        ),

    ]

import datetime

from jsonfield import JSONField

    @staticmethod

    def format_address_data(data):

        result = {}

        result['lon'] = str(data['geometry']['coordinates'][0])

        result['lat'] = str(data['geometry']['coordinates'][1])

        result['address'] = {'country': 'France'}

        for prop, value in data['properties'].items():

            if prop in ('city', 'postcode', 'citycode'):

                result['address'][prop] = value

            elif prop == 'housenumber':

                result['address']['house_number'] = value

            elif prop == 'label':

                result['text'] = result['display_name'] = value

            elif prop == 'name':

                house_number = data['properties'].get('housenumber')

                if house_number and value.startswith(house_number):

                    value = value[len(house_number):].strip()

                result['address']['road'] = value

            elif prop == 'id':

                result['id'] = value

        return result

              description=_('Addresses list'),

                  'id': {'description': _('Address identifier')},

                  'q': {'description': _('Address'), 'example_value': '169 rue du chateau, paris'},

                  'page_limit': {'description': _('Maximum number of results to return. Must be '

                                                  'lower than 20.')},

                  'zipcode': {'description': _('Zipcode'), 'example_value': '75014'},

                  'lat': {'description': _('Prioritize results according to coordinates. "lat" '

                                           'parameter must be present.')},

                  'lon': {'description': _('Prioritize results according to coordinates. "lon" '

                                           'parameter must be present.')},

    def addresses(self, request, id=None, q=None, zipcode='', lat=None, lon=None, page_limit=5):

        if id is not None:

            try:

                address = AddressCacheModel.objects.get(api_id=id)

            except AddressCacheModel.DoesNotExist:

                return {'err': _('Address ID not found')}

            address.update_timestamp()

            return {'data': [address.data]}

            return {'data': []}

        try:

            if int(page_limit) > 20:

                page_limit = 20

        except ValueError:

            page_limit = 5

        query_args = {'q': q, 'limit': page_limit}

            data = self.format_address_data(feature)

            result.append(data)

            AddressCacheModel.objects.update_or_create(api_id=data['id'], data=data)

        return {'data': result}

    @endpoint(pattern='(?P<q>.+)?$', description=_('Geocoding (Nominatim API)'),

              parameters={

                  'q': {'description': _('Address'), 'example_value': '169 rue du chateau, paris'},

              })

    def search(self, request, q, zipcode='', lat=None, lon=None, **kwargs):

        if kwargs.get('format', 'json') != 'json':

            raise NotImplementedError()

        result = self.addresses(request, q=q, zipcode=zipcode, lat=lat, lon=lon, page_limit=1)

        return result['data']

            result = self.format_address_data(feature)

    def clean_addresses_cache(self):

        old_addresses = AddressCacheModel.objects.filter(

            timestamp__lt=timezone.now() - datetime.timedelta(hours=1)

        )

        old_addresses.delete()

        self.clean_addresses_cache()

class AddressCacheModel(models.Model):

    api_id = models.CharField(max_length=30, unique=True)

    data = JSONField()

    timestamp = models.DateTimeField(auto_now=True)

    def update_timestamp(self):

        self.save()

from httmock import urlmatch, HTTMock, response, remember_called

@remember_called

        yield api_adresse_data_gouv_fr_search

import datetime

                                                 DepartmentModel, RegionModel, AddressCacheModel)

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

def test_base_adresse_addresses(mocked_get, app, base_adresse):

    endpoint = utils.generic_endpoint_url('base-adresse', 'addresses', slug=base_adresse.slug)

    assert endpoint == '/base-adresse/base-adresse/addresses'

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

    resp = app.get(endpoint, params={'q': 'plop'}, status=200)

    data = resp.json['data'][0]

    assert data['lat'] == '47.474633'

    assert data['lon'] == '-0.593775'

    assert data['display_name'] == 'Rue Roger Halope 49000 Angers'

    assert data['text'] == 'Rue Roger Halope 49000 Angers'

    assert data['id'] == '49007_6950_be54bd'

    assert data['address']['city'] == 'Angers'

    assert data['address']['postcode'] == '49000'

    assert data['address']['citycode'] == '49007'

    assert data['address']['road'] == 'Rue Roger Halope'

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

def test_base_adresse_addresses_qs_page_limit(mocked_get, app, base_adresse):

    resp = app.get('/base-adresse/%s/addresses?q=plop&page_limit=1' % base_adresse.slug)

    assert 'limit=1' in mocked_get.call_args[0][0]

    resp = app.get('/base-adresse/%s/addresses?q=plop&page_limit=100' % base_adresse.slug)

    assert 'limit=20' in mocked_get.call_args[0][0]

    resp = app.get('/base-adresse/%s/addresses?q=plop&page_limit=blabla' % base_adresse.slug)

    assert 'limit=5' in mocked_get.call_args[0][0]

def test_base_adresse_addresses_cache(app, base_adresse, mock_api_adresse_data_gouv_fr_search):

    resp = app.get('/base-adresse/%s/addresses?q=plop' % base_adresse.slug)

    assert mock_api_adresse_data_gouv_fr_search.call['count'] == 1

    data = resp.json['data'][0]

    assert data['text'] == 'Rue Roger Halope 49000 Angers'

    api_id = data['id']

    assert AddressCacheModel.objects.filter(api_id=api_id).exists()

    assert AddressCacheModel.objects.count() == 1

    resp = app.get('/base-adresse/%s/addresses?id=%s' % (base_adresse.slug, api_id))

    assert mock_api_adresse_data_gouv_fr_search.call['count'] == 1  # no new call

    assert data['text'] == 'Rue Roger Halope 49000 Angers'

    assert 'address' in data

    resp = app.get('/base-adresse/%s/addresses?q=plop' % base_adresse.slug)

    assert AddressCacheModel.objects.count() == 1  # no new object has been created

def test_base_adresse_addresses_cache_err(app, base_adresse,

                                          mock_api_adresse_data_gouv_fr_search):

    resp = app.get('/base-adresse/%s/addresses?id=%s' % (base_adresse.slug, 'wrong_id'))

    assert mock_api_adresse_data_gouv_fr_search.call['count'] == 0

    assert 'err' in resp.json

@pytest.mark.usefixtures('mock_update_api_geo', 'mock_update_streets')

def test_base_adresse_addresses_clean_cache(app, base_adresse, freezer,

                                            mock_api_adresse_data_gouv_fr_search):

    resp = app.get('/base-adresse/%s/addresses?q=plop' % base_adresse.slug)

    assert AddressCacheModel.objects.count() == 1

    freezer.move_to(datetime.timedelta(minutes=30))

    call_command('cron', 'hourly')

    assert AddressCacheModel.objects.count() == 1

    freezer.move_to(datetime.timedelta(minutes=30, seconds=1))

    call_command('cron', 'hourly')

    assert AddressCacheModel.objects.count() == 0

    resp = app.get('/base-adresse/%s/addresses?q=plop' % base_adresse.slug)

    assert AddressCacheModel.objects.count() == 1

    # asking for the address again resets the timestamp

    freezer.move_to(datetime.timedelta(hours=1, seconds=1))

    resp = app.get('/base-adresse/%s/addresses?q=plop' % base_adresse.slug)

    call_command('cron', 'hourly')

    assert AddressCacheModel.objects.count() == 1

    freezer.move_to(datetime.timedelta(hours=1, seconds=1))

    resp = app.get('/base-adresse/%s/addresses?id=%s' % (base_adresse.slug, '49007_6950_be54bd'))

    call_command('cron', 'hourly')

    assert AddressCacheModel.objects.count() == 1