franceconnect_data: allow defining DGFIP API base URL (#71638) #5
|
@ -60,9 +60,10 @@ def base64url_decode(input):
|
||||||
|
|
||||||
|
|
||||||
class FranceConnect:
|
class FranceConnect:
|
||||||
def __init__(self, session, logger):
|
def __init__(self, session, logger, dgfip_api_base_url):
|
||||||
self.session = session
|
self.session = session
|
||||||
self.logger = logger
|
self.logger = logger
|
||||||
|
self.dgfip_api_base_url = dgfip_api_base_url
|
||||||
self.items = []
|
self.items = []
|
||||||
self.correlation_id = str(uuid.uuid4())
|
self.correlation_id = str(uuid.uuid4())
|
||||||
|
|
||||||
|
@ -135,7 +136,7 @@ class FranceConnect:
|
||||||
dgfip_response = self.request(
|
dgfip_response = self.request(
|
||||||
'dgfip token endpoint',
|
'dgfip token endpoint',
|
||||||
'POST',
|
'POST',
|
||||||
'https://gwfc.impots.gouv.fr/token',
|
'token',
|
||||||
data=data,
|
data=data,
|
||||||
auth=(dgfip_username, dgfip_password),
|
auth=(dgfip_username, dgfip_password),
|
||||||
)
|
)
|
||||||
|
@ -162,7 +163,7 @@ class FranceConnect:
|
||||||
dgfip_ressource_ir_response = self.request(
|
dgfip_ressource_ir_response = self.request(
|
||||||
'ressource IR endpoint',
|
'ressource IR endpoint',
|
||||||
'GET',
|
'GET',
|
||||||
'https://gwfc.impots.gouv.fr/impotparticulier/1.0/situations/ir/assiettes/annrev/%s' % annrev,
|
'impotparticulier/1.0/situations/ir/assiettes/annrev/%s' % annrev,
|
||||||
headers=headers,
|
headers=headers,
|
||||||
)
|
)
|
||||||
except FranceConnectError as e:
|
except FranceConnectError as e:
|
||||||
|
@ -185,8 +186,9 @@ class FranceConnect:
|
||||||
def add(self, key, value):
|
def add(self, key, value):
|
||||||
self.items.append((key, value))
|
self.items.append((key, value))
|
||||||
|
|
||||||
def request(self, label, method, url, *args, **kwargs):
|
def request(self, label, method, endpoint, *args, **kwargs):
|
||||||
self.logger.debug('request %s %s args:%s kwargs:%s', label, method, args, kwargs)
|
self.logger.debug('request %s %s args:%s kwargs:%s', label, method, args, kwargs)
|
||||||
|
url = urllib.parse.urljoin(self.dgfip_api_base_url, endpoint)
|
||||||
self.add(label.replace(' ', '_') + '_request', [method, url, args, kwargs])
|
self.add(label.replace(' ', '_') + '_request', [method, url, args, kwargs])
|
||||||
try:
|
try:
|
||||||
response = getattr(self.session, method.lower())(url, *args, **kwargs)
|
response = getattr(self.session, method.lower())(url, *args, **kwargs)
|
||||||
|
|
|
@ -48,23 +48,19 @@ class Migration(migrations.Migration):
|
||||||
),
|
),
|
||||||
(
|
(
|
||||||
'dgfip_username',
|
'dgfip_username',
|
||||||
models.CharField(
|
models.CharField(blank=True, max_length=64, null=True, verbose_name='DGFIP API Username'),
|
||||||
blank=True, max_length=64, null=True, verbose_name='api.impots.gouv.fr username'
|
|
||||||
),
|
|
||||||
),
|
),
|
||||||
(
|
(
|
||||||
'dgfip_password',
|
'dgfip_password',
|
||||||
models.CharField(
|
models.CharField(blank=True, max_length=64, null=True, verbose_name='DGFIP API Password'),
|
||||||
blank=True, max_length=64, null=True, verbose_name='api.impots.gouv.fr password'
|
|
||||||
),
|
|
||||||
),
|
),
|
||||||
(
|
(
|
||||||
'dgfip_scopes',
|
'dgfip_scopes',
|
||||||
models.TextField(blank=True, null=True, verbose_name='api.impots.gouv.fr scopes'),
|
models.TextField(blank=True, null=True, verbose_name='DGFIP API Scopes'),
|
||||||
),
|
),
|
||||||
(
|
(
|
||||||
'dgfip_id_teleservice',
|
'dgfip_id_teleservice',
|
||||||
models.TextField(blank=True, null=True, verbose_name='api.impots.gouv.fr ID_Teleservice'),
|
models.TextField(blank=True, null=True, verbose_name='DGFIP API ID_Teleservice'),
|
||||||
),
|
),
|
||||||
(
|
(
|
||||||
'users',
|
'users',
|
||||||
|
|
|
@ -0,0 +1,22 @@
|
||||||
|
# Generated by Django 2.2.24 on 2022-11-24 10:58
|
||||||
|
|
||||||
|
from django.db import migrations, models
|
||||||
|
|
||||||
|
|
||||||
|
class Migration(migrations.Migration):
|
||||||
|
|
||||||
|
dependencies = [
|
||||||
|
('franceconnect_data', '0002_token'),
|
||||||
|
]
|
||||||
|
|
||||||
|
operations = [
|
||||||
|
migrations.AddField(
|
||||||
|
model_name='resource',
|
||||||
|
name='dgfip_api_base_url',
|
||||||
|
field=models.URLField(
|
||||||
|
default='https://gwfc.dgfip.finances.gouv.fr/',
|
||||||
|
max_length=256,
|
||||||
|
verbose_name='DGFIP API base URL',
|
||||||
|
),
|
||||||
|
),
|
||||||
|
]
|
|
@ -62,13 +62,17 @@ class Resource(BaseResource):
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
dgfip_username = models.CharField(_('api.impots.gouv.fr username'), max_length=64, blank=True, null=True)
|
dgfip_api_base_url = models.URLField(
|
||||||
|
_('DGFIP API base URL'), max_length=256, default='https://gwfc.dgfip.finances.gouv.fr/'
|
||||||
|
)
|
||||||
|
|
||||||
dgfip_password = models.CharField(_('api.impots.gouv.fr password'), max_length=64, blank=True, null=True)
|
dgfip_username = models.CharField(_('DGFIP API Username'), max_length=64, blank=True, null=True)
|
||||||
|
|
||||||
dgfip_scopes = models.TextField(_('api.impots.gouv.fr scopes'), blank=True, null=True)
|
dgfip_password = models.CharField(_('DGFIP API Password'), max_length=64, blank=True, null=True)
|
||||||
|
|
||||||
dgfip_id_teleservice = models.TextField(_('api.impots.gouv.fr ID_Teleservice'), blank=True, null=True)
|
dgfip_scopes = models.TextField(_('DGFIP API Scopes'), blank=True, null=True)
|
||||||
|
|
||||||
|
dgfip_id_teleservice = models.TextField(_('DGFIP API ID_Teleservice'), blank=True, null=True)
|
||||||
|
|
||||||
log_requests_errors = False
|
log_requests_errors = False
|
||||||
|
|
||||||
|
@ -121,7 +125,9 @@ class Resource(BaseResource):
|
||||||
return HttpResponseBadRequest('Missing or invalid origin')
|
return HttpResponseBadRequest('Missing or invalid origin')
|
||||||
|
|
||||||
redirect_uri = self.build_callback_url(request, origin=origin, mode=mode, test=test)
|
redirect_uri = self.build_callback_url(request, origin=origin, mode=mode, test=test)
|
||||||
franceconnect = fc.FranceConnect(session=self.requests, logger=self.logger)
|
franceconnect = fc.FranceConnect(
|
||||||
|
session=self.requests, logger=self.logger, dgfip_api_base_url=self.dgfip_api_base_url
|
||||||
|
)
|
||||||
return HttpResponseRedirect(
|
return HttpResponseRedirect(
|
||||||
franceconnect.authorization_request(
|
franceconnect.authorization_request(
|
||||||
platform=self.fc_platform,
|
platform=self.fc_platform,
|
||||||
|
@ -152,7 +158,9 @@ class Resource(BaseResource):
|
||||||
if test and not request.user.is_superuser:
|
if test and not request.user.is_superuser:
|
||||||
return HttpResponseBadRequest('Only admin can use test mode.')
|
return HttpResponseBadRequest('Only admin can use test mode.')
|
||||||
|
|
||||||
franceconnect = fc.FranceConnect(session=self.requests, logger=self.logger)
|
franceconnect = fc.FranceConnect(
|
||||||
|
session=self.requests, logger=self.logger, dgfip_api_base_url=self.dgfip_api_base_url
|
||||||
|
)
|
||||||
redirect_uri = self.build_callback_url(request, origin=origin, mode=mode, test=test)
|
redirect_uri = self.build_callback_url(request, origin=origin, mode=mode, test=test)
|
||||||
context = {
|
context = {
|
||||||
'origin': origin,
|
'origin': origin,
|
||||||
|
|
|
@ -7,7 +7,7 @@ msgid ""
|
||||||
msgstr ""
|
msgstr ""
|
||||||
"Project-Id-Version: Passerelle 0\n"
|
"Project-Id-Version: Passerelle 0\n"
|
||||||
"Report-Msgid-Bugs-To: \n"
|
"Report-Msgid-Bugs-To: \n"
|
||||||
"POT-Creation-Date: 2022-11-03 16:06-0500\n"
|
"POT-Creation-Date: 2022-11-28 15:13+0100\n"
|
||||||
"PO-Revision-Date: 2022-11-03 22:06+0100\n"
|
"PO-Revision-Date: 2022-11-03 22:06+0100\n"
|
||||||
"Last-Translator: Frederic Peters <fpeters@entrouvert.com>\n"
|
"Last-Translator: Frederic Peters <fpeters@entrouvert.com>\n"
|
||||||
"Language: fr\n"
|
"Language: fr\n"
|
||||||
|
@ -2390,20 +2390,25 @@ msgid "FranceConnect scopes"
|
||||||
msgstr "Domaines (« scopes ») pour FranceConnect"
|
msgstr "Domaines (« scopes ») pour FranceConnect"
|
||||||
|
|
||||||
#: apps/franceconnect_data/models.py
|
#: apps/franceconnect_data/models.py
|
||||||
msgid "api.impots.gouv.fr username"
|
msgid "DGFIP API base URL"
|
||||||
msgstr "identifiant pour api.impots.gouv.fr"
|
msgstr "URL de base de l’API"
|
||||||
|
|
||||||
#: apps/franceconnect_data/models.py
|
#: apps/franceconnect_data/models.py
|
||||||
msgid "api.impots.gouv.fr password"
|
msgid "DGFIP API Username"
|
||||||
msgstr "mot de passe pour api.impots.gouv.fr"
|
msgstr "Identifiant pour l’API DGFIP"
|
||||||
|
|
||||||
#: apps/franceconnect_data/models.py
|
#: apps/franceconnect_data/models.py
|
||||||
msgid "api.impots.gouv.fr scopes"
|
|
||||||
msgstr "Domaines (« scopes ») pour api.impots.gouv.fr"
|
msgid "DGFIP API Password"
|
||||||
|
msgstr "Mot de passe pour l’API DGFIP"
|
||||||
|
|
||||||
#: apps/franceconnect_data/models.py
|
#: apps/franceconnect_data/models.py
|
||||||
msgid "api.impots.gouv.fr ID_Teleservice"
|
msgid "DGFIP API Scopes"
|
||||||
msgstr "ID_Teleservice pour api.impots.gouv.fr"
|
msgstr "Scopes de l’API DGFIP"
|
||||||
|
|
||||||
|
#: apps/franceconnect_data/models.py
|
||||||
|
msgid "DGFIP API ID_Teleservice"
|
||||||
|
msgstr "ID_Teleservice pour l’API DGFIP"
|
||||||
|
|
||||||
#: apps/franceconnect_data/models.py
|
#: apps/franceconnect_data/models.py
|
||||||
msgid "Data sources through FranceConnect"
|
msgid "Data sources through FranceConnect"
|
||||||
|
|
|
@ -18,11 +18,52 @@ import json
|
||||||
from urllib.parse import parse_qs, urlparse, urlunparse
|
from urllib.parse import parse_qs, urlparse, urlunparse
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
from django.utils.timezone import now
|
||||||
|
|
||||||
import tests.utils
|
import tests.utils
|
||||||
from passerelle.apps.franceconnect_data.models import Resource
|
from passerelle.apps.franceconnect_data.models import Resource
|
||||||
from tests.test_rsa13 import mock_response
|
from tests.test_rsa13 import mock_response
|
||||||
|
|
||||||
|
CURRENT_YEAR = now().year
|
||||||
|
|
||||||
|
|
||||||
|
USER_INFO_MOCKED_RESPONSES = [
|
||||||
|
['/api/v1/token', {'access_token': 'at-1234', 'id_token': '.e30=.'}],
|
||||||
|
[
|
||||||
|
'/api/v1/userinfo',
|
||||||
|
{
|
||||||
|
'sub': 'sub-1234',
|
||||||
|
'given_name': 'John',
|
||||||
|
'family_name': 'Doe',
|
||||||
|
'birthdate': '2001-04-28',
|
||||||
|
'birthplace': '13055',
|
||||||
|
'birthcountry': '99100',
|
||||||
|
'gender': 'male',
|
||||||
|
},
|
||||||
|
],
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
DGFIP_MOCKED_RESPONSES = USER_INFO_MOCKED_RESPONSES + [
|
||||||
|
[
|
||||||
|
'/token',
|
||||||
|
{
|
||||||
|
'access_token': 'eyJ4NXQiOi',
|
||||||
|
'expires_in': 3600,
|
||||||
|
'scope': 'RessourceIRDerniere2',
|
||||||
|
'token_type': 'Bearer',
|
||||||
|
},
|
||||||
|
]
|
||||||
|
]
|
||||||
|
|
||||||
|
DGFIP_MOCKED_RESPONSES += [
|
||||||
|
[
|
||||||
|
'/impotparticulier/1.0/situations/ir/assiettes/annrev/%s' % year,
|
||||||
|
{'rfr': 0, 'revenuBrutGlobal': 0},
|
||||||
|
]
|
||||||
|
for year in range(CURRENT_YEAR - 3, CURRENT_YEAR)
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def fc(db):
|
def fc(db):
|
||||||
|
@ -54,21 +95,7 @@ def test_init_request(app, fc):
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@mock_response(
|
@mock_response(*USER_INFO_MOCKED_RESPONSES)
|
||||||
['/api/v1/token', {'access_token': 'at-1234', 'id_token': '.e30=.'}],
|
|
||||||
[
|
|
||||||
'/api/v1/userinfo',
|
|
||||||
{
|
|
||||||
'sub': 'sub-1234',
|
|
||||||
'given_name': 'John',
|
|
||||||
'family_name': 'Doe',
|
|
||||||
'birthdate': '2001-04-28',
|
|
||||||
'birthplace': '13055',
|
|
||||||
'birthcountry': '99100',
|
|
||||||
'gender': 'male',
|
|
||||||
},
|
|
||||||
],
|
|
||||||
)
|
|
||||||
def test_callback(app, fc):
|
def test_callback(app, fc):
|
||||||
resp = app.get(
|
resp = app.get(
|
||||||
'http://testserver/franceconnect-data/test/callback?origin=http%3A%2F%2Ftestserver&code=5678&raise=1'
|
'http://testserver/franceconnect-data/test/callback?origin=http%3A%2F%2Ftestserver&code=5678&raise=1'
|
||||||
|
@ -119,3 +146,20 @@ def test_callback_error(app, fc):
|
||||||
error = json.loads(resp.pyquery('#error').text())
|
error = json.loads(resp.pyquery('#error').text())
|
||||||
assert error
|
assert error
|
||||||
assert 'Error in token endpoint response' in resp
|
assert 'Error in token endpoint response' in resp
|
||||||
|
|
||||||
|
|
||||||
|
@mock_response(*DGFIP_MOCKED_RESPONSES)
|
||||||
|
def test_dgfip_mode(app, fc):
|
||||||
|
resp = app.get(
|
||||||
|
'http://testserver/franceconnect-data/test/callback?origin=http%3A%2F%2Ftestserver&code=5678&raise=1&mode=dgfip'
|
||||||
|
)
|
||||||
|
data = json.loads(resp.pyquery('#data').text())
|
||||||
|
assert data
|
||||||
|
assert 'id' in data
|
||||||
|
assert data['text'] == 'John Doe né le April 28, 2001'
|
||||||
|
|
||||||
|
resp = app.get('/franceconnect-data/test/data_source?mode=dgfip&id=' + data['id'])
|
||||||
|
data = resp.json['data'][0]
|
||||||
|
assert data['dgfip_ir']
|
||||||
|
for year in range(CURRENT_YEAR - 3, CURRENT_YEAR):
|
||||||
|
assert data['dgfip_ir'][str(year)]
|
||||||
|
|
Loading…
Reference in New Issue