Produits Entr'ouvert » Passerelle

# -*- coding: utf-8 -*-

from __future__ import unicode_literals

from django.db import models, migrations

class Migration(migrations.Migration):

    dependencies = [

        ('base', '0002_auto_20151009_0326'),

    ]

    operations = [

        migrations.CreateModel(

            name='APIParticulier',

            fields=[

                ('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),

                ('title', models.CharField(max_length=50)),

                ('slug', models.SlugField()),

                ('description', models.TextField()),

                ('log_level', models.CharField(default=b'NOTSET', max_length=10, verbose_name='Log Level', choices=[(b'NOTSET', b'NOTSET'), (b'DEBUG', b'DEBUG'), (b'INFO', b'INFO'), (b'WARNING', b'WARNING'), (b'ERROR', b'ERROR'), (b'CRITICAL', b'CRITICAL'), (b'FATAL', b'FATAL')])),

                ('_platform', models.CharField(max_length=8, verbose_name='platform', choices=[(b'test', 'Test'), (b'prod', 'Production'), (b'dev', 'Development'), (b'mock', 'Mock')])),

                ('_api_key', models.CharField(default=b'', max_length=64, verbose_name='API key', blank=True)),

                ('users', models.ManyToManyField(to='base.ApiUser', blank=True)),

            ],

            options={

                'abstract': False,

            },

            bases=(models.Model,),

        ),

    ]

# passerelle.apps.api_particulier

# Copyright (C) 2017  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-Particulier web-service from SGMAP:

   https://particulier.api.gouv.fr/

'''

from urlparse import urljoin

from collections import OrderedDict

import requests

from django.db import models

from django.utils.translation import ugettext_lazy as _

from passerelle.base.models import BaseResource

from passerelle.utils.api import endpoint

from passerelle.utils.jsonresponse import APIError

class APIParticulier(BaseResource):

    PLATFORMS = [

        {

            'name': 'prod',

            'label': _('Production'),

            'url': 'https://particulier.api.gouv.fr/api/'

        },

        {

            'name': 'test',

            'label': _('Test'),

            'url': 'https://particulier-sandbox.api.gouv.fr/api/',

            'api_key': 'test-token',

        },

    ]

    PLATFORMS = OrderedDict([(platform['name'], platform) for platform in PLATFORMS])

    _platform = models.CharField(

        verbose_name=_('Platform'),

        max_length=8,

        choices=[(key, platform['label']) for key, platform in PLATFORMS.iteritems()])

    _api_key = models.CharField(

        max_length=64,

        default='',

        blank=True,

        verbose_name=_('API key'))

    @property

    def platform(self):

        return self.PLATFORMS[self._platform]

    @property

    def url(self):

        return self.platform['url']

    @property

    def api_key(self):

        return self.platform.get('api_key', self._api_key)

    def get(self, path, **kwargs):

        user = kwargs.pop('user', None)

        url = urljoin(self.url, path)

        headers = {'X-API-KEY': self.api_key}

        if user:

            headers['X-User'] = user

        response = None

        try:

            response = self.requests.get(

                url,

                headers=headers,

                **kwargs)

        except requests.RequestException as e:

            raise APIError(

                u'API-particulier platform "%s" connection error: %s' %

                (self._platform, response.status_code),

                data={

                    'platform': self._platform,

                    'error': unicode(e),

                })

        try:

            data = response.json()

        except ValueError as e:

            content = repr(response.content[:1000])

            raise APIError(

                u'API-particulier platform "%s" returned non-JSON content with status %s: %s' %

                (self._platform, response.status_code, content),

                data={

                    'status_code': response.status_code,

                    'exception': unicode(e),

                    'platform': self._platform,

                    '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-particulier platform "%s" returned a non 200 status %s: %s' %

                (self._platform, response.status_code, data),

                data={

                    'status_code': response.status_code,

                    'platform': self._platform,

                    'content': data,

                })

        return {

            'err': 0,

            'data': data,

        }

    @endpoint(perm='can_access',

              description=_('Get citizen\'s fiscal informations'),

              parameters={

                  'numero_fiscal': {

                      'description': _('fiscal identifier'),

                      'example_value': '12',

                  },

                  'reference_avis': {

                      'description': _('tax notice number'),

                      'example_value': '15',

                  },

                  'user': {

                      'description': _('requesting user'),

                      'example_value': 'John Doe (agent)',

                  },

              })

    def impots_svair(self, request, numero_fiscal, reference_avis, user=None):

        return self.get('impots/svair', params={

            'numeroFiscal': numero_fiscal,

            'referenceAvis': reference_avis,

        }, user=user)

    @endpoint(perm='can_access',

              description=_('Get citizen\'s fiscal address'),

              parameters={

                  'numero_fiscal': {

                      'description': _('fiscal identifier'),

                      'example_value': '12',

                  },

                  'reference_avis': {

                      'description': _('tax notice number'),

                      'example_value': '15',

                  },

                  'user': {

                      'description': _('requesting user'),

                      'example_value': 'John Doe (agent)',

                  },

              })

    def impots_adresse(self, request, numero_fiscal, reference_avis, user=None):

        return self.get('impots/adress', params={

            'numeroFiscal': numero_fiscal,

            'referenceAvis': reference_avis,

        }, user=user)

    @endpoint(perm='can_access',

              description=_('Get family allowances recipient informations'),

              parameters={

                  'code_postal': {

                      'description': _('postal code'),

                      'example_value': '99148',

                  },

                  'numero_allocataire': {

                      'description': _('recipient identifier'),

                      'example_value': '0000354',

                  },

                  'user': {

                      'description': _('requesting user'),

                      'example_value': 'John Doe (agent)',

                  },

              })

    def caf_famille(self, request, code_postal, numero_allocataire, user=None):

        return self.get('caf/famille', params={

            'codePostal': code_postal,

            'numeroAllocataire': numero_allocataire,

        }, user=user)

    category = _('Business Process Connectors')

    class Meta:

        verbose_name = _('API Particulier')

    'passerelle.apps.api_particulier',

# -*- coding: utf-8 -*-

# tests/test_api_particulier.py

# Copyright (C) 2017  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_particulier.models import APIParticulier

from utils import make_resource, endpoint_get

SVAIR_RESPONSE = {

    "declarant1": {

        "nom": "Martin",

        "nomNaissance": "Martin",

        "prenoms": "Pierre",

        "dateNaissance": "22/03/1985"

    },

    "declarant2": {

        "nom": "Martin",

        "nomNaissance": "Honore",

        "prenoms": "Marie",

        "dateNaissance": "03/04/1986"

    },

    "foyerFiscal": {

        "annee": 2015,

        "adresse": "12 rue Balzac 75008 Paris"

    },

    "dateRecouvrement": "10/10/2015",

    "dateEtablissement": "08/07/2015",

    "nombreParts": 2,

    "situationFamille": "Marié(e)s",

    "nombrePersonnesCharge": 2,

    "revenuBrutGlobal": 29880,

    "revenuImposable": 29880,

    "impotRevenuNetAvantCorrections": 2165,

    "montantImpot": 2165,

    "revenuFiscalReference": 29880,

    "anneeImpots": "2015",

    "anneeRevenus": "2014"

}

IMPOTS_ADRESSE = {

    "adresses": [

        {

            "adresse": {

                "citycode": "75108",

                "street": "Rue Balzac",

                "name": "12 Rue Balzac",

                "housenumber": "12",

                "city": "Paris",

                "type": "housenumber",

                "context": "75, Île-de-France",

                "score": 0.9401454545454544,

                "label": "12 Rue Balzac 75008 Paris",

                "postcode": "75008"

            },

            "geometry": {

                "type": "Point",

                "coordinates": [

                    2.300816,

                    48.873951

                ]

            }

        }

    ],

    "declarant1": {

        "nom": "Martin",

        "nomNaissance": "Martin",

        "prenoms": "Pierre",

        "dateNaissance": "22/03/1985"

    },

    "declarant2": {

        "nom": "Martin",

        "nomNaissance": "Honore",

        "prenoms": "Marie",

        "dateNaissance": "03/04/1986"

    },

    "foyerFiscal": {

        "annee": 2015,

        "adresse": "12 rue Balzac 75008 Paris"

    }

}

CAF_FAMILLE = {

    "adresse": {

        "codePostalVille": "12345 CONDAT",

        "complementIdentiteGeo": "ESCALIER B",

        "identite": "Madame MARIE DUPONT",

        "numeroRue": "123 RUE BIDON",

        "pays": "FRANCE"

    },

    "allocataires": [

        {

            "dateDeNaissance": "12111971",

            "nomPrenom": "MARIE DUPONT",

            "sexe": "F"

        },

        {

            "dateDeNaissance": "18101969",

            "nomPrenom": "JEAN DUPONT",

            "sexe": "M"

        }

    ],

    "annee": 2017,

    "enfants": [

        {

            "dateDeNaissance": "11122016",

            "nomPrenom": "LUCIE DUPONT",

            "sexe": "F"

        }

    ],

    "mois": 4,

    "quotientFamilial": 1754

}

@urlmatch(netloc='^particulier.*\.api\.gouv\.fr$',

          path='^/api/impots/svair$')

def api_particulier_impots_svair(url, request):

    return response(200, SVAIR_RESPONSE, request=request)

@urlmatch(netloc='^particulier.*\.api\.gouv\.fr$',

          path='^/api/impots/adresse$')

def api_particulier_impots_adresse(url, request):

    return response(200, IMPOTS_ADRESSE, request=request)

@urlmatch(netloc='^particulier.*\.api\.gouv\.fr$',

          path='^/api/caf/famille$')

def api_particulier_caf_famille(url, request):

    return response(200, CAF_FAMILLE, request=request)

@urlmatch(netloc='^particulier.*\.api\.gouv\.fr$')

def api_particulier_error_500(url, request):

    return response(500, 'something bad happened', request=request)

@urlmatch(netloc='^particulier.*\.api\.gouv\.fr$')

def api_particulier_error_not_json(url, request):

    return response(200, 'something bad happened', request=request)

@urlmatch(netloc='^particulier.*\.api\.gouv\.fr$')

def api_particulier_error_not_found(url, request):

    return response(404, {

        'error': 'not_found',

        'message': u'Les paramètres fournis sont incorrects ou ne correspondent pas à un avis'

    }, request=request)

@pytest.yield_fixture

def mock_api_particulier():

    with HTTMock(api_particulier_impots_svair, api_particulier_impots_adresse, api_particulier_caf_famille):

        yield None

@pytest.fixture

def resource(db):

    return make_resource(

        APIParticulier,

        slug='test',

        title='API Particulier Prod',

        description='API Particulier Prod',

        _platform='test')

def test_error(app, resource, mock_api_particulier):

    with HTTMock(api_particulier_error_500):

        def do(endpoint, params):

            resp = endpoint_get(

                '/api-particulier/test/%s' % endpoint,

                app,

                resource,

                endpoint,

                params=params)

            assert resp.status_code == 200

            assert resp.json['err'] == 1

            assert resp.json['data']['status_code'] == 500

        vector = [

            (['impots_svair', 'impots_adresse'], {

                'numero_fiscal': 12,

                'reference_avis': 15,

            }),

            (['caf_famille'], {

                'code_postal': 12,

                'numero_allocataire': 15

            }),

        ]

        for endpoints, params in vector:

            for endpoint in endpoints:

                do(endpoint, params)

    with HTTMock(api_particulier_error_not_json):

        def do(endpoint, params):

            resp = endpoint_get(

                '/api-particulier/test/%s' % endpoint,

                app,

                resource,

                endpoint,

                params=params)

            assert resp.status_code == 200

            assert resp.json['err'] == 1

            assert resp.json['data']['exception'] == 'No JSON object could be decoded'

        vector = [

            (['impots_svair', 'impots_adresse'], {

                'numero_fiscal': 12,

                'reference_avis': 15,

            }),

            (['caf_famille'], {

                'code_postal': 12,

                'numero_allocataire': 15

            }),

        ]

        for endpoints, params in vector:

            for endpoint in endpoints:

                do(endpoint, params)

    with HTTMock(api_particulier_error_not_found):

        def do(endpoint, params):

            resp = endpoint_get(

                '/api-particulier/test/%s' % endpoint,

                app,

                resource,

                endpoint,

                params=params)

            assert resp.status_code == 200

            assert resp.json['err'] == 1

            assert resp.json['err_desc'].endswith(u'à un avis')

        vector = [

            (['impots_svair', 'impots_adresse'], {

                'numero_fiscal': 12,

                'reference_avis': 15,

            }),

            (['caf_famille'], {

                'code_postal': 12,

                'numero_allocataire': 15

            }),

        ]

        for endpoints, params in vector:

            for endpoint in endpoints:

                do(endpoint, params)

def test_impots_svair(app, resource, mock_api_particulier):

    resp = endpoint_get(

        '/api-particulier/test/impots_svair',

        app,

        resource,

        'impots_svair',

        params={

            'numero_fiscal': 12,

            'reference_avis': 15,

            'user': 'John Doe',

        })

    assert resp.json['data']['montantImpot'] == 2165

def test_impots_adresse(app, resource, mock_api_particulier):

    resp = endpoint_get(

        '/api-particulier/test/impots_adresse',

        app,

        resource,

        'impots_adresse',

        params={

            'numero_fiscal': 12,

            'reference_avis': 15,

            'user': 'John Doe',

        })

    assert resp.json['data']['adresses'][0]['adresse']['citycode'] == '75108'

def test_caf_famille(app, resource, mock_api_particulier):

    resp = endpoint_get(

        '/api-particulier/test/caf_famille',

        app,

        resource,

        'caf_famille',

        params={

            'code_postal': '99148',

            'numero_allocataire': '000354',

            'user': 'John Doe',

        })

    assert resp.json['data']['adresse']['codePostalVille'] == '12345 CONDAT'

def test_detail_page(app, resource):

    response = app.get(reverse('view-connector', kwargs={

        'connector': 'api-particulier',

        'slug': 'test',

    }))

    assert 'API Particulier Prod' in response.content

    assert 'family allowance' in response.content

    assert 'fiscal information' in response.content

    assert 'fiscal address' in response.content

def make_resource(model_class, **kwargs):

    resource = model_class.objects.create(**kwargs)

    setup_access_rights(resource)

    return resource

def endpoint_get(expected_url, app, ressource, endpoint, **kwargs):

    url = generic_endpoint_url(

        connector=ressource.__class__.get_connector_slug(),

        endpoint=endpoint,

        slug=ressource.slug)

    assert url == expected_url, 'endpoint URL has changed'

    return app.get(url, **kwargs)