Produits Entr'ouvert » Passerelle

# Generated by Django 2.2.24 on 2022-06-21 13:21

from django.db import migrations, models

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0029_auto_20210202_1627'),

    ]

    operations = [

        migrations.CreateModel(

            name='Resource',

            fields=[

                (

                    'id',

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

                ),

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

                ('slug', models.SlugField(unique=True, verbose_name='Identifier')),

                ('description', models.TextField(verbose_name='Description')),

                ('consumer_key', models.CharField(max_length=128, verbose_name='Consumer key')),

                ('consumer_secret', models.CharField(max_length=128, verbose_name='Consumer secret')),

                (

                    'environment',

                    models.CharField(

                        choices=[('test', 'Test'), ('prod', 'Production')],

                        max_length=4,

                        verbose_name='Environment',

                    ),

                ),

                (

                    'users',

                    models.ManyToManyField(

                        blank=True,

                        related_name='_resource_users_+',

                        related_query_name='+',

                        to='base.ApiUser',

                    ),

                ),

            ],

            options={

                'verbose_name': 'SIvin',

            },

        ),

    ]

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2022 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/>.xs

from urllib.parse import urljoin

from django.core.cache import cache

from django.db import models

from django.utils.translation import ugettext_lazy as _

from requests import RequestException

from passerelle.base.models import BaseResource

from passerelle.utils.api import APIError, endpoint

PLATFORMS = {

    'test': {

        'token_url': 'https://api.rec.sivin.fr/token',

        'api_base_url': 'https://api.rec.sivin.fr/sivin/v2/',

    },

    'prod': {'token_url': 'https://api.sivin.fr/token', 'api_base_url': 'https://api.rec.sivin.fr/sivin/v1/'},

}

ENVS = (('test', _('Test')), ('prod', _('Production')))

class Resource(BaseResource):

    consumer_key = models.CharField(_('Consumer key'), max_length=128)

    consumer_secret = models.CharField(_('Consumer secret'), max_length=128)

    environment = models.CharField(_('Environment'), choices=ENVS, max_length=4)

    category = _('Data Sources')

    class Meta:

        verbose_name = _('SIvin')

    def check_status(self):

        self.get_token()

    def get_token(self):

        cache_key = 'sivin-%s-token' % self.pk

        token = cache.get(cache_key)

        if token:

            return token

        resp = self.requests.post(

            PLATFORMS[self.environment]['token_url'],

            auth=(self.consumer_key, self.consumer_secret),

            data={'grant_type': 'client_credentials'},

        )

        if resp.status_code >= 400:

            raise APIError('Failed to get token. Error: %s' % resp.status_code)

        resp = resp.json()

        token = resp['access_token']

        timeout = int(resp['expires_in'])

        cache.set(cache_key, token, timeout)

        self.logger.debug('new token: %s (timeout %ss)', token, timeout)

        return token

    def call(self, endpoint, payload):

        url = urljoin(PLATFORMS[self.environment]['api_base_url'], endpoint)

        try:

            resp = self.requests.post(

                url, headers={'Authorization': 'Bearer %s' % self.get_token()}, json=payload

            )

            if resp.status_code >= 400:

                raise APIError(resp.content)

        except RequestException as e:

            raise APIError('failed to call %s: %s' % (url, e))

        return resp.json()

    def get_infos_by_immat(self, endpoint, immat, codesra=None):

        payload = {'immat': immat}

        if codesra is not None:

            payload['codesra'] = codesra

        return {'data': self.call(endpoint, payload)}

    @endpoint(

        perm='can_access',

        description=_('Get vehicle informations by VIN number'),

        parameters={'vin': {'description': _('VIN number'), 'example_value': 'VF1BA0E0514143067'}},

    )

    def consultervehiculeparvin(self, request, vin):

        return {'data': self.call('consultervehiculeparvin', {'vin': vin})}

    @endpoint(

        perm='can_access',

        description=_('Get vehicles list of a SIREN'),

        parameters={'siren': {'description': _('SIREN number'), 'example_value': '000399634'}},

    )

    def consulterflotteparsiren(self, request, siren):

        result = self.call('consulterflotteparsiren', {'siren': siren})

        return {'data': [{'id': vin, 'text': vin} for vin in result.get('vins', [])]}

    @endpoint(

        perm='can_access',

        description=_('Get vehicle details by registration plate'),

        parameters={'immat': {'description': _('Registration plate number'), 'example_value': '01XT0747'}},

    )

    def consultervehiculeparimmat(self, request, immat, codesra=None):

        return self.get_infos_by_immat('consultervehiculeparimmat', immat, codesra)

    @endpoint(

        perm='can_access',

        description=_('Get vehicle "finition" by registration plate'),

        parameters={'immat': {'description': _('Registration plate number'), 'example_value': '01XT0747'}},

    )

    def consulterfinitionparimmat(self, request, immat, codesra=None):

        result = self.call('consulterfinitionparimmat', {'immat': immat, 'codesra': codesra})

        result = self.get_infos_by_immat('consulterfinitionparimmat', immat, codesra)

        return {'data': result['data']['finitions']}

    @endpoint(

        perm='can_access',

        description=_('Get vehicle "finition" by registration plate, ordered by rangs'),

        parameters={'immat': {'description': _('Registration plate number'), 'example_value': '01XT0747'}},

    )

    def consulterfinitionscoresparimmat(self, request, immat, codesra=None):

        return self.get_infos_by_immat('consulterfinitionscoresparimmat', immat, codesra)

    @endpoint(

        perm='can_access',

        description=_('Get vehicle theorical "finition" by registration plate'),

        parameters={'immat': {'description': _('Registration plate number'), 'example_value': '01XT0747'}},

    )

    def consulterfinitiontheoriqueparimmat(self, request, immat, codesra=None):

        return self.get_infos_by_immat('consulterfinitiontheoriqueparimmat', immat, codesra)

    'passerelle.apps.sivin',

# Copyright (C) 2022  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 json

import mock

import pytest

from django.contrib.contenttypes.models import ContentType

from django.urls import reverse

from requests.exceptions import ReadTimeout

import tests.utils

from passerelle.apps.sivin.models import Resource

from passerelle.base.models import AccessRight, ApiUser

pytestmark = pytest.mark.django_db

VEHICLE_DETAILS = {

    "carrosserie": "BERLINE",

    "clEnvironPrf": "70/220 2001/100EURO3",

    "codifVin": "VF7FCKFVB26857835",

    "genreVCG": "VP",

    "immatSiv": "FS032GM",

    "genreVPrf": "VP",

    "date1erCir": "2003-11-21",

    "nSiren": "000000000",

}

EXPIRED_TOKEN_MESSAGE = (

    '<ams:fault xmlns:ams="http://wso2.org/apimanager/security">'

    '<ams:code>900901</ams:code><ams:message>Invalid Credentials</ams:message>'

    '<ams:description>Invalid Credentials. Make sure you have provided the correct security credentials</ams:description>'

    '</ams:fault>'

)

TOKEN_401 = tests.utils.FakedResponse(

    content='{"error_description": "Client Authentication failed.", "error": "invalid_client"}',

    status_code=401,

    headers={'Content-Type': 'application/json'},

)

TOKEN = tests.utils.FakedResponse(

    content='{"access_token": "token_value", "scope": "default", "token_type": "Bearer", "expires_in": 3600}',

    status_code=200,

    headers={'Content-Type': 'application/json'},

)

EXPIRED_TOKEN = tests.utils.FakedResponse(

    content=EXPIRED_TOKEN_MESSAGE,

    status_code=500,

)

VIN = tests.utils.FakedResponse(

    content=json.dumps(VEHICLE_DETAILS),

    status_code=200,

    headers={'Content-Type': 'application/json'},

)

VEHICLES = tests.utils.FakedResponse(

    content='{"vins":["VF1FB30A511331122"]}', status_code=200, headers={'Content-Type': 'application/json'}

)

@pytest.fixture

def conn():

    api_user = ApiUser.objects.create(username='sivin', keytype='API', key='sivinkey')

    connector = Resource.objects.create(

        title='Test', slug='test', consumer_key='key', consumer_secret='secret', environment='test'

    )

    obj_type = ContentType.objects.get_for_model(Resource)

    AccessRight.objects.create(

        codename='can_access', apiuser=api_user, resource_type=obj_type, resource_pk=connector.pk

    )

    return connector

@mock.patch('passerelle.utils.Request.post', side_effect=TOKEN_401)

def test_no_api_key(mocked_post, app, conn):

    mocked_post.return_value = TOKEN

    url = reverse(

        'generic-endpoint',

        kwargs={'connector': 'sivin', 'endpoint': 'consultervehiculeparvin', 'slug': conn.slug},

    )

    app.get(url, params={'vin': 'vin'}, status=403)

    assert mocked_post.call_count == 0

@mock.patch('passerelle.utils.Request.post')

def test_wrong_credentials(mocked_post, app, conn):

    mocked_post.return_value = TOKEN_401

    url = reverse(

        'generic-endpoint',

        kwargs={'connector': 'sivin', 'endpoint': 'consultervehiculeparvin', 'slug': conn.slug},

    )

    resp = app.get(url, params={'apikey': 'sivinkey', 'vin': 'VF1BA0E0514143067'}).json

    assert mocked_post.call_count == 1

    assert mocked_post.call_args[0][0] == 'https://api.rec.sivin.fr/token'

    assert resp['err']

    assert resp['err_desc'] == 'Failed to get token. Error: 401'

@mock.patch('passerelle.utils.Request.post')

def test_get_token(mocked_post, app, conn):

    mocked_post.return_value = TOKEN

    conn.get_token()

    assert mocked_post.call_count == 1

    assert mocked_post.call_args[0][0] == 'https://api.rec.sivin.fr/token'

    # token is in cache so no more http hits

    conn.get_token()

    assert mocked_post.call_count == 1

@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, VIN))

def test_get_details_by_vin(mocked_post, app, conn):

    url = reverse(

        'generic-endpoint',

        kwargs={'connector': 'sivin', 'endpoint': 'consultervehiculeparvin', 'slug': conn.slug},

    )

    resp = app.get(url, params={'apikey': 'sivinkey', 'vin': 'VF1BA0E0514143067'}).json

    assert mocked_post.call_count == 2

    assert not resp['err']

    assert resp['data'] == VEHICLE_DETAILS

@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, VEHICLES))

def test_get_vehicles_by_siren(mocked_post, app, conn):

    url = reverse(

        'generic-endpoint',

        kwargs={'connector': 'sivin', 'endpoint': 'consulterflotteparsiren', 'slug': conn.slug},

    )

    resp = app.get(url, params={'apikey': 'sivinkey', 'siren': '000399634'}).json

    assert mocked_post.call_count == 2

    assert not resp['err']

    for item in resp['data']:

        assert 'id' in item

        assert 'text' in item

@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, EXPIRED_TOKEN))

def test_get_with_expired_token(mocked_post, app, conn):

    url = reverse(

        'generic-endpoint',

        kwargs={'connector': 'sivin', 'endpoint': 'consulterflotteparsiren', 'slug': conn.slug},

    )

    resp = app.get(url, params={'apikey': 'sivinkey', 'siren': '000399634'}).json

    assert mocked_post.call_count == 2

    assert resp['err']

    assert resp['err_desc'] == EXPIRED_TOKEN_MESSAGE

@mock.patch('passerelle.utils.Request.post', side_effect=ReadTimeout('timeout'))

def test_connection_timeout(mocked_post, app, conn):

    url = reverse(

        'generic-endpoint',

        kwargs={'connector': 'sivin', 'endpoint': 'consulterflotteparsiren', 'slug': conn.slug},

    )

    resp = app.get(url, params={'apikey': 'sivinkey', 'siren': '000399634'}).json

    assert mocked_post.call_count == 1

    assert resp['err']

    assert (

        resp['err_desc']

        == 'failed to call https://api.rec.sivin.fr/sivin/v2/consulterflotteparsiren: timeout'

    )