256 lines
8.8 KiB
Python
256 lines
8.8 KiB
Python
# coding: utf-8
|
|
|
|
import base64
|
|
import os
|
|
|
|
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
|
|
|
|
|
|
def get_file(filename):
|
|
with open(
|
|
os.path.join(
|
|
os.path.dirname(__file__), 'data', 'atal', filename
|
|
),
|
|
'r'
|
|
) as f:
|
|
return f.read()
|
|
|
|
|
|
@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_action_comment(app, connector, monkeypatch):
|
|
mock_soap_call = mock_atal_soap_call(monkeypatch, return_value='DIT19050001')
|
|
params = {
|
|
'numero_demande': 'DIT19050001',
|
|
'commentaire': 'aaa'
|
|
}
|
|
response = app.post_json('/atal/slug-atal/insert-action-comment', 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'] == 'insertActionComment'
|
|
assert call_params['numeroDemande'] == 'DIT19050001'
|
|
assert call_params['commentaire'] == 'aaa'
|
|
|
|
|
|
def test_insert_demande_complet_by_type(app, connector, monkeypatch):
|
|
mock_soap_call = mock_atal_soap_call(monkeypatch, return_value='DIT19050001')
|
|
params = {
|
|
'type_demande': 'VOIRIE',
|
|
'coord_x': '48.866667', 'coord_y': '2.333333'
|
|
}
|
|
response = app.post_json('/atal/slug-atal/insert-demande-complet-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'] == 'insertDemandeCompletByType'
|
|
assert call_params['typeDemande'] == 'VOIRIE'
|
|
assert call_params['coordX'] == '48.866667'
|
|
assert call_params['coordY'] == '2.333333'
|
|
|
|
|
|
def test_upload(app, connector, monkeypatch):
|
|
mock_soap_call = mock_atal_soap_call(monkeypatch, return_value=None)
|
|
base64_str = 'eyJsYXN0X2NoZWNrIjoiMjAxOS0wNC0xMFQxMjowODoyOVoiL' + \
|
|
'CJweXBpX3ZlcnNpb24iOiIxOS4wLjMifQ=='
|
|
params = {
|
|
'numero_demande': 'DIT19050001',
|
|
'nom_fichier': 'data.json',
|
|
'file': {
|
|
'content': base64_str
|
|
}
|
|
}
|
|
response = app.post_json('/atal/slug-atal/upload', params=params)
|
|
assert response.json == {
|
|
'err': 0
|
|
}
|
|
call_params = mock_soap_call.call_args.kwargs
|
|
assert call_params['wsdl'] == 'ChargementPiecesJointesService'
|
|
assert call_params['method'] == 'upload'
|
|
assert call_params['numeroDemande'] == 'DIT19050001'
|
|
assert call_params['nomFichier'] == 'data.json'
|
|
assert call_params['donneesFichier'] == base64.b64decode(base64_str)
|
|
|
|
params = {
|
|
'numero_demande': 'DIT19050001',
|
|
'nom_fichier': 'data.json',
|
|
'file': {
|
|
'content': 'invalidbase64'
|
|
}
|
|
}
|
|
response = app.post_json('/atal/slug-atal/upload', params=params)
|
|
assert response.json == {
|
|
'data': None,
|
|
'err': 1,
|
|
'err_class': 'passerelle.utils.jsonresponse.APIError',
|
|
'err_desc': 'Invalid base64 string'
|
|
}
|
|
|
|
|
|
def test_retrieve_details_demande(app, connector, monkeypatch):
|
|
import passerelle.utils
|
|
|
|
wsdl_response = mock.Mock(
|
|
content=get_file('DemandeService.wsdl'), status_code=200,
|
|
headers={'Content-Type': 'text/xml'}
|
|
)
|
|
monkeypatch.setattr(passerelle.utils.Request, 'get', mock.Mock(return_value=wsdl_response))
|
|
|
|
api_response = mock.Mock(
|
|
content=get_file('details_demande_response.xml') % 'EN ATTENTE', status_code=200,
|
|
headers={'Content-Type': 'text/xml'}
|
|
)
|
|
monkeypatch.setattr(passerelle.utils.Request, 'post', mock.Mock(return_value=api_response))
|
|
|
|
response = app.get('/atal/slug-atal/retrieve-details-demande/DIT18050001/')
|
|
assert response.json['err'] == 0
|
|
assert response.json['data']['etatDemande']['description'] == 'EN ATTENTE'
|
|
|
|
|
|
def test_retrieve_etat_travaux(app, connector, monkeypatch):
|
|
import passerelle.utils
|
|
|
|
wsdl_response = mock.Mock(
|
|
content=get_file('DemandeService.wsdl'), status_code=200,
|
|
headers={'Content-Type': 'text/xml'}
|
|
)
|
|
monkeypatch.setattr(passerelle.utils.Request, 'get', mock.Mock(return_value=wsdl_response))
|
|
|
|
api_response = mock.Mock(
|
|
content=get_file('etat_travaux_response.xml') % u'travaux pas commencés', status_code=200,
|
|
headers={'Content-Type': 'text/xml'}
|
|
)
|
|
monkeypatch.setattr(passerelle.utils.Request, 'post', mock.Mock(return_value=api_response))
|
|
|
|
response = app.get('/atal/slug-atal/retrieve-etat-travaux/DIT18050001/')
|
|
assert response.json['err'] == 0
|
|
assert response.json['data']['libelle'] == u'travaux pas commencés'
|
|
|
|
|
|
def test_status(app, connector, monkeypatch):
|
|
import passerelle.utils
|
|
|
|
wsdl_response = mock.Mock(
|
|
content=get_file('DemandeService.wsdl'), status_code=200,
|
|
headers={'Content-Type': 'text/xml'}
|
|
)
|
|
monkeypatch.setattr(passerelle.utils.Request, 'get', mock.Mock(return_value=wsdl_response))
|
|
|
|
api_response = mock.Mock(
|
|
content=get_file('details_demande_response.xml') % 'EN ATTENTE', status_code=200,
|
|
headers={'Content-Type': 'text/xml'}
|
|
)
|
|
monkeypatch.setattr(passerelle.utils.Request, 'post', mock.Mock(return_value=api_response))
|
|
response = app.get('/atal/slug-atal/status/DIT18050001/')
|
|
|
|
assert response.json['err'] == 0
|
|
assert response.json['data']['status'] == u'EN ATTENTE'
|
|
|
|
api_response1 = mock.Mock(
|
|
content=get_file('details_demande_response.xml') % 'PRISE EN COMPTE', status_code=200,
|
|
headers={'Content-Type': 'text/xml'}
|
|
)
|
|
api_response2 = mock.Mock(
|
|
content=get_file('etat_travaux_response.xml') % u'travaux pas commencés', status_code=200,
|
|
headers={'Content-Type': 'text/xml'}
|
|
)
|
|
monkeypatch.setattr(
|
|
passerelle.utils.Request, 'post', mock.Mock(side_effect=[api_response1, api_response2])
|
|
)
|
|
|
|
response = app.get('/atal/slug-atal/status/DIT18050001/')
|
|
assert response.json['err'] == 0
|
|
assert response.json['data']['status'] == u'travaux pas commencés'
|