Produits Entr'ouvert » Passerelle

# Copyright (C) 2017  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

from collections import OrderedDict

from dateutil.parser import parse as dateutil_parse

from django.core.files.storage import default_storage

from django.core.files.base import ContentFile

from passerelle.utils.jsonresponse import APIError

from .utils import ElementFactory, json_to_xml, zipdir, etree

from .mapping import MAPPING

ALLOWED_VALUES = {

    '_title_raw': ('M', 'Mme', 'Mlle'),  # {applicant_, concerned_, partner_}{father, mother}

    '_gender_raw': ('M', 'F', 'NA'),  # {applicant_, concerned_, partner_}{father, mother}

    'document_kind_raw': ('CPI', 'EXTAF', 'EXTSF', 'EXTPL'),

    'concerned_kind_raw': ('reconnu', 'auteur'),

    'event_kind_raw': ('NAI', 'MAR', 'REC', 'DEC'),

    'canal_raw': ('internet', 'guichet', 'courrier')

}

def is_value_allowed(key, value, allowed_values):

    if value not in allowed_values:

        raise APIError('Invalid value for %s: %s not in %s'

                       % (key, value, allowed_values))

class CivilStatusApplication(object):

    root_element = 'demandeEtatCivil'

    mapping = MAPPING

    namespace = 'http://tempuri.org/XMLSchema.xsd'

    def __init__(self, demand_id, data):

        self.application_id = demand_id

        self.data = self.validate(data)

    def validate(self, data):

        data['application_id'] = self.application_id

        data['application_date'] = data['receipt_time']

        applicant_is_concerned = data.pop('applicant_is_concerned', False)

        if applicant_is_concerned:

            data['concerned_kind'] = 'auteur'

        for key in data.keys():

            # set all dates to ISO format

            if '_date' in key and data.get(key):

                if key == 'application_date':

                    continue

                data[key] = dateutil_parse(data[key]).date().isoformat()

            # check value validity for key with value constraint

            for pattern, allowed_values in ALLOWED_VALUES.items():

                if key.endswith(pattern) and data.get(key):

                    is_value_allowed(pattern, data[key], allowed_values)

            # if the applicant is the concerned person, then all

            # applicant vars values are copied

            if key.startswith('applicant_') and applicant_is_concerned:

                if key == 'applicant_kind':

                    continue

                data[key.replace('applicant', 'concerned')] = data[key]

        return data

    @property

    def xml(self):

        elements = OrderedDict()

        for var, elt in self.mapping.items():

            if self.data.get(var):

                elements[elt] = self.data[var]

        etree.register_namespace('xs', 'http://tempuri.org/XMLSchema.xsd')

        root = ElementFactory(self.root_element, namespace=self.namespace)

        for key, value in elements.items():

            path = key.split('_')

            root.append(json_to_xml(path, value, root, namespace=self.namespace), allow_new=False)

        return root

    def save(self, path):

        filepath = os.path.join(path, self.application_id + '.xml')

        default_storage.save(filepath, ContentFile(self.xml.to_pretty_xml()))

        return zipdir(path)

import os

from dateutil.parser import parse as dateutil_parse

from passerelle.utils.jsonresponse import APIError

from .cityweb import CivilStatusApplication

from .utils import get_resource_base_dir, flatten_payload

        formdata.update(flatten_payload(formdata.pop('fields'), formdata.pop('extra')))

        # check mandatory keys

        for key in ('display_id', 'receipt_time', 'event_kind_raw'):

            if key not in formdata:

                raise APIError('<%s> is required' % key)

        attrs = {

            'demand_id': '%s-%s' % (formdata['display_id'], formdata['event_kind_raw']),

            'kind': formdata['event_kind_raw'], 'resource': self,

            'received_at': dateutil_parse(formdata['receipt_time'])}

        demand, created = Demand.objects.get_or_create(**attrs)

    received_at = models.DateTimeField()

    demand_id = models.CharField(max_length=32, primary_key=True)

        return '%s - %s - %s' % (self.resource.slug, self.demand_id)

    def filepath(self):

        return os.path.join(get_resource_base_dir(), self.resource.slug, self.demand_id)

        application = CivilStatusApplication(self.demand_id, data)

        application.save(self.filepath)

        return self.demand_id

from lxml import etree as letree

def get_test_base_dir():

    return os.path.join(os.path.dirname(__file__), 'data', 'cityweb')

    path = os.path.join(get_test_base_dir(), filename)

def assert_xml_doc(doc):

    schema = letree.XMLSchema(

        letree.parse(os.path.join(get_test_base_dir(), 'cityweb.xsd')))

    schema.assertValid(doc)

@mock.patch('passerelle.contrib.cityweb.models.get_resource_base_dir', get_test_base_dir)

    url = '/cityweb/test/create/'

        response = app.post_json(url, params=payload['naissance'])

        assert response.json['data']['demand_id'] == '15-4-NAI'

        doc = letree.parse(

            os.path.join(get_test_base_dir(), 'test', '15-4-NAI', '15-4-NAI.xml'))

        assert_xml_doc(doc)

        response = app.post_json(url, params=payload['mariage'])

        assert response.json['data']['demand_id'] == '16-1-MAR'

        doc = letree.parse(

            os.path.join(get_test_base_dir(), 'test', '16-1-MAR', '16-1-MAR.xml'))

        assert_xml_doc(doc)

        response = app.post_json(url, params=payload['deces'])

        assert response.json['data']['demand_id'] == '17-1-DEC'

        doc = letree.parse(

            os.path.join(get_test_base_dir(), 'test', '17-1-DEC', '17-1-DEC.xml'))

        assert_xml_doc(doc)

    shutil.rmtree(os.path.join(get_test_base_dir(), 'test'), ignore_errors=True)