0004-base_adresse-add-API-G-o-endpoints-11497.patch
passerelle/apps/base_adresse/migrations/0015_auto_20191206_1244.py | ||
---|---|---|
1 |
# -*- coding: utf-8 -*- |
|
2 |
# Generated by Django 1.11.18 on 2019-12-06 11:44 |
|
3 |
from __future__ import unicode_literals |
|
4 | ||
5 |
from django.db import migrations, models |
|
6 |
import django.db.models.deletion |
|
7 |
import passerelle.apps.base_adresse.models |
|
8 | ||
9 | ||
10 |
class Migration(migrations.Migration): |
|
11 | ||
12 |
dependencies = [ |
|
13 |
('base_adresse', '0014_auto_20190207_0456'), |
|
14 |
] |
|
15 | ||
16 |
operations = [ |
|
17 |
migrations.CreateModel( |
|
18 |
name='CityModel', |
|
19 |
fields=[ |
|
20 |
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), |
|
21 |
('name', models.CharField(max_length=150, verbose_name='City name')), |
|
22 |
('unaccent_name', models.CharField(max_length=150, null=True, verbose_name='City name ascii char')), |
|
23 |
('code', models.CharField(max_length=5, verbose_name='INSEE code')), |
|
24 |
('zipcode', models.CharField(max_length=5, verbose_name='Postal code')), |
|
25 |
('population', models.PositiveIntegerField(verbose_name='Population')), |
|
26 |
('last_update', models.DateTimeField(auto_now=True, null=True, verbose_name='Last update')), |
|
27 |
], |
|
28 |
options={ |
|
29 |
'ordering': ['-population', 'zipcode', 'unaccent_name', 'name'], |
|
30 |
}, |
|
31 |
bases=(passerelle.apps.base_adresse.models.UnaccentNameMixin, models.Model), |
|
32 |
), |
|
33 |
migrations.CreateModel( |
|
34 |
name='DepartmentModel', |
|
35 |
fields=[ |
|
36 |
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), |
|
37 |
('name', models.CharField(max_length=100, verbose_name='Department name')), |
|
38 |
('unaccent_name', models.CharField(max_length=150, null=True, verbose_name='Department name ascii char')), |
|
39 |
('code', models.CharField(max_length=3, unique=True, verbose_name='Department code')), |
|
40 |
('last_update', models.DateTimeField(auto_now=True, null=True, verbose_name='Last update')), |
|
41 |
], |
|
42 |
options={ |
|
43 |
'ordering': ['code'], |
|
44 |
}, |
|
45 |
bases=(passerelle.apps.base_adresse.models.UnaccentNameMixin, models.Model), |
|
46 |
), |
|
47 |
migrations.CreateModel( |
|
48 |
name='RegionModel', |
|
49 |
fields=[ |
|
50 |
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), |
|
51 |
('name', models.CharField(max_length=150, verbose_name='Region name')), |
|
52 |
('unaccent_name', models.CharField(max_length=150, null=True, verbose_name='Region name ascii char')), |
|
53 |
('code', models.CharField(max_length=2, unique=True, verbose_name='Region code')), |
|
54 |
('last_update', models.DateTimeField(auto_now=True, null=True, verbose_name='Last update')), |
|
55 |
], |
|
56 |
options={ |
|
57 |
'ordering': ['code'], |
|
58 |
}, |
|
59 |
bases=(passerelle.apps.base_adresse.models.UnaccentNameMixin, models.Model), |
|
60 |
), |
|
61 |
migrations.AddField( |
|
62 |
model_name='baseadresse', |
|
63 |
name='api_geo_url', |
|
64 |
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'), |
|
65 |
), |
|
66 |
migrations.AlterField( |
|
67 |
model_name='baseadresse', |
|
68 |
name='zipcode', |
|
69 |
field=models.CharField(blank=True, max_length=600, verbose_name='Postal codes or department number to get streets, separated with commas'), |
|
70 |
), |
|
71 |
migrations.AlterField( |
|
72 |
model_name='streetmodel', |
|
73 |
name='city', |
|
74 |
field=models.CharField(max_length=150, verbose_name='City'), |
|
75 |
), |
|
76 |
migrations.AddField( |
|
77 |
model_name='departmentmodel', |
|
78 |
name='region', |
|
79 |
field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='base_adresse.RegionModel'), |
|
80 |
), |
|
81 |
migrations.AddField( |
|
82 |
model_name='citymodel', |
|
83 |
name='department', |
|
84 |
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, to='base_adresse.DepartmentModel'), |
|
85 |
), |
|
86 |
migrations.AddField( |
|
87 |
model_name='citymodel', |
|
88 |
name='region', |
|
89 |
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, to='base_adresse.RegionModel'), |
|
90 |
), |
|
91 |
migrations.AlterUniqueTogether( |
|
92 |
name='citymodel', |
|
93 |
unique_together=set([('code', 'zipcode')]), |
|
94 |
), |
|
95 |
] |
passerelle/apps/base_adresse/models.py | ||
---|---|---|
1 | 1 |
import bz2 |
2 | 2 |
import json |
3 |
import os |
|
4 | 3 |
import urlparse |
5 | 4 |
import unicodedata |
6 | 5 | |
6 |
from requests import RequestException |
|
7 | 7 | |
8 | 8 |
from django.db import connection, models |
9 | 9 |
from django.db.models import Q |
... | ... | |
15 | 15 | |
16 | 16 |
from passerelle.base.models import BaseResource |
17 | 17 |
from passerelle.utils.api import endpoint |
18 |
from passerelle.utils.conversion import simplify |
|
19 |
from passerelle.utils.jsonresponse import APIError |
|
18 | 20 | |
19 | 21 | |
20 | 22 |
class BaseAdresse(BaseResource): |
... | ... | |
24 | 26 |
verbose_name=_('Service URL'), |
25 | 27 |
help_text=_('Base Adresse Web Service URL')) |
26 | 28 | |
29 |
api_geo_url = models.CharField( |
|
30 |
max_length=128, blank=False, |
|
31 |
default='https://geo.api.gouv.fr/', |
|
32 |
verbose_name=_('API Geo URL'), |
|
33 |
help_text=_('Base Adresse API Geo URL')) |
|
34 | ||
27 | 35 |
category = _('Geographic information system') |
28 | 36 | |
29 |
api_description = _("The API is a partial view of OpenStreetMap's Nominatim " |
|
30 |
"own API; it currently doesn't support all parameters and " |
|
31 |
"is limited to the JSON format.") |
|
37 |
api_description = _("The geocoding endpoints are a partial view of OpenStreetMap's " |
|
38 |
"Nominatim own API; it currently doesn't support all parameters and " |
|
39 |
"is limited to the JSON format. The cities, departments and regions " |
|
40 |
"endpoints source data from French API Geo.") |
|
32 | 41 | |
33 | 42 |
zipcode = models.CharField( |
34 | 43 |
max_length=600, |
35 | 44 |
blank=True, |
36 |
verbose_name=_('Postal codes or county number to get streets, separated with commas'))
|
|
45 |
verbose_name=_('Postal codes or department number to get streets, separated with commas'))
|
|
37 | 46 | |
38 | 47 |
class Meta: |
39 | 48 |
verbose_name = _('Base Adresse Web Service') |
... | ... | |
135 | 144 |
else: |
136 | 145 |
streets = StreetModel.objects.all() |
137 | 146 |
if q: |
138 |
unaccented_q = unicodedata.normalize('NFKD', q).encode('ascii', 'ignore').lower() |
|
139 |
streets = streets.filter(unaccent_name__icontains=unaccented_q) |
|
147 |
streets = streets.filter(unaccent_name__icontains=simplify(q)) |
|
140 | 148 | |
141 | 149 |
if zipcode: |
142 | 150 |
streets = streets.filter(zipcode__startswith=zipcode) |
... | ... | |
158 | 166 | |
159 | 167 |
return {'data': result} |
160 | 168 | |
169 |
@endpoint(description=_('Cities list'), |
|
170 |
parameters={ |
|
171 |
'id': {'description': _('Get exactly one city using its code and postal code ' |
|
172 |
'separated with a dot'), |
|
173 |
'example_value': '75056.75014'}, |
|
174 |
'q': {'description': _("Search text in name or postal code"), |
|
175 |
'example_value': 'Paris'}, |
|
176 |
'code': {'description': _('INSEE code'), 'example_value': '75056'}, |
|
177 |
'region_code': {'description': _('Region code'), 'example_value': '11'}, |
|
178 |
'department_code': {'description': _('Department code'), 'example_value': '75'}, |
|
179 |
}) |
|
180 |
def cities(self, request, id=None, q=None, code=None, region_code=None, |
|
181 |
department_code=None): |
|
182 |
cities = CityModel.objects.all() |
|
183 | ||
184 |
if id is not None: |
|
185 |
try: |
|
186 |
code, zipcode = id.split('.') |
|
187 |
except ValueError: |
|
188 |
raise APIError('Invalid id') |
|
189 |
cities = cities.filter(code=code, zipcode=zipcode) |
|
190 |
if q: |
|
191 |
unaccented_q = simplify(q) |
|
192 |
cities = cities.filter(Q(unaccent_name__istartswith=unaccented_q) | |
|
193 |
Q(zipcode__istartswith=unaccented_q)) |
|
194 |
if code: |
|
195 |
cities = cities.filter(code=code) |
|
196 |
if region_code: |
|
197 |
cities = cities.filter(region__code=region_code) |
|
198 |
if department_code: |
|
199 |
cities = cities.filter(department__code=department_code) |
|
200 | ||
201 |
cities = cities.select_related('department', 'region') |
|
202 |
return {'data': [city.to_json() for city in cities]} |
|
203 | ||
204 |
@endpoint(description=_('Departments list'), |
|
205 |
parameters={ |
|
206 |
'id': {'description': _('Get exactly one department using its code'), |
|
207 |
'example_value': '59'}, |
|
208 |
'q': {'description': _('Search text in name or code'), 'example_value': 'Nord'}, |
|
209 |
'region_code': {'description': _('Region code'), 'example_value': '32'}, |
|
210 |
}) |
|
211 |
def departments(self, request, id=None, q=None, region_code=None): |
|
212 |
departments = DepartmentModel.objects.all() |
|
213 | ||
214 |
if id is not None: |
|
215 |
departments = departments.filter(code=id) |
|
216 |
if q: |
|
217 |
unaccented_q = simplify(q) |
|
218 |
departments = departments.filter(Q(unaccent_name__istartswith=unaccented_q) | |
|
219 |
Q(code__istartswith=unaccented_q)) |
|
220 |
if region_code: |
|
221 |
departments = departments.filter(region__code=region_code) |
|
222 | ||
223 |
departments = departments.select_related('region') |
|
224 |
return {'data': [department.to_json() for department in departments]} |
|
225 | ||
226 |
@endpoint(description=_('Regions list'), |
|
227 |
parameters={ |
|
228 |
'id': {'description': _('Get exactly one region using its code'), |
|
229 |
'example_value': '32'}, |
|
230 |
'q': {'description': _('Search text in name or code'), |
|
231 |
'example_value': 'Hauts-de-France'}, |
|
232 |
}) |
|
233 |
def regions(self, request, id=None, q=None): |
|
234 |
regions = RegionModel.objects.all() |
|
235 | ||
236 |
if id is not None: |
|
237 |
regions = regions.filter(code=id) |
|
238 |
if q: |
|
239 |
unaccented_q = simplify(q) |
|
240 |
regions = regions.filter(Q(unaccent_name__istartswith=unaccented_q) | |
|
241 |
Q(code__istartswith=unaccented_q)) |
|
242 | ||
243 |
return {'data': [region.to_json() for region in regions]} |
|
244 | ||
161 | 245 |
def check_status(self): |
162 | 246 |
if self.service_url == 'https://api-adresse.data.gouv.fr/': |
163 | 247 |
result = self.search(None, '169 rue du chateau, paris') |
... | ... | |
174 | 258 |
criteria |= Q(zipcode__startswith=zipcode) |
175 | 259 |
return StreetModel.objects.filter(criteria) |
176 | 260 | |
261 |
def cities_exist(self): |
|
262 |
return CityModel.objects.exists() |
|
263 | ||
177 | 264 |
def update_streets_data(self): |
178 | 265 |
if not self.get_zipcodes(): |
179 | 266 |
return |
... | ... | |
218 | 305 | |
219 | 306 |
self.get_streets_queryset().filter(last_update__lt=start_update).delete() |
220 | 307 | |
308 |
def get_api_geo_endpoint(self, endpoint): |
|
309 |
if not self.api_geo_url: |
|
310 |
return |
|
311 |
error = None |
|
312 |
try: |
|
313 |
response = self.requests.get(urljoin(self.api_geo_url, endpoint)) |
|
314 |
except RequestException as e: |
|
315 |
error = e |
|
316 |
else: |
|
317 |
if response.status_code != 200: |
|
318 |
error = 'bad status code (%s)' % response.status_code |
|
319 |
else: |
|
320 |
try: |
|
321 |
result = response.json() |
|
322 |
except ValueError: |
|
323 |
error = 'invalid json, got: %s' % response.text |
|
324 |
if error: |
|
325 |
self.logger.error('failed to update api geo data for endpoint %s: %s', |
|
326 |
endpoint, error) |
|
327 |
return |
|
328 |
if not result: |
|
329 |
raise Exception('api geo returns empty json') |
|
330 |
return result |
|
331 | ||
332 |
def update_api_geo_data(self): |
|
333 |
regions_json = self.get_api_geo_endpoint('regions') |
|
334 |
departments_json = self.get_api_geo_endpoint('departements') |
|
335 |
cities_json = self.get_api_geo_endpoint('communes') |
|
336 |
if not (regions_json and departments_json and cities_json): |
|
337 |
return |
|
338 |
start_update = timezone.now() |
|
339 | ||
340 |
for data in regions_json: |
|
341 |
defaults = { |
|
342 |
'name': data['nom'], |
|
343 |
} |
|
344 |
RegionModel.objects.update_or_create(code=data['code'], defaults=defaults) |
|
345 |
RegionModel.objects.filter(last_update__lt=start_update).delete() |
|
346 | ||
347 |
for data in departments_json: |
|
348 |
defaults = { |
|
349 |
'name': data['nom'], |
|
350 |
'region': RegionModel.objects.get(code=data['codeRegion']), |
|
351 |
} |
|
352 |
DepartmentModel.objects.update_or_create(code=data['code'], defaults=defaults) |
|
353 |
DepartmentModel.objects.filter(last_update__lt=start_update).delete() |
|
354 | ||
355 |
for data in cities_json: |
|
356 |
for zipcode in data['codesPostaux']: |
|
357 |
defaults = { |
|
358 |
'name': data['nom'], |
|
359 |
'population': data.get('population', 0), |
|
360 |
} |
|
361 |
if data.get('codeDepartement'): |
|
362 |
defaults['department'] = DepartmentModel.objects.get(code=data['codeDepartement']) |
|
363 |
if data.get('codeRegion'): |
|
364 |
defaults['region'] = RegionModel.objects.get(code=data['codeRegion']) |
|
365 |
CityModel.objects.update_or_create( |
|
366 |
code=data['code'], zipcode=zipcode, defaults=defaults) |
|
367 |
CityModel.objects.filter(last_update__lt=start_update).delete() |
|
368 | ||
221 | 369 |
def hourly(self): |
222 | 370 |
super(BaseAdresse, self).hourly() |
371 |
# don't wait for daily job to grab data |
|
223 | 372 |
if self.get_zipcodes() and not self.get_streets_queryset().exists(): |
224 |
# don't wait for daily job to grab streets |
|
225 | 373 |
self.update_streets_data() |
374 |
if not CityModel.objects.exists(): |
|
375 |
self.update_api_geo_data() |
|
226 | 376 | |
227 | 377 |
def daily(self): |
228 | 378 |
super(BaseAdresse, self).daily() |
229 | 379 |
self.update_streets_data() |
380 |
self.update_api_geo_data() |
|
230 | 381 | |
231 | 382 | |
232 |
class StreetModel(models.Model):
|
|
383 |
class UnaccentNameMixin(object):
|
|
233 | 384 | |
234 |
city = models.CharField(_('City'), max_length=100) |
|
385 |
def save(self, *args, **kwargs): |
|
386 |
self.unaccent_name = unicodedata.normalize('NFKD', self.name).encode('ascii', 'ignore').lower() |
|
387 |
super(UnaccentNameMixin, self).save(*args, **kwargs) |
|
388 | ||
389 | ||
390 |
class StreetModel(UnaccentNameMixin, models.Model): |
|
391 | ||
392 |
city = models.CharField(_('City'), max_length=150) |
|
235 | 393 |
name = models.CharField(_('Street name'), max_length=150) |
236 | 394 |
unaccent_name = models.CharField(_('Street name ascii char'), max_length=150, null=True) |
237 | 395 |
zipcode = models.CharField(_('Postal code'), max_length=5) |
... | ... | |
245 | 403 |
def __unicode__(self): |
246 | 404 |
return self.name |
247 | 405 | |
248 |
def save(self, *args, **kwargs): |
|
249 |
self.unaccent_name = unicodedata.normalize('NFKD', self.name).encode('ascii', 'ignore') |
|
250 |
super(StreetModel, self).save(*args, **kwargs) |
|
406 | ||
407 |
@six.python_2_unicode_compatible |
|
408 |
class RegionModel(UnaccentNameMixin, models.Model): |
|
409 | ||
410 |
name = models.CharField(_('Region name'), max_length=150) |
|
411 |
unaccent_name = models.CharField(_('Region name ascii char'), max_length=150, null=True) |
|
412 |
code = models.CharField(_('Region code'), max_length=2, unique=True) |
|
413 |
last_update = models.DateTimeField(_('Last update'), null=True, auto_now=True) |
|
414 | ||
415 |
def to_json(self): |
|
416 |
return { |
|
417 |
'text': str(self), |
|
418 |
'id': self.code, |
|
419 |
'code': self.code, |
|
420 |
'name': self.name, |
|
421 |
} |
|
422 | ||
423 |
class Meta: |
|
424 |
ordering = ['code'] |
|
425 | ||
426 |
def __str__(self): |
|
427 |
return '%s %s' % (self.code, self.name) |
|
428 | ||
429 | ||
430 |
@six.python_2_unicode_compatible |
|
431 |
class DepartmentModel(UnaccentNameMixin, models.Model): |
|
432 | ||
433 |
name = models.CharField(_('Department name'), max_length=100) |
|
434 |
unaccent_name = models.CharField(_('Department name ascii char'), max_length=150, null=True) |
|
435 |
code = models.CharField(_('Department code'), max_length=3, unique=True) |
|
436 |
region = models.ForeignKey(RegionModel, on_delete=models.CASCADE) |
|
437 |
last_update = models.DateTimeField(_('Last update'), null=True, auto_now=True) |
|
438 | ||
439 |
def to_json(self): |
|
440 |
return { |
|
441 |
'text': str(self), |
|
442 |
'id': self.code, |
|
443 |
'code': self.code, |
|
444 |
'name': self.name, |
|
445 |
'region_code': self.region.code, |
|
446 |
} |
|
447 | ||
448 |
class Meta: |
|
449 |
ordering = ['code'] |
|
450 | ||
451 |
def __str__(self): |
|
452 |
return '%s %s' % (self.code, self.name) |
|
453 | ||
454 | ||
455 |
@six.python_2_unicode_compatible |
|
456 |
class CityModel(UnaccentNameMixin, models.Model): |
|
457 | ||
458 |
name = models.CharField(_('City name'), max_length=150) |
|
459 |
unaccent_name = models.CharField(_('City name ascii char'), max_length=150, null=True) |
|
460 |
code = models.CharField(_('INSEE code'), max_length=5) |
|
461 |
zipcode = models.CharField(_('Postal code'), max_length=5) |
|
462 |
population = models.PositiveIntegerField(_('Population')) |
|
463 |
department = models.ForeignKey(DepartmentModel, on_delete=models.CASCADE, blank=True, null=True) |
|
464 |
region = models.ForeignKey(RegionModel, on_delete=models.CASCADE, blank=True, null=True) |
|
465 |
last_update = models.DateTimeField(_('Last update'), null=True, auto_now=True) |
|
466 | ||
467 |
def to_json(self): |
|
468 |
data = { |
|
469 |
'text': str(self), |
|
470 |
'id': '%s.%s' % (self.code, self.zipcode), |
|
471 |
'code': self.code, |
|
472 |
'name': self.name, |
|
473 |
'zipcode': self.zipcode, |
|
474 |
'population': self.population, |
|
475 |
'department_code': self.department.code if self.department else None, |
|
476 |
'region_code': self.region.code if self.region else None, |
|
477 |
} |
|
478 |
return data |
|
479 | ||
480 |
class Meta: |
|
481 |
ordering = ['-population', 'zipcode', 'unaccent_name', 'name'] |
|
482 |
unique_together = ('code', 'zipcode') |
|
483 | ||
484 |
def __str__(self): |
|
485 |
return '%s %s' % (self.zipcode, self.name) |
passerelle/apps/base_adresse/templates/base_adresse/baseadresse_detail.html | ||
---|---|---|
8 | 8 |
{% trans "Street data is not available yet, it should soon be downloaded." %} |
9 | 9 |
</div> |
10 | 10 |
{% endif %} |
11 |
{% if object.api_geo_url and not object.cities_exist %} |
|
12 |
<div class="infonotice"> |
|
13 |
{% trans "API Géo data is not available yet, it should soon be downloaded." %} |
|
14 |
</div> |
|
15 |
{% endif %} |
|
11 | 16 |
{% endblock %} |
12 | 17 | |
13 | 18 |
{% block security %} |
tests/test_base_adresse.py | ||
---|---|---|
6 | 6 |
import utils |
7 | 7 |
import json |
8 | 8 | |
9 |
from requests.exceptions import ConnectionError |
|
10 | ||
9 | 11 |
from django.core.management import call_command |
12 |
from django.core.management.base import CommandError |
|
13 |
from django.utils.six.moves.urllib.parse import urljoin |
|
10 | 14 | |
11 |
from passerelle.apps.base_adresse.models import BaseAdresse, StreetModel |
|
15 |
from passerelle.apps.base_adresse.models import (BaseAdresse, StreetModel, CityModel, |
|
16 |
DepartmentModel, RegionModel) |
|
12 | 17 | |
13 | 18 |
FAKED_CONTENT = json.dumps({ |
14 | 19 |
"limit": 1, |
... | ... | |
41 | 46 | |
42 | 47 |
FAKE_DATA = '' |
43 | 48 | |
49 |
FAKE_API_GEO_LIST = [ |
|
50 |
{ |
|
51 |
"code": "75056", |
|
52 |
"codeDepartement": "75", |
|
53 |
"codeRegion": "11", |
|
54 |
"codesPostaux": [ |
|
55 |
"75001", |
|
56 |
"75002", |
|
57 |
], |
|
58 |
"nom": "Paris", |
|
59 |
"population": 2190327, |
|
60 |
}, |
|
61 |
{ |
|
62 |
"code": "97501", |
|
63 |
"codesPostaux": [ |
|
64 |
"97500" |
|
65 |
], |
|
66 |
"nom": "Miquelon-Langlade", |
|
67 |
"population": 596 |
|
68 |
} |
|
69 |
] |
|
70 | ||
71 |
FAKE_API_GEO = json.dumps(FAKE_API_GEO_LIST) |
|
72 | ||
73 |
FAKE_API_GEO_DEPARTMENTS = json.dumps([ |
|
74 |
{ |
|
75 |
"code": "75", |
|
76 |
"codeRegion": "11", |
|
77 |
"nom": "Paris" |
|
78 |
}, |
|
79 |
{ |
|
80 |
"code": "58", |
|
81 |
"codeRegion": "27", |
|
82 |
"nom": "Nièvre", |
|
83 |
} |
|
84 |
]) |
|
85 | ||
86 |
FAKE_API_GEO_REGIONS = json.dumps([ |
|
87 |
{ |
|
88 |
"code": "11", |
|
89 |
"nom": "Île-de-France" |
|
90 |
}, |
|
91 |
{ |
|
92 |
"code": "27", |
|
93 |
"nom": "Bourgogne-Franche-Comté" |
|
94 |
} |
|
95 |
]) |
|
96 | ||
44 | 97 | |
45 | 98 |
@pytest.fixture |
46 | 99 |
def base_adresse(db): |
... | ... | |
74 | 127 |
citycode=u'73001') |
75 | 128 | |
76 | 129 | |
130 |
@pytest.fixture |
|
131 |
def region(db): |
|
132 |
return RegionModel.objects.create(name=u'Auvergne-Rhône-Alpes', code='84') |
|
133 | ||
134 | ||
135 |
@pytest.fixture |
|
136 |
def department(db, region): |
|
137 |
return DepartmentModel.objects.create(name=u'Savoie', code='73', region=region) |
|
138 | ||
139 | ||
140 |
@pytest.fixture |
|
141 |
def city(db, region, department): |
|
142 |
return CityModel.objects.create(name=u'Chambéry', code='73065', zipcode='73000', |
|
143 |
population=42000, region=region, department=department) |
|
144 | ||
145 | ||
146 |
@pytest.fixture |
|
147 |
def miquelon(db): |
|
148 |
return CityModel.objects.create(name=u'Miquelon-Langlade', code='97501', zipcode='97500', |
|
149 |
population=42) |
|
150 | ||
151 | ||
152 |
@pytest.fixture |
|
153 |
def mock_update_api_geo(): |
|
154 |
with mock.patch('passerelle.apps.base_adresse.models.BaseAdresse.update_api_geo_data', |
|
155 |
new=lambda x: None) as _fixture: |
|
156 |
yield _fixture |
|
157 | ||
158 | ||
159 |
@pytest.fixture |
|
160 |
def mock_update_streets(): |
|
161 |
with mock.patch('passerelle.apps.base_adresse.models.BaseAdresse.update_streets_data', |
|
162 |
new=lambda x: None) as _fixture: |
|
163 |
yield _fixture |
|
164 | ||
165 | ||
77 | 166 |
@mock.patch('passerelle.utils.Request.get') |
78 | 167 |
def test_base_adresse_search(mocked_get, app, base_adresse): |
79 | 168 |
endpoint = utils.generic_endpoint_url('base-adresse', 'search', slug=base_adresse.slug) |
... | ... | |
195 | 284 |
assert len(resp.json['data']) == 0 |
196 | 285 | |
197 | 286 | |
287 |
@pytest.mark.usefixtures('mock_update_api_geo') |
|
198 | 288 |
@mock.patch('passerelle.utils.Request.get') |
199 | 289 |
def test_base_adresse_command_update(mocked_get, db, base_adresse): |
200 | 290 |
filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.bz2') |
... | ... | |
214 | 304 |
assert mocked_get.call_count == 2 |
215 | 305 | |
216 | 306 | |
307 |
@pytest.mark.usefixtures('mock_update_api_geo') |
|
217 | 308 |
@mock.patch('passerelle.utils.Request.get') |
218 | 309 |
def test_base_adresse_command_hourly_update(mocked_get, db, base_adresse): |
310 |
base_adresse.update_api_geo_data = lambda: None |
|
219 | 311 |
filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.bz2') |
220 | 312 |
mocked_get.return_value = utils.FakedResponse(content=open(filepath).read(), status_code=200) |
221 | 313 |
# check the first hourly job downloads streets |
... | ... | |
227 | 319 |
assert mocked_get.call_count == 1 |
228 | 320 | |
229 | 321 | |
322 |
@pytest.mark.usefixtures('mock_update_api_geo') |
|
230 | 323 |
@mock.patch('passerelle.utils.Request.get') |
231 | 324 |
def test_base_adresse_command_update_97x(mocked_get, db, base_adresse_97x): |
325 |
base_adresse_97x.update_api_geo_data = lambda: None |
|
232 | 326 |
filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.bz2') |
233 | 327 |
mocked_get.return_value = utils.FakedResponse(content=open(filepath).read(), status_code=200) |
234 | 328 |
call_command('cron', 'daily') |
... | ... | |
236 | 330 |
assert StreetModel.objects.count() == 2 |
237 | 331 | |
238 | 332 | |
333 |
@pytest.mark.usefixtures('mock_update_api_geo') |
|
239 | 334 |
@mock.patch('passerelle.utils.Request.get') |
240 | 335 |
def test_base_adresse_command_update_corsica(mocked_get, db, base_adresse_corsica): |
336 |
base_adresse_corsica.update_api_geo_data = lambda: None |
|
241 | 337 |
filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.bz2') |
242 | 338 |
mocked_get.return_value = utils.FakedResponse(content=open(filepath).read(), status_code=200) |
243 | 339 |
call_command('cron', 'daily') |
... | ... | |
247 | 343 |
assert StreetModel.objects.count() == 0 |
248 | 344 | |
249 | 345 | |
346 |
@pytest.mark.usefixtures('mock_update_api_geo') |
|
250 | 347 |
@mock.patch('passerelle.utils.Request.get') |
251 | 348 |
def test_base_adresse_command_update_multiple(mocked_get, db, base_adresse_multiple): |
349 |
base_adresse_multiple.update_api_geo_data = lambda: None |
|
252 | 350 |
filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.bz2') |
253 | 351 |
mocked_get.return_value = utils.FakedResponse(content=open(filepath).read(), status_code=200) |
254 | 352 |
call_command('cron', 'daily') |
... | ... | |
258 | 356 |
mocked_get.assert_any_call('http://bano.openstreetmap.fr/BAN_odbl/BAN_odbl_2A-json.bz2') |
259 | 357 |
mocked_get.assert_any_call('http://bano.openstreetmap.fr/BAN_odbl/BAN_odbl_2B-json.bz2') |
260 | 358 |
assert StreetModel.objects.count() == 5 |
359 | ||
360 | ||
361 |
def test_base_adresse_cities(app, base_adresse, city, department, region): |
|
362 |
resp = app.get('/base-adresse/%s/cities?q=chambe' % base_adresse.slug) |
|
363 |
result = resp.json['data'][0] |
|
364 |
assert result['name'] == city.name |
|
365 |
assert result['text'] == '%s %s' % (city.zipcode, city.name) |
|
366 |
assert result['code'] == city.code |
|
367 |
assert result['zipcode'] == city.zipcode |
|
368 |
assert result['id'] == '%s.%s' % (city.code, city.zipcode) |
|
369 |
assert result['population'] == city.population |
|
370 |
assert result['region_code'] == city.region.code |
|
371 |
assert result['department_code'] == city.department.code |
|
372 | ||
373 |
resp = app.get('/base-adresse/%s/cities?q=73' % base_adresse.slug) |
|
374 |
assert resp.json['data'][0] == result |
|
375 | ||
376 |
resp = app.get('/base-adresse/%s/cities?code=73065' % base_adresse.slug) |
|
377 |
assert resp.json['data'][0] == result |
|
378 | ||
379 | ||
380 |
def test_base_adresse_cities_missing_region_and_department(app, base_adresse, miquelon): |
|
381 |
resp = app.get('/base-adresse/%s/cities?q=miqu' % base_adresse.slug) |
|
382 |
result = resp.json['data'][0] |
|
383 |
assert result['name'] == miquelon.name |
|
384 |
assert not result['department_code'] |
|
385 |
assert not result['region_code'] |
|
386 | ||
387 | ||
388 |
def test_base_adresse_cities_region_department(app, base_adresse, miquelon, city): |
|
389 |
reg = RegionModel.objects.create(name=u'IdF', code='11') |
|
390 |
dep = DepartmentModel.objects.create(name=u'Paris', code='75', region=reg) |
|
391 |
paris = CityModel.objects.create(name=u'Paris', code='75056', zipcode='75014', |
|
392 |
population=2000000, region=reg, department=dep) |
|
393 | ||
394 |
resp = app.get('/base-adresse/%s/cities?department_code=73' % base_adresse.slug) |
|
395 |
result = resp.json['data'] |
|
396 |
assert len(result) == 1 |
|
397 |
assert result[0]['name'] == city.name |
|
398 | ||
399 |
resp = app.get('/base-adresse/%s/cities?region_code=84' % base_adresse.slug) |
|
400 |
result = resp.json['data'] |
|
401 |
assert len(result) == 1 |
|
402 |
assert result[0]['name'] == city.name |
|
403 | ||
404 |
resp = app.get('/base-adresse/%s/cities?region_code=84&department_code=75' % base_adresse.slug) |
|
405 |
result = resp.json['data'] |
|
406 |
assert not result |
|
407 | ||
408 | ||
409 |
def test_base_adresse_cities_sort_order(app, base_adresse, miquelon, city): |
|
410 |
assert miquelon.population < city.population |
|
411 |
resp = app.get('/base-adresse/%s/cities' % base_adresse.slug) |
|
412 |
result = resp.json['data'] |
|
413 |
assert result[0]['name'] == city.name |
|
414 |
assert result[1]['name'] == miquelon.name |
|
415 | ||
416 | ||
417 |
def test_base_adresse_cities_get_by_id(app, base_adresse, city): |
|
418 |
for i in range(1, 10): |
|
419 |
# create additional cities |
|
420 |
city.pk = None |
|
421 |
city.zipcode = int(city.zipcode) + i |
|
422 |
city.save() |
|
423 | ||
424 |
resp = app.get('/base-adresse/%s/cities?q=cham' % base_adresse.slug) |
|
425 |
result = resp.json['data'][0] |
|
426 |
assert len(resp.json['data']) == 10 |
|
427 |
city_id = result['id'] |
|
428 | ||
429 |
resp = app.get('/base-adresse/%s/cities?id=%s' % (base_adresse.slug, city_id)) |
|
430 |
assert len(resp.json['data']) == 1 |
|
431 |
result2 = resp.json['data'][0] |
|
432 |
assert result2['text'] == result['text'] |
|
433 | ||
434 |
# non integer id. |
|
435 |
resp = app.get('/base-adresse/%s/cities?id=%s' % (base_adresse.slug, 'XXX')) |
|
436 |
assert resp.json['err'] == 1 |
|
437 | ||
438 |
# integer but without match. |
|
439 |
resp = app.get('/base-adresse/%s/cities?id=%s' % (base_adresse.slug, '1.1')) |
|
440 |
assert len(resp.json['data']) == 0 |
|
441 | ||
442 | ||
443 |
def test_base_adresse_departments(app, base_adresse, department, region): |
|
444 |
resp = app.get('/base-adresse/%s/departments?q=sav' % base_adresse.slug) |
|
445 |
result = resp.json['data'][0] |
|
446 |
assert result['name'] == department.name |
|
447 |
assert result['code'] == department.code |
|
448 |
assert result['id'] == department.code |
|
449 |
assert result['text'] == '%s %s' % (department.code, department.name) |
|
450 |
assert result['region_code'] == region.code |
|
451 | ||
452 |
resp = app.get('/base-adresse/%s/departments?q=73' % base_adresse.slug) |
|
453 |
result = resp.json['data'][0] |
|
454 |
assert result['name'] == department.name |
|
455 | ||
456 |
resp = app.get('/base-adresse/%s/departments?id=%s' % (base_adresse.slug, department.code)) |
|
457 |
result = resp.json['data'][0] |
|
458 |
assert result['name'] == department.name |
|
459 | ||
460 | ||
461 |
def test_base_adresse_departments_region(app, base_adresse, department): |
|
462 |
reg = RegionModel.objects.create(name=u'IdF', code='11') |
|
463 |
paris = DepartmentModel.objects.create(name=u'Paris', code='75', region=reg) |
|
464 | ||
465 |
resp = app.get('/base-adresse/%s/departments?region_code=84' % base_adresse.slug) |
|
466 |
result = resp.json['data'] |
|
467 |
assert len(result) == 1 |
|
468 |
assert result[0]['name'] == department.name |
|
469 | ||
470 | ||
471 |
def test_base_adresse_regions(app, base_adresse, region): |
|
472 |
resp = app.get('/base-adresse/%s/regions?q=au' % base_adresse.slug) |
|
473 |
result = resp.json['data'][0] |
|
474 |
assert result['name'] == region.name |
|
475 |
assert result['code'] == region.code |
|
476 |
assert result['id'] == region.code |
|
477 |
assert result['text'] == '%s %s' % (region.code, region.name) |
|
478 | ||
479 |
resp = app.get('/base-adresse/%s/regions?id=%s' % (base_adresse.slug, region.code)) |
|
480 |
result = resp.json['data'][0] |
|
481 |
assert result['name'] == region.name |
|
482 | ||
483 | ||
484 |
@pytest.mark.usefixtures('mock_update_streets') |
|
485 |
@mock.patch('passerelle.utils.Request.get') |
|
486 |
def test_base_adresse_command_update_geo(mocked_get, db, base_adresse): |
|
487 |
return_values = [utils.FakedResponse(content=content, status_code=200) |
|
488 |
for content in (FAKE_API_GEO_REGIONS, FAKE_API_GEO_DEPARTMENTS, FAKE_API_GEO)] |
|
489 |
mocked_get.side_effect = return_values |
|
490 |
call_command('cron', 'daily') |
|
491 |
assert mocked_get.call_count == 3 |
|
492 |
mocked_get.assert_any_call(urljoin(base_adresse.api_geo_url, 'communes')) |
|
493 |
mocked_get.assert_any_call(urljoin(base_adresse.api_geo_url, 'regions')) |
|
494 |
mocked_get.assert_any_call(urljoin(base_adresse.api_geo_url, 'departements')) |
|
495 | ||
496 |
regions = RegionModel.objects.all() |
|
497 |
assert regions.count() == 2 |
|
498 |
idf = regions.get(name='Île-de-France') |
|
499 |
assert idf.code == '11' |
|
500 |
centre = regions.get(name='Bourgogne-Franche-Comté') |
|
501 |
assert centre.code == '27' |
|
502 | ||
503 |
departments = DepartmentModel.objects.all() |
|
504 |
assert departments.count() == 2 |
|
505 |
paris_dep = departments.get(name='Paris') |
|
506 |
assert paris_dep.code == '75' |
|
507 |
assert paris_dep.region == idf |
|
508 |
nievre = departments.get(name='Nièvre') |
|
509 |
assert nievre.code == '58' |
|
510 |
assert nievre.region == centre |
|
511 | ||
512 |
cities = CityModel.objects.all() |
|
513 |
assert cities.count() == 3 |
|
514 |
paris = cities.get(zipcode='75001') |
|
515 |
assert paris.name == 'Paris' |
|
516 |
assert paris.code == '75056' |
|
517 |
assert paris.population == 2190327 |
|
518 |
assert paris.department.code == '75' |
|
519 |
assert paris.region.code == '11' |
|
520 | ||
521 |
paris2 = cities.get(zipcode='75002') |
|
522 |
paris_json = paris.to_json() |
|
523 |
for key, value in paris2.to_json().items(): |
|
524 |
if not key in ['id', 'text', 'zipcode']: |
|
525 |
assert paris_json[key] == value |
|
526 | ||
527 |
miquelon = cities.get(zipcode='97500') |
|
528 |
assert miquelon.name == 'Miquelon-Langlade' |
|
529 |
assert miquelon.code == '97501' |
|
530 |
assert miquelon.population == 596 |
|
531 |
assert not miquelon.department |
|
532 |
assert not miquelon.region |
|
533 | ||
534 |
# check a new call downloads again |
|
535 |
mocked_get.side_effect = return_values |
|
536 |
call_command('cron', 'daily') |
|
537 |
assert mocked_get.call_count == 6 |
|
538 |
# and doesn't delete anything |
|
539 |
assert CityModel.objects.count() == 3 |
|
540 |
assert DepartmentModel.objects.count() == 2 |
|
541 |
assert RegionModel.objects.count() == 2 |
|
542 | ||
543 | ||
544 |
@pytest.mark.usefixtures('mock_update_streets') |
|
545 |
@mock.patch('passerelle.utils.Request.get') |
|
546 |
def test_base_adresse_command_update_geo_delete(mocked_get, db, base_adresse): |
|
547 |
return_values = [utils.FakedResponse(content=content, status_code=200) |
|
548 |
for content in (FAKE_API_GEO_REGIONS, FAKE_API_GEO_DEPARTMENTS, FAKE_API_GEO)] |
|
549 |
mocked_get.side_effect = return_values |
|
550 |
call_command('cron', 'daily') |
|
551 |
assert CityModel.objects.count() == 3 |
|
552 | ||
553 |
new_fake_api_geo = json.dumps([FAKE_API_GEO_LIST[1]]) |
|
554 |
return_values = [utils.FakedResponse(content=content, status_code=200) |
|
555 |
for content in (FAKE_API_GEO_REGIONS, FAKE_API_GEO_DEPARTMENTS, new_fake_api_geo)] |
|
556 |
mocked_get.side_effect = return_values |
|
557 |
call_command('cron', 'daily') |
|
558 |
assert CityModel.objects.count() == 1 |
|
559 | ||
560 | ||
561 |
@pytest.mark.usefixtures('mock_update_streets') |
|
562 |
@mock.patch('passerelle.utils.Request.get') |
|
563 |
def test_base_adresse_command_hourly_update_geo(mocked_get, db, base_adresse): |
|
564 |
return_values = [utils.FakedResponse(content=content, status_code=200) |
|
565 |
for content in (FAKE_API_GEO_REGIONS, FAKE_API_GEO_DEPARTMENTS, FAKE_API_GEO)] |
|
566 |
mocked_get.side_effect = return_values |
|
567 |
# check the first hourly job downloads data |
|
568 |
call_command('cron', 'hourly') |
|
569 |
assert mocked_get.call_count == 3 |
|
570 |
assert CityModel.objects.count() == 3 |
|
571 |
# check a second call doesn't download anything |
|
572 |
call_command('cron', 'hourly') |
|
573 |
assert mocked_get.call_count == 3 |
|
574 | ||
575 | ||
576 |
@pytest.mark.usefixtures('mock_update_streets') |
|
577 |
@mock.patch('passerelle.utils.Request.get') |
|
578 |
def test_base_adresse_command_update_geo_invalid(mocked_get, db, base_adresse): |
|
579 |
mocked_get.return_value = utils.FakedResponse(content='{}', status_code=200) |
|
580 |
with pytest.raises(CommandError): |
|
581 |
call_command('cron', 'daily') |
|
582 |
assert mocked_get.call_count == 1 |
|
583 |
assert not RegionModel.objects.exists() |
|
584 | ||
585 |
mocked_get.return_value = utils.FakedResponse(content=FAKE_API_GEO, status_code=500) |
|
586 |
call_command('cron', 'daily') |
|
587 |
assert mocked_get.call_count == 4 |
|
588 |
assert not RegionModel.objects.exists() |
|
589 | ||
590 |
mocked_get.return_value = utils.FakedResponse(content='not-json', status_code=200) |
|
591 |
call_command('cron', 'daily') |
|
592 |
assert mocked_get.call_count == 7 |
|
593 |
assert not RegionModel.objects.exists() |
|
594 | ||
595 | ||
596 |
@pytest.mark.usefixtures('mock_update_streets') |
|
597 |
@mock.patch('passerelle.utils.Request.get', side_effect=ConnectionError) |
|
598 |
def test_base_adresse_command_update_geo_no_connection(mocked_get, db, base_adresse): |
|
599 |
call_command('cron', 'daily') |
|
600 |
assert mocked_get.call_count == 3 |
|
601 |
assert not RegionModel.objects.exists() |
|
261 |
- |