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