0001-api_entreprise-add-initial-connector-30010.patch

Serghei Mihai (congés, retour 15/05), 10 avril 2019 00:49

Voir les différences: en ligne côte à côte

Subject: [PATCH] api_entreprise: add initial connector (#30010)

 passerelle/apps/api_entreprise/__init__.py    |   0
 .../api_entreprise/migrations/0001_initial.py |  32 ++
 .../api_entreprise/migrations/__init__.py     |   0
 passerelle/apps/api_entreprise/models.py      | 305 +++++++++++++
 passerelle/settings.py                        |   1 +
 passerelle/static/css/style.css               |   4 +
 tests/test_api_entreprise.py                  | 423 ++++++++++++++++++
 7 files changed, 765 insertions(+)
 create mode 100644 passerelle/apps/api_entreprise/__init__.py
 create mode 100644 passerelle/apps/api_entreprise/migrations/0001_initial.py
 create mode 100644 passerelle/apps/api_entreprise/migrations/__init__.py
 create mode 100644 passerelle/apps/api_entreprise/models.py
 create mode 100644 tests/test_api_entreprise.py

     # -*- coding: utf-8 -*-
     # Generated by Django 1.11.20 on 2019-03-15 09:38
     from __future__ import unicode_literals
     from django.db import migrations, models
     class Migration(migrations.Migration):
         initial = True
         dependencies = [
             ('base', '0012_job'),
+        ]
         operations = [
             migrations.CreateModel(
                 name='APIEntreprise',
                 fields=[
                     ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                     ('title', models.CharField(max_length=50, verbose_name='Title')),
                     ('description', models.TextField(verbose_name='Description')),
                     ('slug', models.SlugField(unique=True, verbose_name='Identifier')),
                     ('url', models.URLField(default=b'https://entreprise.api.gouv.fr/v2/', max_length=256, verbose_name='API URL')),
                     ('token', models.CharField(max_length=1024, verbose_name='API token')),
                     ('users', models.ManyToManyField(blank=True, to='base.ApiUser')),
                 ],
                 options={
                     'verbose_name': 'API Entreprise',
                 },
             ),
+        ]

     # passerelle - uniform access to multiple data sources and services
     # Copyright (C) 2019  Entr'ouvert
+    #
     # This program is free software: you can redistribute it and/or modify it
     # under the terms of the GNU Affero General Public License as published
     # by the Free Software Foundation, either version 3 of the License, or
     # (at your option) any later version.
+    #
     # This program is distributed in the hope that it will be useful,
     # but WITHOUT ANY WARRANTY; without even the implied warranty of
     # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
     # GNU Affero General Public License for more details.
+    #
     # You should have received a copy of the GNU Affero General Public License
     # along with this program.  If not, see <http://www.gnu.org/licenses/>.
     '''Gateway to API-Entreprise web-service from SGMAP:
        https://entreprise.api.gouv.fr
     '''
     from six.moves.urllib_parse import urljoin
     import requests
     from django.db import models
     from django.utils.translation import ugettext_lazy as _
     from django.utils.timezone import datetime, make_aware, timedelta
     from django.http import HttpResponse, Http404, HttpResponseBadRequest
     from django.core import signing
     from django.core.urlresolvers import reverse
     from passerelle.base.models import BaseResource
     from passerelle.utils.api import endpoint
     from passerelle.utils.jsonresponse import APIError
     from passerelle.views import WrongParameter
     DOCUMENT_SIGNATURE_MAX_AGE = timedelta(days=7)
     def normalize_dates(data):
         for key in data:
             if isinstance(data[key], dict):
                 normalize_dates(data[key])
             if isinstance(data[key], list):
                 for item in data[key]:
                     normalize_dates(item)
             if key.startswith('date_'):
                 # try to parse isoformatted dates
                 try:
                     data[key] = datetime.fromtimestamp(int(data[key])).date()
                 except (ValueError, TypeError):
                     pass
             if key.endswith('timestamp'):
                 # timestamps can be integers or strings
                 try:
                     data[key] = make_aware(datetime.fromtimestamp(int(data[key])))
                 except (ValueError, TypeError):
                     pass
     class APIEntreprise(BaseResource):
         url = models.URLField(_('API URL'), max_length=256, default='https://entreprise.api.gouv.fr/v2/')
         token = models.CharField(max_length=1024, verbose_name=_('API token'))
         category = _('Business Process Connectors')
         class Meta:
             verbose_name = _('API Entreprise')
         def get(self, path, **kwargs):
             params = {'token': self.token}
             for param in ('context', 'object', 'recipient'):
                 if not kwargs.get(param):
                     raise WrongParameter([param], [])
                 params[param] = kwargs[param]
             url = urljoin(self.url, path)
             try:
                 response = self.requests.get(url, data=params)
             except requests.RequestException as e:
                 raise APIError(u'API-entreprise connection error: %s' %
                                response.status_code,
                                data={'error': unicode(e)})
             try:
                 data = response.json()
             except ValueError as e:
                 content = repr(response.content[:1000])
                 raise APIError(
                     u'API-entreprise returned non-JSON content with status %s: %s' %
                     (response.status_code, content),
                     data={'status_code': response.status_code,
                           'exception': unicode(e),
                           'content': content,
                     })
             if response.status_code != 200:
                 if data.get('error') == 'not_found':
                     return {
                         'err': 1,
                         'err_desc': data.get('message', 'not-found'),
+                    }
                 raise APIError(
                     u'API-entreprise returned a non 200 status %s: %s' %
                     (response.status_code, data),
                     data={
                         'status_code': response.status_code,
                         'content': data,
                     })
             normalize_dates(data)
             return {
                 'err': 0,
                 'data': data,
+            }
         @endpoint(perm='can_access',
                   pattern='(?P<institution_id>\w+)/$',
                   example_pattern='{institution_id}/',
                   description=_('Get association\'s documents'),
                   parameters={
                       'institution_id': {
                           'description': _('association id'),
                           'example_value': '44317013900036',
                       },
                       'object': {
                           'Description': _('request object'),
                           'example_value': 'MSP'
                       },
                       'context': {
                           'description': _('request context'),
                           'example_value': '42'
                       },
                       'recipient': {
                           'description': _('request recipient: usually customer number'),
                           'example_value': '44317013900036'
+                      }
+                  }
+        )
         def documents_associations(self, request, institution_id, **kwargs):
             data = []
             resp = self.get('documents_associations/%s/' % institution_id, **kwargs)
             for item in resp['data'].get('documents', []):
                 # ignore documents with no type
                 if not item.get('type'):
                     continue
                 signature_elements = {'url': item['url'],
                                       'context': kwargs['context'],
                                       'object': kwargs['object'],
                                       'recipient': kwargs['recipient']}
                 signature = signing.dumps(signature_elements)
                 document_url = request.build_absolute_uri(reverse('generic-endpoint', kwargs={'connector': self.get_connector_slug(),
                                             'slug': self.slug,
                                             'endpoint': 'document',
                                             'rest': '%s/%s/' % (institution_id, signature)}))
                 data.append({'id': item['timestamp'],
                              'text': item['type'],
                              'type': item['type'],
                              'url': document_url})
             # sort data by date
             data.sort(key=lambda i: i['id'])
             return {'err': 0, 'data': data}
         @endpoint(pattern='(?P<institution_id>\w+)/(?P<document_id>[\:\w-]+)/$',
                   example_pattern='{institution_id}/{document_id}/',
                   description=_('Get institution\'s document'),
                   parameters={
                       'institution_id': {
                           'description': _('institution id'),
                           'example_value': '44317013900036',
                       },
                       'document_id': {
                           'description': _('document id'),
                           'example_value': 'A1500660325',
                       },
                       'object': {
                           'Description': _('request object'),
                           'example_value': 'MSP'
                       },
                       'context': {
                           'description': _('request context'),
                           'example_value': '42'
                       },
                       'recipient': {
                           'description': _('request recipient: usually customer number'),
                           'example_value': '44317013900036'
+                      }
+                  }
+        )
         def document(self, request, institution_id, document_id, **kwargs):
             try:
                 params = signing.loads(document_id, max_age=DOCUMENT_SIGNATURE_MAX_AGE)
             except signing.BadSignature:
                 raise Http404('document not found')
             response = self.requests.get(params['url'])
             if response.ok:
                 return HttpResponse(response, content_type='application/pdf')
             raise Http404('document not found')
         @endpoint(perm='can_access',
                   pattern='(?P<institution_id>\w+)/$',
                   example_pattern='{institution_id}/',
                   description=_('Get institution\'s data from Infogreffe'),
                   parameters={
                       'institution_id': {
                           'description': _('institution id'),
                           'example_value': '44317013900036',
                       },
                       'object': {
                           'Description': _('request object'),
                           'example_value': 'MSP'
                       },
                       'context': {
                           'description': _('request context'),
                           'example_value': '42'
                       },
                       'recipient': {
                           'description': _('request recipient: usually customer number'),
                           'example_value': '44317013900036'
+                      }
+                  }
+        )
         def extraits_rcs(self, request, institution_id, **kwargs):
             return self.get('extraits_rcs_infogreffe/%s/' % institution_id, **kwargs)
         @endpoint(perm='can_access',
                   pattern='(?P<institution_id>\w+)/$',
                   example_pattern='{institution_id}/',
                   description=_('Get institution\'s related informations'),
                   parameters={
                       'institution_id': {
                           'description': _('institution id'),
                           'example_value': '44317013900036',
                       },
                       'object': {
                           'Description': _('request object'),
                           'example_value': 'MSP'
                       },
                       'context': {
                           'description': _('request context'),
                           'example_value': '42'
                       },
                       'recipient': {
                           'description': _('request recipient: usually customer number'),
                           'example_value': '44317013900036'
+                      }
+                  }
+        )
         def associations(self, request, institution_id, **kwargs):
             return self.get('associations/%s/' % institution_id, **kwargs)
         @endpoint(perm='can_access',
                   pattern='(?P<institution_id>\w+)/$',
                   example_pattern='{institution_id}/',
                   description=_('Get institution\'s related informations'),
                   parameters={
                       'institution_id': {
                           'description': _('association id'),
                           'example_value': '44317013900036',
                       },
                       'object': {
                           'description': _('request object'),
                           'example_value': 'MSP'
                       },
                       'context': {
                           'description': _('request context'),
                           'example_value': '42'
                       },
                       'recipient': {
                           'description': _('request recipient: usually customer number'),
                           'example_value': '44317013900036'
+                      }
+                  }
+        )
         def entreprises(self, request, institution_id, **kwargs):
             return self.get('entreprises/%s/' % institution_id, **kwargs)
         @endpoint(perm='can_access',
                   methods=['get'],
                   pattern='(?P<institution_id>\w+)/$',
                   example_pattern='{institution_id}/', #&
                   description_get=_('Get institution\'s related informations'),
                   parameters={
                       'institution_id': {
                           'description': _('institution id'),
                           'example_value': '44317013900036',
                       },
                       'object': {
                           'description': _('request object'),
                           'example_value': 'MSP'
                       },
                       'context': {
                           'description': _('request context'),
                           'example_value': '42'
                       },
                       'recipient': {
                           'description': _('request recipient: usually customer number'),
                           'example_value': '44317013900036'
+                      }
+                  }
+        )
         def etablissements(self, request, institution_id, **kwargs):
             return self.get('etablissements/%s/' % institution_id, **kwargs)

passerelle/settings.py
145	145	'passerelle.apps.solis',
146	146	'passerelle.apps.arpege_ecp',
147	147	'passerelle.apps.vivaticket',
	148	'passerelle.apps.api_entreprise',
148	149	# backoffice templates and static
149	150	'gadjo',
150	151	)

passerelle/static/css/style.css
168	168	content: "\f2c3"; /* id-card-o */
169	169	}
170	170
	171	li.connector.apientreprise a::before {
	172	content: "\f1ad"; /* fa-building-o */
	173	}
	174
171	175	li.connector.dpark a::before {
172	176	content: "\f1b9"; /* car */
173	177	}

     # -*- coding: utf-8 -*-
     # tests/test_api_entreprise.py
     # Copyright (C) 2019  Entr'ouvert
+    #
     # This program is free software: you can redistribute it and/or modify it
     # under the terms of the GNU Affero General Public License as published
     # by the Free Software Foundation, either version 3 of the License, or
     # (at your option) any later version.
+    #
     # This program is distributed in the hope that it will be useful,
     # but WITHOUT ANY WARRANTY; without even the implied warranty of
     # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
     # GNU Affero General Public License for more details.
+    #
     # You should have received a copy of the GNU Affero General Public License
     # along with this program.  If not, see <http://www.gnu.org/licenses/>.
     import pytest
     import mock
     from httmock import urlmatch, HTTMock, response
     from django.core.urlresolvers import reverse
     from django.utils import timezone
     from passerelle.apps.api_entreprise.models import APIEntreprise
     from utils import make_resource, endpoint_get, FakedResponse
     ETABLISSEMENTS_RESPONSE = {
       "etablissement": {
         "siege_social": True,
         "siret": "41816609600051",
         "naf": "6202A",
         "libelle_naf": "Conseil en systèmes et logiciels informatiques",
         "date_mise_a_jour": 1449183600,
         "tranche_effectif_salarie_etablissement": {
           "de": 200,
           "a": 249,
           "code": "31",
           "date_reference": "2014",
           "intitule": "200 à 249 salariés"
         },
         "date_creation_etablissement": 1108594800,
         "region_implantation": {
           "code": "11",
           "value": "Île-de-France"
         },
         "commune_implantation": {
           "code": "75108",
           "value": "PARIS 8"
         },
         "adresse": {
           "l1": "OCTO TECHNOLOGY",
           "l4": "50 AVENUE DES CHAMPS ELYSEES",
           "l6": "75008 PARIS",
           "l7": "FRANCE",
           "numero_voie": "50",
           "type_voie": "AV",
           "nom_voie": "DES CHAMPS ELYSEES",
           "code_postal": "75008",
           "localite": "PARIS 8",
           "code_insee_localite": "75108",
         },
         "etat_administratif": {
           "value": "F",
           "date_fermeture": 1315173600
+        }
       },
       "gateway_error": False
+    }
     ENTREPRISES_RESPONSE = {
       "entreprise": {
         "siren": "418166096",
         "capital_social": 459356,
         "numero_tva_intracommunautaire": "FR16418166096",
         "forme_juridique": "SA à directoire (s.a.i.)",
         "forme_juridique_code": "5699",
         "nom_commercial": "OCTO-TECHNOLOGY",
         "procedure_collective": False,
         "naf_entreprise": "6202A",
         "libelle_naf_entreprise": "Conseil en systèmes et logiciels informatiques",
         "raison_sociale": "OCTO-TECHNOLOGY",
         "siret_siege_social": "41816609600051",
         "code_effectif_entreprise": "31",
         "date_creation": 891381600,
         "categorie_entreprise": "PME",
         "tranche_effectif_salarie_entreprise": {
           "de": 200,
           "a": 249,
           "code": "31",
           "date_reference": "2014",
           "intitule": "200 à 249 salariés"
         },
         "mandataires_sociaux": [{
           "nom": "HISQUIN",
           "prenom": "FRANCOIS",
           "fonction": "PRESIDENT DU DIRECTOIRE",
           "dirigeant": True,
           "date_naissance": "1965-01-27",
           "raison_sociale": "",
           "identifiant": "",
           "type": "PP"
         }, {
             "nom": "",
             "prenom": "",
             "fonction": "COMMISSAIRE AUX COMPTES SUPPLEANT",
             "dirigeant": True,
             "date_naissance": "",
             "date_naissance_timestamp": 0,
             "raison_sociale": "BCRH & ASSOCIES - SOCIETE A RESPONSABILITE LIMITEE A ASSOCIE UNIQUE",
             "identifiant": "490092574",
             "type": "PM"
+          }
         ],
         "etat_administratif": {
           "value": "C", # A (actif) ou C (cessé)
           "date_cessation": 1315173600 # null quand actif (A), un timestamp (un entier) quand cessé (C )
+        }
       },
       "etablissement_siege": {
         "siege_social": True,
         "siret": "41816609600051",
         "naf": "6202A",
         "libelle_naf": "Conseil en systèmes et logiciels informatiques",
         "date_mise_a_jour": 1449183600,
         "tranche_effectif_salarie_etablissement": {
           "de": 200,
           "a": 249,
           "code": "31",
           "date_reference": "2014",
           "intitule": "200 à 249 salariés"
         },
         "date_creation_etablissement": 1108594800,
         "region_implantation": {
           "code": "11",
           "value": "Île-de-France"
         },
         "commune_implantation": {
           "code": "75108",
           "value": "PARIS 8"
         },
         "adresse": {
           "l1": "OCTO TECHNOLOGY",
           "l4": "50 AVENUE DES CHAMPS ELYSEES",
           "l6": "75008 PARIS",
           "l7": "FRANCE",
           "numero_voie": "50",
           "type_voie": "AV",
           "nom_voie": "DES CHAMPS ELYSEES",
           "code_postal": "75008",
           "localite": "PARIS 8",
           "code_insee_localite": "75108",
         },
         "etat_administratif": {
             "value": "F",
             "date_fermeture": 1315173600
+        }
       },
       "gateway_error": False
+    }
     EXTRAITS_RCS_RESPONSE = {
       "siren": "418166096",
       "date_immatriculation": "1998-03-27",
       "date_immatriculation_timestamp": 890953200,
       "date_extrait": "21 AVRIL 2017",
       "observations": [
+        {
           "date": "2000-02-23",
           "date_timestamp": 951260400,
           "numero": "12197",
           "libelle": " LA SOCIETE NE CONSERVE AUCUNE ACTIVITE A SON ANCIEN SIEGE "
+        }
+      ]
+    }
     ASSOCIATIONS_RESPONSE = {
       "association" : {
         "id": "W751135389",
         "titre": "ALLIANCE DU COEUR: UNION NATIONALE DES FEDERATIONS ET ASSOCIATIONS DE MALADES CARDIOVASCULAIRES",
         "objet": "information, soutien, solidarité et accompagnement psycho médico social des personnes malades cardiovasculaires et de leurs proches...",
         "siret": "42135938100025",
         "siret_siege_social": "42135938100033",
         "date_creation": "1993-02-11",
         "date_declaration": "2013-06-28",
         "date_publication": "1993-03-03",
         "adresse_siege": {
           "numero_voie": "10",
           "type_voie": "RUE",
           "libelle_voie": "Lebouis",
           "code_insee": "75120",
           "code_postal": "75014",
           "commune": "Paris"
         },
         "groupement": "Simple",
         "mise_a_jour": "2013-06-28"
+      }
+    }
     DOCUMENTS_ASSOCIATION_RESPONSE = {
       "nombre_documents": 2,
       "documents": [
+        {
           "type": "Statuts",
           "url": "https://apientreprise.fr/attestations/40ab0b07d434d0417e8997ce7c5afbef/attestation_document_association.pdf",
           "timestamp": "1500660325"
         },
+        {
           "type": "Récépissé",
           "url": "https://apientreprise.fr/attestations/40ab0b07d434d0417e8997ce7c5afbef/recepisse_association.pdf",
           "timestamp": "1500667325"
         },
+      ]
+    }
     DOCUMENT_ASSOCIATION_RESPONSE = "binary content"
     REQUEST_PARAMS = {'context': 'MSP', 'object': 'demand', 'recipient': 'siret'}
     @urlmatch(netloc='^entreprise.api.gouv.fr$',
               path='^/v2/etablissements/')
     def api_entreprise_etablissements(url, request):
         return response(200, ETABLISSEMENTS_RESPONSE, request=request)
     @urlmatch(netloc='^entreprise.api.gouv.fr$',
               path='^/v2/entreprises/')
     def api_entreprise_entreprises(url, request):
         return response(200, ENTREPRISES_RESPONSE, request=request)
     @urlmatch(netloc='^entreprise.api.gouv.fr$',
               path='^/v2/associations/')
     def api_entreprise_associations(url, request):
         return response(200, ASSOCIATIONS_RESPONSE, request=request)
     @urlmatch(netloc='^entreprise.api.gouv.fr$',
               path='^/v2/extraits_rcs_infogreffe/')
     def api_entreprise_extraits_rcs(url, request):
         return response(200, EXTRAITS_RCS_RESPONSE, request=request)
     @urlmatch(netloc='^entreprise.api.gouv.fr$',
               path='^/v2/documents_associations/')
     def api_entreprise_documents_associations(url, request):
         return response(200, DOCUMENTS_ASSOCIATION_RESPONSE, request=request)
     @urlmatch(netloc='^apientreprise.fr$',
               path='^/attestations/')
     def api_entreprise_document_association(url, request):
         return response(200, DOCUMENT_ASSOCIATION_RESPONSE, request=request)
     @urlmatch(netloc='^entreprise.api.gouv.fr$')
     def api_entreprise_error_500(url, request):
         return response(500, 'bad error happened', request=request)
     @urlmatch(netloc='^entreprise.api.gouv.fr$')
     def api_entreprise_error_not_json(url, request):
         return response(200, 'simple text', request=request)
     @urlmatch(netloc='^entreprise.api.gouv.fr$')
     def api_entreprise_error_not_found(url, request):
         return response(404, {
             'error': 'not_found',
             'message': u'Page not found'
         }, request=request)
     @pytest.yield_fixture
     def mock_api_entreprise():
         with HTTMock(api_entreprise_etablissements, api_entreprise_entreprises, api_entreprise_associations, api_entreprise_extraits_rcs,
                      api_entreprise_associations, api_entreprise_documents_associations, api_entreprise_document_association):
             yield None
     @pytest.fixture
     def resource(db):
         return make_resource(
             APIEntreprise,
             slug='test',
             title='API Entreprise',
             description='API Entreprise',
             token='83c68bf0b6013c4daf3f8213f7212aa5')
     @mock.patch('passerelle.utils.Request.get')
     def test_endpoint_with_no_params(mocked_get, app, resource):
         mocked_get.return_value = FakedResponse(content='{}', status_code=200)
         response = app.get('/api-entreprise/test/documents_associations/443170139/', status=400)
         assert response.json['err_class'] == 'passerelle.views.WrongParameter'
         assert response.json['err_desc'] == "missing parameters: 'context'."
         params = {'context': 'Custom context', 'object': 'Custom object', 'recipient': 'Custom recipient'}
         response = app.get('/api-entreprise/test/documents_associations/443170139/', params=params)
         assert mocked_get.call_args[1]['data']['context'] == 'Custom context'
         assert mocked_get.call_args[1]['data']['object'] == 'Custom object'
         assert mocked_get.call_args[1]['data']['recipient'] == 'Custom recipient'
     def test_entreprises_endpoint(app, resource, mock_api_entreprise):
         response = app.get('/api-entreprise/test/entreprises/443170139/',
                            params=REQUEST_PARAMS)
         data = response.json['data']
         assert data['entreprise']['categorie_entreprise'] == 'PME'
         assert data['entreprise']['numero_tva_intracommunautaire'] == 'FR16418166096'
         assert data['entreprise']['siret_siege_social'] == '41816609600051'
         assert data['entreprise']['forme_juridique_code'] == '5699'
         assert data['entreprise']['siren'] == '418166096'
         assert data['entreprise']['date_creation'] == '1998-03-31'
         assert data['entreprise']['mandataires_sociaux'][0]['date_naissance'] == '1965-01-27'
         assert 'etablissement_siege' in data
         assert data['etablissement_siege']['siret'] == '41816609600051'
         assert data['etablissement_siege']['adresse']['numero_voie'] == '50'
         assert data['etablissement_siege']['adresse']['type_voie'] == 'AV'
         assert data['etablissement_siege']['adresse']['nom_voie'] == 'DES CHAMPS ELYSEES'
         assert data['etablissement_siege']['adresse']['code_postal'] == '75008'
         assert data['etablissement_siege']['adresse']['localite'] == 'PARIS 8'
         assert data['etablissement_siege']['adresse']['code_insee_localite'] == '75108'
         assert data['etablissement_siege']['date_mise_a_jour'] == '2015-12-03'
     def test_etablissements_endpoint(app, resource, mock_api_entreprise):
         response = app.get('/api-entreprise/test/etablissements/44317013900036/',
                            params=REQUEST_PARAMS)
         assert 'data' in response.json
         data = response.json['data']
         assert 'etablissement' in data
         assert data['etablissement']['siret'] == '41816609600051'
         assert data['etablissement']['naf'] == '6202A'
         assert data['etablissement']['date_mise_a_jour'] == '2015-12-03'
         assert data['etablissement']['date_creation_etablissement'] == '2005-02-16'
         assert 'adresse' in data['etablissement']
         assert data['etablissement']['adresse']['code_postal'] == '75008'
         assert data['etablissement']['adresse']['localite'] == 'PARIS 8'
         assert data['etablissement']['adresse']['code_insee_localite'] == '75108'
     def test_associations_endpoint(app, resource, mock_api_entreprise):
         response = app.get('/api-entreprise/test/associations/443170139/',
                            params=REQUEST_PARAMS)
         assert 'data' in response.json
         data = response.json['data']
         assert 'association' in data
         assert data['association']['id'] == 'W751135389'
         assert data['association']['siret'] == '42135938100025'
         assert data['association']['date_creation'] == '1993-02-11'
         assert data['association']['date_declaration'] == '2013-06-28'
         assert data['association']['date_publication'] == '1993-03-03'
         assert 'adresse_siege' in data['association']
         assert data['association']['adresse_siege']['code_postal'] == '75014'
         assert data['association']['adresse_siege']['code_insee'] == '75120'
     def test_documents_associations_endpoint(app, resource, mock_api_entreprise):
         response = app.get('/api-entreprise/test/documents_associations/443170139/',
                            params=REQUEST_PARAMS)
         assert 'data' in response.json
         data = response.json['data']
         assert len(data) == 2
         for document in data:
             assert 'id' in document
             assert 'text' in document
             assert 'url' in document
             assert 'type' in document
     def test_document_association(app, resource, mock_api_entreprise, freezer):
         response = app.get('/api-entreprise/test/documents_associations/443170139/',
                            params=REQUEST_PARAMS)
         assert 'data' in response.json
         data = response.json['data']
         assert len(data) == 2
         document = data[0]
         assert 'url' in document
         resp = app.get(document['url'],
                        params={'context': 'MSP', 'object': 'demand', 'recipient': 'siret'},
                        status=200)
         # try to get document with wrong signature
         url = document['url']
         wrong_url = document['url'] + "wrong/"
         resp = app.get(wrong_url, status=404)
         # try expired url
         freezer.move_to(timezone.now() + timezone.timedelta(days=8))
         resp = app.get(document['url'], status=404)
     def test_error_500(app, resource, mock_api_entreprise):
         with HTTMock(api_entreprise_error_500):
             response = app.get('/api-entreprise/test/entreprises/443170139/',
                                params=REQUEST_PARAMS)
             assert response.status_code == 200
             assert response.json['err'] == 1
             assert response.json['data']['status_code'] == 500
             assert 'error happened' in response.json['err_desc']
     def test_no_json_error(app, resource, mock_api_entreprise):
         with HTTMock(api_entreprise_error_not_json):
             response = app.get('/api-entreprise/test/entreprises/443170139/',
                                params=REQUEST_PARAMS)
             assert response.status_code == 200
             assert response.json['err'] == 1
             assert response.json['err_desc'] == "API-entreprise returned non-JSON content with status 200: 'simple text'"
     def test_error_404(app, resource, mock_api_entreprise):
         with HTTMock(api_entreprise_error_not_found):
             response = app.get('/api-entreprise/test/entreprises/443170139/',
                                params=REQUEST_PARAMS)
             assert response.status_code == 200
             assert response.json['err'] == 1
             assert response.json['err_desc'] == 'Page not found'
+    -

Projet

Général

Profil

Produits Entr'ouvert » Passerelle

0001-api_entreprise-add-initial-connector-30010.patch