Produits Entr'ouvert » Passerelle

# -*- 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 - 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

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

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'))

    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,

                })

        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',

                }

              }

    )

    def documents_associations(self, request, institution_id, **kwargs):

        data = []

        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

            signature = signing.dumps({'type': item['type'], 'timestamp': 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': datetime.fromtimestamp(int(item['timestamp'])),

                         '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):

        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:

                r = signing.loads(document_id, max_age=DOCUMENT_SIGNATURE_MAX_AGE)

                print r

            except signing.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 institution\'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 institution\'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.apps.api_entreprise',

li.connector.apientreprise a::before {

	content: "\f1ad";  /* fa-building-o */

}

# -*- 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

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

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 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/')

    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/')

    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/')

    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, freezer):

    response = app.get('/api-entreprise/test/documents_associations/443170139/')

    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'], 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/')

        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'