Produits Entr'ouvert » Passerelle

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

# Generated by Django 1.11.20 on 2019-06-19 10:24

from __future__ import unicode_literals

from django.db import migrations, models

import django.db.models.deletion

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0012_job'),

    ]

    operations = [

        migrations.CreateModel(

            name='AstreGS',

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

                ('wsdl_base_url', models.URLField(verbose_name='Webservices base URL')),

                ('username', models.CharField(max_length=32, verbose_name='Username')),

                ('password', models.CharField(max_length=32, verbose_name='Password')),

                ('organism', models.CharField(max_length=32, verbose_name='Organism')),

                ('budget', models.CharField(max_length=32, verbose_name='Budget')),

                ('exercice', models.CharField(max_length=32, verbose_name='Exercice')),

                ('users', models.ManyToManyField(blank=True, related_name='_astregs_users_+', related_query_name='+', to='base.ApiUser')),

            ],

            options={

                'verbose_name': 'AstresGS',

            },

        ),

        migrations.CreateModel(

            name='Link',

            fields=[

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

                ('name_id', models.CharField(max_length=32, verbose_name='User NameID')),

                ('association_id', models.CharField(max_length=32, verbose_name='Association ID')),

                ('created', models.DateTimeField(auto_now_add=True, verbose_name='Creation date')),

                ('resource', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='astregs.AstreGS')),

            ],

        ),

        migrations.CreateModel(

            name='LinkRequest',

            fields=[

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

                ('token', models.CharField(max_length=16, verbose_name='Temporary token')),

                ('association_id', models.CharField(max_length=32, verbose_name='Association ID')),

                ('resource', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='astregs.AstreGS')),

            ],

        ),

        migrations.AlterUniqueTogether(

            name='linkrequest',

            unique_together=set([('resource', 'token', 'association_id')]),

        ),

        migrations.AlterUniqueTogether(

            name='link',

            unique_together=set([('resource', 'name_id', 'association_id')]),

        ),

    ]

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

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

from django.db import models

from django.utils.translation import ugettext_lazy as _

from django.utils.crypto import get_random_string

from django.utils.text import slugify

from django.utils.six.moves.urllib import parse as urlparse

from django.http import Http404

from passerelle.base.models import BaseResource

from passerelle.utils.api import endpoint

from passerelle.utils.jsonresponse import APIError

LINK_CREATION_SCHEMA = {

    '$schema': 'http://json-schema.org/draft-03/schema#',

    'title': 'AstreGS Link',

    'description': '',

    'type': 'object',

    'properties': {

        'NameID': {

            'description': 'user name_id',

            'type': 'string',

            'required': True

        },

        'token': {

            'description': 'linking token',

            'type': 'string',

            'required': True

        }

    }

}

class AstreGS(BaseResource):

    wsdl_base_url = models.URLField(_('Webservices base URL'))

    username = models.CharField(_('Username'), max_length=32)

    password = models.CharField(_('Password'), max_length=32)

    organism = models.CharField(_('Organism'), max_length=32)

    budget = models.CharField(_('Budget'), max_length=32)

    exercice = models.CharField(_('Exercice'), max_length=32)

    category = _('Business Process Connectors')

    class Meta:

        verbose_name = u'AstresGS'

    @classmethod

    def get_verbose_name(cls):

        return cls._meta.verbose_name

    def check_status(self):

        response = self.requests.get(self.wsdl_base_url)

        response.raise_for_status()

    @property

    def authentication(self):

        return {'USERNOM': self.username, 'USERPWD': self.password}

    @property

    def context(self):

        return {'Organisme': self.organism,

                'Budget': self.budget,

                'Exercice': self.exercice}

    def get_client(self, wsdl_name):

        url = urlparse.urljoin(self.wsdl_base_url, '%s?wsdl' % wsdl_name)

        return self.soap_client(wsdl_url=url)

    @endpoint(description=_('Find associations by SIREN number'),

              perm='can_access',

              parameters={

                  'siren': {'description': _('SIREN Number'),

                            'example_value': '77567227216096'}

                  }

              )

    def associations(self, request, siren):

        client = self.get_client('RechercheTiersDetails')

        r = client.service.liste(Authentification=self.authentication,

                                 Contexte=self.context,

                                 Criteres={'siren': '%s*' % siren})

        data = []

        if r.liste:

            for item in r.liste.EnregRechercheTiersDetailsReturn:

                association_data = dict(inspect.getmembers(item))

                association_data['id'] = item.Numero_SIRET

                association_data['text'] = '%s - %s' % (item.Numero_SIRET, item.Nom_enregistrement)

                association_data['code'] = item.Code_tiers

                association_data['name'] = item.Nom_enregistrement

                association_data['address'] = item.Nom_enregistrement

                data.append(association_data)

        return {'data': data}

    @endpoint(description=_('Check if association exists by its SIRET number'),

              name='verify-association-presence-by-siret',

              perm='can_access',

              parameters={

                  'siret': {'description': _('SIRET Number'),

                            'example_value': '7756722721609600014'}

                  }

              )

    def verify_association_presence(self, request, siret):

        client = self.get_client('RechercheTiers')

        r = client.service.liste(Authentification=self.authentication,

                                 Contexte=self.context,

                                 Criteres={'siren': siret})

        if r.liste:

            return {'exists': True}

        return {'exists': False}

    @endpoint(name='get-association-link-means',

              description=_('Get association linking means'),

              perm='can_access',

              parameters={

                  'association_id': {'description': _('Association ID'),

                                     'example_value': '42435'}

                  }

    )

    def get_association_link_means(self, request, association_id):

        client = self.get_client('Tiers')

        r = client.service.Chargement({'Authentification': self.authentication,

                                       'Contexte': self.context,

                                       'TiersCle': {'CodeTiers': association_id}})

        data = []

        # assocation contact is defined in EncodeKeyContact attribute

        if not r.EncodeKeyContact:

            return {'data': data}

        client = self.get_client('Contact')

        r = client.service.Chargement({'Authentification': self.authentication,

                                       'Contexte': self.context,

                                       'ContactCle': {'idContact': r.EncodeKeyContact}})

        if r.AdresseMail or r.TelephoneMobile or r.RueVoie:

            try:

                link_request = LinkRequest.objects.get(resource=self, association_id=association_id)

            except LinkRequest.DoesNotExist:

                token = get_random_string()

                link_request = LinkRequest.objects.create(resource=self, association_id=association_id,

                                                          token=token)

        if r.AdresseMail:

            data.append({'id': '1',

                         'text': r.AdresseMail,

                         'value': r.AdresseMail,

                         'token': link_request.token,

                         'type': 'email'})

        if r.TelephoneMobile:

            data.append({'id': '2',

                         'text': r.TelephoneMobile,

                         'value': r.TelephoneMobile,

                         'token': link_request.token,

                         'type': 'mobile'})

        if r.RueVoie:

            full_address = r.RueVoie

            full_address += ', '

            if r.CodePostal:

                full_address += '%s ' % r.CodePostal

            if r.Ville:

                full_address += r.Ville

            data.append({'id': '3',

                         'text': full_address,

                         'address': r.RueVoie,

                         'zicode': r.CodePostal,

                         'city': r.Ville,

                         'value': full_address,

                         'token': link_request.token,

                         'type': 'mail'})

        return {'data': data}

    @endpoint(name='link', perm='can_access',

              post={

                  'description': _('Create link with an association'),

                  'request_body': {

                      'schema': {

                          'application/json': LINK_CREATION_SCHEMA

                      }

                  }

              }

    )

    def link(self, request, post_data):

        try:

            link_request = LinkRequest.objects.get(resource=self, token=post_data['token'])

            # if link request found, create link

            link = Link.objects.create(resource=self, name_id=post_data['NameID'], association_id=link_request.association_id)

            # then delete link request

            link_request.delete()

            return {'link': link.id, 'created': link.created, 'association_id': link_request.association_id}

        except LinkRequest.DoesNotExist:

            raise Http404('token not found')

    @endpoint(description=_('Remove link to an association'),

              perm='can_access',

              parameters={

                  'NameID':{

                      'description': _('Publik NameID'),

                      'example_value': 'xyz24d934',

                  },

                  'association_id':{

                      'description': _('Association ID'),

                      'example_value': '12345',

                  }

              })

    def unlink(self, request, NameID, association_id):

        try:

            link = Link.objects.get(resource=self, name_id=NameID, association_id=association_id)

            link.delete()

            return {'deleted': True}

        except Link.DoesNotExist:

            raise Http404('link not found')

    @endpoint(description=_('List user links to associations'),

              perm='can_access',

              parameters={

                  'NameID':{

                      'description': _('User NameID'),

                      'example_value': 'xyz24d934',

                  }

              })

    def links(self, request, NameID):

        data = []

        for link in Link.objects.filter(resource=self, name_id=NameID):

            data.append({'association_id': link.association_id,

                         'created': link.created})

        return {'data': data}

class Link(models.Model):

    resource = models.ForeignKey(AstreGS)

    name_id = models.CharField(_('User NameID'), max_length=32)

    association_id = models.CharField(_('Association ID'), max_length=32)

    created = models.DateTimeField(_('Creation date'), auto_now_add=True)

    class Meta:

        unique_together = ('resource', 'name_id', 'association_id')

class LinkRequest(models.Model):

    resource = models.ForeignKey(AstreGS)

    token = models.CharField(_('Temporary token'), max_length=16)

    association_id = models.CharField(_('Association ID'), max_length=32)

    class Meta:

        unique_together = ('resource', 'token', 'association_id')

    'passerelle.apps.astregs',

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

import mock

import pytest

from passerelle.apps.astregs.models import AstreGS, LinkRequest, Link

from . import utils

BASE_URL = 'https://test-ws-astre-gs.departement06.fr/axis2/services/'

class FakeSearchDetailsService():

    def liste(self, Authentification, Contexte, Criteres):

        class FakeObject():

            Numero_SIRET='50043390900016'

            Nom_enregistrement='ASSOCIATION OMNISPORTS DES MONTS D AZUR'

            Code_tiers='173957'

        liste = mock.Mock(EnregRechercheTiersDetailsReturn=[FakeObject()])

        return mock.Mock(liste=liste)

class FakeSearchService():

    def liste(self, Authentification, Contexte, Criteres):

        if Criteres['siren'] == 'unknown':

            return mock.Mock(liste=False)

        return mock.Mock(liste=True)

class FakeTiersService():

    def Chargement(self, data):

        if data['TiersCle']['CodeTiers'] == '42':

            return mock.Mock(EncodeKeyContact='4242')

        return mock.Mock(EncodeKeyContact=None)

class FakeContactService():

    def Chargement(self, data):

        return mock.Mock(AdresseMail='foo@example.com',

                         TelephoneMobile='0607080900',

                         RueVoie='169 rue du Château',

                         CodePostal='75014',

                         Ville='Paris')

def search_side_effect(wsdl_url, **kwargs):

    if 'RechercheTiersDetail' in wsdl_url:

        return mock.Mock(service=FakeSearchDetailsService())

    return mock.Mock(service=FakeSearchService())

def contact_search_side_effect(wsdl_url, **kwargs):

    if 'Tiers' in wsdl_url:

        return mock.Mock(service=FakeTiersService())

    return mock.Mock(service=FakeContactService())

@pytest.fixture

def connector(db):

    return utils.make_resource(AstreGS,

            title='Test', slug='test',

            description='test', wsdl_base_url=BASE_URL,

            username='CS-FORML', password='secret',

            organism='CG06', budget='01',

            exercice='2019'

    )

@mock.patch('passerelle.base.models.BaseResource.soap_client', side_effect=search_side_effect)

def test_search_association_by_siren(client, connector, app):

    resp = app.get('/astregs/test/associations', params={'siren': '500433909'})

    assert isinstance(resp.json['data'], list)

    assert len(resp.json['data']) > 0

    assert resp.json['data'][0]['id'] == '50043390900016'

    assert resp.json['data'][0]['text'] == '50043390900016 - ASSOCIATION OMNISPORTS DES MONTS D AZUR'

    assert resp.json['data'][0]['code'] == '173957'

    assert resp.json['data'][0]['name'] == 'ASSOCIATION OMNISPORTS DES MONTS D AZUR'

    assert client.call_args[1]['wsdl_url'] == '%sRechercheTiersDetails?wsdl' % BASE_URL

@mock.patch('passerelle.base.models.BaseResource.soap_client', side_effect=search_side_effect)

def test_check_association_presence(client, connector, app):

    resp = app.get('/astregs/test/verify-association-presence-by-siret', params={'siret': '50043390900014'})

    assert resp.json['exists'] == True

    assert client.call_args[1]['wsdl_url'] == '%sRechercheTiers?wsdl' % BASE_URL

    resp = app.get('/astregs/test/verify-association-presence-by-siret', params={'siret': 'unknown'})

    assert resp.json['exists'] == False

@mock.patch('passerelle.base.models.BaseResource.soap_client', side_effect=contact_search_side_effect)

def test_association_linking_means(client, connector, app):

    resp = app.get('/astregs/test/get-association-link-means', params={'association_id': '000'})

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

    assert LinkRequest.objects.count() == 0

    resp = app.get('/astregs/test/get-association-link-means', params={'association_id': '42'})

    assert len(resp.json['data']) == 3

    assert LinkRequest.objects.filter(association_id='42').count() == 1

    for result in resp.json['data']:

        assert 'id' in result

        assert 'text' in result

        assert 'type' in result

        assert 'token' in result

@mock.patch('passerelle.base.models.BaseResource.soap_client', side_effect=contact_search_side_effect)

def test_link_user_to_association(client, connector, app):

    assert LinkRequest.objects.count() == 0

    resp = app.get('/astregs/test/get-association-link-means', params={'association_id': '42'})

    assert len(resp.json['data']) == 3

    token = resp.json['data'][0]['token']

    assert LinkRequest.objects.filter(association_id='42').count() == 1

    resp = app.post_json('/astregs/test/link', params={'token': token, 'NameID': 'user_name_id'})

    assert LinkRequest.objects.filter(association_id='42').count() == 0

    assert Link.objects.filter(name_id='user_name_id', association_id='42').count() == 1

    link = Link.objects.get(name_id='user_name_id', association_id='42')

    assert resp.json['association_id'] == link.association_id

    assert resp.json['link'] == link.pk

    assert 'created' in resp.json

    # try to link again

    resp = app.post_json('/astregs/test/link', params={'token': token, 'NameID': 'user_name_id'}, status=404)

@mock.patch('passerelle.base.models.BaseResource.soap_client', side_effect=contact_search_side_effect)

def test_list_user_links(client, connector, app):

    resp = app.get('/astregs/test/get-association-link-means', params={'association_id': '42'})

    token = resp.json['data'][0]['token']

    resp = app.post_json('/astregs/test/link', params={'token': token, 'NameID': 'user_name_id'})

    resp = app.get('/astregs/test/links', params={'NameID': 'user_name_id'})

    assert len(resp.json['data']) > 0

    for link in resp.json['data']:

        assert 'association_id' in link

        assert 'created' in link

    resp = app.get('/astregs/test/links', params={'NameID': 'foo_name_id'})

    assert len(resp.json['data']) == 0

@mock.patch('passerelle.base.models.BaseResource.soap_client', side_effect=contact_search_side_effect)

def test_unlink_user_from_association(client, connector, app):

    resp = app.get('/astregs/test/get-association-link-means', params={'association_id': '42'})

    token = resp.json['data'][0]['token']

    resp = app.post_json('/astregs/test/link', params={'token': token, 'NameID': 'user_name_id'})

    resp = app.get('/astregs/test/unlink', params={'NameID': 'user_name_id', 'association_id': '42'})

    assert resp.json['deleted']

    resp = app.get('/astregs/test/unlink', params={'NameID': 'user_name_id', 'association_id': '42'}, status=404)