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