base_adresse: add API Géo endpoints (#11497)
Namely /cities/, /departments/ and /regions/.
This commit is contained in:
parent
27f3bafed5
commit
1a4c7c649b
|
|
@ -0,0 +1,95 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# Generated by Django 1.11.18 on 2019-12-06 11:44
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from django.db import migrations, models
|
||||
import django.db.models.deletion
|
||||
import passerelle.apps.base_adresse.models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('base_adresse', '0014_auto_20190207_0456'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='CityModel',
|
||||
fields=[
|
||||
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('name', models.CharField(max_length=150, verbose_name='City name')),
|
||||
('unaccent_name', models.CharField(max_length=150, null=True, verbose_name='City name ascii char')),
|
||||
('code', models.CharField(max_length=5, verbose_name='INSEE code')),
|
||||
('zipcode', models.CharField(max_length=5, verbose_name='Postal code')),
|
||||
('population', models.PositiveIntegerField(verbose_name='Population')),
|
||||
('last_update', models.DateTimeField(auto_now=True, null=True, verbose_name='Last update')),
|
||||
],
|
||||
options={
|
||||
'ordering': ['-population', 'zipcode', 'unaccent_name', 'name'],
|
||||
},
|
||||
bases=(passerelle.apps.base_adresse.models.UnaccentNameMixin, models.Model),
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='DepartmentModel',
|
||||
fields=[
|
||||
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('name', models.CharField(max_length=100, verbose_name='Department name')),
|
||||
('unaccent_name', models.CharField(max_length=150, null=True, verbose_name='Department name ascii char')),
|
||||
('code', models.CharField(max_length=3, unique=True, verbose_name='Department code')),
|
||||
('last_update', models.DateTimeField(auto_now=True, null=True, verbose_name='Last update')),
|
||||
],
|
||||
options={
|
||||
'ordering': ['code'],
|
||||
},
|
||||
bases=(passerelle.apps.base_adresse.models.UnaccentNameMixin, models.Model),
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='RegionModel',
|
||||
fields=[
|
||||
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('name', models.CharField(max_length=150, verbose_name='Region name')),
|
||||
('unaccent_name', models.CharField(max_length=150, null=True, verbose_name='Region name ascii char')),
|
||||
('code', models.CharField(max_length=2, unique=True, verbose_name='Region code')),
|
||||
('last_update', models.DateTimeField(auto_now=True, null=True, verbose_name='Last update')),
|
||||
],
|
||||
options={
|
||||
'ordering': ['code'],
|
||||
},
|
||||
bases=(passerelle.apps.base_adresse.models.UnaccentNameMixin, models.Model),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='baseadresse',
|
||||
name='api_geo_url',
|
||||
field=models.CharField(default=b'https://geo.api.gouv.fr/', help_text='Base Adresse API Geo URL', max_length=128, verbose_name='API Geo URL'),
|
||||
),
|
||||
migrations.AlterField(
|
||||
model_name='baseadresse',
|
||||
name='zipcode',
|
||||
field=models.CharField(blank=True, max_length=600, verbose_name='Postal codes or department number to get streets, separated with commas'),
|
||||
),
|
||||
migrations.AlterField(
|
||||
model_name='streetmodel',
|
||||
name='city',
|
||||
field=models.CharField(max_length=150, verbose_name='City'),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='departmentmodel',
|
||||
name='region',
|
||||
field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='base_adresse.RegionModel'),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='citymodel',
|
||||
name='department',
|
||||
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, to='base_adresse.DepartmentModel'),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='citymodel',
|
||||
name='region',
|
||||
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, to='base_adresse.RegionModel'),
|
||||
),
|
||||
migrations.AlterUniqueTogether(
|
||||
name='citymodel',
|
||||
unique_together=set([('code', 'zipcode')]),
|
||||
),
|
||||
]
|
||||
|
|
@ -1,9 +1,9 @@
|
|||
import bz2
|
||||
import json
|
||||
import os
|
||||
import urlparse
|
||||
import unicodedata
|
||||
|
||||
from requests import RequestException
|
||||
|
||||
from django.db import connection, models
|
||||
from django.db.models import Q
|
||||
|
|
@ -15,6 +15,8 @@ from django.utils.six.moves.urllib.parse import urljoin
|
|||
|
||||
from passerelle.base.models import BaseResource
|
||||
from passerelle.utils.api import endpoint
|
||||
from passerelle.utils.conversion import simplify
|
||||
from passerelle.utils.jsonresponse import APIError
|
||||
|
||||
|
||||
class BaseAdresse(BaseResource):
|
||||
|
|
@ -24,16 +26,23 @@ class BaseAdresse(BaseResource):
|
|||
verbose_name=_('Service URL'),
|
||||
help_text=_('Base Adresse Web Service URL'))
|
||||
|
||||
api_geo_url = models.CharField(
|
||||
max_length=128, blank=False,
|
||||
default='https://geo.api.gouv.fr/',
|
||||
verbose_name=_('API Geo URL'),
|
||||
help_text=_('Base Adresse API Geo URL'))
|
||||
|
||||
category = _('Geographic information system')
|
||||
|
||||
api_description = _("The API is a partial view of OpenStreetMap's Nominatim "
|
||||
"own API; it currently doesn't support all parameters and "
|
||||
"is limited to the JSON format.")
|
||||
api_description = _("The geocoding endpoints are a partial view of OpenStreetMap's "
|
||||
"Nominatim own API; it currently doesn't support all parameters and "
|
||||
"is limited to the JSON format. The cities, departments and regions "
|
||||
"endpoints source data from French API Geo.")
|
||||
|
||||
zipcode = models.CharField(
|
||||
max_length=600,
|
||||
blank=True,
|
||||
verbose_name=_('Postal codes or county number to get streets, separated with commas'))
|
||||
verbose_name=_('Postal codes or department number to get streets, separated with commas'))
|
||||
|
||||
class Meta:
|
||||
verbose_name = _('Base Adresse Web Service')
|
||||
|
|
@ -135,8 +144,7 @@ class BaseAdresse(BaseResource):
|
|||
else:
|
||||
streets = StreetModel.objects.all()
|
||||
if q:
|
||||
unaccented_q = unicodedata.normalize('NFKD', q).encode('ascii', 'ignore').lower()
|
||||
streets = streets.filter(unaccent_name__icontains=unaccented_q)
|
||||
streets = streets.filter(unaccent_name__icontains=simplify(q))
|
||||
|
||||
if zipcode:
|
||||
streets = streets.filter(zipcode__startswith=zipcode)
|
||||
|
|
@ -158,6 +166,82 @@ class BaseAdresse(BaseResource):
|
|||
|
||||
return {'data': result}
|
||||
|
||||
@endpoint(description=_('Cities list'),
|
||||
parameters={
|
||||
'id': {'description': _('Get exactly one city using its code and postal code '
|
||||
'separated with a dot'),
|
||||
'example_value': '75056.75014'},
|
||||
'q': {'description': _("Search text in name or postal code"),
|
||||
'example_value': 'Paris'},
|
||||
'code': {'description': _('INSEE code'), 'example_value': '75056'},
|
||||
'region_code': {'description': _('Region code'), 'example_value': '11'},
|
||||
'department_code': {'description': _('Department code'), 'example_value': '75'},
|
||||
})
|
||||
def cities(self, request, id=None, q=None, code=None, region_code=None,
|
||||
department_code=None):
|
||||
cities = CityModel.objects.all()
|
||||
|
||||
if id is not None:
|
||||
try:
|
||||
code, zipcode = id.split('.')
|
||||
except ValueError:
|
||||
raise APIError('Invalid id')
|
||||
cities = cities.filter(code=code, zipcode=zipcode)
|
||||
if q:
|
||||
unaccented_q = simplify(q)
|
||||
cities = cities.filter(Q(unaccent_name__istartswith=unaccented_q) |
|
||||
Q(zipcode__istartswith=unaccented_q))
|
||||
if code:
|
||||
cities = cities.filter(code=code)
|
||||
if region_code:
|
||||
cities = cities.filter(region__code=region_code)
|
||||
if department_code:
|
||||
cities = cities.filter(department__code=department_code)
|
||||
|
||||
cities = cities.select_related('department', 'region')
|
||||
return {'data': [city.to_json() for city in cities]}
|
||||
|
||||
@endpoint(description=_('Departments list'),
|
||||
parameters={
|
||||
'id': {'description': _('Get exactly one department using its code'),
|
||||
'example_value': '59'},
|
||||
'q': {'description': _('Search text in name or code'), 'example_value': 'Nord'},
|
||||
'region_code': {'description': _('Region code'), 'example_value': '32'},
|
||||
})
|
||||
def departments(self, request, id=None, q=None, region_code=None):
|
||||
departments = DepartmentModel.objects.all()
|
||||
|
||||
if id is not None:
|
||||
departments = departments.filter(code=id)
|
||||
if q:
|
||||
unaccented_q = simplify(q)
|
||||
departments = departments.filter(Q(unaccent_name__istartswith=unaccented_q) |
|
||||
Q(code__istartswith=unaccented_q))
|
||||
if region_code:
|
||||
departments = departments.filter(region__code=region_code)
|
||||
|
||||
departments = departments.select_related('region')
|
||||
return {'data': [department.to_json() for department in departments]}
|
||||
|
||||
@endpoint(description=_('Regions list'),
|
||||
parameters={
|
||||
'id': {'description': _('Get exactly one region using its code'),
|
||||
'example_value': '32'},
|
||||
'q': {'description': _('Search text in name or code'),
|
||||
'example_value': 'Hauts-de-France'},
|
||||
})
|
||||
def regions(self, request, id=None, q=None):
|
||||
regions = RegionModel.objects.all()
|
||||
|
||||
if id is not None:
|
||||
regions = regions.filter(code=id)
|
||||
if q:
|
||||
unaccented_q = simplify(q)
|
||||
regions = regions.filter(Q(unaccent_name__istartswith=unaccented_q) |
|
||||
Q(code__istartswith=unaccented_q))
|
||||
|
||||
return {'data': [region.to_json() for region in regions]}
|
||||
|
||||
def check_status(self):
|
||||
if self.service_url == 'https://api-adresse.data.gouv.fr/':
|
||||
result = self.search(None, '169 rue du chateau, paris')
|
||||
|
|
@ -174,6 +258,9 @@ class BaseAdresse(BaseResource):
|
|||
criteria |= Q(zipcode__startswith=zipcode)
|
||||
return StreetModel.objects.filter(criteria)
|
||||
|
||||
def cities_exist(self):
|
||||
return CityModel.objects.exists()
|
||||
|
||||
def update_streets_data(self):
|
||||
if not self.get_zipcodes():
|
||||
return
|
||||
|
|
@ -218,20 +305,91 @@ class BaseAdresse(BaseResource):
|
|||
|
||||
self.get_streets_queryset().filter(last_update__lt=start_update).delete()
|
||||
|
||||
def get_api_geo_endpoint(self, endpoint):
|
||||
if not self.api_geo_url:
|
||||
return
|
||||
error = None
|
||||
try:
|
||||
response = self.requests.get(urljoin(self.api_geo_url, endpoint))
|
||||
except RequestException as e:
|
||||
error = e
|
||||
else:
|
||||
if response.status_code != 200:
|
||||
error = 'bad status code (%s)' % response.status_code
|
||||
else:
|
||||
try:
|
||||
result = response.json()
|
||||
except ValueError:
|
||||
error = 'invalid json, got: %s' % response.text
|
||||
if error:
|
||||
self.logger.error('failed to update api geo data for endpoint %s: %s',
|
||||
endpoint, error)
|
||||
return
|
||||
if not result:
|
||||
raise Exception('api geo returns empty json')
|
||||
return result
|
||||
|
||||
def update_api_geo_data(self):
|
||||
regions_json = self.get_api_geo_endpoint('regions')
|
||||
departments_json = self.get_api_geo_endpoint('departements')
|
||||
cities_json = self.get_api_geo_endpoint('communes')
|
||||
if not (regions_json and departments_json and cities_json):
|
||||
return
|
||||
start_update = timezone.now()
|
||||
|
||||
for data in regions_json:
|
||||
defaults = {
|
||||
'name': data['nom'],
|
||||
}
|
||||
RegionModel.objects.update_or_create(code=data['code'], defaults=defaults)
|
||||
RegionModel.objects.filter(last_update__lt=start_update).delete()
|
||||
|
||||
for data in departments_json:
|
||||
defaults = {
|
||||
'name': data['nom'],
|
||||
'region': RegionModel.objects.get(code=data['codeRegion']),
|
||||
}
|
||||
DepartmentModel.objects.update_or_create(code=data['code'], defaults=defaults)
|
||||
DepartmentModel.objects.filter(last_update__lt=start_update).delete()
|
||||
|
||||
for data in cities_json:
|
||||
for zipcode in data['codesPostaux']:
|
||||
defaults = {
|
||||
'name': data['nom'],
|
||||
'population': data.get('population', 0),
|
||||
}
|
||||
if data.get('codeDepartement'):
|
||||
defaults['department'] = DepartmentModel.objects.get(code=data['codeDepartement'])
|
||||
if data.get('codeRegion'):
|
||||
defaults['region'] = RegionModel.objects.get(code=data['codeRegion'])
|
||||
CityModel.objects.update_or_create(
|
||||
code=data['code'], zipcode=zipcode, defaults=defaults)
|
||||
CityModel.objects.filter(last_update__lt=start_update).delete()
|
||||
|
||||
def hourly(self):
|
||||
super(BaseAdresse, self).hourly()
|
||||
# don't wait for daily job to grab data
|
||||
if self.get_zipcodes() and not self.get_streets_queryset().exists():
|
||||
# don't wait for daily job to grab streets
|
||||
self.update_streets_data()
|
||||
if not CityModel.objects.exists():
|
||||
self.update_api_geo_data()
|
||||
|
||||
def daily(self):
|
||||
super(BaseAdresse, self).daily()
|
||||
self.update_streets_data()
|
||||
self.update_api_geo_data()
|
||||
|
||||
|
||||
class StreetModel(models.Model):
|
||||
class UnaccentNameMixin(object):
|
||||
|
||||
city = models.CharField(_('City'), max_length=100)
|
||||
def save(self, *args, **kwargs):
|
||||
self.unaccent_name = unicodedata.normalize('NFKD', self.name).encode('ascii', 'ignore').lower()
|
||||
super(UnaccentNameMixin, self).save(*args, **kwargs)
|
||||
|
||||
|
||||
class StreetModel(UnaccentNameMixin, models.Model):
|
||||
|
||||
city = models.CharField(_('City'), max_length=150)
|
||||
name = models.CharField(_('Street name'), max_length=150)
|
||||
unaccent_name = models.CharField(_('Street name ascii char'), max_length=150, null=True)
|
||||
zipcode = models.CharField(_('Postal code'), max_length=5)
|
||||
|
|
@ -245,6 +403,86 @@ class StreetModel(models.Model):
|
|||
def __unicode__(self):
|
||||
return self.name
|
||||
|
||||
def save(self, *args, **kwargs):
|
||||
self.unaccent_name = unicodedata.normalize('NFKD', self.name).encode('ascii', 'ignore')
|
||||
super(StreetModel, self).save(*args, **kwargs)
|
||||
|
||||
@six.python_2_unicode_compatible
|
||||
class RegionModel(UnaccentNameMixin, models.Model):
|
||||
|
||||
name = models.CharField(_('Region name'), max_length=150)
|
||||
unaccent_name = models.CharField(_('Region name ascii char'), max_length=150, null=True)
|
||||
code = models.CharField(_('Region code'), max_length=2, unique=True)
|
||||
last_update = models.DateTimeField(_('Last update'), null=True, auto_now=True)
|
||||
|
||||
def to_json(self):
|
||||
return {
|
||||
'text': str(self),
|
||||
'id': self.code,
|
||||
'code': self.code,
|
||||
'name': self.name,
|
||||
}
|
||||
|
||||
class Meta:
|
||||
ordering = ['code']
|
||||
|
||||
def __str__(self):
|
||||
return '%s %s' % (self.code, self.name)
|
||||
|
||||
|
||||
@six.python_2_unicode_compatible
|
||||
class DepartmentModel(UnaccentNameMixin, models.Model):
|
||||
|
||||
name = models.CharField(_('Department name'), max_length=100)
|
||||
unaccent_name = models.CharField(_('Department name ascii char'), max_length=150, null=True)
|
||||
code = models.CharField(_('Department code'), max_length=3, unique=True)
|
||||
region = models.ForeignKey(RegionModel, on_delete=models.CASCADE)
|
||||
last_update = models.DateTimeField(_('Last update'), null=True, auto_now=True)
|
||||
|
||||
def to_json(self):
|
||||
return {
|
||||
'text': str(self),
|
||||
'id': self.code,
|
||||
'code': self.code,
|
||||
'name': self.name,
|
||||
'region_code': self.region.code,
|
||||
'region_name': self.region.name,
|
||||
}
|
||||
|
||||
class Meta:
|
||||
ordering = ['code']
|
||||
|
||||
def __str__(self):
|
||||
return '%s %s' % (self.code, self.name)
|
||||
|
||||
|
||||
@six.python_2_unicode_compatible
|
||||
class CityModel(UnaccentNameMixin, models.Model):
|
||||
|
||||
name = models.CharField(_('City name'), max_length=150)
|
||||
unaccent_name = models.CharField(_('City name ascii char'), max_length=150, null=True)
|
||||
code = models.CharField(_('INSEE code'), max_length=5)
|
||||
zipcode = models.CharField(_('Postal code'), max_length=5)
|
||||
population = models.PositiveIntegerField(_('Population'))
|
||||
department = models.ForeignKey(DepartmentModel, on_delete=models.CASCADE, blank=True, null=True)
|
||||
region = models.ForeignKey(RegionModel, on_delete=models.CASCADE, blank=True, null=True)
|
||||
last_update = models.DateTimeField(_('Last update'), null=True, auto_now=True)
|
||||
|
||||
def to_json(self):
|
||||
data = {
|
||||
'text': str(self),
|
||||
'id': '%s.%s' % (self.code, self.zipcode),
|
||||
'code': self.code,
|
||||
'name': self.name,
|
||||
'zipcode': self.zipcode,
|
||||
'population': self.population,
|
||||
'department_code': self.department.code if self.department else None,
|
||||
'department_name': self.department.name if self.department else None,
|
||||
'region_code': self.region.code if self.region else None,
|
||||
'region_name': self.region.name if self.region else None,
|
||||
}
|
||||
return data
|
||||
|
||||
class Meta:
|
||||
ordering = ['-population', 'zipcode', 'unaccent_name', 'name']
|
||||
unique_together = ('code', 'zipcode')
|
||||
|
||||
def __str__(self):
|
||||
return '%s %s' % (self.zipcode, self.name)
|
||||
|
|
|
|||
|
|
@ -8,6 +8,11 @@
|
|||
{% trans "Street data is not available yet, it should soon be downloaded." %}
|
||||
</div>
|
||||
{% endif %}
|
||||
{% if object.api_geo_url and not object.cities_exist %}
|
||||
<div class="infonotice">
|
||||
{% trans "API Géo data is not available yet, it should soon be downloaded." %}
|
||||
</div>
|
||||
{% endif %}
|
||||
{% endblock %}
|
||||
|
||||
{% block security %}
|
||||
|
|
|
|||
|
|
@ -6,9 +6,14 @@ import mock
|
|||
import utils
|
||||
import json
|
||||
|
||||
from django.core.management import call_command
|
||||
from requests.exceptions import ConnectionError
|
||||
|
||||
from passerelle.apps.base_adresse.models import BaseAdresse, StreetModel
|
||||
from django.core.management import call_command
|
||||
from django.core.management.base import CommandError
|
||||
from django.utils.six.moves.urllib.parse import urljoin
|
||||
|
||||
from passerelle.apps.base_adresse.models import (BaseAdresse, StreetModel, CityModel,
|
||||
DepartmentModel, RegionModel)
|
||||
|
||||
FAKED_CONTENT = json.dumps({
|
||||
"limit": 1,
|
||||
|
|
@ -41,6 +46,54 @@ FAKED_CONTENT = json.dumps({
|
|||
|
||||
FAKE_DATA = ''
|
||||
|
||||
FAKE_API_GEO_LIST = [
|
||||
{
|
||||
"code": "75056",
|
||||
"codeDepartement": "75",
|
||||
"codeRegion": "11",
|
||||
"codesPostaux": [
|
||||
"75001",
|
||||
"75002",
|
||||
],
|
||||
"nom": "Paris",
|
||||
"population": 2190327,
|
||||
},
|
||||
{
|
||||
"code": "97501",
|
||||
"codesPostaux": [
|
||||
"97500"
|
||||
],
|
||||
"nom": "Miquelon-Langlade",
|
||||
"population": 596
|
||||
}
|
||||
]
|
||||
|
||||
FAKE_API_GEO = json.dumps(FAKE_API_GEO_LIST)
|
||||
|
||||
FAKE_API_GEO_DEPARTMENTS = json.dumps([
|
||||
{
|
||||
"code": "75",
|
||||
"codeRegion": "11",
|
||||
"nom": "Paris"
|
||||
},
|
||||
{
|
||||
"code": "58",
|
||||
"codeRegion": "27",
|
||||
"nom": "Nièvre",
|
||||
}
|
||||
])
|
||||
|
||||
FAKE_API_GEO_REGIONS = json.dumps([
|
||||
{
|
||||
"code": "11",
|
||||
"nom": "Île-de-France"
|
||||
},
|
||||
{
|
||||
"code": "27",
|
||||
"nom": "Bourgogne-Franche-Comté"
|
||||
}
|
||||
])
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def base_adresse(db):
|
||||
|
|
@ -74,6 +127,42 @@ def street(db):
|
|||
citycode=u'73001')
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def region(db):
|
||||
return RegionModel.objects.create(name=u'Auvergne-Rhône-Alpes', code='84')
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def department(db, region):
|
||||
return DepartmentModel.objects.create(name=u'Savoie', code='73', region=region)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def city(db, region, department):
|
||||
return CityModel.objects.create(name=u'Chambéry', code='73065', zipcode='73000',
|
||||
population=42000, region=region, department=department)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def miquelon(db):
|
||||
return CityModel.objects.create(name=u'Miquelon-Langlade', code='97501', zipcode='97500',
|
||||
population=42)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_update_api_geo():
|
||||
with mock.patch('passerelle.apps.base_adresse.models.BaseAdresse.update_api_geo_data',
|
||||
new=lambda x: None) as _fixture:
|
||||
yield _fixture
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_update_streets():
|
||||
with mock.patch('passerelle.apps.base_adresse.models.BaseAdresse.update_streets_data',
|
||||
new=lambda x: None) as _fixture:
|
||||
yield _fixture
|
||||
|
||||
|
||||
@mock.patch('passerelle.utils.Request.get')
|
||||
def test_base_adresse_search(mocked_get, app, base_adresse):
|
||||
endpoint = utils.generic_endpoint_url('base-adresse', 'search', slug=base_adresse.slug)
|
||||
|
|
@ -195,6 +284,7 @@ def test_base_adresse_streets_get_by_id(app, base_adresse, street):
|
|||
assert len(resp.json['data']) == 0
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('mock_update_api_geo')
|
||||
@mock.patch('passerelle.utils.Request.get')
|
||||
def test_base_adresse_command_update(mocked_get, db, base_adresse):
|
||||
filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.bz2')
|
||||
|
|
@ -214,8 +304,10 @@ def test_base_adresse_command_update(mocked_get, db, base_adresse):
|
|||
assert mocked_get.call_count == 2
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('mock_update_api_geo')
|
||||
@mock.patch('passerelle.utils.Request.get')
|
||||
def test_base_adresse_command_hourly_update(mocked_get, db, base_adresse):
|
||||
base_adresse.update_api_geo_data = lambda: None
|
||||
filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.bz2')
|
||||
mocked_get.return_value = utils.FakedResponse(content=open(filepath).read(), status_code=200)
|
||||
# check the first hourly job downloads streets
|
||||
|
|
@ -227,8 +319,10 @@ def test_base_adresse_command_hourly_update(mocked_get, db, base_adresse):
|
|||
assert mocked_get.call_count == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('mock_update_api_geo')
|
||||
@mock.patch('passerelle.utils.Request.get')
|
||||
def test_base_adresse_command_update_97x(mocked_get, db, base_adresse_97x):
|
||||
base_adresse_97x.update_api_geo_data = lambda: None
|
||||
filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.bz2')
|
||||
mocked_get.return_value = utils.FakedResponse(content=open(filepath).read(), status_code=200)
|
||||
call_command('cron', 'daily')
|
||||
|
|
@ -236,8 +330,10 @@ def test_base_adresse_command_update_97x(mocked_get, db, base_adresse_97x):
|
|||
assert StreetModel.objects.count() == 2
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('mock_update_api_geo')
|
||||
@mock.patch('passerelle.utils.Request.get')
|
||||
def test_base_adresse_command_update_corsica(mocked_get, db, base_adresse_corsica):
|
||||
base_adresse_corsica.update_api_geo_data = lambda: None
|
||||
filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.bz2')
|
||||
mocked_get.return_value = utils.FakedResponse(content=open(filepath).read(), status_code=200)
|
||||
call_command('cron', 'daily')
|
||||
|
|
@ -247,8 +343,10 @@ def test_base_adresse_command_update_corsica(mocked_get, db, base_adresse_corsic
|
|||
assert StreetModel.objects.count() == 0
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('mock_update_api_geo')
|
||||
@mock.patch('passerelle.utils.Request.get')
|
||||
def test_base_adresse_command_update_multiple(mocked_get, db, base_adresse_multiple):
|
||||
base_adresse_multiple.update_api_geo_data = lambda: None
|
||||
filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.bz2')
|
||||
mocked_get.return_value = utils.FakedResponse(content=open(filepath).read(), status_code=200)
|
||||
call_command('cron', 'daily')
|
||||
|
|
@ -258,3 +356,251 @@ def test_base_adresse_command_update_multiple(mocked_get, db, base_adresse_multi
|
|||
mocked_get.assert_any_call('http://bano.openstreetmap.fr/BAN_odbl/BAN_odbl_2A-json.bz2')
|
||||
mocked_get.assert_any_call('http://bano.openstreetmap.fr/BAN_odbl/BAN_odbl_2B-json.bz2')
|
||||
assert StreetModel.objects.count() == 5
|
||||
|
||||
|
||||
def test_base_adresse_cities(app, base_adresse, city, department, region):
|
||||
resp = app.get('/base-adresse/%s/cities?q=chambe' % base_adresse.slug)
|
||||
result = resp.json['data'][0]
|
||||
assert result['name'] == city.name
|
||||
assert result['text'] == '%s %s' % (city.zipcode, city.name)
|
||||
assert result['code'] == city.code
|
||||
assert result['zipcode'] == city.zipcode
|
||||
assert result['id'] == '%s.%s' % (city.code, city.zipcode)
|
||||
assert result['population'] == city.population
|
||||
assert result['region_code'] == city.region.code
|
||||
assert result['region_name'] == city.region.name
|
||||
assert result['department_code'] == city.department.code
|
||||
assert result['department_name'] == city.department.name
|
||||
|
||||
resp = app.get('/base-adresse/%s/cities?q=73' % base_adresse.slug)
|
||||
assert resp.json['data'][0] == result
|
||||
|
||||
resp = app.get('/base-adresse/%s/cities?code=73065' % base_adresse.slug)
|
||||
assert resp.json['data'][0] == result
|
||||
|
||||
|
||||
def test_base_adresse_cities_missing_region_and_department(app, base_adresse, miquelon):
|
||||
resp = app.get('/base-adresse/%s/cities?q=miqu' % base_adresse.slug)
|
||||
result = resp.json['data'][0]
|
||||
assert result['name'] == miquelon.name
|
||||
assert not result['department_code']
|
||||
assert not result['region_code']
|
||||
assert not result['department_name']
|
||||
assert not result['region_name']
|
||||
|
||||
|
||||
def test_base_adresse_cities_region_department(app, base_adresse, miquelon, city):
|
||||
reg = RegionModel.objects.create(name=u'IdF', code='11')
|
||||
dep = DepartmentModel.objects.create(name=u'Paris', code='75', region=reg)
|
||||
paris = CityModel.objects.create(name=u'Paris', code='75056', zipcode='75014',
|
||||
population=2000000, region=reg, department=dep)
|
||||
|
||||
resp = app.get('/base-adresse/%s/cities?department_code=73' % base_adresse.slug)
|
||||
result = resp.json['data']
|
||||
assert len(result) == 1
|
||||
assert result[0]['name'] == city.name
|
||||
|
||||
resp = app.get('/base-adresse/%s/cities?region_code=84' % base_adresse.slug)
|
||||
result = resp.json['data']
|
||||
assert len(result) == 1
|
||||
assert result[0]['name'] == city.name
|
||||
|
||||
resp = app.get('/base-adresse/%s/cities?region_code=84&department_code=75' % base_adresse.slug)
|
||||
result = resp.json['data']
|
||||
assert not result
|
||||
|
||||
|
||||
def test_base_adresse_cities_sort_order(app, base_adresse, miquelon, city):
|
||||
assert miquelon.population < city.population
|
||||
resp = app.get('/base-adresse/%s/cities' % base_adresse.slug)
|
||||
result = resp.json['data']
|
||||
assert result[0]['name'] == city.name
|
||||
assert result[1]['name'] == miquelon.name
|
||||
|
||||
|
||||
def test_base_adresse_cities_get_by_id(app, base_adresse, city):
|
||||
for i in range(1, 10):
|
||||
# create additional cities
|
||||
city.pk = None
|
||||
city.zipcode = int(city.zipcode) + i
|
||||
city.save()
|
||||
|
||||
resp = app.get('/base-adresse/%s/cities?q=cham' % base_adresse.slug)
|
||||
result = resp.json['data'][0]
|
||||
assert len(resp.json['data']) == 10
|
||||
city_id = result['id']
|
||||
|
||||
resp = app.get('/base-adresse/%s/cities?id=%s' % (base_adresse.slug, city_id))
|
||||
assert len(resp.json['data']) == 1
|
||||
result2 = resp.json['data'][0]
|
||||
assert result2['text'] == result['text']
|
||||
|
||||
# non integer id.
|
||||
resp = app.get('/base-adresse/%s/cities?id=%s' % (base_adresse.slug, 'XXX'))
|
||||
assert resp.json['err'] == 1
|
||||
|
||||
# integer but without match.
|
||||
resp = app.get('/base-adresse/%s/cities?id=%s' % (base_adresse.slug, '1.1'))
|
||||
assert len(resp.json['data']) == 0
|
||||
|
||||
|
||||
def test_base_adresse_departments(app, base_adresse, department, region):
|
||||
resp = app.get('/base-adresse/%s/departments?q=sav' % base_adresse.slug)
|
||||
result = resp.json['data'][0]
|
||||
assert result['name'] == department.name
|
||||
assert result['code'] == department.code
|
||||
assert result['id'] == department.code
|
||||
assert result['text'] == '%s %s' % (department.code, department.name)
|
||||
assert result['region_code'] == region.code
|
||||
assert result['region_name'] == region.name
|
||||
|
||||
resp = app.get('/base-adresse/%s/departments?q=73' % base_adresse.slug)
|
||||
result = resp.json['data'][0]
|
||||
assert result['name'] == department.name
|
||||
|
||||
resp = app.get('/base-adresse/%s/departments?id=%s' % (base_adresse.slug, department.code))
|
||||
result = resp.json['data'][0]
|
||||
assert result['name'] == department.name
|
||||
|
||||
|
||||
def test_base_adresse_departments_region(app, base_adresse, department):
|
||||
reg = RegionModel.objects.create(name=u'IdF', code='11')
|
||||
paris = DepartmentModel.objects.create(name=u'Paris', code='75', region=reg)
|
||||
|
||||
resp = app.get('/base-adresse/%s/departments?region_code=84' % base_adresse.slug)
|
||||
result = resp.json['data']
|
||||
assert len(result) == 1
|
||||
assert result[0]['name'] == department.name
|
||||
|
||||
|
||||
def test_base_adresse_regions(app, base_adresse, region):
|
||||
resp = app.get('/base-adresse/%s/regions?q=au' % base_adresse.slug)
|
||||
result = resp.json['data'][0]
|
||||
assert result['name'] == region.name
|
||||
assert result['code'] == region.code
|
||||
assert result['id'] == region.code
|
||||
assert result['text'] == '%s %s' % (region.code, region.name)
|
||||
|
||||
resp = app.get('/base-adresse/%s/regions?id=%s' % (base_adresse.slug, region.code))
|
||||
result = resp.json['data'][0]
|
||||
assert result['name'] == region.name
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('mock_update_streets')
|
||||
@mock.patch('passerelle.utils.Request.get')
|
||||
def test_base_adresse_command_update_geo(mocked_get, db, base_adresse):
|
||||
return_values = [utils.FakedResponse(content=content, status_code=200)
|
||||
for content in (FAKE_API_GEO_REGIONS, FAKE_API_GEO_DEPARTMENTS, FAKE_API_GEO)]
|
||||
mocked_get.side_effect = return_values
|
||||
call_command('cron', 'daily')
|
||||
assert mocked_get.call_count == 3
|
||||
mocked_get.assert_any_call(urljoin(base_adresse.api_geo_url, 'communes'))
|
||||
mocked_get.assert_any_call(urljoin(base_adresse.api_geo_url, 'regions'))
|
||||
mocked_get.assert_any_call(urljoin(base_adresse.api_geo_url, 'departements'))
|
||||
|
||||
regions = RegionModel.objects.all()
|
||||
assert regions.count() == 2
|
||||
idf = regions.get(name='Île-de-France')
|
||||
assert idf.code == '11'
|
||||
centre = regions.get(name='Bourgogne-Franche-Comté')
|
||||
assert centre.code == '27'
|
||||
|
||||
departments = DepartmentModel.objects.all()
|
||||
assert departments.count() == 2
|
||||
paris_dep = departments.get(name='Paris')
|
||||
assert paris_dep.code == '75'
|
||||
assert paris_dep.region == idf
|
||||
nievre = departments.get(name='Nièvre')
|
||||
assert nievre.code == '58'
|
||||
assert nievre.region == centre
|
||||
|
||||
cities = CityModel.objects.all()
|
||||
assert cities.count() == 3
|
||||
paris = cities.get(zipcode='75001')
|
||||
assert paris.name == 'Paris'
|
||||
assert paris.code == '75056'
|
||||
assert paris.population == 2190327
|
||||
assert paris.department.code == '75'
|
||||
assert paris.region.code == '11'
|
||||
|
||||
paris2 = cities.get(zipcode='75002')
|
||||
paris_json = paris.to_json()
|
||||
for key, value in paris2.to_json().items():
|
||||
if not key in ['id', 'text', 'zipcode']:
|
||||
assert paris_json[key] == value
|
||||
|
||||
miquelon = cities.get(zipcode='97500')
|
||||
assert miquelon.name == 'Miquelon-Langlade'
|
||||
assert miquelon.code == '97501'
|
||||
assert miquelon.population == 596
|
||||
assert not miquelon.department
|
||||
assert not miquelon.region
|
||||
|
||||
# check a new call downloads again
|
||||
mocked_get.side_effect = return_values
|
||||
call_command('cron', 'daily')
|
||||
assert mocked_get.call_count == 6
|
||||
# and doesn't delete anything
|
||||
assert CityModel.objects.count() == 3
|
||||
assert DepartmentModel.objects.count() == 2
|
||||
assert RegionModel.objects.count() == 2
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('mock_update_streets')
|
||||
@mock.patch('passerelle.utils.Request.get')
|
||||
def test_base_adresse_command_update_geo_delete(mocked_get, db, base_adresse):
|
||||
return_values = [utils.FakedResponse(content=content, status_code=200)
|
||||
for content in (FAKE_API_GEO_REGIONS, FAKE_API_GEO_DEPARTMENTS, FAKE_API_GEO)]
|
||||
mocked_get.side_effect = return_values
|
||||
call_command('cron', 'daily')
|
||||
assert CityModel.objects.count() == 3
|
||||
|
||||
new_fake_api_geo = json.dumps([FAKE_API_GEO_LIST[1]])
|
||||
return_values = [utils.FakedResponse(content=content, status_code=200)
|
||||
for content in (FAKE_API_GEO_REGIONS, FAKE_API_GEO_DEPARTMENTS, new_fake_api_geo)]
|
||||
mocked_get.side_effect = return_values
|
||||
call_command('cron', 'daily')
|
||||
assert CityModel.objects.count() == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('mock_update_streets')
|
||||
@mock.patch('passerelle.utils.Request.get')
|
||||
def test_base_adresse_command_hourly_update_geo(mocked_get, db, base_adresse):
|
||||
return_values = [utils.FakedResponse(content=content, status_code=200)
|
||||
for content in (FAKE_API_GEO_REGIONS, FAKE_API_GEO_DEPARTMENTS, FAKE_API_GEO)]
|
||||
mocked_get.side_effect = return_values
|
||||
# check the first hourly job downloads data
|
||||
call_command('cron', 'hourly')
|
||||
assert mocked_get.call_count == 3
|
||||
assert CityModel.objects.count() == 3
|
||||
# check a second call doesn't download anything
|
||||
call_command('cron', 'hourly')
|
||||
assert mocked_get.call_count == 3
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('mock_update_streets')
|
||||
@mock.patch('passerelle.utils.Request.get')
|
||||
def test_base_adresse_command_update_geo_invalid(mocked_get, db, base_adresse):
|
||||
mocked_get.return_value = utils.FakedResponse(content='{}', status_code=200)
|
||||
with pytest.raises(CommandError):
|
||||
call_command('cron', 'daily')
|
||||
assert mocked_get.call_count == 1
|
||||
assert not RegionModel.objects.exists()
|
||||
|
||||
mocked_get.return_value = utils.FakedResponse(content=FAKE_API_GEO, status_code=500)
|
||||
call_command('cron', 'daily')
|
||||
assert mocked_get.call_count == 4
|
||||
assert not RegionModel.objects.exists()
|
||||
|
||||
mocked_get.return_value = utils.FakedResponse(content='not-json', status_code=200)
|
||||
call_command('cron', 'daily')
|
||||
assert mocked_get.call_count == 7
|
||||
assert not RegionModel.objects.exists()
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('mock_update_streets')
|
||||
@mock.patch('passerelle.utils.Request.get', side_effect=ConnectionError)
|
||||
def test_base_adresse_command_update_geo_no_connection(mocked_get, db, base_adresse):
|
||||
call_command('cron', 'daily')
|
||||
assert mocked_get.call_count == 3
|
||||
assert not RegionModel.objects.exists()
|
||||
|
|
|
|||
Loading…
Reference in New Issue