Compare commits
30 Commits
183be84c7d
...
72e8bb0c0b
Author | SHA1 | Date |
---|---|---|
Nicolas Roche | 72e8bb0c0b | |
Nicolas Roche | a5bef170be | |
Nicolas Roche | da77201197 | |
Nicolas Roche | 2bbee39441 | |
Nicolas Roche | 62d7af4d4b | |
Nicolas Roche | 1eb705e9ea | |
Nicolas Roche | 0dedbe3b96 | |
Nicolas Roche | adcc9adf1d | |
Nicolas Roche | af18ba7acd | |
Nicolas Roche | 94f8c95bf8 | |
Nicolas Roche | d8107ea4f5 | |
Nicolas Roche | 3aeebd8842 | |
Nicolas Roche | 6c39c8dea5 | |
Nicolas Roche | 47350bdc00 | |
Nicolas Roche | 50e8ec7ebe | |
Nicolas Roche | 5ffaaba031 | |
Nicolas Roche | e3aee14160 | |
Nicolas Roche | 150f96251a | |
Nicolas Roche | 276c760da4 | |
Nicolas Roche | 3424d98969 | |
Nicolas Roche | 1f81071278 | |
Nicolas Roche | 5804ed6616 | |
Frédéric Péters | e75e5287d9 | |
Serghei Mihai | b464442b6b | |
Serghei Mihai | 48db73f027 | |
Serghei Mihai | e2df9aec5b | |
Frédéric Péters | d07523b3c4 | |
Lauréline Guérin | d0bdb9b618 | |
Frédéric Péters | 9b889430d1 | |
Agate | 56b6baea78 |
|
@ -1,9 +0,0 @@
|
||||||
import pytest
|
|
||||||
|
|
||||||
|
|
||||||
def pytest_addoption(parser):
|
|
||||||
parser.addoption("--cmis-connector-url", help="Url of a passerelle CMIS connector instance")
|
|
||||||
parser.addoption("--cmis-endpoint", help="Url of a passerelle CMIS endpoint")
|
|
||||||
parser.addoption("--cmis-username", help="Username for the CMIS endpoint")
|
|
||||||
parser.addoption("--cmis-password", help="Password for the CMIS endpoint")
|
|
||||||
parser.addoption("--preserve-tree", action="store_true", default=False, help="Preserve test directory")
|
|
|
@ -72,7 +72,7 @@ class Esabora(BaseResource, HTTPResource):
|
||||||
url = urllib.parse.urljoin(self.service_url, path)
|
url = urllib.parse.urljoin(self.service_url, path)
|
||||||
headers = {'Authorization': f'Bearer {self.api_key}'}
|
headers = {'Authorization': f'Bearer {self.api_key}'}
|
||||||
try:
|
try:
|
||||||
return self.requests.post(url, json=payload, headers=headers, timeout=5, **kwargs)
|
response = self.requests.post(url, json=payload, headers=headers, timeout=5, **kwargs)
|
||||||
except requests.RequestException as e:
|
except requests.RequestException as e:
|
||||||
raise APIError(
|
raise APIError(
|
||||||
'Esabora platform "%s" connection error: %s' % (self.service_url, exception_to_text(e)),
|
'Esabora platform "%s" connection error: %s' % (self.service_url, exception_to_text(e)),
|
||||||
|
@ -83,6 +83,26 @@ class Esabora(BaseResource, HTTPResource):
|
||||||
'error': str(e),
|
'error': str(e),
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
try:
|
||||||
|
data = response.json()
|
||||||
|
except requests.JSONDecodeError as e:
|
||||||
|
raise APIError(
|
||||||
|
'Esabora platform "%s" invalid JSON response: %s' % (self.service_url, exception_to_text(e)),
|
||||||
|
log_error=True,
|
||||||
|
data={
|
||||||
|
'status_code': response.status_code,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
if not response.ok:
|
||||||
|
raise APIError(
|
||||||
|
'Esabora platform "%s" answered with HTTP error' % (self.service_url),
|
||||||
|
log_error=True,
|
||||||
|
data={
|
||||||
|
'status_code': response.status_code,
|
||||||
|
'content': data,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
return data
|
||||||
|
|
||||||
@endpoint(
|
@endpoint(
|
||||||
name='do-search',
|
name='do-search',
|
||||||
|
@ -100,9 +120,7 @@ class Esabora(BaseResource, HTTPResource):
|
||||||
for name, value in post_data['criterions'].items()
|
for name, value in post_data['criterions'].items()
|
||||||
],
|
],
|
||||||
}
|
}
|
||||||
response = self.post('mult/', payload, params={'task': 'doSearch'})
|
data = self.post('mult/', payload, params={'task': 'doSearch'})
|
||||||
response.raise_for_status()
|
|
||||||
data = response.json()
|
|
||||||
columns = {slugify(c).replace('-', '_'): c for c in data['columnList']}
|
columns = {slugify(c).replace('-', '_'): c for c in data['columnList']}
|
||||||
keys = {slugify(c).replace('-', '_'): c for c in data['keyList']}
|
keys = {slugify(c).replace('-', '_'): c for c in data['keyList']}
|
||||||
cleaned_data = {
|
cleaned_data = {
|
||||||
|
@ -130,9 +148,7 @@ class Esabora(BaseResource, HTTPResource):
|
||||||
endpoint = post_data.pop('endpoint', None) or 'modbdd'
|
endpoint = post_data.pop('endpoint', None) or 'modbdd'
|
||||||
payload = get_treatment_payload(post_data)
|
payload = get_treatment_payload(post_data)
|
||||||
|
|
||||||
response = self.post(f'{endpoint}/', payload, params={'task': 'doTreatment'})
|
data = self.post(f'{endpoint}/', payload, params={'task': 'doTreatment'})
|
||||||
response.raise_for_status()
|
|
||||||
data = response.json()
|
|
||||||
keys = [slugify(c).replace('-', '_') for c in data['keyList']]
|
keys = [slugify(c).replace('-', '_') for c in data['keyList']]
|
||||||
cleaned_data = esabora_row_to_object([], keys, data)
|
cleaned_data = esabora_row_to_object([], keys, data)
|
||||||
cleaned_data['action'] = data['action']
|
cleaned_data['action'] = data['action']
|
||||||
|
|
|
@ -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,
|
||||||
|
@ -222,11 +230,13 @@ class Resource(BaseResource):
|
||||||
)
|
)
|
||||||
def data_source(self, request, id=None, test=None, mode=None, **kwargs):
|
def data_source(self, request, id=None, test=None, mode=None, **kwargs):
|
||||||
if id:
|
if id:
|
||||||
return {
|
token = self.retrieve(id)
|
||||||
'data': [
|
if token:
|
||||||
dict(self.retrieve(id), id=id),
|
return {
|
||||||
]
|
'data': [
|
||||||
}
|
dict(token, id=id),
|
||||||
|
]
|
||||||
|
}
|
||||||
url = request.build_absolute_uri(
|
url = request.build_absolute_uri(
|
||||||
reverse(
|
reverse(
|
||||||
'generic-endpoint',
|
'generic-endpoint',
|
||||||
|
@ -262,6 +272,10 @@ class Resource(BaseResource):
|
||||||
return token.guid.hex
|
return token.guid.hex
|
||||||
|
|
||||||
def retrieve(self, ref):
|
def retrieve(self, ref):
|
||||||
|
try:
|
||||||
|
ref = uuid.UUID(str(ref))
|
||||||
|
except ValueError:
|
||||||
|
return None
|
||||||
token = Token.objects.filter(guid=ref).first()
|
token = Token.objects.filter(guid=ref).first()
|
||||||
return token and token.content
|
return token and token.content
|
||||||
|
|
||||||
|
|
|
@ -131,6 +131,7 @@ class GrandLyonStreetSections(BaseResource):
|
||||||
)
|
)
|
||||||
for attribute in ('nom', 'nomcommune', 'domanialite', 'codeinsee'):
|
for attribute in ('nom', 'nomcommune', 'domanialite', 'codeinsee'):
|
||||||
setattr(section, attribute, value.get(attribute) or '')
|
setattr(section, attribute, value.get(attribute) or '')
|
||||||
|
section.nomcommune = normalize_commune(section.nomcommune)
|
||||||
for attribute in ('bornemindroite', 'bornemingauche', 'bornemaxdroite', 'bornemaxgauche', 'gid'):
|
for attribute in ('bornemindroite', 'bornemingauche', 'bornemaxdroite', 'bornemaxgauche', 'gid'):
|
||||||
if value.get(attribute) in (None, 'None'):
|
if value.get(attribute) in (None, 'None'):
|
||||||
# data.grandlyon returned 'None' as a string at a time
|
# data.grandlyon returned 'None' as a string at a time
|
||||||
|
|
|
@ -131,6 +131,10 @@ class Tcl(BaseResource):
|
||||||
response = self.requests.get(url)
|
response = self.requests.get(url)
|
||||||
response.raise_for_status()
|
response.raise_for_status()
|
||||||
for line_data in response.json()['values']:
|
for line_data in response.json()['values']:
|
||||||
|
if 'code_trace' in line_data:
|
||||||
|
line_data['code_titan'] = line_data['code_trace']
|
||||||
|
if 'nom_trace' in line_data:
|
||||||
|
line_data['libelle'] = line_data['nom_trace']
|
||||||
line, dummy = Line.objects.get_or_create(
|
line, dummy = Line.objects.get_or_create(
|
||||||
code_titan=line_data['code_titan'],
|
code_titan=line_data['code_titan'],
|
||||||
defaults={'transport_key': key, 'ligne': line_data['ligne']},
|
defaults={'transport_key': key, 'ligne': line_data['ligne']},
|
||||||
|
|
|
@ -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"
|
||||||
|
|
|
@ -245,3 +245,31 @@ def test_do_treatment_arbitrary_endpoint(app, connector):
|
||||||
assert responses.calls[0].request.params['task'] == 'doTreatment'
|
assert responses.calls[0].request.params['task'] == 'doTreatment'
|
||||||
response_data = json.loads(responses.calls[0].request.body)
|
response_data = json.loads(responses.calls[0].request.body)
|
||||||
assert response_data == expected_payload
|
assert response_data == expected_payload
|
||||||
|
|
||||||
|
|
||||||
|
@responses.activate
|
||||||
|
def test_post_raises_proper_error(app, connector):
|
||||||
|
url = tests.utils.generic_endpoint_url('esabora', 'do-treatment')
|
||||||
|
responses.add(
|
||||||
|
responses.POST,
|
||||||
|
f'{connector.service_url}addevt/',
|
||||||
|
json={'foo': 'bar'},
|
||||||
|
status=400,
|
||||||
|
)
|
||||||
|
|
||||||
|
payload = {
|
||||||
|
'endpoint': 'addevt',
|
||||||
|
'treatment_name': 'Import Event',
|
||||||
|
'Adresse_Latitude': 12.3,
|
||||||
|
'Adresse_Ville': 'Marseille',
|
||||||
|
}
|
||||||
|
|
||||||
|
expected_response = {
|
||||||
|
'err': 1,
|
||||||
|
'err_class': 'passerelle.utils.jsonresponse.APIError',
|
||||||
|
'data': {'content': {'foo': 'bar'}, 'status_code': 400},
|
||||||
|
'err_desc': 'Esabora platform "http://example.esabora/ws/rest/" answered with HTTP error',
|
||||||
|
}
|
||||||
|
|
||||||
|
response = app.post_json(url, params=payload)
|
||||||
|
assert response.json == expected_response
|
||||||
|
|
|
@ -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'
|
||||||
|
@ -95,6 +122,8 @@ def test_callback(app, fc):
|
||||||
'text': 'John Doe né le April 28, 2001',
|
'text': 'John Doe né le April 28, 2001',
|
||||||
}
|
}
|
||||||
|
|
||||||
|
app.get('/franceconnect-data/test/data_source?id=bad') # no error
|
||||||
|
|
||||||
|
|
||||||
@mock_response(
|
@mock_response(
|
||||||
['/api/v1/token', ''],
|
['/api/v1/token', ''],
|
||||||
|
@ -119,3 +148,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)]
|
||||||
|
|
|
@ -209,6 +209,33 @@ DATA_EXAMPLE = """{
|
||||||
"senscircspecialise" : null,
|
"senscircspecialise" : null,
|
||||||
"senscirculation" : "Double",
|
"senscirculation" : "Double",
|
||||||
"typecirculation" : "générale"
|
"typecirculation" : "générale"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"bornemaxdroite" : 10,
|
||||||
|
"bornemaxgauche" : 11,
|
||||||
|
"bornemindroite" : 2,
|
||||||
|
"bornemingauche" : 1,
|
||||||
|
"codefuv" : "21424",
|
||||||
|
"codeinsee" : "69266",
|
||||||
|
"codetroncon" : "T54753",
|
||||||
|
"datecreation" : "1995-07-17 18:43:00+02:00",
|
||||||
|
"datedomanialite" : null,
|
||||||
|
"datemajalpha" : "2022-11-14",
|
||||||
|
"datemajborne" : "2022-07-26",
|
||||||
|
"datemajgraph" : "2022-07-26",
|
||||||
|
"denomroutiere" : null,
|
||||||
|
"domanialite" : "Métropole",
|
||||||
|
"gestionnaire" : null,
|
||||||
|
"gid" : 30067,
|
||||||
|
"importance" : "Petite rue",
|
||||||
|
"nom" : "Rue de l'Avenir",
|
||||||
|
"nomcommune" : "Villeurbanne",
|
||||||
|
"observation" : null,
|
||||||
|
"particularite" : null,
|
||||||
|
"referencedomanialite" : null,
|
||||||
|
"senscircspecialise" : null,
|
||||||
|
"senscirculation" : "Inverse",
|
||||||
|
"typecirculation" : "Générale"
|
||||||
}
|
}
|
||||||
]
|
]
|
||||||
}
|
}
|
||||||
|
@ -238,3 +265,15 @@ def test_daily_none_bornes(app, connector):
|
||||||
connector.daily()
|
connector.daily()
|
||||||
assert StreetSection.objects.get(codetroncon='T5869').bornemindroite == 0
|
assert StreetSection.objects.get(codetroncon='T5869').bornemindroite == 0
|
||||||
assert StreetSection.objects.get(codetroncon='T5869').bornemaxdroite == 99999
|
assert StreetSection.objects.get(codetroncon='T5869').bornemaxdroite == 99999
|
||||||
|
|
||||||
|
|
||||||
|
def test_non_uppercase_communes(app, connector):
|
||||||
|
StreetSection.objects.all().delete()
|
||||||
|
with HTTMock(data_mock):
|
||||||
|
connector.daily()
|
||||||
|
response = app.get(
|
||||||
|
"/grandlyon-streetsections/gl-streetsections/section_info"
|
||||||
|
"?streetname=Rue de l'Avenir&commune=Villeurbanne&streetnumber=8"
|
||||||
|
)
|
||||||
|
assert response.json['err'] == 0
|
||||||
|
assert response.json['data']['codetroncon'] == 'T54753'
|
||||||
|
|
|
@ -10,16 +10,12 @@ from passerelle.contrib.tcl.models import Line, Stop, Tcl
|
||||||
LIGNE_BUS = {
|
LIGNE_BUS = {
|
||||||
"values": [
|
"values": [
|
||||||
{
|
{
|
||||||
"indice": "",
|
|
||||||
"last_update_fme": "2017-06-27 06:01:10",
|
"last_update_fme": "2017-06-27 06:01:10",
|
||||||
"infos": "",
|
"nom_trace": "Croix Rousse - Plateaux de St Rambert",
|
||||||
"couleur": "164 203 38",
|
|
||||||
"libelle": "Croix Rousse - Plateaux de St Rambert",
|
|
||||||
"last_update": "None",
|
"last_update": "None",
|
||||||
"code_titan": "2Aa1",
|
"code_trace": "2Aa1",
|
||||||
"gid": "1003",
|
"gid": "1003",
|
||||||
"ligne": "2",
|
"ligne": "2",
|
||||||
"ut": "UTV",
|
|
||||||
"sens": "Aller",
|
"sens": "Aller",
|
||||||
},
|
},
|
||||||
]
|
]
|
||||||
|
|
Loading…
Reference in New Issue