Produits Entr'ouvert » Passerelle

def validate_where(format_string):

    formatter = SqlFormatter()

        if ref is None:

            pass

        elif ref == '':

        contact = self.call('Contact', 'Creation', Contact=post_data)

        # address should be set separatedly

        post_data['EncodeKeyContact'] = contact.idContact

        return {'data': serialize_object(contact)}

    @endpoint(

        },

    )

    def link_by_id_per(self, request, NameID, id_per):

        link, created = Link.objects.get_or_create(resource=self, name_id=NameID, id_per=id_per)

        return {'link_id': link.pk, 'new': created}

                        department

                    )

                )

                continue

            if ban_gz.status_code != 200:

                continue

    CartaDSSubscriber = apps.get_model('cartads_cs', 'CartaDSSubscriber')

    for instance in CartaDSDossier.objects.all():

        if instance.name_id:

            instance.subscribers.add(subscriber)

        # communes

        resp = client.service.GetCommunes(self.get_token(), {})

        communes_cache.data_values = {'data': [{'id': str(x['Key']), 'text': x['Value']} for x in resp]}

        communes_cache.save()

            resp = client.service.GetTypesDossier(self.get_token(), int(commune['id']), {})

            if resp is None:

                continue

                data_type='types_dossier', data_parameters={'commune_id': int(commune['id'])}

            )

            data_cache.data_values = {'data': [{'id': str(x['Key']), 'text': x['Value']} for x in resp]}

            resp = client.service.GetObjetsDemande(self.get_token(), type_dossier_id)

            if resp is None:

                continue

                data_type='objets_demande', data_parameters={'type_dossier_id': type_dossier_id}

            )

            data_cache.data_values = {'data': [{'id': str(x['Key']), 'text': x['Value']} for x in resp]}

                        pass

                    return u'%(Nom)s' % x

                    data_type='liste_pdf',

                    data_parameters={

                        'type_dossier_id': type_dossier_id,

            resp = client.service.GetPieces(self.get_token(), type_dossier_id, objet_demande_id)

            if resp is None:

                continue

                data_type='pieces',

                data_parameters={

                    'type_dossier_id': type_dossier_id,

        },

    )

    def pieces(self, request, type_dossier_id, objet_demande_id, tracking_code, demolitions=True):

            data_type='pieces',

            data_parameters={

                'type_dossier_id': type_dossier_id,

        # this cannot be verb DELETE as we have no way to set

        # Access-Control-Allow-Methods

        signer = Signer(salt='cart@ds_cs')

        CartaDSFile.objects.filter(id=signer.unsign(file_upload)).delete()

        return {'err': 0}

                    piece.save()

    def get_file_status(self, dossier):

        response = {}

        if dossier.deleted:

            status_id = 'deleted'

            else:

                try:

                    output = r.json()

                    results.append('Choosit error: bad JSON response')

                else:

                    if not isinstance(output, dict):

        return result

    def cache_data(self):

        with transaction.atomic():

            TableRow.objects.filter(resource=self).delete()

            for block in batch(enumerate(self.get_rows()), 5000):

        rows = [[smart_text(x) for x in y] for y in rows if y]

        titles = [t.strip() for t in self.columns_keynames.split(',')]

        indexes = [titles.index(t) for t in titles if t]

        def get_cell(row, index):

            try:

            except IndexError:

                return ''

    def get_cached_rows(self, initial=True):

        found = False

            post_data['user']['address'] = {}

        post_data['codeRDV'] = id

        return {'data': {'id': id, 'updated': True}}

    @endpoint(

        },

    )

    def delete_appointment(self, request, id):

        return {'data': {'id': id, 'deleted': True}}

                    invoice['online_payment'] = False

                    invoice['no_online_payment_reason'] = 'autobilling'

                resource=self.connector, external_id=row['ID_FAC'], defaults=invoice

            )

            invoice_filename = '%s_%s.pdf' % (

                        'city',

                    ),

                )

                    external_id=family_data['external_id'], resource=self.connector, defaults=data

                )

                    invoice_path = os.path.join(invoices_dir, invoice_filename)

                    # create invoice object only if associated pdf exists

                    if os.path.exists(invoice_path):

                            resource=self.connector,

                            family=family,

                            external_id=invoice_data['external_id'],

                'no_online_payment_reason': None,

                'label': external_id,

            }

                resource=self.connector, external_id=external_id, defaults=invoice

            )

            external_ids.append(external_id)

                    invoice['online_payment'] = False

                    invoice['no_online_payment_reason'] = 'autobilling'

                resource=self.connector, external_id=row['ID_FAC'], defaults=invoice

            )

            invoice_filename = '%s_%s.pdf' % (

                    ('complement', 'address_complement'),

                ),

            )

                external_id=family_data['id'], resource=self, defaults=data

            )

            for adult in family_data.get('adults') or []:

            self.add('fc_token_endpoint_response', response_content)

            self.add('fc_access_token', response_content['access_token'])

            self.add('fc_id_token', response_content['id_token'])

            self.add('fc_id_token_payload', json.loads(base64url_decode(payload.encode())))

        except Exception as e:

            raise FranceConnectError('Error in token endpoint response', sub_exception=repr(e))

                with transaction.atomic():

                    form = Form.objects.create(resource=self, form_id=form_id, counter=counter)

                    break

                continue

        else:

            raise APIError('fail: more than 20 demands')

    """Zip directory"""

    archname = path + '.zip'

    with zipfile.ZipFile(archname, 'w', zipfile.ZIP_DEFLATED) as zipf:

            for f in files:

                fpath = os.path.join(root, f)

                zipf.write(fpath, os.path.basename(fpath))

        access_token = self.get_access_token()

        group_id = self.group_id_from_name(access_token)

        return None  # credit consumed is unknown

        else:

            try:

                result = response.json()

                raise APIError('OVH error: bad JSON response')

        try:

            response.raise_for_status()

        else:

            try:

                result = response.json()

                raise APIError('OVH error: bad JSON response')

            else:

                if not isinstance(result, dict):

            return connector.get_absolute_url()

        try:

            result = resp.json()

            messages.error(self.request, _('There has been an error requesting token: bad JSON response.'))

            return connector.get_absolute_url()

        try:

            resp.raise_for_status()

            error_text = result.get('message', result)

            messages.error(self.request, _('There has been an error requesting token: %s.') % error_text)

            return connector.get_absolute_url()

        return token

    def request(self, path='', method='GET', params=None, json=None):

        url = urlunsplit((scheme, netloc, base_path + '/%s' % path, '', fragment))

        headers = {'Accept': 'application/json'}

        kwargs = {'method': method, 'url': url, 'headers': headers, 'params': params, 'json': json}

        if response.status_code != 204:  # No Content

            try:

                json_response = response.json()

                raise APIError('PloneRestApi: bad JSON response')

        try:

            response.raise_for_status()

        token = self.apa_token(user_id, code)  # invalid credentials raise APIError here

        information = self.apa_get_information(information='exportDonneesIndividu', token=token)

        text = get_template(self.text_template_name).render(information).strip()

            resource=self, name_id=name_id, user_id=user_id, defaults={'code': code, 'text': text}

        )

        return {'data': {'user_id': user_id, 'created': created, 'updated': not created}}

        self.rsa_token(user_id, code, dob)  # invalid credentials raise APIError here

        information = self.rsa_get_information('individu', user_id, code, dob)

        text = get_template(self.text_template_name_rsa).render(information).strip()

            resource=self, name_id=name_id, user_id=user_id, defaults={'code': code, 'dob': dob, 'text': text}

        )

        return {'data': {'user_id': user_id, 'text': text, 'created': created, 'updated': not created}}

    def variables(self):

        yield 'insee_code'

        yield 'email'

            names = [simplify(tag.localname) for tag in path]

            yield '_'.join(names)

        if hasattr(self, 'variables_%s' % self.procedure):

        else:

            internal_code = response.json()['InternalCode']

            # update contact data

            url = urlparse.urljoin(self.url, 'Contact/Put')

            response = self.requests.put(

                url,

            continue

        content_type = ContentType.objects.get_for_model(model)

        for instance in model.objects.all():

                resource_type=content_type, resource_pk=instance.id

            )

            parameters.log_level = instance.log_level

    def get_endpoints_infos(self):

        endpoints = []

            if hasattr(method, 'endpoint_info'):

                method.endpoint_info.object = self

                endpoint_name = method.endpoint_info.name

        app_label, model_name = d['resource_type'].split('.')

        try:

            model = apps.get_model(app_label, model_name)

            raise BaseResource.UnknownBaseResourceError(app_label)

        try:

            instance = model.objects.get(slug=d['slug'])

    def handle_job_error(self, job, exc_info):

        from passerelle.utils.conversion import exception_to_text

        job.status = 'failed'

        job.done_timestamp = timezone.now()

        job.status_details = {

            attr['sourceip'] = sourceip

            if kwargs.get('exc_info'):

                attr['extra']['error_summary'] = traceback.format_exception_only(exc_type, exc_value)

            ResourceLog.objects.create(**attr)

        },

    )

    def feature_info(self, request, lat, lon):

        query_args['token'] = self.api_token

        query_args['sector_type'] = self.sector_type

        response = self.requests.get(

        for value in json_loads(sections).get('values'):

            if not value.get('codefuv') or not value.get('codetroncon'):

                continue

                codefuv=value.get('codefuv'), codetroncon=value.get('codetroncon')

            )

            for attribute in ('nom', 'nomcommune', 'domanialite', 'codeinsee'):

                        message.attach(part)

                    message._write_headers = lambda x: None

                    # RFC 2045 defines MIME multipart boundaries:

                    # * boundary := 0*69<bchars> bcharsnospace

                    # * dash-boundary := "--" boundary

    message['InformationsBancaires'] = get_info_bancaire(fields, wf)

    message['Patrimoine'] = get_patrimoine(fields, wf)

    demande_apa = {

        'statutDemande': statut_demande,

        try:

            ret = response.json()

            return ret

            raise APIError('Response content is not a valid JSON')

    def get_resource_url(self, uri):

    def get_lieux(self, q, commune, departement):

        # si commune est un code solis de la forme commune-dep-com

        if commune and commune.startswith('commune-'):

        call = self._conciliation(

            conciliation.CONCILIATION_ADRESSE, commune=commune, departement=departement, lieu='%%%s%%' % q

        )

                response = self.requests.get(url)

                response.raise_for_status()

                for line_data in response.json()['values']:

                        code_titan=line_data['code_titan'],

                        defaults={'transport_key': key, 'ligne': line_data['ligne']},

                    )

            for feature in response.json()['features']:

                arret_data = feature['properties']

                arret_data['id_data'] = arret_data.pop('id')

                stop.__dict__.update(arret_data)

                stop.pmr = bool(stop.pmr == 't')

                stop.escalator = bool(stop.escalator == 't')

    def lock(self, request, key, locker):

        if not key:

            raise APIError('key is empty', err_code='bad-request', http_status=400)

        return {'key': key, 'locked': True, 'locker': lock.locker, 'lock_date': lock.lock_date}

    @endpoint(

        files = {'media': (self.filename, self.content.open('rb'), self.content_type)}

        try:

            instance.request(url, method='PUT', files=files)

            return False

        self.content.delete()

        return True

    result = JSONField(null=True)

    def get_wcs_api(self, base_url):

        services = settings.KNOWN_SERVICES.get('wcs', {})

        service = None

        for service in services.values():

            remote_url = service.get('url')

            if r_scheme == scheme and r_netloc == netloc:

                break

        else:

def client_to_jsondict(client):

    """return description of the client, as dict (for json export)"""

    res = {}

        d = {}

        d['tns'] = sd.wsdl.tns[1]

        d['prefixes'] = dict(p for p in sd.prefixes)

        self.init_stuff(request, *args, **kwargs)

        self.connector = self.get_object()

        self.endpoint = None

            if not hasattr(method, 'endpoint_info'):

                continue

            if not method.endpoint_info.name == kwargs.get('endpoint'):

def test_airquality_details_unknown_city(app, airquality):

    endpoint = tests.utils.generic_endpoint_url('airquality', 'details', slug=airquality.slug)

    assert len(data) == 3

    document = data[0]

    assert 'url' in document

    # try to get document with wrong signature

    wrong_url = document['url'] + "wrong/"

    # try expired url

    freezer.move_to(timezone.now() + timezone.timedelta(days=8))

def test_effectifs_annuels_acoss_covid(app, resource, mock_api_entreprise, freezer):

)

def test_api_particulier_dont_log_not_found(app, resource, mock, should_log):

    with HTTMock(mock):

            '/api-particulier/test/avis-imposition',

            app,

            resource,

    del hello_response['Data']

    mocked_get.return_value = tests.utils.FakedResponse(content=json.dumps(hello_response), status_code=200)

    with pytest.raises(Exception) as error:

    assert str(error.value) == 'Invalid credentials'

    mocked_post.return_value = tests.utils.FakedResponse(content=FAKE_LOGIN_OIDC_RESPONSE, status_code=200)

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

    resp = app.get(endpoint)

    assert resp.json['data']

    for item in resp.json['data']:

        assert item['status'] == 'Deposee'

        status=200,

    )

    assert len(responses.calls) == 1

    assert responses.calls[0].request.headers['login'] == 'admin'

    assert responses.calls[0].request.headers['database'] == 'db'

    connector.auth = 'AD'

    connector.save()

    assert len(responses.calls) == 2

    assert responses.calls[1].request.headers['login'] == 'admin'

    assert responses.calls[1].request.headers['database'] == 'db'

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

def test_base_adresse_search_qs_lat_lon(mocked_get, app, base_adresse):

    assert 'lat=0' in mocked_get.call_args[0][0]

    assert 'lon=1' in mocked_get.call_args[0][0]

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

    reg = RegionModel.objects.create(name=u'IdF', code='11', resource=base_adresse)

    dep = DepartmentModel.objects.create(name=u'Paris', code='75', region=reg, resource=base_adresse)

        name=u'Paris',

        code='75056',

        zipcode='75014',

def test_base_adresse_departments_region(app, base_adresse, department):

    reg = RegionModel.objects.create(name=u'IdF', code='11', resource=base_adresse)

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

    result = resp.json['data']

@pytest.mark.usefixtures('mock_update_api_geo')

@mock.patch('passerelle.utils.Request.get', side_effect=ConnectionError)

def test_base_adresse_command_update_street_timeout(mocked_get, db, base_adresse):

    assert mocked_get.call_count == 1

    assert not RegionModel.objects.exists()

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

def test_base_adresse_addresses_qs_coordinates(mocked_get, app, base_adresse_coordinates):

    assert 'lat=%s' % base_adresse_coordinates.latitude in mocked_get.call_args[0][0]

    assert 'lon=%s' % base_adresse_coordinates.longitude in mocked_get.call_args[0][0]

    assert 'lat=42' in mocked_get.call_args[0][0]

    assert 'lon=43' in mocked_get.call_args[0][0]

@pytest.mark.usefixtures('mock_update_api_geo', 'mock_update_streets')

def test_base_adresse_addresses_clean_cache(app, base_adresse, freezer, mock_api_adresse_data_gouv_fr_search):

    assert AddressCacheModel.objects.count() == 1

    freezer.move_to(datetime.timedelta(minutes=30))

    call_command('cron', 'hourly')

    assert AddressCacheModel.objects.count() == 0

    assert AddressCacheModel.objects.count() == 1

    # asking for the address again resets the timestamp

    freezer.move_to(datetime.timedelta(hours=1, seconds=1))

    call_command('cron', 'hourly')

    assert AddressCacheModel.objects.count() == 1

    freezer.move_to(datetime.timedelta(hours=1, seconds=1))

        '/base-adresse/%s/addresses?id=%s'

        % (base_adresse.slug, '49007_6950_be54bd~47.474633~-0.593775~Rue%20Roger%20Halope%2049000%20Angers')

    )

    # test_pack

    with mock.patch('passerelle.apps.cartads_cs.models.CartaDSCS.soap_client') as client:

        client.return_value = mock.Mock(service=FakeService())

            connector.jobs()

            assert Job.objects.filter(method_name='pack', status='completed').count()

            assert Job.objects.filter(method_name='send_to_cartads', status='completed').count()

    # test_pack

    with mock.patch('passerelle.apps.cartads_cs.models.CartaDSCS.soap_client') as client:

        client.return_value = mock.Mock(service=FakeService())

            connector.jobs()

            assert Job.objects.filter(method_name='pack', status='completed').count()

            assert Job.objects.filter(method_name='send_to_cartads', status='completed').count()

def test_list_of_files(connector, app, cached_data):

    CartaDSDossier.objects.all().delete()

    test_send(connector, app, cached_data)

    with mock.patch('passerelle.apps.cartads_cs.models.CartaDSCS.soap_client') as client:

        client.return_value = mock.Mock(service=FakeService())

def test_cron_error(db, caplog):

    excep = Exception('hello')

    with mock.patch(

        'passerelle.apps.base_adresse.models.AddressResource.hourly', new=mock.Mock(side_effect=excep)

def test_sheet_name_error(setup, app, filetype, admin_user):

    app = login(app)

    resp = app.get('/manage/csvdatasource/test/edit')

    edit_form = resp.forms[0]

def test_unfiltered_data(client, setup, filetype):

    resp = client.get(url)

    result = parse_response(resp)

    for item in result:

def test_empty_file(client, setup):

        'field,,another_field,', filename='data-empty.ods', data=get_file_content('data-empty.ods')

    )

    resp = client.get(url)

def test_view_manage_page(setup, app, filetype, admin_user):

    app = login(app)

    app.get(csvdata.get_absolute_url())

def test_good_filter_data(client, setup, filetype):

    filter_criteria = 'Zakia'

    filters = {'text': filter_criteria}

    resp = client.get(url, filters)

    result = parse_response(resp)

def test_bad_filter_data(client, setup, filetype):

    filter_criteria = 'bad'

    filters = {'text': filter_criteria}

    resp = client.get(url, filters)

    result = parse_response(resp)

def test_useless_filter_data(client, setup, filetype):

    filters = {'text': 'Ali'}

    resp = client.get(url, filters)

    result = parse_response(resp)

def test_columns_keynames_with_spaces(client, setup, filetype):

    filters = {'text': 'Yaniss'}

    resp = client.get(url, filters)

    result = parse_response(resp)

def test_skipped_header_data(client, setup, filetype):

    filters = {'q': 'Eliot'}

    resp = client.get(url, filters)

    result = parse_response(resp)

def test_data(client, setup, filetype):

    filters = {'text': 'Sacha'}

    resp = client.get(url, filters)

    result = parse_response(resp)

def test_unicode_filter_data(client, setup, filetype):

    filter_criteria = u'Benoît'

    filters = {'text': filter_criteria}

    resp = client.get(url, filters)

    result = parse_response(resp)

def test_unicode_case_insensitive_filter_data(client, setup, filetype):

    filter_criteria = u'anaëlle'

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

    resp = client.get(url, filters)

def test_data_bom(client, setup):

    filters = {'text': 'Eliot'}

    resp = client.get(url, filters)

    result = parse_response(resp)

def test_multi_filter(client, setup, filetype):

    filters = {'sexe': 'F'}

    resp = client.get(url, filters)

    result = parse_response(resp)

def test_query(client, setup, filetype):

    filters = {'q': 'liot'}

    resp = client.get(url, filters)

    result = parse_response(resp)

def test_query_insensitive_and_unicode(client, setup, filetype):

    filters = {'q': 'elIo', 'case-insensitive': ''}

    resp = client.get(url, filters)

    result = parse_response(resp)

def test_query_insensitive_and_filter(client, setup, filetype):

    filters = {'q': 'elIo', 'sexe': 'H', 'case-insensitive': ''}

    resp = client.get(url, filters)

    result = parse_response(resp)

def test_on_the_fly_dialect_detection(client, setup):

    # fake a connector that was not initialized during .save(), because it's

    # been migrated and we didn't do dialect detection at save() time.

    CsvDataSource.objects.all().update(_dialect_options=None)

    resp = client.get(url)

    result = json_loads(resp.content)

def test_missing_columns(client, setup):

    resp = client.get(url)

    result = json_loads(resp.content)

    assert result['err'] == 0

def test_download_file(app, setup, filetype, admin_user):

    assert '/login' in app.get('/manage/csvdatasource/test/download/').location

    app = login(app)

    resp = app.get('/manage/csvdatasource/test/download/', status=200)

def test_change_csv_command(setup):

    call_command('change-csv', 'test', os.path.join(TEST_BASE_DIR, 'data-empty.ods'))

    csv.refresh_from_db()

    assert list(csv.get_rows()) == []

    test_orleans_data_import_command()

    resource = GenericFamily.objects.get()

    family = Family.objects.get(external_id='22380')

    family = Family.objects.get(external_id='1228')

    links = resource.get_pending_invoices_by_nameid(None)

    assert len(links['data']) == 2

    for uuid, invoices in links['data'].items():

def test_incorrect_orleans_data(caplog):

    filepath = os.path.join(os.path.dirname(__file__), 'data', 'family_incorrect_data_orleans.zip')

        title='test orleans',

        slug='test-orleans',

        archive=File(open(filepath, 'rb'), 'family_incorrect_data_orleans.zip'),

    arcgis.log_evel = 'DEBUG'

    arcgis.base_url = 'https://example.net/'

    arcgis.save()

        '/arcgis/test/mapservice-query',

        params={'lon': 6.172122, 'lat': 48.673836, 'service': 'test'},

        status=200,

def test_generic_endpoint_superuser_access(db, app, admin_user, simple_user):

    filename = os.path.join(os.path.dirname(__file__), 'data', 'mdel', 'formdata.json')

    payload = json.load(open(filename))

        'card_demand_purpose': 1,

        'cards_quantity': 1,

    }

        response = app.post_json('/gesbac/test/create-demand/', params=payload)

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

    payload = copy.copy(CREATE_PAYLOAD)

    del payload['application']

    assert mocked_post.call_count == 2

    def to_json(root):

def test_export_csvdatasource(app, setup, filetype):

    query = Query(slug='query-1_', resource=csvdata, structure='array')

    query.projections = '\n'.join(['id:int(id)', 'prenom:prenom'])

    query.save()

        )

        response._content = force_bytes(soap_response)

        mocked_post.return_value = response

        base64_data = 'VGVzdCBEb2N1bWVudA=='

        data = {

            'type': typ,

    response = Response()

    response.status_code = 502

    response._content = '<p>Bad Gateway</p>'

    response.raison = 'Bad Gateway'

    mocked_post.return_value = response

    jsondatastore.text_value_template = '{{foo}}'

    jsondatastore.save()

    resp = app.post_json('/jsondatastore/foobar/data/create', params={'foo': 'bar'})

    resp = app.get('/jsondatastore/foobar/data/')

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

    assert resp.json['data'][0]['text'] == 'bar'

    # check entries are alphabetically sorted

    resp = app.post_json('/jsondatastore/foobar/data/create', params={'foo': 'aaa'})

    resp = app.get('/jsondatastore/foobar/data/')

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

    assert resp.json['data'][0]['text'] == 'aaa'

        title='a title',

        description='a description',

    )

        csv_file=File(data, 't.csv'),

        columns_keynames='id, text',

        slug='test2',

    AEC_PAYLOAD = dict(aec_payload)

    display_id = AEC_PAYLOAD['display_id']

    AEC_PAYLOAD['fields']['logitud_commentaire_usager'] = 'gentle user comment'

    # checking that attached files are referenced in -ent-.xml file

    basedir = os.path.join(get_resource_base_dir(), 'test', 'inputs', '%s-EtatCivil-0' % display_id)

    from passerelle.utils.jsonresponse import APIError

    with pytest.raises(APIError) as error:

    assert 'day is out of range for month' in str(error)

    with pytest.raises(APIError) as error:

    for text in ('date', '28-02-2018', 'not iso-formated'):

        assert text in str(error)

    endpoint = tests.utils.generic_endpoint_url('opengis', 'features', slug=connector.slug)

    assert endpoint == '/opengis/test/features'

    mocked_get.side_effect = server_responses

    assert mocked_get.call_args[1]['params']['request'] == 'GetFeature'

    assert mocked_get.call_args[1]['params']['version'] == version

    assert typename_label in mocked_get.call_args[1]['params'].keys()

    photon.latitude = 1.2

    photon.longitude = 2.1

    photon.save()

    assert 'lat=%s' % photon.latitude in mocked_get.call_args[0][0]

    assert 'lon=%s' % photon.longitude in mocked_get.call_args[0][0]

    assert 'lat=42' in mocked_get.call_args[0][0]

    assert 'lon=43' in mocked_get.call_args[0][0]

def test_create_wrong_payload(app, connector, token):

    endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'create', slug=connector.slug)

    assert endpoint == '/plone-restapi/my_connector/create'

    payload = 'not json'

    resp = app.post(endpoint + '?uri=braine-l-alleud', params=payload, status=400)

    assert resp.json['err']

def test_update_wrong_payload(app, connector, token):

    endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'update', slug=connector.slug)

    assert endpoint == '/plone-restapi/my_connector/update'

    query_string = '?uri=braine-l-alleud&uid=dccd85d12cf54b6899dff41e5a56ee7f'

    payload = 'not json'

    resp = app.post(endpoint + query_string, params=payload, status=400)

@pytest.fixture

def connector():

    connector.set_log_level('DEBUG')

    connector.url = 'http://example.net/'

    connector.save()

        raise requests.ConnectionError('timeout')

    monkeypatch.setattr(Feed, 'json', json)

    assert any('Traceback:' in mail.body for mail in mailoutbox)

    settings.LOGGED_RESPONSES_MAX_SIZE = 7

    with HTTMock(http400_mock):

        requests = Request(logger=logger)

    if logger.level == 10:  # DEBUG

        records = [record for record in caplog.records if record.name == 'requests']

    logger = logging.getLogger('requests')

    logger.setLevel(logging.DEBUG)

    requests = Request(logger=logger)

    records = [record for record in caplog.records if record.name == 'requests']

    if 'xml' in endpoint_response.headers.get('Content-Type'):

    credentials = {'id': 'id', 'key': 'key', 'algorithm': 'sha256'}

    hawk_auth = HawkAuth(**credentials)

    prepared_method = mocked_send.call_args[0][0]

    assert 'Authorization' in prepared_method.headers

    generated_header = prepared_method.headers['Authorization']

    assert dict(generated_parts) == dict(expected_parts)

    hawk_auth = HawkAuth(ext='extra attribute', **credentials)

    prepared_method = mocked_send.call_args[0][0]

    assert 'Authorization' in prepared_method.headers

    generated_header = prepared_method.headers['Authorization']

    }

    with mock.patch.object(OVHSMSGateway, 'send_msg') as send_function:

        send_function.return_value = None

        connector.jobs()

        assert send_function.call_args[1]['text'] == 'a' * connector.max_message_length

    }

    with mock.patch.object(OVHSMSGateway, 'send_msg') as send_function:

        send_function.return_value = 1

        connector.jobs()

        assert SMSLog.objects.filter(

            appname=connector.get_connector_slug(), slug=connector.slug, credits=1

    with mock.patch.object(OVHSMSGateway, 'send_msg') as send_function:

        send_function.return_value = 2

        connector.jobs()

        assert SMSLog.objects.filter(

            appname=connector.get_connector_slug(), slug=connector.slug, credits=2

        )

        with send_patch as send_function:

            send_function.return_value = None

            connector.jobs()

            assert send_function.call_args[1]['text'] == 'not a spam'

            assert send_function.call_args[1]['stop'] == ('nostop' not in path)

        'creditsLeft': 123,

    }

    ovh_url = connector.API_URL % {'serviceName': 'sms-test42'}

        connector.jobs()

    connector.refresh_from_db()

    assert connector.credit_left == 123

    resp = {

        'creditsLeft': 456,

    }

        connector.hourly()

    assert connector.credit_left == 456

    freezer.move_to('2019-01-01 00:00:00')

    resp = {'creditsLeft': 101}

    ovh_url = connector.API_URL % {'serviceName': 'sms-test42'}

        connector.hourly()

    assert len(mailoutbox) == 0

    resp = {'creditsLeft': 99}

    ovh_url = connector.API_URL % {'serviceName': 'sms-test42'}

        connector.hourly()

    assert len(mailoutbox) == 1

    freezer.move_to('2019-01-01 12:00:00')

    resp = {'creditsLeft': 99}

    ovh_url = connector.API_URL % {'serviceName': 'sms-test42'}

        connector.hourly()

    assert len(mailoutbox) == 0

    freezer.move_to('2019-01-02 01:00:07')

        connector.hourly()

    assert len(mailoutbox) == 1

    ovh_request_token_url = 'https://eu.api.ovh.com/1.0/auth/credential'

    ovh_response = {'message': 'Invalid application key'}

        resp = resp.click('request access').follow()

    assert 'error requesting token: Invalid application key.' in resp.text

    ovh_response = 'not-json'

        resp = resp.click('request access').follow()

    assert 'error requesting token: bad JSON response' in resp.text

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

    freezer.move_to('2021-01-01 12:00')

        SMSLog.objects.create(appname='ovh', slug='ovhsmsgateway')

    freezer.move_to('2021-02-03 13:00')

        SMSLog.objects.create(appname='ovh', slug='ovhsmsgateway')

    freezer.move_to('2021-02-06 13:00')

            connector_model = app.get_connector_model()

            if connector_model is None:

                continue

                if not hasattr(method, 'endpoint_info'):

                    continue

                if method.endpoint_info.post and method.endpoint_info.post.get('request_body', {}).get(

@mock.patch("django.db.models.fields.UUIDField.get_default", return_value=UUID)

def test_create_intervention_inconsistency_id_error(mocked_uuid4, app, freezer, smart):

    freezer.move_to('2021-07-08 00:00:00')

    wcs_request = smart.wcs_requests.get(uuid=UUID)

    assert wcs_request.status == 'registered'

    job = Job.objects.get(method_name='create_intervention_job')

@mock.patch("django.db.models.fields.UUIDField.get_default", return_value=UUID)

def test_create_intervention_content_error(mocked_uuid, app, freezer, smart):

    freezer.move_to('2021-07-08 00:00:00')

    wcs_request = smart.wcs_requests.get(uuid=UUID)

    assert wcs_request.status == 'registered'

    assert 'invalid json' in wcs_request.result

@mock.patch("django.db.models.fields.UUIDField.get_default", return_value=UUID)

def test_create_intervention_client_error(mocked_uuid, app, freezer, smart):

    freezer.move_to('2021-07-08 00:00:00')

    wcs_request = smart.wcs_requests.get(uuid=UUID)

    assert '400 Client Error' in wcs_request.result

    assert wcs_request.tries == 1

    assert CREATE_INTERVENTION_QUERY[

        'notification_url'

    ] == 'http://testserver/toulouse-smart/test/update-intervention?uuid=%s' % str(UUID)

    mocked_push = mock.patch(

        "passerelle.contrib.toulouse_smart.models.SmartRequest.push",

    wcs_request = smart.wcs_requests.get(uuid=UUID)

    wcs_request.status = 'failed'

    wcs_request.save()

    smart.jobs()

    job = Job.objects.get(method_name='add_media_job')

def setup_access_rights(obj):

    obj_type = ContentType.objects.get_for_model(obj)

    AccessRight.objects.create(codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=obj.pk)

    return obj