479 lines
17 KiB
Python
479 lines
17 KiB
Python
# passerelle - uniform access to multiple data sources and services
|
|
# Copyright (C) 2020 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 json
|
|
import os
|
|
|
|
import mock
|
|
import pytest
|
|
|
|
import utils
|
|
|
|
from passerelle.contrib.solis_afi_mss.models import SolisAfiMss
|
|
from passerelle.utils.jsonresponse import APIError
|
|
|
|
@pytest.fixture
|
|
def connector(db):
|
|
return utils.setup_access_rights(SolisAfiMss.objects.create(
|
|
slug='test',
|
|
base_url='https://dummy-server.org'
|
|
))
|
|
|
|
|
|
TEST_BASE_DIR = os.path.join(os.path.dirname(__file__), 'data', 'solis_afi_mss')
|
|
|
|
def json_get_data(filename):
|
|
with open(os.path.join(TEST_BASE_DIR, "%s.json" % filename)) as fd:
|
|
return json.dumps(json.load(fd))
|
|
|
|
def response(status_code, content):
|
|
return utils.FakedResponse(content=content, status_code=status_code)
|
|
|
|
|
|
IS_ALIVE = response(200, json_get_data('isAlive'))
|
|
|
|
ERROR_MSG = 'Solis API MSS AFI - Erreur: XXX'
|
|
ERROR = response(400, json_get_data('error'))
|
|
|
|
RECHERCHE_PAR_EMAIL_1 = response(200, json_get_data('rechercherParEmail_jacques_rousseau'))
|
|
RECHERCHE_PAR_EMAIL_2 = response(200, json_get_data('rechercherParEmail_michelle.delaunay'))
|
|
RECHERCHE_PAR_EMAIL_3 = response(200, json_get_data('rechercherParEmail_evelyne_pied'))
|
|
RECHERCHE_PAR_EMAIL_4 = response(200, json_get_data('rechercherParEmail_marie_noelle_basdevant'))
|
|
RECHERCHE_PAR_EMAIL_5 = response(200, json_get_data('rechercherParEmail_jean_christophe_hurel'))
|
|
|
|
GET_IMPOSITION_PAR_AGENT_5 = response(200, json_get_data('getImpositionsParAgent_388405'))
|
|
GET_IMPOSITION_5 = response(200, json_get_data('getImposition_388405_2019'))
|
|
GET_IMPOSITION_2 = response(204, "")
|
|
GET_IMPOSITION_0 = response(200, "")
|
|
|
|
DECLARER_IMPOT_1 = response(200, json_get_data('declarerImpot_389227'))
|
|
CALCULER_1 = response(200, json_get_data('calculer'))
|
|
DEPOSER_5 = response(200, json_get_data('deposer_388405'))
|
|
|
|
GET_AIDES_PAR_AGENT_2 = response(200, json_get_data('getAidesParAgent_394404'))
|
|
GET_AIDES_PAR_AGENT_5 = response(200, json_get_data('getAidesParAgent_388405'))
|
|
|
|
|
|
def get_endpoint(name):
|
|
return utils.generic_endpoint_url('solis-afi-mss', name)
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
@pytest.mark.parametrize('status_code, json_content, a_dict', [
|
|
(200, 'not json', None),
|
|
(500, '{"message": "help"}', {'message': 'help'}),
|
|
(500, 'not json', None),
|
|
])
|
|
def test_request_error(mocked_get, app, connector, status_code, json_content, a_dict):
|
|
mocked_get.side_effect = [response(status_code, json_content)]
|
|
with pytest.raises(APIError) as exc:
|
|
connector.request('some-url')
|
|
assert exc.value.err
|
|
if status_code == 200:
|
|
assert exc.value.http_status == 200
|
|
assert exc.value.args[0] == "invalid JSON content:'%s'" % json_content
|
|
else:
|
|
assert exc.value.data['status_code'] == status_code
|
|
assert exc.value.data['json_content'] == a_dict
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_check_status(mocked_get, app, connector):
|
|
mocked_get.side_effect = [IS_ALIVE]
|
|
connector.check_status()
|
|
assert mocked_get.mock_calls == [mock.call(
|
|
'https://dummy-server.org/main/isAlive/',
|
|
headers={'Accept': 'application/json'},
|
|
params=None
|
|
)]
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_check_status_error(mocked_get, app, connector):
|
|
mocked_get.side_effect = [response(500, '')]
|
|
with pytest.raises(APIError):
|
|
connector.check_status()
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
@pytest.mark.parametrize('response1, adults, children', [
|
|
(RECHERCHE_PAR_EMAIL_1, [
|
|
(389227, 'Jacques ROUSSEAU'),
|
|
(434729, 'Rina DI MARINO'),
|
|
], [
|
|
(389229, 'Lola ROUSSEAU'),
|
|
(389230, 'Nicolas ROUSSEAU'),
|
|
(389231, 'Mélina ROUSSEAU'),
|
|
]),
|
|
(RECHERCHE_PAR_EMAIL_2, [
|
|
(394404, 'Michelle DELAUNAY'),
|
|
], []),
|
|
(RECHERCHE_PAR_EMAIL_3, [
|
|
(388412, 'Louise PIED'),
|
|
], [
|
|
(388413, 'KEVIN PIED'),
|
|
]),
|
|
(RECHERCHE_PAR_EMAIL_4, [
|
|
(388420, 'Marie-Noëlle BASDEVANT'),
|
|
(434728, 'PIETRO BARTOLOMEO'),
|
|
], []),
|
|
(RECHERCHE_PAR_EMAIL_5, [
|
|
(388405, 'Jean-Christophe HUREL'),
|
|
(434727, 'CAROLE HUREL'),
|
|
], [
|
|
(388407, 'Camille HUREL'),
|
|
(388408, 'Valentin HUREL'),
|
|
]),
|
|
])
|
|
def test_search_from_email(mocked_get, app, connector, response1, adults, children):
|
|
mocked_get.side_effect = [response1]
|
|
result = connector.search_from_email('foo@dummy.org')
|
|
assert mocked_get.mock_calls == [mock.call(
|
|
'https://dummy-server.org/afi/agent/rechercherParEmail/',
|
|
headers={'Accept': 'application/json'},
|
|
params={'adresseMail': 'foo@dummy.org'})]
|
|
assert result[0] == adults[0][0] # agent index
|
|
assert [(x['id'], x['text']) for x in result[1]] == adults
|
|
assert [(x['id'], x['text']) for x in result[2]] == children
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_search_from_email_error(mocked_get, app, connector):
|
|
mocked_get.side_effect = [ERROR]
|
|
with pytest.raises(APIError) as exc:
|
|
connector.search_from_email('foo@dummy.org')
|
|
assert exc.value.data['status_code'] == 400
|
|
assert exc.value.data['json_content'][0]['message'] == ERROR_MSG
|
|
assert mocked_get.mock_calls == [mock.call(
|
|
'https://dummy-server.org/afi/agent/rechercherParEmail/',
|
|
headers={'Accept': 'application/json'},
|
|
params={'adresseMail': 'foo@dummy.org'})]
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
@pytest.mark.parametrize('response1, family', [
|
|
(RECHERCHE_PAR_EMAIL_1, [
|
|
(389227, 'Jacques ROUSSEAU'),
|
|
(434729, 'Rina DI MARINO'),
|
|
(389229, 'Lola ROUSSEAU'),
|
|
(389230, 'Nicolas ROUSSEAU'),
|
|
(389231, 'Mélina ROUSSEAU'),
|
|
]),
|
|
(RECHERCHE_PAR_EMAIL_2, [
|
|
(394404, 'Michelle DELAUNAY'),
|
|
]),
|
|
(RECHERCHE_PAR_EMAIL_3, [
|
|
(388412, 'Louise PIED'),
|
|
(388413, 'KEVIN PIED'),
|
|
]),
|
|
(RECHERCHE_PAR_EMAIL_4, [
|
|
(388420, 'Marie-Noëlle BASDEVANT'),
|
|
(434728, 'PIETRO BARTOLOMEO'),
|
|
]),
|
|
(RECHERCHE_PAR_EMAIL_5, [
|
|
(388405, 'Jean-Christophe HUREL'),
|
|
(434727, 'CAROLE HUREL'),
|
|
(388407, 'Camille HUREL'),
|
|
(388408, 'Valentin HUREL'),
|
|
]),
|
|
])
|
|
def test_family(mocked_get, app, connector, response1, family):
|
|
mocked_get.side_effect = [response1]
|
|
endpoint = get_endpoint('family') + '?email=foo@dummy.org'
|
|
resp = app.get(endpoint)
|
|
assert not resp.json['err']
|
|
assert [(x['id'], x['text']) for x in resp.json['data']] == family
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_family_error(mocked_get, app, connector):
|
|
mocked_get.side_effect = [ERROR]
|
|
endpoint = get_endpoint('family') + '?email=foo@dummy.org'
|
|
resp = app.get(endpoint)
|
|
assert resp.json['err']
|
|
assert ERROR_MSG in resp.json['err_desc']
|
|
assert resp.json['data']['json_content'][0]['message'] == ERROR_MSG
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
@pytest.mark.parametrize('response1, adults', [
|
|
(RECHERCHE_PAR_EMAIL_1, [
|
|
(389227, 'Jacques ROUSSEAU'),
|
|
(434729, 'Rina DI MARINO'),
|
|
]),
|
|
(RECHERCHE_PAR_EMAIL_2, [
|
|
(394404, 'Michelle DELAUNAY'),
|
|
]),
|
|
(RECHERCHE_PAR_EMAIL_3, [
|
|
(388412, 'Louise PIED'),
|
|
]),
|
|
(RECHERCHE_PAR_EMAIL_4, [
|
|
(388420, 'Marie-Noëlle BASDEVANT'),
|
|
(434728, 'PIETRO BARTOLOMEO'),
|
|
]),
|
|
(RECHERCHE_PAR_EMAIL_5, [
|
|
(388405, 'Jean-Christophe HUREL'),
|
|
(434727, 'CAROLE HUREL'),
|
|
]),
|
|
])
|
|
def test_adults(mocked_get, app, connector, response1, adults):
|
|
mocked_get.side_effect = [response1]
|
|
endpoint = get_endpoint('adults') + '?email=foo@dummy.org'
|
|
resp = app.get(endpoint)
|
|
assert not resp.json['err']
|
|
assert [(x['id'], x['text']) for x in resp.json['data']] == adults
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
@pytest.mark.parametrize('response1, children', [
|
|
(RECHERCHE_PAR_EMAIL_1, [
|
|
(389229, 'Lola ROUSSEAU'),
|
|
(389230, 'Nicolas ROUSSEAU'),
|
|
(389231, 'Mélina ROUSSEAU'),
|
|
]),
|
|
(RECHERCHE_PAR_EMAIL_2, [
|
|
]),
|
|
(RECHERCHE_PAR_EMAIL_3, [
|
|
(388413, 'KEVIN PIED'),
|
|
]),
|
|
(RECHERCHE_PAR_EMAIL_4, [
|
|
]),
|
|
(RECHERCHE_PAR_EMAIL_5, [
|
|
(388407, 'Camille HUREL'),
|
|
(388408, 'Valentin HUREL'),
|
|
]),
|
|
])
|
|
def test_children(mocked_get, app, connector, response1, children):
|
|
mocked_get.side_effect = [response1]
|
|
endpoint = get_endpoint('children') + '?email=foo@dummy.org'
|
|
resp = app.get(endpoint)
|
|
assert not resp.json['err']
|
|
assert [(x['id'], x['text']) for x in resp.json['data']] == children
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
@pytest.mark.parametrize('response1, response2, taxes', [
|
|
(RECHERCHE_PAR_EMAIL_5, GET_IMPOSITION_PAR_AGENT_5,
|
|
[(2018, '2018: 15000'), (2019, '2019: 1000')])
|
|
])
|
|
def test_taxes(mocked_get, app, connector, response1, response2, taxes):
|
|
mocked_get.side_effect = [response1, response2]
|
|
endpoint = get_endpoint('taxes') + '?email=foo@dummy.org'
|
|
resp = app.get(endpoint)
|
|
assert mocked_get.mock_calls[1] == mock.call(
|
|
'https://dummy-server.org/afi/budget/getImpositionsParAgent/',
|
|
headers={'Accept': 'application/json'},
|
|
params={'indexAgent': str(json.loads(response1.content)['indexAgent'])})
|
|
assert not resp.json['err']
|
|
assert [(x['id'], x['text']) for x in resp.json['data']] == taxes
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
@pytest.mark.parametrize('response1, response2, tax', [
|
|
(RECHERCHE_PAR_EMAIL_5, GET_IMPOSITION_5, 1000),
|
|
(RECHERCHE_PAR_EMAIL_5, GET_IMPOSITION_2, None),
|
|
(RECHERCHE_PAR_EMAIL_5, GET_IMPOSITION_0, None),
|
|
])
|
|
def test_taxes_for_year(mocked_get, app, connector, response1, response2, tax):
|
|
mocked_get.side_effect = [response1, response2]
|
|
endpoint = get_endpoint('taxes') + '?email=foo@dummy.org&year=2019'
|
|
resp = app.get(endpoint)
|
|
assert mocked_get.mock_calls[1] == mock.call(
|
|
'https://dummy-server.org/afi/budget/getImposition/',
|
|
headers={'Accept': 'application/json'},
|
|
params={
|
|
'indexAgent': str(json.loads(response1.content)['indexAgent']),
|
|
'anneeImposition': '2019'})
|
|
assert not resp.json['err']
|
|
if tax:
|
|
assert len(resp.json['data']) == 1
|
|
assert resp.json['data'][0]['montantImposition'] == tax
|
|
else:
|
|
assert resp.json['data'] == []
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
@pytest.mark.parametrize('response1, response2', [
|
|
(ERROR, None)
|
|
])
|
|
def test_taxes_error(mocked_get, app, connector, response1, response2):
|
|
mocked_get.side_effect = [response1, response2]
|
|
endpoint = get_endpoint('taxes') + '?email=foo@dummy.org'
|
|
resp = app.get(endpoint)
|
|
assert resp.json['err']
|
|
assert resp.json['data']['json_content'][0]['message'] == ERROR_MSG
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
@mock.patch('passerelle.utils.Request.post')
|
|
@pytest.mark.parametrize('response1, response2', [
|
|
(RECHERCHE_PAR_EMAIL_1, DECLARER_IMPOT_1),
|
|
])
|
|
def test_declare_tax(mocked_post, mocked_get, app, connector, response1, response2):
|
|
mocked_get.side_effect = [response1]
|
|
mocked_post.side_effect = [response2]
|
|
endpoint = get_endpoint('declare-tax')
|
|
payload = {
|
|
'email': 'foo@dummy.org',
|
|
'indexImposition': '222',
|
|
'anneeImposition': '2020',
|
|
'nombrePartImposition': '3.2',
|
|
'montantImposition': '777.77',
|
|
}
|
|
resp = app.post_json(endpoint, params=payload)
|
|
assert mocked_post.mock_calls == [mock.call(
|
|
'https://dummy-server.org/afi/budget/declarerImpot/',
|
|
headers={'Accept': 'application/json'},
|
|
json={
|
|
'indexAgent': '389227',
|
|
'indexImposition': '222',
|
|
'anneeImposition': '2020',
|
|
'nombrePartImposition': '3.2',
|
|
'montantImposition': '777.77',
|
|
})]
|
|
assert resp.json == {'err': 0, 'data': json.loads(response2.content)}
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
@mock.patch('passerelle.utils.Request.post')
|
|
@pytest.mark.parametrize('response1, response2', [
|
|
(ERROR, None),
|
|
(RECHERCHE_PAR_EMAIL_5, ERROR),
|
|
])
|
|
def test_declare_tax_error(mocked_post, mocked_get, app, connector, response1, response2):
|
|
mocked_get.side_effect = [response1]
|
|
mocked_post.side_effect = [response2]
|
|
endpoint = get_endpoint('declare-tax')
|
|
payload = {
|
|
'email': 'foo@dummy.org',
|
|
'indexImposition': '222',
|
|
'anneeImposition': '2020',
|
|
'nombrePartImposition': '3.2',
|
|
'montantImposition': '777.77',
|
|
}
|
|
resp = app.post_json(endpoint, params=payload)
|
|
assert resp.json['err']
|
|
assert resp.json['data']['json_content'][0]['message'] == ERROR_MSG
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
@pytest.mark.parametrize('response1, ratio', [
|
|
(CALCULER_1, 52.33),
|
|
])
|
|
def test_simulate_quotient(mocked_get, app, connector, response1, ratio):
|
|
mocked_get.side_effect = [response1]
|
|
endpoint = get_endpoint('simulate-quotient') + '?code=2&nb_parts=2.2&amount=222.22'
|
|
resp = app.get(endpoint)
|
|
assert mocked_get.mock_calls == [mock.call(
|
|
'https://dummy-server.org/afi/budget/calculer/',
|
|
headers={'Accept': 'application/json'},
|
|
params={'codeCalcul': '2', 'nbrPartImposition': '2.2', 'mntImposition': '222.22'})]
|
|
assert resp.json == {'err': 0, 'data': json.loads(response1.content)}
|
|
assert resp.json['data']['resultatCalcul'] == ratio
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_simulate_quotient_error(mocked_get, app, connector):
|
|
mocked_get.side_effect = [ERROR]
|
|
endpoint = get_endpoint('simulate-quotient') + '?code=2&nb_parts=2.2&amount=222.22'
|
|
resp = app.get(endpoint)
|
|
assert resp.json['err']
|
|
assert resp.json['data']['json_content'][0]['message'] == ERROR_MSG
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
@pytest.mark.parametrize('response1, response2, helps', [
|
|
(RECHERCHE_PAR_EMAIL_5, GET_AIDES_PAR_AGENT_2, []),
|
|
(RECHERCHE_PAR_EMAIL_5, GET_AIDES_PAR_AGENT_5, [
|
|
(37145, '2020-05-26 (En attente)'),
|
|
(37146, '2020-05-26 (En attente)'),
|
|
(37149, '2020-06-11 (En attente)'),
|
|
(37152, '2020-09-29 (En attente)'),
|
|
(37153, '2020-09-29 (En attente)'),
|
|
(37154, '2020-09-29 (En attente)'),
|
|
(37155, '2020-09-29 (En attente)')]),
|
|
])
|
|
def test_helps(mocked_get, app, connector, response1, response2, helps):
|
|
mocked_get.side_effect = [response1, response2]
|
|
endpoint = get_endpoint('helps') + '?email=foo@dummy.org'
|
|
resp = app.get(endpoint)
|
|
assert mocked_get.mock_calls[1] == mock.call(
|
|
'https://dummy-server.org/afi/aide/getAidesParAgent/',
|
|
headers={'Accept': 'application/json'},
|
|
params={'indexAgent': str(json.loads(response1.content)['indexAgent'])})
|
|
assert not resp.json['err']
|
|
assert [(x['id'], x['text']) for x in resp.json['data']] == helps
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
@mock.patch('passerelle.utils.Request.post')
|
|
@pytest.mark.parametrize('response1, response2', [
|
|
(RECHERCHE_PAR_EMAIL_5, DEPOSER_5),
|
|
])
|
|
def test_demand_help(mocked_post, mocked_get, app, connector, response1, response2):
|
|
mocked_get.side_effect = [response1]
|
|
mocked_post.side_effect = [response2]
|
|
endpoint = get_endpoint('demand-help')
|
|
|
|
payload = {
|
|
'email': 'foo@dummy.org',
|
|
'codeTypeAide': '24',
|
|
'natureTypeAide': 'A',
|
|
'individusConcernes': '388407:388408',
|
|
'dateDebut': '2020-07-15',
|
|
'dateFin': '2020-07-31',
|
|
'montantFacture': '2222.22',
|
|
}
|
|
resp = app.post_json(endpoint, params=payload)
|
|
assert mocked_post.mock_calls == [mock.call(
|
|
'https://dummy-server.org/afi/aide/deposer/',
|
|
headers={'Accept': 'application/json'},
|
|
json={
|
|
'indexAgent': '388405',
|
|
'codeTypeAide': '24',
|
|
'natureTypeAide': 'A',
|
|
'individusConcernes': [{'indexIndividu': '388407'}, {'indexIndividu': '388408'}],
|
|
'dateDebut': '2020-07-15',
|
|
'dateFin': '2020-07-31',
|
|
'montantFacture': '2222.22'
|
|
})]
|
|
assert resp.json == {'err': 0, 'data': json.loads(response2.content)}
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
@mock.patch('passerelle.utils.Request.post')
|
|
@pytest.mark.parametrize('response1, response2', [
|
|
(ERROR, None),
|
|
(RECHERCHE_PAR_EMAIL_5, ERROR),
|
|
])
|
|
def test_demand_help_error(mocked_post, mocked_get, app, connector, response1, response2):
|
|
mocked_get.side_effect = [response1]
|
|
mocked_post.side_effect = [response2]
|
|
endpoint = get_endpoint('demand-help')
|
|
|
|
payload = {
|
|
'email': 'foo@dummy.org',
|
|
'codeTypeAide': '24',
|
|
'natureTypeAide': 'A',
|
|
'individusConcernes': '388407:388408',
|
|
'dateDebut': '2020-07-15',
|
|
'dateFin': '2020-07-31',
|
|
'montantFacture': '2222.22',
|
|
}
|
|
resp = app.post_json(endpoint, params=payload)
|
|
assert resp.json['err']
|
|
assert resp.json['data']['json_content'][0]['message'] == ERROR_MSG
|