passerelle/tests/test_base_adresse.py

Ignoring revisions in .git-blame-ignore-revs. Click here to bypass and see the normal blame view.

1231 lines
48 KiB
Python
Raw Normal View History

import datetime
import json
import os
from unittest import mock
from urllib.parse import parse_qs, urljoin
2021-05-07 11:53:34 +02:00
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,
)
2021-02-20 16:26:01 +01:00
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'
2021-02-20 16:26:01 +01:00
@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)
2020-01-06 18:47:53 +01:00
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)
2021-02-20 16:26:01 +01:00
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