Produits Entr'ouvert » Passerelle

# 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

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

            'label': _('Test'),

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

            'api_key': 'test-token',

        },

        {

            'name': 'prod',

            'label': _('Production'),

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

        },

        {

            'name': 'dev',

            'label': _('Development'),

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

        },

        {

            'name': 'mock',

            'label': _('Mock'),

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

        },

    ]

    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 = self.requests.get(

            url,

            headers=headers,

            **kwargs)

        if response.status_code != 200:

            raise APIError(

                u'API-particulier platform "%s" returned non-200 code: %s' %

                (self._platform, response.status_code),

                data={

                    'platform': self._platform,

                    'code': response.status_code,

                    'content': repr(response.content[:1000]),

                })

        try:

            return response.json()

        except ValueError as e:

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

            raise APIError(

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

                (self._platform, content),

                data={

                    'exception': unicode(e),

                    'platform': self._platform,

                    'content': content,

                })

    @endpoint(serializer_type='json-api', perm='can_access')

    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(serializer_type='json-api', perm='can_access')

    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(serializer_type='json-api', perm='can_access')

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

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

            'codePostal': code_postal,

            'numeroAllocataire': numero_allocataire,

        }, user=user)

    @endpoint(serializer_type='json-api', perm='can_access')

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

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

            'codePostal': code_postal,

            'numeroAllocataire': numero_allocataire,

        }, user=user)

    @endpoint(serializer_type='json-api', perm='can_access')

    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)

    '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 api_particulier.models import APIParticulier

from utils import make_ressource, 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"

    }

}

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

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)

@pytest.yield_fixture

def mock_api_particulier():

    with HTTMock(api_particulier_impots_svair, api_particulier_impots_adresse):

        yield None

@pytest.fixture

def ressource(db):

    return make_ressource(

        APIParticulier,

        slug='test',

        title='API Particulier Prod',

        description='API Particulier Prod',

        _platform='test')

def test_error(app, ressource, mock_api_particulier):

    with HTTMock(api_particulier_error_500):

        def do(endpoint, params):

            resp = endpoint_get(

                app,

                ressource,

                endpoint,

                params=params)

            assert resp.status_code == 200

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

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

        vector = [

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

                'numero_fiscal': 12,

                'reference_avis': 15,

                'user': 'john.doe',

            }),

            (['caf_qf', 'caf_adresse', '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(

                app,

                ressource,

                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,

                'user': 'john.doe',

            }),

            (['caf_qf', 'caf_adresse', '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, ressource, mock_api_particulier):

    resp = endpoint_get(

        app,

        ressource,

        'impots_svair',

        params={

            'numero_fiscal': 12,

            'reference_avis': 15

        })

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

def test_impots_adresse(app, ressource, mock_api_particulier):

    resp = endpoint_get(

        app,

        ressource,

        'impots_adresse',

        params={

            'numero_fiscal': 12,

            'reference_avis': 15

        })

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

# FIXME: CAF web services are currently broken, add test eventually when we can test them

def make_ressource(model_class, **kwargs):

    api, created = ApiUser.objects.get_or_create(

        username='all',

        keytype='',

        key='')

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

    obj_type = ContentType.objects.get_for_model(model_class)

    AccessRight.objects.get_or_create(

        codename='can_access',

        apiuser=api,

        resource_type=obj_type,

        resource_pk=ressource.pk)

    return ressource

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

    url = generic_endpoint_url(

        connector=ressource.__class__.get_connector_slug(),

        endpoint=endpoint,

        slug=ressource.slug)

    return app.get(url, **kwargs)