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

Serghei Mihai, 20 mars 2019 12:05

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 |  35 ++
 .../api_entreprise/migrations/__init__.py     |   0
 passerelle/apps/api_entreprise/models.py      | 214 ++++++++++
 passerelle/settings.py                        |   1 +
 passerelle/static/css/style.css               |   3 +-
 tests/test_api_entreprise.py                  | 393 ++++++++++++++++++
 7 files changed, 645 insertions(+), 1 deletion(-)
 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')),
                     ('recipient', models.CharField(help_text='ex: siret, siren, waldec, etc...', max_length=256, verbose_name='Call recipient')),
                     ('context', models.CharField(help_text='ex: APS, MPS, etc...', max_length=256, verbose_name='Call context')),
                     ('obj', models.CharField(max_length=49, verbose_name='Calls object')),
                     ('users', models.ManyToManyField(blank=True, to='base.ApiUser')),
                 ],
                 options={
                     'verbose_name': 'API Entreprise',
                 },
             ),
+        ]

     # passerelle.apps.api_entreprise
     # 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 urlparse import urljoin
     import requests
     from django.db import models
     from django.utils.translation import ugettext_lazy as _
     from django.utils.timezone import datetime
     from django.http import HttpResponse, Http404
     from django.core.signing import Signer, BadSignature
     from django.core.urlresolvers import reverse
     from passerelle.base.models import BaseResource
     from passerelle.utils.api import endpoint
     from passerelle.utils.jsonresponse import APIError
     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'))
         recipient = models.CharField(max_length=256, verbose_name=_('Call recipient'),
                                      help_text=_('ex: siret, institution_id, waldec, etc...'))
         context = models.CharField(max_length=256, verbose_name=_('Call context'),
                                      help_text=_('ex: APS, MPS, etc...'))
         obj = models.CharField(max_length=49, verbose_name=_('Calls object'))
         category = _('Business Process Connectors')
         class Meta:
             verbose_name = _('API Entreprise')
         def get(self, path, **kwargs):
             params = {'token': self.token,
                 'context': kwargs.get('context', self.context),
                 'object': kwargs.get('object', self.obj),
                 'recipient': kwargs.get('recipient', self.recipient)
+            }
             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,
                     })
             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',
+                    }
+                  }
+        )
         def documents_associations(self, request, institution_id, **kwargs):
             data = []
             signer = Signer()
             resp = self.get('documents_associations/%s/' % institution_id)
             for item in resp['data'].get('documents', []):
                 # ignore documents with no type
                 if not item.get('type'):
                     continue
                 dt = datetime.fromtimestamp(int(item['timestamp']))
                 signature = signer.signature('%s-%s' % (item['type'], item['timestamp']))
                 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'],
                              'datetime': dt,
                              'type': item['type'],
                              'url': document_url})
             # sort data by date
             data.sort(key=lambda i: i['datetime'])
             return {'err': 0, 'data': data}
         @endpoint(perm='can_access',
                   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',
+                      }
+                  }
+        )
         def document(self, request, institution_id, document_id, **kwargs):
             signer = Signer()
             resp = self.get('documents_associations/%s/' % institution_id)
             for item in resp['data'].get('documents', []):
                 # ignore documents with no type
                 if not item.get('type'):
                     continue
                 try:
                     signer.unsign('%s-%s:%s' % (item['type'], item['timestamp'], document_id))
                 except BadSignature:
                     continue
                 return HttpResponse(self.requests.get(item['url']),
                                     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',
+                    }
+                  }
+        )
         def extraits_rcs(self, request, institution_id, **kwargs):
             return self.get('extraits_rcs_infogreffe/%s/' % institution_id)
         @endpoint(perm='can_access',
                   pattern='(?P<institution_id>\w+)/$',
                   example_pattern='{institution_id}/',
                   description=_('Get association\'s related informations'),
                   parameters={
                       'institution_id': {
                           'description': _('institution id'),
                           'example_value': '44317013900036',
+                    }
+                  }
+        )
         def associations(self, request, institution_id, **kwargs):
             return self.get('associations/%s/' % institution_id)
         @endpoint(perm='can_access',
                   pattern='(?P<institution_id>\w+)/$',
                   example_pattern='{institution_id}/',
                   description=_('Get firm\'s related informations'),
                   parameters={
                       'institution_id': {
                           'description': _('association id'),
                           'example_value': '44317013900036',
+                    }
+                  }
+        )
         def entreprises(self, request, institution_id, **kwargs):
             return self.get('entreprises/%s/' % institution_id)
         @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',
+                    }
+                  }
+        )
         def etablissements(self, request, institution_id, **kwargs):
             return self.get('etablissements/%s/' % institution_id)

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
164	164	content: "\f15b"; /* file */
165	165	}
166	166
167		li.connector.apiparticulier a::before {
	167	li.connector.apiparticulier a::before,
	168	li.connector.apientreprise a::before {
168	169	content: "\f2c3"; /* id-card-o */
169	170	}
170	171

     # -*- coding: utf-8 -*-
     # tests/test_api_entreprise.py
     # Copyright (C) 2018  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
     from httmock import urlmatch, HTTMock, response
     from django.core.urlresolvers import reverse
     from passerelle.apps.api_entreprise.models import APIEntreprise
     from utils import make_resource, endpoint_get
     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"
     @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',
             recipient='Test', context='Test',
             obj='Test')
     def test_entreprises_endpoint(app, resource, mock_api_entreprise):
         response = app.get('/api-entreprise/test/entreprises/443170139/')
         data = response.json['data']
         assert 'entreprise' in 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'] == 891381600
         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'
     def test_etablissements_endpoint(app, resource, mock_api_entreprise):
         response = app.get('/api-entreprise/test/etablissements/44317013900036/')
         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'] == 1449183600
         assert data['etablissement']['date_creation_etablissement'] == 1108594800
         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/')
         assert 'data' in response.json
         data = response.json['data']
         assert 'association' in data
         assert data['association']['id'] == 'W751135389'
         assert data['association']['siret'] == '42135938100025'
         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/')
         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 'datetime' in document
             assert 'type' in document
     def test_document_association(app, resource, mock_api_entreprise):
         response = app.get('/api-entreprise/test/documents_associations/443170139/')
         assert 'data' in response.json
         data = response.json['data']
         assert len(data) == 2
         for document in data:
             assert 'url' in document
             resp = app.get(document['url'], status=200)
             # try to get document with wrong signature
             url = document['url']
             wrong_url = document['url'] + "wrong/"
             resp = app.get(wrong_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/')
             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/')
             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/')
             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