passerelle/tests/test_api_particulier.py

494 lines
16 KiB
Python

# tests/test_api_particulier.py
# Copyright (C) 2017 Entr'ouvert
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import logging
import pytest
import requests
from django.urls import reverse
from httmock import HTTMock, response, urlmatch
from passerelle.apps.api_particulier.models import APIParticulier
from passerelle.base.models import ResourceLog
from tests.test_manager import login
from tests.utils import endpoint_get, make_resource
SVAIR_RESPONSE = {
"declarant1": {
"nom": "Martin",
"nomNaissance": "Martin",
"prenoms": "Pierre",
"dateNaissance": "22/03/1985",
},
"declarant2": {
"nom": "Martin",
"nomNaissance": "Honore",
"prenoms": "Marie",
"dateNaissance": "03/04/1986",
},
"foyerFiscal": {"annee": 2015, "adresse": "12 rue Balzac 75008 Paris"},
"dateRecouvrement": "10/10/2015",
"dateEtablissement": "08/07/2015",
"nombreParts": 2,
"situationFamille": "Marié(e)s",
"nombrePersonnesCharge": 2,
"revenuBrutGlobal": 29880,
"revenuImposable": 29880,
"impotRevenuNetAvantCorrections": 2165,
"montantImpot": 2165,
"revenuFiscalReference": 29880,
"anneeImpots": "2015",
"anneeRevenus": "2014",
}
CAF_FAMILLE = {
"adresse": {
"codePostalVille": "12345 CONDAT",
"complementIdentiteGeo": "ESCALIER B",
"identite": "Madame MARIE DUPONT",
"numeroRue": "123 RUE BIDON",
"pays": "FRANCE",
},
"allocataires": [
{"dateDeNaissance": "12111971", "nomPrenom": "MARIE DUPONT", "sexe": "F"},
{"dateDeNaissance": "18101969", "nomPrenom": "JEAN DUPONT", "sexe": "M"},
],
"annee": 2017,
"enfants": [{"dateDeNaissance": "11122016", "nomPrenom": "LUCIE DUPONT", "sexe": "F"}],
"mois": 4,
"quotientFamilial": 1754,
}
INTROSPECT = {
"_id": "1d99db5a-a099-4314-ad2f-2707c6b505a6",
"name": "Application de sandbox",
"scopes": [
"dgfip_avis_imposition",
"dgfip_adresse",
"cnaf_allocataires",
"cnaf_enfants",
"cnaf_adresse",
"cnaf_quotient_familial",
"mesri_statut_etudiant",
],
}
@urlmatch(netloc=r'^particulier.*\.api\.gouv\.fr$', path=r'^/api/v2/avis-imposition$')
def api_particulier_v2_avis_imposition(url, request):
return response(200, SVAIR_RESPONSE, request=request)
@urlmatch(netloc=r'^particulier.*\.api\.gouv\.fr$', path=r'^/api/v2/composition-familiale$')
def api_particulier_v2_situation_familiale(url, request):
return response(200, CAF_FAMILLE, request=request)
@urlmatch(netloc=r'^particulier.*\.api\.gouv\.fr$', path=r'^/api/introspect$')
def api_particulier_introspect(url, request):
return response(200, INTROSPECT, request=request)
@urlmatch(netloc=r'^particulier.*\.api\.gouv\.fr$')
def api_particulier_error_500(url, request):
return response(500, {'error': 500}, request=request)
@urlmatch(netloc=r'^particulier.*\.api\.gouv\.fr$')
def api_particulier_connection_error(url, request):
raise requests.RequestException('connection timed-out')
@urlmatch(netloc=r'^particulier.*\.api\.gouv\.fr$')
def api_particulier_error_not_json(url, request):
return response(200, 'something bad happened', request=request)
@urlmatch(netloc=r'^particulier.*\.api\.gouv\.fr$')
def api_particulier_error_not_found(url, request):
return response(
404,
{
'error': 'not_found',
'message': 'Les paramètres fournis sont incorrects ou ne correspondent pas à un avis',
},
request=request,
)
@urlmatch(netloc=r'^particulier.*\.api\.gouv\.fr$')
def api_particulier_error_not_found_caf(url, request):
return response(
404,
{'error': 'not_found', 'message': 'Dossier allocataire inexistant. Le document ne peut être édité.'},
request=request,
)
@urlmatch(netloc=r'^particulier.*\.api\.gouv\.fr$')
def api_particulier_error_not_found_deregistrated(url, request):
return response(
404,
{'error': 'not_found', 'message': 'Dossier radié. Le document ne peut être édité.'},
request=request,
)
@pytest.fixture
def mock_api_particulier():
with HTTMock(
api_particulier_v2_avis_imposition, api_particulier_v2_situation_familiale, api_particulier_introspect
):
yield None
@pytest.fixture
def resource(db):
return make_resource(
APIParticulier,
slug='test',
title='API Particulier Prod',
description='API Particulier Prod',
platform='test',
api_key='83c68bf0b6013c4daf3f8213f7212aa5',
)
vector = [
(
['impots_svair', 'avis-imposition'],
{
'numero_fiscal': '1234567890123',
'reference_avis': '3210987654321',
},
),
(['caf_famille', 'situation-familiale'], {'code_postal': 12, 'numero_allocataire': '0000015'}),
]
def test_error_500(app, resource, mock_api_particulier):
with HTTMock(api_particulier_error_500):
def do(endpoint, params):
resp = endpoint_get('/api-particulier/test/%s' % endpoint, app, resource, endpoint, params=params)
assert resp.status_code == 200
assert resp.json['err'] == 1
assert resp.json['data']['status_code'] == 500
assert resp.json['data']['code'] == 'non-200'
for endpoints, params in vector:
for endpoint in endpoints:
do(endpoint, params)
def test_not_json(app, resource, mock_api_particulier):
with HTTMock(api_particulier_error_not_json):
def do(endpoint, params):
resp = endpoint_get('/api-particulier/test/%s' % endpoint, app, resource, endpoint, params=params)
assert resp.status_code == 200
assert resp.json['err'] == 1
assert 'returned non-JSON content' in resp.json['err_desc']
assert resp.json['data']['code'] == 'non-json'
for endpoints, params in vector:
for endpoint in endpoints:
do(endpoint, params)
def test_not_found(app, resource, mock_api_particulier):
with HTTMock(api_particulier_error_not_found):
def do(endpoint, params):
resp = endpoint_get('/api-particulier/test/%s' % endpoint, app, resource, endpoint, params=params)
assert resp.status_code == 200
assert resp.json['err'] == 1
assert 'incorrects ou ne correspondent pas' in resp.json['err_desc']
assert resp.json['data']['code'] == 'not-found'
for endpoints, params in vector:
for endpoint in endpoints:
do(endpoint, params)
def test_connection_error(app, resource, mock_api_particulier):
with HTTMock(api_particulier_connection_error):
def do(endpoint, params):
resp = endpoint_get('/api-particulier/test/%s' % endpoint, app, resource, endpoint, params=params)
assert resp.status_code == 200
assert resp.json['err'] == 1
assert (
resp.json['err_desc']
== 'API-particulier platform "test" connection error: connection timed-out'
)
for endpoints, params in vector:
for endpoint in endpoints:
do(endpoint, params)
def test_numero_fiscal_too_short(app, resource, mock_api_particulier):
resp = endpoint_get(
'/api-particulier/test/avis-imposition',
app,
resource,
'avis-imposition',
params={
'numero_fiscal': ' 1234567890', # too short
'reference_avis': '3210987654321',
'user': 'John Doe',
},
)
assert resp.status_code == 200
assert resp.json['err'] == 1
assert resp.json['data'] is None
assert 'bad numero_fiscal' in resp.json['err_desc']
def test_reference_avis_too_short(app, resource, mock_api_particulier):
resp = endpoint_get(
'/api-particulier/test/avis-imposition',
app,
resource,
'avis-imposition',
params={
'numero_fiscal': '1234567890123',
'reference_avis': '32109876543 ', # too short
'user': 'John Doe',
},
)
assert resp.status_code == 200
assert resp.json['err'] == 1
assert resp.json['data'] is None
assert 'bad reference_avis' in resp.json['err_desc']
def test_avis_imposition(app, resource, mock_api_particulier):
resp = endpoint_get(
'/api-particulier/test/avis-imposition',
app,
resource,
'avis-imposition',
params={
'numero_fiscal': '1234567890123',
'reference_avis': '3210987654321',
'user': 'John Doe',
},
)
assert resp.status_code == 200
assert resp.json['data']['montantImpot'] == 2165
assert resp.json['err'] == 0
resp = endpoint_get(
'/api-particulier/test/avis-imposition',
app,
resource,
'avis-imposition',
params={
'numero_fiscal': '1234567890123X', # 14 chars : will be cutted
'reference_avis': '3210987654321X', # idem
'user': 'John Doe',
},
)
assert resp.status_code == 200
assert resp.json['data']['montantImpot'] == 2165
assert resp.json['err'] == 0
def test_situation_familiale(app, resource, mock_api_particulier):
params = {
'code_postal': '99148',
'numero_allocataire': '0000354',
'user': 'John Doe',
}
resp = endpoint_get(
'/api-particulier/test/situation-familiale', app, resource, 'situation-familiale', params=params
)
assert resp.json['data']['adresse']['codePostalVille'] == '12345 CONDAT'
params['numero_allocataire'] = '11'
resp = endpoint_get(
'/api-particulier/test/situation-familiale', app, resource, 'situation-familiale', params=params
)
assert resp.status_code == 200
assert resp.json['err'] == 1
assert '7 digits' in resp.json['err_desc']
params['numero_allocataire'] = '123456a'
resp = endpoint_get(
'/api-particulier/test/situation-familiale', app, resource, 'situation-familiale', params=params
)
assert resp.status_code == 200
assert resp.json['err'] == 1
assert '7 digits' in resp.json['err_desc']
# last letter truncated automatically
params['numero_allocataire'] = '1234567a'
resp = endpoint_get(
'/api-particulier/test/situation-familiale', app, resource, 'situation-familiale', params=params
)
assert resp.json['data']['adresse']['codePostalVille'] == '12345 CONDAT'
# cleaned data is also inlcuded in the response
assert resp.json['data']['numero_allocataire'] == '1234567'
assert resp.json['data']['code_postal'] == params['code_postal']
params['code_postal'] = ' '
resp = endpoint_get(
'/api-particulier/test/situation-familiale', app, resource, 'situation-familiale', params=params
)
assert resp.status_code == 200
assert resp.json['err'] == 1
assert 'missing' in resp.json['err_desc']
def test_detail_page(app, resource, admin_user):
login(app)
response = app.get(
reverse(
'view-connector',
kwargs={
'connector': 'api-particulier',
'slug': 'test',
},
)
)
assert 'API Particulier Prod' in response.text
assert 'family allowance' in response.text
assert 'fiscal information' in response.text
@pytest.mark.parametrize(
'mock,should_log',
[
(api_particulier_error_not_found, False),
(api_particulier_error_500, True),
(api_particulier_error_not_json, True),
(api_particulier_error_not_found_caf, False),
(api_particulier_error_not_found_deregistrated, False),
],
)
def test_api_particulier_dont_log_not_found(app, resource, mock, should_log):
with HTTMock(mock):
endpoint_get(
'/api-particulier/test/avis-imposition',
app,
resource,
'avis-imposition',
params={
'numero_fiscal': '1234567890123',
'reference_avis': '3210987654321',
},
)
logs = ResourceLog.objects.all()
if should_log:
assert logs.count() == 3
assert logs.filter(levelno=logging.ERROR).count() == 1
else:
assert logs.count() == 2
assert not logs.filter(levelno=logging.ERROR).exists()
def test_scopes(app, resource, mock_api_particulier):
assert not resource.accessible_scopes
resp = endpoint_get('/api-particulier/test/scopes', app, resource, 'scopes')
assert resp.json['data'] == [
"cnaf_adresse",
"cnaf_allocataires",
"cnaf_enfants",
"cnaf_quotient_familial",
"dgfip_adresse",
"dgfip_avis_imposition",
"mesri_statut_etudiant",
]
assert len(APIParticulier.objects.get(slug=resource.slug).accessible_scopes) == 7
def test_scopes_error(app, resource):
resource.accessible_scopes = ['some', 'scopes']
resource.save()
assert len(APIParticulier.objects.get(slug=resource.slug).accessible_scopes) == 2
with HTTMock(api_particulier_error_500):
resp = endpoint_get('/api-particulier/test/scopes', app, resource, 'scopes')
assert resp.json['err']
assert len(APIParticulier.objects.get(slug=resource.slug).accessible_scopes) == 2
def test_cron(resource):
assert not resource.accessible_scopes
with HTTMock(api_particulier_introspect):
resource.daily()
assert len(resource.accessible_scopes) == 7
with HTTMock(api_particulier_error_500):
resource.daily()
assert len(resource.accessible_scopes) == 0
def test_manager_creation(db, app, admin_user, resource):
app = login(app)
path = '/manage/%s/add' % resource.get_connector_slug()
resp = app.get(path)
resp.form['slug'] = 'test2'
resp.form['title'] = 'API Particulier Test2'
resp.form['description'] = 'API Particulier Test2'
resp.form['platform'] = 'test'
resp.form['api_key'] = '83c68bf0b6013c4daf3f8213f7212aa5'
with HTTMock(api_particulier_introspect):
resp = resp.form.submit()
assert len(APIParticulier.objects.get(slug='test2').accessible_scopes) == 7
resp = resp.follow()
assert len(resp.html.find('ul', {'class': 'accessible-scopes'}).find_all('li')) == 7
resp = app.get(path)
resp.form['slug'] = 'test3'
resp.form['title'] = 'API Particulier Test3'
resp.form['description'] = 'API Particulier Test3'
resp.form['platform'] = 'test'
resp.form['api_key'] = '83c68bf0b6013c4daf3f8213f7212aa5'
with HTTMock(api_particulier_error_500):
resp = resp.form.submit()
assert not APIParticulier.objects.get(slug='test3').accessible_scopes
resp = resp.follow()
assert len(resp.html.find('ul', {'class': 'accessible-scopes'}).find_all('li')) == 0
def test_manager_edition(db, app, admin_user, resource):
app = login(app)
path = '/%s/%s/' % (resource.get_connector_slug(), resource.slug)
assert not APIParticulier.objects.get(slug=resource.slug).accessible_scopes
with HTTMock(api_particulier_introspect):
resp = app.get(path)
assert len(resp.html.find('ul', {'class': 'accessible-scopes'}).find_all('li')) == 0
path = '/manage/%s/%s/edit' % (resource.get_connector_slug(), resource.slug)
resp = app.get(path)
with HTTMock(api_particulier_introspect):
resp = resp.form.submit()
assert len(APIParticulier.objects.get(slug=resource.slug).accessible_scopes) == 7
resp = resp.follow()
assert len(resp.html.find('ul', {'class': 'accessible-scopes'}).find_all('li')) == 7
path = '/manage/%s/%s/edit' % (resource.get_connector_slug(), resource.slug)
resp = app.get(path)
with HTTMock(api_particulier_error_500):
resp = resp.form.submit()
assert not APIParticulier.objects.get(slug=resource.slug).accessible_scopes
resp = resp.follow()
assert len(resp.html.find('ul', {'class': 'accessible-scopes'}).find_all('li')) == 0