Produits Entr'ouvert » Passerelle

from .conftest import diff, link, unlink

def test_direct_debit_order(conn, create_data):

    unlink(conn, create_data['name_id'])

    link(conn, create_data)

    payload = {

        'codeRegie': '1',

        'bank/bankBIC': 'BDFEFR2T',

        'bank/bankIBAN': 'FR7630001007941234567890185',

        'bank/bankRUM': 'xxx',

        'bank/dateStart': '2023-01-01',

        'bank/bankAddress': '75049 PARIS cedex 01',

        'bank/civility': 'x',

        'bank/lastName': 'Ewing',

        'bank/firstName': 'John Ross',

    }

    resp = requests.post(url, json=payload)

    resp.raise_for_status()

    res = resp.json()

    assert res['data'] == 'ok'

    payload = {

        'codeRegie': '1',

        'dateRef': '2023-01-01',

    }

    resp = requests.post(url, json=payload)

    resp.raise_for_status()

    res = resp.json()

    res['data']['numPerson'] = 'N/A'

}

ADD_DIRECT_DEBIT_ORDER_SCHEMA = {

    '$schema': 'http://json-schema.org/draft-04/schema#',

    'title': 'Add direct debit order',

    'description': "Ajout d'autorisation de prélèvement",

    'type': 'object',

    'properties': {

        'bank': PERSON_BANK_SCHEMA,

        'codeRegie': {

            'description': 'Code de la régie',

            'type': 'string',

        },

    },

    'additionalProperties': False,

    'unflatten': True,

}

GET_DIRECT_DEBIT_ORDER_SCHEMA = {

    '$schema': 'http://json-schema.org/draft-04/schema#',

    'title': 'Get direct debit order',

    'description': "Lecture des informations relatives à l'autorisation de prélèvement en cours à la date de référence",

    'type': 'object',

    'properties': {

        'codeRegie': {

            'description': 'Code de la régie',

            'type': 'string',

        },

        'dateRef': {

            'description': 'Date de référence ',

            'type': 'string',

            'pattern': '^[0-9]{4}-[0-9]{2}-[0-9]{2}$',

        self.assert_update_indicator_payload_in_referential('ChildIndicator', post_data)

        self.call('Family', 'updatePersonIndicatorList', numPerson=child_id, **post_data)

        return {'data': 'ok'}

    @endpoint(

        display_category='Facture',

        description="Ajout d'autorisation de prélèvement",

        perm='can_access',

        parameters={'NameID': {'description': 'Publik NameID'}},

        post={

            'request_body': {'schema': {'application/json': invoice_schemas.ADD_DIRECT_DEBIT_ORDER_SCHEMA}}

        },

    )

        family_id = self.get_link(NameID).family_id

        self.call('Invoice', 'addDirectDebitOrder', dossierNumber=family_id, **post_data)

        return {'data': 'ok'}

    @endpoint(

        display_category='Facture',

        description="Lecture des informations relatives à l'autorisation de prélèvement en cours à la date de référence",

        perm='can_access',

        parameters={'NameID': {'description': 'Publik NameID'}},

        post={

            'request_body': {'schema': {'application/json': invoice_schemas.GET_DIRECT_DEBIT_ORDER_SCHEMA}}

        },

    )

        family_id = self.get_link(NameID).family_id

        response = self.call('Invoice', 'getDirectDebitOrder', dossierNumber=family_id, **post_data)

        data = serialize_object(response)

        return {'data': data}

class Link(models.Model):

    resource = models.ForeignKey(ToulouseMaelis, on_delete=models.CASCADE)

    name_id = models.CharField(blank=False, max_length=256)

# 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 logging

import os

from unittest import mock

import pytest

from requests.exceptions import ConnectionError

from zeep import Settings

from passerelle.contrib.toulouse_maelis.models import Link, Referential, ToulouseMaelis

from passerelle.utils.jsonresponse import APIError

from passerelle.utils.soap import SOAPError

from tests.utils import FakedResponse, ResponsesSoap, generic_endpoint_url, setup_access_rights

def get_endpoint(name):

    url = generic_endpoint_url('toulouse-maelis', name)

    assert url == '/toulouse-maelis/test/%s' % name

    return url

@pytest.fixture()

        wsdl_url='https://example.org/FamilyService?wsdl',

        wsdl_content=get_xml_file('FamilyService.wsdl'),

        settings=Settings(strict=False, xsd_ignore_sequence_order=True),

@pytest.fixture()

        wsdl_url='https://example.org/ActivityService?wsdl',

        wsdl_content=get_xml_file('ActivityService.wsdl'),

        settings=Settings(strict=False, xsd_ignore_sequence_order=True),

@pytest.fixture()

        wsdl_url='https://example.org/InvoiceService?wsdl',

        wsdl_content=get_xml_file('InvoiceService.wsdl'),

        settings=Settings(strict=False, xsd_ignore_sequence_order=True),

@pytest.fixture(scope='module')

def django_db_setup(django_db_setup, django_db_blocker):

    with django_db_blocker.unblock():

        con = setup_access_rights(

            ToulouseMaelis.objects.get_or_create(

                slug='test',

            )[0]

        )

        family_service = ResponsesSoap(

            wsdl_url='https://example.org/FamilyService?wsdl',

            wsdl_content=get_xml_file('FamilyService.wsdl'),

            settings=Settings(strict=False, xsd_ignore_sequence_order=True),

        )

                'readChildIndicatorList', get_xml_file('R_read_child_indicator_list.xml')

            )

            con.daily()

    # reset change in zeep private interface to bypass clear_cache fixture

    from zeep.cache import InMemoryCache

    InMemoryCache._cache = {}

    resp = app.post_json(url + '?NameID=local&child_id=613880', params=params)

    assert resp.json['err'] == 'wrong-key'

    assert (

        resp.json['err_desc']

        == "indicatorList/0/code key value 'plop' do not belong to 'ChildIndicator' required referential"

    )

    def request_check(request):

        assert request.dossierNumber == 1312

    invoice_service.add_soap_response(

    )

    params = {

        'codeRegie': '22',

        'bank/bankBIC': 'BDFEFR2T',

        'bank/bankIBAN': 'FR7630001007941234567890185',

        'bank/bankRUM': 'xxx',

        'bank/dateStart': '2023-01-01',

        'bank/bankAddress': '75049 PARIS cedex 01',

        'bank/civility': 'x',

        'bank/lastName': 'Ewing',

    }

    Link.objects.create(resource=con, family_id='1312', name_id='local')

    resp = app.post_json(url + '?NameID=local', params=params)

    assert resp.json['err'] == 0

    assert resp.json['data'] == 'ok'

    params = {

        'codeRegie': '22',

        'bank/bankBIC': 'BDFEFR2T',

        'bank/bankIBAN': 'FR7630001007941234567890185',

        'bank/bankRUM': 'xxx',

        'bank/dateStart': '2023-01-01',

        'bank/bankAddress': '75049 PARIS cedex 01',

        'bank/civility': 'x',

        'bank/lastName': 'Ewing',

        'bank/firstName': 'John Ross',

    }

    resp = app.post_json(url + '?NameID=local', params=params)

    assert resp.json['err'] == 'not-linked'

    assert resp.json['err_desc'] == 'User not linked to family'

    invoice_service.add_soap_response(

        'addDirectDebitOrder', get_xml_file('R_direct_debit_order_soap_error.xml'), status=500

    )

    params = {

        'codeRegie': '12345',

        'bank/bankBIC': 'BDFEFR2T',

        'bank/bankIBAN': 'FR7630001007941234567890185',

        'bank/bankRUM': 'xxx',

        'bank/dateStart': '2023-01-01',

        'bank/bankAddress': '75049 PARIS cedex 01',

        'bank/civility': 'x',

        'bank/lastName': 'Ewing',

    }

    Link.objects.create(resource=con, family_id='1312', name_id='local')

    resp = app.post_json(url + '?NameID=local', params=params)

    assert resp.json['err'] == 'Invoice-addDirectDebitOrder-soap:Server'

    assert resp.json['err_desc'] == "E520 : La REGIE CS [{0}] n''existe pas dans la base Maelis"

    def request_check(request):

        assert request.dossierNumber == 1312

    invoice_service.add_soap_response(

    )

    params = {

        'codeRegie': '22',

        'dateRef': '2023-01-01',

    }

    Link.objects.create(resource=con, family_id='1312', name_id='local')

    resp = app.post_json(url + '?NameID=local', params=params)

    assert resp.json['err'] == 0

    assert resp.json['data'] == {

        'dateStart': '2023-01-01T00:00:00+01:00',

        'bankAddress': '75049 PARIS cedex 01',

        'civility': 'x',

        'lastName': 'Ewing',

        'firstName': 'John Ross',

    }

    params = {

        'codeRegie': '22',

        'dateRef': '2023-01-01',

    }

    resp = app.post_json(url + '?NameID=local', params=params)

    assert resp.json['err'] == 'not-linked'

    assert resp.json['err_desc'] == 'User not linked to family'

    invoice_service.add_soap_response(

        'getDirectDebitOrder', get_xml_file('R_direct_debit_order_soap_error.xml'), status=500

    )

    params = {

        'codeRegie': '12345',

        'dateRef': '2023-01-01',

    }

    Link.objects.create(resource=con, family_id='1312', name_id='local')

    resp = app.post_json(url + '?NameID=local', params=params)

    assert resp.json['err'] == 'Invoice-getDirectDebitOrder-soap:Server'

    assert resp.json['err_desc'] == "E520 : La REGIE CS [{0}] n''existe pas dans la base Maelis"