• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    公众号

Python utils.http_request_full函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中vumi.utils.http_request_full函数的典型用法代码示例。如果您正苦于以下问题:Python http_request_full函数的具体用法?Python http_request_full怎么用?Python http_request_full使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。



在下文中一共展示了http_request_full函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: test_http_basic_auth

    def test_http_basic_auth(self):
        http_auth_config = self.config.copy()
        http_auth_config.update({
            'identities': {
                'username': 'password',
            }
        })
        transport = yield self.tx_helper.get_transport(http_auth_config)
        url = '%s%s?%s' % (
            transport.get_transport_url(),
            self.config['web_path'],
            urlencode({
                'to_msisdn': '123',
                'from_msisdn': '456',
                'message': 'hello',
            }))

        response = yield http_request_full(url, '', method='GET')
        self.assertEqual(response.code, http.UNAUTHORIZED)
        self.assertEqual([], self.tx_helper.get_dispatched_inbound())

        response = yield http_request_full(url, '', headers={
            'Authorization': ['Basic %s' % b64encode('username:password')]
        }, method='GET')
        self.assertEqual(response.code, http.OK)
        [msg] = self.tx_helper.get_dispatched_inbound()
        self.assertEqual(msg['content'], 'hello')
        self.assertEqual(msg['transport_metadata'], {
            'http_user': 'username',
        })
开发者ID:AndrewCvekl,项目名称:vumi,代码行数:30,代码来源:test_oldapi.py


示例2: test_health_response

    def test_health_response(self):
        health_url = 'http://%s:%s%s' % (
            self.addr.host, self.addr.port, self.config['health_path'])

        response = yield http_request_full(health_url, method='GET')
        self.assertEqual(response.delivered_body, '0')

        yield self.app_helper.make_dispatch_inbound(
            'in 1', message_id='1', conv=self.conversation)

        queue = DeferredQueue()
        stream_url = '%s/%s/messages.json' % (self.url, self.conversation.key)
        stream_receiver = self.client.stream(
            TransportUserMessage, queue.put, queue.put, stream_url,
            Headers(self.auth_headers))

        yield queue.get()

        response = yield http_request_full(health_url, method='GET')
        self.assertEqual(response.delivered_body, '1')

        stream_receiver.disconnect()

        response = yield http_request_full(health_url, method='GET')
        self.assertEqual(response.delivered_body, '0')

        self.assertEqual(self.app.client_manager.clients, {
            'sphex.stream.message.%s' % (self.conversation.key,): []
        })
开发者ID:ChrisNolan1992,项目名称:vumi-go,代码行数:29,代码来源:test_vumi_app.py


示例3: test_http_request_full_headers

    def test_http_request_full_headers(self):
        def check_ua(request):
            self.assertEqual('blah', request.getHeader('user-agent'))
            return "Yay"
        self.set_render(check_ua)

        request = yield http_request_full(self.url, '',
                                          {'User-Agent': ['blah']})
        self.assertEqual(request.delivered_body, "Yay")
        self.assertEqual(request.code, http.OK)

        request = yield http_request_full(self.url, '', {'User-Agent': 'blah'})
        self.assertEqual(request.delivered_body, "Yay")
        self.assertEqual(request.code, http.OK)
开发者ID:Nagato23,项目名称:vumi,代码行数:14,代码来源:test_utils.py


示例4: test_missing_parameters

 def test_missing_parameters(self):
     url = self.mkurl_raw(ServiceNumber='12345', SMSBODY='hello',
         USN='user', PWD='pass', Operator='foo')
     response = yield http_request_full(url, '', method='GET')
     self.assertEqual(400, response.code)
     self.assertEqual(json.loads(response.delivered_body),
                      {'missing_parameter': ['PhoneNumber']})
开发者ID:TouK,项目名称:vumi,代码行数:7,代码来源:test_mediaedgegsm.py


示例5: test_health_response

    def test_health_response(self):
        yield self.start_app_worker()
        health_url = 'http://%s:%s%s' % (
            self.addr.host, self.addr.port, self.config['health_path'])

        response = yield http_request_full(health_url, method='GET')
        self.assertEqual(response.delivered_body, 'OK')
开发者ID:praekelt,项目名称:vumi-http-api,代码行数:7,代码来源:test_vumi_app.py


示例6: test_send_to_within_content_length_limit

    def test_send_to_within_content_length_limit(self):
        yield self.start_app_worker({
            'content_length_limit': 182,
        })

        msg = {
            'content': 'foo',
            'to_addr': '+1234',
        }

        url = '%s/%s/messages.json' % (self.url, self.conversation)
        response = yield http_request_full(url, json.dumps(msg),
                                           self.auth_headers, method='PUT')
        self.assertEqual(
            response.headers.getRawHeaders('content-type'),
            ['application/json; charset=utf-8'])
        put_msg = json.loads(response.delivered_body)
        self.assertEqual(response.code, http.OK)

        [sent_msg] = self.app_helper.get_dispatched_outbound()
        self.assertEqual(sent_msg['to_addr'], put_msg['to_addr'])
        self.assertEqual(sent_msg['message_id'], put_msg['message_id'])
        self.assertEqual(sent_msg['session_event'], None)
        self.assertEqual(sent_msg['to_addr'], '+1234')
        self.assertEqual(sent_msg['from_addr'], None)
开发者ID:praekelt,项目名称:vumi-http-api,代码行数:25,代码来源:test_vumi_app.py


示例7: test_send_to

    def test_send_to(self):
        yield self.start_app_worker()
        msg = {
            'to_addr': '+2345',
            'content': 'foo',
            'helper_metadata': {'voice': {'foo': 'bar'}},
            'session_event': 'new',
            'message_id': 'evil_id',
        }

        url = '%s/%s/messages.json' % (self.url, self.conversation)
        response = yield http_request_full(url, json.dumps(msg),
                                           self.auth_headers, method='PUT')

        self.assertEqual(response.code, http.OK)
        self.assertEqual(
            response.headers.getRawHeaders('content-type'),
            ['application/json; charset=utf-8'])
        put_msg = json.loads(response.delivered_body)

        [sent_msg] = self.app_helper.get_dispatched_outbound()
        self.assertEqual(sent_msg['to_addr'], sent_msg['to_addr'])
        # We do not respect the message_id that's been given.
        self.assertNotEqual(sent_msg['message_id'], msg['message_id'])
        self.assertEqual(sent_msg['message_id'], put_msg['message_id'])
        self.assertEqual(sent_msg['to_addr'], msg['to_addr'])
        self.assertEqual(sent_msg['from_addr'], None)
        self.assertEqual(sent_msg['session_event'], 'new')
        self.assertEqual(
            sent_msg['helper_metadata'], {'voice': {'foo': 'bar'}})
开发者ID:praekelt,项目名称:vumi-http-api,代码行数:30,代码来源:test_vumi_app.py


示例8: handle_outbound_message

    def handle_outbound_message(self, message):
        headers = {
            'Content-Type': 'application/json; charset=utf-8',
        }
        headers.update(self.get_auth_headers())

        params = {
            'to_addr': message['to_addr'],
            'content': message['content'],
            'message_id': message['message_id'],
            'in_reply_to': message['in_reply_to'],
            'session_event': message['session_event']
        }
        if 'helper_metadata' in message:
            params['helper_metadata'] = message['helper_metadata']

        resp = yield http_request_full(
            self.get_url('messages.json'),
            data=json.dumps(params).encode('utf-8'),
            headers=headers,
            method='PUT')

        if resp.code != http.OK:
            log.warning('Unexpected status code: %s, body: %s' % (
                resp.code, resp.delivered_body))
            yield self.publish_nack(message['message_id'],
                                    reason='Unexpected status code: %s' % (
                                        resp.code,))
            return

        remote_message = json.loads(resp.delivered_body)
        yield self.map_message_id(
            remote_message['message_id'], message['message_id'])
        yield self.publish_ack(user_message_id=message['message_id'],
                               sent_message_id=remote_message['message_id'])
开发者ID:Nagato23,项目名称:vumi,代码行数:35,代码来源:vumi_bridge.py


示例9: test_unicode_rendering

 def test_unicode_rendering(self):
     resp_d = http_request_full(self.url, headers=self.sample_req_headers)
     [msg] = yield self.tx_helper.wait_for_dispatched_inbound(1)
     self.tx_helper.make_dispatch_reply(msg, self.sample_unicode_menu_resp)
     resp = yield resp_d
     self.assertTrue("Hell\xc3\xb8" in resp.delivered_body)
     self.assertTrue("\xc3\xb8pti\xc3\xb8n 1" in resp.delivered_body)
开发者ID:Nagato23,项目名称:vumi,代码行数:7,代码来源:test_mxit.py


示例10: test_cleanup_as_seen_in_production

 def test_cleanup_as_seen_in_production(self):
     """what's a technical spec between friends?"""
     yield self.session_manager.create_session('13697502734175597',
         to_addr='*167*7#', from_addr='254XXXXXXXXX')
     query_string = ("msisdn=254XXXXXXXXX&clean=cleann&error=523"
                     "&SessionID=13697502734175597&MSC=254XXXXXXXXX"
                     "&=&=en&=9031510005344&=&=&=postpaid"
                     "&=20130528171235405&=200220130528171113956582"
                     "&userid=%s&password=%s" % (self.airtel_username,
                                                 self.airtel_password))
     response = yield http_request_full(
         '%s?%s' % (self.transport_url, query_string),
         data='', method='GET')
     self.assertEqual(response.code, http.OK)
     self.assertEqual(response.delivered_body, '')
     [msg] = yield self.tx_helper.wait_for_dispatched_inbound(1)
     self.assertEqual(msg['session_event'],
                      TransportUserMessage.SESSION_CLOSE)
     self.assertEqual(msg['to_addr'], '*167*7#')
     self.assertEqual(msg['from_addr'], '254XXXXXXXXX')
     self.assertEqual(msg['transport_metadata'], {
         'airtel': {
             'clean': 'cleann',
             'error': '523',
         }
     })
开发者ID:AndrewCvekl,项目名称:vumi,代码行数:26,代码来源:test_airtel.py


示例11: request_new_access_token

    def request_new_access_token(self):
        config = self.get_static_config()
        response = yield http_request_full(
            self.make_url(
                "token",
                {"grant_type": "client_credential", "appid": config.wechat_appid, "secret": config.wechat_secret},
            ),
            method="GET",
        )
        if not http_ok(response):
            raise WeChatApiException(
                ("Received HTTP status code %r when " "requesting access token.") % (response.code,)
            )

        data = json.loads(response.delivered_body)
        if "errcode" in data:
            raise WeChatApiException(
                "Error when requesting access token. " "Errcode: %(errcode)s, Errmsg: %(errmsg)s." % data
            )

        # make sure we're always ahead of the WeChat expiry
        access_token = data["access_token"]
        expiry = int(data["expires_in"]) * 0.90
        yield self.redis.setex(self.ACCESS_TOKEN_KEY, int(expiry), access_token)
        returnValue(access_token)
开发者ID:Nagato23,项目名称:vumi,代码行数:25,代码来源:wechat.py


示例12: push

    def push(self, url, vumi_message):
        config = self.get_static_config()
        data = vumi_message.to_json().encode('utf-8')
        try:
            auth, url = extract_auth_from_url(url.encode('utf-8'))
            headers = {
                'Content-Type': 'application/json; charset=utf-8',
            }
            if auth is not None:
                username, password = auth

                if username is None:
                    username = ''

                if password is None:
                    password = ''

                headers.update({
                    'Authorization': 'Basic %s' % (
                        base64.b64encode('%s:%s' % (username, password)),)
                })
            resp = yield http_request_full(
                url, data=data, headers=headers, timeout=config.timeout)
            if not (200 <= resp.code < 300):
                # We didn't get a 2xx response.
                log.warning('Got unexpected response code %s from %s' % (
                    resp.code, url))
        except SchemeNotSupported:
            log.warning('Unsupported scheme for URL: %s' % (url,))
        except HttpTimeoutError:
            log.warning("Timeout pushing message to %s" % (url,))
        except DNSLookupError:
            log.warning("DNS lookup error pushing message to %s" % (url,))
        except ConnectionRefusedError:
            log.warning("Connection refused pushing message to %s" % (url,))
开发者ID:ChrisNolan1992,项目名称:vumi-go,代码行数:35,代码来源:vumi_app.py


示例13: handle_outbound_message

    def handle_outbound_message(self, message):
        log.msg("Outbound message to be processed %s" % repr(message))
        params = {
            'ybsacctno': self.config['ybsacctno'],
            'password': self.config['password'],
            'origin': message['from_addr'],
            'sms_content': message['content'],
            'destinations': self.phone_format_to_yo(message['to_addr']),
        }
        log.msg('Hitting %s with %s' % (self.config['url'], urlencode(params)))
        try:
            response = yield http_request_full(
                "%s?%s" % (self.config['url'], urlencode(params)),
                "",
                {'User-Agent': ['Vumi Yo Transport'],
                 'Content-Type': ['application/json;charset=UTF-8'], },
                'GET')
        except ConnectionRefusedError:
            log.msg("Connection failed sending message: %s" % message)
            #raise TemporaryFailure('connection refused')
        except Exception as ex:
            log.msg("Unexpected error %s" % repr(ex))

        if response.code != 200:
            log.msg("Http Error %s: %s"
                    % (response.code, response.delivered_body))
            yield self.publish_delivery_report(
                user_message_id=message['message_id'],
                delivery_status='failed',
                failure_level='http',
                failure_code=response.code,
                failure_reason=response.delivered_body
                )
            return

        try:
            response_attr = parse_qs(unquote(response.delivered_body))
            [ybs_status] = response_attr['ybs_autocreate_status']
            ybs_msg = response_attr['ybs_autocreate_message'][0] if 'ybs_autocreate_message' in response_attr else None
            if (ybs_status == 'ERROR'):
                log.msg("Yo Error %s: %s" % (response.code,
                                             response.delivered_body))
                yield self.publish_delivery_report(
                    user_message_id=message['message_id'],
                    delivery_status='failed',
                    failure_level='service',
                    failure_code=ybs_status,
                    failure_reason=ybs_msg
                )
                return

            log.msg("Sms received and accepted by Yo %s" %
                    response.delivered_body)
            yield self.publish_delivery_report(
                user_message_id=message['message_id'],
                delivery_status='delivered',
                to_addr=message['to_addr']
            )
        except Exception as ex:
            log.msg("Unexpected error %s" % repr(ex))
开发者ID:hacklabcbba,项目名称:vusion-backend,代码行数:60,代码来源:yo_ug_http.py


示例14: handle_outbound_message

    def handle_outbound_message(self, message):
        log.msg("Outbound message to be processed %s" % repr(message))
        try:
            cmparser = CMXMLParser()
            params = cmparser.build({
                'customer_id': self.config['customer_id'],
                'login': self.config['login'],
                'password': self.config['password'],
                'from_addr': self.config['default_origin'],
                'to_addr': message['to_addr'],
                'content': message['content']})
            log.msg('Hitting %s with %r' % (self.config['outbound_url'], params))
            response = yield http_request_full(
                self.config['outbound_url'],
                params,
                {'User-Agent': ['Vumi CM YO Transport'],
                 'Content-Type': ['application/json;charset=UTF-8'], },
                method='POST')
            log.msg("Response: (%s) %r" % (response.code, response.delivered_body))
            content = response.delivered_body.strip()            

            if response.code != 200 or response.delivered_body:
                error = self.KNOWN_ERROR_RESPONSE_CODES.get(content,
                                                            'Unknown response code: %s' % (content,))
                yield self.publish_nack(message['message_id'], error)
                return    
            yield self.publish_ack(user_message_id=message['message_id'],
                                   sent_message_id=message['message_id'])
        except Exception as ex:
            log.msg("Unexpected error %s" % repr(ex))
开发者ID:hacklabcbba,项目名称:vumi-simagri,代码行数:30,代码来源:cm_nl_yo_ug_http.py


示例15: test_invalid_credentials

 def test_invalid_credentials(self):
     url = self.mkurl_raw(ServiceNumber='12345', SMSBODY='hello',
         USN='something', PWD='wrong', Operator='foo', PhoneNumber='1234')
     response = yield http_request_full(url, '', method='GET')
     self.assertEqual(400, response.code)
     self.assertEqual(json.loads(response.delivered_body),
                      {'credentials': 'invalid'})
开发者ID:TouK,项目名称:vumi,代码行数:7,代码来源:test_mediaedgegsm.py


示例16: test_bad_parameter_in_permissive_mode

 def test_bad_parameter_in_permissive_mode(self):
     url = self.mkurl('hello', foo='bar')
     response = yield http_request_full(url, '', method='GET')
     [msg] = self.tx_helper.get_dispatched_inbound()
     self.assertEqual(200, response.code)
     self.assertEqual(json.loads(response.delivered_body),
                      {'message_id': msg['message_id']})
开发者ID:AndrewCvekl,项目名称:vumi,代码行数:7,代码来源:test_cellulant_sms.py


示例17: handle_outbound_message

    def handle_outbound_message(self, message):
        creds = self._credentials.get(message['from_addr'], {})
        username = creds.get('username', '')
        password = creds.get('password', '')
        params = {
            'username': username,
            'password': password,
            'source': message['from_addr'],
            'destination': message['to_addr'],
            'message': message['content'],
            }
        log.msg("Sending outbound message: %s" % (message,))
        url = '%s?%s' % (self._outbound_url, urlencode(params))
        log.msg("Making HTTP request: %s" % (url,))
        response = yield http_request_full(url, '', method='GET')
        log.msg("Response: (%s) %r" % (response.code, response.delivered_body))
        content = response.delivered_body.strip()

        # we'll only send 1 message at a time and so the API can only
        # return this on a valid ack
        if content == '1':
            yield self.publish_ack(user_message_id=message['message_id'],
                                sent_message_id=message['message_id'])
        else:
            error = self.KNOWN_ERROR_RESPONSE_CODES.get(content,
                'Unknown response code: %s' % (content,))
            yield self.publish_nack(message['message_id'], error)
开发者ID:TouK,项目名称:vumi,代码行数:27,代码来源:cellulant_sms.py


示例18: test_http_request_full_timeout_after_first_receive

    def test_http_request_full_timeout_after_first_receive(self):
        # This tests the case where the client connects, receives
        # some data and creates its receiver but then times out
        # because the server takes too long to finish sending the data.

        def interrupt(r):
            raise self.InterruptHttp
        request_started = Deferred()
        self.set_render(interrupt, request_started)

        client_done = http_request_full(self.url, '', timeout=0.1)

        request = yield request_started
        request.write("some data")

        def check_server_response(reason):
            self.assertTrue(reason.check('twisted.internet.error'
                                         '.ConnectionDone'))

        request_done = request.notifyFinish()
        request_done.addBoth(check_server_response)
        yield request_done

        def check_client_response(reason):
            self.assertTrue(reason.check('vumi.utils.HttpTimeoutError'))
        client_done.addBoth(check_client_response)
        yield client_done
开发者ID:Nagato23,项目名称:vumi,代码行数:27,代码来源:test_utils.py


示例19: test_http_request_full_err

 def test_http_request_full_err(self):
     def err(r):
         raise ValueError("Bad")
     self.set_render(err)
     request = yield http_request_full(self.url, '')
     self.assertEqual(request.delivered_body, "Bad")
     self.assertEqual(request.code, http.INTERNAL_SERVER_ERROR)
开发者ID:Nagato23,项目名称:vumi,代码行数:7,代码来源:test_utils.py


示例20: test_http_request_full_ok_with_timeout_set

 def test_http_request_full_ok_with_timeout_set(self):
     # If we don't cancel the pending timeout check this test will fail with
     # a dirty reactor.
     self.set_render(lambda r: "Yay")
     request = yield http_request_full(self.url, '', timeout=100)
     self.assertEqual(request.delivered_body, "Yay")
     self.assertEqual(request.code, http.OK)
开发者ID:Nagato23,项目名称:vumi,代码行数:7,代码来源:test_utils.py



注:本文中的vumi.utils.http_request_full函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python vunit.VUnit类代码示例发布时间:2022-05-26
下一篇:
Python utils.http_request函数代码示例发布时间:2022-05-26
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap