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

Python resources.test_root_certificates函数代码示例

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

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



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

示例1: __init__

    def __init__(self, server, config, hist):
        # Create the stub
        if config.HasField('security_params'):
            creds = grpc.ssl_channel_credentials(
                resources.test_root_certificates())
            channel = test_common.test_secure_channel(
                server, creds, config.security_params.server_host_override)
        else:
            channel = grpc.insecure_channel(server)

        # waits for the channel to be ready before we start sending messages
        grpc.channel_ready_future(channel).result()

        if config.payload_config.WhichOneof('payload') == 'simple_params':
            self._generic = False
            self._stub = services_pb2.BenchmarkServiceStub(channel)
            payload = messages_pb2.Payload(
                body='\0' * config.payload_config.simple_params.req_size)
            self._request = messages_pb2.SimpleRequest(
                payload=payload,
                response_size=config.payload_config.simple_params.resp_size)
        else:
            self._generic = True
            self._stub = GenericStub(channel)
            self._request = '\0' * config.payload_config.bytebuf_params.req_size

        self._hist = hist
        self._response_callbacks = []
开发者ID:aaronjheng,项目名称:grpc,代码行数:28,代码来源:benchmark_client.py


示例2: __init__

  def __init__(self, server, config, hist):
    # Create the stub
    if config.HasField('security_params'):
      creds = grpc.ssl_channel_credentials(resources.test_root_certificates())
      channel = test_common.test_secure_channel(
        server, creds, config.security_params.server_host_override)
    else:
      channel = grpc.insecure_channel(server)

    connected_event = threading.Event()
    def wait_for_ready(connectivity):
      if connectivity == grpc.ChannelConnectivity.READY:
        connected_event.set()
    channel.subscribe(wait_for_ready, try_to_connect=True)
    connected_event.wait()

    if config.payload_config.WhichOneof('payload') == 'simple_params':
      self._generic = False
      self._stub = services_pb2.BenchmarkServiceStub(channel)
      payload = messages_pb2.Payload(
          body='\0' * config.payload_config.simple_params.req_size)
      self._request = messages_pb2.SimpleRequest(
          payload=payload,
          response_size=config.payload_config.simple_params.resp_size)
    else:
      self._generic = True
      self._stub = GenericStub(channel)
      self._request = '\0' * config.payload_config.bytebuf_params.req_size

    self._hist = hist
    self._response_callbacks = []
开发者ID:ConfusedReality,项目名称:pkg_network_grpc,代码行数:31,代码来源:benchmark_client.py


示例3: setUp

  def setUp(self):
    self._servicer = _Servicer()
    self._method_implementations = {
        (_GROUP, _UNARY_UNARY):
            utilities.unary_unary_inline(self._servicer.unary_unary),
        (_GROUP, _UNARY_STREAM):
            utilities.unary_stream_inline(self._servicer.unary_stream),
        (_GROUP, _STREAM_UNARY):
            utilities.stream_unary_inline(self._servicer.stream_unary),
        (_GROUP, _STREAM_STREAM):
            utilities.stream_stream_inline(self._servicer.stream_stream),
    }

    self._cardinalities = {
        _UNARY_UNARY: cardinality.Cardinality.UNARY_UNARY,
        _UNARY_STREAM: cardinality.Cardinality.UNARY_STREAM,
        _STREAM_UNARY: cardinality.Cardinality.STREAM_UNARY,
        _STREAM_STREAM: cardinality.Cardinality.STREAM_STREAM,
    }

    self._server_options = implementations.server_options(
        thread_pool_size=test_constants.POOL_SIZE)
    self._server_credentials = implementations.ssl_server_credentials(
        [(resources.private_key(), resources.certificate_chain(),),])
    self._client_credentials = implementations.ssl_client_credentials(
        resources.test_root_certificates(), None, None)
    self._stub_options = implementations.stub_options(
        thread_pool_size=test_constants.POOL_SIZE)
开发者ID:coreyglory,项目名称:grpc,代码行数:28,代码来源:_beta_features_test.py


示例4: __init__

  def __init__(self, server, config, hist):
    # Create the stub
    host, port = server.split(':')
    port = int(port)
    if config.HasField('security_params'):
      creds = implementations.ssl_channel_credentials(
          resources.test_root_certificates())
      channel = test_utilities.not_really_secure_channel(
          host, port, creds, config.security_params.server_host_override)
    else:
      channel = implementations.insecure_channel(host, port)

    if config.payload_config.WhichOneof('payload') == 'simple_params':
      self._generic = False
      self._stub = services_pb2.beta_create_BenchmarkService_stub(channel)
      payload = messages_pb2.Payload(
          body='\0' * config.payload_config.simple_params.req_size)
      self._request = messages_pb2.SimpleRequest(
          payload=payload,
          response_size=config.payload_config.simple_params.resp_size)
    else:
      self._generic = True
      self._stub = implementations.generic_stub(channel)
      self._request = '\0' * config.payload_config.bytebuf_params.req_size

    self._hist = hist
    self._response_callbacks = []
开发者ID:ash32,项目名称:grpc,代码行数:27,代码来源:benchmark_client.py


示例5: instantiate

  def instantiate(
      self, methods, method_implementations, multi_method_implementation):
    serialization_behaviors = _serialization_behaviors_from_test_methods(
        methods)
    # TODO(nathaniel): Add a "groups" attribute to _digest.TestServiceDigest.
    service = next(iter(methods))[0]
    # TODO(nathaniel): Add a "cardinalities_by_group" attribute to
    # _digest.TestServiceDigest.
    cardinalities = {
        method: method_object.cardinality()
        for (group, method), method_object in six.iteritems(methods)}

    server_options = implementations.server_options(
        request_deserializers=serialization_behaviors.request_deserializers,
        response_serializers=serialization_behaviors.response_serializers,
        thread_pool_size=test_constants.POOL_SIZE)
    server = implementations.server(
        method_implementations, options=server_options)
    server_credentials = implementations.ssl_server_credentials(
        [(resources.private_key(), resources.certificate_chain(),),])
    port = server.add_secure_port('[::]:0', server_credentials)
    server.start()
    channel_credentials = implementations.ssl_channel_credentials(
        resources.test_root_certificates())
    channel = test_utilities.not_really_secure_channel(
        'localhost', port, channel_credentials, _SERVER_HOST_OVERRIDE)
    stub_options = implementations.stub_options(
        request_serializers=serialization_behaviors.request_serializers,
        response_deserializers=serialization_behaviors.response_deserializers,
        thread_pool_size=test_constants.POOL_SIZE)
    generic_stub = implementations.generic_stub(channel, options=stub_options)
    dynamic_stub = implementations.dynamic_stub(
        channel, service, cardinalities, options=stub_options)
    return generic_stub, {service: dynamic_stub}, server
开发者ID:201528013359030,项目名称:grpc,代码行数:34,代码来源:_face_interface_test.py


示例6: setUp

 def setUp(self):
   server_credentials = cygrpc.server_credentials_ssl(
       None, [cygrpc.SslPemKeyCertPair(resources.private_key(),
                                       resources.certificate_chain())], False)
   client_credentials = cygrpc.channel_credentials_ssl(
       resources.test_root_certificates(), None)
   self.setUpMixin(server_credentials, client_credentials, _SSL_HOST_OVERRIDE)
开发者ID:DrSnowbird,项目名称:grpc,代码行数:7,代码来源:cygrpc_test.py


示例7: setUp

 def setUp(self):
   server_credentials = cygrpc.server_credentials_ssl(
       None, [cygrpc.SslPemKeyCertPair(resources.private_key(),
                                       resources.certificate_chain())], False)
   channel_credentials = cygrpc.channel_credentials_ssl(
       resources.test_root_certificates(), None)
   self.server_completion_queue = cygrpc.CompletionQueue()
   self.server = cygrpc.Server()
   self.server.register_completion_queue(self.server_completion_queue)
   self.port = self.server.add_http2_port('[::]:0', server_credentials)
   self.server.start()
   self.client_completion_queue = cygrpc.CompletionQueue()
   client_channel_arguments = cygrpc.ChannelArgs([
       cygrpc.ChannelArg(cygrpc.ChannelArgKey.ssl_target_name_override,
                         _SSL_HOST_OVERRIDE)])
   self.client_channel = cygrpc.Channel(
       'localhost:{}'.format(self.port), client_channel_arguments,
       channel_credentials)
开发者ID:Aj0Ay,项目名称:grpc,代码行数:18,代码来源:cygrpc_test.py


示例8: setUp

    def setUp(self):
        self._servicer = _Servicer()
        method_implementations = {
            (_GROUP, _UNARY_UNARY):
            utilities.unary_unary_inline(self._servicer.unary_unary),
            (_GROUP, _UNARY_STREAM):
            utilities.unary_stream_inline(self._servicer.unary_stream),
            (_GROUP, _STREAM_UNARY):
            utilities.stream_unary_inline(self._servicer.stream_unary),
            (_GROUP, _STREAM_STREAM):
            utilities.stream_stream_inline(self._servicer.stream_stream),
        }

        cardinalities = {
            _UNARY_UNARY: cardinality.Cardinality.UNARY_UNARY,
            _UNARY_STREAM: cardinality.Cardinality.UNARY_STREAM,
            _STREAM_UNARY: cardinality.Cardinality.STREAM_UNARY,
            _STREAM_STREAM: cardinality.Cardinality.STREAM_STREAM,
        }

        server_options = implementations.server_options(
            thread_pool_size=test_constants.POOL_SIZE)
        self._server = implementations.server(
            method_implementations, options=server_options)
        server_credentials = implementations.ssl_server_credentials([
            (
                resources.private_key(),
                resources.certificate_chain(),
            ),
        ])
        port = self._server.add_secure_port('[::]:0', server_credentials)
        self._server.start()
        self._channel_credentials = implementations.ssl_channel_credentials(
            resources.test_root_certificates())
        self._call_credentials = implementations.metadata_call_credentials(
            _metadata_plugin)
        channel = test_utilities.not_really_secure_channel(
            'localhost', port, self._channel_credentials, _SERVER_HOST_OVERRIDE)
        stub_options = implementations.stub_options(
            thread_pool_size=test_constants.POOL_SIZE)
        self._dynamic_stub = implementations.dynamic_stub(
            channel, _GROUP, cardinalities, options=stub_options)
开发者ID:CCNITSilchar,项目名称:grpc,代码行数:42,代码来源:_beta_features_test.py


示例9: handle_unary_unary

_UNARY_UNARY = '/test/UnaryUnary'

_SERVER_HOST_OVERRIDE = 'foo.test.google.fr'
_CLIENT_IDS = (
    b'*.test.google.fr',
    b'waterzooi.test.google.be',
    b'*.test.youtube.com',
    b'192.168.1.3',
)
_ID = 'id'
_ID_KEY = 'id_key'
_AUTH_CTX = 'auth_ctx'

_PRIVATE_KEY = resources.private_key()
_CERTIFICATE_CHAIN = resources.certificate_chain()
_TEST_ROOT_CERTIFICATES = resources.test_root_certificates()
_SERVER_CERTS = ((_PRIVATE_KEY, _CERTIFICATE_CHAIN),)
_PROPERTY_OPTIONS = ((
    'grpc.ssl_target_name_override',
    _SERVER_HOST_OVERRIDE,
),)


def handle_unary_unary(request, servicer_context):
    return pickle.dumps({
        _ID: servicer_context.peer_identities(),
        _ID_KEY: servicer_context.peer_identity_key(),
        _AUTH_CTX: servicer_context.auth_context()
    })

开发者ID:Falco20019,项目名称:grpc,代码行数:29,代码来源:_auth_context_test.py


示例10: test_application_provided_root_certificates

 def test_application_provided_root_certificates(self):
     channel_credentials = implementations.ssl_channel_credentials(
         resources.test_root_certificates())
     self.assertIsInstance(channel_credentials,
                           implementations.ChannelCredentials)
开发者ID:CCNITSilchar,项目名称:grpc,代码行数:5,代码来源:_implementations_test.py



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


鲜花

握手

雷人

路过

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

请发表评论

全部评论

专题导读
上一篇:
Python test.get_test_context函数代码示例发布时间:2022-05-27
下一篇:
Python fakes.FakeResponse类代码示例发布时间: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