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