entrouvert
/
passerelle
15
0
Fork 0
Code Issues Pull Requests 15 Releases Activity

r2p: start connector (#86000)
gitea/passerelle/pipeline/head This commit looks good Details

This commit is contained in:
Emmanuel Cazenave 2024-01-22 18:11:02 +01:00
parent eac67cb852
commit 0f1117f483
6 changed files with 546 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
passerelle/apps/r2p/__init__.py Normal file
View File

45
passerelle/apps/r2p/migrations/0001_inital.py Normal file
View File

@ -0,0 +1,45 @@
# Generated by Django 3.2.18 on 2024-01-23 10:26
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
('base', '0030_resourcelog_base_resour_appname_298cbc_idx'),
]
operations = [
migrations.CreateModel(
name='R2P',
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')),
(
'api_url',
models.URLField(
default='https://gw.dgfip.finances.gouv.fr',
max_length=256,
verbose_name='DGFIP API base URL',
),
),
('oauth_username', models.CharField(max_length=128, verbose_name='DGFIP API Username')),
('oauth_password', models.CharField(max_length=128, verbose_name='DGFIP API Password')),
(
'users',
models.ManyToManyField(
blank=True, related_name='_r2p_r2p_users_+', related_query_name='+', to='base.ApiUser'
),
),
],
options={
'verbose_name': 'API Recherche des personnes physiques',
},
),
]

0
passerelle/apps/r2p/migrations/__init__.py Normal file
View File

192
passerelle/apps/r2p/models.py Normal file
View File

@ -0,0 +1,192 @@
# passerelle - uniform access to multiple data sources and services
# Copyright (C) 2024 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 hashlib
import json
import uuid
from urllib.parse import urljoin
import requests
from django.core.cache import cache
from django.db import models
from django.utils.translation import gettext_lazy as _
from passerelle.base.models import BaseResource
from passerelle.utils.api import endpoint
from passerelle.utils.jsonresponse import APIError
BASE_PARAMETERS = {
'prenom': {
'description': _('First name'),
'example_value': 'Jean',
},
'nom': {
'description': _('Last name'),
'example_value': 'Dubignot',
},
'sexe': {
'description': _('Gender (1 = male, 2 = female)'),
'example_value': '2',
},
'naisDate': {
'description': _('Birthdate (YYYY or MM/YYYY or DD/MM/YYYY)'),
'example_value': '15/06/1943',
},
'naisCodePays': {
'description': _('Birth country (5 caracters - INSEE)'),
'example_value': '99100',
},
'naisCodeDept': {
'description': _('Birth department (2 caracters or 3 caracters for TOM and DOM)'),
'example_value': '75',
},
'naisCodeCommune': {
'description': _('Birth commune (3 caracters or 2 caracters for TOM and DOM - INSEE)'),
'example_value': '109',
},
}
PERSONNE_BY_CRITERIA_PARAMETERS = BASE_PARAMETERS.copy()
PERSONNE_BY_CRITERIA_PARAMETERS.update(
{
'adrCodePays': {
'description': _('Adress country (5 caracters - INSEE)'),
'example_value': '99100',
},
'adrCodeDept': {
'description': _('Adress department (2 caracters or 3 caracters for TOM and DOM)'),
'example_value': '75',
},
'adrCodeCommune': {
'description': _('Adress commune (3 caracters or 2 caracters for TOM and DOM - INSEE)'),
'example_value': '109',
},
'adrLibelleVoie': {
'description': _('Adress street'),
'example_value': 'danton',
},
'adrIndiceRepetition': {
'description': _('Adress repeat index (B = bis, T = ter, Q = quater)'),
'example_value': 'B',
},
'adrNumVoie': {
'description': _('Adress street number'),
'example_value': '22',
},
}
)
def parse_spi(value):
value = value.strip().replace(' ', '')
if not (value and value.isascii() and value.isdigit()):
raise APIError(_('invalid spi'))
return value
class R2P(BaseResource):
api_url = models.URLField(
_('DGFIP API base URL'),
max_length=256,
default='https://gw.dgfip.finances.gouv.fr',
)
oauth_username = models.CharField(_('DGFIP API Username'), max_length=128)
oauth_password = models.CharField(_('DGFIP API Password'), max_length=128)
category = _('Business Process Connectors')
class Meta:
verbose_name = _('API Recherche des personnes physiques')
def _get_access_token(self):
key = (
'r2p-at-'
+ hashlib.sha256(
f'{self.oauth_username}-{self.oauth_password}-{self.api_url}'.encode()
).hexdigest()
)
access_token = cache.get(key)
if not access_token:
data = {
'grant_type': 'client_credentials',
'scope': 'api_r2p_recherche_personne_physique api_r2p_recherche_personne_physique api_r2p_resolution_spi_degrade',
}
url = urljoin(self.api_url, '/token')
try:
response = self.requests.post(url, data=data, auth=(self.oauth_username, self.oauth_password))
response.raise_for_status()
except requests.RequestException:
raise APIError('Could not obtain a token')
try:
response_data = response.json()
access_token = response_data['access_token']
except (ValueError, KeyError, TypeError):
raise APIError('Could not obtain a token')
cache.set(key, access_token, 300)
return access_token
def _call(self, path, params=None):
url = urljoin(self.api_url, path)
token = self._get_access_token()
headers = {
'Authorization': f'Bearer {token}',
'X-Correlation-ID': str(uuid.uuid4().hex),
}
params = params if params else {}
try:
resp = self.requests.get(url=url, headers=headers, params=params)
except (requests.Timeout, requests.RequestException) as e:
raise APIError(str(e))
try:
resp.raise_for_status()
except requests.RequestException as main_exc:
try:
err_data = resp.json()
except (json.JSONDecodeError, requests.exceptions.RequestException):
err_data = {'response_text': resp.text}
raise APIError(str(main_exc), data=err_data)
try:
return resp.json()
except (json.JSONDecodeError, requests.exceptions.RequestException) as e:
raise APIError(str(e))
@endpoint(
name='personne-by-criteria',
description=_('Search natural person by criteria'),
parameters=PERSONNE_BY_CRITERIA_PARAMETERS,
)
def personne_by_criteria(self, request, **kwargs):
return {'data': self._call('/r2p/v1/personne', params=kwargs)}
@endpoint(
name='personne-by-spi',
description=_('Search natural person by SPI'),
parameters={
'spi': {
'description': _('Tax number of the person'),
'example_value': '1257553562447',
},
},
)
def personne_by_spi(self, request, spi):
spi = parse_spi(spi)
return {'data': self._call(f'/r2p/v1/personne/{spi}')}
@endpoint(name='spi-by-criteria', description=_('Search SPI by criteria'), parameters=BASE_PARAMETERS)
def spi_by_criteria(self, request, **kwargs):
return {'data': self._call('/r2p/v1/personne/spi', params=kwargs)}

1
passerelle/settings.py
View File

@ -178,6 +178,7 @@ INSTALLED_APPS = (
'passerelle.apps.photon',
'passerelle.apps.plone_restapi',
'passerelle.apps.proxy',
'passerelle.apps.r2p',
'passerelle.apps.sector',
'passerelle.apps.sfr_dmc',
'passerelle.apps.signal_arretes',

308
tests/test_r2p.py Normal file
View File

@ -0,0 +1,308 @@
import pytest
import responses
from django.contrib.contenttypes.models import ContentType
from django.core.cache import cache
from passerelle.apps.r2p.models import R2P
from passerelle.base.models import AccessRight, ApiUser
@pytest.fixture()
def connector(db):
api = ApiUser.objects.create(username='all', keytype='', key='')
connector = R2P.objects.create(
api_url='https://r2p.invalid', slug='test', oauth_username='foo', oauth_password='bar'
)
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_tocken(rsps):
rsps.post(
'https://r2p.invalid/token',
status=200,
json={'access_token': '1234'},
)
def test_token(app, connector):
cache.clear()
with responses.RequestsMock() as rsps:
mock_tocken(rsps)
rsps.get(
'https://r2p.invalid/r2p/v1/personne/1257553562447',
status=200,
json={
'personnePhysique': {
'identifiant': {'spi': '1257553562447'},
'etatCivil': {'cdSexe': '1', 'lbNomNaissance': 'SIMON'},
}
},
)
resp = app.get('/r2p/test/personne-by-spi?spi=1257553562447')
assert resp.json['err'] == 0
api_call = rsps.calls[1]
assert api_call.request.headers['Authorization'] == 'Bearer 1234'
rsps.reset()
with responses.RequestsMock() as rsps:
# no more call to obtain a token, get it from cache
rsps.get(
'https://r2p.invalid/r2p/v1/personne/1257553562447',
status=200,
json={
'personnePhysique': {
'identifiant': {'spi': '1257553562447'},
'etatCivil': {'cdSexe': '1', 'lbNomNaissance': 'SIMON'},
}
},
)
resp = app.get('/r2p/test/personne-by-spi?spi=1257553562447')
assert resp.json['err'] == 0
api_call = rsps.calls[0]
assert api_call.request.headers['Authorization'] == 'Bearer 1234'
def test_personne_by_spi(app, connector):
cache.clear()
with responses.RequestsMock() as rsps:
mock_tocken(rsps)
rsps.get(
'https://r2p.invalid/r2p/v1/personne/1257553562447',
status=200,
json={
'personnePhysique': {
'identifiant': {'spi': '1257553562447'},
'etatCivil': {'cdSexe': '1', 'lbNomNaissance': 'SIMON'},
}
},
)
resp = app.get('/r2p/test/personne-by-spi?spi=1257553562447')
json_resp = resp.json
assert json_resp['err'] == 0
assert json_resp['data']['personnePhysique']['identifiant']['spi'] == '1257553562447'
def test_personne_by_spi_error(app, connector):
cache.clear()
with responses.RequestsMock() as rsps:
mock_tocken(rsps)
rsps.get(
'https://r2p.invalid/r2p/v1/personne/123',
status=400,
json={'erreur': {'code': '40015', 'message': '[spi] Format du SPI erroné'}},
)
resp = app.get('/r2p/test/personne-by-spi?spi=123')
json_resp = resp.json
assert json_resp['err'] == 1
assert json_resp['err_class'] == 'passerelle.utils.jsonresponse.APIError'
assert (
json_resp['err_desc']
== '400 Client Error: Bad Request for url: https://r2p.invalid/r2p/v1/personne/123'
)
assert json_resp['data']['erreur']['code'] == '40015'
assert json_resp['data']['erreur']['message'] == '[spi] Format du SPI erroné'
def test_personne_by_criteria(app, connector):
cache.clear()
with responses.RequestsMock() as rsps:
mock_tocken(rsps)
params = {
'prenom': 'samuel',
'nom': 'garcia',
'sexe': '1',
'naisDate': '15/06/1943',
'naisCodePays': '99100',
'naisCodeCommune': '109',
'naisCodeDept': '75',
}
rsps.get(
'https://r2p.invalid/r2p/v1/personne',
status=200,
match=[responses.matchers.query_param_matcher(params)],
json={
'personnePhysique': {
'identifiant': {'spi': '7540305732558'},
'etatCivil': {
'cdSexe': '1',
'lbNomNaissance': 'garcia',
'lbPrenomNaissance': 'samuel',
'lbNomUsage': 'garcia',
'lbPrenomUsage': 'samuel',
'anneeNaissance': '1943',
'moisNaissance': '06',
'jourNaissance': '15',
'cdPaysNaissance': '99100',
'cdDeptNaissance': '75',
'cdTOMNaissance': None,
'cdCommuneNaissance': '109',
},
'adresse': {
'cdPays': '99100',
'lbPays': 'FRANCE',
'cdDepartement': '75',
'lbDepartement': 'PARIS',
'cdTOM': None,
'lbTOM': None,
'cdCommune': '112',
'lbCommune': 'PARIS',
'cdVoie': '0881',
'lbVoie': 'RUE DE BERCY',
'numeroVoie': '139',
'indiceDeRepetition': None,
},
}
},
)
resp = app.get('/r2p/test/personne-by-criteria', params=params)
json_resp = resp.json
assert json_resp['err'] == 0
assert json_resp['data'] == {
'personnePhysique': {
'identifiant': {'spi': '7540305732558'},
'etatCivil': {
'cdSexe': '1',
'lbNomNaissance': 'garcia',
'lbPrenomNaissance': 'samuel',
'lbNomUsage': 'garcia',
'lbPrenomUsage': 'samuel',
'anneeNaissance': '1943',
'moisNaissance': '06',
'jourNaissance': '15',
'cdPaysNaissance': '99100',
'cdDeptNaissance': '75',
'cdTOMNaissance': None,
'cdCommuneNaissance': '109',
},
'adresse': {
'cdPays': '99100',
'lbPays': 'FRANCE',
'cdDepartement': '75',
'lbDepartement': 'PARIS',
'cdTOM': None,
'lbTOM': None,
'cdCommune': '112',
'lbCommune': 'PARIS',
'cdVoie': '0881',
'lbVoie': 'RUE DE BERCY',
'numeroVoie': '139',
'indiceDeRepetition': None,
},
}
}
def test_personne_by_criteria_error(app, connector):
cache.clear()
with responses.RequestsMock() as rsps:
mock_tocken(rsps)
params = {
'prenom': 'samuel',
'nom': 'garcia',
'sexe': '1',
}
rsps.get(
'https://r2p.invalid/r2p/v1/personne',
status=400,
match=[responses.matchers.query_param_matcher(params)],
json={'erreur': {'code': '40005', 'message': '[naisDate] Date de naissance absente'}},
)
resp = app.get('/r2p/test/personne-by-criteria', params=params)
json_resp = resp.json
assert json_resp['err'] == 1
assert json_resp['data'] == {
'erreur': {'code': '40005', 'message': '[naisDate] Date de naissance absente'}
}
def test_spi_by_criteria(app, connector):
cache.clear()
with responses.RequestsMock() as rsps:
mock_tocken(rsps)
params = {
'prenom': 'samuel',
'nom': 'garcia',
'sexe': '1',
'naisDate': '15/06/1943',
'naisCodePays': '99100',
'naisCodeCommune': '109',
'naisCodeDept': '75',
}
rsps.get(
'https://r2p.invalid/r2p/v1/personne/spi',
status=200,
match=[responses.matchers.query_param_matcher(params)],
json={
'personnePhysique': {
'identifiant': {'spi': '7540305732558'},
'etatCivil': {
'cdSexe': '1',
'lbNomNaissance': 'garcia',
'lbPrenomNaissance': 'samuel',
'lbNomUsage': 'garcia',
'lbPrenomUsage': 'samuel',
'anneeNaissance': '1943',
'moisNaissance': '06',
'jourNaissance': '15',
'cdPaysNaissance': '99100',
'cdDeptNaissance': '75',
'cdTOMNaissance': None,
'cdCommuneNaissance': '109',
},
'adresse': {
'cdPays': '99100',
'lbPays': 'FRANCE',
'cdDepartement': '75',
'lbDepartement': 'PARIS',
'cdTOM': None,
'lbTOM': None,
'cdCommune': '112',
'lbCommune': 'PARIS',
'cdVoie': '0881',
'lbVoie': 'RUE DE BERCY',
'numeroVoie': '139',
'indiceDeRepetition': None,
},
}
},
)
resp = app.get('/r2p/test/spi-by-criteria', params=params)
json_resp = resp.json
assert json_resp['err'] == 0
assert json_resp['data'] == {
'personnePhysique': {
'identifiant': {'spi': '7540305732558'},
'etatCivil': {
'cdSexe': '1',
'lbNomNaissance': 'garcia',
'lbPrenomNaissance': 'samuel',
'lbNomUsage': 'garcia',
'lbPrenomUsage': 'samuel',
'anneeNaissance': '1943',
'moisNaissance': '06',
'jourNaissance': '15',
'cdPaysNaissance': '99100',
'cdDeptNaissance': '75',
'cdTOMNaissance': None,
'cdCommuneNaissance': '109',
},
'adresse': {
'cdPays': '99100',
'lbPays': 'FRANCE',
'cdDepartement': '75',
'lbDepartement': 'PARIS',
'cdTOM': None,
'lbTOM': None,
'cdCommune': '112',
'lbCommune': 'PARIS',
'cdVoie': '0881',
'lbVoie': 'RUE DE BERCY',
'numeroVoie': '139',
'indiceDeRepetition': None,
},
}
}