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

Python util.spinWhile函数代码示例

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

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



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

示例1: testBrokenTimer

 def testBrokenTimer(self):
     t = internet.TimerService(1, lambda: 1 / 0)
     t.startService()
     util.spinWhile(lambda :t._loop.running, timeout=30)
     t.stopService()
     self.assertEquals([ZeroDivisionError],
                       [o.value.__class__ for o in log.flushErrors(ZeroDivisionError)])
开发者ID:pwarren,项目名称:AGDeviceControl,代码行数:7,代码来源:test_application.py


示例2: testTCP

    def testTCP(self):
        s = service.MultiService()
        s.startService()
        factory = protocol.ServerFactory()
        factory.protocol = TestEcho
        TestEcho.d = defer.Deferred()
        t = internet.TCPServer(0, factory)
        t.setServiceParent(s)
        num = t._port.getHost().port
        class Foo(basic.LineReceiver):
            def connectionMade(self):
                self.transport.write('lalala\r\n')
            def lineReceived(self, line):
                self.factory.line = line
                self.transport.loseConnection()
        factory = protocol.ClientFactory()
        factory.protocol = Foo
        factory.line = None
        internet.TCPClient('127.0.0.1', num, factory).setServiceParent(s)
        util.spinWhile(lambda :factory.line is None)
        self.assertEqual(factory.line, 'lalala')

        # Cleanup the reactor
        util.wait(defer.maybeDeferred(s.stopService))
        util.wait(TestEcho.d)
开发者ID:pwarren,项目名称:AGDeviceControl,代码行数:25,代码来源:test_application.py


示例3: testClientBind

    def testClientBind(self):
        f = MyServerFactory()
        p = reactor.listenTCP(0, f, interface="127.0.0.1")
        self.ports.append(p)
        
        factory = MyClientFactory()
        reactor.connectTCP("127.0.0.1", p.getHost().port, factory,
                           bindAddress=("127.0.0.1", 0))

        spinUntil(lambda :factory.protocol is not None)
        
        self.assertEquals(factory.protocol.made, 1)

        port = factory.protocol.transport.getHost().port
        f2 = MyClientFactory()
        reactor.connectTCP("127.0.0.1", p.getHost().port, f2,
                           bindAddress=("127.0.0.1", port))


        spinUntil(lambda :f2.failed)
        
        self.assertEquals(f2.failed, 1)
        f2.reason.trap(error.ConnectBindError)
        self.assert_(f2.reason.check(error.ConnectBindError))
        self.assertEquals(f2.stopped, 1)
        
        p.stopListening()
        factory.protocol.transport.loseConnection()

        spinWhile(lambda :p.connected)

        self.assertEquals(factory.stopped, 1)
        self.cleanPorts(*self.ports)
开发者ID:pwarren,项目名称:AGDeviceControl,代码行数:33,代码来源:test_tcp.py


示例4: waitOnShutDown

 def waitOnShutDown(self):
    if not self.done:
         try:
             self.transport.signalProcess('KILL')
             util.spinWhile(lambda: not self.done)
         except error.ProcessExitedAlready:
             pass
开发者ID:pwarren,项目名称:AGDeviceControl,代码行数:7,代码来源:subprocessprotocol.py


示例5: testSignal

 def testSignal(self):
     exe = sys.executable
     scriptPath = util.sibpath(__file__, "process_signal.py")
     signals = ("HUP", "INT", "KILL")
     for sig in signals:
         p = SignalProtocol(sig, self)
         reactor.spawnProcess(p, exe, [exe, "-u", scriptPath, sig], env=None, usePTY=self.usePTY)
         spinWhile(lambda: p.going)
         self.failIf(p.failure, p.failure)
开发者ID:pwarren,项目名称:AGDeviceControl,代码行数:9,代码来源:test_process.py


示例6: testUNIX

    def testUNIX(self):
        # FIXME: This test is far too dense.  It needs comments.
        #  -- spiv, 2004-11-07
        if not interfaces.IReactorUNIX(reactor, None):
            raise unittest.SkipTest, "This reactor does not support UNIX domain sockets"
        s = service.MultiService()
        s.startService()
        factory = protocol.ServerFactory()
        factory.protocol = TestEcho
        TestEcho.d = defer.Deferred()
        t = internet.UNIXServer('echo.skt', factory)
        t.setServiceParent(s)
        class Foo(basic.LineReceiver):
            def connectionMade(self):
                self.transport.write('lalala\r\n')
            def lineReceived(self, line):
                self.factory.line = line
                self.transport.loseConnection()
        factory = protocol.ClientFactory()
        factory.protocol = Foo
        factory.line = None
        internet.UNIXClient('echo.skt', factory).setServiceParent(s)
        util.spinWhile(lambda :factory.line is None)
        self.assertEqual(factory.line, 'lalala')
        util.wait(defer.maybeDeferred(s.stopService))
        util.wait(TestEcho.d)

        TestEcho.d = defer.Deferred()
        factory.line = None
        s.startService()
        util.spinWhile(lambda :factory.line is None)
        self.assertEqual(factory.line, 'lalala')

        # Cleanup the reactor
        util.wait(defer.maybeDeferred(s.stopService))
        util.wait(TestEcho.d)
开发者ID:pwarren,项目名称:AGDeviceControl,代码行数:36,代码来源:test_application.py


示例7: cleanPorts

 def cleanPorts(self, *ports):
     for p in ports:
         if p.connected:
             p.loseConnection()
             util.spinWhile(lambda :p.connected)
开发者ID:pwarren,项目名称:AGDeviceControl,代码行数:5,代码来源:test_tcp.py


示例8: wait

 def wait(self, sec_delay):
     self.wake_up = time.time()+sec_delay
     util.spinWhile(self._waiting, sec_delay+1)
开发者ID:BackupTheBerlios,项目名称:solipsis-svn,代码行数:3,代码来源:unittest_network.py


示例9: wait

 def wait(self, sec_delay):
     wake_up = time.time() + sec_delay
     def _waiting():
         return time.time() < wake_up
     util.spinWhile(_waiting, sec_delay+1)
开发者ID:BackupTheBerlios,项目名称:solipsis-svn,代码行数:5,代码来源:unittest_protocols.py



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


鲜花

握手

雷人

路过

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

请发表评论

全部评论

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