368 lines
12 KiB
Python
368 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
|
# 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/>.
|
|
|
|
from contextlib import contextmanager
|
|
|
|
import mock
|
|
import pytest
|
|
import utils
|
|
import xmlschema
|
|
|
|
from passerelle.contrib.caluire_axel import schemas
|
|
from passerelle.contrib.caluire_axel.models import CaluireAxel, Link
|
|
from passerelle.contrib.utils.axel import AxelError
|
|
from passerelle.utils.soap import SOAPError
|
|
|
|
|
|
@pytest.fixture
|
|
def resource(db):
|
|
return utils.make_resource(
|
|
CaluireAxel, slug='test', wsdl_url='http://example.net/AXEL_WS/AxelWS.php?wsdl'
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def link_params():
|
|
return {
|
|
'IDENTFAMILLE': '12345',
|
|
'NOM': 'Doe',
|
|
'PRENOM': 'John',
|
|
}
|
|
|
|
|
|
@contextmanager
|
|
def mock_getdata(content, operation):
|
|
with mock.patch('passerelle.contrib.caluire_axel.models.CaluireAxel.soap_client') as client:
|
|
resp = '''
|
|
<?xml version="1.0"?>
|
|
<PORTAILSERVICE>
|
|
<RESULTAT>
|
|
<TYPE>%s</TYPE>
|
|
<STATUS>OK</STATUS>
|
|
<DATE>10/10/2010 10:10:01</DATE>
|
|
<COMMENTAIRES><![CDATA[]]></COMMENTAIRES>
|
|
</RESULTAT>
|
|
<DATA>
|
|
%s
|
|
</DATA>
|
|
</PORTAILSERVICE>
|
|
'''.strip() % (
|
|
operation,
|
|
content,
|
|
)
|
|
client.return_value.service.getData.return_value = resp
|
|
yield
|
|
|
|
|
|
def test_operation_status_error(resource):
|
|
resp = '''
|
|
<?xml version="1.0"?>
|
|
<PORTAILSERVICE>
|
|
<RESULTAT>
|
|
<TYPE>FindIndividus</TYPE>
|
|
<STATUS>NOK</STATUS>
|
|
<COMMENTAIRES><![CDATA[Foo reason]]></COMMENTAIRES>
|
|
</RESULTAT>
|
|
<DATA>
|
|
<PORTAIL/>
|
|
</DATA>
|
|
</PORTAILSERVICE>
|
|
'''.strip()
|
|
with mock.patch('passerelle.contrib.caluire_axel.models.CaluireAxel.soap_client') as client:
|
|
client.return_value.service.getData.return_value = resp
|
|
with pytest.raises(AxelError, match='Foo reason'):
|
|
schemas.find_individus(
|
|
resource,
|
|
{
|
|
'PORTAIL': {
|
|
'FINDINDIVIDU': {
|
|
'NOM': 'Doe',
|
|
'PRENOM': 'John',
|
|
'NAISSANCE': None,
|
|
'CODEPOSTAL': None,
|
|
'VILLE': None,
|
|
'TEL': None,
|
|
'MAIL': None,
|
|
}
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'content',
|
|
[
|
|
'<PORTAIL><FINDINDIVIDUS/></PORTAIL>',
|
|
],
|
|
)
|
|
def test_operation_find_individus(resource, content):
|
|
with mock_getdata(content, 'FindIndividus'):
|
|
with pytest.raises(AxelError):
|
|
schemas.find_individus(
|
|
resource,
|
|
{
|
|
'PORTAIL': {
|
|
'FINDINDIVIDU': {
|
|
'NOM': 'Doe',
|
|
'PRENOM': 'John',
|
|
'NAISSANCE': None,
|
|
'CODEPOSTAL': None,
|
|
'VILLE': None,
|
|
'TEL': None,
|
|
'MAIL': None,
|
|
}
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
def test_link_endpoint_nameid_empty(app, resource, link_params):
|
|
resp = app.post_json('/caluire-axel/test/link?NameID=', params=link_params, status=400)
|
|
assert resp.json['err_desc'] == "NameID is empty"
|
|
assert resp.json['err'] == 'bad-request'
|
|
|
|
|
|
def test_link_endpoint_axel_error(app, resource, link_params):
|
|
with mock.patch('passerelle.contrib.caluire_axel.schemas.find_individus') as operation:
|
|
operation.side_effect = AxelError('FooBar')
|
|
resp = app.post_json('/caluire-axel/test/link?NameID=yyy', params=link_params)
|
|
assert resp.json['err_desc'] == "Axel error: FooBar"
|
|
assert resp.json['err'] == 'error'
|
|
assert resp.json['data'] == {'xml_request': None, 'xml_response': None}
|
|
|
|
# test xml_request and xml_response only for this endpoint
|
|
xml_request = """<PORTAIL>
|
|
<FINDINDIVIDU>
|
|
<NOM>Doe</NOM>
|
|
<PRENOM>John</PRENOM>
|
|
<NAISSANCE />
|
|
<CODEPOSTAL />
|
|
<VILLE />
|
|
<TEL />
|
|
<MAIL />
|
|
</FINDINDIVIDU>
|
|
</PORTAIL>
|
|
"""
|
|
|
|
with mock_getdata('', 'FindIndividus'):
|
|
with mock.patch('xmlschema.XMLSchema.validate') as xml_validate:
|
|
xml_validate.side_effect = xmlschema.XMLSchemaValidationError(None, None)
|
|
resp = app.post_json('/caluire-axel/test/link?NameID=yyy', params=link_params)
|
|
assert resp.json['err_desc'].startswith("Axel error: invalid request")
|
|
assert resp.json['err'] == 'error'
|
|
assert resp.json['data']['xml_request'] == xml_request
|
|
assert resp.json['data']['xml_response'] is None
|
|
|
|
xml_response = """<PORTAILSERVICE>
|
|
<RESULTAT>
|
|
<TYPE>FINDINDIVIDUS</TYPE>
|
|
<STATUS>NOK</STATUS>
|
|
<COMMENTAIRES>Foo reason</COMMENTAIRES>
|
|
</RESULTAT>
|
|
<DATA>
|
|
<PORTAIL />
|
|
</DATA>
|
|
</PORTAILSERVICE>"""
|
|
response = """
|
|
<?xml version="1.0"?>
|
|
<PORTAILSERVICE>
|
|
<RESULTAT>
|
|
<TYPE>FINDINDIVIDUS</TYPE>
|
|
<STATUS>NOK</STATUS>
|
|
<COMMENTAIRES><![CDATA[Foo reason]]></COMMENTAIRES>
|
|
</RESULTAT>
|
|
<DATA>
|
|
<PORTAIL/>
|
|
</DATA>
|
|
</PORTAILSERVICE>
|
|
""".strip()
|
|
with mock.patch('passerelle.contrib.caluire_axel.models.CaluireAxel.soap_client') as client:
|
|
client.return_value.service.getData.return_value = response
|
|
resp = app.post_json('/caluire-axel/test/link?NameID=yyy', params=link_params)
|
|
assert resp.json['err_desc'] == 'Axel error: Foo reason'
|
|
assert resp.json['err'] == 'error'
|
|
assert resp.json['data']['xml_request'] == xml_request
|
|
assert resp.json['data']['xml_response'] == xml_response
|
|
|
|
content = """<PORTAIL>
|
|
<FINDINDIVIDUS>
|
|
<INDIVIDU>
|
|
<IDENT>123</IDENT>
|
|
<NOM>Doe</NOM>
|
|
<PRENOM>John</PRENOM>
|
|
<FAMILLE>
|
|
<IDENTFAMILLE>12345</IDENTFAMILLE>
|
|
<PLACE>2</PLACE>
|
|
</FAMILLE>
|
|
</INDIVIDU>
|
|
</FINDINDIVIDUS>
|
|
</PORTAIL>"""
|
|
xml_response = (
|
|
"""<PORTAILSERVICE>
|
|
<RESULTAT>
|
|
<TYPE>FindIndividus</TYPE>
|
|
<STATUS>OK</STATUS>
|
|
<DATE>10/10/2010 10:10:01</DATE>
|
|
<COMMENTAIRES />
|
|
</RESULTAT>
|
|
<DATA>
|
|
%s
|
|
</DATA>
|
|
</PORTAILSERVICE>"""
|
|
% content
|
|
)
|
|
with mock_getdata(content, 'FindIndividus'):
|
|
with mock.patch('passerelle.contrib.utils.axel.AxelSchema.decode') as decode:
|
|
decode.side_effect = xmlschema.XMLSchemaValidationError(None, None)
|
|
resp = app.post_json('/caluire-axel/test/link?NameID=yyy', params=link_params)
|
|
assert resp.json['err_desc'].startswith("Axel error: invalid response")
|
|
assert resp.json['err'] == 'error'
|
|
assert resp.json['data']['xml_request'] == xml_request
|
|
assert resp.json['data']['xml_response'] == xml_response
|
|
|
|
with mock.patch('passerelle.contrib.caluire_axel.models.CaluireAxel.soap_client') as client:
|
|
client.side_effect = SOAPError('SOAP service is down')
|
|
resp = app.post_json('/caluire-axel/test/link?NameID=yyy', params=link_params)
|
|
assert resp.json['err_desc'] == "SOAP service is down"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'xml_response',
|
|
[
|
|
'',
|
|
"""<INDIVIDU>
|
|
<IDENT>123</IDENT>
|
|
<CIVILITE/><NOM>A</NOM><PRENOM/><NAISSANCE/><SEXE/><NOMJF/><TELFIXE/><TELPORTABLE/><MAIL/><CSP/><EMPLOYEUR/><VILLEEMP/><PAI/><GARDEALTERNEE/>
|
|
<FAMILLE>
|
|
<IDENTFAMILLE>12346</IDENTFAMILLE>
|
|
<PLACE>2</PLACE>
|
|
<SITUATION/>
|
|
</FAMILLE>
|
|
<FAMILLE>
|
|
<IDENTFAMILLE>12347</IDENTFAMILLE>
|
|
<PLACE>1</PLACE>
|
|
</FAMILLE>
|
|
</INDIVIDU>""",
|
|
"""<INDIVIDU>
|
|
<IDENT>123</IDENT>
|
|
<CIVILITE/><NOM>A</NOM><PRENOM/><NAISSANCE/><SEXE/><NOMJF/><TELFIXE/><TELPORTABLE/><MAIL/><CSP/><EMPLOYEUR/><VILLEEMP/><PAI/><GARDEALTERNEE/>
|
|
<FAMILLE>
|
|
<IDENTFAMILLE>12345</IDENTFAMILLE>
|
|
<PLACE>3</PLACE>
|
|
<SITUATION/>
|
|
</FAMILLE>
|
|
</INDIVIDU>""",
|
|
],
|
|
)
|
|
def test_link_endpoint_no_result(app, resource, link_params, xml_response):
|
|
content = (
|
|
'''<PORTAIL>
|
|
<FINDINDIVIDUS>
|
|
<CODE>42</CODE>
|
|
%s
|
|
</FINDINDIVIDUS>
|
|
</PORTAIL>'''
|
|
% xml_response
|
|
)
|
|
with mock_getdata(content, 'FindIndividus'):
|
|
resp = app.post_json('/caluire-axel/test/link?NameID=yyy', params=link_params)
|
|
assert resp.json['err_desc'] == "Person not found"
|
|
assert resp.json['err'] == 'not-found'
|
|
|
|
|
|
def test_link_endpoint_conflict(app, resource, link_params):
|
|
content = """<PORTAIL><FINDINDIVIDUS>
|
|
<CODE>42</CODE>
|
|
<INDIVIDU>
|
|
<IDENT>123</IDENT>
|
|
<CIVILITE/><NOM>A</NOM><PRENOM/><NAISSANCE/><SEXE/><NOMJF/><TELFIXE/><TELPORTABLE/><MAIL/><CSP/><EMPLOYEUR/><VILLEEMP/><PAI/><GARDEALTERNEE/>
|
|
<FAMILLE>
|
|
<IDENTFAMILLE>12345</IDENTFAMILLE>
|
|
<PLACE>2</PLACE>
|
|
<SITUATION/>
|
|
</FAMILLE>
|
|
</INDIVIDU>
|
|
</FINDINDIVIDUS></PORTAIL>"""
|
|
# existing link but family_id is wrong
|
|
link = Link.objects.create(resource=resource, name_id='yyy', family_id='YYY', person_id='42')
|
|
with mock_getdata(content, 'FindIndividus'):
|
|
resp = app.post_json('/caluire-axel/test/link?NameID=yyy', params=link_params)
|
|
assert resp.json['err_desc'] == "Data conflict"
|
|
assert resp.json['err'] == 'conflict'
|
|
|
|
# existing link but person_id is wrong
|
|
link.family_id = '12345'
|
|
link.person_id = '35'
|
|
link.save()
|
|
with mock_getdata(content, 'FindIndividus'):
|
|
resp = app.post_json('/caluire-axel/test/link?NameID=yyy', params=link_params)
|
|
assert resp.json['err_desc'] == "Data conflict"
|
|
assert resp.json['err'] == 'conflict'
|
|
|
|
|
|
@pytest.mark.parametrize('place', [1, 2])
|
|
def test_link_endpoint(app, resource, link_params, place):
|
|
content = (
|
|
"""<PORTAIL><FINDINDIVIDUS>
|
|
<CODE>42</CODE>
|
|
<INDIVIDU>
|
|
<IDENT>123</IDENT>
|
|
<CIVILITE/><NOM>A</NOM><PRENOM/><NAISSANCE/><SEXE/><NOMJF/><TELFIXE/><TELPORTABLE/><MAIL/><CSP/><EMPLOYEUR/><VILLEEMP/><PAI/><GARDEALTERNEE/>
|
|
<FAMILLE>
|
|
<IDENTFAMILLE>12345</IDENTFAMILLE>
|
|
<PLACE>%s</PLACE>
|
|
<SITUATION/>
|
|
</FAMILLE>
|
|
</INDIVIDU>
|
|
</FINDINDIVIDUS></PORTAIL>"""
|
|
% place
|
|
)
|
|
with mock_getdata(content, 'FindIndividus'):
|
|
resp = app.post_json('/caluire-axel/test/link?NameID=yyy', params=link_params)
|
|
assert set(resp.json.keys()) == set(['err', 'link', 'created', 'family_id', 'data'])
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['family_id'] == '12345'
|
|
assert resp.json['created'] is True
|
|
assert 'xml_request' in resp.json['data']
|
|
assert 'xml_response' in resp.json['data']
|
|
|
|
# again
|
|
with mock_getdata(content, 'FindIndividus'):
|
|
resp = app.post_json('/caluire-axel/test/link?NameID=yyy', params=link_params)
|
|
assert set(resp.json.keys()) == set(['err', 'link', 'created', 'family_id', 'data'])
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['family_id'] == '12345'
|
|
assert resp.json['created'] is False # link already exists
|
|
assert 'xml_request' in resp.json['data']
|
|
assert 'xml_response' in resp.json['data']
|
|
|
|
|
|
def test_unlink_endpoint_no_result(app, resource):
|
|
resp = app.post('/caluire-axel/test/unlink?NameID=yyy')
|
|
assert resp.json['err_desc'] == "Person not found"
|
|
assert resp.json['err'] == 'not-found'
|
|
|
|
|
|
def test_unlink_endpoint(app, resource):
|
|
link = Link.objects.create(resource=resource, name_id='yyy', family_id='XXX', person_id='42')
|
|
resp = app.post('/caluire-axel/test/unlink?NameID=yyy')
|
|
assert Link.objects.exists() is False
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['link'] == link.pk
|
|
assert resp.json['family_id'] == 'XXX'
|
|
assert resp.json['deleted'] is True
|