atos-genesys: add new endpoints search and link-by-id-per (#32618)
xml_request had to be factorized to created xml_request_multiple as the Genesys web-service for search return multiple results (all other WS returns only one). call_select_usager was also factorized to serve as basis for call_select_usager_by_ref. The search endpoint does a first call to Genesys for listing REF_PER then for each REF_PER a call to selectUsagerByRefPer to get the ID_PER and contact informations necessary for appairage.
This commit is contained in:
parent
6427389205
commit
d643e759b7
|
|
@ -14,6 +14,7 @@
|
|||
# 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 datetime
|
||||
import urlparse
|
||||
import requests
|
||||
import xml.etree.ElementTree as ET
|
||||
|
|
@ -51,6 +52,21 @@ class Resource(BaseResource, HTTPResource):
|
|||
return urlparse.urljoin(self.base_url, 'WSUsagerPublik/services/PublikService/selectCodifications')
|
||||
|
||||
def xml_request(self, url, *args, **kwargs):
|
||||
root, response = self._xml_request_base(url, *args, **kwargs)
|
||||
row = root.find('ROWSET/ROW')
|
||||
if row is None:
|
||||
raise APIError('no ROWSET/ROW node', data={'content': response.text[:1024]})
|
||||
return row
|
||||
|
||||
def xml_request_multiple(self, url, *args, **kwargs):
|
||||
root, response = self._xml_request_base(url, *args, **kwargs)
|
||||
rowset = root.find('ROWSET')
|
||||
if rowset is None:
|
||||
raise APIError('no ROWSET node', data={'content': response.text[:1024]})
|
||||
rows = rowset.findall('ROW')
|
||||
return rows
|
||||
|
||||
def _xml_request_base(self, url, *args, **kwargs):
|
||||
try:
|
||||
response = self.requests.get(url, *args, **kwargs)
|
||||
response.raise_for_status()
|
||||
|
|
@ -62,10 +78,7 @@ class Resource(BaseResource, HTTPResource):
|
|||
raise APIError('XML parsing failed', data={'exception': six.text_type(e)})
|
||||
if root.tag != 'return':
|
||||
raise APIError('root XML node is not return', data={'content': response.text[:1024]})
|
||||
row = root.find('ROWSET/ROW')
|
||||
if row is None:
|
||||
raise APIError('no ROWSET/ROW node', data={'content': response.text[:1024]})
|
||||
return row
|
||||
return root, response
|
||||
|
||||
def call_select_codifications(self):
|
||||
root = self.xml_request(self.select_codifications_url)
|
||||
|
|
@ -239,6 +252,9 @@ class Resource(BaseResource, HTTPResource):
|
|||
'idPer': id_per,
|
||||
'codRgp': self.cod_rgp,
|
||||
})
|
||||
return self._select_usager_row_to_json(row)
|
||||
|
||||
def _select_usager_row_to_json(self, row):
|
||||
d = xmlutils.to_json(row)
|
||||
# sort demandes and droits using COD_APPLI
|
||||
demandes = {}
|
||||
|
|
@ -304,6 +320,87 @@ class Resource(BaseResource, HTTPResource):
|
|||
return {'data': data[0] if data else None}
|
||||
return {'data': data}
|
||||
|
||||
@property
|
||||
def select_usager_by_ref_url(self):
|
||||
return urlparse.urljoin(self.base_url, 'WSUsagerPublik/services/PublikService/selectUsagerByRef')
|
||||
|
||||
def call_select_usager_by_ref(self, ref_per):
|
||||
row = self.xml_request(self.select_usager_by_ref_url, params={
|
||||
'refPer': ref_per,
|
||||
'codRgp': self.cod_rgp,
|
||||
})
|
||||
return self._select_usager_row_to_json(row)
|
||||
|
||||
@property
|
||||
def cherche_beneficiaire_url(self):
|
||||
return urlparse.urljoin(self.base_url, 'WSIntermed/services/IntermediationService/chercheBeneficiaire')
|
||||
|
||||
def call_cherche_beneficiaire(self, nom, prenom, dob):
|
||||
rows = self.xml_request_multiple(self.cherche_beneficiaire_url, params={
|
||||
'nmPer': nom,
|
||||
'prPer': prenom,
|
||||
'dtNaissance': dob.strftime('%d/%m/%Y'),
|
||||
})
|
||||
beneficiaires = [xmlutils.to_json(row) for row in rows]
|
||||
return beneficiaires
|
||||
|
||||
@endpoint(name='search',
|
||||
description=_('Search for beneficiaries'),
|
||||
perm='can_access',
|
||||
parameters={
|
||||
'first_name': {
|
||||
'description': _('Beneficiary first name'),
|
||||
'example_value': 'John',
|
||||
},
|
||||
'last_name': {
|
||||
'description': _('Beneficiary last name'),
|
||||
'example_value': 'Doe',
|
||||
},
|
||||
'date_of_birth': {
|
||||
'description': _('Beneficiary date of birth'),
|
||||
'example_value': '1987-10-23',
|
||||
}
|
||||
})
|
||||
def search(self, request, first_name, last_name, date_of_birth):
|
||||
try:
|
||||
date_of_birth = datetime.datetime.strptime(date_of_birth, '%Y-%m-%d')
|
||||
except (ValueError, TypeError):
|
||||
raise APIError('invalid date_of_birth: %r' % date_of_birth)
|
||||
beneficiaires = self.call_cherche_beneficiaire(first_name, last_name, date_of_birth)
|
||||
for beneficiaire in beneficiaires:
|
||||
ref_per = beneficiaire.get('REF_PER')
|
||||
if not ref_per:
|
||||
continue
|
||||
dossier = self.call_select_usager_by_ref(ref_per)
|
||||
identification = dossier['IDENTIFICATION'][0]
|
||||
beneficiaire['ID_PER'] = identification['ID_PER']
|
||||
beneficiaire['TEL_FIXE'] = identification.get('TEL_FIXE', '')
|
||||
beneficiaire['TEL_MOBILE'] = identification.get('TEL_MOBILE', '')
|
||||
beneficiaire['MAIL'] = identification.get('MAIL', '')
|
||||
return {'data': beneficiaires}
|
||||
|
||||
@endpoint(name='link-by-id-per',
|
||||
methods=['post'],
|
||||
description=_('Create link with an extranet account'),
|
||||
perm='can_access',
|
||||
parameters={
|
||||
'NameID':{
|
||||
'description': _('Publik NameID'),
|
||||
'example_value': 'xyz24d934',
|
||||
},
|
||||
'id_per': {
|
||||
'description': _('ATOS Genesys ID_PER'),
|
||||
'example_value': '767676',
|
||||
}
|
||||
})
|
||||
def link_by_id_per(self, request, NameID, id_per):
|
||||
dossier = self.call_select_usager(id_per)
|
||||
link, created = Link.objects.get_or_create(
|
||||
resource=self,
|
||||
name_id=NameID,
|
||||
id_per=id_per)
|
||||
return {'link_id': link.pk, 'new': created}
|
||||
|
||||
|
||||
class Link(models.Model):
|
||||
resource = models.ForeignKey(
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@
|
|||
<ROW num="1">
|
||||
<IDENTIFICATION>
|
||||
<IDENTIFICATION_ROW num="1">
|
||||
<ID_PER>1234</ID_PER>
|
||||
<NOM>DOE</NOM>
|
||||
<NOM_NAISSANCE>TEST</NOM_NAISSANCE>
|
||||
<PRENOM>John</PRENOM>
|
||||
|
|
|
|||
|
|
@ -230,3 +230,58 @@ def test_row_locked_cache(genesys, freezer):
|
|||
counter += 1
|
||||
assert rlc() == 2
|
||||
assert f.calls == 2
|
||||
|
||||
|
||||
def test_ws_search(app, genesys):
|
||||
url = utils.generic_endpoint_url('atos-genesys', 'search', slug=genesys.slug)
|
||||
|
||||
RESPONSE_SEARCH = '''<?xml version="1.0" encoding="UTF-8"?><return><ROWSET>
|
||||
<ROW num="1">
|
||||
<NOMPER>John</NOMPER>
|
||||
<PRENOMPER>Doe</PRENOMPER>
|
||||
<DATE_NAISSANCE>01/01/1925</DATE_NAISSANCE>
|
||||
<REF_PER>951858</REF_PER>
|
||||
<LIEU_NAIS>ANTIBES (006)</LIEU_NAIS>
|
||||
<PAY_NAIS>FRANCE</PAY_NAIS>
|
||||
</ROW>
|
||||
</ROWSET>
|
||||
</return>'''
|
||||
|
||||
with utils.mock_url(FAKE_URL + 'WSIntermed/services/IntermediationService/chercheBeneficiaire', RESPONSE_SEARCH):
|
||||
with utils.mock_url(FAKE_URL + 'WSUsagerPublik/services/PublikService/selectUsagerByRef',
|
||||
RESPONSE_SELECT_USAGER):
|
||||
response = app.get(url + '?' + urlencode({
|
||||
'first_name': 'John',
|
||||
'last_name': 'Doe',
|
||||
'date_of_birth': '1925-01-01',
|
||||
}))
|
||||
assert response.json['err'] == 0
|
||||
assert len(response.json['data']) == 1
|
||||
data = response.json['data'][0]
|
||||
assert data['REF_PER'] == '951858'
|
||||
assert data['ID_PER'] == '1234'
|
||||
assert data['NOMPER'] == 'John'
|
||||
assert data['PRENOMPER'] == 'Doe'
|
||||
assert data['DATE_NAISSANCE'] == '01/01/1925'
|
||||
assert data['TEL_FIXE'] == '06.44.44.44.44'
|
||||
assert data['TEL_MOBILE'] == '06.55.55.55.55'
|
||||
assert data['MAIL'] == 'test@sirus.fr'
|
||||
|
||||
|
||||
def test_ws_link_by_id_per(app, genesys):
|
||||
url = utils.generic_endpoint_url('atos-genesys', 'link-by-id-per', slug=genesys.slug)
|
||||
|
||||
assert Link.objects.count() == 0
|
||||
with utils.mock_url(FAKE_URL + 'WSUsagerPublik/services/PublikService/selectUsager',
|
||||
RESPONSE_SELECT_USAGER):
|
||||
response = app.post(url + '?' + urlencode({
|
||||
'NameID': 'zob',
|
||||
'id_per': '1234',
|
||||
}))
|
||||
|
||||
assert response.json['err'] == 0
|
||||
assert Link.objects.count() == 1
|
||||
link = Link.objects.get()
|
||||
data = response.json
|
||||
assert data['new']
|
||||
assert data['link_id'] == link.pk
|
||||
|
|
|
|||
Loading…
Reference in New Issue