0001-caluire-axel-code-factorization-53704.patch
passerelle/contrib/toulouse_axel/models.py | ||
---|---|---|
32 | 32 |
from passerelle.utils.jsonresponse import APIError |
33 | 33 |
from . import schemas |
34 | 34 |
from . import utils |
35 |
from .exceptions import AxelError
|
|
35 |
from passerelle.contrib.utils import axel
|
|
36 | 36 | |
37 | 37 |
logger = logging.getLogger('passerelle.contrib.toulouse_axel') |
38 | 38 | |
... | ... | |
120 | 120 | |
121 | 121 |
try: |
122 | 122 |
result = schemas.ref_date_gestion_dui(self) |
123 |
except AxelError as e: |
|
123 |
except axel.AxelError as e:
|
|
124 | 124 |
raise APIError( |
125 | 125 |
'Axel error: %s' % e, |
126 | 126 |
err_code='error', |
... | ... | |
141 | 141 |
def check_dui(self, post_data): |
142 | 142 |
try: |
143 | 143 |
result = schemas.ref_verif_dui(self, {'PORTAIL': {'DUI': post_data}}) |
144 |
except AxelError as e: |
|
144 |
except axel.AxelError as e:
|
|
145 | 145 |
raise APIError( |
146 | 146 |
'Axel error: %s' % e, |
147 | 147 |
err_code='error', |
... | ... | |
292 | 292 |
def get_family_data(self, dui, check_registrations=False, with_management_dates=False): |
293 | 293 |
try: |
294 | 294 |
result = schemas.ref_famille_dui(self, {'PORTAIL': {'DUI': {'IDDUI': dui}}}) |
295 |
except AxelError as e: |
|
295 |
except axel.AxelError as e:
|
|
296 | 296 |
raise APIError( |
297 | 297 |
'Axel error: %s' % e, |
298 | 298 |
err_code='error', |
... | ... | |
452 | 452 |
flags = sorted(schemas.UPDATE_FAMILY_FLAGS.keys()) |
453 | 453 |
for flag in flags: |
454 | 454 |
flag_value = post_data.get(flag) |
455 |
flag_value = utils.encode_bool(flag_value)
|
|
455 |
flag_value = axel.encode_bool(flag_value)
|
|
456 | 456 | |
457 | 457 |
# no update for the related block |
458 | 458 |
if flag_value == 'OUI': |
... | ... | |
579 | 579 |
# transform ALLERGIE block |
580 | 580 |
new_allergie = [] |
581 | 581 |
for key in ['ASTHME', 'MEDICAMENTEUSES', 'ALIMENTAIRES']: |
582 |
if utils.encode_bool(child['SANITAIRE']['ALLERGIE'][key]) == 'OUI':
|
|
582 |
if axel.encode_bool(child['SANITAIRE']['ALLERGIE'][key]) == 'OUI':
|
|
583 | 583 |
new_allergie.append( |
584 | 584 |
{ |
585 | 585 |
'TYPE': key, |
... | ... | |
642 | 642 | |
643 | 643 |
# prepare data |
644 | 644 |
post_data['IDDUI'] = link.dui |
645 |
post_data['DATEDEMANDE'] = datetime.date.today().strftime(utils.json_date_format)
|
|
645 |
post_data['DATEDEMANDE'] = datetime.date.today().strftime(axel.json_date_format)
|
|
646 | 646 | |
647 | 647 |
self.sanitize_update_family_data(dui=link.dui, post_data=post_data) |
648 | 648 | |
... | ... | |
653 | 653 | |
654 | 654 |
try: |
655 | 655 |
result = schemas.form_maj_famille_dui(self, {'PORTAIL': {'DUI': post_data}}) |
656 |
except AxelError as e: |
|
656 |
except axel.AxelError as e:
|
|
657 | 657 |
raise APIError( |
658 | 658 |
'Axel error: %s' % e, |
659 | 659 |
err_code='error', |
... | ... | |
680 | 680 | |
681 | 681 |
try: |
682 | 682 |
result = schemas.ref_facture_a_payer(self, {'PORTAIL': {'DUI': {'IDDUI': dui}}}) |
683 |
except AxelError as e: |
|
683 |
except axel.AxelError as e:
|
|
684 | 684 |
raise APIError( |
685 | 685 |
'Axel error: %s' % e, |
686 | 686 |
err_code='error', |
... | ... | |
701 | 701 |
result = schemas.list_dui_factures( |
702 | 702 |
self, {'LISTFACTURE': {'NUMDUI': link.dui, 'DEBUT': '1970-01-01'}} |
703 | 703 |
) |
704 |
except AxelError as e: |
|
704 |
except axel.AxelError as e:
|
|
705 | 705 |
raise APIError( |
706 | 706 |
'Axel error: %s' % e, |
707 | 707 |
err_code='error', |
... | ... | |
829 | 829 |
result = schemas.ref_facture_pdf( |
830 | 830 |
self, {'PORTAIL': {'FACTUREPDF': {'IDFACTURE': int(invoice['display_id'])}}} |
831 | 831 |
) |
832 |
except AxelError as e: |
|
832 |
except axel.AxelError as e:
|
|
833 | 833 |
raise APIError( |
834 | 834 |
'Axel error: %s' % e, |
835 | 835 |
err_code='error', |
... | ... | |
876 | 876 | |
877 | 877 |
transaction_amount = invoice['amount'] |
878 | 878 |
transaction_id = data['transaction_id'] |
879 |
transaction_date = utils.parse_datetime(data['transaction_date'])
|
|
879 |
transaction_date = axel.parse_datetime(data['transaction_date'])
|
|
880 | 880 |
if transaction_date is None: |
881 | 881 |
raise APIError('invalid transaction_date') |
882 |
transaction_date = utils.encode_datetime(transaction_date)
|
|
882 |
transaction_date = axel.encode_datetime(transaction_date)
|
|
883 | 883 |
post_data = { |
884 | 884 |
'IDFACTURE': int(invoice_id), |
885 | 885 |
'IDREGIEENCAISSEMENT': '', |
... | ... | |
889 | 889 |
} |
890 | 890 |
try: |
891 | 891 |
schemas.form_paiement_dui(self, {'PORTAIL': {'DUI': post_data}}) |
892 |
except AxelError as e: |
|
892 |
except axel.AxelError as e:
|
|
893 | 893 |
raise APIError( |
894 | 894 |
'Axel error: %s' % e, |
895 | 895 |
err_code='error', |
... | ... | |
913 | 913 |
} |
914 | 914 |
}, |
915 | 915 |
) |
916 |
except AxelError as e: |
|
916 |
except axel.AxelError as e:
|
|
917 | 917 |
raise APIError( |
918 | 918 |
'Axel error: %s' % e, |
919 | 919 |
err_code='error', |
... | ... | |
993 | 993 |
{ |
994 | 994 |
'IDACTIVITE': activity_id, |
995 | 995 |
'ANNEEREFERENCE': str(reference_year), |
996 |
'DATEDEBUT': start_date.strftime(utils.json_date_format),
|
|
997 |
'DATEDFIN': end_date.strftime(utils.json_date_format),
|
|
996 |
'DATEDEBUT': start_date.strftime(axel.json_date_format),
|
|
997 |
'DATEDFIN': end_date.strftime(axel.json_date_format),
|
|
998 | 998 |
} |
999 | 999 |
) |
1000 | 1000 |
try: |
... | ... | |
1012 | 1012 |
} |
1013 | 1013 |
}, |
1014 | 1014 |
) |
1015 |
except AxelError as e: |
|
1015 |
except axel.AxelError as e:
|
|
1016 | 1016 |
raise APIError( |
1017 | 1017 |
'Axel error: %s' % e, |
1018 | 1018 |
err_code='error', |
... | ... | |
1044 | 1044 | |
1045 | 1045 |
for activity in child_activities.get('ACTIVITE', []): |
1046 | 1046 |
activity['id'] = activity['IDACTIVITE'] |
1047 |
start_date = datetime.datetime.strptime(activity['DATEENTREE'], utils.json_date_format)
|
|
1048 |
end_date = datetime.datetime.strptime(activity['DATESORTIE'], utils.json_date_format)
|
|
1047 |
start_date = datetime.datetime.strptime(activity['DATEENTREE'], axel.json_date_format)
|
|
1048 |
end_date = datetime.datetime.strptime(activity['DATESORTIE'], axel.json_date_format)
|
|
1049 | 1049 |
activity['text'] = u'{} (inscription du {} au {})'.format( |
1050 | 1050 |
activity['LIBELLEACTIVITE'], |
1051 |
start_date.strftime(utils.xml_date_format),
|
|
1052 |
end_date.strftime(utils.xml_date_format),
|
|
1051 |
start_date.strftime(axel.xml_date_format),
|
|
1052 |
end_date.strftime(axel.xml_date_format),
|
|
1053 | 1053 |
) |
1054 | 1054 |
activity['annee_reference'] = reference_year |
1055 | 1055 |
activity['annee_reference_short'] = str(reference_year)[2:] |
... | ... | |
1080 | 1080 |
# get pivot date |
1081 | 1081 |
try: |
1082 | 1082 |
pivot_date = datetime.datetime.strptime( |
1083 |
'%s-%s' % (reference_year, pivot_date), utils.json_date_format
|
|
1083 |
'%s-%s' % (reference_year, pivot_date), axel.json_date_format
|
|
1084 | 1084 |
).date() |
1085 | 1085 |
except ValueError: |
1086 | 1086 |
raise APIError('bad date format, should be MM-DD', err_code='bad-request', http_status=400) |
... | ... | |
1094 | 1094 |
'id': str(reference_year), |
1095 | 1095 |
'text': '%s/%s' % (reference_year, reference_year + 1), |
1096 | 1096 |
'type': 'encours', |
1097 |
'refdate': today.strftime(utils.json_date_format),
|
|
1097 |
'refdate': today.strftime(axel.json_date_format),
|
|
1098 | 1098 |
} |
1099 | 1099 |
] |
1100 | 1100 |
if today < pivot_date: |
... | ... | |
1115 | 1115 |
'id': str(reference_year + 1), |
1116 | 1116 |
'text': '%s/%s' % (reference_year + 1, reference_year + 2), |
1117 | 1117 |
'type': 'suivante', |
1118 |
'refdate': next_ref_date.strftime(utils.json_date_format),
|
|
1118 |
'refdate': next_ref_date.strftime(axel.json_date_format),
|
|
1119 | 1119 |
} |
1120 | 1120 |
) |
1121 | 1121 |
return {'data': data} |
... | ... | |
1133 | 1133 |
def clae_children_activities_info(self, request, NameID, booking_date): |
1134 | 1134 |
link = self.get_link(NameID) |
1135 | 1135 |
try: |
1136 |
booking_date = datetime.datetime.strptime(booking_date, utils.json_date_format)
|
|
1136 |
booking_date = datetime.datetime.strptime(booking_date, axel.json_date_format)
|
|
1137 | 1137 |
except ValueError: |
1138 | 1138 |
raise APIError('bad date format, should be YYYY-MM-DD', err_code='bad-request', http_status=400) |
1139 | 1139 | |
... | ... | |
1159 | 1159 |
def clae_booking_activities_info(self, request, NameID, idpersonne, start_date, end_date): |
1160 | 1160 |
link = self.get_link(NameID) |
1161 | 1161 |
try: |
1162 |
start_date = datetime.datetime.strptime(start_date, utils.json_date_format).date()
|
|
1163 |
end_date = datetime.datetime.strptime(end_date, utils.json_date_format).date()
|
|
1162 |
start_date = datetime.datetime.strptime(start_date, axel.json_date_format).date()
|
|
1163 |
end_date = datetime.datetime.strptime(end_date, axel.json_date_format).date()
|
|
1164 | 1164 |
except ValueError: |
1165 | 1165 |
raise APIError('bad date format, should be YYYY-MM-DD', err_code='bad-request', http_status=400) |
1166 | 1166 | |
... | ... | |
1174 | 1174 |
booking_data = {d['TYPEACTIVITE']: d for d in booking_data} |
1175 | 1175 |
start_date, end_date = utils.get_week_dates_from_date(week_start_date) |
1176 | 1176 |
week = 'week:%s:%s' % ( |
1177 |
start_date.strftime(utils.json_date_format),
|
|
1178 |
end_date.strftime(utils.json_date_format),
|
|
1177 |
start_date.strftime(axel.json_date_format),
|
|
1178 |
end_date.strftime(axel.json_date_format),
|
|
1179 | 1179 |
) |
1180 | 1180 |
day_date = week_start_date |
1181 | 1181 |
while day_date <= week_end_date: |
... | ... | |
1187 | 1187 |
booked = activity['booking']['days'][day] |
1188 | 1188 |
if booked is not None: |
1189 | 1189 |
yield { |
1190 |
'day': day_date.strftime(utils.json_date_format),
|
|
1190 |
'day': day_date.strftime(axel.json_date_format),
|
|
1191 | 1191 |
'activity_id': activity['id'], |
1192 | 1192 |
'activity_type': activity_type, |
1193 | 1193 |
'activity_label': activity['LIBELLEACTIVITE'], |
... | ... | |
1218 | 1218 |
entree_dates = [act['DATEENTREE'] for act in child_activities.get('ACTIVITE', [])] |
1219 | 1219 |
sortie_dates = [act['DATESORTIE'] for act in child_activities.get('ACTIVITE', [])] |
1220 | 1220 |
return ( |
1221 |
datetime.datetime.strptime(max(entree_dates), utils.json_date_format).date(),
|
|
1222 |
datetime.datetime.strptime(min(sortie_dates), utils.json_date_format).date(),
|
|
1221 |
datetime.datetime.strptime(max(entree_dates), axel.json_date_format).date(),
|
|
1222 |
datetime.datetime.strptime(min(sortie_dates), axel.json_date_format).date(),
|
|
1223 | 1223 |
) |
1224 | 1224 | |
1225 | 1225 |
@endpoint( |
... | ... | |
1240 | 1240 |
): |
1241 | 1241 |
link = self.get_link(NameID) |
1242 | 1242 |
try: |
1243 |
start_date = datetime.datetime.strptime(start_date, utils.json_date_format).date()
|
|
1244 |
end_date = datetime.datetime.strptime(end_date, utils.json_date_format).date()
|
|
1243 |
start_date = datetime.datetime.strptime(start_date, axel.json_date_format).date()
|
|
1244 |
end_date = datetime.datetime.strptime(end_date, axel.json_date_format).date()
|
|
1245 | 1245 |
except ValueError: |
1246 | 1246 |
raise APIError('bad date format, should be YYYY-MM-DD', err_code='bad-request', http_status=400) |
1247 | 1247 |
if activity_type not in ['MAT', 'MIDI', 'SOIR', 'GARD']: |
... | ... | |
1286 | 1286 |
day = WEEKDAYS[day_date.weekday()] |
1287 | 1287 |
activity_day = { |
1288 | 1288 |
'id': '{}:{}:{}:{}'.format( |
1289 |
idpersonne, activity_type, activity['id'], day_date.strftime(utils.json_date_format)
|
|
1289 |
idpersonne, activity_type, activity['id'], day_date.strftime(axel.json_date_format)
|
|
1290 | 1290 |
), |
1291 | 1291 |
'text': dateformat.format(day_date, 'l j F Y'), |
1292 | 1292 |
'disabled': activity['booking']['days'][day] is None, |
... | ... | |
1328 | 1328 |
): |
1329 | 1329 |
link = self.get_link(NameID) |
1330 | 1330 |
try: |
1331 |
booking_date = datetime.datetime.strptime(booking_date, utils.json_date_format).date()
|
|
1331 |
booking_date = datetime.datetime.strptime(booking_date, axel.json_date_format).date()
|
|
1332 | 1332 |
except ValueError: |
1333 | 1333 |
raise APIError('bad date format, should be YYYY-MM-DD', err_code='bad-request', http_status=400) |
1334 | 1334 |
if activity_type not in ['MAT', 'MIDI', 'SOIR', 'GARD']: |
... | ... | |
1408 | 1408 |
# check dates |
1409 | 1409 |
today = datetime.date.today() |
1410 | 1410 |
start_date_min = today + datetime.timedelta(days=8) |
1411 |
start_date = datetime.datetime.strptime( |
|
1412 |
post_data['booking_start_date'], utils.json_date_format |
|
1413 |
).date() |
|
1411 |
start_date = datetime.datetime.strptime(post_data['booking_start_date'], axel.json_date_format).date() |
|
1414 | 1412 |
reference_year = utils.get_reference_year_from_date(start_date) |
1415 | 1413 |
end_date_max = datetime.date(reference_year + 1, 7, 31) |
1416 |
end_date = datetime.datetime.strptime(post_data['booking_end_date'], utils.json_date_format).date()
|
|
1414 |
end_date = datetime.datetime.strptime(post_data['booking_end_date'], axel.json_date_format).date()
|
|
1417 | 1415 |
if start_date > end_date: |
1418 | 1416 |
raise APIError( |
1419 | 1417 |
'booking_start_date should be before booking_end_date', |
... | ... | |
1461 | 1459 |
post_data['child_id'], |
1462 | 1460 |
activity_type, |
1463 | 1461 |
activity_id, |
1464 |
day_date.strftime(utils.json_date_format),
|
|
1462 |
day_date.strftime(axel.json_date_format),
|
|
1465 | 1463 |
) |
1466 | 1464 |
if key in post_data['booking_list_%s' % activity_type]: |
1467 | 1465 |
week_pattern = ( |
... | ... | |
1483 | 1481 |
week_pattern = get_week_pattern(real_start_date, real_end_date, activity_type, activity_id) |
1484 | 1482 |
activity['PERIODE'].append( |
1485 | 1483 |
{ |
1486 |
'DATEDEBUT': real_start_date.strftime(utils.json_date_format),
|
|
1487 |
'DATEDFIN': real_end_date.strftime(utils.json_date_format),
|
|
1484 |
'DATEDEBUT': real_start_date.strftime(axel.json_date_format),
|
|
1485 |
'DATEDFIN': real_end_date.strftime(axel.json_date_format),
|
|
1488 | 1486 |
'SEMAINETYPE': week_pattern, |
1489 | 1487 |
} |
1490 | 1488 |
) |
... | ... | |
1494 | 1492 |
# build data |
1495 | 1493 |
data = { |
1496 | 1494 |
'IDDUI': link.dui, |
1497 |
'DATEDEMANDE': today.strftime(utils.json_date_format),
|
|
1495 |
'DATEDEMANDE': today.strftime(axel.json_date_format),
|
|
1498 | 1496 |
'ENFANT': [ |
1499 | 1497 |
{ |
1500 | 1498 |
'IDPERSONNE': post_data['child_id'], |
... | ... | |
1510 | 1508 | |
1511 | 1509 |
try: |
1512 | 1510 |
result = schemas.reservation_annuelle(self, {'PORTAIL': {'DUI': data}}) |
1513 |
except AxelError as e: |
|
1511 |
except axel.AxelError as e:
|
|
1514 | 1512 |
raise APIError( |
1515 | 1513 |
'Axel error: %s' % e, |
1516 | 1514 |
err_code='error', |
... | ... | |
1562 | 1560 |
# build dates of the period |
1563 | 1561 |
today = datetime.date.today() |
1564 | 1562 |
start_date_min = today + datetime.timedelta(days=8) |
1565 |
start_date = datetime.datetime.strptime(post_data['booking_date'], utils.json_date_format).date()
|
|
1563 |
start_date = datetime.datetime.strptime(post_data['booking_date'], axel.json_date_format).date()
|
|
1566 | 1564 |
start_date = max(start_date, start_date_min) |
1567 | 1565 |
reference_year = utils.get_reference_year_from_date(start_date) |
1568 | 1566 |
end_date = datetime.date(reference_year + 1, 7, 31) |
... | ... | |
1593 | 1591 |
'ANNEEREFERENCE': str(reference_year), |
1594 | 1592 |
'PERIODE': [ |
1595 | 1593 |
{ |
1596 |
'DATEDEBUT': start_date.strftime(utils.json_date_format),
|
|
1597 |
'DATEDFIN': end_date.strftime(utils.json_date_format),
|
|
1594 |
'DATEDEBUT': start_date.strftime(axel.json_date_format),
|
|
1595 |
'DATEDFIN': end_date.strftime(axel.json_date_format),
|
|
1598 | 1596 |
'SEMAINETYPE': week_pattern, |
1599 | 1597 |
} |
1600 | 1598 |
], |
... | ... | |
1603 | 1601 |
# build data |
1604 | 1602 |
data = { |
1605 | 1603 |
'IDDUI': link.dui, |
1606 |
'DATEDEMANDE': today.strftime(utils.json_date_format),
|
|
1604 |
'DATEDEMANDE': today.strftime(axel.json_date_format),
|
|
1607 | 1605 |
'ENFANT': [ |
1608 | 1606 |
{ |
1609 | 1607 |
'IDPERSONNE': post_data['child_id'], |
... | ... | |
1619 | 1617 | |
1620 | 1618 |
try: |
1621 | 1619 |
result = schemas.reservation_annuelle(self, {'PORTAIL': {'DUI': data}}) |
1622 |
except AxelError as e: |
|
1620 |
except axel.AxelError as e:
|
|
1623 | 1621 |
raise APIError( |
1624 | 1622 |
'Axel error: %s' % e, |
1625 | 1623 |
err_code='error', |
passerelle/contrib/toulouse_axel/schemas.py | ||
---|---|---|
15 | 15 |
# along with this program. If not, see <http://www.gnu.org/licenses/>. |
16 | 16 | |
17 | 17 |
import copy |
18 |
import datetime |
|
19 | 18 |
import os |
20 |
import re |
|
21 |
import xml.etree.ElementTree as ET |
|
22 |
from collections import namedtuple |
|
23 | 19 | |
24 |
from django.utils.encoding import force_text |
|
25 | ||
26 |
import xmlschema |
|
27 | ||
28 |
from passerelle.utils.xml import JSONSchemaFromXMLSchema |
|
29 |
from . import utils |
|
30 |
from .exceptions import AxelError |
|
20 |
from passerelle.contrib.utils import axel |
|
31 | 21 | |
32 | 22 |
BASE_XSD_PATH = os.path.join(os.path.dirname(__file__), 'xsd') |
33 | 23 | |
34 | 24 | |
35 |
class AxelSchema(JSONSchemaFromXMLSchema): |
|
36 |
type_map = { |
|
37 |
'{urn:AllAxelTypes}DATEREQUIREDType': 'date', |
|
38 |
'{urn:AllAxelTypes}DATEType': 'date_optional', |
|
39 |
'{urn:AllAxelTypes}OUINONREQUIREDType': 'bool', |
|
40 |
'{urn:AllAxelTypes}OUINONType': 'bool_optional', |
|
41 |
} |
|
42 | ||
43 |
@classmethod |
|
44 |
def schema_date(cls): |
|
45 |
return { |
|
46 |
'type': 'string', |
|
47 |
'pattern': '[0-9]{4}-[0-9]{2}-[0-9]{2}', |
|
48 |
} |
|
49 | ||
50 |
def encode_date(self, obj): |
|
51 |
try: |
|
52 |
return datetime.datetime.strptime(obj, utils.json_date_format).strftime(utils.xml_date_format) |
|
53 |
except ValueError: |
|
54 |
return obj |
|
55 | ||
56 |
def encode_date_optional(self, obj): |
|
57 |
if not obj: |
|
58 |
return obj |
|
59 |
return self.encode_date(obj) |
|
60 | ||
61 |
def decode_date(self, data): |
|
62 |
value = datetime.datetime.strptime(data.text, utils.xml_date_format).strftime(utils.json_date_format) |
|
63 |
return xmlschema.ElementData( |
|
64 |
tag=data.tag, text=value, content=data.content, attributes=data.attributes |
|
65 |
) |
|
66 | ||
67 |
def decode_date_optional(self, data): |
|
68 |
if not data.text: |
|
69 |
return data |
|
70 |
return self.decode_date(data) |
|
71 | ||
72 |
@classmethod |
|
73 |
def schema_bool(cls): |
|
74 |
return copy.deepcopy(utils.boolean_type) |
|
75 | ||
76 |
def encode_bool(self, obj): |
|
77 |
return utils.encode_bool(obj) |
|
78 | ||
79 |
def decode_bool(self, data): |
|
80 |
value = False |
|
81 |
if data.text.lower() == 'oui': |
|
82 |
value = True |
|
83 |
return xmlschema.ElementData( |
|
84 |
tag=data.tag, text=value, content=data.content, attributes=data.attributes |
|
85 |
) |
|
86 | ||
87 |
@classmethod |
|
88 |
def schema_bool_optional(cls): |
|
89 |
schema_bool_optional = cls.schema_bool() |
|
90 |
schema_bool_optional['oneOf'].append({'type': 'string', 'enum': ['']}) |
|
91 |
return schema_bool_optional |
|
92 | ||
93 |
def encode_bool_optional(self, obj): |
|
94 |
return self.encode_bool(obj) |
|
95 | ||
96 |
def decode_bool_optional(self, data): |
|
97 |
if not data.text: |
|
98 |
return data |
|
99 |
return self.decode_bool(data) |
|
100 | ||
101 | ||
102 |
def xml_schema_converter(name, root_element): |
|
103 |
xsd_path = os.path.join(BASE_XSD_PATH, name) |
|
104 |
if not os.path.exists(xsd_path): |
|
105 |
return None |
|
106 |
return AxelSchema(xsd_path, root_element) |
|
107 | ||
108 | ||
109 |
OperationResult = namedtuple('OperationResult', ['json_response', 'xml_request', 'xml_response']) |
|
110 | ||
111 | ||
112 |
class Operation(object): |
|
113 |
def __init__(self, operation, prefix='Dui/', request_root_element='PORTAIL'): |
|
114 |
self.operation = operation |
|
115 |
self.request_converter = xml_schema_converter( |
|
116 |
'%sQ_%s.xsd' % (prefix, operation), request_root_element |
|
117 |
) |
|
118 |
self.response_converter = xml_schema_converter('%sR_%s.xsd' % (prefix, operation), 'PORTAILSERVICE') |
|
119 |
self.name = re.sub( |
|
120 |
'(.?)([A-Z])', lambda s: s.group(1) + ('-' if s.group(1) else '') + s.group(2).lower(), operation |
|
121 |
) |
|
122 |
self.snake_name = self.name.replace('-', '_') |
|
123 | ||
124 |
@property |
|
125 |
def request_schema(self): |
|
126 |
schema = self.request_converter.json_schema |
|
127 |
schema['flatten'] = True |
|
128 |
schema['merge_extra'] = True |
|
129 |
return schema |
|
130 | ||
131 |
def __call__(self, resource, request_data=None): |
|
132 |
client = resource.soap_client() |
|
133 | ||
134 |
serialized_request = '' |
|
135 |
if self.request_converter: |
|
136 |
try: |
|
137 |
serialized_request = self.request_converter.encode(request_data) |
|
138 |
except xmlschema.XMLSchemaValidationError as e: |
|
139 |
raise AxelError('invalid request %s' % str(e)) |
|
140 |
utils.indent(serialized_request) |
|
141 |
serialized_request = force_text(ET.tostring(serialized_request)) |
|
142 |
try: |
|
143 |
self.request_converter.xml_schema.validate(serialized_request) |
|
144 |
except xmlschema.XMLSchemaValidationError as e: |
|
145 |
raise AxelError('invalid request %s' % str(e), xml_request=serialized_request) |
|
146 | ||
147 |
result = client.service.getData( |
|
148 |
self.operation, serialized_request, '' |
|
149 |
) # FIXME: What is the user parameter for ? |
|
150 | ||
151 |
xml_result = ET.fromstring(result.encode('utf-8')) |
|
152 |
utils.indent(xml_result) |
|
153 |
pretty_result = force_text(ET.tostring(xml_result)) |
|
154 |
if xml_result.find('RESULTAT/STATUS').text != 'OK': |
|
155 |
msg = xml_result.find('RESULTAT/COMMENTAIRES').text |
|
156 |
raise AxelError(msg, xml_request=serialized_request, xml_response=pretty_result) |
|
157 | ||
158 |
try: |
|
159 |
return OperationResult( |
|
160 |
json_response=self.response_converter.decode(xml_result), |
|
161 |
xml_request=serialized_request, |
|
162 |
xml_response=pretty_result, |
|
163 |
) |
|
164 |
except xmlschema.XMLSchemaValidationError as e: |
|
165 |
raise AxelError( |
|
166 |
'invalid response %s' % str(e), xml_request=serialized_request, xml_response=pretty_result |
|
167 |
) |
|
25 |
class Operation(axel.Operation): |
|
26 |
base_xsd_path = BASE_XSD_PATH |
|
27 |
default_prefix = 'Dui/' |
|
168 | 28 | |
169 | 29 | |
170 | 30 |
ref_date_gestion_dui = Operation('RefDateGestionDui') |
... | ... | |
183 | 43 |
PAYMENT_SCHEMA = { |
184 | 44 |
'type': 'object', |
185 | 45 |
'properties': { |
186 |
'transaction_date': copy.deepcopy(utils.datetime_type),
|
|
46 |
'transaction_date': copy.deepcopy(axel.datetime_type),
|
|
187 | 47 |
'transaction_id': { |
188 | 48 |
'type': 'string', |
189 | 49 |
}, |
... | ... | |
229 | 89 |
) |
230 | 90 | |
231 | 91 |
for flag in sorted(UPDATE_FAMILY_FLAGS.keys()): |
232 |
flag_type = copy.deepcopy(utils.boolean_type)
|
|
92 |
flag_type = copy.deepcopy(axel.boolean_type)
|
|
233 | 93 |
if flag not in UPDATE_FAMILY_REQUIRED_FLAGS: |
234 | 94 |
flag_type['oneOf'].append({'type': 'null'}) |
235 | 95 |
flag_type['oneOf'].append({'type': 'string', 'enum': ['']}) |
... | ... | |
283 | 143 |
} |
284 | 144 |
sanitaire_required.append('ALLERGIE') |
285 | 145 |
for key in ['ASTHME', 'MEDICAMENTEUSES', 'ALIMENTAIRES']: |
286 |
flag_type = copy.deepcopy(utils.boolean_type)
|
|
146 |
flag_type = copy.deepcopy(axel.boolean_type)
|
|
287 | 147 |
flag_type['oneOf'].append({'type': 'null'}) |
288 | 148 |
flag_type['oneOf'].append({'type': 'string', 'enum': ['']}) |
289 | 149 |
sanitaire_properties['ALLERGIE']['properties'][key] = flag_type |
... | ... | |
303 | 163 |
BOOKING_SCHEMA = { |
304 | 164 |
'type': 'object', |
305 | 165 |
'properties': { |
306 |
'booking_start_date': copy.deepcopy(utils.date_type),
|
|
307 |
'booking_end_date': copy.deepcopy(utils.date_type),
|
|
166 |
'booking_start_date': copy.deepcopy(axel.date_type),
|
|
167 |
'booking_end_date': copy.deepcopy(axel.date_type),
|
|
308 | 168 |
'booking_list_MAT': { |
309 | 169 |
'oneOf': [ |
310 | 170 |
{'type': 'null'}, |
... | ... | |
429 | 289 |
'maxLength': 8, |
430 | 290 |
}, |
431 | 291 |
'regime': {'oneOf': [{'type': 'null'}, {'type': 'string', 'enum': ['', 'SV', 'AV']}]}, |
432 |
'booking_date': copy.deepcopy(utils.date_type),
|
|
292 |
'booking_date': copy.deepcopy(axel.date_type),
|
|
433 | 293 |
}, |
434 | 294 |
'required': [ |
435 | 295 |
'booking_list_MAT', |
passerelle/contrib/toulouse_axel/utils.py | ||
---|---|---|
19 | 19 | |
20 | 20 |
from collections import OrderedDict |
21 | 21 |
import datetime |
22 |
import unicodedata |
|
23 |
import xml.etree.ElementTree as ET |
|
24 | ||
25 |
import pytz |
|
26 | 22 | |
27 | 23 |
from django.utils.six import string_types |
28 | 24 | |
29 | 25 |
from passerelle.utils.conversion import normalize |
30 | 26 | |
31 | 27 | |
32 |
boolean_type = { |
|
33 |
'oneOf': [ |
|
34 |
{'type': 'boolean'}, |
|
35 |
{ |
|
36 |
'type': 'string', |
|
37 |
'pattern': '[Oo][Uu][Ii]|[Nn][Oo][Nn]|[Tt][Rr][Uu][Ee]|[Ff][Aa][Ll][Ss][Ee]|1|0', |
|
38 |
}, |
|
39 |
] |
|
40 |
} |
|
41 |
date_type = { |
|
42 |
'type': 'string', |
|
43 |
'pattern': '[0-9]{4}-[0-9]{2}-[0-9]{2}', |
|
44 |
} |
|
45 |
datetime_type = { |
|
46 |
'type': 'string', |
|
47 |
'pattern': '[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}', |
|
48 |
} |
|
49 |
json_date_format = '%Y-%m-%d' |
|
50 |
json_datetime_format = '%Y-%m-%dT%H:%M:%S' |
|
51 |
xml_date_format = '%d/%m/%Y' |
|
52 |
xml_datetime_format = '%d/%m/%Y %H:%M:%S' |
|
53 | ||
54 | ||
55 | 28 |
situation_familiale_mapping = OrderedDict( |
56 | 29 |
[ |
57 | 30 |
('C', 'Célibataire'), |
... | ... | |
113 | 86 |
return mapping.get(code, code) |
114 | 87 | |
115 | 88 | |
116 |
def indent(tree, space=" ", level=0): |
|
117 |
# backport from Lib/xml/etree/ElementTree.py python 3.9 |
|
118 |
if isinstance(tree, ET.ElementTree): |
|
119 |
tree = tree.getroot() |
|
120 |
if level < 0: |
|
121 |
raise ValueError("Initial indentation level must be >= 0, got {level}".format(level)) |
|
122 |
if not len(tree): |
|
123 |
return |
|
124 | ||
125 |
# Reduce the memory consumption by reusing indentation strings. |
|
126 |
indentations = ["\n" + level * space] |
|
127 | ||
128 |
def _indent_children(elem, level): |
|
129 |
# Start a new indentation level for the first child. |
|
130 |
child_level = level + 1 |
|
131 |
try: |
|
132 |
child_indentation = indentations[child_level] |
|
133 |
except IndexError: |
|
134 |
child_indentation = indentations[level] + space |
|
135 |
indentations.append(child_indentation) |
|
136 | ||
137 |
if not elem.text or not elem.text.strip(): |
|
138 |
elem.text = child_indentation |
|
139 | ||
140 |
for child in elem: |
|
141 |
if len(child): |
|
142 |
_indent_children(child, child_level) |
|
143 |
if not child.tail or not child.tail.strip(): |
|
144 |
child.tail = child_indentation |
|
145 | ||
146 |
# Dedent after the last child by overwriting the previous indentation. |
|
147 |
if not child.tail.strip(): |
|
148 |
child.tail = indentations[level] |
|
149 | ||
150 |
_indent_children(tree, 0) |
|
151 | ||
152 | ||
153 |
def encode_bool(obj): |
|
154 |
if obj is True or str(obj).lower() in ['true', 'oui', '1']: |
|
155 |
return 'OUI' |
|
156 |
if obj is False or str(obj).lower() in ['false', 'non', '0']: |
|
157 |
return 'NON' |
|
158 |
return obj |
|
159 | ||
160 | ||
161 |
def parse_datetime(value): |
|
162 |
try: |
|
163 |
dt = datetime.datetime.strptime(value, json_datetime_format) |
|
164 |
except ValueError: |
|
165 |
return None |
|
166 |
return pytz.utc.localize(dt) |
|
167 | ||
168 | ||
169 |
def encode_datetime(dt): |
|
170 |
return dt.astimezone(pytz.timezone('Europe/Paris')).strftime(xml_datetime_format) |
|
171 | ||
172 | ||
173 | 89 |
def upperize(data): |
174 | 90 |
if isinstance(data, dict): |
175 | 91 |
for key, val in data.items(): |
passerelle/contrib/utils/axel.py | ||
---|---|---|
1 |
# passerelle - uniform access to multiple data sources and services |
|
2 |
# Copyright (C) 2021 Entr'ouvert |
|
3 |
# |
|
4 |
# This program is free software: you can redistribute it and/or modify it |
|
5 |
# under the terms of the GNU Affero General Public License as published |
|
6 |
# by the Free Software Foundation, either version 3 of the License, or |
|
7 |
# (at your option) any later version. |
|
8 |
# |
|
9 |
# This program is distributed in the hope that it will be useful, |
|
10 |
# but WITHOUT ANY WARRANTY; without even the implied warranty of |
|
11 |
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|
12 |
# GNU Affero General Public License for more details. |
|
13 |
# |
|
14 |
# You should have received a copy of the GNU Affero General Public License |
|
15 |
# along with this program. If not, see <http://www.gnu.org/licenses/>. |
|
16 | ||
17 | ||
18 |
import copy |
|
19 |
import datetime |
|
20 |
import os |
|
21 |
import re |
|
22 |
import xml.etree.ElementTree as ET |
|
23 |
from collections import namedtuple |
|
24 | ||
25 |
import pytz |
|
26 | ||
27 |
from django.utils.encoding import force_text |
|
28 | ||
29 |
import xmlschema |
|
30 | ||
31 |
from passerelle.utils.xml import JSONSchemaFromXMLSchema |
|
32 | ||
33 | ||
34 |
boolean_type = { |
|
35 |
'oneOf': [ |
|
36 |
{'type': 'boolean'}, |
|
37 |
{ |
|
38 |
'type': 'string', |
|
39 |
'pattern': '[Oo][Uu][Ii]|[Nn][Oo][Nn]|[Tt][Rr][Uu][Ee]|[Ff][Aa][Ll][Ss][Ee]|1|0', |
|
40 |
}, |
|
41 |
] |
|
42 |
} |
|
43 |
date_type = { |
|
44 |
'type': 'string', |
|
45 |
'pattern': '[0-9]{4}-[0-9]{2}-[0-9]{2}', |
|
46 |
} |
|
47 |
datetime_type = { |
|
48 |
'type': 'string', |
|
49 |
'pattern': '[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}', |
|
50 |
} |
|
51 |
json_date_format = '%Y-%m-%d' |
|
52 |
json_datetime_format = '%Y-%m-%dT%H:%M:%S' |
|
53 |
xml_date_format = '%d/%m/%Y' |
|
54 |
xml_datetime_format = '%d/%m/%Y %H:%M:%S' |
|
55 | ||
56 | ||
57 |
def indent(tree, space=" ", level=0): |
|
58 |
# backport from Lib/xml/etree/ElementTree.py python 3.9 |
|
59 |
if isinstance(tree, ET.ElementTree): |
|
60 |
tree = tree.getroot() |
|
61 |
if level < 0: |
|
62 |
raise ValueError("Initial indentation level must be >= 0, got {level}".format(level=level)) |
|
63 |
if not len(tree): |
|
64 |
return |
|
65 | ||
66 |
# Reduce the memory consumption by reusing indentation strings. |
|
67 |
indentations = ["\n" + level * space] |
|
68 | ||
69 |
def _indent_children(elem, level): |
|
70 |
# Start a new indentation level for the first child. |
|
71 |
child_level = level + 1 |
|
72 |
try: |
|
73 |
child_indentation = indentations[child_level] |
|
74 |
except IndexError: |
|
75 |
child_indentation = indentations[level] + space |
|
76 |
indentations.append(child_indentation) |
|
77 | ||
78 |
if not elem.text or not elem.text.strip(): |
|
79 |
elem.text = child_indentation |
|
80 | ||
81 |
for child in elem: |
|
82 |
if len(child): |
|
83 |
_indent_children(child, child_level) |
|
84 |
if not child.tail or not child.tail.strip(): |
|
85 |
child.tail = child_indentation |
|
86 | ||
87 |
# Dedent after the last child by overwriting the previous indentation. |
|
88 |
if not child.tail.strip(): |
|
89 |
child.tail = indentations[level] |
|
90 | ||
91 |
_indent_children(tree, 0) |
|
92 | ||
93 | ||
94 |
def encode_bool(obj): |
|
95 |
if obj is True or str(obj).lower() in ['true', 'oui', '1']: |
|
96 |
return 'OUI' |
|
97 |
if obj is False or str(obj).lower() in ['false', 'non', '0']: |
|
98 |
return 'NON' |
|
99 |
return obj |
|
100 | ||
101 | ||
102 |
def parse_datetime(value): |
|
103 |
try: |
|
104 |
dt = datetime.datetime.strptime(value, json_datetime_format) |
|
105 |
except ValueError: |
|
106 |
return None |
|
107 |
return pytz.utc.localize(dt) |
|
108 | ||
109 | ||
110 |
def encode_datetime(dt): |
|
111 |
return dt.astimezone(pytz.timezone('Europe/Paris')).strftime(xml_datetime_format) |
|
112 | ||
113 | ||
114 |
class AxelError(Exception): |
|
115 |
def __init__(self, message, xml_request=None, xml_response=None, *args): |
|
116 |
self.message = message |
|
117 |
self.xml_request = xml_request |
|
118 |
self.xml_response = xml_response |
|
119 |
super(AxelError, self).__init__(message, *args) |
|
120 | ||
121 |
def __str__(self): |
|
122 |
return self.message |
|
123 | ||
124 | ||
125 |
class AxelSchema(JSONSchemaFromXMLSchema): |
|
126 |
type_map = { |
|
127 |
'{urn:AllAxelTypes}DATEREQUIREDType': 'date', |
|
128 |
'{urn:AllAxelTypes}DATEType': 'date_optional', |
|
129 |
'{urn:AllAxelTypes}OUINONREQUIREDType': 'bool', |
|
130 |
'{urn:AllAxelTypes}OUINONType': 'bool_optional', |
|
131 |
} |
|
132 | ||
133 |
@classmethod |
|
134 |
def schema_date(cls): |
|
135 |
return { |
|
136 |
'type': 'string', |
|
137 |
'pattern': '[0-9]{4}-[0-9]{2}-[0-9]{2}', |
|
138 |
} |
|
139 | ||
140 |
def encode_date(self, obj): |
|
141 |
try: |
|
142 |
return datetime.datetime.strptime(obj, json_date_format).strftime(xml_date_format) |
|
143 |
except ValueError: |
|
144 |
return obj |
|
145 | ||
146 |
def encode_date_optional(self, obj): |
|
147 |
if not obj: |
|
148 |
return obj |
|
149 |
return self.encode_date(obj) |
|
150 | ||
151 |
def decode_date(self, data): |
|
152 |
value = datetime.datetime.strptime(data.text, xml_date_format).strftime(json_date_format) |
|
153 |
return xmlschema.ElementData( |
|
154 |
tag=data.tag, text=value, content=data.content, attributes=data.attributes |
|
155 |
) |
|
156 | ||
157 |
def decode_date_optional(self, data): |
|
158 |
if not data.text: |
|
159 |
return data |
|
160 |
return self.decode_date(data) |
|
161 | ||
162 |
@classmethod |
|
163 |
def schema_bool(cls): |
|
164 |
return copy.deepcopy(boolean_type) |
|
165 | ||
166 |
def encode_bool(self, obj): |
|
167 |
return encode_bool(obj) |
|
168 | ||
169 |
def decode_bool(self, data): |
|
170 |
value = False |
|
171 |
if data.text.lower() == 'oui': |
|
172 |
value = True |
|
173 |
return xmlschema.ElementData( |
|
174 |
tag=data.tag, text=value, content=data.content, attributes=data.attributes |
|
175 |
) |
|
176 | ||
177 |
@classmethod |
|
178 |
def schema_bool_optional(cls): |
|
179 |
schema_bool_optional = cls.schema_bool() |
|
180 |
schema_bool_optional['oneOf'].append({'type': 'string', 'enum': ['']}) |
|
181 |
return schema_bool_optional |
|
182 | ||
183 |
def encode_bool_optional(self, obj): |
|
184 |
return self.encode_bool(obj) |
|
185 | ||
186 |
def decode_bool_optional(self, data): |
|
187 |
if not data.text: |
|
188 |
return data |
|
189 |
return self.decode_bool(data) |
|
190 | ||
191 | ||
192 |
def xml_schema_converter(base_xsd_path, name, root_element): |
|
193 |
xsd_path = os.path.join(base_xsd_path, name) |
|
194 |
if not os.path.exists(xsd_path): |
|
195 |
return None |
|
196 |
return AxelSchema(xsd_path, root_element) |
|
197 | ||
198 | ||
199 |
OperationResult = namedtuple('OperationResult', ['json_response', 'xml_request', 'xml_response']) |
|
200 | ||
201 | ||
202 |
class Operation(object): |
|
203 |
base_xsd_path = None |
|
204 |
default_prefix = '' |
|
205 | ||
206 |
def __init__(self, operation, prefix=None, request_root_element='PORTAIL'): |
|
207 |
if prefix is None: |
|
208 |
prefix = self.default_prefix |
|
209 |
self.operation = operation |
|
210 |
self.request_converter = xml_schema_converter( |
|
211 |
self.base_xsd_path, '%sQ_%s.xsd' % (prefix, operation), request_root_element |
|
212 |
) |
|
213 |
self.response_converter = xml_schema_converter( |
|
214 |
self.base_xsd_path, '%sR_%s.xsd' % (prefix, operation), 'PORTAILSERVICE' |
|
215 |
) |
|
216 |
self.name = re.sub( |
|
217 |
'(.?)([A-Z])', lambda s: s.group(1) + ('-' if s.group(1) else '') + s.group(2).lower(), operation |
|
218 |
) |
|
219 |
self.snake_name = self.name.replace('-', '_') |
|
220 | ||
221 |
@property |
|
222 |
def request_schema(self): |
|
223 |
schema = self.request_converter.json_schema |
|
224 |
schema['flatten'] = True |
|
225 |
schema['merge_extra'] = True |
|
226 |
return schema |
|
227 | ||
228 |
def __call__(self, resource, request_data=None): |
|
229 |
client = resource.soap_client() |
|
230 | ||
231 |
serialized_request = '' |
|
232 |
if self.request_converter: |
|
233 |
try: |
|
234 |
serialized_request = self.request_converter.encode(request_data) |
|
235 |
except xmlschema.XMLSchemaValidationError as e: |
|
236 |
raise AxelError('invalid request %s' % str(e)) |
|
237 |
indent(serialized_request) |
|
238 |
serialized_request = force_text(ET.tostring(serialized_request)) |
|
239 |
try: |
|
240 |
self.request_converter.xml_schema.validate(serialized_request) |
|
241 |
except xmlschema.XMLSchemaValidationError as e: |
|
242 |
raise AxelError('invalid request %s' % str(e), xml_request=serialized_request) |
|
243 | ||
244 |
result = client.service.getData( |
|
245 |
self.operation, serialized_request, '' |
|
246 |
) # FIXME: What is the user parameter for ? |
|
247 | ||
248 |
xml_result = ET.fromstring(result.encode('utf-8')) |
|
249 |
indent(xml_result) |
|
250 |
pretty_result = force_text(ET.tostring(xml_result)) |
|
251 |
if xml_result.find('RESULTAT/STATUS').text != 'OK': |
|
252 |
msg = xml_result.find('RESULTAT/COMMENTAIRES').text |
|
253 |
raise AxelError(msg, xml_request=serialized_request, xml_response=pretty_result) |
|
254 | ||
255 |
try: |
|
256 |
return OperationResult( |
|
257 |
json_response=self.response_converter.decode(xml_result), |
|
258 |
xml_request=serialized_request, |
|
259 |
xml_response=pretty_result, |
|
260 |
) |
|
261 |
except xmlschema.XMLSchemaValidationError as e: |
|
262 |
raise AxelError( |
|
263 |
'invalid response %s' % str(e), xml_request=serialized_request, xml_response=pretty_result |
|
264 |
) |
passerelle/contrib/toulouse_axel/exceptions.py → tests/test_axel_utils.py | ||
---|---|---|
1 | 1 |
# passerelle - uniform access to multiple data sources and services |
2 |
# Copyright (C) 2020 Entr'ouvert
|
|
2 |
# Copyright (C) 2021 Entr'ouvert
|
|
3 | 3 |
# |
4 | 4 |
# This program is free software: you can redistribute it and/or modify it |
5 | 5 |
# under the terms of the GNU Affero General Public License as published |
... | ... | |
14 | 14 |
# You should have received a copy of the GNU Affero General Public License |
15 | 15 |
# along with this program. If not, see <http://www.gnu.org/licenses/>. |
16 | 16 | |
17 |
from passerelle.contrib.utils.axel import ( |
|
18 |
parse_datetime, |
|
19 |
encode_datetime, |
|
20 |
) |
|
17 | 21 | |
18 |
class AxelError(Exception): |
|
19 |
def __init__(self, message, xml_request=None, xml_response=None, *args): |
|
20 |
self.message = message |
|
21 |
self.xml_request = xml_request |
|
22 |
self.xml_response = xml_response |
|
23 |
super(AxelError, self).__init__(message, *args) |
|
24 | 22 | |
25 |
def __str__(self): |
|
26 |
return self.message |
|
23 |
def test_parse_datetime(): |
|
24 |
# wrong format |
|
25 |
assert parse_datetime('foo') is None |
|
26 |
assert parse_datetime('2019-12-12') is None |
|
27 |
assert parse_datetime('2019-12-12T12:01:72') is None |
|
28 |
# ok |
|
29 |
assert parse_datetime('2019-12-12T12:01:42').isoformat() == '2019-12-12T12:01:42+00:00' |
|
30 | ||
31 | ||
32 |
def test_encode_datetime(): |
|
33 |
assert encode_datetime(parse_datetime('2019-12-12T23:40:42')) == '13/12/2019 00:40:42' |
tests/test_toulouse_axel.py | ||
---|---|---|
33 | 33 |
import xmlschema |
34 | 34 |
import jsonschema |
35 | 35 | |
36 |
from passerelle.contrib.toulouse_axel.exceptions import AxelError |
|
37 | 36 |
from passerelle.contrib.toulouse_axel.models import ( |
38 | 37 |
Link, |
39 | 38 |
Lock, |
... | ... | |
48 | 47 |
regime_mapping, |
49 | 48 |
upperize, |
50 | 49 |
) |
50 |
from passerelle.contrib.utils.axel import AxelError |
|
51 |
from passerelle.contrib.utils.axel import OperationResult |
|
51 | 52 |
from passerelle.utils.jsonresponse import APIError |
52 | 53 |
from passerelle.utils.soap import SOAPError |
53 | 54 |
import utils |
... | ... | |
675 | 676 |
% content |
676 | 677 |
) |
677 | 678 |
with mock_getdata(content, 'RefVerifDui'): |
678 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.AxelSchema.decode') as decode:
|
|
679 |
with mock.patch('passerelle.contrib.utils.axel.AxelSchema.decode') as decode:
|
|
679 | 680 |
decode.side_effect = xmlschema.XMLSchemaValidationError(None, None) |
680 | 681 |
resp = app.post_json('/toulouse-axel/test/link?NameID=yyy', params=link_params) |
681 | 682 |
assert resp.json['err_desc'].startswith("Axel error: invalid response") |
... | ... | |
1278 | 1279 |
assert 'xml_response' in resp.json['data'] |
1279 | 1280 | |
1280 | 1281 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.form_maj_famille_dui') as operation: |
1281 |
operation.return_value = schemas.OperationResult(json_response={}, xml_request='', xml_response='')
|
|
1282 |
operation.return_value = OperationResult(json_response={}, xml_request='', xml_response='') |
|
1282 | 1283 |
with mock.patch( |
1283 | 1284 |
'passerelle.contrib.toulouse_axel.models.ToulouseAxel.get_family_data', return_value=family_data |
1284 | 1285 |
): |
... | ... | |
1292 | 1293 |
link.person_id = '35' |
1293 | 1294 |
link.save() |
1294 | 1295 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.form_maj_famille_dui') as operation: |
1295 |
operation.return_value = schemas.OperationResult(json_response={}, xml_request='', xml_response='')
|
|
1296 |
operation.return_value = OperationResult(json_response={}, xml_request='', xml_response='') |
|
1296 | 1297 |
with mock.patch( |
1297 | 1298 |
'passerelle.contrib.toulouse_axel.models.ToulouseAxel.get_family_data', return_value=family_data |
1298 | 1299 |
): |
... | ... | |
2735 | 2736 |
} |
2736 | 2737 |
} |
2737 | 2738 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.enfants_activites') as operation: |
2738 |
operation.return_value = schemas.OperationResult( |
|
2739 |
json_response=result, xml_request='', xml_response='' |
|
2740 |
) |
|
2739 |
operation.return_value = OperationResult(json_response=result, xml_request='', xml_response='') |
|
2741 | 2740 |
resp = app.get('/toulouse-axel/test/clae_children_activities_info?NameID=yyy&booking_date=2020-01-20') |
2742 | 2741 |
assert resp.json['err'] == 0 |
2743 | 2742 |
assert len(resp.json['data']) == expected and 1 or 0 |
... | ... | |
4000 | 3999 |
assert 'xml_response' in resp.json['data'] |
4001 | 4000 | |
4002 | 4001 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.reservation_annuelle') as operation: |
4003 |
operation.return_value = schemas.OperationResult( |
|
4004 |
json_response={}, xml_request='', xml_response='' |
|
4005 |
) |
|
4002 |
operation.return_value = OperationResult(json_response={}, xml_request='', xml_response='') |
|
4006 | 4003 |
resp = app.post_json('/toulouse-axel/test/clae_booking?NameID=yyy', params=booking_params) |
4007 | 4004 |
payload = operation.call_args_list[0][0][1]['PORTAIL']['DUI'] |
4008 | 4005 |
assert payload == { |
... | ... | |
4069 | 4066 |
new_booking_params['booking_list_SOIR'] = [] |
4070 | 4067 |
new_booking_params['booking_list_GARD'] = [] |
4071 | 4068 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.reservation_annuelle') as operation: |
4072 |
operation.return_value = schemas.OperationResult( |
|
4073 |
json_response={}, xml_request='', xml_response='' |
|
4074 |
) |
|
4069 |
operation.return_value = OperationResult(json_response={}, xml_request='', xml_response='') |
|
4075 | 4070 |
resp = app.post_json('/toulouse-axel/test/clae_booking?NameID=yyy', params=new_booking_params) |
4076 | 4071 |
payload = operation.call_args_list[0][0][1]['PORTAIL']['DUI'] |
4077 | 4072 |
assert len(payload['ENFANT']) == 1 |
... | ... | |
4109 | 4104 |
new_booking_params['booking_list_SOIR'] = None |
4110 | 4105 |
new_booking_params['booking_list_GARD'] = None |
4111 | 4106 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.reservation_annuelle') as operation: |
4112 |
operation.return_value = schemas.OperationResult( |
|
4113 |
json_response={}, xml_request='', xml_response='' |
|
4114 |
) |
|
4107 |
operation.return_value = OperationResult(json_response={}, xml_request='', xml_response='') |
|
4115 | 4108 |
resp = app.post_json('/toulouse-axel/test/clae_booking?NameID=yyy', params=new_booking_params) |
4116 | 4109 |
payload = operation.call_args_list[0][0][1]['PORTAIL']['DUI'] |
4117 | 4110 |
assert 'ACTIVITE' not in payload['ENFANT'][0] |
... | ... | |
4134 | 4127 |
new_booking_params['booking_list_SOIR'] = None |
4135 | 4128 |
new_booking_params['booking_list_GARD'] = None |
4136 | 4129 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.reservation_annuelle') as operation: |
4137 |
operation.return_value = schemas.OperationResult( |
|
4138 |
json_response={}, xml_request='', xml_response='' |
|
4139 |
) |
|
4130 |
operation.return_value = OperationResult(json_response={}, xml_request='', xml_response='') |
|
4140 | 4131 |
with mock.patch('django.core.cache.cache.delete') as mock_cache_delete: |
4141 | 4132 |
resp = app.post_json('/toulouse-axel/test/clae_booking?NameID=yyy', params=new_booking_params) |
4142 | 4133 |
assert mock_cache_delete.call_args_list == [ |
... | ... | |
4180 | 4171 |
): |
4181 | 4172 |
booking_params['regime'] = None |
4182 | 4173 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.reservation_annuelle') as operation: |
4183 |
operation.return_value = schemas.OperationResult( |
|
4184 |
json_response={}, xml_request='', xml_response='' |
|
4185 |
) |
|
4174 |
operation.return_value = OperationResult(json_response={}, xml_request='', xml_response='') |
|
4186 | 4175 |
app.post_json('/toulouse-axel/test/clae_booking?NameID=yyy', params=booking_params) |
4187 | 4176 |
payload = operation.call_args_list[0][0][1]['PORTAIL']['DUI'] |
4188 | 4177 |
assert payload['ENFANT'][0]['REGIME'] == 'SV' |
4189 | 4178 |
booking_params['regime'] = '' |
4190 | 4179 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.reservation_annuelle') as operation: |
4191 |
operation.return_value = schemas.OperationResult( |
|
4192 |
json_response={}, xml_request='', xml_response='' |
|
4193 |
) |
|
4180 |
operation.return_value = OperationResult(json_response={}, xml_request='', xml_response='') |
|
4194 | 4181 |
app.post_json('/toulouse-axel/test/clae_booking?NameID=yyy', params=booking_params) |
4195 | 4182 |
payload = operation.call_args_list[0][0][1]['PORTAIL']['DUI'] |
4196 | 4183 |
assert payload['ENFANT'][0]['REGIME'] == 'SV' |
4197 | 4184 |
del booking_params['regime'] |
4198 | 4185 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.reservation_annuelle') as operation: |
4199 |
operation.return_value = schemas.OperationResult( |
|
4200 |
json_response={}, xml_request='', xml_response='' |
|
4201 |
) |
|
4186 |
operation.return_value = OperationResult(json_response={}, xml_request='', xml_response='') |
|
4202 | 4187 |
app.post_json('/toulouse-axel/test/clae_booking?NameID=yyy', params=booking_params) |
4203 | 4188 |
payload = operation.call_args_list[0][0][1]['PORTAIL']['DUI'] |
4204 | 4189 |
assert payload['ENFANT'][0]['REGIME'] == 'SV' |
... | ... | |
4228 | 4213 |
) as mock_activities: |
4229 | 4214 |
mock_activities.return_value = activities |
4230 | 4215 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.reservation_annuelle') as operation: |
4231 |
operation.return_value = schemas.OperationResult( |
|
4232 |
json_response={}, xml_request='', xml_response='' |
|
4233 |
) |
|
4216 |
operation.return_value = OperationResult(json_response={}, xml_request='', xml_response='') |
|
4234 | 4217 |
app.post_json('/toulouse-axel/test/clae_booking?NameID=yyy', params=booking_params) |
4235 | 4218 |
assert mock_activities.call_args_list == [ |
4236 | 4219 |
mock.call(child_id='3535', dui='XXX', reference_year=2020), |
... | ... | |
4305 | 4288 |
assert 'xml_response' in resp.json['data'] |
4306 | 4289 | |
4307 | 4290 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.reservation_annuelle') as operation: |
4308 |
operation.return_value = schemas.OperationResult( |
|
4309 |
json_response={}, xml_request='', xml_response='' |
|
4310 |
) |
|
4291 |
operation.return_value = OperationResult(json_response={}, xml_request='', xml_response='') |
|
4311 | 4292 |
resp = app.post_json( |
4312 | 4293 |
'/toulouse-axel/test/clae_booking_annual?NameID=yyy', params=annual_booking_params |
4313 | 4294 |
) |
... | ... | |
4360 | 4341 |
new_booking_params['booking_list_SOIR'] = [] |
4361 | 4342 |
new_booking_params['booking_list_GARD'] = [] |
4362 | 4343 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.reservation_annuelle') as operation: |
4363 |
operation.return_value = schemas.OperationResult( |
|
4364 |
json_response={}, xml_request='', xml_response='' |
|
4365 |
) |
|
4344 |
operation.return_value = OperationResult(json_response={}, xml_request='', xml_response='') |
|
4366 | 4345 |
resp = app.post_json( |
4367 | 4346 |
'/toulouse-axel/test/clae_booking_annual?NameID=yyy', params=new_booking_params |
4368 | 4347 |
) |
... | ... | |
4402 | 4381 |
new_booking_params['booking_list_SOIR'] = None |
4403 | 4382 |
new_booking_params['booking_list_GARD'] = None |
4404 | 4383 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.reservation_annuelle') as operation: |
4405 |
operation.return_value = schemas.OperationResult( |
|
4406 |
json_response={}, xml_request='', xml_response='' |
|
4407 |
) |
|
4384 |
operation.return_value = OperationResult(json_response={}, xml_request='', xml_response='') |
|
4408 | 4385 |
resp = app.post_json( |
4409 | 4386 |
'/toulouse-axel/test/clae_booking_annual?NameID=yyy', params=new_booking_params |
4410 | 4387 |
) |
... | ... | |
4434 | 4411 |
'passerelle.contrib.toulouse_axel.models.ToulouseAxel.get_child_activities', return_value=activities |
4435 | 4412 |
): |
4436 | 4413 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.reservation_annuelle') as operation: |
4437 |
operation.return_value = schemas.OperationResult( |
|
4438 |
json_response={}, xml_request='', xml_response='' |
|
4439 |
) |
|
4414 |
operation.return_value = OperationResult(json_response={}, xml_request='', xml_response='') |
|
4440 | 4415 |
app.post_json('/toulouse-axel/test/clae_booking_annual?NameID=yyy', params=annual_booking_params) |
4441 | 4416 |
payload = operation.call_args_list[0][0][1]['PORTAIL']['DUI'] |
4442 | 4417 |
assert payload['ENFANT'][0]['ACTIVITE'][0]['PERIODE'][0]['DATEDEBUT'] == '2020-08-01' |
... | ... | |
4452 | 4427 |
): |
4453 | 4428 |
annual_booking_params['regime'] = None |
4454 | 4429 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.reservation_annuelle') as operation: |
4455 |
operation.return_value = schemas.OperationResult( |
|
4456 |
json_response={}, xml_request='', xml_response='' |
|
4457 |
) |
|
4430 |
operation.return_value = OperationResult(json_response={}, xml_request='', xml_response='') |
|
4458 | 4431 |
app.post_json('/toulouse-axel/test/clae_booking_annual?NameID=yyy', params=annual_booking_params) |
4459 | 4432 |
payload = operation.call_args_list[0][0][1]['PORTAIL']['DUI'] |
4460 | 4433 |
assert payload['ENFANT'][0]['REGIME'] == 'SV' |
4461 | 4434 |
annual_booking_params['regime'] = '' |
4462 | 4435 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.reservation_annuelle') as operation: |
4463 |
operation.return_value = schemas.OperationResult( |
|
4464 |
json_response={}, xml_request='', xml_response='' |
|
4465 |
) |
|
4436 |
operation.return_value = OperationResult(json_response={}, xml_request='', xml_response='') |
|
4466 | 4437 |
app.post_json('/toulouse-axel/test/clae_booking_annual?NameID=yyy', params=annual_booking_params) |
4467 | 4438 |
payload = operation.call_args_list[0][0][1]['PORTAIL']['DUI'] |
4468 | 4439 |
assert payload['ENFANT'][0]['REGIME'] == 'SV' |
4469 | 4440 |
del annual_booking_params['regime'] |
4470 | 4441 |
with mock.patch('passerelle.contrib.toulouse_axel.schemas.reservation_annuelle') as operation: |
4471 |
operation.return_value = schemas.OperationResult( |
|
4472 |
json_response={}, xml_request='', xml_response='' |
|
4473 |
) |
|
4442 |
operation.return_value = OperationResult(json_response={}, xml_request='', xml_response='') |
|
4474 | 4443 |
app.post_json('/toulouse-axel/test/clae_booking_annual?NameID=yyy', params=annual_booking_params) |
4475 | 4444 |
payload = operation.call_args_list[0][0][1]['PORTAIL']['DUI'] |
4476 | 4445 |
assert payload['ENFANT'][0]['REGIME'] == 'SV' |
tests/test_toulouse_axel_schema.py | ||
---|---|---|
16 | 16 | |
17 | 17 |
import os |
18 | 18 | |
19 |
from passerelle.contrib.toulouse_axel.schemas import AxelSchema
|
|
19 |
from passerelle.contrib.utils.axel import AxelSchema
|
|
20 | 20 | |
21 | 21 |
import pytest |
22 | 22 |
import xmlschema |
tests/test_toulouse_axel_utils.py | ||
---|---|---|
18 | 18 |
import pytest |
19 | 19 | |
20 | 20 |
from passerelle.contrib.toulouse_axel.utils import ( |
21 |
parse_datetime, |
|
22 |
encode_datetime, |
|
23 | 21 |
get_booking, |
24 | 22 |
get_reference_year_from_date, |
25 | 23 |
get_week_dates_from_date, |
26 |
json_date_format, |
|
27 | 24 |
) |
28 | ||
29 | ||
30 |
def test_parse_datetime(): |
|
31 |
# wrong format |
|
32 |
assert parse_datetime('foo') is None |
|
33 |
assert parse_datetime('2019-12-12') is None |
|
34 |
assert parse_datetime('2019-12-12T12:01:72') is None |
|
35 |
# ok |
|
36 |
assert parse_datetime('2019-12-12T12:01:42').isoformat() == '2019-12-12T12:01:42+00:00' |
|
37 | ||
38 | ||
39 |
def test_encode_datetime(): |
|
40 |
assert encode_datetime(parse_datetime('2019-12-12T23:40:42')) == '13/12/2019 00:40:42' |
|
25 |
from passerelle.contrib.utils.axel import json_date_format |
|
41 | 26 | |
42 | 27 | |
43 | 28 |
@pytest.mark.parametrize( |
44 |
- |