start atal connector (#33348)
This commit is contained in:
parent
61bb06b143
commit
1554c5c634
|
|
@ -0,0 +1,31 @@
|
|||
# -*- 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',
|
||||
},
|
||||
),
|
||||
]
|
||||
|
|
@ -0,0 +1,169 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# passerelle - uniform access to multiple data sources and services
|
||||
# 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": "Objet 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)}
|
||||
|
|
@ -123,6 +123,7 @@ INSTALLED_APPS = (
|
|||
'passerelle.apps.api_particulier',
|
||||
'passerelle.apps.arcgis',
|
||||
'passerelle.apps.arpege_ecp',
|
||||
'passerelle.apps.atal',
|
||||
'passerelle.apps.atos_genesys',
|
||||
'passerelle.apps.base_adresse',
|
||||
'passerelle.apps.bdp',
|
||||
|
|
|
|||
|
|
@ -0,0 +1,141 @@
|
|||
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'
|
||||
Loading…
Reference in New Issue