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

Merge branch 'main' into wip/73808-parsifal-site-service-fixture
gitea-wip/passerelle/pipeline/pr-main This commit looks good Details

This commit is contained in:
Nicolas Roche 2023-01-31 18:25:34 +01:00
parent 5bbd8256a6 70e03e7721
commit b0f495d0e2
21 changed files with 2127 additions and 933 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
passerelle/apps/cmis/__init__.py
View File

@ -1,17 +0,0 @@
# 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/>.
default_app_config = 'passerelle.apps.cmis.apps.CmisAppConfig'

62
passerelle/apps/cmis/apps.py
View File

@ -1,62 +0,0 @@
# 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 django.apps import AppConfig
def add_logging_to_cmislib():
'''Monkeypatch cmislib request module to log requests and responses.'''
from cmislib.atompub import binding
from cmislib.net import RESTService as CMISRESTService
class RESTService(CMISRESTService):
def get(self, url, *args, **kwargs):
logger = kwargs.pop('passerelle_logger')
logger.debug('cmislib GET request to %s', url)
resp, content = super().get(url, *args, **kwargs)
logger.debug('cmislib GET response (%s)', resp['status'], extra={'response': content.decode()})
return resp, content
def delete(self, url, *args, **kwargs):
logger = kwargs.pop('passerelle_logger')
logger.debug('cmislib DELETE request to %s', url)
resp, content = super().delete(url, *args, **kwargs)
logger.debug('cmislib DELETE response (%s)', resp['status'], extra={'response': content.decode()})
return resp, content
def post(self, url, payload, *args, **kwargs):
logger = kwargs.pop('passerelle_logger')
logger.debug('cmislib POST request to %s', url, extra={'payload': payload.decode()})
resp, content = super().post(url, payload, *args, **kwargs)
logger.debug('cmislib POST response (%s)', resp['status'], extra={'response': content.decode()})
return resp, content
def put(self, url, payload, *args, **kwargs):
logger = kwargs.pop('passerelle_logger')
logger.debug('cmislib PUT request to %s', url, payload, extra={'payload': payload.decode()})
resp, content = super().put(url, *args, **kwargs)
logger.debug('cmislib PUT response (%s)', resp['status'], extra={'response': content.decode()})
return resp, content
binding.Rest = RESTService
class CmisAppConfig(AppConfig):
name = 'passerelle.apps.cmis'
def ready(self):
add_logging_to_cmislib()

50
passerelle/apps/cmis/models.py
View File

@ -34,6 +34,7 @@ from cmislib.exceptions import (
from django.db import models
from django.http import HttpResponse
from django.utils.functional import cached_property
from django.utils.http import urlencode
from django.utils.translation import gettext_lazy as _
from passerelle.base.models import BaseResource
@ -147,7 +148,14 @@ class CmisConnector(BaseResource):
@contextmanager
def get_cmis_gateway(self):
with ignore_loggers('cmislib', 'cmislib.atompub.binding'):
yield CMISGateway(self.cmis_endpoint, self.username, self.password, self.logger)
import cmislib.atompub.binding as atompub_binding
old_Rest = atompub_binding.Rest
atompub_binding.Rest = lambda: RESTService(self)
try:
yield CMISGateway(self.cmis_endpoint, self.username, self.password, self.logger)
finally:
atompub_binding.Rest = old_Rest
def _validate_inputs(self, data):
"""process dict
@ -236,7 +244,7 @@ def wrap_cmis_error(f):
class CMISGateway:
def __init__(self, cmis_endpoint, username, password, logger):
self._cmis_client = CmisClient(cmis_endpoint, username, password, passerelle_logger=logger)
self._cmis_client = CmisClient(cmis_endpoint, username, password)
self._logger = logger
@cached_property
@ -284,3 +292,41 @@ class CMISGateway:
@wrap_cmis_error
def get_object(self, object_id):
return self.repo.getObject(object_id)
# Mock API from cmilib.net.RESTService
class RESTService:
def __init__(self, resource):
self.resource = resource
def request(self, method, url, username, password, body=None, content_type=None, **kwargs):
if username or password:
auth = (username, password)
else:
auth = None
headers = kwargs.pop('headers', {})
if kwargs:
url = url + ('&' if '?' in url else '?') + urlencode(kwargs)
if content_type:
headers['Content-Type'] = content_type
response = self.resource.requests.request(
method=method, url=url, auth=auth, headers=headers, data=body
)
return {'status': str(response.status_code)}, response.content
def get(self, url, username=None, password=None, **kwargs):
return self.request('GET', url, username, password, **kwargs)
def delete(self, url, username=None, password=None, **kwargs):
return self.request('DELETE', url, username, password, **kwargs)
def put(self, url, payload, contentType, username=None, password=None, **kwargs):
return self.request('PUT', url, username, password, body=payload, content_type=contentType, **kwargs)
def post(self, url, payload, contentType, username=None, password=None, **kwargs):
return self.request('POST', url, username, password, body=payload, content_type=contentType, **kwargs)

10
passerelle/apps/franceconnect_data/fc.py
View File

@ -177,6 +177,16 @@ class FranceConnect:
data[annrev] = dgfip_ressource_ir_response
self.add('dgfip_ressource_ir_response', data)
def request_dgfip_last_known_ir(self):
ir = {}
for year, response in self.dgfip_ressource_ir_response.items():
if response.get('error'):
continue
if ir.get('year', '0') < year:
ir = response
ir['year'] = year
self.add('dgfip_ressource_last_known_ir_response', ir)
def __getattr__(self, name):
try:
return dict(self.items)[name]

2
passerelle/apps/franceconnect_data/models.py
View File

@ -186,7 +186,9 @@ class Resource(BaseResource):
current_year = now().year
for year in range(current_year - 3, current_year):
franceconnect.request_dgfip_ir(str(year), id_teleservice=self.dgfip_id_teleservice)
franceconnect.request_dgfip_last_known_ir()
token['dgfip_ir'] = franceconnect.dgfip_ressource_ir_response
token['dgfip_ir']['last_known'] = franceconnect.dgfip_ressource_last_known_ir_response
try:
template = Template(self.text_template)
text_template_context = {

2
passerelle/apps/signal_arretes/models.py
View File

@ -87,6 +87,7 @@ REQUEST_SCHEMA = {
'format': 'date',
},
'occupation_end_date': {'description': _('Occupation end date'), 'type': 'string', 'format': 'date'},
'comment': {'description': _('Comment'), 'type': 'string'},
},
}
@ -218,6 +219,7 @@ class SignalArretes(BaseResource, HTTPResource):
'qualite': post_data['declarant_quality'],
'SIRET': post_data['declarant_siret'],
'numeroDossier': post_data['file_number'],
'commentaire': post_data['comment'],
'contact': {
'civilite': post_data['declarant_civility'],
'nom': post_data['declarant_name'],

10
passerelle/apps/vivaticket/models.py
View File

@ -188,7 +188,7 @@ class VivaTicket(BaseResource):
return self.requests.post(url, json=payload, headers=headers)
return response
def get_setting(self, endpoint, **kwargs):
def get_list_of_settings(self, endpoint, **kwargs):
response = self.get(endpoint, **kwargs)
json = response.json()
data = []
@ -198,25 +198,25 @@ class VivaTicket(BaseResource):
@endpoint(perm='can_access', methods=['get'], description=_('Get event categories'))
def events(self, request):
return self.get_setting('Settings/GetEventCategory')
return self.get_list_of_settings('Settings/GetEventCategory')
@endpoint(perm='can_access', methods=['get'], description=_('Get rooms'))
def rooms(self, request, event=None):
query = {}
if event is not None:
query['eventCategory'] = event
return self.get_setting('Settings/GetRooms', **query)
return self.get_list_of_settings('Settings/GetRooms', **query)
@endpoint(perm='can_access', methods=['get'], description=_('Get themes'))
def themes(self, request, room=None):
query = {}
if room is not None:
query['room'] = room
return self.get_setting('Settings/GetThemes', **query)
return self.get_list_of_settings('Settings/GetThemes', **query)
@endpoint(name='school-levels', perm='can_access', methods=['get'], description=_('Get school levels'))
def school_levels(self, request):
return self.get_setting('Settings/GetSchoolLevel')
return self.get_list_of_settings('Settings/GetSchoolLevel')
def get_or_create_contact(self, data, name_id=None):
contact_payload = {

10
passerelle/base/models.py
View File

@ -667,6 +667,16 @@ class BaseResource(models.Model):
resource_type=ContentType.objects.get_for_model(self), resource_pk=self.pk, apiuser__key=''
).exists()
def get_setting(self, name):
connectors_settings = settings.CONNECTORS_SETTINGS
if not isinstance(connectors_settings, dict):
return None
connector_identifier = f'{self.get_connector_slug()}/{self.slug}'
connector_settings = connectors_settings.get(connector_identifier)
if not isinstance(connector_settings, dict):
return None
return connector_settings.get(name)
class AccessRight(models.Model):
codename = models.CharField(max_length=100, verbose_name='codename')

301
passerelle/contrib/toulouse_maelis/models.py
View File

@ -496,6 +496,23 @@ class ToulouseMaelis(BaseResource, HTTPResource):
if value is None:
dico[key] = ''
def get_person_activity_list_raw(
self, family_id, person_id, nature_id=None, reference_year=None, start_date=None, end_date=None
):
params = {
'numDossier': family_id,
'numPerson': person_id,
'codeNatureActivity': nature_id,
'yearSchool': reference_year,
'dateStartActivity': start_date,
'dateEndActivity': end_date,
}
response = self.call(
'Activity', 'getPersonCatalogueActivity', getPersonCatalogueActivityRequestBean=params
)
data = serialize_object(response)
return data
@endpoint(
display_category='Famille',
description='Liste des catégories',
@ -2099,6 +2116,290 @@ class ToulouseMaelis(BaseResource, HTTPResource):
response = self.call('Family', 'presubscribeSchoolSibling', **post_data)
return {'data': serialize_object(response)}
@endpoint(
display_category='Inscriptions',
description="Catalogue des activités d'une personne",
name='get-person-activity-list',
perm='can_access',
parameters={
'NameID': {'description': 'Publik NameID'},
'family_id': {'description': 'Numéro de DUI'},
'person_id': {'description': "Numéro du responsale légal ou de l'enfant"},
'nature_id': {'description': "Numéro de la nature des activités"},
'start_date': {'description': 'Début de la période'},
'end_date': {'description': 'Fin de la période'},
'text_template': {
'description': "template utilisé pour la valeur text (URL encoding)",
'example_value': '{{ activity.libelle2 }}',
},
},
)
def get_person_activity_list(
self,
request,
person_id,
NameID=None,
family_id=None,
nature_id=None,
start_date=None,
end_date=None,
text_template=None,
):
family_id = family_id or self.get_link(NameID).family_id
reference_year = None
if start_date and end_date:
start_date, end_date, reference_year = self.get_start_and_end_dates(start_date, end_date)
if not text_template:
text_template = '{{ activity.libelle2|default:activity.libelle1 }}'
response = self.get_person_activity_list_raw(
family_id,
person_id,
nature_id=nature_id,
reference_year=reference_year,
start_date=start_date and start_date.strftime(utils.json_date_format),
end_date=start_date and end_date.strftime(utils.json_date_format),
)
for item in response['catalogueActivityList']:
item['id'] = item['activity']['idActivity']
item['text'] = render_to_string(text_template, item).strip()
return {'data': response['catalogueActivityList'], 'meta': {'person': response['person']}}
@endpoint(
display_category='Inscriptions',
description="Liste des unités d'une activité pour une personne",
name='get-person-unit-list',
perm='can_access',
parameters={
'NameID': {'description': 'Publik NameID'},
'family_id': {'description': 'Numéro de DUI'},
'person_id': {'description': "Numéro du responsale légal ou de l'enfant"},
'activity_id': {'description': "Numéro de l'activités"},
'start_date': {'description': 'Début de la période'},
'end_date': {'description': 'Fin de la période'},
'text_template': {
'description': 'template utilisée pour la valeur text (URL encoding)',
'example_value': '{{ libelle }}',
},
},
)
def get_person_unit_list(
self,
request,
person_id,
activity_id,
NameID=None,
family_id=None,
start_date=None,
end_date=None,
text_template=None,
):
family_id = family_id or self.get_link(NameID).family_id
reference_year = None
if start_date and end_date:
start_date, end_date, reference_year = self.get_start_and_end_dates(start_date, end_date)
if not text_template:
text_template = '{{ libelle }}'
response = self.get_person_activity_list_raw(
family_id,
person_id,
reference_year=reference_year,
start_date=start_date and start_date.strftime(utils.json_date_format),
end_date=start_date and end_date.strftime(utils.json_date_format),
)
for activity in response['catalogueActivityList']:
if activity['activity']['idActivity'] == activity_id:
break
else:
raise APIError('No activity %s for person' % activity_id, err_code='no-activity')
data = activity.pop('unitInfoList')
meta = {'person': response['person'], 'activity': activity}
for item in data:
item['id'] = item['idUnit']
context = dict(item)
context['meta'] = meta
item['text'] = render_to_string(text_template, context).strip()
return {'data': data, 'meta': meta}
@endpoint(
display_category='Inscriptions',
description="Liste des lieux d'une unité pour une personne",
name='get-person-place-list',
perm='can_access',
parameters={
'NameID': {'description': 'Publik NameID'},
'family_id': {'description': 'Numéro de DUI'},
'person_id': {'description': "Numéro du responsale légal ou de l'enfant"},
'activity_id': {'description': "Numéro de l'activités"},
'unit_id': {'description': "Numéro de l'unité"},
'start_date': {'description': 'Début de la période'},
'end_date': {'description': 'Fin de la période'},
'text_template': {
'description': 'template utilisée pour la valeur text (URL encoding)',
'example_value': '{{ libelle }}',
},
},
)
def get_person_place_list(
self,
request,
person_id,
activity_id,
unit_id,
NameID=None,
family_id=None,
start_date=None,
end_date=None,
text_template=None,
):
family_id = family_id or self.get_link(NameID).family_id
reference_year = None
if start_date and end_date:
start_date, end_date, reference_year = self.get_start_and_end_dates(start_date, end_date)
if not text_template:
text_template = '{{ place.lib2|default:place.lib1 }}'
response = self.get_person_activity_list_raw(
family_id,
person_id,
reference_year=reference_year,
start_date=start_date and start_date.strftime(utils.json_date_format),
end_date=start_date and end_date.strftime(utils.json_date_format),
)
for activity in response['catalogueActivityList']:
if activity['activity']['idActivity'] == activity_id:
break
else:
raise APIError('No activity %s for person' % activity_id, err_code='no-activity')
for unit in activity['unitInfoList']:
if unit['idUnit'] == unit_id:
break
else:
raise APIError('No unit %s for person' % unit_id, err_code='no-unit')
data = unit.pop('placeInfoList')
del activity['unitInfoList']
meta = {'person': response['person'], 'activity': activity, 'unit': unit}
for item in data:
item['id'] = item['place']['idPlace']
context = dict(item)
context['meta'] = meta
item['text'] = render_to_string(text_template, context).strip()
return {'data': data, 'meta': meta}
@endpoint(
display_category='Inscriptions',
description="Catalog geojson pour une personne",
name='get-person-catalog-geojson',
perm='can_access',
parameters={
'NameID': {'description': 'Publik NameID'},
'family_id': {'description': 'Numéro de DUI'},
'person_id': {'description': "Numéro du responsale légal ou de l'enfant"},
'nature_id': {'description': "Numéro de la nature des activités"},
'start_date': {'description': 'Début de la période'},
'end_date': {'description': 'Fin de la période'},
'activity_id': {'description': "Numéro de l'activités"},
'unit_id': {'description': "Numéro de l'unité"},
'place_id': {'description': "Numéro du lieu"},
},
)
def get_person_catalog_geojson(
self,
request,
person_id,
NameID=None,
family_id=None,
start_date=None,
end_date=None,
nature_id=None,
activity_id=None,
unit_id=None,
place_id=None,
):
family_id = family_id or self.get_link(NameID).family_id
reference_year = None
if start_date and end_date:
start_date, end_date, reference_year = self.get_start_and_end_dates(start_date, end_date)
response = self.get_person_activity_list_raw(
family_id,
person_id,
nature_id=nature_id,
reference_year=reference_year,
start_date=start_date and start_date.strftime(utils.json_date_format),
end_date=start_date and end_date.strftime(utils.json_date_format),
)
def places(properties, place_list, place_id=None):
for place in place_list:
properties['place'] = place
properties['place_id'] = place['place']['idPlace']
if not place['place']['longitude'] or not place['place']['latitude']:
continue
if place_id:
if properties['place_id'] == place_id:
yield properties
break
else:
yield properties
def units(properties, unit_list, unit_id=None, place_id=None):
for unit in unit_list:
place_list = unit.pop('placeInfoList')
properties['unit'] = unit
properties['unit_id'] = unit['idUnit']
if unit_id:
if properties['unit_id'] == unit_id:
yield from places(properties, place_list, place_id)
break
else:
yield from places(properties, place_list, place_id)
def activities(activity_id=None, unit_id=None, place_id=None):
for activity in response['catalogueActivityList']:
unit_list = activity.pop('unitInfoList')
properties = {
'person': response['person'],
'activity_id': activity['activity']['idActivity'],
'activity': activity,
}
if activity_id:
if properties['activity_id'] == activity_id:
yield from units(properties, unit_list, unit_id, place_id)
break
else:
yield from units(properties, unit_list, unit_id, place_id)
geojson = {
'type': 'FeatureCollection',
'features': [],
}
for item in activities(activity_id, unit_id, place_id):
geojson['features'].append(
{
'type': 'Feature',
'geometry': {
'coordinates': [
float(item['place']['place']['longitude']),
float(item['place']['place']['latitude']),
],
'type': 'Point',
},
'properties': {
'id': '%s:%s:%s' % (item['activity_id'], item['unit_id'], item['place_id']),
'text': '%s / %s / %s'
% (
item['activity']['activity']['libelle1'],
item['unit']['libelle'],
item['place']['place']['lib1'],
),
**item,
},
}
)
return geojson
class Link(models.Model):
resource = models.ForeignKey(ToulouseMaelis, on_delete=models.CASCADE)

6
passerelle/locale/fr/LC_MESSAGES/django.po
View File

@ -7,7 +7,7 @@ msgid ""
msgstr ""
"Project-Id-Version: Passerelle 0\n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2023-01-18 15:55+0100\n"
"POT-Creation-Date: 2023-01-27 14:15+0100\n"
"PO-Revision-Date: 2022-11-03 22:06+0100\n"
"Last-Translator: Frederic Peters <fpeters@entrouvert.com>\n"
"Language: fr\n"
@ -4162,6 +4162,10 @@ msgstr "Date de début de l'occupation"
msgid "Occupation end date"
msgstr "Date de fin de l'occupation"
#: apps/signal_arretes/models.py
msgid "Comment"
msgstr "Commentaire :"
#: apps/signal_arretes/models.py
msgid "Get cities available in Signal Arrêtés"
msgstr "Récupérer les communes disponibles dans Signal Arrêtés"

25
passerelle/settings.py
View File

@ -245,13 +245,36 @@ REQUESTS_TIMEOUT = 25
# }
REQUESTS_MAX_RETRIES = {}
# Connectors settings - extra settings for connectors
#
# CONNECTORS_SETTINGS = {
# "cmis/test": {
# "requests_substitutions": [
# {
# 'url': 'https://service.example.com/api/',
# 'search': 'http://service.example.internal/software/api/',
# 'replace': 'https://service.example.com/api/'
# }
# ]
# ]
# ]
#
# * requests_substitutions:
# Apply substitutions to HTTP responses obtained through self.requests
# search is a python regular expression for re.sub(), and replace the replacement string.
# The 'url' key is optional, if absent the replacement is done on all URLs.
CONNECTORS_SETTINGS = {}
# List of authorized content-types, as regular expressions, for substitutions
REQUESTS_SUBSTITUTIONS_CONTENT_TYPES = [r'text/.*', r'application/(.*\+)?json', r'application/(.*\+)?xml']
# Passerelle can receive big requests (for example base64 encoded files)
DATA_UPLOAD_MAX_MEMORY_SIZE = 100 * 1024 * 1024
SITE_BASE_URL = 'http://localhost'
# List of passerelle.utils.Request response Content-Type to log
LOGGED_CONTENT_TYPES_MESSAGES = (r'text/', r'application/(json|xml)')
LOGGED_CONTENT_TYPES_MESSAGES = [r'text/.*', r'application/(.*\+)?json', r'application/(.*\+)?xml']
# Max size of the response to log
LOGGED_RESPONSES_MAX_SIZE = 5000

78
passerelle/utils/__init__.py
View File

@ -176,7 +176,7 @@ def protected_api(perm):
return decorator
def content_type_match(ctype):
def should_content_type_body_be_logged(ctype):
content_types = settings.LOGGED_CONTENT_TYPES_MESSAGES
if not ctype:
return False
@ -219,7 +219,8 @@ def log_http_request(
if logger.level == 10: # DEBUG
extra['response_headers'] = make_headers_safe(response.headers)
# log body only if content type is allowed
if content_type_match(response.headers.get('Content-Type')):
content_type = response.headers.get('Content-Type', '').split(';')[0].strip().lower()
if should_content_type_body_be_logged(content_type):
max_size = settings.LOGGED_RESPONSES_MAX_SIZE
if hasattr(logger, 'connector'):
max_size = logger.connector.logging_parameters.responses_max_size or max_size
@ -280,6 +281,72 @@ class Request(RequestSession):
self.mount('http://', adapter)
self.timeout = timeout if timeout is not None else settings.REQUESTS_TIMEOUT
def _substitute(self, search, replace, value):
if isinstance(value, str):
value, nsub = re.subn(search, replace, value)
if nsub:
self.logger.debug('substitution: %d occurences', nsub)
elif isinstance(value, list):
value = [self._substitute(search, replace, v) for v in value]
elif isinstance(value, dict):
value = {
self._substitute(search, replace, k): self._substitute(search, replace, v)
for k, v in value.items()
}
return value
def apply_requests_substitution(self, response, substitution):
if not isinstance(substitution, dict):
self.logger.warning('substitution: invalid substitution, %r', substitution)
return
for key in ['search', 'replace']:
if key not in substitution:
self.logger.warning('substitution: missing field "%s": %s', key, substitution)
return
if not isinstance(substitution[key], str):
self.logger.warning(
'substitution: invalid type for field "%s", must be str: %s', key, substitution
)
return
search = substitution['search']
replace = substitution['replace']
# filter on url
if isinstance(substitution.get('url'), str):
url = urllib.parse.urlparse(substitution['url'])
request_url = urllib.parse.urlparse(response.request.url)
if url.scheme and url.scheme != request_url.scheme:
return
# substitution without a netloc are ignored
if not url.netloc:
return
if request_url.netloc != url.netloc:
return
if url.path and url.path != '/' and not request_url.path.startswith(url.path):
return
# filter on content-type
content_type = response.headers.get('Content-Type', '').split(';')[0].strip().lower()
for content_type_re in settings.REQUESTS_SUBSTITUTIONS_CONTENT_TYPES:
if re.match(content_type_re, content_type):
break
else:
self.logger.debug('substitution: content_type did not match %s', content_type)
return
self.logger.debug('substitution: try %s', substitution)
try:
if re.match(r'application/([^;]\+)?json', content_type):
import json
response._content = json.dumps(self._substitute(search, replace, response.json())).encode()
else:
response._content = self._substitute(search, replace, response.text).encode()
response.encoding = 'utf-8'
return True
except Exception:
self.logger.exception('substitution: "%s" failed', substitution)
def request(self, method, url, **kwargs):
cache_duration = kwargs.pop('cache_duration', None)
invalidate_cache = kwargs.pop('invalidate_cache', False)
@ -336,6 +403,13 @@ class Request(RequestSession):
warnings.simplefilter(action='ignore', category=InsecureRequestWarning)
response = super().request(method, url, **kwargs)
if self.resource:
requests_substitutions = self.resource.get_setting('requests_substitutions')
if isinstance(requests_substitutions, list):
for requests_substitution in requests_substitutions:
if not self.apply_requests_substitution(response, requests_substitution):
self.logger.debug('substitution: %s does not match', requests_substitution)
if method == 'GET' and cache_duration and (response.status_code // 100 == 2):
cache.set(
cache_key,

3
setup.py
View File

@ -144,6 +144,9 @@ setup(
'gadjo',
'phpserialize',
'suds',
# pyexcel-xlsx is unmaintained (no change since 2020) and is incompatible
# with a change introduced in openpyxl 3.1.0
'openpyxl<3.1',
'pyexcel-io',
'pyexcel-ods',
'pyexcel-xls',

1342
tests/data/toulouse_maelis/ActivityService.wsdl
View File

File diff suppressed because it is too large Load Diff

18
tests/data/toulouse_maelis/FamilyService.wsdl
View File

@ -800,6 +800,15 @@
<xs:enumeration value="CHOICE"/>
</xs:restriction>
</xs:simpleType>
<xs:simpleType name="subscribeAction">
<xs:restriction base="xs:string">
<xs:enumeration value="ADD_SUBSCRIBE"/>
<xs:enumeration value="DELETE_SUBSCRIBE"/>
<xs:enumeration value="UPDATE_SUBSCRIBE_DOSS"/>
<xs:enumeration value="ADD_DEROG"/>
<xs:enumeration value="COMPLETE_DEROG_DOSS"/>
</xs:restriction>
</xs:simpleType>
<xs:simpleType name="levelCode">
<xs:restriction base="xs:string">
<xs:enumeration value="SCHOOL"/>
@ -817,15 +826,6 @@
<xs:enumeration value="PERSON"/>
</xs:restriction>
</xs:simpleType>
<xs:simpleType name="subscribeAction">
<xs:restriction base="xs:string">
<xs:enumeration value="ADD_SUBSCRIBE"/>
<xs:enumeration value="DELETE_SUBSCRIBE"/>
<xs:enumeration value="UPDATE_SUBSCRIBE_DOSS"/>
<xs:enumeration value="ADD_DEROG"/>
<xs:enumeration value="COMPLETE_DEROG_DOSS"/>
</xs:restriction>
</xs:simpleType>
<xs:element name="MaelisFamilyException" type="tns:MaelisFamilyException"/>
<xs:complexType name="MaelisFamilyException">
<xs:sequence>

191
tests/data/toulouse_maelis/R_read_person_catalog_activity.xml Normal file
View File

@ -0,0 +1,191 @@
<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
<soap:Body>
<ns2:getPersonCatalogueActivityResponse xmlns:ns2="activity.ws.maelis.sigec.com" xmlns:ns3="bean.persistence.activity.ws.maelis.sigec.com">
<getPersonCatalogueActivityResultBean>
<person>
<dateBirth>2014-04-01T00:00:00+02:00</dateBirth>
<firstname>BART</firstname>
<lastname>SIMPSON</lastname>
<numPerson>246423</numPerson>
<sexe>M</sexe>
</person>
<catalogueActivityList>
<activity>
<idActivity>A10051141965</idActivity>
<libelle1>Vitrail Fusing 1/2 Je Adultes 2022/2023 - Mardi 14h-17h</libelle1>
<libelle2>Activité modèle</libelle2>
<activityType>
<code>LOI_ADU</code>
<libelle>Loisirs Adultes</libelle>
<natureSpec>
<code>P</code>
<libelle>Loisirs</libelle>
</natureSpec>
</activityType>
<typInsPortal>I</typInsPortal>
<paiementPortal>I</paiementPortal>
</activity>
<unitInfoList>
<idUnit>A10051141970</idUnit>
<libelle>Inscription 2ème semestre</libelle>
<dateStart>2023-02-01T00:00:00+01:00</dateStart>
<dateEnd>2023-06-30T00:00:00+02:00</dateEnd>
<placeInfoList>
<place>
<idPlace>A10053179226</idPlace>
<lib1>Centre Culturel ALBAN MINVILLE</lib1>
<ctrlPlaces>H</ctrlPlaces>
</place>
<capacityInfo>
<controlOK>true</controlOK>
</capacityInfo>
</placeInfoList>
</unitInfoList>
<unitInfoList>
<idUnit>A10051141990</idUnit>
<libelle>Inscription 1er semestre</libelle>
<dateStart>2022-09-01T00:00:00+02:00</dateStart>
<dateEnd>2023-01-31T00:00:00+01:00</dateEnd>
<placeInfoList>
<place>
<idPlace>A10053179226</idPlace>
<lib1>Centre Culturel ALBAN MINVILLE</lib1>
<ctrlPlaces>H</ctrlPlaces>
</place>
<capacityInfo>
<controlOK>true</controlOK>
</capacityInfo>
</placeInfoList>
</unitInfoList>
<unitInfoList>
<idUnit>A10051141968</idUnit>
<libelle>Inscription annuelle</libelle>
<dateStart>2022-09-01T00:00:00+02:00</dateStart>
<dateEnd>2023-06-30T00:00:00+02:00</dateEnd>
<placeInfoList>
<place>
<idPlace>A10053179226</idPlace>
<lib1>Centre Culturel ALBAN MINVILLE</lib1>
<ctrlPlaces>N</ctrlPlaces>
</place>
<capacityInfo>
<controlOK>true</controlOK>
</capacityInfo>
</placeInfoList>
</unitInfoList>
<incompleteFamilyFile>false</incompleteFamilyFile>
</catalogueActivityList>
<catalogueActivityList>
<activity>
<idActivity>A10053187087</idActivity>
<libelle1>Vacances Ete 2023</libelle1>
<activityType>
<code>LOI_VAC</code>
<libelle>Loisirs - Vacances</libelle>
<natureSpec>
<code>V</code>
<libelle>Vacances Enfants</libelle>
</natureSpec>
</activityType>
<typInsPortal>I</typInsPortal>
<paiementPortal>I</paiementPortal>
</activity>
<unitInfoList>
<idUnit>A10053187241</idUnit>
<libelle>Juillet</libelle>
<dateStart>2023-07-10T00:00:00+02:00</dateStart>
<dateEnd>2023-07-31T00:00:00+02:00</dateEnd>
<placeInfoList>
<place>
<idPlace>A10053179604</idPlace>
<lib1>ALEX JANY</lib1>
<ctrlPlaces>H</ctrlPlaces>
<longitude>2.0</longitude>
<latitude>1.0</latitude>
</place>
<capacityInfo>
<controlOK>true</controlOK>
</capacityInfo>
</placeInfoList>
</unitInfoList>
<unitInfoList>
<idUnit>A10053187242</idUnit>
<libelle>Aout</libelle>
<dateStart>2023-08-01T00:00:00+02:00</dateStart>
<dateEnd>2023-08-31T00:00:00+02:00</dateEnd>
<placeInfoList>
<place>
<idPlace>A10053179604</idPlace>
<lib1>ALEX JANY</lib1>
<ctrlPlaces>M</ctrlPlaces>
<longitude>2.0</longitude>
<latitude>1.0</latitude>
</place>
<capacityInfo>
<controlOK>true</controlOK>
</capacityInfo>
</placeInfoList>
</unitInfoList>
<incompleteFamilyFile>false</incompleteFamilyFile>
</catalogueActivityList>
<catalogueActivityList>
<activity>
<idActivity>A10053187065</idActivity>
<libelle1>Vacances Hivers 2023</libelle1>
<activityType>
<code>LOI_VAC</code>
<libelle>Loisirs - Vacances</libelle>
<natureSpec>
<code>V</code>
<libelle>Vacances Enfants</libelle>
</natureSpec>
</activityType>
<typInsPortal>I</typInsPortal>
<paiementPortal>I</paiementPortal>
</activity>
<unitInfoList>
<idUnit>A10053187086</idUnit>
<libelle>Semaine 2</libelle>
<dateStart>2023-02-27T00:00:00+01:00</dateStart>
<dateEnd>2023-03-03T00:00:00+01:00</dateEnd>
<placeInfoList>
<place>
<idPlace>A10053179604</idPlace>
<lib1>ALEX JANY</lib1>
<ageStart>2011-01-01T00:00:00+01:00</ageStart>
<ageEnd>2018-12-31T00:00:00+01:00</ageEnd>
<ctrlPlaces>H</ctrlPlaces>
<longitude>2.0</longitude>
<latitude>1.0</latitude>
</place>
<capacityInfo>
<controlOK>true</controlOK>
</capacityInfo>
</placeInfoList>
</unitInfoList>
<unitInfoList>
<idUnit>A10053187085</idUnit>
<libelle>Semaine 1</libelle>
<dateStart>2023-02-20T00:00:00+01:00</dateStart>
<dateEnd>2023-02-24T00:00:00+01:00</dateEnd>
<placeInfoList>
<place>
<idPlace>A10053179604</idPlace>
<lib1>ALEX JANY</lib1>
<ageStart>2011-01-01T00:00:00+01:00</ageStart>
<ageEnd>2020-12-31T00:00:00+01:00</ageEnd>
<ctrlPlaces>H</ctrlPlaces>
<longitude>2.0</longitude>
<latitude>1.0</latitude>
</place>
<capacityInfo>
<controlOK>true</controlOK>
</capacityInfo>
</placeInfoList>
</unitInfoList>
<incompleteFamilyFile>false</incompleteFamilyFile>
</catalogueActivityList>
</getPersonCatalogueActivityResultBean>
</ns2:getPersonCatalogueActivityResponse>
</soap:Body>
</soap:Envelope>

195
tests/test_cmis.py
View File

@ -1,14 +1,28 @@
# passerelle - uniform access to multiple data sources and services
# Copyright (C) 2023 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 base64
import os
import re
import xml.etree.ElementTree as ET
from unittest import mock
from unittest.mock import Mock, call
from urllib import error as urllib2
import httplib2
import py
import pytest
import responses
from cmislib import CmisClient
from cmislib.exceptions import (
CmisException,
@ -22,7 +36,7 @@ from django.urls import reverse
from django.utils.encoding import force_bytes, force_str
from passerelle.apps.cmis.models import CmisConnector
from passerelle.base.models import AccessRight, ApiUser, ResourceLog
from passerelle.base.models import AccessRight, ApiUser
from tests.test_manager import login
@ -392,9 +406,6 @@ def test_create_doc():
@pytest.mark.parametrize(
"cmis_exc,err_msg",
[
(httplib2.HttpLib2Error, "connection error"),
# FIXME used for cmslib 0.5 compat
(urllib2.URLError, "connection error"),
(PermissionDeniedException, "permission denied"),
(UpdateConflictException, "update conflict"),
(InvalidArgumentException, "invalid property"),
@ -509,8 +520,8 @@ def test_cmis_types_view(setup, app, admin_user, monkeypatch):
@pytest.mark.parametrize('debug', (False, True))
@mock.patch('httplib2.Http.request')
def test_raw_uploadfile(mocked_request, app, setup, debug, caplog):
@responses.activate
def test_raw_uploadfile(app, setup, debug, caplog):
""" Simulate the bellow bash query :
$ http https://passerelle.dev.publik.love/cmis/ged/uploadfile \
file:='{"filename": "test2", "content": "c2FsdXQK"}' path=/test-eo
@ -525,46 +536,21 @@ def test_raw_uploadfile(mocked_request, app, setup, debug, caplog):
if debug:
setup.set_log_level('DEBUG')
def cmis_mocked_request(uri, method="GET", body=None, **kwargs):
"""simulate the 3 (ordered) HTTP queries involved"""
response = {'status': '200'}
if method == 'GET' and uri == 'http://example.com/cmisatom':
with open('%s/tests/data/cmis/cmis1.out.xml' % os.getcwd(), 'rb') as fd:
content = fd.read()
elif method == 'GET' and uri == (
'http://example.com/cmisatom/test/path?path=/test-eo&filter=&includeAllowableActions=false&includeACL=false&'
'includePolicyIds=false&includeRelationships=&renditionFilter='
):
with open('%s/tests/data/cmis/cmis2.out.xml' % os.getcwd(), 'rb') as fd:
content = fd.read()
elif method == 'POST' and uri == 'http://example.com/cmisatom/test/children?id=L3Rlc3QtZW8%3D':
with open('%s/tests/data/cmis/cmis3.in.xml' % os.getcwd()) as fd:
expected_input = fd.read()
expected_input = expected_input.replace('\n', '')
expected_input = re.sub('> *<', '><', expected_input)
input1 = ET.tostring(ET.XML(expected_input))
with open('%s/tests/data/cmis/cmis1.out.xml' % os.getcwd(), 'rb') as fd:
cmis1_body = fd.read()
# reorder properties
input2 = ET.XML(body)
objects = input2.find('{http://docs.oasis-open.org/ns/cmis/restatom/200908/}object')
properties = objects.find('{http://docs.oasis-open.org/ns/cmis/core/200908/}properties')
data = []
for elem in properties:
key = elem.tag
data.append((key, elem))
data.sort()
properties[:] = [item[-1] for item in data]
input2 = ET.tostring(input2)
with open('%s/tests/data/cmis/cmis2.out.xml' % os.getcwd(), 'rb') as fd:
cmis2_body = fd.read()
if input1 != input2:
raise Exception('expect [[%s]] but get [[%s]]' % (body, expected_input))
with open('%s/tests/data/cmis/cmis3.out.xml' % os.getcwd(), 'rb') as fd:
content = fd.read()
else:
raise Exception('my fault error, url is not yet mocked: %s' % uri)
return (response, content)
with open('%s/tests/data/cmis/cmis3.out.xml' % os.getcwd(), 'rb') as fd:
cmis3_body = fd.read()
responses.add(responses.GET, 'http://example.com/cmisatom', body=cmis1_body, status=200)
responses.add(responses.GET, 'http://example.com/cmisatom/test/path', body=cmis2_body, status=200)
responses.add(responses.POST, 'http://example.com/cmisatom/test/children', body=cmis3_body, status=200)
mocked_request.side_effect = cmis_mocked_request
params = {
"path": path,
"file": {"filename": file_name, "content": b64encode(file_content), "content_type": "image/jpeg"},
@ -575,22 +561,6 @@ def test_raw_uploadfile(mocked_request, app, setup, debug, caplog):
assert json_result['data']['properties']['cmis:objectTypeId'] == "cmis:document"
assert json_result['data']['properties']['cmis:name'] == file_name
if not debug:
assert ResourceLog.objects.count() == 2
else:
assert ResourceLog.objects.count() == 11
logs = list(ResourceLog.objects.all())
assert logs[3].message == 'cmislib GET request to http://example.com/cmisatom'
assert logs[4].message == 'cmislib GET response (200)'
assert logs[4].extra['response'].startswith('<?xml')
assert (
logs[8].message
== 'cmislib POST request to http://example.com/cmisatom/test/children?id=L3Rlc3QtZW8%3D'
)
assert logs[8].extra['payload'].startswith('<?xml')
assert logs[9].message == 'cmislib POST response (200)'
assert logs[9].extra['response'].startswith('<?xml')
assert not any('cmislib' in record.name for record in caplog.records)
@ -606,73 +576,66 @@ def test_cmis_check_status(app, setup, monkeypatch):
setup.check_status()
@mock.patch('httplib2.Http.request')
def test_get_file(mocked_request, app, setup):
url = reverse('generic-endpoint', kwargs={'connector': 'cmis', 'endpoint': 'getfile', 'slug': setup.slug})
@responses.activate
def test_get_file(app, setup):
url = (
reverse('generic-endpoint', kwargs={'connector': 'cmis', 'endpoint': 'getfile', 'slug': setup.slug})
+ '?raise=1'
)
def cmis_mocked_request(uri, method="GET", body=None, **kwargs):
"""simulate the HTTP queries involved"""
response = {'status': '200'}
if method == 'GET' and uri == 'http://example.com/cmisatom':
with open('%s/tests/data/cmis/cmis1.out.xml' % os.getcwd(), 'rb') as fd:
content = fd.read()
elif method == 'GET' and (
uri.startswith('http://example.com/cmisatom/test/path?path=/test/file')
or uri.startswith(
'http://example.com/cmisatom/test/id?id=c4bc9d00-5bf0-404d-8f0a-a6260f6d21ae;1.0'
)
):
with open('%s/tests/data/cmis/cmis3.out.xml' % os.getcwd(), 'rb') as fd:
content = fd.read()
elif (
method == 'GET'
and uri == 'http://example.com/cmisatom/test/content/test2?id=L3Rlc3QtZW8vdGVzdDI%3D'
):
content = b'hello world'
else:
raise Exception('url is not yet mocked: %s' % uri)
return (response, content)
with open('%s/tests/data/cmis/cmis1.out.xml' % os.getcwd(), 'rb') as fd:
cmis1_body = fd.read()
mocked_request.side_effect = cmis_mocked_request
response = app.get(url, params={'object_id': 'c4bc9d00-5bf0-404d-8f0a-a6260f6d21ae;1.0'})
assert response.content_type == 'application/octet-stream'
assert response.content == b'hello world'
with open('%s/tests/data/cmis/cmis3.out.xml' % os.getcwd(), 'rb') as fd:
cmis3_body = fd.read()
responses.add(responses.GET, 'http://example.com/cmisatom', body=cmis1_body, status=200)
responses.add(responses.GET, 'http://example.com/cmisatom/test/path', body=cmis3_body, status=200)
responses.add(responses.GET, 'http://example.com/cmisatom/test/id', body=cmis3_body, status=200)
responses.add(
responses.GET,
'http://example.com/cmisatom/test/content/test2?id=L3Rlc3QtZW8vdGVzdDI%3D',
body=b'hello world',
status=200,
)
response = app.get(url, params={'object_id': '/test/file'})
assert response.content_type == 'application/octet-stream'
assert response.content == b'hello world'
response = app.get(url, params={'object_id': 'c4bc9d00-5bf0-404d-8f0a-a6260f6d21ae;1.0'})
assert response.content_type == 'application/octet-stream'
assert response.content == b'hello world'
@mock.patch('httplib2.Http.request')
def test_get_metadata(mocked_request, app, setup):
@responses.activate
def test_get_metadata(app, setup):
url = reverse(
'generic-endpoint', kwargs={'connector': 'cmis', 'endpoint': 'getmetadata', 'slug': setup.slug}
)
def cmis_mocked_request(uri, method="GET", body=None, **kwargs):
"""simulate the HTTP queries involved"""
response = {'status': '200'}
if method == 'GET' and uri == 'http://example.com/cmisatom':
with open('%s/tests/data/cmis/cmis1.out.xml' % os.getcwd(), 'rb') as fd:
content = fd.read()
elif method == 'GET' and (
uri.startswith('http://example.com/cmisatom/test/path?path=/test/file')
or uri.startswith(
'http://example.com/cmisatom/test/id?id=c4bc9d00-5bf0-404d-8f0a-a6260f6d21ae;1.0'
)
):
with open('%s/tests/data/cmis/cmis3.out.xml' % os.getcwd(), 'rb') as fd:
content = fd.read()
elif (
method == 'GET'
and uri == 'http://example.com/cmisatom/test/content/test2?id=L3Rlc3QtZW8vdGVzdDI%3D'
):
content = b'hello world'
else:
raise Exception('url is not yet mocked: %s' % uri)
return (response, content)
with open('%s/tests/data/cmis/cmis1.out.xml' % os.getcwd(), 'rb') as fd:
cmis1_body = fd.read()
with open('%s/tests/data/cmis/cmis3.out.xml' % os.getcwd(), 'rb') as fd:
cmis3_body = fd.read()
responses.add(responses.GET, 'http://example.com/cmisatom', body=cmis1_body, status=200)
responses.add(responses.GET, 'http://example.com/cmisatom/test/path', body=cmis3_body, status=200)
responses.add(responses.GET, 'http://example.com/cmisatom/test/id', body=cmis3_body, status=200)
responses.add(
responses.GET,
'http://example.com/cmisatom/test/content/test2?id=L3Rlc3QtZW8vdGVzdDI%3D',
body=b'hello world',
status=200,
)
mocked_request.side_effect = cmis_mocked_request
response = app.get(url, params={'object_id': 'c4bc9d00-5bf0-404d-8f0a-a6260f6d21ae;1.0'})
assert response.json['data']['cmis']['contentStreamFileName'] == 'test2'
assert response.json['data']['rsj']['idInsertis'] == '21N284563'

8
tests/test_franceconnect_data.py
View File

@ -59,7 +59,7 @@ DGFIP_MOCKED_RESPONSES = USER_INFO_MOCKED_RESPONSES + [
DGFIP_MOCKED_RESPONSES += [
[
'/impotparticulier/1.0/situations/ir/assiettes/annrev/%s' % year,
{'rfr': 0, 'revenuBrutGlobal': 0},
{'rfr': year * 2, 'revenuBrutGlobal': year * 10},
]
for year in range(CURRENT_YEAR - 3, CURRENT_YEAR)
]
@ -165,3 +165,9 @@ def test_dgfip_mode(app, fc):
assert data['dgfip_ir']
for year in range(CURRENT_YEAR - 3, CURRENT_YEAR):
assert data['dgfip_ir'][str(year)]
assert 'last_known' in data['dgfip_ir']
last_known_ir = data['dgfip_ir']['last_known']
past_year = CURRENT_YEAR - 1
assert last_known_ir['year'] == str(past_year)
assert last_known_ir['revenuBrutGlobal'] == past_year * 10
assert last_known_ir['rfr'] == past_year * 2

133
tests/test_requests.py
View File

@ -33,6 +33,10 @@ class MockResource:
verify_cert = True
http_proxy = ''
@classmethod
def get_setting(cls, name):
return None
@pytest.fixture(params=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'])
def log_level(request):
@ -531,3 +535,132 @@ def test_requests_to_legacy_urls(log_level):
resp = requests.get('https://old.org/foobar')
assert resp.json() == {"foo": "bar"}
assert resp.request.url == 'https://new.org/foobar'
@responses.activate
def test_requests_substitution(settings):
from passerelle.base.models import BaseResource
resource = mock.Mock()
resource.requests_max_retries = {}
resource.slug = 'test'
resource.get_connector_slug.return_value = 'cmis'
resource.get_setting = lambda name: BaseResource.get_setting(resource, name)
requests = Request(logger=logging.getLogger(), resource=resource)
settings.CONNECTORS_SETTINGS = {
"cmis/test": {
'requests_substitutions': [
{
'url': 'https://example.com/',
'search': 'http://example.internal',
'replace': 'https://example.com',
}
]
}
}
responses.add(
responses.GET,
"https://example.com/html",
content_type='text/html',
body=b'<html>\n<a href="http://example.internal/path/">\n<a/></html>',
status=200,
)
assert (
requests.get('https://example.com/html?bar=foo', params={'foo': 'bar'}).text
== '<html>\n<a href="https://example.com/path/">\n<a/></html>'
)
responses.add(
responses.GET,
"https://example.com/xml",
content_type='application/xml',
body=b'<a href="http://example.internal/path/"><a/>',
status=200,
)
assert requests.get('https://example.com/xml').text == '<a href="https://example.com/path/"><a/>'
# check substitution is applied inside JSON, even if some characters are escaped
responses.add(
responses.GET,
"https://example.com/json",
content_type='application/json',
body=b'{"url": "http:\\/\\/example.internal/path/"}',
status=200,
)
assert requests.get('https://example.com/json').json() == {'url': 'https://example.com/path/'}
responses.add(
responses.GET,
"https://example.com/binary",
content_type='application/octet-stream',
body=b'\00<a href="http://example.internal/path/"><a/>',
status=200,
)
assert (
requests.get('https://example.com/binary').content
== b'\00<a href="http://example.internal/path/"><a/>'
)
responses.add(
responses.GET,
"https://example.com/binary2",
content_type='',
body=b'\00<a href="http://example.internal/path/"><a/>',
status=200,
)
assert (
requests.get('https://example.com/binary2').content
== b'\00<a href="http://example.internal/path/"><a/>'
)
responses.add(
responses.GET,
"https://example2.com/html",
content_type='text/html',
body=b'<html>\n<a href="http://example.internal/path/">\n<a/></html>',
status=200,
)
# wrong hostname
assert (
requests.get('https://example2.com/html?query=1').text
== '<html>\n<a href="http://example.internal/path/">\n<a/></html>'
)
# check that url field is optional
settings.CONNECTORS_SETTINGS = {
"cmis/test": {
'requests_substitutions': [
{
'search': 'http://example.internal',
'replace': 'https://example.com',
}
]
}
}
responses.add(
responses.GET,
"https://whatever.com/html",
content_type='text/html',
body=b'<html>\n<a href="http://example.internal/path/">\n<a/></html>',
status=200,
)
assert (
requests.get('https://whatever.com/html?bar=foo', params={'foo': 'bar'}).text
== '<html>\n<a href="https://example.com/path/">\n<a/></html>'
)
# check setting is applied per connector slug
resource.get_connector_slug.return_value = 'pas-cmis'
requests = Request(logger=logging.getLogger(), resource=resource)
responses.add(
responses.GET,
"https://example.com/html",
content_type='text/html',
body=b'<html>\n<a href="http://example.internal/path/">\n<a/></html>',
status=200,
)
assert (
requests.get('https://example.com/html?bar=foo', params={'foo': 'bar'}).text
== '<html>\n<a href="http://example.internal/path/">\n<a/></html>'
)

3
tests/test_signal_arretes.py
View File

@ -68,6 +68,7 @@ def mock_creation_dodp(url, request):
assert 'SIRET' not in data or data['SIRET'] == '00000000000000'
assert 'numeroDossier' not in data or data['numeroDossier'] == 'reference_dossier'
assert 'commentaire' not in data or data['commentaire'] == 'Wubba Lubba Dub Dub'
assert 'adresseLigne1' not in contact or contact['adresseLigne1'] == '6 Sesame street'
assert 'CP' not in contact or contact['CP'] == '42 42420'
assert 'ville' not in contact or contact['ville'] == 'Melun'
@ -211,6 +212,7 @@ REQUIRED_PARAMETERS = {
'occupation_type': 'Base de vie',
'occupation_start_date': '02/06/2022',
'occupation_end_date': '03/06/2022',
'comment': '',
}
@ -229,6 +231,7 @@ def test_create_request(app, connector, mock_signal_arretes):
'declarant_city': 'Melun',
'declarant_phone': '0636656565',
'occupation_lane': 'Sesame Street',
'comment': 'Wubba Lubba Dub Dub',
}
)

594
tests/test_toulouse_maelis.py
View File

@ -4648,3 +4648,597 @@ def test_create_child_school_pre_registration_with_sibling(family_service, con,
assert resp.json['data']['codeWait'] == 'MO_FRATERIE'
assert resp.json['data']['derogReason'] == '01PRIO-5'
assert resp.json['data']['derogComment'] == 'SERGHEI3 LISA'
def test_get_person_activity_list(activity_service, con, app):
def request_check(request):
assert request.yearSchool == 2022
activity_service.add_soap_response(
'getPersonCatalogueActivity',
get_xml_file('R_read_person_catalog_activity.xml'),
request_check=request_check,
)
url = get_endpoint('get-person-activity-list')
params = {
'NameID': '',
'family_id': '311323',
'person_id': '246423',
'nature_id': '',
'start_date': '2022-09-01',
'end_date': '2023-08-31',
'text_template': '',
}
resp = app.get(url, params=params)
assert resp.json['err'] == 0
Link.objects.create(resource=con, family_id='311323', name_id='local')
params['NameID'] = 'local'
params['family_id'] = ''
resp = app.get(url, params=params)
assert resp.json['err'] == 0
assert [(x['id'], x['text']) for x in resp.json['data']] == [
('A10051141965', 'Activité modèle'),
('A10053187087', 'Vacances Ete 2023'),
('A10053187065', 'Vacances Hivers 2023'),
]
data = resp.json['data'][1]
del data['unitInfoList'][1]
assert data == {
'activity': {
'activityType': {
'code': 'LOI_VAC',
'libelle': 'Loisirs - Vacances',
'natureSpec': {'code': 'V', 'libelle': 'Vacances Enfants'},
},
'idActivity': 'A10053187087',
'libelle1': 'Vacances Ete 2023',
'libelle2': None,
'paiementPortal': 'I',
'typInsPortal': 'I',
},
'id': 'A10053187087',
'incompleteFamilyFile': False,
'indicatorBlockSubscribeList': [],
'text': 'Vacances Ete 2023',
'unitInfoList': [
{
'dateEnd': '2023-07-31T00:00:00+02:00',
'dateStart': '2023-07-10T00:00:00+02:00',
'idIns': None,
'idUnit': 'A10053187241',
'libelle': 'Juillet',
'placeInfoList': [
{
'capacityInfo': {'controlOK': True, 'message': None},
'idIns': None,
'place': {
'ageEnd': None,
'ageStart': None,
'ctrlPlaces': 'H',
'etatIns': None,
'idIns': None,
'idPlace': 'A10053179604',
'latitude': 1,
'lib1': 'ALEX JANY',
'lib2': None,
'listBlocNoteBean': [],
'longitude': 2,
},
}
],
},
],
}
assert resp.json['meta']['person']['numPerson'] == 246423
params['text_template'] = '{{ unitInfoList.0.libelle }}'
resp = app.get(url, params=params)
assert resp.json['err'] == 0
assert [(x['id'], x['text']) for x in resp.json['data']] == [
('A10051141965', 'Inscription 2ème semestre'),
('A10053187087', 'Juillet'),
('A10053187065', 'Semaine 2'),
]
def test_get_person_activity_list_not_linked_error(con, app):
url = get_endpoint('get-person-activity-list')
params = {
'NameID': 'local',
'family_id': '',
'person_id': '246423',
'nature_id': '',
'start_date': '2022-09-01',
'end_date': '2023-08-31',
'text_template': '',
}
resp = app.get(url, params=params)
assert resp.json['err'] == 'not-linked'
assert resp.json['err_desc'] == 'User not linked to family'
def test_get_person_activity_list_date_error(con, app):
url = get_endpoint('get-person-activity-list')
params = {
'NameID': '',
'family_id': '311323',
'person_id': '246423',
'nature_id': '',
'start_date': 'bad',
'end_date': '2023-08-31',
'text_template': '',
}
resp = app.get(url, params=params, status=400)
assert resp.json['err'] == 'bad-request'
assert resp.json['err_desc'] == 'bad date format, should be YYYY-MM-DD'
params['start_date'] = '2022-09-01'
params['end_date'] = 'bad'
resp = app.get(url, params=params, status=400)
assert resp.json['err'] == 'bad-request'
assert resp.json['err_desc'] == 'bad date format, should be YYYY-MM-DD'
params['start_date'] = '2023-09-01'
params['end_date'] = '2023-08-31'
resp = app.get(url, params=params, status=400)
assert resp.json['err'] == 'bad-request'
assert resp.json['err_desc'] == 'start_date should be before end_date'
params['start_date'] = '2022-09-01'
params['end_date'] = '2024-08-31'
resp = app.get(url, params=params, status=400)
assert resp.json['err'] == 'bad-request'
assert resp.json['err_desc'] == 'start_date and end_date are in different reference year (2022 != 2023)'
def test_get_person_unit_list(activity_service, con, app):
def request_check(request):
assert request.yearSchool == 2022
activity_service.add_soap_response(
'getPersonCatalogueActivity',
get_xml_file('R_read_person_catalog_activity.xml'),
request_check=request_check,
)
url = get_endpoint('get-person-unit-list')
params = {
'NameID': '',
'family_id': '311323',
'person_id': '246423',
'activity_id': 'A10053187087',
'start_date': '2022-09-01',
'end_date': '2023-08-31',
'text_template': '',
}
resp = app.get(url, params=params)
assert resp.json['err'] == 0
Link.objects.create(resource=con, family_id='311323', name_id='local')
params['NameID'] = 'local'
params['family_id'] = ''
resp = app.get(url, params=params)
assert resp.json['err'] == 0
assert [(x['id'], x['text']) for x in resp.json['data']] == [
('A10053187241', 'Juillet'),
('A10053187242', 'Aout'),
]
assert resp.json['data'][0] == {
'dateEnd': '2023-07-31T00:00:00+02:00',
'dateStart': '2023-07-10T00:00:00+02:00',
'id': 'A10053187241',
'idIns': None,
'idUnit': 'A10053187241',
'libelle': 'Juillet',
'placeInfoList': [
{
'capacityInfo': {'controlOK': True, 'message': None},
'idIns': None,
'place': {
'ageEnd': None,
'ageStart': None,
'ctrlPlaces': 'H',
'etatIns': None,
'idIns': None,
'idPlace': 'A10053179604',
'latitude': 1,
'lib1': 'ALEX JANY',
'lib2': None,
'listBlocNoteBean': [],
'longitude': 2,
},
}
],
'text': 'Juillet',
}
assert resp.json['meta']['person']['numPerson'] == 246423
assert resp.json['meta']['activity']['activity']['idActivity'] == 'A10053187087'
params['text_template'] = '{{ meta.activity.activity.activityType.natureSpec.libelle }} - {{ libelle }}'
resp = app.get(url, params=params)
assert resp.json['err'] == 0
assert [(x['id'], x['text']) for x in resp.json['data']] == [
('A10053187241', 'Vacances Enfants - Juillet'),
('A10053187242', 'Vacances Enfants - Aout'),
]
def test_get_person_unit_list_not_linked_error(con, app):
url = get_endpoint('get-person-unit-list')
params = {
'NameID': 'local',
'family_id': '',
'person_id': '246423',
'activity_id': 'A10053187087',
'start_date': '2022-09-01',
'end_date': '2023-08-31',
'text_template': '',
}
resp = app.get(url, params=params)
assert resp.json['err'] == 'not-linked'
assert resp.json['err_desc'] == 'User not linked to family'
def test_get_person_unit_list_date_error(con, app):
url = get_endpoint('get-person-unit-list')
params = {
'NameID': '',
'family_id': '311323',
'person_id': '246423',
'activity_id': 'A10053187087',
'start_date': 'bad',
'end_date': '2023-08-31',
'text_template': '',
}
resp = app.get(url, params=params, status=400)
assert resp.json['err'] == 'bad-request'
assert resp.json['err_desc'] == 'bad date format, should be YYYY-MM-DD'
params['start_date'] = '2022-09-01'
params['end_date'] = 'bad'
resp = app.get(url, params=params, status=400)
assert resp.json['err'] == 'bad-request'
assert resp.json['err_desc'] == 'bad date format, should be YYYY-MM-DD'
params['start_date'] = '2023-09-01'
params['end_date'] = '2023-08-31'
resp = app.get(url, params=params, status=400)
assert resp.json['err'] == 'bad-request'
assert resp.json['err_desc'] == 'start_date should be before end_date'
params['start_date'] = '2022-09-01'
params['end_date'] = '2024-08-31'
resp = app.get(url, params=params, status=400)
assert resp.json['err'] == 'bad-request'
assert resp.json['err_desc'] == 'start_date and end_date are in different reference year (2022 != 2023)'
def test_get_person_unit_list_no_activity_error(activity_service, con, app):
activity_service.add_soap_response(
'getPersonCatalogueActivity',
get_xml_file('R_read_person_catalog_activity.xml'),
)
url = get_endpoint('get-person-unit-list')
params = {
'NameID': '',
'family_id': '311323',
'person_id': '246423',
'activity_id': 'plop',
'start_date': '2022-09-01',
'end_date': '2023-08-31',
'text_template': '',
}
resp = app.get(url, params=params)
assert resp.json['err'] == 'no-activity'
assert resp.json['err_desc'] == 'No activity plop for person'
def test_get_person_place_list(activity_service, con, app):
def request_check(request):
assert request.yearSchool == 2022
activity_service.add_soap_response(
'getPersonCatalogueActivity',
get_xml_file('R_read_person_catalog_activity.xml'),
request_check=request_check,
)
url = get_endpoint('get-person-place-list')
params = {
'NameID': '',
'family_id': '311323',
'person_id': '246423',
'activity_id': 'A10053187087',
'unit_id': 'A10053187241',
'start_date': '2022-09-01',
'end_date': '2023-08-31',
'text_template': '',
}
resp = app.get(url, params=params)
assert resp.json['err'] == 0
Link.objects.create(resource=con, family_id='311323', name_id='local')
params['NameID'] = 'local'
params['family_id'] = ''
resp = app.get(url, params=params)
assert resp.json['err'] == 0
assert [(x['id'], x['text']) for x in resp.json['data']] == [('A10053179604', 'ALEX JANY')]
assert resp.json['data'] == [
{
'capacityInfo': {'controlOK': True, 'message': None},
'id': 'A10053179604',
'idIns': None,
'place': {
'ageEnd': None,
'ageStart': None,
'ctrlPlaces': 'H',
'etatIns': None,
'idIns': None,
'idPlace': 'A10053179604',
'latitude': 1.0,
'lib1': 'ALEX JANY',
'lib2': None,
'listBlocNoteBean': [],
'longitude': 2.0,
},
'text': 'ALEX JANY',
}
]
assert resp.json['meta']['person']['numPerson'] == 246423
assert resp.json['meta']['activity']['activity']['idActivity'] == 'A10053187087'
assert resp.json['meta']['unit']['idUnit'] == 'A10053187241'
params['text_template'] = '{{ meta.unit.libelle }} - {{ place.lib1 }}'
resp = app.get(url, params=params)
assert resp.json['err'] == 0
assert [(x['id'], x['text']) for x in resp.json['data']] == [('A10053179604', 'Juillet - ALEX JANY')]
def test_get_person_place_list_not_linked_error(con, app):
url = get_endpoint('get-person-place-list')
params = {
'NameID': 'local',
'family_id': '',
'person_id': '246423',
'activity_id': 'A10053187087',
'unit_id': 'A10053187241',
'start_date': '2022-09-01',
'end_date': '2023-08-31',
'text_template': '',
}
resp = app.get(url, params=params)
assert resp.json['err'] == 'not-linked'
assert resp.json['err_desc'] == 'User not linked to family'
def test_get_person_place_list_date_error(con, app):
url = get_endpoint('get-person-place-list')
params = {
'NameID': '',
'family_id': '311323',
'person_id': '246423',
'activity_id': 'A10053187087',
'unit_id': 'A10053187241',
'start_date': 'bad',
'end_date': '2023-08-31',
'text_template': '',
}
resp = app.get(url, params=params, status=400)
assert resp.json['err'] == 'bad-request'
assert resp.json['err_desc'] == 'bad date format, should be YYYY-MM-DD'
params['start_date'] = '2022-09-01'
params['end_date'] = 'bad'
resp = app.get(url, params=params, status=400)
assert resp.json['err'] == 'bad-request'
assert resp.json['err_desc'] == 'bad date format, should be YYYY-MM-DD'
params['start_date'] = '2023-09-01'
params['end_date'] = '2023-08-31'
resp = app.get(url, params=params, status=400)
assert resp.json['err'] == 'bad-request'
assert resp.json['err_desc'] == 'start_date should be before end_date'
params['start_date'] = '2022-09-01'
params['end_date'] = '2024-08-31'
resp = app.get(url, params=params, status=400)
assert resp.json['err'] == 'bad-request'
assert resp.json['err_desc'] == 'start_date and end_date are in different reference year (2022 != 2023)'
def test_get_person_place_list_no_unit_error(activity_service, con, app):
activity_service.add_soap_response(
'getPersonCatalogueActivity',
get_xml_file('R_read_person_catalog_activity.xml'),
)
url = get_endpoint('get-person-place-list')
params = {
'NameID': '',
'family_id': '311323',
'person_id': '246423',
'activity_id': 'plop',
'unit_id': 'plop',
'start_date': '2022-09-01',
'end_date': '2023-08-31',
'text_template': '',
}
resp = app.get(url, params=params)
assert resp.json['err'] == 'no-activity'
assert resp.json['err_desc'] == 'No activity plop for person'
params['activity_id'] = 'A10053187087'
resp = app.get(url, params=params)
assert resp.json['err'] == 'no-unit'
assert resp.json['err_desc'] == 'No unit plop for person'
def test_get_person_catalog_geojson(activity_service, con, app):
def request_check(request):
assert request.yearSchool == 2022
activity_service.add_soap_response(
'getPersonCatalogueActivity',
get_xml_file('R_read_person_catalog_activity.xml'),
request_check=request_check,
)
url = get_endpoint('get-person-catalog-geojson')
params = {
'NameID': '',
'family_id': '311323',
'person_id': '246423',
'nature_id': '',
'start_date': '2022-09-01',
'end_date': '2023-08-31',
'activity_id': 'A10053187087',
'unit_id': 'A10053187241',
'place_id': 'A10053179604',
}
resp = app.get(url, params=params)
assert resp.json['err'] == 0
Link.objects.create(resource=con, family_id='311323', name_id='local')
params['NameID'] = 'local'
params['family_id'] = ''
resp = app.get(url, params=params)
assert len(resp.json['features']) == 1
assert resp.json == {
'err': 0,
'features': [
{
'geometry': {'coordinates': [2.0, 1.0], 'type': 'Point'},
'properties': {
'id': 'A10053187087:A10053187241:A10053179604',
'text': 'Vacances Ete 2023 / Juillet / ALEX JANY',
'activity_id': 'A10053187087',
'unit_id': 'A10053187241',
'place_id': 'A10053179604',
'person': {
'dateBirth': '2014-04-01T00:00:00+02:00',
'firstname': 'BART',
'lastname': 'SIMPSON',
'numPerson': 246423,
'sexe': 'M',
},
'activity': {
'activity': {
'activityType': {
'code': 'LOI_VAC',
'libelle': 'Loisirs - Vacances',
'natureSpec': {'code': 'V', 'libelle': 'Vacances ' 'Enfants'},
},
'idActivity': 'A10053187087',
'libelle1': 'Vacances ' 'Ete 2023',
'libelle2': None,
'paiementPortal': 'I',
'typInsPortal': 'I',
},
'incompleteFamilyFile': False,
'indicatorBlockSubscribeList': [],
},
'unit': {
'dateEnd': '2023-07-31T00:00:00+02:00',
'dateStart': '2023-07-10T00:00:00+02:00',
'idIns': None,
'idUnit': 'A10053187241',
'libelle': 'Juillet',
},
'place': {
'capacityInfo': {'controlOK': True, 'message': None},
'idIns': None,
'place': {
'ageEnd': None,
'ageStart': None,
'ctrlPlaces': 'H',
'etatIns': None,
'idIns': None,
'idPlace': 'A10053179604',
'latitude': 1.0,
'lib1': 'ALEX JANY',
'lib2': None,
'listBlocNoteBean': [],
'longitude': 2.0,
},
},
},
'type': 'Feature',
},
],
'type': 'FeatureCollection',
}
params['place_id'] = 'plop'
resp = app.get(url, params=params)
assert len(resp.json['features']) == 0
del params['place_id']
resp = app.get(url, params=params)
assert len(resp.json['features']) == 1
params['unit_id'] = 'plop'
resp = app.get(url, params=params)
assert len(resp.json['features']) == 0
del params['unit_id']
resp = app.get(url, params=params)
assert len(resp.json['features']) == 2
params['activity_id'] = 'plop'
resp = app.get(url, params=params)
assert len(resp.json['features']) == 0
del params['activity_id']
resp = app.get(url, params=params)
assert len(resp.json['features']) == 4
def test_get_person_catalog_geojson_not_linked_error(con, app):
url = get_endpoint('get-person-catalog-geojson')
params = {
'NameID': '',
'family_id': '',
'person_id': '246423',
}
resp = app.get(url, params=params)
assert resp.json['err'] == 'not-linked'
assert resp.json['err_desc'] == 'User not linked to family'
def test_get_person_catalog_geojson_date_error(con, app):
url = get_endpoint('get-person-catalog-geojson')
params = {
'NameID': '',
'family_id': '311323',
'person_id': '246423',
'start_date': 'bad',
'end_date': '2023-08-31',
}
resp = app.get(url, params=params, status=400)
assert resp.json['err'] == 'bad-request'
assert resp.json['err_desc'] == 'bad date format, should be YYYY-MM-DD'
params['start_date'] = '2022-09-01'
params['end_date'] = 'bad'
resp = app.get(url, params=params, status=400)
assert resp.json['err'] == 'bad-request'
assert resp.json['err_desc'] == 'bad date format, should be YYYY-MM-DD'
params['start_date'] = '2023-09-01'
params['end_date'] = '2023-08-31'
resp = app.get(url, params=params, status=400)
assert resp.json['err'] == 'bad-request'
assert resp.json['err_desc'] == 'start_date should be before end_date'
params['start_date'] = '2022-09-01'
params['end_date'] = '2024-08-31'
resp = app.get(url, params=params, status=400)
assert resp.json['err'] == 'bad-request'
assert resp.json['err_desc'] == 'start_date and end_date are in different reference year (2022 != 2023)'