1136 lines
60 KiB
Python
1136 lines
60 KiB
Python
from io import StringIO
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.core.files import File
|
|
|
|
import tests.utils
|
|
from passerelle.apps.solis.models import Solis, SolisAPALink, SolisRSALink, unflat
|
|
from passerelle.base.models import AccessRight, ApiUser
|
|
|
|
NAMEID = 'bebe'
|
|
APATOKEN = '''{"token":"1c2562e6-b0a9-4bcf-b669-e33a42397147","endDate":"2017-10-11T10:22:40.342"}'''
|
|
APATOKEN_403 = '''[{"logref":"db15cb8a-4d05-4e4f-b4e1-44ec39dc11e3","message":
|
|
"Erreur d'authentification m\xc3\xa9tier ASG APA: Code confidentiel non valide pour l'individu 2823255","links":[]}]'''
|
|
APAINFOS = {
|
|
'exportDonneesIndividu': '''{"individu":{"civilite":"Mme","nomUsuel":"PYPPENNE","nomNaissance":"NPYNEZ","prenom":"Pecile","dateNaissance":"1922-12-17",
|
|
"contact":{"telephone":"0344480774","mail":""},"adresse":{"complementDestinataire":"compl dest","numeroLieu":"38","natureLieu":null,
|
|
"nomLieu":"RUE MARTIN","complementLieu":"MARPA LES NACRES - APPARTEMENT 9","finLieu":"fin adresse","codePostal":"80370","commune":"BERNAVILLE"},
|
|
"tutelles":{"tutelle":[{"type":"Organisme","identite":"Association Tut\xc3\xa9laire de la Somme","mesure":null,"natureAccord":"Juridique",
|
|
"dateEffet":"2014-01-01","dateFin":null,"adresse":{"numeroLieu":"21","natureLieu":null,"nomLieu":"RUE SULLY","complementLieu":"BP 11660","finLieu":"",
|
|
"codePostal":"80016","commune":"AMIENS","cedex":"1"}},{"type":"Individu/Particulier","identite":"Ehmet TYEP","mesure":"Curatelle simple",
|
|
"natureAccord":"Juridique","dateEffet":"2017-01-01","dateFin":"2017-12-31","adresse":{"numeroLieu":"89","natureLieu":null,"nomLieu":"AVENUE LEON BLUM",
|
|
"complementLieu":"","finLieu":"","codePostal":"80100","commune":"ABBEVILLE","cedex":""}},{"type":"Individu/Particulier",
|
|
"identite":"Esg TYTEYP PE PEPPOXE OEX","mesure":null,"natureAccord":null,"dateEffet":null,"dateFin":null,"adresse":{"numeroLieu":"1","natureLieu":null,
|
|
"nomLieu":"BOULEVARD DU PORT","complementLieu":"CAD","finLieu":"","codePostal":"80000","commune":"AMIENS","cedex":""}}]},"apa":{"classotheque":"05-2834",
|
|
"centreAutonomie":"Centre Autonomie nord ouest"}}}''',
|
|
'consultationDeMesDroits': '''{"demandeAsg":[{"demande":{"indexDemande":42,"type":"Allocation Personnalis\xc3\xa9e Autonomie","nature":"Domicile"},
|
|
"droit":{"dateDebut":"2017-01-05","dateFin":"2019-01-31"},"complementDossier":{"dateDepot":"2016-11-15","dateArrivee":"2016-11-16",
|
|
"dateDossierComplet":"2016-11-17"},"gir":{"type":"Synth\xc3\xa8se","gir":3,"dateEvaluation":"2017-02-01"},"suivi":{"instructeur":{"civilite":"madame",
|
|
"nom":"ZEPEQPE","prenom":"EPOZOE","telephone":"0344974383","mail":"e.zepeqpe@xoppe.pp"},"dateDecision":"2017-01-05"},
|
|
"bilan":{"pourcentageTM":2.1973443031311035},"prestationAccordeeAsg":[{"prestation":"Accueil de jour GIR 1-2","periode":{"dateEffet":"2017-01-05",
|
|
"dateFin":"2019-01-31"},"tiers":{"type":"Ind\xc3\xa9termin\xc3\xa9","identite":null,"tarif":null,"quantitatif":null},"quantite":0,"montant":{"accorde":0,
|
|
"participation":0,"verse":0},"attributaire":{"type":"Tuteur","identite":"Association Tut\xc3\xa9laire de la Somme"}},
|
|
{"prestation":"Articles d\'hygi\xc3\xa8ne forfait 45\xc2\x80","periode":{"dateEffet":"2017-01-05","dateFin":null},"tiers":{"type":"Ind\xc3\xa9termin\xc3\xa9",
|
|
"identite":null,"tarif":45,"quantitatif":"Mois"},"quantite":1,"montant":{"accorde":45,"participation":1.68,"verse":43.32},
|
|
"attributaire":{"type":"B\xc3\xa9n\xc3\xa9ficiaire","identite":"PYPPENNE Pecile"}},{"prestation":"Petite structure","periode":{"dateEffet":"2017-01-05",
|
|
"dateFin":"2019-01-31"},"tiers":{"type":"Etablissement","identite":"MARPA LES NACRES","tarif":null,"quantitatif":null},"quantite":0,
|
|
"montant":{"accorde":440.42,"participation":7.68,"verse":432.74},"attributaire":{"type":"Etablissement","identite":"MARPA LES NACRES"}},
|
|
{"prestation":"Aide humaine prestataire","periode":{"dateEffet":"2017-01-05","dateFin":"2019-01-31"},"tiers":{"type":"Prestataire",
|
|
"identite":"COMMUNAUTE DE COMMUNES DU TERRITOIRE NORD PICARDIE BERNAVILLE","tarif":19,"quantitatif":"Heure(s)"},"quantite":45.5,"montant":{"accorde":864.5,
|
|
"participation":18.93,"verse":845.57},"attributaire":{"type":"Prestataire","identite":"COMMUNAUTE DE COMMUNES DU TERRITOIRE NORD PICARDIE BERNAVILLE"}},
|
|
{"prestation":"Articles d\'hygi\xc3\xa8ne forfait 90\xc2\x80","periode":{"dateEffet":"2017-01-05","dateFin":null},"tiers":{"type":"Ind\xc3\xa9termin\xc3\xa9",
|
|
"identite":null,"tarif":90,"quantitatif":"Mois"},"quantite":1,"montant":{"accorde":90,"participation":3.35,"verse":86.65},
|
|
"attributaire":{"type":"B\xc3\xa9n\xc3\xa9ficiaire","identite":"PYPPENNE Pecile"}}]}]}''',
|
|
'suiviDemandeEnInstruction': '{"demandeAsg":[]}',
|
|
'suiviDemandeHistorique': '''{"demandeAsg":[{"demande":{"indexDemande":42,"type":"Allocation Personnalis\xc3\xa9e Autonomie","nature":"Domicile"},
|
|
"droit":{"dateDebut":"2013-03-01","dateFin":"2013-06-19"},"complementDossier":{"dateArrivee":null,"dateDossierComplet":"2012-10-25"},
|
|
"suivi":{"decision":"Accord","dateDecision":"2013-02-12"}},{"demande":{"indexDemande":43,"type":"Allocation Personnalis\xc3\xa9e Autonomie",
|
|
"nature":"Domicile"},"droit":{"dateDebut":"2013-06-20","dateFin":"2016-03-31"},"complementDossier":{"dateArrivee":null,"dateDossierComplet":"2012-10-25"},
|
|
"suivi":{"decision":"Accord","dateDecision":"2013-06-25"}},{"demande":{"indexDemande":44,"type":"Allocation Personnalis\xc3\xa9e Autonomie",
|
|
"nature":"Domicile"},"droit":{"dateDebut":"2016-04-01","dateFin":"2017-01-04"},"complementDossier":{"dateArrivee":"2016-06-06",
|
|
"dateDossierComplet":"2016-06-06"},"suivi":{"decision":"Accord","dateDecision":"2016-06-14"}}]}''',
|
|
'propositionPlanAide': '''{"demandeAsg":[{"demande":{"type":"Allocation Personnalis\\u00e9e Autonomie","indexDemande":42,"nature":"Domicile"},
|
|
"droit":{"dateDebut":"2016-08-23","dateFin":"2018-08-31"},"planAide":{"commentaire":"",
|
|
"prestationsPlanAide":{"prestationPlanAide":[{"tiers":{"tarif":12.8,"identite":"CCAS DE MERS LES BAINS","quantitatif":"Heure(s)","type":"Prestataire"},
|
|
"quantite":84,"montant":1075.2,"prestation":"Aide humaine mandataire"},{"tiers":{"tarif":90,"identite":null,"quantitatif":"Mois",
|
|
"type":"Ind\\u00e9termin\\u00e9"},"quantite":1,"montant":90,"prestation":"Articles d\'hygi\\u00e8ne forfait 90"}]},"dateReponse":null,
|
|
"datePropositionPlan":null,"avis":""},"complementDossier":{"dateDepot":"2016-06-22"},"suivi":{"dateDecision":"2016-08-23"}}]}''',
|
|
}
|
|
APAREQUEST = '''{"demandeAsg":{"visite":{"date":"2016-07-07","heure":"1330"},"demande":{"type":"Allocation Personnalis\\u00e9e Autonomie",
|
|
"indexDemande":42,"nature":"Domicile"},"droit":{"dateDebut":"2016-08-23","dateFin":"2018-08-31"},"complementDossier":{"dateArrivee":"2016-06-22",
|
|
"dateDossierComplet":"2016-06-22"},"suivi":{"dateDecision":"2016-08-23","decision":"Accord"}}}'''
|
|
DEPARTEMENTS = '''{"departements":[{"code":"1","libelle":"Ain","pays":{"code":"79","libelle":"France"}},{"code":"2","libelle":"Aisne",
|
|
"pays":{"code":"79","libelle":"France"}},{"code":"3","libelle":"Allier","pays":{"code":"79","libelle":"France"}},
|
|
{"code":"4","libelle":"Alpes de Haute Provence","pays":{"code":"79","libelle":"France"}},{"code":"5","libelle":"Hautes Alpes",
|
|
"pays":{"code":"79","libelle":"France"}},{"code":"6","libelle":"Alpes Maritimes","pays":{"code":"79","libelle":"France"}},
|
|
{"code":"7","libelle":"Ardèche","pays":{"code":"79","libelle":"France"}},{"code":"8","libelle":"Ardennes","pays":{"code":"79","libelle":"France"}}]}'''
|
|
CIVI_INDIVIDU = '''{"index":4273,"referentDossier":true,"presentFoyer":true,"etatCivil":{"genre":"MME","sexe":"F","nom":"NOM",
|
|
"prenom":"Prenom","nomNaissance":"NEENOM","dateNaissance":"1950-12-10","lieuNaissance":"NOWHERE"}}'''
|
|
|
|
RSATOKEN = '''{
|
|
"token": "e18f0967-1b8b-4ae5-8e7a-3e89076429bd",
|
|
"endDate": "2019-05-21T12:23:08.004"
|
|
}'''
|
|
RSATOKEN_403 = '''[
|
|
{
|
|
"logref": "62215971-42dc-41af-a0c2-fdbe8115be73",
|
|
"message": "rSa - Référentiels - Accès non autorisé aux référentiels rSa",
|
|
"links": []
|
|
}
|
|
]'''
|
|
RSAALLOCATAIRES = '''{
|
|
"identifiant": {
|
|
"codeAllocataire": "12345",
|
|
"dateNaissance": "10/10/1966"
|
|
},
|
|
"poleEmploi": null,
|
|
"nir": "2760150025241",
|
|
"soumisDD": "Non soumis à droit et devoir",
|
|
"oriente": true,
|
|
"_links": {
|
|
"etatCivil": {
|
|
"href": "https://solis.example.net/solisapi/referentiels/civi/individu/4242/"
|
|
},
|
|
"conjoint": {
|
|
"href": "http://solis.example.net/solisapi/referentiels/civi/individu/4273/"
|
|
}
|
|
}
|
|
}'''
|
|
RSA_3LINKS = '''{
|
|
"_links": {
|
|
"etatCivil": {
|
|
"href": "https://solis.example.net/solisapi/referentiels/civi/individu/4242/"
|
|
},
|
|
"refOrientStructAcc": [
|
|
{"href": "https://solis.example.net/solisapi/referentiels/civi/individu/4242/"},
|
|
{"href": "http://solis.example.net/solisapi/referentiels/civi/individu/4242/"}
|
|
]
|
|
}
|
|
}'''
|
|
RSA_ACTIONS = '''[
|
|
{"idStructure":42, "prescriptionPlacement":"Placement",
|
|
"_links": {"structure":
|
|
{"href": "https://solis.example.net/solisapi/referentiels/trans/organisme/6481/"}}},
|
|
{"idStructure":42, "prescriptionPlacement":"Prescription"},
|
|
{"idStructure":43, "prescriptionPlacement":"Prescription",
|
|
"_links": {"structure":
|
|
{"href": "https://solis.example.net/solisapi/referentiels/trans/organisme/6482/"}}},
|
|
{"idStructure":44, "prescriptionPlacement":"Placement"}
|
|
]'''
|
|
|
|
|
|
@pytest.fixture
|
|
def solis(db):
|
|
return Solis.objects.create(
|
|
slug='test',
|
|
service_url='https://solis.example.net/solisapi/',
|
|
basic_auth_username='usertest',
|
|
basic_auth_password='userpass',
|
|
)
|
|
|
|
|
|
def test_solis_restricted_access(app, solis):
|
|
for service in ('apa-link', 'apa-unlink', 'rsa-link', 'rsa-unlink'):
|
|
endpoint = tests.utils.generic_endpoint_url('solis', service, slug=solis.slug)
|
|
assert endpoint == '/solis/test/%s' % service
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
resp = app.post(endpoint, status=403)
|
|
assert requests_post.call_count == 0
|
|
assert resp.json['err'] == 1
|
|
assert 'PermissionDenied' in resp.json['err_class']
|
|
resp = app.get(endpoint, status=405)
|
|
assert requests_get.call_count == 0
|
|
for service in ('apa-links', 'apa-user-info', 'apa-users', 'rsa-links', 'rsa-user-info'):
|
|
endpoint = tests.utils.generic_endpoint_url('solis', service, slug=solis.slug)
|
|
assert endpoint == '/solis/test/%s' % service
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
resp = app.get(endpoint, status=403)
|
|
assert requests_get.call_count == 0
|
|
assert resp.json['err'] == 1
|
|
assert 'PermissionDenied' in resp.json['err_class']
|
|
resp = app.post(endpoint, status=405)
|
|
assert requests_post.call_count == 0
|
|
|
|
|
|
@pytest.fixture
|
|
def ping_response():
|
|
response_request = mock.Mock(headers={'Accept': '*/*'}, body=None)
|
|
return tests.utils.FakedResponse(
|
|
headers={'Content-Type': 'text/plain'}, status_code=200, request=response_request
|
|
)
|
|
|
|
|
|
def test_solis_ping(app, solis, ping_response):
|
|
# full opened access
|
|
api = ApiUser.objects.create(username='all', keytype='', key='')
|
|
obj_type = ContentType.objects.get_for_model(solis)
|
|
AccessRight.objects.create(
|
|
codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=solis.pk
|
|
)
|
|
|
|
endpoint = tests.utils.generic_endpoint_url('solis', 'ping', slug=solis.slug)
|
|
with mock.patch('passerelle.utils.RequestSession.request') as requests_get:
|
|
for bad_content in (
|
|
'error',
|
|
'{"foo": "bar"}',
|
|
'["not", "a", "dict"]',
|
|
'{"response": "Solis API est en panne"}',
|
|
):
|
|
ping_response.content = bad_content
|
|
requests_get.return_value = ping_response
|
|
resp = app.get(endpoint, status=200)
|
|
assert resp.json['err'] == 1
|
|
|
|
ping_response.content = '{"response":"Solis API est opérationnel."}'
|
|
requests_get.return_value = ping_response
|
|
resp = app.get(endpoint, status=200)
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data'] == 'pong'
|
|
assert resp.json['response'] == 'Solis API est opérationnel.'
|
|
|
|
assert requests_get.call_args[1]['auth'] == ('usertest', 'userpass')
|
|
assert requests_get.call_args[1]['verify'] is True
|
|
assert 'cert' not in requests_get.call_args[1]
|
|
assert 'proxies' not in requests_get.call_args[1]
|
|
assert requests_get.call_args[1]['timeout'] == 25
|
|
|
|
# try certificates parameters
|
|
solis.verify_cert = False
|
|
solis.save()
|
|
resp = app.get(endpoint, status=200)
|
|
assert requests_get.call_args[1]['verify'] is False
|
|
|
|
solis.trusted_certificate_authorities = File(StringIO('CA'), 'ca.pem')
|
|
solis.save()
|
|
resp = app.get(endpoint, status=200)
|
|
assert requests_get.call_args[1]['verify'] == solis.trusted_certificate_authorities.path
|
|
assert 'cert' not in requests_get.call_args[1]
|
|
|
|
solis.client_certificate = File(StringIO('KS'), 'crt.pem')
|
|
solis.save()
|
|
resp = app.get(endpoint, status=200)
|
|
assert requests_get.call_args[1]['cert'] == solis.client_certificate.path
|
|
assert requests_get.call_args[1]['verify'] == solis.trusted_certificate_authorities.path
|
|
|
|
solis.trusted_certificate_authorities = None
|
|
solis.save()
|
|
resp = app.get(endpoint, status=200)
|
|
assert requests_get.call_args[1]['verify'] is False
|
|
assert requests_get.call_args[1]['cert'] == solis.client_certificate.path
|
|
|
|
# try proxy parameter
|
|
solis.http_proxy = 'http://proxy:3128/'
|
|
solis.save()
|
|
resp = app.get(endpoint, status=200)
|
|
assert requests_get.call_args[1]['proxies'] == {
|
|
'http': 'http://proxy:3128/',
|
|
'https': 'http://proxy:3128/',
|
|
}
|
|
|
|
|
|
def test_solis_apa_link_infos_unlink(app, solis):
|
|
# full opened access
|
|
api = ApiUser.objects.create(username='all', keytype='', key='')
|
|
obj_type = ContentType.objects.get_for_model(solis)
|
|
AccessRight.objects.create(
|
|
codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=solis.pk
|
|
)
|
|
|
|
# link
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post: # get solis token
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get: # get solis informations
|
|
endpoint = tests.utils.generic_endpoint_url('solis', 'apa-link', slug=solis.slug)
|
|
for params in (None, '', []):
|
|
resp = app.post_json(endpoint, params=params, status=200)
|
|
assert requests_post.call_count == 0
|
|
assert resp.json['err'] == 1
|
|
assert 'payload is not a JSON dict' in resp.json['err_desc']
|
|
|
|
for params in ({}, {'user_id': 'x'}, {'code': 'x'}, {'foo': 'bar'}):
|
|
resp = app.post_json(endpoint, params=params, status=200)
|
|
assert requests_post.call_count == 0
|
|
assert resp.json['err'] == 1
|
|
assert 'missing name_id' in resp.json['err_desc']
|
|
params['name_id'] = 'xx'
|
|
resp = app.post_json(endpoint, params=params, status=200)
|
|
assert requests_post.call_count == 0
|
|
assert resp.json['err'] == 1
|
|
assert 'missing user_id/code credentials' in resp.json['err_desc']
|
|
|
|
requests_post.return_value = tests.utils.FakedResponse(content=APATOKEN_403, status_code=403)
|
|
resp = app.post_json(
|
|
endpoint, params={'user_id': 'x', 'code': 'x', 'name_id': NAMEID}, status=200
|
|
)
|
|
assert requests_post.call_count == 1
|
|
assert requests_get.call_count == 0
|
|
assert resp.json['err'] == 1
|
|
assert 'Code confidentiel non valide' in resp.json['err_desc']
|
|
|
|
assert SolisAPALink.objects.count() == 0
|
|
|
|
requests_post.return_value = tests.utils.FakedResponse(content=APATOKEN, status_code=200)
|
|
requests_get.return_value = tests.utils.FakedResponse(
|
|
content=APAINFOS['exportDonneesIndividu'], status_code=200
|
|
)
|
|
resp = app.post_json(
|
|
endpoint, params={'name_id': NAMEID, 'user_id': '42', 'code': 'foo'}, status=200
|
|
)
|
|
assert requests_post.call_count == 2
|
|
assert requests_get.call_count == 1
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['user_id'] == '42'
|
|
assert resp.json['data']['created']
|
|
assert not resp.json['data']['updated']
|
|
assert SolisAPALink.objects.count() == 1
|
|
assert SolisAPALink.objects.first().name_id == NAMEID
|
|
assert SolisAPALink.objects.first().user_id == '42'
|
|
assert SolisAPALink.objects.first().code == 'foo'
|
|
assert SolisAPALink.objects.first().text == 'Mme Pecile PYPPENNE (NPYNEZ)'
|
|
|
|
# change code
|
|
resp = app.post_json(
|
|
endpoint, params={'name_id': NAMEID, 'user_id': '42', 'code': 'bar'}, status=200
|
|
)
|
|
assert requests_post.call_count == 3
|
|
assert requests_get.call_count == 2
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['user_id'] == '42'
|
|
assert not resp.json['data']['created']
|
|
assert resp.json['data']['updated']
|
|
assert SolisAPALink.objects.count() == 1
|
|
assert SolisAPALink.objects.first().name_id == NAMEID
|
|
assert SolisAPALink.objects.first().user_id == '42'
|
|
assert SolisAPALink.objects.first().code == 'bar'
|
|
assert SolisAPALink.objects.first().text == 'Mme Pecile PYPPENNE (NPYNEZ)'
|
|
|
|
# second link
|
|
resp = app.post_json(
|
|
endpoint, params={'name_id': NAMEID, 'user_id': '53', 'code': 'bar'}, status=200
|
|
)
|
|
assert requests_post.call_count == 4
|
|
assert requests_get.call_count == 3
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['user_id'] == '53'
|
|
assert resp.json['data']['created']
|
|
assert not resp.json['data']['updated']
|
|
assert SolisAPALink.objects.count() == 2
|
|
|
|
# verify recorded names after link
|
|
assert [x['text'] for x in SolisAPALink.objects.values('text')] == [
|
|
'Mme Pecile PYPPENNE (NPYNEZ)',
|
|
'Mme Pecile PYPPENNE (NPYNEZ)',
|
|
]
|
|
endpoint = tests.utils.generic_endpoint_url('solis', 'apa-links', slug=solis.slug)
|
|
resp = app.get(endpoint, status=400) # missing name_id
|
|
assert resp.json['err'] == 1
|
|
endpoint += '?name_id=%s' % NAMEID
|
|
resp = app.get(endpoint, status=200)
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 2
|
|
assert resp.json['data'][0]['text'] == resp.json['data'][1]['text'] == 'Mme Pecile PYPPENNE (NPYNEZ)'
|
|
|
|
# get base informations from a linked user (exportDonneesIndividu)
|
|
changed_name = APAINFOS['exportDonneesIndividu'].replace('PYPPENNE', 'PEPONE')
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = tests.utils.FakedResponse(content=APATOKEN, status_code=200)
|
|
requests_get.return_value = tests.utils.FakedResponse(content=changed_name, status_code=200)
|
|
endpoint = tests.utils.generic_endpoint_url('solis', 'apa-user-info', slug=solis.slug)
|
|
endpoint += '?name_id=%s&user_id=42' % NAMEID
|
|
resp = app.get(endpoint, status=200)
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['individu']['nomUsuel'] == 'PEPONE'
|
|
# user "text" updated in link:
|
|
assert SolisAPALink.objects.get(name_id=NAMEID, user_id='42').text == 'Mme Pecile PEPONE (NPYNEZ)'
|
|
|
|
# get all kind of informations
|
|
for apa_endpoint, content in APAINFOS.items():
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = tests.utils.FakedResponse(content=APATOKEN, status_code=200)
|
|
|
|
endpoint_base = tests.utils.generic_endpoint_url('solis', 'apa-user-info', slug=solis.slug)
|
|
resp = app.get(endpoint_base, status=400) # missing name_id
|
|
assert resp.json['err'] == 1
|
|
|
|
endpoint = endpoint_base + '?name_id=%s&user_id=53&information=%s' % (NAMEID, apa_endpoint)
|
|
requests_get.return_value = tests.utils.FakedResponse(content=content, status_code=200)
|
|
resp = app.get(endpoint, status=200)
|
|
assert requests_post.call_count == 1 # get a token
|
|
assert requests_get.call_count == 1 # get informations
|
|
assert ('/asg/apa/%s' % apa_endpoint) in requests_get.call_args[0][0]
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']
|
|
|
|
# solis api crash
|
|
requests_get.return_value = tests.utils.FakedResponse(content='boum', status_code=500)
|
|
resp = app.get(endpoint, status=200)
|
|
assert requests_post.call_count == 2 # get a token
|
|
assert requests_get.call_count == 2 # get informations
|
|
assert ('/asg/apa/%s' % apa_endpoint) in requests_get.call_args[0][0]
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'].startswith('error status:500')
|
|
assert resp.json['data'] == {'json_content': None, 'status_code': 500}
|
|
|
|
requests_get.return_value = tests.utils.FakedResponse(
|
|
content='{"error":"foobar"}', status_code=500
|
|
)
|
|
resp = app.get(endpoint, status=200)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'].startswith('error status:500')
|
|
assert resp.json['data'] == {'json_content': {'error': 'foobar'}, 'status_code': 500}
|
|
|
|
# unknown name_id or user_id
|
|
for qs in ('name_id=%s&user_id=XXX' % NAMEID, 'name_id=unlinked&user_id=53'):
|
|
endpoint = endpoint_base + ('?information=%s' % apa_endpoint) + '&' + qs
|
|
resp = app.get(endpoint, status=200)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'unknown link'
|
|
assert resp.json['data'] is None
|
|
|
|
# get info about a specific request
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = tests.utils.FakedResponse(content=APATOKEN, status_code=200)
|
|
endpoint_base = tests.utils.generic_endpoint_url('solis', 'apa-user-info', slug=solis.slug)
|
|
|
|
# via demandeUnitaire
|
|
endpoint = endpoint_base + '?name_id=%s&user_id=53&information=demandeUnitaire&index=42' % NAMEID
|
|
requests_get.return_value = tests.utils.FakedResponse(content=APAREQUEST, status_code=200)
|
|
resp = app.get(endpoint, status=200)
|
|
assert requests_post.call_count == 1 # get a token
|
|
assert requests_get.call_count == 1 # get demandeUnitaire
|
|
url = requests_get.call_args[0][0]
|
|
assert '/asg/apa/demandeUnitaire/' in url
|
|
assert url.endswith('/42')
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['demandeAsg']['demande']['indexDemande'] == 42
|
|
|
|
requests_post.reset_mock()
|
|
requests_get.reset_mock()
|
|
requests_get.return_value = tests.utils.FakedResponse(content='nothing here', status_code=404)
|
|
resp = app.get(endpoint, status=200)
|
|
assert requests_post.call_count == 1 # get a token
|
|
assert requests_get.call_count == 1 # get demandeUnitaire
|
|
url = requests_get.call_args[0][0]
|
|
assert '/asg/apa/demandeUnitaire/' in url
|
|
assert url.endswith('/42')
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'].startswith('error status:404')
|
|
|
|
# missing index
|
|
requests_post.reset_mock()
|
|
requests_get.reset_mock()
|
|
endpoint = endpoint_base + '?name_id=%s&user_id=53&information=demandeUnitaire' % NAMEID
|
|
resp = app.get(endpoint, status=400)
|
|
requests_get.assert_not_called()
|
|
requests_post.assert_not_called()
|
|
assert resp.json['err'] == 1
|
|
|
|
# get indexDemande 42 in lists
|
|
for information in (
|
|
'consultationDeMesDroits',
|
|
'suiviDemandeHistorique',
|
|
'propositionPlanAide',
|
|
):
|
|
requests_post.reset_mock()
|
|
requests_get.reset_mock()
|
|
endpoint = endpoint_base + '?name_id=%s&user_id=53&information=%s&index=42' % (
|
|
NAMEID,
|
|
information,
|
|
)
|
|
requests_get.return_value = tests.utils.FakedResponse(
|
|
content=APAINFOS[information], status_code=200
|
|
)
|
|
resp = app.get(endpoint, status=200)
|
|
assert ('/asg/apa/%s/' % information) in requests_get.call_args[0][0]
|
|
requests_post.assert_called_once()
|
|
requests_get.assert_called_once()
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['demandeAsg']['demande']['indexDemande'] == 42
|
|
|
|
endpoint = endpoint_base + '?name_id=%s&user_id=53&information=%s&index=57' % (
|
|
NAMEID,
|
|
information,
|
|
)
|
|
resp = app.get(endpoint, status=200)
|
|
assert ('/asg/apa/%s/' % information) in requests_get.call_args[0][0]
|
|
assert requests_post.call_count == 2
|
|
assert requests_get.call_count == 2
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'cannot find indexDemande=57 in demandeAsg list'
|
|
|
|
# get informations for all users (exportDonneesIndividu)
|
|
change_info = APAINFOS['exportDonneesIndividu'].replace('PYPPENNE', 'PEPPYNE')
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = tests.utils.FakedResponse(content=APATOKEN, status_code=200)
|
|
requests_get.return_value = tests.utils.FakedResponse(content=change_info, status_code=200)
|
|
endpoint = tests.utils.generic_endpoint_url('solis', 'apa-users', slug=solis.slug)
|
|
endpoint += '?name_id=%s' % NAMEID
|
|
resp = app.get(endpoint, status=200)
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 2
|
|
assert requests_post.call_count == 2
|
|
assert requests_get.call_count == 2
|
|
assert {x['id'] for x in resp.json['data']} == {'42', '53'}
|
|
assert resp.json['data'][0]['text'] == 'Mme Pecile PEPPYNE (NPYNEZ)'
|
|
# user "text" updated in links:
|
|
assert [x['text'] for x in SolisAPALink.objects.values('text')] == [
|
|
'Mme Pecile PEPPYNE (NPYNEZ)',
|
|
'Mme Pecile PEPPYNE (NPYNEZ)',
|
|
]
|
|
|
|
# unlink
|
|
endpoint = tests.utils.generic_endpoint_url('solis', 'apa-unlink', slug=solis.slug)
|
|
for bad_params in ({}, {'user_id': '42'}, {'name_id': NAMEID}):
|
|
resp = app.post_json(endpoint, params=bad_params, status=200)
|
|
assert resp.json['err'] == 1
|
|
|
|
resp = app.post_json(endpoint, params={'user_id': 'xxx', 'name_id': 'xxx'}, status=200)
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['deleted']
|
|
assert resp.json['data']['user_id'] == 'xxx'
|
|
assert SolisAPALink.objects.count() == 2
|
|
|
|
resp = app.post_json(endpoint, params={'user_id': '42', 'name_id': NAMEID}, status=200)
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['deleted']
|
|
assert resp.json['data']['user_id'] == '42'
|
|
assert SolisAPALink.objects.count() == 1
|
|
|
|
# unlink again, no trouble
|
|
resp = app.post_json(endpoint, params={'user_id': '42', 'name_id': NAMEID}, status=200)
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['deleted']
|
|
assert resp.json['data']['user_id'] == '42'
|
|
assert SolisAPALink.objects.count() == 1
|
|
|
|
# can not get informations from unlinked user
|
|
endpoint = tests.utils.generic_endpoint_url('solis', 'apa-user-info', slug=solis.slug)
|
|
endpoint += '?name_id=%s&user_id=42' % NAMEID
|
|
resp = app.get(endpoint, status=200)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'unknown link'
|
|
assert resp.json['data'] is None
|
|
|
|
|
|
def test_solis_referentiels(app, solis):
|
|
# full opened access
|
|
api = ApiUser.objects.create(username='all', keytype='', key='')
|
|
obj_type = ContentType.objects.get_for_model(solis)
|
|
AccessRight.objects.create(
|
|
codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=solis.pk
|
|
)
|
|
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = tests.utils.FakedResponse(content=DEPARTEMENTS, status_code=200)
|
|
url = tests.utils.generic_endpoint_url('solis', 'referential', slug=solis.slug)
|
|
|
|
resp = app.get(url + '/trans/departement/', status=200)
|
|
assert requests_get.call_count == 1
|
|
assert requests_get.call_args[0][0].endswith('/solisapi/referentiels/trans/departement')
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 8
|
|
|
|
resp = app.get(url + '/trans/departement/?q=ardeche', status=200)
|
|
assert requests_get.call_args[0][0].endswith('/solisapi/referentiels/trans/departement')
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 1
|
|
assert resp.json['data'][0]['text'] == 'Ardèche'
|
|
|
|
resp = app.get(url + '/trans/departement/?q=arde', status=200)
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 2
|
|
assert (resp.json['data'][0]['text'], resp.json['data'][1]['text']) == ('Ardèche', 'Ardennes')
|
|
|
|
resp = app.get(url + '/trans/departement/?id=7', status=200)
|
|
assert requests_get.call_args[0][0].endswith('/solisapi/referentiels/trans/departement')
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 1
|
|
assert resp.json['data'][0]['text'] == 'Ardèche'
|
|
|
|
resp = app.get(url + '/trans/departement/?id=99', status=200)
|
|
assert requests_get.call_args[0][0].endswith('/solisapi/referentiels/trans/departement')
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 0
|
|
|
|
resp = app.get(url + '/trans/departement/?q=arde&ignore=8', status=200)
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 1
|
|
assert resp.json['data'][0]['text'] == 'Ardèche'
|
|
|
|
resp = app.get(url + '/trans/departement/?q=arde&ignore=8,, 7', status=200)
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 0
|
|
|
|
resp = app.get(url + '/trans/departement/?codePays=79&foo=bar', status=200)
|
|
assert requests_get.call_args[0][0].endswith('/solisapi/referentiels/trans/departement?codePays=79')
|
|
|
|
resp = app.get(url + '/trans/commune/?codePays=79&filtreSurInactivite=true&bar=foo', status=200)
|
|
called_url = requests_get.call_args[0][0]
|
|
assert '/solisapi/referentiels/trans/commune?' in called_url
|
|
assert 'filtreSurInactivite=true' in called_url
|
|
assert 'codePays=79' in called_url
|
|
assert 'foo=bar' not in called_url
|
|
|
|
requests_get.return_value = tests.utils.FakedResponse(
|
|
content='{"nada":0}', status_code=404, reason='Not found'
|
|
)
|
|
resp = app.get(url + '/foo/bar/', status=200)
|
|
assert requests_get.call_args[0][0].endswith('/solisapi/referentiels/foo/bar')
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == "error status:404 'Not found', content:'{\"nada\":0}'"
|
|
assert resp.json['data'] == {'json_content': {'nada': 0}, 'status_code': 404}
|
|
|
|
requests_get.return_value = tests.utils.FakedResponse(content='crash', status_code=500, reason='boum')
|
|
resp = app.get(url + '/foo/bar/', status=200)
|
|
assert requests_get.call_args[0][0].endswith('/solisapi/referentiels/foo/bar')
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == "error status:500 'boum', content:'crash'"
|
|
assert resp.json['data'] == {'json_content': None, 'status_code': 500}
|
|
|
|
# optimized trans/lieu search
|
|
requests_get.return_value = tests.utils.FakedResponse(content='{"lieux":[]}', status_code=200)
|
|
resp = app.get(url + '/trans/lieu/', status=200)
|
|
assert requests_get.call_args[0][0].endswith('/solisapi/referentiels/trans/lieu')
|
|
resp = app.get(url + '/trans/lieu/?codeDepartement=1&codeCommune=2&q=search', status=200)
|
|
assert requests_get.call_args[0][0].endswith('/solisapi/referentiels/trans/nomlieu/1/2/search')
|
|
resp = app.get(url + '/trans/lieu/?codeDepartement=1&codeCommune=2&codeLieu=3', status=200)
|
|
assert 'referentiels/trans/lieu?' in requests_get.call_args[0][0]
|
|
assert 'codeLieu=3' in requests_get.call_args[0][0]
|
|
|
|
|
|
def test_solis_referential_item(app, solis):
|
|
# full opened access
|
|
api = ApiUser.objects.create(username='all', keytype='', key='')
|
|
obj_type = ContentType.objects.get_for_model(solis)
|
|
AccessRight.objects.create(
|
|
codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=solis.pk
|
|
)
|
|
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = tests.utils.FakedResponse(content=CIVI_INDIVIDU, status_code=200)
|
|
url = tests.utils.generic_endpoint_url('solis', 'referential-item', slug=solis.slug)
|
|
resp = app.get(url + '/civi/individu/4373/', status=200)
|
|
assert requests_get.call_count == 1
|
|
assert requests_get.call_args[0][0].endswith('/solisapi/referentiels/civi/individu/4373/')
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['etatCivil']['nom'] == 'NOM'
|
|
|
|
requests_get.return_value = tests.utils.FakedResponse(
|
|
content='{"nada":0}', status_code=404, reason='Not found'
|
|
)
|
|
url = tests.utils.generic_endpoint_url('solis', 'referential-item', slug=solis.slug)
|
|
resp = app.get(url + '/civi/individu/424242/', status=200)
|
|
assert requests_get.call_args[0][0].endswith('/solisapi/referentiels/civi/individu/424242/')
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == "error status:404 'Not found', content:'{\"nada\":0}'"
|
|
assert resp.json['data'] == {'json_content': {'nada': 0}, 'status_code': 404}
|
|
|
|
requests_get.return_value = tests.utils.FakedResponse(content='["not a dict"]', status_code=200)
|
|
url = tests.utils.generic_endpoint_url('solis', 'referential-item', slug=solis.slug)
|
|
resp = app.get(url + '/civi/individu/0/', status=200)
|
|
assert requests_get.call_args[0][0].endswith('/solisapi/referentiels/civi/individu/0/')
|
|
assert resp.json['err'] == 1
|
|
assert 'not a dict' in resp.json['err_desc']
|
|
|
|
|
|
def test_unflat_dict():
|
|
assert unflat({'foo': 'bar', 'two_foo': 'one', 'two_bar': 'two'}) == {
|
|
'foo': 'bar',
|
|
'two': {'foo': 'one', 'bar': 'two'},
|
|
}
|
|
|
|
|
|
def test_solis_apa_integration(app, solis):
|
|
api = ApiUser.objects.create(username='all', keytype='', key='')
|
|
obj_type = ContentType.objects.get_for_model(solis)
|
|
AccessRight.objects.create(
|
|
codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=solis.pk
|
|
)
|
|
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
|
|
def integration_ok(*args, **kwargs):
|
|
return tests.utils.FakedResponse(content='', status_code=204)
|
|
|
|
requests_post.return_value = tests.utils.FakedResponse(content='', status_code=204)
|
|
# requests_post.side_effect = [utils.FakedResponse(content='', status_code=204)]
|
|
url = tests.utils.generic_endpoint_url('solis', 'apa-integration', slug=solis.slug)
|
|
|
|
demande = {
|
|
"beneficiaire_demande_aide": "APAD",
|
|
"beneficiaire_demande_dateDepot": "2018-02-09",
|
|
"beneficiaire_etatCivil_civilite": "M",
|
|
"beneficiaire_etatCivil_contact_courriel": "benef@yopmail.com",
|
|
"conjoint_nom": "Conjnom",
|
|
"conjoint_prenom": "Conjprenom",
|
|
}
|
|
|
|
resp = app.post_json(url, params=demande, status=200)
|
|
|
|
requests_post.assert_called_once()
|
|
assert requests_post.call_args[0][0].endswith('/solisapi/asg/apa/integrationDemandeApa')
|
|
assert requests_post.call_args[1]['json']['demandeApa']['beneficiaire']['demande']['aide'] == 'APAD'
|
|
assert requests_post.call_args[1]['json']['demandeApa']['conjoint']['nom'] == 'Conjnom'
|
|
assert requests_post.call_args[1]['json']['demandeApa'] == unflat(demande)
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data'] is None
|
|
|
|
# don't send "conjoint" dict to Solis
|
|
requests_post.reset_mock()
|
|
demande['del:conjoint'] = True
|
|
resp = app.post_json(url, params=demande, status=200)
|
|
requests_post.assert_called_once()
|
|
assert 'conjoint' not in requests_post.call_args[1]['json']['demandeApa']
|
|
assert resp.json['err'] == 0
|
|
|
|
# send "conjoint" dict to Solis
|
|
requests_post.reset_mock()
|
|
demande['del:conjoint'] = False
|
|
resp = app.post_json(url, params=demande, status=200)
|
|
requests_post.assert_called_once()
|
|
assert requests_post.call_args[1]['json']['demandeApa']['conjoint']['nom'] == 'Conjnom'
|
|
assert resp.json['err'] == 0
|
|
|
|
# add files
|
|
requests_post.reset_mock()
|
|
requests_post.side_effect = [
|
|
tests.utils.FakedResponse(content='{"id": "foo", "nbFichiersAcceptes": 3}', status_code=200),
|
|
tests.utils.FakedResponse(content='', status_code=204),
|
|
]
|
|
demande['file:etat_civil_001.pdf'] = {
|
|
'content': 'JVBERmZha2U=',
|
|
'content_type': 'application/pdf',
|
|
'filename': 'whatever.pdf',
|
|
}
|
|
demande['file:etat_civil_002.pdf'] = {
|
|
# jpeg, will be converted to PDF
|
|
'content': '/9j/2wBDAAMCAgICAgMCAgIDAwMDBAYEBAQEBAgGBgUGCQgKCgkICQkKDA8MCgsOCw'
|
|
'kJDRENDg8QEBEQCgwSExIQEw8QEBD/yQALCAABAAEBAREA/8wABgAQEAX/2gAIAQEAAD8A0s8g/9k=',
|
|
'content_type': 'image/jpeg',
|
|
'filename': 'image.jpg',
|
|
}
|
|
demande['file:etat_civil_003.pdf'] = {
|
|
# transparent png (RGBA), will be converted to RGB and then PDF
|
|
'content': 'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAACklEQVR4nGMAAQAABQ'
|
|
'ABDQottAAAAABJRU5ErkJggg==',
|
|
'content_type': 'image/png',
|
|
'filename': 'image.png',
|
|
}
|
|
demande['file:etat_civil_004.pdf'] = {
|
|
'content': 'Y29pbg==', # bad content, conversion will fail
|
|
'content_type': 'image/png',
|
|
'filename': 'image.png',
|
|
}
|
|
demande['file:etat_civil_005.pdf'] = {
|
|
'content': 'Y29pbg==',
|
|
'content_type': 'video/mp4', # not a image, cannot convert
|
|
'filename': 'video.mp4',
|
|
}
|
|
demande['file:etat_civil_006.pdf'] = {
|
|
'content_type': 'video/mp4', # no content, cannot convert
|
|
}
|
|
demande['file:etat_civil_007.pdf'] = None
|
|
resp = app.post_json(url, params=demande, status=200)
|
|
assert requests_post.call_count == 2 # post files + demandeApa
|
|
sent_files = requests_post.call_args_list[0][1]['files']
|
|
assert len(sent_files) == 3
|
|
for file_ in sent_files:
|
|
assert file_[1][1].startswith(b'%PDF')
|
|
# file entries are removed from demandeApa JSON dict
|
|
assert 'file:etat_civil_001.pdf' not in requests_post.call_args[1]['json']['demandeApa']
|
|
assert 'file:etat_civil_002.pdf' not in requests_post.call_args[1]['json']['demandeApa']
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data'] is None
|
|
assert resp.json['files_sent'] == {'id': 'foo', 'nbFichiersAcceptes': 3}
|
|
assert set(resp.json['files_failed_pdf_conversion']) == {
|
|
'etat_civil_004.pdf',
|
|
'etat_civil_005.pdf',
|
|
'etat_civil_006.pdf',
|
|
}
|
|
|
|
# invalid inputs
|
|
requests_post.reset_mock()
|
|
resp = app.post_json(url, params=['not', 'a', 'dict'], status=400)
|
|
requests_post.assert_not_called()
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'payload is not a JSON dict'
|
|
resp = app.post(url, params='coin', status=400)
|
|
requests_post.assert_not_called()
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'payload is not a JSON object'
|
|
|
|
# bad file
|
|
requests_post.reset_mock()
|
|
requests_post.side_effect = [
|
|
tests.utils.FakedResponse(content='{"id": "foo", "nbFichiersAcceptes": 0}', status_code=200),
|
|
tests.utils.FakedResponse(content='', status_code=204),
|
|
]
|
|
resp = app.post_json(url, params=demande, status=200)
|
|
requests_post.assert_called_once() # don't try to post request
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'fail to send all files'
|
|
assert resp.json['data'] == {'id': 'foo', 'nbFichiersAcceptes': 0}
|
|
|
|
# error on sending file
|
|
requests_post.reset_mock()
|
|
requests_post.side_effect = [
|
|
tests.utils.FakedResponse(content='{"error": 1}', status_code=500),
|
|
tests.utils.FakedResponse(content='', status_code=204),
|
|
]
|
|
resp = app.post_json(url, params=demande, status=200)
|
|
requests_post.assert_called_once() # don't try to post request
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'].startswith('error status:500')
|
|
|
|
|
|
def test_solis_rsa_link_infos_unlink(app, solis):
|
|
# full opened access
|
|
api = ApiUser.objects.create(username='all', keytype='', key='')
|
|
obj_type = ContentType.objects.get_for_model(solis)
|
|
AccessRight.objects.create(
|
|
codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=solis.pk
|
|
)
|
|
|
|
# link
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post: # get solis token
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get: # get solis informations
|
|
endpoint = tests.utils.generic_endpoint_url('solis', 'rsa-link', slug=solis.slug)
|
|
for params in (None, '', []):
|
|
resp = app.post_json(endpoint, params=params, status=200)
|
|
assert requests_post.call_count == 0
|
|
assert resp.json['err'] == 1
|
|
assert 'payload is not a JSON dict' in resp.json['err_desc']
|
|
|
|
for params in (
|
|
{},
|
|
{'user_id': 'x'},
|
|
{'code': 'x'},
|
|
{'foo': 'bar'},
|
|
{'name_id': ''},
|
|
{'user_id': '', 'code': ''},
|
|
):
|
|
resp = app.post_json(endpoint, params=params, status=200)
|
|
assert requests_post.call_count == 0
|
|
assert resp.json['err'] == 1
|
|
assert 'missing name_id' in resp.json['err_desc']
|
|
params['name_id'] = 'xx'
|
|
resp = app.post_json(endpoint, params=params, status=200)
|
|
assert requests_post.call_count == 0
|
|
assert resp.json['err'] == 1
|
|
assert 'missing user_id/code credentials' in resp.json['err_desc']
|
|
|
|
requests_post.return_value = tests.utils.FakedResponse(content=RSATOKEN_403, status_code=403)
|
|
resp = app.post_json(
|
|
endpoint, params={'user_id': 'x', 'code': 'x', 'name_id': NAMEID}, status=200
|
|
)
|
|
assert requests_post.call_count == 1
|
|
assert requests_get.call_count == 0
|
|
assert resp.json['err'] == 1
|
|
assert 'non autor' in resp.json['err_desc']
|
|
|
|
assert SolisRSALink.objects.count() == 0
|
|
|
|
requests_post.return_value = tests.utils.FakedResponse(content=RSATOKEN, status_code=200)
|
|
requests_get.return_value = tests.utils.FakedResponse(content=CIVI_INDIVIDU, status_code=200)
|
|
resp = app.post_json(
|
|
endpoint, params={'name_id': NAMEID, 'user_id': '4273', 'code': 'foo'}, status=200
|
|
)
|
|
assert requests_post.call_count == 2
|
|
assert requests_get.call_count == 1
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['user_id'] == '4273'
|
|
assert resp.json['data']['created']
|
|
assert not resp.json['data']['updated']
|
|
assert SolisRSALink.objects.count() == 1
|
|
assert SolisRSALink.objects.first().name_id == NAMEID
|
|
assert SolisRSALink.objects.first().user_id == '4273'
|
|
assert SolisRSALink.objects.first().code == 'foo'
|
|
assert SolisRSALink.objects.first().text == 'MME Prenom NOM'
|
|
assert SolisRSALink.objects.first().dob is None
|
|
|
|
# change code, add dob
|
|
resp = app.post_json(
|
|
endpoint,
|
|
params={'name_id': NAMEID, 'user_id': '4273', 'code': 'bar', 'dob': '01/01/1950'},
|
|
status=200,
|
|
)
|
|
assert requests_post.call_count == 3
|
|
assert requests_get.call_count == 2
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['user_id'] == '4273'
|
|
assert not resp.json['data']['created']
|
|
assert resp.json['data']['updated']
|
|
assert SolisRSALink.objects.count() == 1
|
|
assert SolisRSALink.objects.first().name_id == NAMEID
|
|
assert SolisRSALink.objects.first().user_id == '4273'
|
|
assert SolisRSALink.objects.first().code == 'bar'
|
|
assert SolisRSALink.objects.first().text == 'MME Prenom NOM'
|
|
assert SolisRSALink.objects.first().dob == '01/01/1950'
|
|
|
|
# second link
|
|
requests_post.reset_mock()
|
|
requests_get.reset_mock()
|
|
resp = app.post_json(
|
|
endpoint,
|
|
params={'name_id': NAMEID, 'user_id': '4242', 'code': 'bar', 'dob': '10/10/1960'},
|
|
status=200,
|
|
)
|
|
assert requests_post.call_count == 1
|
|
assert requests_get.call_count == 1
|
|
assert requests_post.call_args[1]['json'] == {
|
|
'codeConfidentiel': 'bar',
|
|
'dateNaissance': '10/10/1960',
|
|
'indexIndividu': '4242',
|
|
}
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['user_id'] == '4242'
|
|
assert resp.json['data']['created']
|
|
assert not resp.json['data']['updated']
|
|
assert SolisRSALink.objects.count() == 2
|
|
|
|
# verify recorded names after link
|
|
assert [x['text'] for x in SolisRSALink.objects.values('text')] == ['MME Prenom NOM', 'MME Prenom NOM']
|
|
|
|
endpoint = tests.utils.generic_endpoint_url('solis', 'rsa-links', slug=solis.slug)
|
|
resp = app.get(endpoint, status=400) # missing name_id
|
|
assert resp.json['err'] == 1
|
|
endpoint += '?name_id=%s' % NAMEID
|
|
resp = app.get(endpoint, status=200)
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 2
|
|
assert resp.json['data'][0]['text'] == resp.json['data'][1]['text'] == 'MME Prenom NOM'
|
|
|
|
# get base informations from a linked user
|
|
changed_name = CIVI_INDIVIDU.replace('Prenom', 'Postnom')
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = tests.utils.FakedResponse(content=RSATOKEN, status_code=200)
|
|
requests_get.return_value = tests.utils.FakedResponse(content=changed_name, status_code=200)
|
|
endpoint = tests.utils.generic_endpoint_url('solis', 'rsa-user-info', slug=solis.slug)
|
|
endpoint += '?name_id=%s&user_id=4242' % NAMEID
|
|
resp = app.get(endpoint, status=200)
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['etatCivil']['prenom'] == 'Postnom'
|
|
# user "text" updated in link:
|
|
assert SolisRSALink.objects.get(name_id=NAMEID, user_id='4242').text == 'MME Postnom NOM'
|
|
|
|
# get referential for a linked user
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = tests.utils.FakedResponse(content=RSATOKEN, status_code=200)
|
|
|
|
endpoint_base = tests.utils.generic_endpoint_url('solis', 'rsa-user-info', slug=solis.slug)
|
|
resp = app.get(endpoint_base, status=400) # missing name_id
|
|
assert resp.json['err'] == 1
|
|
|
|
endpoint = endpoint_base + '?name_id=%s&user_id=4242&information=allocataires' % NAMEID
|
|
requests_get.return_value = tests.utils.FakedResponse(content=RSAALLOCATAIRES, status_code=200)
|
|
resp = app.get(endpoint, status=200)
|
|
assert requests_post.call_count == 1 # get a token
|
|
assert requests_get.call_count == 1 # get informations
|
|
assert '/referentiels/grsa/allocataires/search/' in requests_get.call_args[0][0]
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']
|
|
|
|
# solis api crash
|
|
requests_get.return_value = tests.utils.FakedResponse(content='boum', status_code=500)
|
|
resp = app.get(endpoint, status=200)
|
|
assert requests_post.call_count == 2 # get a token
|
|
assert requests_get.call_count == 2 # get informations
|
|
assert '/referentiels/grsa/allocataires/search/' in requests_get.call_args[0][0]
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'].startswith('error status:500')
|
|
assert resp.json['data'] == {'json_content': None, 'status_code': 500}
|
|
|
|
requests_get.return_value = tests.utils.FakedResponse(
|
|
content='{"error":"foobar"}', status_code=500
|
|
)
|
|
resp = app.get(endpoint, status=200)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'].startswith('error status:500')
|
|
assert resp.json['data'] == {'json_content': {'error': 'foobar'}, 'status_code': 500}
|
|
|
|
# unknown name_id or user_id
|
|
for qs in ('name_id=%s&user_id=53' % NAMEID, 'name_id=unlinked&user_id=4242'):
|
|
endpoint = endpoint_base + '?information=allocataires&' + qs
|
|
resp = app.get(endpoint, status=200)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'unknown link'
|
|
assert resp.json['data'] is None
|
|
|
|
# get actions, with filters
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = tests.utils.FakedResponse(content=RSATOKEN, status_code=200)
|
|
endpoint_base = tests.utils.generic_endpoint_url('solis', 'rsa-user-info', slug=solis.slug)
|
|
endpoint = endpoint_base + '?name_id=%s&user_id=4242&information=actions' % NAMEID
|
|
requests_get.return_value = tests.utils.FakedResponse(content=RSA_ACTIONS, status_code=200)
|
|
resp = app.get(endpoint, status=200)
|
|
assert requests_post.call_count == 1 # get a token
|
|
assert requests_get.call_count == 1 # get actions
|
|
assert '/referentiels/grsa/actions/search/' in requests_get.call_args[0][0]
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 4 # all actions
|
|
|
|
filtered = endpoint + '&filters=idStructure=42'
|
|
resp = app.get(filtered, status=200)
|
|
assert len(resp.json['data']) == 2
|
|
assert resp.json['data'][0]['idStructure'] == 42
|
|
|
|
filtered = endpoint + '&filters=idStructure!=42'
|
|
resp = app.get(filtered, status=200)
|
|
assert len(resp.json['data']) == 2
|
|
assert resp.json['data'][0]['idStructure'] == 43
|
|
assert resp.json['data'][1]['idStructure'] == 44
|
|
|
|
filtered = endpoint + '&filters=idStructure!=42,idStructure!=43,prescriptionPlacement=Placement'
|
|
resp = app.get(filtered, status=200)
|
|
assert len(resp.json['data']) == 1
|
|
assert resp.json['data'][0]['idStructure'] == 44
|
|
|
|
filtered = endpoint + '&filters='
|
|
resp = app.get(filtered, status=200)
|
|
assert len(resp.json['data']) == 4 # empty filters: all actions
|
|
|
|
requests_get.reset_mock()
|
|
requests_get.side_effect = [
|
|
tests.utils.FakedResponse(status_code=200, content=RSA_ACTIONS), # base info
|
|
tests.utils.FakedResponse(status_code=200, content='{"nom":"Structure1"}'), # link 1
|
|
tests.utils.FakedResponse(status_code=200, content='{"nom":"Structure2"}'), # link 2
|
|
]
|
|
filtered = endpoint + '&links'
|
|
resp = app.get(filtered, status=200)
|
|
assert len(resp.json['data']) == 4
|
|
assert requests_get.call_count == 3 # actions + two _links
|
|
assert resp.json['data'][0]['rsa_links']['structure']['content'] == {'nom': 'Structure1'}
|
|
assert resp.json['data'][2]['rsa_links']['structure']['content'] == {'nom': 'Structure2'}
|
|
assert 'rsa_links' not in resp.json['data'][1]
|
|
assert 'rsa_links' not in resp.json['data'][3]
|
|
|
|
# get referential for a linked user with links
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = tests.utils.FakedResponse(content=RSATOKEN, status_code=200)
|
|
endpoint_base = tests.utils.generic_endpoint_url('solis', 'rsa-user-info', slug=solis.slug)
|
|
|
|
requests_get.side_effect = [
|
|
tests.utils.FakedResponse(status_code=200, content=RSAALLOCATAIRES), # base info
|
|
tests.utils.FakedResponse(status_code=200, content=CIVI_INDIVIDU), # link 1
|
|
tests.utils.FakedResponse(status_code=200, content=CIVI_INDIVIDU),
|
|
] # link 2
|
|
endpoint = endpoint_base + '?name_id=%s&user_id=4242&information=allocataires&links' % NAMEID
|
|
resp = app.get(endpoint, status=200)
|
|
assert requests_post.call_count == 1 # get a token
|
|
assert requests_get.call_count == 3 # get informations + two links
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['rsa_links']['etatCivil']['content']['index'] == 4273
|
|
assert resp.json['data']['rsa_links']['conjoint']['content']['index'] == 4273
|
|
|
|
# complex links
|
|
requests_post.reset_mock()
|
|
requests_get.reset_mock()
|
|
requests_get.side_effect = [
|
|
tests.utils.FakedResponse(status_code=200, content=RSA_3LINKS), # base info
|
|
tests.utils.FakedResponse(status_code=200, content=CIVI_INDIVIDU), # link 1
|
|
tests.utils.FakedResponse(status_code=200, content=CIVI_INDIVIDU), # link 2.1
|
|
tests.utils.FakedResponse(status_code=200, content=CIVI_INDIVIDU),
|
|
] # link 2.2
|
|
endpoint = endpoint_base + '?name_id=%s&user_id=4242&information=evaluations&links' % NAMEID
|
|
resp = app.get(endpoint, status=200)
|
|
assert requests_post.call_count == 1 # get a token
|
|
assert requests_get.call_count == 4 # get informations + 2+1 links
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['rsa_links']['etatCivil']['content']['index'] == 4273
|
|
assert resp.json['data']['rsa_links']['refOrientStructAcc'][0]['content']['index'] == 4273
|
|
assert resp.json['data']['rsa_links']['refOrientStructAcc'][1]['content']['index'] == 4273
|
|
|
|
# get only conjoint
|
|
requests_post.reset_mock()
|
|
requests_get.reset_mock()
|
|
requests_get.side_effect = [
|
|
tests.utils.FakedResponse(status_code=200, content=RSAALLOCATAIRES), # base info
|
|
tests.utils.FakedResponse(status_code=200, content=CIVI_INDIVIDU),
|
|
] # link 1
|
|
endpoint = (
|
|
endpoint_base
|
|
+ '?name_id=%s&user_id=4242&information=allocataires&links=conjoint, ,xx,' % NAMEID
|
|
)
|
|
resp = app.get(endpoint, status=200)
|
|
assert requests_post.call_count == 1 # get a token
|
|
assert requests_get.call_count == 2 # get informations + conjoint
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['rsa_links']['conjoint']['content']['index'] == 4273
|
|
assert 'content' not in resp.json['data']['rsa_links']['etatCivil']
|
|
|
|
# error on link retreival
|
|
requests_post.reset_mock()
|
|
requests_get.reset_mock()
|
|
requests_get.side_effect = [
|
|
tests.utils.FakedResponse(status_code=200, content=RSAALLOCATAIRES), # base info
|
|
tests.utils.FakedResponse(status_code=404, content='{"foo": "bar"}'), # link 1
|
|
tests.utils.FakedResponse(status_code=500, content='boom'),
|
|
] # link 2
|
|
endpoint = endpoint_base + '?name_id=%s&user_id=4242&information=allocataires&links' % NAMEID
|
|
resp = app.get(endpoint, status=200)
|
|
assert requests_post.call_count == 1 # get a token
|
|
assert requests_get.call_count == 3 # get informations + two links
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['rsa_links']['etatCivil']['content']['err'] == 1
|
|
assert resp.json['data']['rsa_links']['conjoint']['content']['err'] == 1
|
|
|
|
# bad links, do nothing
|
|
for content in (
|
|
RSAALLOCATAIRES.replace('solis.example.net', 'solis.example.org'),
|
|
RSAALLOCATAIRES.replace('href', 'xxxx'),
|
|
):
|
|
requests_post.reset_mock()
|
|
requests_get.reset_mock()
|
|
requests_get.side_effect = [tests.utils.FakedResponse(status_code=200, content=content)]
|
|
endpoint = endpoint_base + '?name_id=%s&user_id=4242&information=allocataires&links' % NAMEID
|
|
resp = app.get(endpoint, status=200)
|
|
assert requests_post.call_count == 1 # get a token
|
|
assert requests_get.call_count == 1 # get informations only, not links
|
|
assert resp.json['err'] == 0
|
|
assert 'content' not in resp.json['data']['rsa_links']['conjoint']
|
|
assert 'content' not in resp.json['data']['rsa_links']['etatCivil']
|
|
|
|
# unlink
|
|
endpoint = tests.utils.generic_endpoint_url('solis', 'rsa-unlink', slug=solis.slug)
|
|
for bad_params in ({}, {'user_id': '4273'}, {'name_id': NAMEID}):
|
|
resp = app.post_json(endpoint, params=bad_params, status=200)
|
|
assert resp.json['err'] == 1
|
|
|
|
resp = app.post_json(endpoint, params={'user_id': 'xxx', 'name_id': 'xxx'}, status=200)
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['deleted']
|
|
assert resp.json['data']['user_id'] == 'xxx'
|
|
assert SolisRSALink.objects.count() == 2
|
|
|
|
resp = app.post_json(endpoint, params={'user_id': '4242', 'name_id': NAMEID}, status=200)
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['deleted']
|
|
assert resp.json['data']['user_id'] == '4242'
|
|
assert SolisRSALink.objects.count() == 1
|
|
|
|
# unlink again, no trouble
|
|
resp = app.post_json(endpoint, params={'user_id': '4242', 'name_id': NAMEID}, status=200)
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['deleted']
|
|
assert resp.json['data']['user_id'] == '4242'
|
|
assert SolisRSALink.objects.count() == 1
|
|
|
|
# can not get informations from unlinked user
|
|
endpoint = tests.utils.generic_endpoint_url('solis', 'rsa-user-info', slug=solis.slug)
|
|
endpoint += '?name_id=%s&user_id=4242' % NAMEID
|
|
resp = app.get(endpoint, status=200)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'unknown link'
|
|
assert resp.json['data'] is None
|