caluire-axel: boolean types (#53825)
This commit is contained in:
Lauréline Guérin
parent
fe5aa8da1c
commit
4e893cb433
|
|
@ -17,13 +17,59 @@
|
|||
import copy
|
||||
import os
|
||||
|
||||
import xmlschema
|
||||
|
||||
from passerelle.contrib.utils import axel
|
||||
|
||||
BASE_XSD_PATH = os.path.join(os.path.dirname(__file__), 'xsd')
|
||||
|
||||
|
||||
boolean_type = {
|
||||
'oneOf': [
|
||||
{'type': 'boolean'},
|
||||
{
|
||||
'type': 'string',
|
||||
'pattern': '[Oo]|[Nn]|[Tt][Rr][Uu][Ee]|[Ff][Aa][Ll][Ss][Ee]|1|0',
|
||||
},
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
def encode_bool(obj):
|
||||
if obj is True or str(obj).lower() in ['true', 'o', '1']:
|
||||
return 'O'
|
||||
if obj is False or str(obj).lower() in ['false', 'n', '0']:
|
||||
return 'N'
|
||||
return obj
|
||||
|
||||
|
||||
class CaluireAxelSchema(axel.AxelSchema):
|
||||
type_map = {
|
||||
'{urn:AllAxelTypes}DATEREQUIREDType': 'date',
|
||||
'{urn:AllAxelTypes}DATEType': 'date_optional',
|
||||
'{urn:AllAxelTypes}ONType': 'bool',
|
||||
'{urn:AllAxelTypes}ONEmptyType': 'bool_optional',
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def schema_bool(cls):
|
||||
return copy.deepcopy(boolean_type)
|
||||
|
||||
def encode_bool(self, obj):
|
||||
return encode_bool(obj)
|
||||
|
||||
def decode_bool(self, data):
|
||||
value = False
|
||||
if data.text.lower() == 'o':
|
||||
value = True
|
||||
return xmlschema.ElementData(
|
||||
tag=data.tag, text=value, content=data.content, attributes=data.attributes
|
||||
)
|
||||
|
||||
|
||||
class Operation(axel.Operation):
|
||||
base_xsd_path = BASE_XSD_PATH
|
||||
axel_schema = CaluireAxelSchema
|
||||
|
||||
|
||||
find_individus = Operation('FindIndividus')
|
||||
|
|
|
|||
|
|
@ -54,8 +54,8 @@
|
|||
<xsd:element name="EMPLOYEUR" type="all:FOOBARType" minOccurs="0" maxOccurs="1"/><!-- CUSTOM -->
|
||||
<xsd:element name="VILLEEMP" type="all:FOOBARType" minOccurs="0" maxOccurs="1"/><!-- CUSTOM -->
|
||||
<xsd:element name="REGIME1" type="all:FOOBARType" minOccurs="0" maxOccurs="1"/><!-- CUSTOM -->
|
||||
<xsd:element name="PAI" type="all:FOOBARType" minOccurs="0" maxOccurs="1"/><!-- CUSTOM -->
|
||||
<xsd:element name="GARDEALTERNEE" type="all:FOOBARType"/><!-- CUSTOM -->
|
||||
<xsd:element name="PAI" type="all:ONEmptyType" minOccurs="0" maxOccurs="1"/><!-- CUSTOM -->
|
||||
<xsd:element name="GARDEALTERNEE" type="all:ONEmptyType"/><!-- CUSTOM -->
|
||||
<xsd:element name="ADRESSE" type="adr:ADRESSEType" minOccurs="0" maxOccurs="1"/>
|
||||
<xsd:element name="FAMILLE" type="ind:FAMILLEType" minOccurs="0" maxOccurs="unbounded"/><!-- CUSTOM -->
|
||||
</xsd:sequence>
|
||||
|
|
|
|||
|
|
@ -186,11 +186,11 @@ class AxelSchema(JSONSchemaFromXMLSchema):
|
|||
return self.decode_bool(data)
|
||||
|
||||
|
||||
def xml_schema_converter(base_xsd_path, name, root_element):
|
||||
def xml_schema_converter(axel_schema, base_xsd_path, name, root_element):
|
||||
xsd_path = os.path.join(base_xsd_path, name)
|
||||
if not os.path.exists(xsd_path):
|
||||
return None
|
||||
return AxelSchema(xsd_path, root_element)
|
||||
return axel_schema(xsd_path, root_element)
|
||||
|
||||
|
||||
OperationResult = namedtuple('OperationResult', ['json_response', 'xml_request', 'xml_response'])
|
||||
|
|
@ -199,16 +199,17 @@ OperationResult = namedtuple('OperationResult', ['json_response', 'xml_request',
|
|||
class Operation(object):
|
||||
base_xsd_path = None
|
||||
default_prefix = ''
|
||||
axel_schema = AxelSchema
|
||||
|
||||
def __init__(self, operation, prefix=None, request_root_element='PORTAIL'):
|
||||
if prefix is None:
|
||||
prefix = self.default_prefix
|
||||
self.operation = operation
|
||||
self.request_converter = xml_schema_converter(
|
||||
self.base_xsd_path, '%sQ_%s.xsd' % (prefix, operation), request_root_element
|
||||
self.axel_schema, self.base_xsd_path, '%sQ_%s.xsd' % (prefix, operation), request_root_element
|
||||
)
|
||||
self.response_converter = xml_schema_converter(
|
||||
self.base_xsd_path, '%sR_%s.xsd' % (prefix, operation), 'PORTAILSERVICE'
|
||||
self.axel_schema, self.base_xsd_path, '%sR_%s.xsd' % (prefix, operation), 'PORTAILSERVICE'
|
||||
)
|
||||
self.name = re.sub(
|
||||
'(.?)([A-Z])', lambda s: s.group(1) + ('-' if s.group(1) else '') + s.group(2).lower(), operation
|
||||
|
|
|
|||
|
|
@ -0,0 +1,80 @@
|
|||
# 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/>.
|
||||
|
||||
import os
|
||||
|
||||
import pytest
|
||||
import xmlschema
|
||||
|
||||
from passerelle.contrib.caluire_axel.schemas import CaluireAxelSchema
|
||||
|
||||
XSD_BASE_DIR = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)), '../passerelle/contrib/caluire_axel/xsd'
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('bool_type', ['ONType', 'ONEmptyType'])
|
||||
@pytest.mark.parametrize(
|
||||
'value, expected, py_expected',
|
||||
[
|
||||
('O', 'O', True),
|
||||
('o', 'O', True),
|
||||
('TRUE', 'O', True),
|
||||
('true', 'O', True),
|
||||
('True', 'O', True),
|
||||
(True, 'O', True),
|
||||
('1', 'O', True),
|
||||
('N', 'N', False),
|
||||
('n', 'N', False),
|
||||
('FALSE', 'N', False),
|
||||
('false', 'N', False),
|
||||
('False', 'N', False),
|
||||
(False, 'N', False),
|
||||
('0', 'N', False),
|
||||
('FOOBAR', 'FOOBAR', None),
|
||||
('42', '42', None),
|
||||
('OUIFOOBAR', 'OUIFOOBAR', None),
|
||||
('FOONONBAR', 'FOONONBAR', None),
|
||||
],
|
||||
)
|
||||
def test_bool_mapping(bool_type, value, expected, py_expected):
|
||||
if expected == '' and bool_type == 'ONType':
|
||||
# required, can't be empty
|
||||
return
|
||||
|
||||
xsd = """<?xml version="1.0" encoding="utf-8" ?>
|
||||
<xsd:schema xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:all="urn:AllAxelTypes">
|
||||
<xsd:import schemaLocation="{path}/AllAxelTypes.xsd" namespace="urn:AllAxelTypes" />
|
||||
<xsd:complexType name="PORTAILType">
|
||||
<xsd:sequence>
|
||||
<xsd:element name="BOOL" type="all:{bool_type}"/>
|
||||
</xsd:sequence>
|
||||
</xsd:complexType>
|
||||
<xsd:element name="PORTAIL" type="PORTAILType"/>
|
||||
</xsd:schema>""".format(
|
||||
path=XSD_BASE_DIR, bool_type=bool_type
|
||||
)
|
||||
|
||||
schema = CaluireAxelSchema(xsd, 'PORTAIL')
|
||||
xml_data = schema.encode({'PORTAIL': {'BOOL': value}})
|
||||
assert xml_data.find('BOOL').text == expected
|
||||
|
||||
if py_expected is None:
|
||||
with pytest.raises(xmlschema.XMLSchemaValidationError):
|
||||
schema.decode(xml_data)
|
||||
else:
|
||||
json_data = schema.decode(xml_data)
|
||||
assert json_data['BOOL'] is py_expected
|
||||
Loading…
Reference in New Issue