261 lines
11 KiB
Python
261 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import os
|
|
import pytest
|
|
import mock
|
|
import utils
|
|
import json
|
|
|
|
from django.core.management import call_command
|
|
|
|
from passerelle.apps.base_adresse.models import BaseAdresse, 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"
|
|
},
|
|
"type": "Feature"
|
|
}
|
|
]
|
|
})
|
|
|
|
FAKE_DATA = ''
|
|
|
|
|
|
@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')
|
|
|
|
|
|
@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
|
|
|
|
|
|
@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')
|
|
mocked_get.call_count == 2
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_base_adresse_command_hourly_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)
|
|
# 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')
|
|
mocked_get.call_count == 1
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_base_adresse_command_update_97x(mocked_get, db, base_adresse_97x):
|
|
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
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_base_adresse_command_update_corsica(mocked_get, db, base_adresse_corsica):
|
|
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
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_base_adresse_command_update_multiple(mocked_get, db, base_adresse_multiple):
|
|
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
|