Produits Entr'ouvert » Passerelle

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

# Generated by Django 1.11.18 on 2019-05-24 10:25

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

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

                ('base_soap_url', models.URLField(help_text='URL of the base SOAP endpoint', max_length=400, verbose_name='Base SOAP endpoint')),

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

            ],

            options={

                'verbose_name': 'ATAL connector',

            },

        ),

    ]

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

from django.db import models

from django.utils.six.moves import urllib

from django.utils.translation import ugettext_lazy as _

import lxml.etree

from zeep import helpers

from passerelle.base.models import BaseResource

from passerelle.utils.api import endpoint

from passerelle.utils.jsonresponse import APIError

INSERT_DEMANDE_BY_TYPE_SCHEMA = {

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

    "title": "",

    "description": "",

    "type": "object",

    "properties": {

        "contact_nom": {

            "description": "Nom du contact",

            "required": True

        },

        "contact_tel": {

            "description": "Téléphone du contact",

            "type": "string",

        },

        "contact_email": {

            "description": "Email du contact",

            "type": "string",

        },

        "contact_adresse": {

            "description": "Adresse du contact",

            "type": "string",

        },

        "demande_objet": {

            "description": "Object de la demande",

            "type": "string",

        },

        "demande_lieu": {

            "description": "Lieu de la demande",

            "type": "string",

        },

        "demande_description": {

            "description": "Description de la demande",

            "type": "string",

        },

        "remote_adresse": {

            "description": "",

            "type": "string"

        },

        "code_equipement": {

            "description": "Code de l'équipement",

            "type": "string"

        },

        "code_service_demandeur": {

            "description": "Code du service demandeur",

            "type": "string"

        },

        "date_souhaite": {

            "description": "Date souhaitée",

            "type": "string"

        },

        "type_demande": {

            "description": "Type demande",

            "type": "string"

        }

    }

}

class ATALConnector(BaseResource):

    base_soap_url = models.URLField(

        max_length=400, verbose_name=_('Base SOAP endpoint'),

        help_text=_('URL of the base SOAP endpoint'))

    category = _('Business Process Connectors')

    class Meta:

        verbose_name = _('ATAL connector')

    def _soap_call(self, wsdl, method, **kwargs):

        wsdl_url = urllib.parse.urljoin(self.base_soap_url, '%s?wsdl' % wsdl)

        client = self.soap_client(wsdl_url=wsdl_url)

        return getattr(client.service, method)(**kwargs)

    def _basic_ref(self, wsdl, method):

        soap_res = self._soap_call(wsdl=wsdl, method=method)

        res = []

        for elem in soap_res:

            res.append({'id': elem.code, 'text': elem.libelle})

        return {'data': res}

    @endpoint(methods=['get'], perm='can_access')

    def get_type_activite(self, request):

        return self._basic_ref('VilleAgileService', 'getTypeActivite')

    @endpoint(methods=['get'], perm='can_access')

    def get_type_de_voie(self, request):

        return self._basic_ref('VilleAgileService', 'getTypeDeVoie')

    @endpoint(methods=['get'], perm='can_access')

    def get_types_equipement(self, request):

        soap_res = self._soap_call(wsdl='VilleAgileService', method='getTypesEquipement')

        tree = lxml.etree.fromstring(soap_res.encode('utf-8')).getroottree()

        types = tree.xpath('//types')[0]

        res = []

        for type_elem in types.getchildren():

            res.append({'id': type_elem.get('id'), 'text': type_elem.get('label')})

        return {'data': res}

    @endpoint(

        perm='can_access',

        post={

            'description': _('Insert Demande By Type'),

            'request_body': {

                'schema': {

                    'application/json': INSERT_DEMANDE_BY_TYPE_SCHEMA

                }

            }

        }

    )

    def insert_demande_by_type(self, request, post_data):

        demande_number = self._soap_call(

            wsdl='DemandeService', method='insertDemandeByType',

            contactNom=post_data['contact_nom'],

            contactTelephone=post_data['contact_telephone'],

            contactCourriel=post_data['contact_email'],

            contactAdresse=post_data['contact_adresse'], demandeObjet=post_data['demande_objet'],

            demandeLieu=post_data['demande_lieu'],

            demandeDescription=post_data['demande_description'],

            remoteAddress=post_data['remote_adresse'], codeEquipement=post_data['code_equipement'],

            codeServiceDemandeur=post_data['code_service_demandeur'],

            dateSouhaitee=post_data['date_souhaite'], typeDemande=post_data['type_demande']

        )

        return {'data': {'demande_number': demande_number}}

    @endpoint(

        methods=['get'], perm='can_access', example_pattern='{demande_number}/',

        pattern='^(?P<demande_number>\w+)/$',

        parameters={

            'demande_number': {

                'description': _('Demande number'), 'example_value': 'DIT18050001'

            }

        }

    )

    def retrieve_details_demande(self, request, demande_number, **kwargs):

        if not demande_number:

            raise APIError('A demande_number parameter must be specified')

        soap_res = self._soap_call(

            wsdl='DemandeService', method='retrieveDetailsDemande',

            demandeNumberParam=demande_number)

        return {'data': helpers.serialize_object(soap_res)}

    'passerelle.apps.atal',

from django.contrib.contenttypes.models import ContentType

import mock

import pytest

from passerelle.apps.atal.models import ATALConnector

from passerelle.base.models import ApiUser, AccessRight

@pytest.fixture()

def connector(db):

    api = ApiUser.objects.create(username='all', keytype='', key='')

    connector = ATALConnector.objects.create(

        base_soap_url='http://example.atal.com/', slug='slug-atal')

    obj_type = ContentType.objects.get_for_model(connector)

    AccessRight.objects.create(

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

    return connector

def mock_atal_soap_call(monkeypatch, return_value=None, side_effect=None):

    kwargs = {}

    if return_value is not None:

        kwargs['return_value'] = return_value

    if side_effect is not None:

        kwargs['side_effect'] = side_effect

    mock_soap_call = mock.Mock(**kwargs)

    monkeypatch.setattr(ATALConnector, '_soap_call', mock_soap_call)

    return mock_soap_call

class SoapElem(object):

    def __init__(self, **kwargs):

        for attr, value in kwargs.items():

            setattr(self, attr, value)

REFS = [

    SoapElem(code='code1', libelle='elem1'), SoapElem(code='code2', libelle='elem2')

]

def test_get_type_activite(app, connector, monkeypatch):

    mock_soap_call = mock_atal_soap_call(monkeypatch, return_value=REFS)

    response = app.get('/atal/slug-atal/get_type_activite')

    assert response.json == {

        'err': 0,

        'data': [

            {'text': 'elem1', 'id': 'code1'},

            {'text': 'elem2', 'id': 'code2'}]

    }

    call_params = mock_soap_call.call_args.kwargs

    assert call_params['wsdl'] == 'VilleAgileService'

    assert call_params['method'] == 'getTypeActivite'

def test_get_type_de_voie(app, connector, monkeypatch):

    mock_soap_call = mock_atal_soap_call(monkeypatch, return_value=REFS)

    response = app.get('/atal/slug-atal/get_type_de_voie')

    assert response.json == {

        'err': 0,

        'data': [

            {'text': 'elem1', 'id': 'code1'},

            {'text': 'elem2', 'id': 'code2'}]

    }

    call_params = mock_soap_call.call_args.kwargs

    assert call_params['wsdl'] == 'VilleAgileService'

    assert call_params['method'] == 'getTypeDeVoie'

def test_get_types_equipement(app, connector, monkeypatch):

    return_value = u"""<?xml version="1.0" encoding="UTF-8" standalone="no"?>

    <types>

        <type id="2" label="Espaces Verts"></type>

        <type id="4" label="Voirie">

    </type>

</types>

"""

    mock_soap_call = mock_atal_soap_call(monkeypatch, return_value=return_value)

    response = app.get('/atal/slug-atal/get_types_equipement')

    assert response.json == {

        'err': 0,

        'data': [

            {'text': 'Espaces Verts', 'id': '2'},

            {'text': 'Voirie', 'id': '4'}]

    }

    call_params = mock_soap_call.call_args.kwargs

    assert call_params['wsdl'] == 'VilleAgileService'

    assert call_params['method'] == 'getTypesEquipement'

def test_insert_demande_by_type(app, connector, monkeypatch):

    mock_soap_call = mock_atal_soap_call(monkeypatch, return_value='DIT19050001')

    params = {

        'contact_nom': 'John Doe',

        'contact_telephone': '0101010101',

        'contact_email': 'john@doe.com',

        'contact_adresse': '1 doe street',

        'demande_objet': 'sarah connor',

        'demande_lieu': 'LA',

        'demande_description': 'poker face',

        'remote_adresse': 'hollywood bd',

        'code_equipement': 'MAC10',

        'code_service_demandeur': 'skynet',

        'date_souhaite': 'now',

        'type_demande': 'scary'

    }

    response = app.post_json('/atal/slug-atal/insert_demande_by_type', params=params)

    assert response.json == {

        'err': 0,

        'data': {'demande_number': 'DIT19050001'}

    }

    call_params = mock_soap_call.call_args.kwargs

    assert call_params['wsdl'] == 'DemandeService'

    assert call_params['method'] == 'insertDemandeByType'

    assert call_params['contactNom'] == 'John Doe'

    assert call_params['contactTelephone'] == '0101010101'

    assert call_params['contactCourriel'] == 'john@doe.com'

    assert call_params['contactAdresse'] == '1 doe street'

    assert call_params['demandeObjet'] == 'sarah connor'

    assert call_params['demandeLieu'] == 'LA'

    assert call_params['demandeDescription'] == 'poker face'

    assert call_params['remoteAddress'] == 'hollywood bd'

    assert call_params['codeEquipement'] == 'MAC10'

    assert call_params['codeServiceDemandeur'] == 'skynet'

    assert call_params['dateSouhaitee'] == 'now'

    assert call_params['typeDemande'] == 'scary'

def test_retrieve_details_demande(app, connector, monkeypatch):

    mock_soap_call = mock_atal_soap_call(

        monkeypatch, return_value=dict(code='code1', libelle='elem1'))

    response = app.get('/atal/slug-atal/retrieve_details_demande/DIT19050001/')

    assert response.json == {

        'err': 0,

        'data': {'code': 'code1', 'libelle': 'elem1'}

    }

    call_params = mock_soap_call.call_args.kwargs

    assert call_params['wsdl'] == 'DemandeService'

    assert call_params['method'] == 'retrieveDetailsDemande'

    assert call_params['demandeNumberParam'] == 'DIT19050001'