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