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

Python test_common.metadata_transmitted函数代码示例

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

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



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

示例1: testAbortedStreamStream

    def testAbortedStreamStream(self):
        test_cases = zip(_ABORT_CODES, _EXPECTED_CLIENT_CODES,
                         _EXPECTED_DETAILS)
        for abort_code, expected_code, expected_details in test_cases:
            self._servicer.set_code(abort_code)
            self._servicer.set_details(_DETAILS)
            self._servicer.set_abort_call()

            response_iterator_call = self._stream_stream(
                iter([object()] * test_constants.STREAM_LENGTH),
                metadata=_CLIENT_METADATA)
            received_initial_metadata = \
                response_iterator_call.initial_metadata()
            with self.assertRaises(grpc.RpcError):
                self.assertEqual(len(list(response_iterator_call)), 0)

            self.assertTrue(
                test_common.metadata_transmitted(
                    _CLIENT_METADATA,
                    self._servicer.received_client_metadata()))
            self.assertTrue(
                test_common.metadata_transmitted(_SERVER_INITIAL_METADATA,
                                                 received_initial_metadata))
            self.assertTrue(
                test_common.metadata_transmitted(
                    _SERVER_TRAILING_METADATA,
                    response_iterator_call.trailing_metadata()))
            self.assertIs(expected_code, response_iterator_call.code())
            self.assertEqual(expected_details, response_iterator_call.details())
开发者ID:CCNITSilchar,项目名称:grpc,代码行数:29,代码来源:_metadata_code_details_test.py


示例2: testAbortedStreamUnary

    def testAbortedStreamUnary(self):
        test_cases = zip(_ABORT_CODES, _EXPECTED_CLIENT_CODES,
                         _EXPECTED_DETAILS)
        for abort_code, expected_code, expected_details in test_cases:
            self._servicer.set_code(abort_code)
            self._servicer.set_details(_DETAILS)
            self._servicer.set_abort_call()

            with self.assertRaises(grpc.RpcError) as exception_context:
                self._stream_unary.with_call(
                    iter([_SERIALIZED_REQUEST] * test_constants.STREAM_LENGTH),
                    metadata=_CLIENT_METADATA)

            self.assertTrue(
                test_common.metadata_transmitted(
                    _CLIENT_METADATA,
                    self._servicer.received_client_metadata()))
            self.assertTrue(
                test_common.metadata_transmitted(
                    _SERVER_INITIAL_METADATA,
                    exception_context.exception.initial_metadata()))
            self.assertTrue(
                test_common.metadata_transmitted(
                    _SERVER_TRAILING_METADATA,
                    exception_context.exception.trailing_metadata()))
            self.assertIs(expected_code, exception_context.exception.code())
            self.assertEqual(expected_details,
                             exception_context.exception.details())
开发者ID:CCNITSilchar,项目名称:grpc,代码行数:28,代码来源:_metadata_code_details_test.py


示例3: testUnaryUnary

 def testUnaryUnary(self):
   multi_callable = self._channel.unary_unary(_UNARY_UNARY)
   unused_response, call = multi_callable.with_call(
       _REQUEST, metadata=_CLIENT_METADATA)
   self.assertTrue(test_common.metadata_transmitted(
       _SERVER_INITIAL_METADATA, call.initial_metadata()))
   self.assertTrue(test_common.metadata_transmitted(
       _SERVER_TRAILING_METADATA, call.trailing_metadata()))
开发者ID:ConfusedReality,项目名称:pkg_network_grpc,代码行数:8,代码来源:_metadata_test.py


示例4: testUnaryStream

 def testUnaryStream(self):
   multi_callable = self._channel.unary_stream(_UNARY_STREAM)
   call = multi_callable(_REQUEST, metadata=_CLIENT_METADATA)
   self.assertTrue(test_common.metadata_transmitted(
       _SERVER_INITIAL_METADATA, call.initial_metadata()))
   for _ in call:
     pass
   self.assertTrue(test_common.metadata_transmitted(
       _SERVER_TRAILING_METADATA, call.trailing_metadata()))
开发者ID:ConfusedReality,项目名称:pkg_network_grpc,代码行数:9,代码来源:_metadata_test.py


示例5: testStreamUnary

 def testStreamUnary(self):
   multi_callable = self._channel.stream_unary(_STREAM_UNARY)
   unused_response, call = multi_callable.with_call(
       [_REQUEST] * test_constants.STREAM_LENGTH,
       metadata=_CLIENT_METADATA)
   self.assertTrue(test_common.metadata_transmitted(
       _SERVER_INITIAL_METADATA, call.initial_metadata()))
   self.assertTrue(test_common.metadata_transmitted(
       _SERVER_TRAILING_METADATA, call.trailing_metadata()))
开发者ID:ConfusedReality,项目名称:pkg_network_grpc,代码行数:9,代码来源:_metadata_test.py


示例6: testStreamStream

 def testStreamStream(self):
   multi_callable = self._channel.stream_stream(_STREAM_STREAM)
   call = multi_callable([_REQUEST] * test_constants.STREAM_LENGTH,
                         metadata=_CLIENT_METADATA)
   self.assertTrue(test_common.metadata_transmitted(
       _SERVER_INITIAL_METADATA, call.initial_metadata()))
   for _ in call:
     pass
   self.assertTrue(test_common.metadata_transmitted(
       _SERVER_TRAILING_METADATA, call.trailing_metadata()))
开发者ID:ConfusedReality,项目名称:pkg_network_grpc,代码行数:10,代码来源:_metadata_test.py


示例7: validate_client_metadata

def validate_client_metadata(test, servicer_context):
  test.assertTrue(test_common.metadata_transmitted(
      _CLIENT_METADATA, servicer_context.invocation_metadata()))
  test.assertTrue(user_agent(servicer_context.invocation_metadata())
                  .startswith('primary-agent ' + _USER_AGENT))
  test.assertTrue(user_agent(servicer_context.invocation_metadata())
                  .endswith('secondary-agent'))
开发者ID:ConfusedReality,项目名称:pkg_network_grpc,代码行数:7,代码来源:_metadata_test.py


示例8: testSuccessfulUnaryUnary

    def testSuccessfulUnaryUnary(self):
        self._servicer.set_details(_DETAILS)

        unused_response, call = self._unary_unary.with_call(
            object(), metadata=_CLIENT_METADATA)

        self.assertTrue(
            test_common.metadata_transmitted(
                _CLIENT_METADATA, self._servicer.received_client_metadata()))
        self.assertTrue(
            test_common.metadata_transmitted(_SERVER_INITIAL_METADATA,
                                             call.initial_metadata()))
        self.assertTrue(
            test_common.metadata_transmitted(_SERVER_TRAILING_METADATA,
                                             call.trailing_metadata()))
        self.assertIs(grpc.StatusCode.OK, call.code())
        self.assertEqual(_DETAILS, call.details())
开发者ID:deepaklukose,项目名称:grpc,代码行数:17,代码来源:_metadata_code_details_test.py


示例9: testSuccessfulStreamUnary

    def testSuccessfulStreamUnary(self):
        self._servicer.set_details(_DETAILS)

        unused_response, call = self._stream_unary.with_call(
            iter([_SERIALIZED_REQUEST] * test_constants.STREAM_LENGTH),
            metadata=_CLIENT_METADATA)

        self.assertTrue(
            test_common.metadata_transmitted(
                _CLIENT_METADATA, self._servicer.received_client_metadata()))
        self.assertTrue(
            test_common.metadata_transmitted(_SERVER_INITIAL_METADATA,
                                             call.initial_metadata()))
        self.assertTrue(
            test_common.metadata_transmitted(_SERVER_TRAILING_METADATA,
                                             call.trailing_metadata()))
        self.assertIs(grpc.StatusCode.OK, call.code())
        self.assertEqual(_DETAILS, call.details())
开发者ID:deepaklukose,项目名称:grpc,代码行数:18,代码来源:_metadata_code_details_test.py


示例10: validate_client_metadata

def validate_client_metadata(test, servicer_context):
    invocation_metadata = servicer_context.invocation_metadata()
    test.assertTrue(
        test_common.metadata_transmitted(_EXPECTED_INVOCATION_METADATA,
                                         invocation_metadata))
    user_agent = _user_agent(invocation_metadata)
    test.assertTrue(
        user_agent.startswith('primary-agent ' + _channel._USER_AGENT))
    test.assertTrue(user_agent.endswith('secondary-agent'))
开发者ID:Falco20019,项目名称:grpc,代码行数:9,代码来源:_metadata_test.py


示例11: testSuccessfulUnaryStream

  def testSuccessfulUnaryStream(self):
    self._servicer.set_details(_DETAILS)

    call = self._unary_stream(_SERIALIZED_REQUEST, metadata=_CLIENT_METADATA)
    received_initial_metadata = call.initial_metadata()
    for _ in call:
      pass

    self.assertTrue(
        test_common.metadata_transmitted(
            _CLIENT_METADATA, self._servicer.received_client_metadata()))
    self.assertTrue(
        test_common.metadata_transmitted(
            _SERVER_INITIAL_METADATA, received_initial_metadata))
    self.assertTrue(
        test_common.metadata_transmitted(
            _SERVER_TRAILING_METADATA, call.trailing_metadata()))
    self.assertIs(grpc.StatusCode.OK, call.code())
    self.assertEqual(_DETAILS, call.details())
开发者ID:FabioBatSilva,项目名称:grpc,代码行数:19,代码来源:_metadata_code_details_test.py


示例12: testCustomCodeUnaryUnary

    def testCustomCodeUnaryUnary(self):
        self._servicer.set_code(_NON_OK_CODE)
        self._servicer.set_details(_DETAILS)

        with self.assertRaises(grpc.RpcError) as exception_context:
            self._unary_unary.with_call(object(), metadata=_CLIENT_METADATA)

        self.assertTrue(
            test_common.metadata_transmitted(
                _CLIENT_METADATA, self._servicer.received_client_metadata()))
        self.assertTrue(
            test_common.metadata_transmitted(
                _SERVER_INITIAL_METADATA,
                exception_context.exception.initial_metadata()))
        self.assertTrue(
            test_common.metadata_transmitted(
                _SERVER_TRAILING_METADATA,
                exception_context.exception.trailing_metadata()))
        self.assertIs(_NON_OK_CODE, exception_context.exception.code())
        self.assertEqual(_DETAILS, exception_context.exception.details())
开发者ID:deepaklukose,项目名称:grpc,代码行数:20,代码来源:_metadata_code_details_test.py


示例13: testCustomCodeUnaryStream

  def testCustomCodeUnaryStream(self):
    self._servicer.set_code(_NON_OK_CODE)
    self._servicer.set_details(_DETAILS)

    call = self._unary_stream(_SERIALIZED_REQUEST, metadata=_CLIENT_METADATA)
    received_initial_metadata = call.initial_metadata()
    with self.assertRaises(grpc.RpcError):
      for _ in call:
        pass

    self.assertTrue(
        test_common.metadata_transmitted(
            _CLIENT_METADATA, self._servicer.received_client_metadata()))
    self.assertTrue(
        test_common.metadata_transmitted(
            _SERVER_INITIAL_METADATA, received_initial_metadata))
    self.assertTrue(
        test_common.metadata_transmitted(
            _SERVER_TRAILING_METADATA, call.trailing_metadata()))
    self.assertIs(_NON_OK_CODE, call.code())
    self.assertEqual(_DETAILS, call.details())
开发者ID:FabioBatSilva,项目名称:grpc,代码行数:21,代码来源:_metadata_code_details_test.py


示例14: completion_transmitted

 def completion_transmitted(self, original_completion, transmitted_completion):
   if (original_completion.terminal_metadata is not None and
       not grpc_test_common.metadata_transmitted(
           original_completion.terminal_metadata,
           transmitted_completion.terminal_metadata)):
       return False
   elif original_completion.code is not transmitted_completion.code:
     return False
   elif original_completion.message != transmitted_completion.message:
     return False
   else:
     return True
开发者ID:201528013359030,项目名称:grpc,代码行数:12,代码来源:_core_over_links_base_interface_test.py


示例15: testSuccessfulStreamStream

    def testSuccessfulStreamStream(self):
        self._servicer.set_details(_DETAILS)

        response_iterator_call = self._stream_stream(
            iter([object()] * test_constants.STREAM_LENGTH),
            metadata=_CLIENT_METADATA)
        received_initial_metadata = response_iterator_call.initial_metadata()
        list(response_iterator_call)

        self.assertTrue(
            test_common.metadata_transmitted(
                _CLIENT_METADATA, self._servicer.received_client_metadata()))
        self.assertTrue(
            test_common.metadata_transmitted(_SERVER_INITIAL_METADATA,
                                             received_initial_metadata))
        self.assertTrue(
            test_common.metadata_transmitted(
                _SERVER_TRAILING_METADATA,
                response_iterator_call.trailing_metadata()))
        self.assertIs(grpc.StatusCode.OK, response_iterator_call.code())
        self.assertEqual(_DETAILS, response_iterator_call.details())
开发者ID:deepaklukose,项目名称:grpc,代码行数:21,代码来源:_metadata_code_details_test.py


示例16: testCustomCodeStreamUnary

    def testCustomCodeStreamUnary(self):
        self._servicer.set_code(_NON_OK_CODE)
        self._servicer.set_details(_DETAILS)

        with self.assertRaises(grpc.RpcError) as exception_context:
            self._stream_unary.with_call(
                iter([_SERIALIZED_REQUEST] * test_constants.STREAM_LENGTH),
                metadata=_CLIENT_METADATA)

        self.assertTrue(
            test_common.metadata_transmitted(
                _CLIENT_METADATA, self._servicer.received_client_metadata()))
        self.assertTrue(
            test_common.metadata_transmitted(
                _SERVER_INITIAL_METADATA,
                exception_context.exception.initial_metadata()))
        self.assertTrue(
            test_common.metadata_transmitted(
                _SERVER_TRAILING_METADATA,
                exception_context.exception.trailing_metadata()))
        self.assertIs(_NON_OK_CODE, exception_context.exception.code())
        self.assertEqual(_DETAILS, exception_context.exception.details())
开发者ID:deepaklukose,项目名称:grpc,代码行数:22,代码来源:_metadata_code_details_test.py


示例17: testAbortedUnaryStream

    def testAbortedUnaryStream(self):
        self._servicer.set_code(_NON_OK_CODE)
        self._servicer.set_details(_DETAILS)
        self._servicer.set_abort_call()

        response_iterator_call = self._unary_stream(
            _SERIALIZED_REQUEST, metadata=_CLIENT_METADATA)
        received_initial_metadata = response_iterator_call.initial_metadata()
        with self.assertRaises(grpc.RpcError):
            self.assertEqual(len(list(response_iterator_call)), 0)

        self.assertTrue(
            test_common.metadata_transmitted(
                _CLIENT_METADATA, self._servicer.received_client_metadata()))
        self.assertTrue(
            test_common.metadata_transmitted(_SERVER_INITIAL_METADATA,
                                             received_initial_metadata))
        self.assertTrue(
            test_common.metadata_transmitted(
                _SERVER_TRAILING_METADATA,
                response_iterator_call.trailing_metadata()))
        self.assertIs(_NON_OK_CODE, response_iterator_call.code())
        self.assertEqual(_DETAILS, response_iterator_call.details())
开发者ID:deepaklukose,项目名称:grpc,代码行数:23,代码来源:_metadata_code_details_test.py


示例18: testCustomCodeStreamStream

    def testCustomCodeStreamStream(self):
        self._servicer.set_code(_NON_OK_CODE)
        self._servicer.set_details(_DETAILS)

        response_iterator_call = self._stream_stream(
            iter([object()] * test_constants.STREAM_LENGTH),
            metadata=_CLIENT_METADATA)
        received_initial_metadata = response_iterator_call.initial_metadata()
        with self.assertRaises(grpc.RpcError) as exception_context:
            list(response_iterator_call)

        self.assertTrue(
            test_common.metadata_transmitted(
                _CLIENT_METADATA, self._servicer.received_client_metadata()))
        self.assertTrue(
            test_common.metadata_transmitted(_SERVER_INITIAL_METADATA,
                                             received_initial_metadata))
        self.assertTrue(
            test_common.metadata_transmitted(
                _SERVER_TRAILING_METADATA,
                exception_context.exception.trailing_metadata()))
        self.assertIs(_NON_OK_CODE, exception_context.exception.code())
        self.assertEqual(_DETAILS, exception_context.exception.details())
开发者ID:deepaklukose,项目名称:grpc,代码行数:23,代码来源:_metadata_code_details_test.py


示例19: testCustomCodeExceptionStreamStream

  def testCustomCodeExceptionStreamStream(self):
    self._servicer.set_code(_NON_OK_CODE)
    self._servicer.set_details(_DETAILS)
    self._servicer.set_exception()

    call = self._stream_stream(
        iter([object()] * test_constants.STREAM_LENGTH),
        metadata=_CLIENT_METADATA)
    received_initial_metadata = call.initial_metadata()
    with self.assertRaises(grpc.RpcError):
      for _ in call:
        pass

    self.assertTrue(
        test_common.metadata_transmitted(
            _CLIENT_METADATA, self._servicer.received_client_metadata()))
    self.assertTrue(
        test_common.metadata_transmitted(
            _SERVER_INITIAL_METADATA, received_initial_metadata))
    self.assertTrue(
        test_common.metadata_transmitted(
            _SERVER_TRAILING_METADATA, call.trailing_metadata()))
    self.assertIs(_NON_OK_CODE, call.code())
    self.assertEqual(_DETAILS, call.details())
开发者ID:FabioBatSilva,项目名称:grpc,代码行数:24,代码来源:_metadata_code_details_test.py


示例20: test_echo

    def test_echo(self):
        DEADLINE = time.time() + 5
        DEADLINE_TOLERANCE = 0.25
        CLIENT_METADATA_ASCII_KEY = 'key'
        CLIENT_METADATA_ASCII_VALUE = 'val'
        CLIENT_METADATA_BIN_KEY = 'key-bin'
        CLIENT_METADATA_BIN_VALUE = b'\0' * 1000
        SERVER_INITIAL_METADATA_KEY = 'init_me_me_me'
        SERVER_INITIAL_METADATA_VALUE = 'whodawha?'
        SERVER_TRAILING_METADATA_KEY = 'california_is_in_a_drought'
        SERVER_TRAILING_METADATA_VALUE = 'zomg it is'
        SERVER_STATUS_CODE = cygrpc.StatusCode.ok
        SERVER_STATUS_DETAILS = 'our work is never over'
        REQUEST = b'in death a member of project mayhem has a name'
        RESPONSE = b'his name is robert paulson'
        METHOD = b'twinkies'

        server_request_tag = object()
        request_call_result = self.server.request_call(
            self.server_completion_queue, self.server_completion_queue,
            server_request_tag)

        self.assertEqual(cygrpc.CallError.ok, request_call_result)

        client_call_tag = object()
        client_initial_metadata = (
            (
                CLIENT_METADATA_ASCII_KEY,
                CLIENT_METADATA_ASCII_VALUE,
            ),
            (
                CLIENT_METADATA_BIN_KEY,
                CLIENT_METADATA_BIN_VALUE,
            ),
        )
        client_call = self.client_channel.integrated_call(
            0, METHOD, self.host_argument, DEADLINE, client_initial_metadata,
            None, [
                (
                    [
                        cygrpc.SendInitialMetadataOperation(
                            client_initial_metadata, _EMPTY_FLAGS),
                        cygrpc.SendMessageOperation(REQUEST, _EMPTY_FLAGS),
                        cygrpc.SendCloseFromClientOperation(_EMPTY_FLAGS),
                        cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS),
                        cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),
                        cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS),
                    ],
                    client_call_tag,
                ),
            ])
        client_event_future = test_utilities.SimpleFuture(
            self.client_channel.next_call_event)

        request_event = self.server_completion_queue.poll(deadline=DEADLINE)
        self.assertEqual(cygrpc.CompletionType.operation_complete,
                         request_event.completion_type)
        self.assertIsInstance(request_event.call, cygrpc.Call)
        self.assertIs(server_request_tag, request_event.tag)
        self.assertTrue(
            test_common.metadata_transmitted(client_initial_metadata,
                                             request_event.invocation_metadata))
        self.assertEqual(METHOD, request_event.call_details.method)
        self.assertEqual(self.expected_host, request_event.call_details.host)
        self.assertLess(
            abs(DEADLINE - request_event.call_details.deadline),
            DEADLINE_TOLERANCE)

        server_call_tag = object()
        server_call = request_event.call
        server_initial_metadata = ((
            SERVER_INITIAL_METADATA_KEY,
            SERVER_INITIAL_METADATA_VALUE,
        ),)
        server_trailing_metadata = ((
            SERVER_TRAILING_METADATA_KEY,
            SERVER_TRAILING_METADATA_VALUE,
        ),)
        server_start_batch_result = server_call.start_server_batch([
            cygrpc.SendInitialMetadataOperation(server_initial_metadata,
                                                _EMPTY_FLAGS),
            cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),
            cygrpc.SendMessageOperation(RESPONSE, _EMPTY_FLAGS),
            cygrpc.ReceiveCloseOnServerOperation(_EMPTY_FLAGS),
            cygrpc.SendStatusFromServerOperation(
                server_trailing_metadata, SERVER_STATUS_CODE,
                SERVER_STATUS_DETAILS, _EMPTY_FLAGS)
        ], server_call_tag)
        self.assertEqual(cygrpc.CallError.ok, server_start_batch_result)

        server_event = self.server_completion_queue.poll(deadline=DEADLINE)
        client_event = client_event_future.result()

        self.assertEqual(6, len(client_event.batch_operations))
        found_client_op_types = set()
        for client_result in client_event.batch_operations:
            # we expect each op type to be unique
            self.assertNotIn(client_result.type(), found_client_op_types)
            found_client_op_types.add(client_result.type())
            if client_result.type(
#.........这里部分代码省略.........
开发者ID:CCNITSilchar,项目名称:grpc,代码行数:101,代码来源:cygrpc_test.py



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


鲜花

握手

雷人

路过

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

请发表评论

全部评论

专题导读
上一篇:
Python test_common.test_server函数代码示例发布时间:2022-05-27
下一篇:
Python test.get_test_context函数代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

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

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

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