607 lines
23 KiB
Python
607 lines
23 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import os
|
|
import pytest
|
|
import mock
|
|
import utils
|
|
import json
|
|
|
|
from requests.exceptions import ConnectionError
|
|
|
|
from django.core.management import call_command
|
|
from django.core.management.base import CommandError
|
|
from django.utils.six.moves.urllib.parse import urljoin
|
|
|
|
from passerelle.apps.base_adresse.models import (BaseAdresse, StreetModel, CityModel,
|
|
DepartmentModel, RegionModel)
|
|
|
|
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"
|
|
},
|
|
"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 utils.setup_access_rights(BaseAdresse.objects.create(slug='base-adresse', zipcode='73'))
|
|
|
|
|
|
@pytest.fixture
|
|
def base_adresse_97x(db):
|
|
return utils.setup_access_rights(BaseAdresse.objects.create(slug='base-adresse',
|
|
zipcode='97425'))
|
|
|
|
|
|
@pytest.fixture
|
|
def base_adresse_corsica(db):
|
|
return utils.setup_access_rights(BaseAdresse.objects.create(slug='base-adresse',
|
|
zipcode='20000, 20100 '))
|
|
|
|
|
|
@pytest.fixture
|
|
def base_adresse_multiple(db):
|
|
return utils.setup_access_rights(BaseAdresse.objects.create(slug='base-adresse',
|
|
zipcode='73, 73100, 97425,20000 '))
|
|
|
|
|
|
@pytest.fixture
|
|
def street(db):
|
|
return StreetModel.objects.create(city=u'Chambéry',
|
|
name=u'Une rüê très äccentuéè',
|
|
zipcode=u'73000',
|
|
type=u'street',
|
|
citycode=u'73001')
|
|
|
|
|
|
@pytest.fixture
|
|
def region(db):
|
|
return RegionModel.objects.create(name=u'Auvergne-Rhône-Alpes', code='84')
|
|
|
|
|
|
@pytest.fixture
|
|
def department(db, region):
|
|
return DepartmentModel.objects.create(name=u'Savoie', code='73', region=region)
|
|
|
|
|
|
@pytest.fixture
|
|
def city(db, region, department):
|
|
return CityModel.objects.create(name=u'Chambéry', code='73065', zipcode='73000',
|
|
population=42000, region=region, department=department)
|
|
|
|
|
|
@pytest.fixture
|
|
def miquelon(db):
|
|
return CityModel.objects.create(name=u'Miquelon-Langlade', code='97501', zipcode='97500',
|
|
population=42)
|
|
|
|
|
|
@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 = utils.generic_endpoint_url('base-adresse', 'search', slug=base_adresse.slug)
|
|
assert endpoint == '/base-adresse/base-adresse/search'
|
|
mocked_get.return_value = 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_path(mocked_get, app, base_adresse):
|
|
base_adresse.service_url = 'http://example.net/path/'
|
|
base_adresse.save()
|
|
endpoint = utils.generic_endpoint_url('base-adresse', 'search', slug=base_adresse.slug)
|
|
mocked_get.return_value = 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_lat_lon(mocked_get, app, base_adresse):
|
|
resp = 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
|
|
|
|
|
|
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'] == u"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)
|
|
|
|
|
|
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_path(mocked_get, app, base_adresse):
|
|
mocked_get.return_value = 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/?')
|
|
|
|
|
|
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.body)
|
|
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'] == street.id
|
|
|
|
def test_base_adresse_streets_get_by_id(app, base_adresse, street):
|
|
for i in range(10):
|
|
# create additional streets
|
|
StreetModel.objects.create(city=u'Chambéry',
|
|
name=u'Une rue différente',
|
|
zipcode=str(73001 + i),
|
|
type='street',
|
|
citycode=str(73001 + i))
|
|
|
|
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]
|
|
street_id = result['id']
|
|
|
|
resp = app.get('/base-adresse/%s/streets?id=%s' % (base_adresse.slug, street_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/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
|
|
|
|
|
|
@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.bz2')
|
|
mocked_get.return_value = utils.FakedResponse(content=open(filepath).read(), status_code=200)
|
|
call_command('cron', 'daily')
|
|
mocked_get.assert_called_once_with('http://bano.openstreetmap.fr/BAN_odbl/BAN_odbl_73-json.bz2')
|
|
streets = StreetModel.objects.all()
|
|
assert len(streets) == 3
|
|
street = StreetModel.objects.order_by('id').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'
|
|
# 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_hourly_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.bz2')
|
|
mocked_get.return_value = utils.FakedResponse(content=open(filepath).read(), status_code=200)
|
|
# check the first hourly job downloads streets
|
|
call_command('cron', 'hourly')
|
|
mocked_get.assert_called_once_with('http://bano.openstreetmap.fr/BAN_odbl/BAN_odbl_73-json.bz2')
|
|
assert StreetModel.objects.all().count() == 3
|
|
# check a second call doesn't download anything
|
|
call_command('cron', 'hourly')
|
|
assert mocked_get.call_count == 1
|
|
|
|
|
|
@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.bz2')
|
|
mocked_get.return_value = utils.FakedResponse(content=open(filepath).read(), status_code=200)
|
|
call_command('cron', 'daily')
|
|
mocked_get.assert_called_once_with('http://bano.openstreetmap.fr/BAN_odbl/BAN_odbl_974-json.bz2')
|
|
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.bz2')
|
|
mocked_get.return_value = utils.FakedResponse(content=open(filepath).read(), status_code=200)
|
|
call_command('cron', 'daily')
|
|
assert mocked_get.call_count == 2
|
|
mocked_get.assert_any_call('http://bano.openstreetmap.fr/BAN_odbl/BAN_odbl_2A-json.bz2')
|
|
mocked_get.assert_any_call('http://bano.openstreetmap.fr/BAN_odbl/BAN_odbl_2B-json.bz2')
|
|
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.bz2')
|
|
mocked_get.return_value = utils.FakedResponse(content=open(filepath).read(), status_code=200)
|
|
call_command('cron', 'daily')
|
|
assert mocked_get.call_count == 4
|
|
mocked_get.assert_any_call('http://bano.openstreetmap.fr/BAN_odbl/BAN_odbl_73-json.bz2')
|
|
mocked_get.assert_any_call('http://bano.openstreetmap.fr/BAN_odbl/BAN_odbl_974-json.bz2')
|
|
mocked_get.assert_any_call('http://bano.openstreetmap.fr/BAN_odbl/BAN_odbl_2A-json.bz2')
|
|
mocked_get.assert_any_call('http://bano.openstreetmap.fr/BAN_odbl/BAN_odbl_2B-json.bz2')
|
|
assert StreetModel.objects.count() == 5
|
|
|
|
|
|
def test_base_adresse_cities(app, base_adresse, city, department, region):
|
|
resp = app.get('/base-adresse/%s/cities?q=chambe' % base_adresse.slug)
|
|
result = resp.json['data'][0]
|
|
assert result['name'] == city.name
|
|
assert result['text'] == '%s %s' % (city.zipcode, city.name)
|
|
assert result['code'] == city.code
|
|
assert result['zipcode'] == 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 resp.json['data'][0] == result
|
|
|
|
resp = app.get('/base-adresse/%s/cities?code=73065' % base_adresse.slug)
|
|
assert resp.json['data'][0] == result
|
|
|
|
|
|
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_region_department(app, base_adresse, miquelon, city):
|
|
reg = RegionModel.objects.create(name=u'IdF', code='11')
|
|
dep = DepartmentModel.objects.create(name=u'Paris', code='75', region=reg)
|
|
paris = CityModel.objects.create(name=u'Paris', code='75056', zipcode='75014',
|
|
population=2000000, region=reg, department=dep)
|
|
|
|
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=u'IdF', code='11')
|
|
paris = DepartmentModel.objects.create(name=u'Paris', code='75', region=reg)
|
|
|
|
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):
|
|
return_values = [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 mocked_get.call_count == 3
|
|
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'))
|
|
|
|
regions = RegionModel.objects.all()
|
|
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 = DepartmentModel.objects.all()
|
|
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 = CityModel.objects.all()
|
|
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 not key in ['id', 'text', 'zipcode']:
|
|
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
|
|
|
|
# check a new call downloads again
|
|
mocked_get.side_effect = return_values
|
|
call_command('cron', 'daily')
|
|
assert mocked_get.call_count == 6
|
|
# and doesn't delete anything
|
|
assert CityModel.objects.count() == 3
|
|
assert DepartmentModel.objects.count() == 2
|
|
assert RegionModel.objects.count() == 2
|
|
|
|
|
|
@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 = [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 = [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_hourly_update_geo(mocked_get, db, base_adresse):
|
|
return_values = [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 first hourly job downloads data
|
|
call_command('cron', 'hourly')
|
|
assert mocked_get.call_count == 3
|
|
assert CityModel.objects.count() == 3
|
|
# check a second call doesn't download anything
|
|
call_command('cron', 'hourly')
|
|
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 = 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 = 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 = 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', 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()
|