378 lines
16 KiB
Python
378 lines
16 KiB
Python
import base64
|
|
import os
|
|
from datetime import datetime
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.utils.http import urlencode
|
|
from requests import RequestException
|
|
from zeep.exceptions import Fault
|
|
|
|
from passerelle.apps.atal.models import ATALConnector
|
|
from passerelle.base.models import AccessRight, ApiUser
|
|
from passerelle.utils.soap import SOAPError
|
|
|
|
|
|
def get_file(filename):
|
|
with open(os.path.join(os.path.dirname(__file__), 'data', 'atal', filename), 'rb') 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:
|
|
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_thematique(app, connector, monkeypatch):
|
|
return_value = '''<?xml version="1.0" encoding="UTF-8" standalone="no"?><thematiques><thematique id="9" label="Entretien - Maintenance"></thematique>
|
|
<thematique id="12" label="Logistique"></thematique><thematique id="11" label="Réparation- Dépannage"></thematique>
|
|
<thematique id="13" label="Sinistre"></thematique><thematique id="15" label="Sécurité - Contrôle"></thematique>
|
|
<thematique id="10" label="Travaux Neufs"></thematique><thematique id="14" label="Vandalisme"></thematique></thematiques>
|
|
'''
|
|
mock_atal_soap_call(monkeypatch, return_value=return_value)
|
|
response = app.get('/atal/slug-atal/get-thematique')
|
|
assert response.json['err'] == 0
|
|
assert len(response.json['data']) == 7
|
|
assert {'id': '9', 'text': 'Entretien - Maintenance'} in response.json['data']
|
|
assert {'id': '14', 'text': 'Vandalisme'} in response.json['data']
|
|
|
|
|
|
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 = """<?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)
|
|
|
|
# use filename sub param
|
|
params = {'numero_demande': 'DIT19050001', 'file': {'content': base64_str, 'filename': 'nicefile.json'}}
|
|
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'] == 'nicefile.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',
|
|
}
|
|
|
|
# empty file
|
|
params = {'numero_demande': 'DIT19050001', 'nom_fichier': 'data.json', 'file': {}}
|
|
response = app.post_json('/atal/slug-atal/upload', params=params, status=400)
|
|
assert response.json == {
|
|
'data': None,
|
|
'err': 1,
|
|
'err_class': 'passerelle.utils.json.JSONValidationError',
|
|
'err_desc': "file: 'content' is a required property",
|
|
}
|
|
|
|
# no file
|
|
params = {'numero_demande': 'DIT19050001', 'nom_fichier': 'data.json'}
|
|
response = app.post_json('/atal/slug-atal/upload', params=params, status=400)
|
|
assert response.json == {
|
|
'data': None,
|
|
'err': 1,
|
|
'err_class': 'passerelle.utils.json.JSONValidationError',
|
|
'err_desc': "'file' is a required property",
|
|
}
|
|
|
|
|
|
def test_upload_file_too_big(app, connector, monkeypatch):
|
|
mock_atal_soap_call(monkeypatch, side_effect=Fault('File too big'))
|
|
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'] == 1
|
|
assert response.json['err_desc'] == 'File too big'
|
|
|
|
|
|
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') % b'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') % b'travaux pas commences',
|
|
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'] == 'travaux pas commences'
|
|
|
|
|
|
def test_infos(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') % b'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/infos/DIT18050001/')
|
|
|
|
assert response.json['err'] == 0
|
|
assert response.json['data']['status'] == 'EN ATTENTE'
|
|
|
|
api_response1 = mock.Mock(
|
|
content=get_file('details_demande_response.xml') % b'PRISE EN COMPTE',
|
|
status_code=200,
|
|
headers={'Content-Type': 'text/xml'},
|
|
)
|
|
api_response2 = mock.Mock(
|
|
content=get_file('etat_travaux_response.xml') % b'travaux pas commences',
|
|
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/infos/DIT18050001/')
|
|
assert response.json['err'] == 0
|
|
assert response.json['data']['status'] == 'travaux pas commences'
|
|
|
|
# User comments in response
|
|
api_response1 = mock.Mock(
|
|
content=get_file('details_demande_response_with_comments.xml'),
|
|
status_code=200,
|
|
headers={'Content-Type': 'text/xml'},
|
|
)
|
|
api_response2 = mock.Mock(
|
|
content=get_file('etat_travaux_response.xml') % b'travaux pas commences',
|
|
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/infos/DIT18050001/')
|
|
assert response.json['err'] == 0
|
|
data = response.json['data']
|
|
assert data['status'] == 'travaux pas commences'
|
|
assert data['works_comment'] == {'text': 'bonjour atal', 'date': 'Thursday 24 October 2019, 16:51'}
|
|
assert data['works_comments'] == []
|
|
assert data['demand_comment'] is None
|
|
assert data['works_status'] is None
|
|
|
|
# full in query string
|
|
monkeypatch.setattr(
|
|
passerelle.utils.Request, 'post', mock.Mock(side_effect=[api_response1, api_response2])
|
|
)
|
|
response = app.get('/atal/slug-atal/infos/DIT18050001/?full=true')
|
|
data = response.json['data']
|
|
assert len(data['works_comments']) == 2
|
|
assert data['works_comments'][0] == {'text': 'OK', 'date': 'Thursday 24 October 2019, 16:48'}
|
|
last_comment = {'text': 'bonjour atal', 'date': 'Thursday 24 October 2019, 16:51'}
|
|
assert data['works_comments'][1] == last_comment
|
|
assert data['works_comment'] == last_comment
|
|
|
|
|
|
def test_new_comments(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_with_comments.xml'),
|
|
status_code=200,
|
|
headers={'Content-Type': 'text/xml'},
|
|
)
|
|
monkeypatch.setattr(passerelle.utils.Request, 'post', mock.Mock(return_value=api_response))
|
|
|
|
all_comments = [
|
|
{'text': 'OK', 'date': 'Thursday 24 October 2019, 16:48', 'date_raw': '2019-10-24T16:48:34+02:00'},
|
|
{
|
|
'text': 'bonjour atal',
|
|
'date': 'Thursday 24 October 2019, 16:51',
|
|
'date_raw': '2019-10-24T16:51:37+02:00',
|
|
},
|
|
]
|
|
|
|
last_datetime = datetime(year=2019, month=10, day=23)
|
|
params = urlencode({'last_datetime': last_datetime})
|
|
response = app.get('/atal/slug-atal/new-comments/DIT18050001/?%s' % params)
|
|
assert response.json['err'] == 0
|
|
assert response.json['data']['new_comments'] == all_comments
|
|
assert response.json['data']['all_comments'] == all_comments
|
|
assert response.json['data']['last_date'] == '2019-10-24T16:51:37+02:00'
|
|
|
|
last_datetime = datetime(year=2019, month=10, day=24, hour=16, minute=49)
|
|
params = urlencode({'last_datetime': last_datetime})
|
|
response = app.get('/atal/slug-atal/new-comments/DIT18050001/?%s' % params)
|
|
assert response.json['err'] == 0
|
|
assert response.json['data']['new_comments'] == all_comments[1:]
|
|
assert response.json['data']['all_comments'] == all_comments
|
|
assert response.json['data']['last_date'] == '2019-10-24T16:51:37+02:00'
|
|
|
|
last_datetime = datetime(year=2019, month=10, day=24, hour=16, minute=52)
|
|
params = urlencode({'last_datetime': last_datetime})
|
|
response = app.get('/atal/slug-atal/new-comments/DIT18050001/?%s' % params)
|
|
assert response.json['err'] == 0
|
|
assert response.json['data']['new_comments'] == []
|
|
assert response.json['data']['all_comments'] == all_comments
|
|
assert response.json['data']['last_date'] == '2019-10-24T16:51:37+02:00'
|
|
|
|
|
|
def test_check_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))
|
|
connector.check_status()
|
|
|
|
monkeypatch.setattr(passerelle.utils.Request, 'get', mock.Mock(side_effect=RequestException))
|
|
with pytest.raises(SOAPError):
|
|
connector.check_status()
|