Produits Entr'ouvert » Passerelle

    category = _('Civil Status Connectors')

    class Meta:

    @property

    def basepath(self):

class AirQuality(BaseResource):

    category = _('Misc')

    api_description = _(

        This API provides a unique format for the air quality data of various places.

        (But only supports the Rhône-Alpes region for now).

        '''

        try:

            response = self.requests.get(url, data=params, cache_duration=300)

        except requests.RequestException as e:

        try:

            data = response.json()

        except ValueError as e:

            content = response.text[:1000]

            raise APIError(

                % (response.status_code, content),

                data={

                    'status_code': response.status_code,

                    'err_desc': data.get('message', 'not-found'),

                }

            raise APIError(

                data={

                    'status_code': response.status_code,

                    'content': data,

            response = self.requests.get(url, headers=headers, timeout=5, **kwargs)

        except requests.RequestException as e:

            raise APIError(

                log_error=True,

                data={

                    'code': 'connection-error',

        except JSONDecodeError as e:

            content = repr(response.content[:1000])

            raise APIError(

                % (self.platform, response.status_code, content),

                log_error=True,

                data={

                    }

                raise APIError(message, data=error_data)

            raise APIError(

                % (self.platform, response.status_code, data),

                log_error=True,

                data={

            )

            response.raise_for_status()

        except RequestException as e:

        try:

            result = response.json()

        except ValueError:

        if result.get('Data'):

            if 'AccessToken' not in result['Data']:

            if not isinstance(result['Data']['AccessToken'], six.string_types):

            return result['Data']['AccessToken']

    @endpoint(

        name='api',

            response = self.requests.get(url, params=params, auth=auth)

            response.raise_for_status()

        except RequestException as e:

        data = []

        try:

            result = response.json()

        except ValueError:

        if not result.get('Data'):

            raise APIError("%s (%s)" % (result.get('LibErreur'), result.get('CodErreur')))

        for demand in result['Data']['results']:

                receipt_time = parse_time(data_administratives['heure_depot'])

                receipt_date = parse_date(data_administratives['date_depot'])

            except (KeyError, TypeError) as e:

            d = {

                'url': demand['url'],

                'title': data_administratives.get('LibelleQualificationTypeDemande'),

    category = _('Business Process Connectors')

    class Meta:

    def check_status(self):

        self._astre_call('astre/webservices/gf/documents/referentiel/typedocument/list/json')

    log_requests_errors = False

    class Meta:

    def check_status(self):

        response = self.requests.get(self.wsdl_base_url)

        parameters={

            'category': {

                'description': _('Category of codification'),

            }

        },

    )

        # create CIVILITE

        for identification in d.get('IDENTIFICATION', []):

            sexe = identification.get('SEXE', '')

        return d

    @endpoint(

            data.append(

                {

                    'id': str(link.id),

                    'id_per': link.id_per,

                    'dossier': dossier,

                }

                cmu_nais = to_ascii(identification.get('CMU_NAIS', '')).lower()

                if cmu_nais and commune_naissance != cmu_nais:

                    self.logger.debug(

                        id_per,

                        cmu_nais,

                        commune_naissance,

                def format_cerfa_label(x):

                    try:

                        if x['Description']:

                    except KeyError:

                        pass

                data_cache, dummy = CartaDSDataCache.objects.get_or_create(

                    data_type='liste_pdf',

                    'err': 1,

                    'err_desc': 'Choosit error: some destinations failed',

                    'data': [

                    ],

                },

            },

                    'err': 1,

                    'err_desc': 'Choosit error: some destinations failed',

                    'data': [

                    ],

                },

            },

            {

                'response': {

                    'sms_id': 1234,

                },

                'result': {

                    'err': 0,

                    'data': [

                    ],

                },

            },

                    if not isinstance(output, dict):

                        results.append('Choosit error: JSON response is not a dict %r' % output)

                    elif 'error' in output:

                    else:

                        results.append(output)

        if any(isinstance(result, string_types) for result in results):

    category = _('Civil Status Connectors')

    class Meta:

    @classmethod

    def get_verbose_name(cls):

        try:

            query = Query.objects.get(resource=self.id, slug=query_name)

        except Query.DoesNotExist:

        return self.execute_query(request, query, query_params=kwargs)

    def execute_query(self, request, query, query_params=None):

                        data['name'] = titles[i]

                    else:

                        data['idx'] = i

                codes.append((code, expr))

            for row in data:

                new_row = []

                            data['name'] = titles[i]

                        else:

                            data['idx'] = i

                    new_row.append(result)

                yield new_row, row

                    hash(new_row)

                except TypeError:

                    raise APIError(

                        data={

                            'row': repr(row),

                            'distinct': repr(new_row),

            for mapping in projection:

                name, expr = mapping.split(':', 1)

                if not identifier_re.match(name):

                titles.append(name)

                expressions.append(expr)

            new_data = []

    @property

    def fullname(self):

    def __str__(self):

        return self.fullname

    category = _('Business Process Connectors')

    class Meta:

    def check_status(self):

        with self.outcoming_sftp.client() as out_sftp:

    category = _('Business Process Connectors')

    class Meta:

    @classmethod

    def get_verbose_name(cls):

            # Set aec_nature if aec_type_raw == DECES

            if formdata.get('aec_type_raw') == 'DECES' and not formdata.get('aec_nature_raw'):

                formdata['aec_nature_raw'] = 'COPIE-INTEGRALE'

            # Set motif_demand if none

        except ValueError:

            self.handle_opengis_error(response)

            # if handle_opengis_error did not raise an error, we raise a generic one

        for feature in response['features']:

            feature['text'] = feature['properties'].get(property_name)

            data.append(feature)

            return None

        content = exception_text.text

        content = html.unescape(content)

    def convert_coordinates(self, lon, lat, reverse=False):

        lon, lat = float(lon), float(lat)

            with self.resource.output_sftp.client() as client:

                with client.open(self.request.response_zip_filename, mode='w') as fd:

                    self.request.build_response_zip(

                    )

            with self.resource.input_sftp.client() as client:

                with client.open('DONE/' + self.request.response_zip_filename, mode='w') as fd:

                    self.request.build_response_zip(

                    )

        def get_data(self, data, name):

            numero_permis_construire = get('doc_declarant_designation_permis_numero_permis_construire')

            numero_permis_amenager = get('doc_declarant_designation_permis_numero_permis_amenager')

            data['type_permis'] = (

            )

            data['numero_permis'] = numero_permis_construire or numero_permis_amenager

            particulier = get('doc_declarant_identite_type_personne').strip().lower() == 'true'

            if particulier:

                data['nom'] = get('doc_declarant_identite_personne_physique_nom')

                data['prenoms'] = get('doc_declarant_identite_personne_physique_prenom')

                    get('doc_declarant_identite_personne_morale_representant_personne_morale_civilite'), ''

                )

            data['portee'] = (

                if get('doc_ouverture_chantier_totalite_travaux').lower().strip() == 'true'

            )

        def update_data_recensementCitoyen(self, data):

            data['motif'] = {'RECENSEMENT': '1', 'EXEMPTION': '2'}[motif]

            if data['motif'] == '2':

                data['motif_exempte'] = (

                    if get('recensementcitoyen_formalite_formalitemotifcode_2') == 'INFIRME'

                )

            data['justificatif_exemption'] = get('pj_je')

            data['double_nationalite'] = 'Oui' if get('recensementcitoyen_personne_nationalite') else 'Non'

            data['justificatif_identite'] = get('pj_ji')

            situation_matrimoniale = get('recensementcitoyen_personne_situationfamille_situationmatrimoniale')

            data['situation_familiale'] = {

                data['situation_familiale_precision'] = situation_matrimoniale

            pupille = get('recensementcitoyen_personne_situationfamille_pupille')

            data['pupille'] = 'Oui' if pupille else 'Non'

            data['pupille_categorie'] = {

            }.get(pupille)

            for idx in ['', '_1', '_2']:

                code = get('recensementcitoyen_personne_methodecontact%s_canalcode' % idx)

            './/{%(mdel)s}MotDePasse' % ns

        ).text

        response.find('.//{%(pec)s}Etat' % ns).text = '100'

        return response

    def build_response_zip(self, fd_or_filename, etat, commentaire):

    log_requests_errors = False

    class Meta:

    @classmethod

    def get_verbose_name(cls):

    )

    def __str__(self):

    def clean(self):

        if self.keytype and not self.key:

            if status == 'down' and not self.down():  # new downtime

                if availability_parameters.has_zero_delay():

                    self.logger.error(

                    )

                else:

                    self.logger.warning(

                    )

            ResourceStatus(

                resource_type=resource_type, resource_pk=self.pk, status=status, message=message

            ).save()

            if status == 'up' and current_status:

        elif status == 'down':

            # check last_notification_downtime and current downtime to see if it matches a new notification delay

            last_notification_timestamp = (

                        else:

                            human_duration = 'for %d minutes' % downtime

                        self.logger.error(

                            self,

                            self.__class__.__name__,

                            human_duration,

        # log changes to notification delays

        if 'notification_delays' in form.changed_data:

        # log changes to run_check, if enabled immediately check for availability

        if 'run_check' in form.changed_data:

            resource.logger.info(

            )

            if form.instance.run_check:

                resource.availability()

            }

            if code_retour in errors:

                return Error(errors[code_retour], msg=msg_retour)

    @endpoint(perm='can_access', description=_('Check service availibity'))

    def ping(self, request, *args, **kwargs):

            try:

                pdf_content = to_pdf(file_raw_content)

            except ValueError as e:

                continue

            filename = value['filename']

            if not filename.lower().endswith('.pdf'):

    category = _('Business Process Connectors')

    class Meta:

    @classmethod

    def get_verbose_name(cls):

from .formdata import CREATION_SCHEMA, FormData, list_schema_fields

# taken from https://lsimons.wordpress.com/2011/03/17/stripping-illegal-characters-out-of-xml-in-python/

def escape_xml_illegal_chars(val, replacement='?'):

            for person in entourage:

                if not isinstance(person, dict):

                    raise APIError('entourage-content-must-be-dicts', data=content)

                    new_entourage.setdefault('parents', []).append(person)

                else:

                    new_entourage.setdefault('aidants', []).append(person)

                raise APIError('demandes-content-must-be-dicts', data=content)

            new_demandes = {}

            typologies = {

            }

            if not all(isinstance(demande.get('typologie'), six.text_type) for demande in demandes):

                raise APIError('typologie-must-be-a-string', data=content)

            parts.append(prenom)

        if nom:

            parts.append(nom)

        return ' '.join(parts)

    def __str__(self):

    class Meta:

        unique_together = (

            )

    criterium = []

    for xpath, local in config['input'].items():

    returndata = []

    for xpath in config['output']:

        returndata.append(xpath)

        x = entity

        for k in xpath.split('/'):

            x = x.get(k, {})

    return d

        ret = []

        for v in villes:

            #

            # However, the lack of some informations has already been observed,

            # so there are some test/continue here

        lieux = call.get('results')

        ret = []

        for l in lieux:

            for k, v in l.items():

                l[k] = v.strip()

            ret.append(

        individus = call.get('results')

        ret = []

        for i in individus:

            for k, v in i.items():

                i[k] = v.strip()

            njf = i['EtatCivil/NomJeuneFille/@V']

            if njf:

            if not i['EtatCivil/DateNaissance/@V']:

            ret.append(

                {

                    'id': '%(PK/IndexIndividu/@V)s' % i,

                    intervenant['nom'] = _intervenant['Nom']['@V']

                    intervenant['prenom'] = _intervenant['Prenom']['@V']

                    coords = _intervenant.get('Coordonnees', {})

                    intervenants.append(intervenant)

            info['visite_intervenants'] = intervenants

            select = {}

        streamId = operation

        xmlParams = smart_text(ET.tostring(portail, encoding='UTF-8'))

        user = ''

        result = soap.get_client(self).service.getData(streamId, smart_text(xmlParams), user)

        xml_result = ET.fromstring(result)

        if xml_result.find('RESULTAT/STATUS').text != 'OK':

            msg = xml_result.find('RESULTAT/COMMENTAIRES').text

            activity['id'] = activity['IDACTIVITE']

            start_date = datetime.datetime.strptime(activity['DATEENTREE'], axel.json_date_format)

            end_date = datetime.datetime.strptime(activity['DATESORTIE'], axel.json_date_format)

                activity['LIBELLEACTIVITE'],

                start_date.strftime(axel.xml_date_format),

                end_date.strftime(axel.xml_date_format),

    default_country_code = models.CharField(

        verbose_name=_('Default country code'),

        max_length=3,

        validators=[RegexValidator('^[0-9]*$', _('The country must only contain numbers'))],

    )

    default_trunk_prefix = models.CharField(

        verbose_name=_('Default trunk prefix'),

        max_length=2,

        validators=[RegexValidator('^[0-9]*$', _('The trunk prefix must only contain numbers'))],

    )  # Yeah France first !

    max_message_length = models.IntegerField(

            content = '<exception-while-rendering-args>'

    except AttributeError:

        content = ''

def normalize(s):

                s.append(child.tail)

        return s

def to_json(root):

    openssl_cnf_path = tmpdir / 'openssl.cnf'

    with openssl_cnf_path.open('w') as fd:

        fd.write(

[default_conf]

ssl_conf = ssl_sect

@urlmatch(netloc='^entreprise.api.gouv.fr$')

def api_entreprise_error_not_found(url, request):

@pytest.yield_fixture

        404,

        {

            'error': 'not_found',

        },

        request=request,

    )

def test_check_status(mocked_get, connector):

    mocked_get.return_value = tests.utils.FakedResponse(content=FAKE_HELLO_RESPONSE, status_code=200)

    resp = connector.check_status()

@mock.patch('passerelle.utils.Request.get')

def test_get_types_equipement(app, connector, monkeypatch):

    <types>

        <type id="2" label="Espaces Verts"></type>

        <type id="4" label="Voirie">

        assert response.json['data'][0]['id_per'] == '1234'

        assert response.json['data'][0]['dossier']

        assert response.json['data'][0]['id'] == str(link.id)

        assert response.json['data'][0]['dossier']['IDENTIFICATION'][0]['CIVILITE'] == 'Madame'

        assert len(response.json['data'][0]['dossier']['DEMANDES']) == 1

        assert len(response.json['data'][0]['dossier']['DEMANDES']['AD']) == 1

        assert response.json['data'][0]['id_per'] == '1234'

        assert response.json['data'][0]['dossier']

        assert response.json['data'][0]['id'] == str(link.id)

        assert response.json['data'][1]['id_per'] == '4567'

        assert response.json['data'][1]['dossier']

        assert response.json['data'][1]['id'] == str(link2.id)

    with tests.utils.mock_url(FAKE_URL, RESPONSE_SELECT_USAGER):

        response = app.get(

        assert response.json['data']['id_per'] == '4567'

        assert response.json['data']['dossier']

        assert response.json['data']['id'] == str(link2.id)

def test_row_locked_cache(genesys, freezer):

    data = response.json['data']

    assert data == [

        {

        },

        {

        },

        {

        },

    ]

import tests.utils

from passerelle.apps.feeds.models import Feed

<feed xmlns="http://www.w3.org/2005/Atom" xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#" xmlns:sy="http://purl.org/rss/1.0/modules/syndication/" xmlns:dc="http://purl.org/dc/elements/1.1/" xmlns:taxo="http://purl.org/rss/1.0/modules/taxonomy/">

  <title>Actualités de Strasbourg</title>

  <link rel="alternate" href="http://www.strasbourg.eu/fr/actualites/-/asset_publisher/lG7u/rss" />

def street(db):

    return StreetModel.objects.create(

        ban_id='73001_0000',

        resource=BaseAdresse.objects.first(),

    )

@pytest.fixture

def region(db):

    return RegionModel.objects.create(

    )

@pytest.fixture

def department(db, region):

    return DepartmentModel.objects.create(

    )

@pytest.fixture

def city(db, region, department):

    return CityModel.objects.create(

        code='73065',

        zipcode='73000',

        population=42000,

@pytest.fixture

def miquelon(db):

    return CityModel.objects.create(

        code='97501',

        zipcode='97500',

        population=42,

    resp = app.get('/base-adresse/%s/search' % base_adresse.slug, status=400)

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

    assert resp.json['err_class'] == 'passerelle.views.WrongParameter'

    # json-api serializer

    resp = app.get('/base-adresse/%s/streets?zipcode=13400&coin=zz' % base_adresse.slug, status=400)

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

        # create additional streets

        other_street = StreetModel.objects.create(

            ban_id='%d_000T' % (73001 + i),

            zipcode=str(73001 + i),

            type='street',

            citycode=str(73001 + i),

def test_base_adresse_streets_get_by_codes(app, base_adresse, street):

    for i in range(20):

        StreetModel.objects.create(

            zipcode=str(75000 + i * 10),

            type='street',

            citycode=str(75000 + i * 11),

def test_base_adresse_cities_region_department(app, base_adresse, miquelon, city):

    CityModel.objects.create(

        code='75056',

        zipcode='75014',

        population=2000000,

def test_base_adresse_departments_region(app, base_adresse, department):

    resp = app.get('/base-adresse/%s/departments?region_code=84' % base_adresse.slug)

    result = resp.json['data']

            },

            {

                'IdPiece': 62,

                'Descriptif': 'Un plan de masse des constructions...',

                'CodePiece': 'PCA1',

                'Reglementaire': False,

        return [

            OrderedDict(

                [

                ]

            ),

            OrderedDict(

                [

                ]

            ),

        ]

        return [

            OrderedDict(

                [

                ]

            ),

            OrderedDict(

                [

                ]

            ),

        ]

        assert id_dossier in (123, '135792')

        return OrderedDict(

            [

            ]

        )

    assert len(data) == 5

    assert data[0]['text'] == 'Cerfa rempli'

    assert data[0]['max_files'] == 1

    assert data[1]['max_files'] == 6

    assert data[2]['text'] == 'Plan de situation du terrain'

    assert data[2]['max_files'] == 6

    resp = app.get('/cartads-cs/test/status?dossier_id=%s' % dossier.id)

    assert (

        resp.json['status_label']

    )

    dossier.zip_ack_response = 'False'

    dossier.save()

    resp = app.get('/cartads-cs/test/status?dossier_id=%s' % dossier.id)

def test_additional_pieces_management(connector, app, cached_data):

    assert len(data) == 2

    assert data[0]['text'] == 'Document graphique permettant...'

    assert data[0]['max_files'] == 6

    assert data[1]['max_files'] == 6

    for piece in data:

        assert len(piece['files']) == 1

def test_unicode_filter_data(client, setup, filetype):

    _, url = setup(',id,,text,', filename=filetype, data=get_file_content(filetype))

    filters = {'text': filter_criteria}

    resp = client.get(url, filters)

def test_unicode_case_insensitive_filter_data(client, setup, filetype):

    _, url = setup(',id,,text,', filename=filetype, data=get_file_content(filetype))

    filters = {'text': filter_criteria, 'case-insensitive': ''}

    resp = client.get(url, filters)

    result = parse_response(resp)

        elif self.error_class is TransportError:

            raise self.error_class('connection error occured', None)

        elif self.error_class is WebFaultHavingLatin1:

        else:

            raise Exception('random error')

        assert 'Error: random error' in resp.json['err_desc']

        client.return_value = get_client(error_class=WebFaultHavingLatin1)

        resp = app.get('/dpark/test/ping/')

def test_ping(dpark, app):

        assert resp.json['code'] == 'unknown-file'

        # invalid cardnumber

        client.return_value = get_client(

        )

        resp = app.get('/dpark/test/search/', params=params)

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

        assert resp.json['code'] == 'unknown-file'

        # invalid cardnumber

        client.return_value = get_client(

        )

        resp = app.post_json('/dpark/test/link/', params=params)

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

        url = '/dpark/test/check-renewal-time/'

        params = {'firstnames': 'spam eggs', 'lastname': 'bar', 'filenumber': '1' * 9, 'badgenumber': '2' * 9}

        client.return_value = get_client(

        )

        resp = app.get(url, params=params)

        assert resp.json['data'] is False

        client.return_value = get_client(replydata={'CodeRetour': '01', 'MessageRetour': 'Dossier existant'})

        resp = app.get(url, params=params)

        assert resp.json['data'] is True

        url = '/dpark/test/check-renewal-time/'

        params = {'firstnames': 'spam eggs', 'lastname': 'bar', 'filenumber': '1' * 9, 'badgenumber': '2' * 9}

        client.return_value = get_client(

        )

        resp = app.get(url, params=params)

        assert resp.json['data'] is False

        client.return_value = get_client(replydata={'CodeRetour': '01', 'MessageRetour': 'Dossier existant'})

        resp = app.get(url, params=params)

        assert resp.json['data'] is True

            'applicant_lastname': 'Ham',

        }

        client.return_value = get_client(

        )

        resp = app.get(url, params=params)

        assert resp.json['data'] is False

        client.return_value = get_client(replydata={'CodeRetour': '01', 'MessageRetour': ''})

        resp = app.get(url, params=params)

        assert resp.json['data'] is True

            'applicant_lastname': 'Ham',

        }

        client.return_value = get_client(

        )

        resp = app.get(url, params=params)

        assert resp.json['data'] is False

        client.return_value = get_client(replydata={'CodeRetour': '01', 'MessageRetour': ''})

        resp = app.get(url, params=params)

        assert resp.json['data'] is True

            'badgenumber': '2' * 9,

        }

        Pairing.objects.create(resource=dpark, **params)

        resp = app.get(url)

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

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

        replydata = {

            'CodeRetour': '01',

            'NumeroDemande': '55555',

            'Montant': 12500,

            'TypePaiement': 10,

                'badgenumber': '2' * 9,

            },

        )

        resp = app.post_json(url, params=params)

        assert operation.call_args_list[-1].args[5] == expected_date

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

    )

    assert demande_abonnementtiers in force_text(dpark.mock_requests[0].body)

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

    # with complete application

    dpark.mock_responses.append(

        """<?xml version="1.0" encoding="utf-8"?>

        soap_call.side_effect = APIError('Numéro de demande introuvable')

        resp = app.post_json(url, params=params)

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

    params['justif_domicile'] = {

        'filename': 'address_proof.pDf',

    params['impot_revenu'] = 'this is my tax notice'

    resp = app.post_json(url, params=params)

    errors = [

    ]

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

    assert resp.json['err_desc'] == '%s' % errors

import tests.utils

from passerelle.apps.feeds.models import Feed

<feed xmlns="http://www.w3.org/2005/Atom" xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#" xmlns:sy="http://purl.org/rss/1.0/modules/syndication/" xmlns:dc="http://purl.org/dc/elements/1.1/" xmlns:taxo="http://purl.org/rss/1.0/modules/taxonomy/">

  <title>Actualités de Strasbourg</title>

  <link rel="alternate" href="http://www.strasbourg.eu/fr/actualites/-/asset_publisher/lG7u/rss" />

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

        assert len(resp.json['data']) == 3

        assert resp.json['data'][0]['id'] == '16151'

        resp = app.get(endpoint + '/typology/', status=200)

        assert requests_get.call_count == 2

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

        assert len(resp.json['data']) == 12

        assert resp.json['data'][0]['id'] == '13067'

        assert resp.json['data'][0]['service_id'] == '10173'

        assert (

            resp.json['data'][0]['service_text']

        )

        resp = app.get(endpoint + '/typology/?service_id=10426', status=200)

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

        assert len(resp.json['data']) == 7

        assert resp.json['data'][0]['id'] == '10804'

        assert resp.json['data'][0]['service_id'] == '10426'

def test_gdema_referentiel(app, gdema):

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

        assert resp.json['data']['Id'] == '1'

        assert resp.json['data']['AnswerToProvide'] is True

        assert resp.json['data']['ExpectedDate'] is None

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

        assert resp.json['data']['Handler']['CivilityId'] == '0'

        assert requests_get.call_args[0][0] == 'https://gdema.example.net/api/request/1/state'

        assert 'data' in resp.json

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

def test_gdema_create_request(app, gdema):

    assert record.levelno == 20

    assert record.levelname == 'INFO'

    assert record.name == 'passerelle.resource.arcgis.test'

    assert not hasattr(record, 'connector_result')

    record = caplog.records[1]

    assert record.levelno == 10

    assert record.levelname == 'DEBUG'

    assert record.name == 'passerelle.resource.arcgis.test'

    assert hasattr(record, 'connector_result')

    data = resp.json['data']

    with patch_init, patch_object:

        resp = app.post_json(url_foo, params=payload, status=400)

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

    with patch_init, patch_object:

        resp = app.post_json(url_bar, params=payload)

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

    with mock.patch('passerelle.utils.Request.post') as request_post:

        request_post.return_value = get_typo_response()

        payload = BASIC_PAYLOAD.copy()

        app.post_json(

            reverse(

                'generic-endpoint',

            ),

            params=payload,

        )

        assert request_post.call_args[1]['data']['obs_demande_urgente'] == 1

        assert request_post.call_args[1]['data']['dem_comp'] == 'Voirie'

        assert request_post.call_args[1]['data']['dem_reponse'] == 0

        assert request_post.call_args[1]['data']['obs_description_probleme'] == 'thank you '

    )

    json_result = response.json_body

    assert json_result['err'] == 1

def test_checkdate_sti_code_optionnal_last_char(app, setup, endpoint_dummy_cache, monkeypatch):

        'companyLocation': {'number': '', 'road': '', 'zipCode': '', 'city': ''},

        'pictures': [],

        'subType': '/api/sub_types/164',

    }

    assert response.json['data'] == json.loads(DEMAND_CREATION_RESPONSE)

    assert mocked_post.call_count == 2

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

        assert len(resp.json['data']) == 4

        assert resp.json['data'][0]['id'] == '1'

def test_okina_institutions(app, okina):

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

            assert len(resp.json['data']) == 2

            assert resp.json['data'][0]['id'] == '277'

            assert resp.json['data'][0]['lat'] == '46.791332586468016'

            assert resp.json['data'][0]['lon'] == '1.6926355997775921'

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

            assert len(resp.json['data']) == 2

            assert resp.json['data'][0]['id'] == '277'

            resp = app.get(endpoint + '/from-city/36005/', status=200)

            assert (

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

            assert len(resp.json['data']) == 2

            assert resp.json['data'][0]['id'] == '277'

def test_okina_search(app, okina):

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

        assert len(resp.json['data']) == 3

        assert resp.json['data'][0]['id'] == '3281'

def test_okina_ods(app, okina):

        assert len(resp.json['data']) == 1

        assert resp.json['data'][0]['id'] == 'inst:276-seq:1-6-84'