add astech connector (#52684)
This commit is contained in:
parent
0114f77618
commit
d3f0e5803c
|
|
@ -0,0 +1,82 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# Generated by Django 1.11.18 on 2021-04-08 22:32
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
initial = True
|
||||
|
||||
dependencies = [
|
||||
('base', '0029_auto_20210202_1627'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='ASTech',
|
||||
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')),
|
||||
(
|
||||
'basic_auth_username',
|
||||
models.CharField(
|
||||
blank=True, max_length=128, verbose_name='Basic authentication username'
|
||||
),
|
||||
),
|
||||
(
|
||||
'basic_auth_password',
|
||||
models.CharField(
|
||||
blank=True, max_length=128, verbose_name='Basic authentication password'
|
||||
),
|
||||
),
|
||||
(
|
||||
'client_certificate',
|
||||
models.FileField(
|
||||
blank=True, null=True, upload_to='', verbose_name='TLS client certificate'
|
||||
),
|
||||
),
|
||||
(
|
||||
'trusted_certificate_authorities',
|
||||
models.FileField(blank=True, null=True, upload_to='', verbose_name='TLS trusted CAs'),
|
||||
),
|
||||
('verify_cert', models.BooleanField(default=True, verbose_name='TLS verify certificates')),
|
||||
(
|
||||
'http_proxy',
|
||||
models.CharField(blank=True, max_length=128, verbose_name='HTTP and HTTPS proxy'),
|
||||
),
|
||||
(
|
||||
'base_url',
|
||||
models.URLField(
|
||||
help_text='Base API URL (example: https://astech-symphonie.com/app.php/)',
|
||||
verbose_name='Webservice Base URL',
|
||||
),
|
||||
),
|
||||
(
|
||||
'connection',
|
||||
models.CharField(
|
||||
blank=True,
|
||||
help_text='See possibilities with "connections" endpoint. If empty, default connection is used',
|
||||
max_length=64,
|
||||
null=True,
|
||||
verbose_name='Connection code',
|
||||
),
|
||||
),
|
||||
(
|
||||
'users',
|
||||
models.ManyToManyField(
|
||||
blank=True, related_name='_astech_users_+', related_query_name='+', to='base.ApiUser'
|
||||
),
|
||||
),
|
||||
],
|
||||
options={
|
||||
'verbose_name': 'ASTech Symphonie API',
|
||||
},
|
||||
),
|
||||
]
|
||||
|
|
@ -0,0 +1,444 @@
|
|||
# passerelle - uniform access to multiple data sources and services
|
||||
# Copyright (C) 2021 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 base64
|
||||
from collections import OrderedDict
|
||||
from requests.exceptions import ConnectionError
|
||||
|
||||
from django.core.cache import cache
|
||||
from django.db import models
|
||||
from django.utils.six.moves.urllib import parse as urlparse
|
||||
from django.utils.translation import ugettext_lazy as _
|
||||
|
||||
from passerelle.utils.jsonresponse import APIError
|
||||
from passerelle.utils.api import endpoint
|
||||
from passerelle.base.models import BaseResource, HTTPResource
|
||||
|
||||
|
||||
DEMAND_SCHEMA = {
|
||||
'$schema': 'http://json-schema.org/draft-04/schema#',
|
||||
'title': 'Intervention demand',
|
||||
'description': '',
|
||||
'type': 'object',
|
||||
'required': ['company', 'service', 'subject', 'name'],
|
||||
'properties': OrderedDict(
|
||||
{
|
||||
'company': {
|
||||
'description': 'Company code (sgesdemSoc)',
|
||||
'type': 'string',
|
||||
},
|
||||
'service': {
|
||||
'description': 'Service code (sgesdemSserv)',
|
||||
'type': 'string',
|
||||
},
|
||||
'label': {
|
||||
'description': 'Defined label code, required if LIBELDEMDEF=O (sgesdemLibdef)',
|
||||
'type': 'string',
|
||||
},
|
||||
'subject': {
|
||||
'description': 'Subject. Use defined label text if LIBELDEMDEF=O (sgesdemNdt)',
|
||||
'type': 'string',
|
||||
},
|
||||
'description': {
|
||||
'description': 'Description, up to 4000 chars (sgesdemLibelle)',
|
||||
'type': 'string',
|
||||
},
|
||||
'name': {
|
||||
'description': 'Applicant first name and last name (sgesdemCplnom)',
|
||||
'type': 'string',
|
||||
},
|
||||
'email': {
|
||||
'description': 'Email address (sgesdemCplemail)',
|
||||
'type': 'string',
|
||||
},
|
||||
'phone1': {
|
||||
'description': 'Phone 1 (sgesdemCpltel1)',
|
||||
'type': 'string',
|
||||
},
|
||||
'phone2': {
|
||||
'description': 'Phone 2 (sgesdemCpltel2)',
|
||||
'type': 'string',
|
||||
},
|
||||
'address1': {
|
||||
'description': 'Address, first line (sgesdemCpladr1)',
|
||||
'type': 'string',
|
||||
},
|
||||
'address2': {
|
||||
'description': 'Address, second line (sgesdemCpladr2)',
|
||||
'type': 'string',
|
||||
},
|
||||
'address3': {
|
||||
'description': 'Address, third line (sgesdemCpladr2)',
|
||||
'type': 'string',
|
||||
},
|
||||
}
|
||||
),
|
||||
}
|
||||
|
||||
ADD_DOCUMENT_SCHEMA = {
|
||||
'$schema': 'http://json-schema.org/draft-04/schema#',
|
||||
'title': 'Intervention demand',
|
||||
'description': '',
|
||||
'type': 'object',
|
||||
'required': ['demand_id', 'document', 'title'],
|
||||
'properties': OrderedDict(
|
||||
{
|
||||
'demand_id': {
|
||||
'description': 'Document title (docTitre)',
|
||||
'type': 'string',
|
||||
},
|
||||
'title': {
|
||||
'description': 'Document title (docTitre)',
|
||||
'type': 'string',
|
||||
},
|
||||
'filename': {
|
||||
'description': 'Document filename (docFile)',
|
||||
'type': 'string',
|
||||
},
|
||||
'document': {
|
||||
'type': 'object',
|
||||
'description': 'File object (file0)',
|
||||
'required': ['filename', 'content_type', 'content'],
|
||||
'properties': {
|
||||
'filename': {
|
||||
'type': 'string',
|
||||
},
|
||||
'content_type': {
|
||||
'type': 'string',
|
||||
'description': 'MIME content-type',
|
||||
},
|
||||
'content': {
|
||||
'type': 'string',
|
||||
'description': 'Content, base64 encoded',
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
class ASTech(BaseResource, HTTPResource):
|
||||
category = _('Business Process Connectors')
|
||||
|
||||
base_url = models.URLField(
|
||||
verbose_name=_('Webservice Base URL'),
|
||||
help_text=_('Base API URL (example: https://astech-symphonie.com/app.php/)'),
|
||||
)
|
||||
connection = models.CharField(
|
||||
verbose_name=_('Connection code'),
|
||||
help_text=_('See possibilities with "connections" endpoint. If empty, default code is used'),
|
||||
max_length=64,
|
||||
blank=True,
|
||||
null=True,
|
||||
)
|
||||
|
||||
_category_ordering = [_('Parameters'), _('Rules'), _('Demand'), 'Tech & Debug']
|
||||
|
||||
class Meta:
|
||||
verbose_name = _('AS-TECH')
|
||||
|
||||
def call_json(self, method, url, **kwargs):
|
||||
try:
|
||||
response = self.requests.request(method, url, timeout=25, **kwargs)
|
||||
except ConnectionError as e:
|
||||
raise APIError('connection error: %s' % e)
|
||||
if response.status_code not in (200, 201):
|
||||
try:
|
||||
content = response.json()
|
||||
except ValueError:
|
||||
content = response.content[:1024]
|
||||
raise APIError(
|
||||
'AS-TECH response: %s %s' % (response.status_code, response.reason),
|
||||
data={
|
||||
'error': {
|
||||
'status': response.status_code,
|
||||
'reason': response.reason,
|
||||
'content': content,
|
||||
}
|
||||
},
|
||||
)
|
||||
try:
|
||||
json_data = response.json()
|
||||
except ValueError:
|
||||
raise APIError('invalid JSON in response: %r' % response.content[:1024])
|
||||
return json_data
|
||||
|
||||
def get_connections(self):
|
||||
url = urlparse.urljoin(self.base_url, 'connection/all')
|
||||
connections = self.call_json('get', url)
|
||||
if self.connection:
|
||||
if self.connection not in connections['connections']:
|
||||
raise APIError(
|
||||
'connection %s not found in available connections: %r' % (self.connection, connections)
|
||||
)
|
||||
connections['default'] = self.connection
|
||||
connections['connection'] = connections['connections'][connections['default']]
|
||||
return connections
|
||||
|
||||
def get_authorization(self):
|
||||
cache_key = 'astech-%s-%s-authorization' % (self.connection or 'default', self.id)
|
||||
authorization = cache.get(cache_key)
|
||||
if authorization:
|
||||
return authorization
|
||||
connection = self.get_connections()['connection']
|
||||
params = {
|
||||
'client_id': connection['clientId'],
|
||||
'connection_id': connection['code'],
|
||||
'format': 'json',
|
||||
'response_type': 'token',
|
||||
}
|
||||
url = urlparse.urljoin(self.base_url, 'oauth/v2/auth')
|
||||
authorization = self.call_json('post', url, params=params, json={'application': 'interfaceCitoyenne'})
|
||||
timeout = min(120, authorization['expires_in']) # do not trust expires_in delay (1 day in tests)
|
||||
cache.set(cache_key, authorization, timeout)
|
||||
return authorization
|
||||
|
||||
def call(self, endpoint, **kwargs):
|
||||
url = urlparse.urljoin(self.base_url, endpoint)
|
||||
params = kwargs.pop('params', {})
|
||||
json = kwargs.pop('json', None)
|
||||
method = kwargs.pop('method', 'get')
|
||||
authorization = self.get_authorization()
|
||||
params['access_token'] = authorization['access_token']
|
||||
params['connection_id'] = authorization['connection_id']
|
||||
if json is not None:
|
||||
json_response = self.call_json('post', url, params=params, json=json, **kwargs)
|
||||
else:
|
||||
json_response = self.call_json(method, url, params=params, **kwargs)
|
||||
return json_response
|
||||
|
||||
@endpoint(
|
||||
name='connections',
|
||||
description=_('See all possible connections codes (see configuration)'),
|
||||
perm='can_access',
|
||||
display_category='Tech & Debug',
|
||||
display_order=1,
|
||||
)
|
||||
def connections(self, request):
|
||||
return {'data': self.get_connections()}
|
||||
|
||||
@endpoint(
|
||||
name='authorization',
|
||||
description=_('See authorization tokens (testing only)'),
|
||||
perm='can_access',
|
||||
display_category='Tech & Debug',
|
||||
display_order=2,
|
||||
)
|
||||
def authorization(self, request):
|
||||
return {'data': self.get_authorization()}
|
||||
|
||||
@endpoint(
|
||||
name='services',
|
||||
description=_("List authorized services for connected user"),
|
||||
perm='can_access',
|
||||
display_category=_('Rules'),
|
||||
display_order=1,
|
||||
)
|
||||
def services(self, request):
|
||||
services = self.call('apicli/rule-call-by-alias/sousservices/invoke', method='post')
|
||||
services = [{'id': key, 'text': value} for key, value in services.items()]
|
||||
services.sort(key=lambda item: item['id']) # "same as output" sort
|
||||
return {'data': services}
|
||||
|
||||
@endpoint(
|
||||
name='company',
|
||||
description=_('Company code of the applicant'),
|
||||
perm='can_access',
|
||||
parameters={
|
||||
'applicant': {
|
||||
'description': _(
|
||||
'Applicant code (codeDemandeur). If absent, use HTTP basic authentication username'
|
||||
)
|
||||
},
|
||||
},
|
||||
display_category=_('Rules'),
|
||||
display_order=2,
|
||||
)
|
||||
def company(self, request, applicant=None):
|
||||
if applicant is None:
|
||||
applicant = self.basic_auth_username
|
||||
company = self.call(
|
||||
'apicli/rule-call-by-alias/code_societe_demandeur/invoke', json={'codeDemandeur': applicant}
|
||||
)
|
||||
return {'data': {'company': company}}
|
||||
|
||||
@endpoint(
|
||||
name='companies',
|
||||
description=_('List of authorized companies for an applicant'),
|
||||
perm='can_access',
|
||||
parameters={
|
||||
'applicant': {
|
||||
'description': _(
|
||||
'Applicant code (codeDemandeur). If absent, use HTTP basic authentication username'
|
||||
)
|
||||
},
|
||||
'signatory': {'description': _('Signatory code (codeSignataire), supersede applicant')},
|
||||
'kind': {
|
||||
'description': _('Kind(s) of request: 1=parts, 2=interventions, 3=loans'),
|
||||
'example_value': '1,3',
|
||||
},
|
||||
},
|
||||
display_category=_('Rules'),
|
||||
display_order=3,
|
||||
)
|
||||
def companies(self, request, applicant=None, signatory=None, kind='1,2,3'):
|
||||
if applicant is None and signatory is None:
|
||||
applicant = self.basic_auth_username
|
||||
kind = ','.join([item.strip() for item in kind.split(',') if item.strip()])
|
||||
payload = {
|
||||
'codeDemandeur': applicant or '',
|
||||
'codeSignataire': signatory or '',
|
||||
'typeDemande': kind,
|
||||
'designation': True,
|
||||
}
|
||||
companies = self.call('apicli/rule-call-by-alias/societes_demandeur/invoke', json=payload)
|
||||
companies = [{'id': str(key), 'text': value} for key, value in companies.items()]
|
||||
companies.sort(key=lambda item: item['id']) # "same as output" sort
|
||||
return {'data': companies}
|
||||
|
||||
@endpoint(
|
||||
name='labels',
|
||||
description=_('List of predefined labels for a company'),
|
||||
perm='can_access',
|
||||
parameters={
|
||||
'company': {
|
||||
'description': _('Company code (societeDemandeur). If absent, use "company" endpoint result')
|
||||
},
|
||||
},
|
||||
display_category=_('Rules'),
|
||||
display_order=4,
|
||||
)
|
||||
def labels(self, request, company=None):
|
||||
if company is None:
|
||||
company = self.company(request)['data']['company']
|
||||
labels = self.call(
|
||||
'apicli/rule-call-by-alias/libelles_predefinis/invoke', json={'societeDemandeur': company}
|
||||
)
|
||||
labels = [{'id': str(key), 'text': value} for key, value in labels.items()]
|
||||
labels.sort(key=lambda item: item['id']) # "same as output" sort
|
||||
return {'data': labels}
|
||||
|
||||
@endpoint(
|
||||
name='parameter',
|
||||
description=_("Value of a parameter"),
|
||||
perm='can_access',
|
||||
parameters={
|
||||
'name': {'description': _('Name of the parameter'), 'example_value': 'LIBELDEMDEF'},
|
||||
'company': {'description': _('Company code. If absent, use "company" endpoint result')},
|
||||
},
|
||||
display_category=_('Parameters'),
|
||||
display_order=5,
|
||||
)
|
||||
def parameter(self, request, name, company=None):
|
||||
if company is None:
|
||||
company = self.company(request)['data']['company']
|
||||
endpoint = 'apicli/common/getparam/' + name
|
||||
if company:
|
||||
endpoint += '/' + company
|
||||
value = self.call(endpoint)
|
||||
return {'data': value}
|
||||
|
||||
@endpoint(
|
||||
name='create-demand',
|
||||
description=_('Create a demand'),
|
||||
perm='can_access',
|
||||
methods=['post'],
|
||||
post={'request_body': {'schema': {'application/json': DEMAND_SCHEMA}}},
|
||||
display_category=_('Demand'),
|
||||
display_order=1,
|
||||
)
|
||||
def create_demand(self, request, post_data):
|
||||
payload = {
|
||||
'interface_citoyenne_demande': {
|
||||
'sgesdemSoc': post_data['company'],
|
||||
'sgesdemSserv': post_data['service'],
|
||||
'sgesdemLibdef': post_data.get('label') or '',
|
||||
'sgesdemNdt': post_data['subject'],
|
||||
'sgesdemLibelle': post_data.get('description') or '',
|
||||
'sgesdemCplnom': post_data['name'] or '',
|
||||
'sgesdemCplemail': post_data.get('email') or '',
|
||||
'sgesdemCpltel1': post_data.get('phone1') or '',
|
||||
'sgesdemCpltel2': post_data.get('phone2') or '',
|
||||
'sgesdemCpladr1': post_data.get('address1') or '',
|
||||
'sgesdemCpladr2': post_data.get('address2') or '',
|
||||
'sgesdemCpladr3': post_data.get('address3') or '',
|
||||
}
|
||||
}
|
||||
create = self.call('apicli/interface-citoyenne/demande-intervention', json=payload)
|
||||
if not isinstance(create, dict) or not create.get('sgesdemNum'):
|
||||
raise APIError('no sgesdemNum in response: %s' % create)
|
||||
create['demand_id'] = create['sgesdemNum']
|
||||
return {'data': create}
|
||||
|
||||
@endpoint(
|
||||
name='add-document',
|
||||
description=_('Add a document in a demand'),
|
||||
perm='can_access',
|
||||
methods=['post'],
|
||||
post={'request_body': {'schema': {'application/json': ADD_DOCUMENT_SCHEMA}}},
|
||||
display_category=_('Demand'),
|
||||
display_order=2,
|
||||
)
|
||||
def add_document(self, request, post_data):
|
||||
endpoint = 'apicli/interface-citoyenne/document/sgesdemNum/' + post_data['demand_id']
|
||||
filename = post_data.get('filename') or post_data['document']['filename']
|
||||
params = {
|
||||
'docTitre': post_data['title'],
|
||||
'docFile': filename,
|
||||
}
|
||||
content = base64.b64decode(post_data['document']['content'])
|
||||
content_type = post_data['document']['content_type']
|
||||
files = {'file0': (filename, content, content_type)}
|
||||
added = self.call(endpoint, method='post', params=params, files=files)
|
||||
return {'data': added}
|
||||
|
||||
@endpoint(
|
||||
name='demand-position',
|
||||
description=_('Get demand position'),
|
||||
perm='can_access',
|
||||
parameters={
|
||||
'demand_id': {
|
||||
'description': _('Demand id'),
|
||||
'example_value': '000000000001234',
|
||||
},
|
||||
},
|
||||
display_category=_('Demand'),
|
||||
display_order=3,
|
||||
)
|
||||
def demand_position(self, request, demand_id):
|
||||
endpoint = 'apicli/demande/position/' + demand_id
|
||||
position = self.call(endpoint)
|
||||
if not isinstance(position, dict) or not position.get('position'):
|
||||
raise APIError('no position in response: %s' % position)
|
||||
position['id'] = position['position']
|
||||
position['text'] = position.get('positionLib')
|
||||
return {'data': position}
|
||||
|
||||
@endpoint(
|
||||
name='demand-all-positions',
|
||||
description=_('List all demand possible positions'),
|
||||
perm='can_access',
|
||||
display_category=_('Demand'),
|
||||
display_order=4,
|
||||
)
|
||||
def demand_all_positions(self, request):
|
||||
endpoint = 'apicli/demande/positions'
|
||||
positions = self.call(endpoint)
|
||||
for position in positions:
|
||||
position['id'] = position['position']
|
||||
position['text'] = position['positionLib']
|
||||
return {'data': positions}
|
||||
|
|
@ -124,6 +124,7 @@ INSTALLED_APPS = (
|
|||
'passerelle.apps.api_particulier',
|
||||
'passerelle.apps.arcgis',
|
||||
'passerelle.apps.arpege_ecp',
|
||||
'passerelle.apps.astech',
|
||||
'passerelle.apps.astregs',
|
||||
'passerelle.apps.atal',
|
||||
'passerelle.apps.atos_genesys',
|
||||
|
|
|
|||
|
|
@ -0,0 +1,422 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# Passerelle - uniform access to data and services
|
||||
# Copyright (C) 2021 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; exclude 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.deepcopy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
|
||||
import mock
|
||||
import pytest
|
||||
from requests.exceptions import ConnectionError
|
||||
from requests import Request
|
||||
|
||||
|
||||
import utils
|
||||
|
||||
from django.urls import reverse
|
||||
|
||||
from passerelle.apps.astech.models import ASTech
|
||||
from passerelle.utils.jsonresponse import APIError
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def setup(db):
|
||||
return utils.setup_access_rights(
|
||||
ASTech.objects.create(
|
||||
slug='test',
|
||||
base_url='https://astech.example.net/app.php/',
|
||||
basic_auth_username='ENTROUV',
|
||||
basic_auth_password='password',
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
CONNECTION_RESPONSE = """
|
||||
{"default":"TEST","connections":{"TEST":{"name":"TEST","code":"TEST","clientId":"1234","secret":"9876"}}}
|
||||
"""
|
||||
|
||||
AUTH_RESPONSE = """
|
||||
{"access_token":"4242","expires_in":604800,"token_type":"bearer","scope":null,"refresh_token":"4343","database_version":"6.06.210","gamme_nomade":null,"connection_id":"TEST","connection_name":"TEST","user_detail":"Demande Publik","mercure_token":"eyJ0-BsWy","mercure_topics":null,"mercure_base_url":"/.well-known/mercure","module_options":null,"astech_serveur":"6.06.218.00.00"}
|
||||
"""
|
||||
|
||||
SERVICES_RESPONSE = """
|
||||
{"ABC":"ABC / AH BE CE","123":"123 / FOOBAR","XFO":"XFO / BARFOO"}
|
||||
"""
|
||||
|
||||
COMPANY_RESPONSE = '"99"'
|
||||
|
||||
COMPANIES_RESPONSE = """
|
||||
{"01":"01 / SERVICES TECHNIQUES","10":"10 / DIRECTION BATIMENT","11":"11 / PLOMBERIE"}
|
||||
"""
|
||||
|
||||
LABELS_RESPONSE = """
|
||||
{"1":"1 / CHANGEMENT AMPOULE","2":"2 / FUITE","3":"3 / SERRURE CASSEE","4":"4 / WC BOUCHE"}
|
||||
"""
|
||||
|
||||
PARAMETER_RESPONSE = '{"LIBELDEMDEF":"O"}'
|
||||
|
||||
CREATE_DEMAND_RESPONSE = """
|
||||
{"sgesdemNum": "000000000001234"}
|
||||
"""
|
||||
|
||||
ADD_DOCUMENT_RESPONSE = """
|
||||
""
|
||||
"""
|
||||
|
||||
POSITION_RESPONSE = """
|
||||
{"position":"E","positionLib":"Envoi atelier","info":null}
|
||||
"""
|
||||
|
||||
POSITIONS_RESPONSE = """
|
||||
[{"position":"A","positionLib":"En attente","color":"0, 0, 0"},{"position":"E","positionLib":"Envoi","color":"190, 190, 0"},{"position":"C","positionLib":"En cours","color":"255, 0, 0"},{"position":"D","positionLib":"Envoi signataire","color":"255, 255, 113"},{"position":"T","positionLib":"Termin\u00e9","color":"0, 0, 0"},{"position":"I","positionLib":"\u00c9dition devis","color":"0, 255, 255"},{"position":"R","positionLib":"Refus","color":"255, 0, 0"},{"position":"V","positionLib":"V\u00e9rification","color":"0, 255, 0"},{"position":"F","positionLib":"Devis effectu\u00e9","color":"153, 204, 255"},{"position":"P","positionLib":"Livraison partielle","color":"255, 102, 0"},{"position":"L","positionLib":"Livraison","color":"128, 0, 0"}]
|
||||
"""
|
||||
|
||||
|
||||
@mock.patch("passerelle.utils.Request.request")
|
||||
def test_connections(mocked_request, app, setup):
|
||||
mocked_request.return_value = utils.FakedResponse(content=CONNECTION_RESPONSE, status_code=200)
|
||||
endpoint = reverse(
|
||||
"generic-endpoint",
|
||||
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "connections"},
|
||||
)
|
||||
response = app.get(endpoint)
|
||||
assert mocked_request.call_args[0][0] == 'get'
|
||||
assert mocked_request.call_args[0][1].endswith("connection/all")
|
||||
assert mocked_request.call_count == 1
|
||||
assert response.json["err"] == 0
|
||||
assert response.json["data"]["default"] == "TEST"
|
||||
|
||||
ASTech.objects.update(connection="OTHER")
|
||||
response = app.get(endpoint)
|
||||
assert response.json["err"] == 1
|
||||
|
||||
ASTech.objects.update(connection="TEST")
|
||||
response = app.get(endpoint)
|
||||
assert response.json["err"] == 0
|
||||
assert response.json["data"]["default"] == "TEST"
|
||||
|
||||
# bad response
|
||||
mocked_request.return_value = utils.FakedResponse(
|
||||
content='{"msg": "not found"}', status_code=404, reason="Not Found"
|
||||
)
|
||||
response = app.get(endpoint)
|
||||
assert response.json["err"] == 1
|
||||
assert response.json["err_class"].endswith("APIError")
|
||||
assert response.json["err_desc"] == "AS-TECH response: 404 Not Found"
|
||||
assert response.json["data"]["error"]["status"] == 404
|
||||
assert response.json["data"]["error"]["content"]["msg"] == "not found"
|
||||
mocked_request.return_value = utils.FakedResponse(content="crash", status_code=500, reason="Crashhhh")
|
||||
response = app.get(endpoint)
|
||||
assert response.json["err"] == 1
|
||||
assert response.json["err_class"].endswith("APIError")
|
||||
assert response.json["err_desc"] == "AS-TECH response: 500 Crashhhh"
|
||||
assert response.json["data"]["error"]["status"] == 500
|
||||
assert response.json["data"]["error"]["content"] == "crash"
|
||||
mocked_request.return_value = utils.FakedResponse(content="not json", status_code=200, reason="OK")
|
||||
response = app.get(endpoint)
|
||||
assert response.json["err"] == 1
|
||||
assert response.json["err_class"].endswith("APIError")
|
||||
assert response.json["err_desc"].startswith("invalid JSON in response:")
|
||||
mocked_request.side_effect = ConnectionError('mocked error', request=Request())
|
||||
response = app.get(endpoint)
|
||||
assert response.json["err"] == 1
|
||||
assert response.json["err_class"].endswith("APIError")
|
||||
assert response.json["err_desc"] == 'connection error: mocked error'
|
||||
|
||||
|
||||
@mock.patch("passerelle.utils.Request.request")
|
||||
def test_authorization(mocked_request, app, setup):
|
||||
mocked_request.side_effect = [
|
||||
utils.FakedResponse(content=CONNECTION_RESPONSE, status_code=200),
|
||||
utils.FakedResponse(content=AUTH_RESPONSE, status_code=200),
|
||||
]
|
||||
endpoint = reverse(
|
||||
"generic-endpoint",
|
||||
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "authorization"},
|
||||
)
|
||||
response = app.get(endpoint)
|
||||
assert mocked_request.call_count == 2
|
||||
assert mocked_request.call_args_list[0][0][0] == 'get'
|
||||
assert mocked_request.call_args_list[0][0][1].endswith('connection/all') # get client_id
|
||||
assert mocked_request.call_args_list[1][0][0] == 'post'
|
||||
assert mocked_request.call_args_list[1][0][1].endswith('oauth/v2/auth') # get access_token
|
||||
assert response.json['err'] == 0
|
||||
assert response.json['data']['access_token'] == '4242'
|
||||
assert response.json['data']['connection_id'] == 'TEST'
|
||||
|
||||
# test cache
|
||||
response = app.get(endpoint)
|
||||
assert mocked_request.call_count == 2
|
||||
|
||||
|
||||
@mock.patch("passerelle.utils.Request.request")
|
||||
@mock.patch("passerelle.apps.astech.models.ASTech.get_authorization")
|
||||
def test_services(mocked_auth, mocked_request, app, setup):
|
||||
mocked_auth.return_value = {'access_token': '4242', 'connection_id': 'TEST'}
|
||||
endpoint = reverse(
|
||||
"generic-endpoint",
|
||||
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "services"},
|
||||
)
|
||||
|
||||
mocked_request.return_value = utils.FakedResponse(content=SERVICES_RESPONSE, status_code=200)
|
||||
response = app.get(endpoint)
|
||||
assert mocked_request.call_count == 1
|
||||
assert mocked_request.call_args[0][0] == 'post'
|
||||
assert mocked_request.call_args[0][1].endswith('/sousservices/invoke')
|
||||
assert mocked_request.call_args[1]['params']['access_token'] == '4242'
|
||||
assert mocked_request.call_args[1]['params']['connection_id'] == 'TEST'
|
||||
assert response.json['err'] == 0
|
||||
assert response.json['data'] == [
|
||||
{"id": "123", "text": "123 / FOOBAR"},
|
||||
{"id": "ABC", "text": "ABC / AH BE CE"},
|
||||
{"id": "XFO", "text": "XFO / BARFOO"},
|
||||
]
|
||||
|
||||
|
||||
@mock.patch("passerelle.utils.Request.request")
|
||||
@mock.patch("passerelle.apps.astech.models.ASTech.get_authorization")
|
||||
def test_companies(mocked_auth, mocked_request, app, setup):
|
||||
mocked_auth.return_value = {'access_token': '4242', 'connection_id': 'TEST'}
|
||||
endpoint = reverse(
|
||||
"generic-endpoint",
|
||||
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "companies"},
|
||||
)
|
||||
|
||||
mocked_request.return_value = utils.FakedResponse(content=COMPANIES_RESPONSE, status_code=200)
|
||||
response = app.get(endpoint)
|
||||
assert mocked_request.call_count == 1
|
||||
assert mocked_request.call_args[0][0] == 'post'
|
||||
assert mocked_request.call_args[0][1].endswith('/societes_demandeur/invoke')
|
||||
assert mocked_request.call_args[1]['json']['codeDemandeur'] == 'ENTROUV'
|
||||
assert response.json['err'] == 0
|
||||
assert response.json['data'] == [
|
||||
{"id": "01", "text": "01 / SERVICES TECHNIQUES"},
|
||||
{"id": "10", "text": "10 / DIRECTION BATIMENT"},
|
||||
{"id": "11", "text": "11 / PLOMBERIE"},
|
||||
]
|
||||
|
||||
|
||||
@mock.patch("passerelle.utils.Request.request")
|
||||
@mock.patch("passerelle.apps.astech.models.ASTech.get_authorization")
|
||||
def test_labels(mocked_auth, mocked_request, app, setup):
|
||||
mocked_auth.return_value = {'access_token': '4242', 'connection_id': 'TEST'}
|
||||
endpoint = reverse(
|
||||
"generic-endpoint",
|
||||
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "labels"},
|
||||
)
|
||||
|
||||
mocked_request.side_effect = [
|
||||
utils.FakedResponse(content=COMPANY_RESPONSE, status_code=200),
|
||||
utils.FakedResponse(content=LABELS_RESPONSE, status_code=200),
|
||||
]
|
||||
response = app.get(endpoint, status=200)
|
||||
assert mocked_request.call_count == 2
|
||||
assert mocked_request.call_args_list[0][0][0] == 'post'
|
||||
assert mocked_request.call_args_list[0][0][1].endswith('societe_demandeur/invoke') # get company (99)
|
||||
assert mocked_request.call_args_list[1][0][0] == 'post'
|
||||
assert mocked_request.call_args_list[1][0][1].endswith('libelles_predefinis/invoke') # get labels
|
||||
assert mocked_request.call_args_list[1][1]['json'] == {'societeDemandeur': '99'}
|
||||
assert response.json['err'] == 0
|
||||
assert response.json['data'] == [
|
||||
{"id": "1", "text": "1 / CHANGEMENT AMPOULE"},
|
||||
{"id": "2", "text": "2 / FUITE"},
|
||||
{"id": "3", "text": "3 / SERRURE CASSEE"},
|
||||
{"id": "4", "text": "4 / WC BOUCHE"},
|
||||
]
|
||||
|
||||
mocked_request.reset_mock(side_effect=True)
|
||||
mocked_request.return_value = utils.FakedResponse(content=LABELS_RESPONSE, status_code=200)
|
||||
response = app.get(endpoint + '?company=42', status=200)
|
||||
assert mocked_request.call_count == 1
|
||||
assert mocked_request.call_args[0][0] == 'post'
|
||||
assert mocked_request.call_args[0][1].endswith('libelles_predefinis/invoke') # get labels
|
||||
assert mocked_request.call_args[1]['json'] == {'societeDemandeur': '42'}
|
||||
assert response.json['err'] == 0
|
||||
assert len(response.json['data']) == 4
|
||||
|
||||
|
||||
@mock.patch("passerelle.utils.Request.request")
|
||||
@mock.patch("passerelle.apps.astech.models.ASTech.get_authorization")
|
||||
def test_parameter(mocked_auth, mocked_request, app, setup):
|
||||
mocked_auth.return_value = {'access_token': '4242', 'connection_id': 'TEST'}
|
||||
endpoint = reverse(
|
||||
"generic-endpoint",
|
||||
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "parameter"},
|
||||
)
|
||||
|
||||
mocked_request.side_effect = [
|
||||
utils.FakedResponse(content=COMPANY_RESPONSE, status_code=200),
|
||||
utils.FakedResponse(content=PARAMETER_RESPONSE, status_code=200),
|
||||
]
|
||||
response = app.get(endpoint + '?name=LIBELDEMDEF', status=200)
|
||||
assert mocked_request.call_count == 2
|
||||
assert mocked_request.call_args_list[0][0][0] == 'post'
|
||||
assert mocked_request.call_args_list[0][0][1].endswith('societe_demandeur/invoke') # get company (99)
|
||||
assert mocked_request.call_args_list[1][0][0] == 'get'
|
||||
assert mocked_request.call_args_list[1][0][1].endswith('/common/getparam/LIBELDEMDEF/99') # get param
|
||||
assert response.json['err'] == 0
|
||||
assert response.json['data']['LIBELDEMDEF'] == 'O'
|
||||
|
||||
mocked_request.reset_mock(side_effect=True)
|
||||
mocked_request.return_value = utils.FakedResponse(content=PARAMETER_RESPONSE, status_code=200)
|
||||
response = app.get(endpoint + '?name=LIBELDEMDEF&company=00', status=200)
|
||||
assert mocked_request.call_count == 1
|
||||
assert mocked_request.call_args[0][0] == 'get'
|
||||
assert mocked_request.call_args[0][1].endswith('/common/getparam/LIBELDEMDEF/00') # get param
|
||||
assert response.json['err'] == 0
|
||||
assert response.json['data']['LIBELDEMDEF'] == 'O'
|
||||
|
||||
response = app.get(endpoint, status=400)
|
||||
|
||||
|
||||
@mock.patch("passerelle.utils.Request.request")
|
||||
@mock.patch("passerelle.apps.astech.models.ASTech.get_authorization")
|
||||
def test_create_demand(mocked_auth, mocked_request, app, setup):
|
||||
mocked_auth.return_value = {'access_token': '4242', 'connection_id': 'TEST'}
|
||||
endpoint = reverse(
|
||||
"generic-endpoint",
|
||||
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "create-demand"},
|
||||
)
|
||||
|
||||
mocked_request.return_value = utils.FakedResponse(content=CREATE_DEMAND_RESPONSE, status_code=201)
|
||||
demand = {
|
||||
'company': '99',
|
||||
'service': '11',
|
||||
'label': '4',
|
||||
'subject': '4 / WC BOUCHE',
|
||||
'name': 'Super Mario',
|
||||
'description': 'Come with Luigi please',
|
||||
'email': 'peach@example.net',
|
||||
'phone1': '123',
|
||||
'phone2': '321',
|
||||
'address1': 'Nintendo Tokyo Branch Office',
|
||||
'address2': 'Chiyoda-ku',
|
||||
'address3': 'Tokyo 101-0054',
|
||||
}
|
||||
response = app.post_json(endpoint, params=demand, status=200)
|
||||
assert mocked_request.call_count == 1
|
||||
assert mocked_request.call_args[0][0] == 'post'
|
||||
assert mocked_request.call_args[0][1].endswith('interface-citoyenne/demande-intervention')
|
||||
assert mocked_request.call_args[1]['json'] == {
|
||||
'interface_citoyenne_demande': {
|
||||
'sgesdemSoc': '99',
|
||||
'sgesdemSserv': '11',
|
||||
'sgesdemLibdef': '4',
|
||||
'sgesdemNdt': '4 / WC BOUCHE',
|
||||
'sgesdemLibelle': 'Come with Luigi please',
|
||||
'sgesdemCplnom': 'Super Mario',
|
||||
'sgesdemCplemail': 'peach@example.net',
|
||||
'sgesdemCpltel1': '123',
|
||||
'sgesdemCpltel2': '321',
|
||||
'sgesdemCpladr1': 'Nintendo Tokyo Branch Office',
|
||||
'sgesdemCpladr2': 'Chiyoda-ku',
|
||||
'sgesdemCpladr3': 'Tokyo 101-0054',
|
||||
}
|
||||
}
|
||||
assert response.json['err'] == 0
|
||||
assert response.json['data']['demand_id'] == '000000000001234'
|
||||
|
||||
# mock invalid AS-TECH response
|
||||
mocked_request.return_value = utils.FakedResponse(content='{"foo":"bar"}', status_code=200)
|
||||
response = app.post_json(endpoint, params=demand, status=200)
|
||||
assert mocked_request.call_count == 2
|
||||
assert response.json['err'] == 1
|
||||
assert response.json['err_desc'].startswith('no sgesdemNum in response: ')
|
||||
|
||||
# test invalid requests
|
||||
response = app.get(endpoint, status=405)
|
||||
response = app.post_json(endpoint, status=400)
|
||||
response = app.post_json(endpoint, params={'foo': 'bar'}, status=400)
|
||||
|
||||
# add a document
|
||||
mocked_request.reset_mock()
|
||||
endpoint = reverse(
|
||||
"generic-endpoint",
|
||||
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "add-document"},
|
||||
)
|
||||
mocked_request.return_value = utils.FakedResponse(content=ADD_DOCUMENT_RESPONSE, status_code=201)
|
||||
document = {
|
||||
'demand_id': '000000000001234',
|
||||
'title': 'test document',
|
||||
'document': {
|
||||
'filename': 'test.txt',
|
||||
'content_type': 'text/plain',
|
||||
'content': 'Zm9vCg==', # base64(foo)
|
||||
},
|
||||
}
|
||||
response = app.post_json(endpoint, params=document, status=200)
|
||||
assert mocked_request.call_count == 1
|
||||
assert mocked_request.call_args[0][0] == 'post'
|
||||
assert mocked_request.call_args[0][1].endswith('document/sgesdemNum/000000000001234')
|
||||
assert mocked_request.call_args[1]['params']['docTitre'] == 'test document'
|
||||
assert mocked_request.call_args[1]['params']['docFile'] == 'test.txt'
|
||||
assert mocked_request.call_args[1]['files'] == {'file0': ('test.txt', b'foo\n', 'text/plain')}
|
||||
assert response.json['err'] == 0
|
||||
assert response.json['data'] == ""
|
||||
|
||||
|
||||
@mock.patch("passerelle.utils.Request.request")
|
||||
@mock.patch("passerelle.apps.astech.models.ASTech.get_authorization")
|
||||
def test_positions(mocked_auth, mocked_request, app, setup):
|
||||
mocked_auth.return_value = {'access_token': '4242', 'connection_id': 'TEST'}
|
||||
|
||||
# position of a demand
|
||||
endpoint = reverse(
|
||||
"generic-endpoint",
|
||||
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "demand-position"},
|
||||
)
|
||||
mocked_request.return_value = utils.FakedResponse(content=POSITION_RESPONSE, status_code=200)
|
||||
response = app.get(endpoint + '?demand_id=000000000001234', status=200)
|
||||
assert mocked_request.call_count == 1
|
||||
assert mocked_request.call_args[0][0] == 'get'
|
||||
assert mocked_request.call_args[0][1].endswith('apicli/demande/position/000000000001234')
|
||||
assert response.json['err'] == 0
|
||||
assert response.json['data'] == {
|
||||
"position": "E",
|
||||
"positionLib": "Envoi atelier",
|
||||
"info": None,
|
||||
"id": "E",
|
||||
"text": "Envoi atelier",
|
||||
}
|
||||
|
||||
# invalid AS-TECH response
|
||||
mocked_request.return_value = utils.FakedResponse(content='{"foo":"bar"}', status_code=200)
|
||||
response = app.get(endpoint + '?demand_id=000000000001234', status=200)
|
||||
assert mocked_request.call_count == 2
|
||||
assert response.json['err'] == 1
|
||||
assert response.json['err_desc'].startswith('no position in response: ')
|
||||
# invalid request
|
||||
response = app.get(endpoint, status=400)
|
||||
response = app.post_json(endpoint, status=405)
|
||||
|
||||
# get all possible positions
|
||||
mocked_request.reset_mock()
|
||||
endpoint = reverse(
|
||||
"generic-endpoint",
|
||||
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "demand-all-positions"},
|
||||
)
|
||||
mocked_request.return_value = utils.FakedResponse(content=POSITIONS_RESPONSE, status_code=200)
|
||||
response = app.get(endpoint, status=200)
|
||||
assert mocked_request.call_count == 1
|
||||
assert mocked_request.call_args[0][0] == 'get'
|
||||
assert mocked_request.call_args[0][1].endswith('apicli/demande/positions')
|
||||
assert response.json['err'] == 0
|
||||
assert len(response.json['data']) == 11
|
||||
assert response.json['data'][0] == {
|
||||
"position": "A",
|
||||
"positionLib": "En attente",
|
||||
"color": "0, 0, 0",
|
||||
"id": "A",
|
||||
"text": "En attente",
|
||||
}
|
||||
Loading…
Reference in New Issue