Produits Entr'ouvert » Passerelle

recursive-include passerelle *.html *.mako *.txt *.xsd README

    python-pil,

    python-jsonschema,

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

from __future__ import unicode_literals

from django.db import migrations, models

class Migration(migrations.Migration):

    dependencies = [

        ('base', '0007_auto_20180619_1456'),

    ]

    operations = [

        migrations.CreateModel(

            name='IWSConnector',

            fields=[

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

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

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

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

                ('log_level', models.CharField(default=b'INFO', 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')])),

                ('wsdl_endpoint', models.URLField(help_text='URL of the SOAP wsdl endpoint', max_length=400, verbose_name='SOAP wsdl endpoint')),

                ('operation_endpoint', models.URLField(help_text='URL of SOAP operation endpoint', max_length=400, verbose_name='SOAP operation endpoint')),

                ('username', models.CharField(max_length=128, verbose_name='Service username')),

                ('password', models.CharField(max_length=128, null=True, verbose_name='Service password', blank=True)),

                ('database', models.CharField(max_length=128, verbose_name='Service database')),

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

            ],

            options={

                'verbose_name': 'IWS connector',

            },

        ),

    ]

# passerelle.contrib.iws

# Copyright (C) 2016  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 datetime import datetime

import json

from django.db import models

from django.utils import dateformat

from django.utils.translation import ugettext_lazy as _

from jsonschema import validate, ValidationError

import lxml.etree

import pkg_resources

from suds.client import Client

from passerelle.base.models import BaseResource

from passerelle.utils.api import endpoint

from passerelle.utils.jsonresponse import APIError

CODE_EQUIPE = {"DECHET": "DMT", "ENCOMBRANT": "VPVIGIE"}

BOOKDATE_SCHEMA = {

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

    "title": "IWS",

    "description": "",

    "type": "object",

    "properties": {

        "firstname": {

            "description": "Firstname",

            "type": "string",

            "required": True

        },

        "lastname": {

            "description": "Lastname",

            "type": "string",

            "required": True

        },

        "email": {

            "description": "Email",

            "type": "string",

            "required": True

        },

        "description": {

            "description": "Description of the request",

            "type": "string",

        },

        "tel_number": {

            "description": "Telephone number",

            "type": "string",

        },

        "date": {

            "description": "Booking date",

            "type": "string",

            "required": True

        },

        "token": {

            "description": "Booking token",

            "type": "string",

            "required": True

        }

    }

}

def create_client(wsdl_endpoint, operation_endpoint, username, password, database):

    client = Client(wsdl_endpoint, location=operation_endpoint)

    auth_header = client.factory.create('IsiWsAuthHeader')

    auth_header.IsiLogin = username

    auth_header.IsiPassword = password

    auth_header.IsiDataBaseID = database

    client.set_options(soapheaders=auth_header)

    return client

def soap_call(logger, client, iws_data, method):

    logger.debug("sending to iws : %s", iws_data)

    IsiWsEntity = client.factory.create('IsiWsEntity')

    ArrayOfIsiWsDataField = client.factory.create('ArrayOfIsiWsDataField')

    for field, value in iws_data.items():

        IsiWsDataField = client.factory.create('IsiWsDataField')

        IsiWsDataField.IsiField = field

        IsiWsDataField.IsiValue = value

        ArrayOfIsiWsDataField.IsiWsDataField.append(IsiWsDataField)

    IsiWsEntity.IsiFields = ArrayOfIsiWsDataField

    iws_res = getattr(client.service, method)(IsiWsEntity)

    schema_root = lxml.etree.parse(pkg_resources.resource_stream(

        'passerelle.contrib.iws', 'xsd/ReponseWS.xsd'))

    schema = lxml.etree.XMLSchema(schema_root)

    parser = lxml.etree.XMLParser(schema=schema, encoding='utf-8')

    try:

        tree = lxml.etree.fromstring(iws_res.encode('utf-8'), parser).getroottree()

    except lxml.etree.XMLSyntaxError:

        raise APIError("IWS response is not valid")

    result = {

        "status": tree.find('//Statut').text,

        "trace": tree.find('//Trace').text

    }

    fields = {}

    for data_field in tree.xpath('//IsiWsDataField'):

        fields[data_field.find('IsiField').text] = data_field.find('IsiValue').text

    result['fields'] = fields

    logger.debug("recieved from iws : %s", result)

    return result

class IWSConnector(BaseResource):

    wsdl_endpoint = models.URLField(

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

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

    operation_endpoint = models.URLField(

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

        help_text=_('URL of SOAP operation endpoint'))

    username = models.CharField(max_length=128, verbose_name=_('Service username'))

    password = models.CharField(

        max_length=128, verbose_name=_('Service password'), null=True, blank=True)

    database = models.CharField(max_length=128, verbose_name=_('Service database'))

    category = _('Business Process Connectors')

    class Meta:

        verbose_name = _('IWS connector')

    def _create_client(self):

        return create_client(

            self.wsdl_endpoint, self.operation_endpoint, self.username, self.password,

            self.database)

    @endpoint(

        methods=['get'], perm='can_access', example_pattern='{sti_code}/{request_type}/{volume}/',

        pattern='^(?P<sti_code>[0-9]{16})/(?P<request_type>\w+)/(?P<volume>[0-9]+)/$',

        parameters={

            'sti_code': {

                'description': _('Adrress STI code'), 'example_value': '3155570464130003'

            },

            'request_type': {

                'description': _('DECHET or ENCOMBRANT'),

                'example_value': 'DECHET'

            },

            'volume': {

                'description': _('Volume of waste'),

                'example_value': '1'

            },

            'city': {

                'description': _('City'),

                'example_value': 'TOULOUSE'

            },

        }

    )

    def checkdate(self, request, sti_code, request_type, volume, city):

        if request_type not in ('DECHET', 'ENCOMBRANT'):

            raise APIError("request_type should be 'DECHET' or 'ENCOMBRANT'")

        iws_data = {

            'C_ORIGINE': 'TELESERVICE',

            'I_APP_TYPEDEM': request_type,

            'I_AP_QTEAGENDA': '1' if request_type == 'DECHET' else volume,

            'C_STAPPEL': 'E',

            'C_NATURE': 'INC',

            'DE_SYMPAPPEL': 'booking description',

            'I_AP_COMMUNE': city,

            'I_AP_COMMUNEINTER': sti_code,

            'J_PRJACTPREV': '5',

            'C_EQUIPE': CODE_EQUIPE[request_type],

            'I_APP_DEMANDEUR': 'booking, demandeur',

            'I_AP_ADRESSEMAIL': 'booking@localhost'

        }

        iws_res = soap_call(self.logger, self._create_client(), iws_data, 'IsiAddAndGetCall')

        self._check_status(iws_res)

        iws_fields = iws_res['fields']

        token = iws_fields['NO_APPEL']

        if not token:

            raise APIError('iws error, missing token')

        dates = []

        result = {'data': dates}

        iws_dates = iws_fields['I_APP_DATESPOSSIBLES']

        if iws_dates == 'Aucune dates disponibles':

            return result

        for raw_date in iws_dates.split(';'):

            if raw_date:

                raw_date = raw_date.strip()

                date_obj = datetime.strptime(raw_date, '%d/%m/%Y').date()

                date_text = dateformat.format(date_obj, 'l d F Y')

                dates.append({"id": raw_date, "text": date_text, "token": token})

        return result

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

    def bookdate(self, request):

        error, error_msg, data = self._validate_inputs(request.body, BOOKDATE_SCHEMA)

        if error:

            self.logger.debug("received invalid data: %s", error_msg)

            raise APIError(error_msg, http_status=400)

        iws_data = {

            'NO_APPEL': data['token'],

            'I_APP_DEMANDEUR': '%s, %s' % (data['lastname'], data['firstname']),

            'I_AP_TEL_DEMANDEU': data['tel_number'] or '',

            'I_AP_ADRESSEMAIL': data['email'],

            'I_AP_DATRDVAGENDA': data['date'],

            'C_STAPPEL': 'E',

            'I_AP_SERVICE': 'TELESERVICE',

            'I_AP_SOURCE': 'TELESERVICE',

            'C_ORIGINE': 'TELESERVICE',

            'C_BLOCAGE': 2,

            'I_AP_EMAIL': 'OUI',

            'I_AP_CNIL': 1,

            'I_AP_SMS': 'NON',

            'DE_SYMPAPPEL': data['description'] or '',

            'C_QUALIFICATIF': 'INTERVENTION',

        }

        iws_res = soap_call(self.logger, self._create_client(), iws_data, 'IsiUpdateAndGetCall')

        self._check_status(iws_res)

        return {'data': iws_res['fields']}

    def _check_status(self, iws_res):

        if iws_res['status'] != 'responseOk':

            raise APIError('iws error, status: "%(status)s", trace: "%(trace)s"' % iws_res)

    def _validate_inputs(self, body, schema):

        """ process JSON body

        return a tuple (error, error_msg, data)

        """

        try:

            data = json.loads(body)

        except ValueError as e:

            return True, "could not decode body to json: %s" % e, None

        try:

            validate(data, schema)

        except ValidationError as e:

            return True, e.message, None

        return False, '', data

{% extends "passerelle/manage/service_view.html" %}

{% load i18n passerelle %}

{% block extra-sections %}

<div class="section">

  <ul>

    <li>

      <h4>{% trans 'Book date' %}</h4>

      {% url "generic-endpoint" connector="iws" slug=object.slug endpoint="bookdate" as bookdate %}

      <p> <strong>POST</strong> <a href="{{bookdate}}">{{bookdate}}</a></p>

      <pre>

        data_send = {

          'firstname': 'jon'

          'lastname': 'doe'

          'email': 'jon.doe@jondoe.com'

          'description': 'a refrigerator',

          'tel_number': '0102030405',

          'date': '26/10/2018',

          'token': 'XBNDNFT34'

        }

      </pre>

    </li>

  </ul>

</div>

{% endblock %}

{% block security %}

<p>

{% trans 'Book date is limited to the following API users:' %}

</p>

{% access_rights_table resource=object permission='can_access' %}

{% endblock %}

<?xml version="1.0" encoding="utf-8" ?>

<xsd:schema xmlns:xsd="http://www.w3.org/2001/XMLSchema">

	<xsd:element name="IsiWsResponse">

		<xsd:complexType>

			<xsd:sequence>

				<xsd:element name="Statut" type="xsd:string"></xsd:element>

				<xsd:element name="Trace" type="xsd:string"></xsd:element>

				<xsd:element name="Objects" minOccurs="0">

					<xsd:complexType>

						<xsd:sequence>

							<xsd:element name="anyType" type="IsiWsEntity">

							</xsd:element>

						</xsd:sequence>

					</xsd:complexType>

				</xsd:element>

			</xsd:sequence>

		</xsd:complexType>

	</xsd:element>

	<xsd:complexType name="IsiWsEntity">

		<xsd:sequence>

			<xsd:element minOccurs="0" maxOccurs="1" name="IsiFields"

				type="ArrayOfIsiWsDataField" />

		</xsd:sequence>

	</xsd:complexType>

	<xsd:complexType name="ArrayOfIsiWsDataField">

		<xsd:sequence>

			<xsd:element minOccurs="0" maxOccurs="unbounded" name="IsiWsDataField"

				nillable="true" type="IsiWsDataField" />

		</xsd:sequence>

	</xsd:complexType>

	<xsd:complexType name="IsiWsDataField">

		<xsd:sequence>

			<xsd:element minOccurs="0" maxOccurs="1" name="IsiField"

				type="xsd:string" />

			<xsd:element minOccurs="0" maxOccurs="1" name="IsiValue"

				type="xsd:string" />

		</xsd:sequence>

	</xsd:complexType>

</xsd:schema>

            'python-magic',

            'jsonschema'

      package_data={'passerelle': ['*.xsd']}

        'passerelle.contrib.iws',

from django.contrib.contenttypes.models import ContentType

from mock import Mock

import pytest

from passerelle.base.models import ApiUser, AccessRight

from passerelle.contrib.iws.models import IWSConnector

@pytest.fixture()

def setup(db):

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

    conn = IWSConnector.objects.create(

        wsdl_endpoint='http://example.com/iws?wsdl',

        operation_endpoint='http://example.com/iws', username='admin', password='admin',

        database='somedb', slug='slug-iws')

    obj_type = ContentType.objects.get_for_model(conn)

    AccessRight.objects.create(

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

    return conn

def create_params(**kwargs):

    res = {

        "firstname": "John", "lastname": "Doe", "email": "john.doe@localhost",

        "description": "four : 1", "tel_number": "0101010101", "date": "28/10/2018",

        "token": "token"

    }

    res.update(kwargs)

    return res

def mock_soap_call(monkeypatch, return_value):

    mock_soap_call = Mock(return_value=return_value)

    mock_soap_client = Mock()

    import passerelle.contrib.iws.models

    monkeypatch.setattr(passerelle.contrib.iws.models, 'soap_call', mock_soap_call)

    monkeypatch.setattr(passerelle.contrib.iws.models, 'create_client', mock_soap_client)

    return mock_soap_call, mock_soap_client

def test_checkdate_dechet_or_encombrant(app, setup):

    response = app.get(

        '/iws/slug-iws/checkdate/3155570464130003/error/3/?city=toulouse', expect_errors=True)

    json_result = response.json_body

    assert json_result['err'] == 1

    assert u'DECHET' in json_result['err_desc']

    assert u'ENCOMBRANT' in json_result['err_desc']

def test_checkdate_iws_error_status(app, setup, monkeypatch):

    mock_soap_call(monkeypatch, {'status': 'KO', 'trace': 'some trace'})

    response = app.get(

        '/iws/slug-iws/checkdate/3155570464130003/DECHET/3/?city=toulouse', expect_errors=True)

    json_result = response.json_body

    assert json_result['err'] == 1

    assert json_result['err_desc'] == 'iws error, status: "KO", trace: "some trace"'

def test_checkdate_iws_error_no_appel(app, setup, monkeypatch):

    mock_soap_call(

        monkeypatch, {

            'status': 'responseOk', 'trace': '',

            'fields': {'NO_APPEL': ''}})

    response = app.get(

        '/iws/slug-iws/checkdate/3155570464130003/DECHET/3/?city=toulouse', expect_errors=True)

    json_result = response.json_body

    assert json_result['err'] == 1

    assert json_result['err_desc'] == 'iws error, missing token'

def test_checkdate_iws_no_dates(app, setup, monkeypatch):

    mock_soap_call(

        monkeypatch, {

            'status': 'responseOk', 'trace': '',

            'fields': {

                'NO_APPEL': 'sometoken',

                'I_APP_DATESPOSSIBLES': 'Aucune dates disponibles'

            }

        })

    response = app.get('/iws/slug-iws/checkdate/3155570464130003/DECHET/3/?city=toulouse')

    json_result = response.json_body

    assert json_result['err'] == 0

    assert json_result['data'] == []

def test_checkdate_iws_has_dates(app, setup, monkeypatch, settings):

    settings.LANGUAGE_CODE = 'fr-fr'

    mock_soap_call(

        monkeypatch, {

            'status': 'responseOk', 'trace': '',

            'fields': {

                'NO_APPEL': 'sometoken',

                'I_APP_DATESPOSSIBLES': '18/06/2018; 19/06/2018'

            }

        })

    response = app.get('/iws/slug-iws/checkdate/3155570464130003/DECHET/3/?city=toulouse')

    json_result = response.json_body

    assert json_result['err'] == 0

    dates = json_result['data']

    assert len(dates) == 2

    assert dates[0] == {"id": "18/06/2018", "text": "lundi 18 juin 2018", "token": "sometoken"}

    assert dates[1] == {"id": "19/06/2018", "text": "mardi 19 juin 2018", "token": "sometoken"}

def test_bookdate(app, setup, monkeypatch):

    mock_soap_call(

        monkeypatch, {

            'status': 'responseOk', 'trace': '',

            'fields': {

                'NO_APPEL': 'sometoken',

                'I_APP_DATESPOSSIBLES': '18/06/2018;'

            }

        })

    response = app.post_json('/iws/slug-iws/bookdate/', params=create_params())

    json_result = response.json_body

    assert json_result['err'] == 0

    assert json_result['data'] == {'NO_APPEL': 'sometoken', 'I_APP_DATESPOSSIBLES': '18/06/2018;'}