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

Python endpoint.ProcessEndpointUnitMixin类代码示例

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

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



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

示例1: test__intercept_msg_out

    def test__intercept_msg_out(self, mocknpi, mockipi):
        mockipi.return_value = sentinel.inv2
        mocknpi.return_value = sentinel.inv2
        ep = ProcessEndpointUnitMixin(process=sentinel.proc, interceptors=sentinel_interceptors)

        ep._intercept_msg_out(sentinel.inv)

        mockipi.assert_has_calls([call(sentinel.proc_outgoing, sentinel.inv)])
        mocknpi.assert_has_calls([call(sentinel.msg_outgoing, sentinel.inv2)])
开发者ID:daf,项目名称:pyon,代码行数:9,代码来源:test_endpoint.py


示例2: test__build_header_no_context

    def test__build_header_no_context(self, mockgci):
        ep = ProcessEndpointUnitMixin(process=Mock())
        header = ep._build_header(sentinel.raw_msg, {})

        self.assertIn('sender-name', header)
        self.assertIn('sender', header)
        self.assertIn('sender-type', header)
        self.assertNotIn('sender-service', header)
        self.assertIn('origin-container-id', header)

        self.assertEquals(header['sender-name'], ep._process.name)
        self.assertEquals(header['sender'], ep._process.id)
        self.assertEquals(header['sender-type'], ep._process.process_type)
        self.assertEquals(header['origin-container-id'], mockgci().id)
开发者ID:daf,项目名称:pyon,代码行数:14,代码来源:test_endpoint.py


示例3: test__build_header_service

    def test__build_header_service(self, mockgci):
        procmock = Mock()
        procmock.process_type = 'service'

        ep = ProcessEndpointUnitMixin(process=procmock)
        ep.channel = Mock(spec=SendChannel)
        header = ep._build_header(sentinel.raw_msg, {})

        self.assertIn('sender-name', header)
        self.assertIn('sender', header)
        self.assertIn('sender-type', header)
        self.assertIn('sender-service', header)
        self.assertIn('origin-container-id', header)

        self.assertEquals(header['sender-name'], ep._process.name)
        self.assertEquals(header['sender'], ep._process.id)
        self.assertEquals(header['sender-type'], ep._process.process_type)
        self.assertEquals(header['sender-service'], "%s,%s" % (ep.channel._send_name.exchange, procmock.name))
        self.assertEquals(header['origin-container-id'], mockgci().id)
开发者ID:daf,项目名称:pyon,代码行数:19,代码来源:test_endpoint.py


示例4: _build_security_headers

    def _build_security_headers(self):
        """
        Builds additional security headers to be passed through to EMS.
        """
        # pull context from container
        ctx = self.container.context.get_context()

        if isinstance(ctx, dict):
            return ProcessEndpointUnitMixin.build_security_headers(ctx)

        return None
开发者ID:daf,项目名称:pyon,代码行数:11,代码来源:exchange.py


示例5: test__build_header_with_context

    def test__build_header_with_context(self, mockgci):
        procmock = Mock()
        procmock.get_context.return_value={'ion-actor-id': sentinel.ion_actor_id,
                                           'ion-actor-roles': sentinel.ion_actor_roles,
                                           'ion-actor-tokens': sentinel.ion_actor_tokens,
                                           'expiry': sentinel.expiry,
                                           'origin-container-id': sentinel.container_id}

        ep = ProcessEndpointUnitMixin(process=procmock)
        ep.channel = Mock(spec=SendChannel)
        header = ep._build_header(sentinel.raw_msg, {})

        self.assertIn('ion-actor-id', header)
        self.assertIn('ion-actor-roles', header)
        self.assertIn('ion-actor-tokens', header)
        self.assertIn('expiry', header)
        self.assertIn('origin-container-id', header)

        self.assertEquals(header['ion-actor-id'], sentinel.ion_actor_id)
        self.assertEquals(header['ion-actor-roles'], sentinel.ion_actor_roles)
        self.assertEquals(header['ion-actor-tokens'], sentinel.ion_actor_tokens)
        self.assertEquals(header['expiry'], sentinel.expiry)
        self.assertEquals(header['origin-container-id'], sentinel.container_id)
开发者ID:daf,项目名称:pyon,代码行数:23,代码来源:test_endpoint.py


示例6: test__get_sflow_manager

 def test__get_sflow_manager(self):
     ep = ProcessEndpointUnitMixin(process=Mock())
     self.assertEquals(ep._get_sflow_manager(), ep._process.container.sflow_manager)
开发者ID:daf,项目名称:pyon,代码行数:3,代码来源:test_endpoint.py


示例7: test__build_invocation

    def test__build_invocation(self, mockbi):
        ep = ProcessEndpointUnitMixin(process=sentinel.proc)
        ep._build_invocation(invother=sentinel.anything)

        mockbi.assert_called_once_with(ep, process=sentinel.proc, invother=sentinel.anything)
开发者ID:daf,项目名称:pyon,代码行数:5,代码来源:test_endpoint.py


示例8: test__get_sample_name

 def test__get_sample_name(self):
     ep = ProcessEndpointUnitMixin(process=Mock())
     self.assertEquals(ep._get_sample_name(), str(ep._process.id))
开发者ID:swarbhanu,项目名称:pyon,代码行数:3,代码来源:test_endpoint.py



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


鲜花

握手

雷人

路过

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

请发表评论

全部评论

专题导读
上一篇:
Python event.EventSubscriber类代码示例发布时间:2022-05-27
下一篇:
Python event.EventSubscriber类代码示例发布时间: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