entrouvert
/
passerelle
15
0
Fork 0
Code Issues Pull Requests 14 Releases Activity
passerelle/tests/test_base_adresse.py

1231 lines
48 KiB
Python
Raw Permalink Blame History

import datetime
import json
import os
from unittest import mock
from urllib.parse import parse_qs, urljoin
import pytest
import responses
from django.core.management import call_command
from django.core.management.base import CommandError
from requests.exceptions import ConnectionError, HTTPError
from responses.registries import OrderedRegistry
import tests.utils
from passerelle.apps.base_adresse.models import (
AddressCacheModel,
BaseAdresse,
CityModel,
DepartmentModel,
RegionModel,
StreetModel,
)
FAKED_CONTENT = json.dumps(
{
'limit': 1,
'attribution': 'BAN',
'version': 'draft',
'licence': 'ODbL 1.0',
'query': 'plop',
'type': 'FeatureCollection',
'features': [
{
'geometry': {'type': 'Point', 'coordinates': [-0.593775, 47.474633]},
'properties': {
'citycode': '49007',
'name': 'Rue Roger Halope',
'id': '49007_6950_be54bd',
'city': 'Angers',
'context': '49, Maine-et-Loire, Pays de la Loire',
'score': 0.14097272727272728,
'label': 'Rue Roger Halope 49000 Angers',
'postcode': '49000',
'type': 'street',
'info1': 'xxx',
'info2': 'yyy',
},
'type': 'Feature',
}
],
}
)
FAKE_DATA = ''
FAKE_API_GEO_LIST = [
{
'code': '75056',
'codeDepartement': '75',
'codeRegion': '11',
'codesPostaux': [
'75001',
'75002',
],
'nom': 'Paris',
'population': 2190327,
},
{'code': '97501', 'codesPostaux': ['97500'], 'nom': 'Miquelon-Langlade', 'population': 596},
]
FAKE_API_GEO = json.dumps(FAKE_API_GEO_LIST)
FAKE_API_GEO_DEPARTMENTS = json.dumps(
[
{'code': '75', 'codeRegion': '11', 'nom': 'Paris'},
{
'code': '58',
'codeRegion': '27',
'nom': 'Nièvre',
},
]
)
FAKE_API_GEO_REGIONS = json.dumps(
[{'code': '11', 'nom': 'Île-de-France'}, {'code': '27', 'nom': 'Bourgogne-Franche-Comté'}]
)
@pytest.fixture
def base_adresse(db):
return tests.utils.setup_access_rights(BaseAdresse.objects.create(slug='base-adresse', zipcode='73'))
@pytest.fixture
def base_adresse_97x(db):
return tests.utils.setup_access_rights(
BaseAdresse.objects.create(slug='base-adresse-97x', zipcode='97425')
)
@pytest.fixture
def base_adresse_corsica(db):
return tests.utils.setup_access_rights(
BaseAdresse.objects.create(slug='base-adresse', zipcode='20000, 20100 ')
)
@pytest.fixture
def base_adresse_multiple(db):
return tests.utils.setup_access_rights(
BaseAdresse.objects.create(slug='base-adresse', zipcode='73, 73100, 97425,20000 ')
)
@pytest.fixture
def base_adresse_coordinates(db):
return tests.utils.setup_access_rights(
BaseAdresse.objects.create(slug='base-adresse', latitude=1.2, longitude=2.1)
)
@pytest.fixture
def street(db):
return StreetModel.objects.create(
ban_id='73001_0000',
city='Chambéry',
name='Une rüê très äccentuéè',
zipcode='73000',
type='street',
citycode='73001',
resource=BaseAdresse.objects.first(),
)
@pytest.fixture
def region(db):
return RegionModel.objects.create(
name='Auvergne-Rhône-Alpes', code='84', resource=BaseAdresse.objects.first()
)
@pytest.fixture
def department(db, region):
return DepartmentModel.objects.create(
name='Savoie', code='73', region=region, resource=BaseAdresse.objects.first()
)
@pytest.fixture
def city(db, region, department):
return CityModel.objects.create(
name='Chambéry',
code='73065',
zipcode='73000',
population=42000,
region=region,
department=department,
resource=BaseAdresse.objects.first(),
)
@pytest.fixture
def city2(db, region, department):
return CityModel.objects.create(
name='Aix-les-Bains',
code='73008',
zipcode='73010',
population=30000,
region=region,
department=department,
resource=BaseAdresse.objects.first(),
)
@pytest.fixture
def miquelon(db):
return CityModel.objects.create(
name='Miquelon-Langlade',
code='97501',
zipcode='97500',
population=42,
resource=BaseAdresse.objects.first(),
)
@pytest.fixture
def mock_update_api_geo():
with mock.patch(
'passerelle.apps.base_adresse.models.BaseAdresse.update_api_geo_data', new=lambda x: None
) as _fixture:
yield _fixture
@pytest.fixture
def mock_update_streets():
with mock.patch(
'passerelle.apps.base_adresse.models.BaseAdresse.update_streets_data', new=lambda x: None
) as _fixture:
yield _fixture
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_search(mocked_get, app, base_adresse):
endpoint = tests.utils.generic_endpoint_url('base-adresse', 'search', slug=base_adresse.slug)
assert endpoint == '/base-adresse/base-adresse/search'
mocked_get.return_value = tests.utils.FakedResponse(content=FAKED_CONTENT, status_code=200)
resp = app.get(endpoint, params={'q': 'plop'}, status=200)
data = resp.json[0]
assert data['lat'] == '47.474633'
assert data['lon'] == '-0.593775'
assert data['display_name'] == 'Rue Roger Halope 49000 Angers'
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_search_type_housenumber(mocked_get, app, base_adresse):
endpoint = tests.utils.generic_endpoint_url('base-adresse', 'search', slug=base_adresse.slug)
mocked_get.return_value = tests.utils.FakedResponse(content=FAKED_CONTENT, status_code=200)
app.get(endpoint, params={'q': 'plop'}, status=200)
assert 'type=' not in mocked_get.call_args[0][0]
for type in ['housenumber', 'street', 'locality', 'municipality']:
app.get(endpoint, params={'q': 'plop', 'type': type}, status=200)
assert f'type={type}' in mocked_get.call_args[0][0]
app.get(endpoint, params={'q': 'plop', 'type': 'foo'}, status=200)
assert 'type=foo' not in mocked_get.call_args[0][0]
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_search_limit_to_200(mocked_get, app, base_adresse):
endpoint = tests.utils.generic_endpoint_url('base-adresse', 'search', slug=base_adresse.slug)
assert endpoint == '/base-adresse/base-adresse/search'
mocked_get.return_value = tests.utils.FakedResponse(content=FAKED_CONTENT, status_code=200)
app.get(endpoint, params={'q': 'plop' * 200}, status=200)
assert len(parse_qs(mocked_get.call_args[0][0].split('?')[1])['q'][0]) == 200
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_search_path(mocked_get, app, base_adresse):
base_adresse.service_url = 'http://example.net/path/'
base_adresse.save()
endpoint = tests.utils.generic_endpoint_url('base-adresse', 'search', slug=base_adresse.slug)
mocked_get.return_value = tests.utils.FakedResponse(content=FAKED_CONTENT, status_code=200)
resp = app.get(endpoint, params={'q': 'plop'}, status=200)
assert mocked_get.call_args[0][0].startswith('http://example.net/path/search/?')
data = resp.json[0]
assert data['lat'] == '47.474633'
assert data['lon'] == '-0.593775'
assert data['display_name'] == 'Rue Roger Halope 49000 Angers'
def test_base_adresse_search_qs(app, base_adresse, mock_api_adresse_data_gouv_fr_search):
resp = app.get('/base-adresse/%s/search?q=plop' % base_adresse.slug)
assert 'display_name' in resp.json[0]
def test_base_adresse_search_qs_zipcode(app, base_adresse, mock_api_adresse_data_gouv_fr_search):
resp = app.get('/base-adresse/%s/search?q=plop&zipcode=49000' % base_adresse.slug)
assert 'display_name' in resp.json[0]
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_search_qs_citycode(mocked_get, app, base_adresse):
app.get('/base-adresse/%s/search?q=plop&citycode=31555' % base_adresse.slug)
assert 'citycode=31555' in mocked_get.call_args[0][0]
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_search_qs_lat_lon(mocked_get, app, base_adresse):
app.get('/base-adresse/%s/search?q=plop&lat=0&lon=1' % base_adresse.slug)
assert 'lat=0' in mocked_get.call_args[0][0]
assert 'lon=1' in mocked_get.call_args[0][0]
def test_base_adresse_search_qs_empty(app, base_adresse, mock_api_adresse_data_gouv_fr_search):
resp = app.get('/base-adresse/%s/search?q=' % base_adresse.slug)
assert len(resp.json) == 0
resp = app.get('/base-adresse/%s/search?q= ' % base_adresse.slug)
assert len(resp.json) == 0
def test_base_adresse_search_qs_not_alphanumeric(app, base_adresse, mock_api_adresse_data_gouv_fr_search):
resp = app.get('/base-adresse/%s/search?q=**notalphanumeric' % base_adresse.slug)
assert len(resp.json) == 0
resp = app.get('/base-adresse/%s/search?q= **notalpha ' % base_adresse.slug)
assert len(resp.json) == 0
def test_base_adresse_search_qs_too_short(app, base_adresse, mock_api_adresse_data_gouv_fr_search):
resp = app.get('/base-adresse/%s/search?q=12' % base_adresse.slug)
assert len(resp.json) == 0
resp = app.get('/base-adresse/%s/search?q= ab ' % base_adresse.slug)
assert len(resp.json) == 0
def test_base_adresse_search_qs_parameters_error(app, base_adresse, mock_api_adresse_data_gouv_fr_search):
resp = app.get('/base-adresse/%s/search' % base_adresse.slug, status=400)
assert resp.json['err'] == 1
assert resp.json['err_class'] == 'passerelle.views.WrongParameter'
assert resp.json['err_desc'] == "missing parameters: 'q'."
# json-api serializer
resp = app.get('/base-adresse/%s/streets?zipcode=13400&coin=zz' % base_adresse.slug, status=400)
assert resp.json['err'] == 1
assert 'coin' in resp.json['err_desc']
# signature and format are ignored
app.get(
'/base-adresse/%s/streets?zipcode=13400&signature=zz&format=jsonp'
'&raise=1&jsonpCallback=f' % base_adresse.slug
)
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_search_api_error(mocked_get, app, base_adresse):
def raise_for_status():
raise HTTPError('400 Client Error: Bad Request for url: xxx')
response = tests.utils.FakedResponse(content=json.dumps({'title': 'error'}), status_code=400)
response.raise_for_status = raise_for_status
mocked_get.return_value = response
resp = app.get('/base-adresse/%s/search' % base_adresse.slug, params={'q': 'plop'}, status=200)
assert resp.json['err'] == 1
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_search_api_timeout(mocked_get, app, base_adresse):
mocked_get.side_effect = ConnectionError('Remote end closed connection without response')
resp = app.get('/base-adresse/%s/search' % base_adresse.slug, params={'q': 'plop'})
assert resp.status_code == 200
assert resp.json['err'] == 1
assert (
resp.json['err_desc']
== 'failed to get https://api-adresse.data.gouv.fr/search/?q=plop&limit=1: Remote end closed connection without response'
)
def test_base_adresse_reverse(app, base_adresse, mock_api_adresse_data_gouv_fr_reverse):
resp = app.get('/base-adresse/%s/reverse?lon=-0.593775&lat=47.474633' % base_adresse.slug)
data = resp.json
assert 'display_name' in data
assert 'address' in data
assert data['address']['city'] == 'Angers'
assert data['address']['postcode'] == '49000'
assert data['address']['citycode'] == '49007'
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_reverse_having_several(mocked_get, app, base_adresse):
content = json.loads(FAKED_CONTENT)
content['features'].append(
{
'geometry': {'type': 'Point', 'coordinates': [-0.593775, 47.474633]},
'properties': {
'citycode': '49007',
'name': 'Rue Eugène Bardon',
'id': '49007_6950_aaaaa',
'city': 'Angers',
'context': '49, Maine-et-Loire, Pays de la Loire',
'score': 0.2,
'label': 'Rue Eugène Bardon 49000 Angers',
'postcode': '49000',
'type': 'street',
},
'type': 'Feature',
}
)
faked_content = json.dumps(content)
mocked_get.return_value = tests.utils.FakedResponse(content=faked_content, status_code=200)
resp = app.get('/base-adresse/%s/reverse?lon=-0.593775&lat=47.474633' % base_adresse.slug)
data = resp.json
assert data['address']['road'] == 'Rue Roger Halope'
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_reverse_path(mocked_get, app, base_adresse):
mocked_get.return_value = tests.utils.FakedResponse(content=json.dumps({'features': []}), status_code=200)
base_adresse.service_url = 'http://example.net/path/'
base_adresse.save()
app.get('/base-adresse/%s/reverse?lon=-0.593775&lat=47.474633' % base_adresse.slug)
assert mocked_get.call_args[0][0].startswith('http://example.net/path/reverse/?')
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_reverse_type_housenumber(mocked_get, app, base_adresse):
mocked_get.return_value = tests.utils.FakedResponse(content=json.dumps({'features': []}), status_code=200)
app.get('/base-adresse/%s/reverse?lon=-0.593775&lat=47.474633' % base_adresse.slug)
assert 'type=' not in mocked_get.call_args[0][0]
app.get('/base-adresse/%s/reverse?lon=-0.593775&lat=47.474633&type=truc' % base_adresse.slug)
assert 'type=' not in mocked_get.call_args[0][0]
for type in ['housenumber', 'street', 'locality', 'municipality']:
app.get(f'/base-adresse/%s/reverse?lon=-0.593775&lat=47.474633&type={type}' % base_adresse.slug)
assert f'type={type}' in mocked_get.call_args[0][0]
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_reverse_api_timeout(mocked_get, app, base_adresse):
mocked_get.side_effect = ConnectionError('Remote end closed connection without response')
resp = app.get('/base-adresse/%s/reverse?lon=-0.593775&lat=47.474633' % base_adresse.slug)
assert resp.status_code == 200
assert resp.json['err'] == 1
assert (
resp.json['err_desc']
== 'failed to get https://api-adresse.data.gouv.fr/reverse/?lat=47.474633&lon=-0.593775: Remote end closed connection without response'
)
def test_base_adresse_streets_unaccent(app, base_adresse, street):
resp = app.get('/base-adresse/%s/streets?q=une rue tres acc' % base_adresse.slug)
data = json.loads(resp.text)
assert 'data' in data
result = data['data'][0]
assert result['city'] == street.city
assert result['text'] == street.name
assert result['citycode'] == street.citycode
assert result['zipcode'] == street.zipcode
assert result['id'] == str(street.ban_id)
def test_base_adresse_streets_get_by_id(app, base_adresse, street):
for i in range(10):
# create additional streets
other_street = StreetModel.objects.create(
ban_id='%d_000T' % (73001 + i),
city='Chambéry',
name='Une rue différente',
zipcode=str(73001 + i),
type='street',
citycode=str(73001 + i),
resource=base_adresse,
)
resp = app.get('/base-adresse/%s/streets?q=une rue tres acc' % base_adresse.slug)
assert 'data' in resp.json
result = resp.json['data'][0]
assert result['id'] == '73001_0000' # it's the "street" fixture
resp = app.get('/base-adresse/%s/streets?id=73001_0000' % base_adresse.slug)
assert len(resp.json['data']) == 1
result2 = resp.json['data'][0]
assert result2['text'] == result['text']
# get by legacy id
resp = app.get('/base-adresse/%s/streets?id=%d' % (base_adresse.slug, other_street.id))
assert len(resp.json['data']) == 1
result3 = resp.json['data'][0]
assert result3['text'] == other_street.name
# non existing and non integer id
resp = app.get('/base-adresse/%s/streets?id=%s' % (base_adresse.slug, 'XXX'))
assert len(resp.json['data']) == 0
# integer but without match.
resp = app.get('/base-adresse/%s/streets?id=%s' % (base_adresse.slug, '-20'))
assert len(resp.json['data']) == 0
def test_base_adresse_streets_get_by_codes(app, base_adresse, street):
for i in range(20):
StreetModel.objects.create(
city='Paris %d' % i,
name='La rue %d' % i,
zipcode=str(75000 + i * 10),
type='street',
citycode=str(75000 + i * 11),
resource=base_adresse,
)
resp = app.get('/base-adresse/%s/streets?zipcode=75' % base_adresse.slug)
assert 'data' in resp.json
assert resp.json['err'] == 0
assert len(resp.json['data']) == 20
assert {street['zipcode'][:2] for street in resp.json['data']} == {'75'}
resp = app.get('/base-adresse/%s/streets?zipcode=75010' % base_adresse.slug)
assert 'data' in resp.json
assert resp.json['err'] == 0
assert len(resp.json['data']) == 1
assert resp.json['data'][0]['zipcode'] == '75010'
resp = app.get('/base-adresse/%s/streets?zipcode=12345' % base_adresse.slug)
assert 'data' in resp.json
assert resp.json['err'] == 0
assert len(resp.json['data']) == 0
resp = app.get('/base-adresse/%s/streets?citycode=75' % base_adresse.slug)
assert 'data' in resp.json
assert resp.json['err'] == 0
assert len(resp.json['data']) == 20
assert {street['citycode'][:2] for street in resp.json['data']} == {'75'}
resp = app.get('/base-adresse/%s/streets?citycode=75044' % base_adresse.slug)
assert 'data' in resp.json
assert resp.json['err'] == 0
assert len(resp.json['data']) == 1
assert resp.json['data'][0]['citycode'] == '75044'
resp = app.get('/base-adresse/%s/streets?citycode=12345' % base_adresse.slug)
assert 'data' in resp.json
assert resp.json['err'] == 0
assert len(resp.json['data']) == 0
@pytest.mark.usefixtures('mock_update_api_geo')
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_command_update(mocked_get, db, base_adresse):
filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.gz')
with open(filepath, 'rb') as ban_file:
mocked_get.return_value = tests.utils.FakedResponse(content=ban_file.read(), status_code=200)
call_command('cron', 'daily')
mocked_get.assert_called_once_with(
'https://adresse.data.gouv.fr/data/ban/adresses/latest/addok/adresses-addok-73.ndjson.gz'
)
streets = StreetModel.objects.all()
assert len(streets) == 3
streets = streets.filter(ban_id='73001_0004')
assert streets.count() == 1
street = streets.first()
assert street.name == 'Chemin de la Vie, LA GRANGE DU TRIEU'
assert street.zipcode == '73610'
assert street.type == 'street'
assert street.city == 'Aiguebelette-le-Lac'
assert street.citycode == '73001'
assert street.ban_id == '73001_0004'
# check a new call downloads again
call_command('cron', 'daily')
assert mocked_get.call_count == 2
@pytest.mark.usefixtures('mock_update_api_geo')
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_command_job_update(mocked_get, db, base_adresse):
base_adresse.update_api_geo_data = lambda: None
filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.gz')
with open(filepath, 'rb') as ban_file:
mocked_get.return_value = tests.utils.FakedResponse(content=ban_file.read(), status_code=200)
# check the job added at save() downloads streets
base_adresse.jobs()
mocked_get.assert_called_once_with(
'https://adresse.data.gouv.fr/data/ban/adresses/latest/addok/adresses-addok-73.ndjson.gz'
)
assert StreetModel.objects.all().count() == 3
# second save doesn't download anything
base_adresse.save()
base_adresse.jobs()
assert mocked_get.call_count == 1
# but changing zipcode does
base_adresse.zipcode = '74'
base_adresse.save()
base_adresse.jobs()
assert mocked_get.call_count == 2
@pytest.mark.usefixtures('mock_update_api_geo')
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_command_update_97x(mocked_get, db, base_adresse_97x):
base_adresse_97x.update_api_geo_data = lambda: None
filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.gz')
with open(filepath, 'rb') as ban_file:
mocked_get.return_value = tests.utils.FakedResponse(content=ban_file.read(), status_code=200)
call_command('cron', 'daily')
mocked_get.assert_called_once_with(
'https://adresse.data.gouv.fr/data/ban/adresses/latest/addok/adresses-addok-974.ndjson.gz'
)
assert StreetModel.objects.count() == 2
@pytest.mark.usefixtures('mock_update_api_geo')
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_command_update_corsica(mocked_get, db, base_adresse_corsica):
base_adresse_corsica.update_api_geo_data = lambda: None
filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.gz')
with open(filepath, 'rb') as ban_file:
mocked_get.return_value = tests.utils.FakedResponse(content=ban_file.read(), status_code=200)
call_command('cron', 'daily')
assert mocked_get.call_count == 2
mocked_get.assert_any_call(
'https://adresse.data.gouv.fr/data/ban/adresses/latest/addok/adresses-addok-2A.ndjson.gz'
)
mocked_get.assert_any_call(
'https://adresse.data.gouv.fr/data/ban/adresses/latest/addok/adresses-addok-2B.ndjson.gz'
)
assert StreetModel.objects.count() == 0
@pytest.mark.usefixtures('mock_update_api_geo')
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_command_update_multiple(mocked_get, db, base_adresse_multiple):
base_adresse_multiple.update_api_geo_data = lambda: None
filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.gz')
with open(filepath, 'rb') as ban_file:
mocked_get.return_value = tests.utils.FakedResponse(content=ban_file.read(), status_code=200)
call_command('cron', 'daily')
assert mocked_get.call_count == 4
mocked_get.assert_any_call(
'https://adresse.data.gouv.fr/data/ban/adresses/latest/addok/adresses-addok-73.ndjson.gz'
)
mocked_get.assert_any_call(
'https://adresse.data.gouv.fr/data/ban/adresses/latest/addok/adresses-addok-974.ndjson.gz'
)
mocked_get.assert_any_call(
'https://adresse.data.gouv.fr/data/ban/adresses/latest/addok/adresses-addok-2A.ndjson.gz'
)
mocked_get.assert_any_call(
'https://adresse.data.gouv.fr/data/ban/adresses/latest/addok/adresses-addok-2B.ndjson.gz'
)
assert StreetModel.objects.count() == 5
def test_base_adresse_cities(app, base_adresse, city, city2, miquelon, department, region):
resp = app.get('/base-adresse/%s/cities?q=chambe' % base_adresse.slug)
assert len(resp.json['data']) == 1
result = resp.json['data'][0]
assert result['name'] == city.name
assert result['city'] == city.name
assert result['text'] == '%s %s' % (city.zipcode, city.name)
assert result['code'] == city.code
assert result['zipcode'] == city.zipcode
assert result['postcode'] == city.zipcode
assert result['id'] == '%s.%s' % (city.code, city.zipcode)
assert result['population'] == city.population
assert result['region_code'] == city.region.code
assert result['region_name'] == city.region.name
assert result['department_code'] == city.department.code
assert result['department_name'] == city.department.name
resp = app.get('/base-adresse/%s/cities?q=73' % base_adresse.slug)
assert len(resp.json['data']) == 2
assert resp.json['data'][0] == result
assert resp.json['data'][1]['name'] == city2.name
assert resp.json['data'][1]['zipcode'] == city2.zipcode
assert resp.json['data'][1]['id'] == '%s.%s' % (city2.code, city2.zipcode)
resp = app.get('/base-adresse/%s/cities?code=73065' % base_adresse.slug)
assert len(resp.json['data']) == 1
assert resp.json['data'][0] == result
resp = app.get('/base-adresse/%s/cities?code=73065,97501,75056' % base_adresse.slug)
assert len(resp.json['data']) == 2
assert resp.json['data'][0] == result
assert resp.json['data'][1]['name'] == 'Miquelon-Langlade'
# default ordering, descending number of inhabitants
resp = app.get('/base-adresse/%s/cities?department_code=73' % base_adresse.slug)
assert len(resp.json['data']) == 2
assert resp.json['data'][0]['name'] == city.name
assert resp.json['data'][1]['name'] == city2.name
# sorted by population (ascending) then name
resp = app.get('/base-adresse/%s/cities?department_code=73&ordering=population,name' % base_adresse.slug)
assert resp.json['data'][0]['name'] == city2.name
assert resp.json['data'][1]['name'] == city.name
# sorted by name
resp = app.get('/base-adresse/%s/cities?department_code=73&ordering=name' % base_adresse.slug)
assert resp.json['data'][0]['name'] == city2.name
assert resp.json['data'][1]['name'] == city.name
# sorted by name (reverse) then by code
resp = app.get('/base-adresse/%s/cities?department_code=73&ordering=-name,code' % base_adresse.slug)
assert resp.json['data'][0]['name'] == city.name
assert resp.json['data'][1]['name'] == city2.name
# sorted by code
resp = app.get('/base-adresse/%s/cities?department_code=73&ordering=code' % base_adresse.slug)
assert resp.json['data'][0]['name'] == city2.name
assert resp.json['data'][1]['name'] == city.name
# nonexistent ordering field, fallback on default ordering
resp = app.get('/base-adresse/%s/cities?department_code=73&ordering=foobar' % base_adresse.slug)
assert resp.json['err'] == 1
assert resp.json['err_desc'] == 'cities: erroneous ordering query foobar'
def test_base_adresse_cities_missing_region_and_department(app, base_adresse, miquelon):
resp = app.get('/base-adresse/%s/cities?q=miqu' % base_adresse.slug)
result = resp.json['data'][0]
assert result['name'] == miquelon.name
assert not result['department_code']
assert not result['region_code']
assert not result['department_name']
assert not result['region_name']
def test_base_adresse_cities_dash_in_q(app, base_adresse, miquelon):
resp = app.get('/base-adresse/%s/cities?q=miquelon-langlad' % base_adresse.slug)
result = resp.json['data'][0]
assert result['name'] == miquelon.name
def test_base_adresse_cities_region_department(app, base_adresse, miquelon, city):
reg = RegionModel.objects.create(name='IdF', code='11', resource=base_adresse)
dep = DepartmentModel.objects.create(name='Paris', code='75', region=reg, resource=base_adresse)
CityModel.objects.create(
name='Paris',
code='75056',
zipcode='75014',
population=2000000,
region=reg,
department=dep,
resource=base_adresse,
)
resp = app.get('/base-adresse/%s/cities?department_code=73' % base_adresse.slug)
result = resp.json['data']
assert len(result) == 1
assert result[0]['name'] == city.name
resp = app.get('/base-adresse/%s/cities?region_code=84' % base_adresse.slug)
result = resp.json['data']
assert len(result) == 1
assert result[0]['name'] == city.name
resp = app.get('/base-adresse/%s/cities?region_code=84&department_code=75' % base_adresse.slug)
result = resp.json['data']
assert not result
def test_base_adresse_cities_sort_order(app, base_adresse, miquelon, city):
assert miquelon.population < city.population
resp = app.get('/base-adresse/%s/cities' % base_adresse.slug)
result = resp.json['data']
assert result[0]['name'] == city.name
assert result[1]['name'] == miquelon.name
def test_base_adresse_cities_get_by_id(app, base_adresse, city):
for i in range(1, 10):
# create additional cities
city.pk = None
city.zipcode = int(city.zipcode) + i
city.save()
resp = app.get('/base-adresse/%s/cities?q=cham' % base_adresse.slug)
result = resp.json['data'][0]
assert len(resp.json['data']) == 10
city_id = result['id']
resp = app.get('/base-adresse/%s/cities?id=%s' % (base_adresse.slug, city_id))
assert len(resp.json['data']) == 1
result2 = resp.json['data'][0]
assert result2['text'] == result['text']
# non integer id.
resp = app.get('/base-adresse/%s/cities?id=%s' % (base_adresse.slug, 'XXX'))
assert resp.json['err'] == 1
# integer but without match.
resp = app.get('/base-adresse/%s/cities?id=%s' % (base_adresse.slug, '1.1'))
assert len(resp.json['data']) == 0
def test_base_adresse_departments(app, base_adresse, department, region):
resp = app.get('/base-adresse/%s/departments?q=sav' % base_adresse.slug)
result = resp.json['data'][0]
assert result['name'] == department.name
assert result['code'] == department.code
assert result['id'] == department.code
assert result['text'] == '%s %s' % (department.code, department.name)
assert result['region_code'] == region.code
assert result['region_name'] == region.name
resp = app.get('/base-adresse/%s/departments?q=73' % base_adresse.slug)
result = resp.json['data'][0]
assert result['name'] == department.name
resp = app.get('/base-adresse/%s/departments?id=%s' % (base_adresse.slug, department.code))
result = resp.json['data'][0]
assert result['name'] == department.name
def test_base_adresse_departments_region(app, base_adresse, department):
reg = RegionModel.objects.create(name='IdF', code='11', resource=base_adresse)
DepartmentModel.objects.create(name='Paris', code='75', region=reg, resource=base_adresse)
resp = app.get('/base-adresse/%s/departments?region_code=84' % base_adresse.slug)
result = resp.json['data']
assert len(result) == 1
assert result[0]['name'] == department.name
def test_base_adresse_regions(app, base_adresse, region):
resp = app.get('/base-adresse/%s/regions?q=au' % base_adresse.slug)
result = resp.json['data'][0]
assert result['name'] == region.name
assert result['code'] == region.code
assert result['id'] == region.code
assert result['text'] == '%s %s' % (region.code, region.name)
resp = app.get('/base-adresse/%s/regions?id=%s' % (base_adresse.slug, region.code))
result = resp.json['data'][0]
assert result['name'] == region.name
@pytest.mark.usefixtures('mock_update_streets')
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_command_update_geo(mocked_get, db, base_adresse, base_adresse_97x):
return_values = [
tests.utils.FakedResponse(content=content, status_code=200)
for content in (FAKE_API_GEO_REGIONS, FAKE_API_GEO_DEPARTMENTS, FAKE_API_GEO) * 2
]
mocked_get.side_effect = return_values
call_command('cron', 'daily')
assert mocked_get.call_count == 6 # 3 downloads * 2 BaseAdresse instances
mocked_get.assert_any_call(urljoin(base_adresse.api_geo_url, 'communes'))
mocked_get.assert_any_call(urljoin(base_adresse.api_geo_url, 'regions'))
mocked_get.assert_any_call(urljoin(base_adresse.api_geo_url, 'departements'))
for resource in (base_adresse, base_adresse_97x):
regions = resource.regionmodel_set
assert regions.count() == 2
idf = regions.get(name='Île-de-France')
assert idf.code == '11'
centre = regions.get(name='Bourgogne-Franche-Comté')
assert centre.code == '27'
departments = resource.departmentmodel_set
assert departments.count() == 2
paris_dep = departments.get(name='Paris')
assert paris_dep.code == '75'
assert paris_dep.region == idf
nievre = departments.get(name='Nièvre')
assert nievre.code == '58'
assert nievre.region == centre
cities = resource.citymodel_set
assert cities.count() == 3
paris = cities.get(zipcode='75001')
assert paris.name == 'Paris'
assert paris.code == '75056'
assert paris.population == 2190327
assert paris.department.code == '75'
assert paris.region.code == '11'
paris2 = cities.get(zipcode='75002')
paris_json = paris.to_json()
for key, value in paris2.to_json().items():
if key not in ['id', 'text', 'zipcode', 'postcode']:
assert paris_json[key] == value
miquelon = cities.get(zipcode='97500')
assert miquelon.name == 'Miquelon-Langlade'
assert miquelon.code == '97501'
assert miquelon.population == 596
assert not miquelon.department
assert not miquelon.region
assert CityModel.objects.count() == 6
assert DepartmentModel.objects.count() == 4
assert RegionModel.objects.count() == 4
# check a new call downloads again
mocked_get.side_effect = return_values
call_command('cron', 'daily')
assert mocked_get.call_count == 12
# and doesn't delete anything
assert CityModel.objects.count() == 6
assert DepartmentModel.objects.count() == 4
assert RegionModel.objects.count() == 4
@pytest.mark.usefixtures('mock_update_streets')
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_command_update_geo_delete(mocked_get, db, base_adresse):
return_values = [
tests.utils.FakedResponse(content=content, status_code=200)
for content in (FAKE_API_GEO_REGIONS, FAKE_API_GEO_DEPARTMENTS, FAKE_API_GEO)
]
mocked_get.side_effect = return_values
call_command('cron', 'daily')
assert CityModel.objects.count() == 3
new_fake_api_geo = json.dumps([FAKE_API_GEO_LIST[1]])
return_values = [
tests.utils.FakedResponse(content=content, status_code=200)
for content in (FAKE_API_GEO_REGIONS, FAKE_API_GEO_DEPARTMENTS, new_fake_api_geo)
]
mocked_get.side_effect = return_values
call_command('cron', 'daily')
assert CityModel.objects.count() == 1
@pytest.mark.usefixtures('mock_update_streets')
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_command_job_update_geo(mocked_get, db, base_adresse):
return_values = [
tests.utils.FakedResponse(content=content, status_code=200)
for content in (FAKE_API_GEO_REGIONS, FAKE_API_GEO_DEPARTMENTS, FAKE_API_GEO)
]
mocked_get.side_effect = return_values
# check the job added at save() downloads data
base_adresse.jobs()
assert mocked_get.call_count == 3
assert CityModel.objects.count() == 3
# second save doesn't download anything
base_adresse.save()
base_adresse.jobs()
assert mocked_get.call_count == 3
@pytest.mark.usefixtures('mock_update_streets')
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_command_update_geo_invalid(mocked_get, db, base_adresse):
mocked_get.return_value = tests.utils.FakedResponse(content='{}', status_code=200)
with pytest.raises(CommandError):
call_command('cron', 'daily')
assert mocked_get.call_count == 1
assert not RegionModel.objects.exists()
mocked_get.return_value = tests.utils.FakedResponse(content=FAKE_API_GEO, status_code=500)
call_command('cron', 'daily')
assert mocked_get.call_count == 4
assert not RegionModel.objects.exists()
mocked_get.return_value = tests.utils.FakedResponse(content='not-json', status_code=200)
call_command('cron', 'daily')
assert mocked_get.call_count == 7
assert not RegionModel.objects.exists()
@pytest.mark.usefixtures('mock_update_streets')
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_command_update_geo_region_not_found(mocked_get, db, base_adresse):
new_fake_api_geo_regions = json.dumps([json.loads(FAKE_API_GEO_REGIONS)[1]])
return_values = [
tests.utils.FakedResponse(content=content, status_code=200)
for content in (
new_fake_api_geo_regions, # first call, get regions
FAKE_API_GEO_DEPARTMENTS, # then, get departments
FAKE_API_GEO, # and get communes
)
] + [
# region code 11 not found, try to get it again
tests.utils.FakedResponse(content='not-json', status_code=200),
tests.utils.FakedResponse(content='not-json', status_code=200),
tests.utils.FakedResponse(content='not-json', status_code=200),
]
mocked_get.side_effect = return_values
call_command('cron', 'daily')
assert mocked_get.call_args_list == [
mock.call('https://geo.api.gouv.fr/regions'),
mock.call('https://geo.api.gouv.fr/departements'),
mock.call('https://geo.api.gouv.fr/communes'),
mock.call('https://geo.api.gouv.fr/regions/11'),
mock.call('https://geo.api.gouv.fr/regions/11'),
mock.call('https://geo.api.gouv.fr/regions/11'),
]
assert RegionModel.objects.count() == 1
assert RegionModel.objects.get().code == '27'
assert DepartmentModel.objects.count() == 1
assert DepartmentModel.objects.get().code == '58'
@pytest.mark.usefixtures('mock_update_api_geo')
@mock.patch('passerelle.utils.Request.get', side_effect=ConnectionError)
def test_base_adresse_command_update_street_timeout(mocked_get, db, base_adresse):
call_command('cron', 'daily')
assert mocked_get.call_count == 1
assert not RegionModel.objects.exists()
@pytest.mark.usefixtures('mock_update_streets')
@mock.patch('passerelle.utils.Request.get', side_effect=ConnectionError)
def test_base_adresse_command_update_geo_no_connection(mocked_get, db, base_adresse):
call_command('cron', 'daily')
assert mocked_get.call_count == 3
assert not RegionModel.objects.exists()
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_addresses(mocked_get, app, base_adresse):
endpoint = tests.utils.generic_endpoint_url('base-adresse', 'addresses', slug=base_adresse.slug)
assert endpoint == '/base-adresse/base-adresse/addresses'
mocked_get.return_value = tests.utils.FakedResponse(content=FAKED_CONTENT, status_code=200)
resp = app.get(endpoint, params={'q': 'plop'}, status=200)
data = resp.json['data'][0]
assert data['lat'] == '47.474633'
assert data['lon'] == '-0.593775'
assert data['display_name'] == 'Rue Roger Halope 49000 Angers'
assert data['text'] == 'Rue Roger Halope 49000 Angers'
assert data['id'] == '49007_6950_be54bd~47.474633~-0.593775~Rue Roger Halope 49000 Angers'
assert data['address']['city'] == 'Angers'
assert data['address']['postcode'] == '49000'
assert data['address']['citycode'] == '49007'
assert data['address']['road'] == 'Rue Roger Halope'
assert data['ban_id'] == '49007_6950_be54bd'
assert data['extra']['info1'] == 'xxx'
assert data['extra']['info2'] == 'yyy'
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_addresses_qs_page_limit(mocked_get, app, base_adresse):
resp = app.get('/base-adresse/%s/addresses?q=plop&page_limit=1' % base_adresse.slug)
assert 'limit=1' in mocked_get.call_args[0][0]
resp = app.get('/base-adresse/%s/addresses?q=plop&page_limit=100' % base_adresse.slug)
assert 'limit=20' in mocked_get.call_args[0][0]
resp = app.get('/base-adresse/%s/addresses?q=plop&page_limit=blabla' % base_adresse.slug, status=400)
assert 'invalid value' in resp.json['err_desc']
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_addresses_qs_citycode(mocked_get, app, base_adresse):
app.get('/base-adresse/%s/addresses?q=plop&citycode=31555' % base_adresse.slug)
assert 'citycode=31555' in mocked_get.call_args[0][0]
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_addresses_qs_coordinates(mocked_get, app, base_adresse_coordinates):
app.get('/base-adresse/%s/addresses?q=plop' % base_adresse_coordinates.slug)
assert 'lat=%s' % base_adresse_coordinates.latitude in mocked_get.call_args[0][0]
assert 'lon=%s' % base_adresse_coordinates.longitude in mocked_get.call_args[0][0]
app.get('/base-adresse/%s/addresses?q=plop&lat=42&lon=43' % base_adresse_coordinates.slug)
assert 'lat=42' in mocked_get.call_args[0][0]
assert 'lon=43' in mocked_get.call_args[0][0]
def test_base_adresse_addresses_cache(app, base_adresse, mock_api_adresse_data_gouv_fr_search, caplog):
resp = app.get('/base-adresse/%s/addresses?q=plop' % base_adresse.slug)
assert mock_api_adresse_data_gouv_fr_search.call['count'] == 1
data = resp.json['data'][0]
assert data['text'] == 'Rue Roger Halope 49000 Angers'
api_id = data['id']
assert AddressCacheModel.objects.filter(api_id=api_id[:30]).exists()
assert AddressCacheModel.objects.count() == 1
resp = app.get('/base-adresse/%s/addresses?id=%s' % (base_adresse.slug, api_id))
assert mock_api_adresse_data_gouv_fr_search.call['count'] == 1 # no new call
assert data['text'] == 'Rue Roger Halope 49000 Angers'
assert 'address' in data
resp = app.get('/base-adresse/%s/addresses?q=plop' % base_adresse.slug)
assert AddressCacheModel.objects.count() == 1 # no new object has been created
assert mock_api_adresse_data_gouv_fr_search.call['count'] == 2
# no cache
AddressCacheModel.objects.all().delete()
resp = app.get('/base-adresse/%s/addresses?id=%s' % (base_adresse.slug, api_id))
assert AddressCacheModel.objects.count() == 1
assert mock_api_adresse_data_gouv_fr_search.call['count'] == 3
assert data['text'] == 'Rue Roger Halope 49000 Angers'
assert 'address' in data
# no cache and id has changed
AddressCacheModel.objects.all().delete()
api_id = '49007_XXXX_be54bd~47.474633~-0.593775~Rue%20Roger%20Halope%2049000%20Angers'
resp = app.get('/base-adresse/%s/addresses?id=%s' % (base_adresse.slug, api_id))
assert resp.json['err'] == 'Address ID not found'
def test_base_adresse_addresses_cache_err(app, base_adresse, mock_api_adresse_data_gouv_fr_search):
resp = app.get('/base-adresse/%s/addresses?id=%s' % (base_adresse.slug, 'wrong_id'))
assert mock_api_adresse_data_gouv_fr_search.call['count'] == 0
assert 'err' in resp.json
@pytest.mark.usefixtures('mock_update_api_geo', 'mock_update_streets')
def test_base_adresse_addresses_clean_cache(app, base_adresse, freezer, mock_api_adresse_data_gouv_fr_search):
app.get('/base-adresse/%s/addresses?q=plop' % base_adresse.slug)
assert AddressCacheModel.objects.count() == 1
freezer.move_to(datetime.timedelta(minutes=30))
call_command('cron', 'hourly')
assert AddressCacheModel.objects.count() == 1
freezer.move_to(datetime.timedelta(minutes=30, seconds=1))
call_command('cron', 'hourly')
assert AddressCacheModel.objects.count() == 0
app.get('/base-adresse/%s/addresses?q=plop' % base_adresse.slug)
assert AddressCacheModel.objects.count() == 1
# asking for the address again resets the timestamp
freezer.move_to(datetime.timedelta(hours=1, seconds=1))
app.get('/base-adresse/%s/addresses?q=plop' % base_adresse.slug)
call_command('cron', 'hourly')
assert AddressCacheModel.objects.count() == 1
freezer.move_to(datetime.timedelta(hours=1, seconds=1))
app.get(
'/base-adresse/%s/addresses?id=%s'
% (base_adresse.slug, '49007_6950_be54bd~47.474633~-0.593775~Rue%20Roger%20Halope%2049000%20Angers')
)
call_command('cron', 'hourly')
assert AddressCacheModel.objects.count() == 1
@mock.patch('passerelle.utils.Request.get')
def test_base_adresse_addresses_data_change(mocked_get, app, base_adresse):
endpoint = tests.utils.generic_endpoint_url('base-adresse', 'addresses', slug=base_adresse.slug)
mocked_get.return_value = tests.utils.FakedResponse(content=FAKED_CONTENT, status_code=200)
# one user selects an address
resp = app.get(endpoint, params={'q': 'plop'}, status=200)
data = resp.json['data'][0]
address_id, address_text = data['id'], data['text']
# another requests the same while upstream data has been updated
new_content = json.loads(FAKED_CONTENT)
new_content['features'][0]['properties']['label'] = 'changed'
mocked_get.return_value = tests.utils.FakedResponse(content=json.dumps(new_content), status_code=200)
resp = app.get(endpoint, params={'q': 'plop'}, status=200)
# first user saves the form, data should not have changed
resp = app.get(endpoint, params={'id': address_id}, status=200)
assert resp.json['data'][0]['text'] == address_text
# when cache is cleared, we get the updated data
AddressCacheModel.objects.all().delete()
resp = app.get(endpoint, params={'q': 'plop'}, status=200)
assert resp.json['data'][0]['text'] == 'changed'
def test_base_adresse_reverse_cache(
app, base_adresse, freezer, mock_api_adresse_data_gouv_fr_reverse, mock_api_adresse_data_gouv_fr_search
):
assert AddressCacheModel.objects.count() == 0
resp = app.get('/base-adresse/%s/reverse?lon=-0.593775&lat=47.474633' % base_adresse.slug)
assert mock_api_adresse_data_gouv_fr_reverse.call['count'] == 1
data = resp.json
assert data['text'] == 'Rue Roger Halope 49000 Angers'
api_id = data['id']
assert AddressCacheModel.objects.filter(api_id=api_id[:30]).exists()
assert AddressCacheModel.objects.count() == 1
first_timestamp = AddressCacheModel.objects.get().timestamp
resp = app.get('/base-adresse/%s/addresses?id=%s' % (base_adresse.slug, api_id))
data = resp.json['data'][0]
assert mock_api_adresse_data_gouv_fr_search.call['count'] == 0
assert data['text'] == 'Rue Roger Halope 49000 Angers'
assert 'address' in data
# check caching timestamp update
freezer.move_to(datetime.timedelta(hours=1, seconds=1))
resp = app.get('/base-adresse/%s/reverse?lon=-0.593775&lat=47.474633' % base_adresse.slug)
assert mock_api_adresse_data_gouv_fr_reverse.call['count'] == 2
assert AddressCacheModel.objects.get().timestamp > first_timestamp
# check lookup id is kept
resp = app.get('/base-adresse/%s/addresses?id=%s' % (base_adresse.slug, api_id.lower()))
data = resp.json['data'][0]
assert mock_api_adresse_data_gouv_fr_search.call['count'] == 0
assert data['id'] != api_id
assert data['id'] == api_id.lower()
# without cache
assert AddressCacheModel.objects.all().delete()
resp = app.get('/base-adresse/%s/addresses?id=%s' % (base_adresse.slug, api_id.lower()))
data = resp.json['data'][0]
assert mock_api_adresse_data_gouv_fr_search.call['count'] == 1
assert data['id'] != api_id
assert data['id'] == api_id.lower()
@responses.activate(registry=OrderedRegistry) # pylint: disable=unexpected-keyword-arg,no-value-for-parameter
def test_base_adresse_search_retry(app, base_adresse):
from passerelle.utils import Request
Request.ADAPTER_REGISTRY.clear()
responses.add(
responses.GET,
'https://api-adresse.data.gouv.fr/search/',
body='Error',
status=504,
)
responses.add(
responses.GET,
'https://api-adresse.data.gouv.fr/search/',
body='Error',
status=504,
)
responses.add(
responses.GET,
'https://api-adresse.data.gouv.fr/search/',
body='Error',
status=504,
)
responses.add(
responses.GET,
'https://api-adresse.data.gouv.fr/search/',
body=FAKED_CONTENT,
content_type='application/json',
status=200,
)
resp = app.get('/base-adresse/base-adresse/search/', params={'q': 'plop'})
data = resp.json[0]
assert data['lat'] == '47.474633'
assert data['lon'] == '-0.593775'
assert data['display_name'] == 'Rue Roger Halope 49000 Angers'
responses.add(
responses.GET,
'https://api-adresse.data.gouv.fr/search/',
body='Error',
status=504,
)
responses.add(
responses.GET,
'https://api-adresse.data.gouv.fr/search/',
body='Error',
status=504,
)
responses.add(
responses.GET,
'https://api-adresse.data.gouv.fr/search/',
body='Error',
status=504,
)
responses.add(
responses.GET,
'https://api-adresse.data.gouv.fr/search/',
body='Error',
status=504,
)
responses.add(
responses.GET,
'https://api-adresse.data.gouv.fr/search/',
body=FAKED_CONTENT,
content_type='application/json',
status=200,
)
resp = app.get('/base-adresse/base-adresse/search/', params={'q': 'plop'})
assert resp.json['err'] == 1
View Git Blame Copy Permalink