def testStopTrying(self):
        f = Factory()
        f.protocol = In
        f.connections = 0
        f.allMessages = []

        c = ReconnectingClientFactory()
        c.initialDelay = c.delay = 0.2
        c.protocol = Out
        c.howManyTimes = 2

        port = reactor.listenTCP(0, f)
        PORT = port.getHost().port
        reactor.connectTCP('127.0.0.1', PORT, c)

        now = time.time()
        while len(f.allMessages) != 2 and (time.time() < now + 10):
            reactor.iterate(0.1)
        util.wait(defer.maybeDeferred(port.stopListening))
        
        self.assertEquals(len(f.allMessages), 2,
                          "not enough messages -- %s" % f.allMessages)
        self.assertEquals(f.connections, 2,
                          "Number of successful connections incorrect %d" %
                          f.connections)
        self.assertEquals(f.allMessages, [Out.msgs] * 2)
        self.failIf(c.continueTrying, "stopTrying never called or ineffective")

 def testScheduling(self):
     c = Counter()
     for i in range(100):
         self.schedule(c.add)
     for i in range(100):
         reactor.iterate()
     self.assertEquals(c.index, 100)

    def setUp(self):
        CFTPClientTestBase.setUp(self)

        self.startServer()
        cmds = ('-p %i -l testuser '
               '--known-hosts kh_test '
               '--user-authentications publickey '
               '--host-key-algorithms ssh-rsa '
               '-K direct '
               '-i dsa_test '
               '-a --nocache '
               '-v '
               '127.0.0.1')
        port = self.server.getHost().port
        cmds = test_conch._makeArgs((cmds % port).split(), mod='cftp')
        log.msg('running %s %s' % (sys.executable, cmds))
        self.processProtocol = SFTPTestProcess()

        env = os.environ.copy()
        env['PYTHONPATH'] = os.pathsep.join(sys.path)
        reactor.spawnProcess(self.processProtocol, sys.executable, cmds,
                             env=env)

        timeout = time.time() + 10
        while (not self.processProtocol.buffer) and (time.time() < timeout):
            reactor.iterate(0.1)
        if time.time() > timeout:
            self.skip = "couldn't start process"
        else:
            self.processProtocol.clearBuffer()

 def test_script_kill(self):
     self._skipCheck()
     try:
         result = yield self.db.script_kill()
     except redis.NoScriptRunning:
         pass
     except:
         raise self.failureException('%s raised instead of %s:\n %s'
                                     % (sys.exc_info()[0],
                                        'txredisapi.NoScriptRunning',
                                        failure.Failure().getTraceback()))
     else:
         raise self.failureException('%s not raised (%r returned)'
                                     % ('txredisapi.ResponseError', result))
     # Run an infinite loop script from one connection
     # and kill it from another.
     inf_loop = "while 1 do end"
     self.db1 = yield redis.Connection(REDIS_HOST, REDIS_PORT,
                                       reconnect=False)
     eval_deferred = self.db1.eval(inf_loop)
     reactor.iterate()
     r = yield self.db.script_kill()
     self.assertEqual(r, 'OK')
     try:
         result = yield eval_deferred
     except redis.ResponseError:
         pass
     except:
         raise self.failureException('%s raised instead of %s:\n %s'
                                     % (sys.exc_info()[0],
                                        'txredisapi.ResponseError',
                                        failure.Failure().getTraceback()))
     else:
         raise self.failureException('%s not raised (%r returned)'
                                     % ('txredisapi.ResponseError', result))

 def _getBatchOutput(self, f):
     fn = tempfile.mktemp()
     open(fn, 'w').write(f)
     l = []
     port = self.server.getHost().port
     cmds = ('-p %i -l testuser '
                 '--known-hosts kh_test '
                 '--user-authentications publickey '
                 '--host-key-algorithms ssh-rsa '
                 '-K direct '
                 '-i dsa_test '
                 '-a --nocache '
                 '-v -b %s 127.0.0.1') % (port, fn)
     cmds = test_conch._makeArgs(cmds.split(), mod='cftp')[1:]
     log.msg('running %s %s' % (sys.executable, cmds))
     env = os.environ.copy()
     env['PYTHONPATH'] = os.pathsep.join(sys.path)
     d = getProcessOutputAndValue(sys.executable, cmds, env=env)
     d.setTimeout(10)
     d.addBoth(l.append)
     while not l:
         if hasattr(self.server.factory, 'proto'):
             self.server.factory.proto.expectedLoseConnection = 1
         reactor.iterate(0.1)
     os.remove(fn)
     result = l[0]
     if isinstance(result, failure.Failure):
         raise result.value
     else:
         log.msg(result[1])
         return result[0]

 def test_bind(self):
     server = self.createServer([ pureldap.LDAPBindResponse(resultCode=0),
                                  ])
     server.dataReceived(str(pureldap.LDAPMessage(pureldap.LDAPBindRequest(), id=4)))
     reactor.iterate() #TODO
     self.assertEquals(server.transport.value(),
                       str(pureldap.LDAPMessage(pureldap.LDAPBindResponse(resultCode=0), id=4)))

    def _shutdownTestReactor(self, prevent_stop=False):
        """
        Called at the end of a test reactor run.

        When prevent_stop=True, the reactor will not be stopped.
        """
        if not self._timeout_reached:
            # Everything fine, disable timeout.
            if not self._reactor_timeout_call.cancelled:
                self._reactor_timeout_call.cancel()

        if prevent_stop:
            # Don't continue with stop procedure.
            return

        # Let the reactor know that we want to stop reactor.
        reactor.stop()
        # Let the reactor run one more time to execute the stop code.
        reactor.iterate()

        # Set flag to fake a clean reactor.
        reactor._startedBefore = False
        reactor._started = False
        reactor._justStopped = False
        reactor.running = False
        # Start running has consumed the startup events, so we need
        # to restore them.
        reactor.addSystemEventTrigger(
            'during', 'startup', reactor._reallyStartRunning)

    def testRetr(self):
        # Connect
        client = ftp.FTPClient(passive=self.passive)
        factory = ClientFactory()
        factory.noisy = 0
        factory.buildProtocol = lambda s, c=client: c
        reactor.connectTCP('localhost', self.ftp_port, factory)

        # download ftp_crap
        

        proto = BufferingProtocol()
        d = client.retr(os.path.basename('ftp_crap'), proto)
        d.addCallbacks(self.callback, self.errback)

        # Wait for a result
        id = reactor.callLater(5, self.errback, "timed out") # timeout so we don't freeze
        while not hasattr(self, 'result') and not hasattr(self, 'error'):
            reactor.iterate()
        try:
            id.cancel()
        except ValueError: pass

        error = getattr(self, 'error', None)
        if error:
            raise error[0], error[1], error[2]

        # Check that the file is the same as read directly off the disk
        self.failUnless(type(self.result) == types.ListType,
                        'callback result is wrong type: ' + str(self.result))
        data = proto.buf.getvalue()
        self.failUnless(data == open('ftp_crap', "rb").read(),
                        'RETRieved file does not match original')

    def testStor(self):
        # Connect
        client = ftp.FTPClient(passive=self.passive)
        client.debug = 1
        factory = ClientFactory()
        factory.noisy = 0
        factory.buildProtocol = lambda s, c=client: c
        reactor.connectTCP('localhost', self.ftp_port, factory)

        expectedContent = "Hello\n"*4
        
        def gotResult(c):
            c.write(expectedContent)
            c.finish()

        def gotErr(f):
            self.errback(f)

        t = client.storeFile("HelloThere")
        t[0].addCallbacks(gotResult, gotErr)
        t[1].addCallbacks(self.callback, self.errback)

        # Wait for a result
        id = reactor.callLater(5, self.errback, "timed out") # timeout so we don't freeze
        while not hasattr(self, 'result') and not hasattr(self, 'error'):
            reactor.iterate()
        try:
            id.cancel()
        except ValueError: pass

        error = getattr(self, 'error', None)
        if error:
            raise error[0], error[1], error[2]

        self.assertEquals(open('HelloThere').read(), expectedContent)

    def testCallLaterDelayAndReset(self):
        self._resetcallbackTime = None
        self._delaycallbackTime = None
        ireset = reactor.callLater(0.4, self._resetcallback)
        idelay = reactor.callLater(0.5, self._delaycallback)
        start = time.time()
        # chug a little before delaying
        while time.time() - start < 0.2:
            reactor.iterate(0.01)
        ireset.reset(0.4) # move expiration from 0.4 to (now)0.2+0.4=0.6
        idelay.delay(0.3) # move expiration from 0.5 to (orig)0.5+0.3=0.8
        # both should be called sometime during this
        while time.time() - start < 2:
            reactor.iterate(0.01)
        ireset_elapsed = self._resetcallbackTime - start
        idelay_elapsed = self._delaycallbackTime - start
        #self.assertApproximates(ireset_elapsed, 0.6, 0.4,
        #                        "ireset fired at %f (wanted 0.6)" % \
        #                        ireset_elapsed)
        #self.assertApproximates(idelay_elapsed, 0.8, 0.4,
        #                        "idelay fired at %f (wanted 0.8)" % \
        #                        idelay_elapsed)
        self.failUnless(idelay_elapsed >= ireset_elapsed,
                        "got %f, %f" % (idelay_elapsed, ireset_elapsed))
        self.failUnless(ireset_elapsed >= (0.6-0.05),
                        "got %f (wanted 0.6)" % ireset_elapsed)
        self.failUnless(idelay_elapsed >= (0.8-0.05),
                        "got %f (wanted 0.8)" % idelay_elapsed)

        del self._resetcallbackTime
        del self._delaycallbackTime

    def testShortFileListings(self):

        # Connect
        client = ftp.FTPClient(passive=self.passive)
        factory = ClientFactory()
        factory.noisy = 0
        factory.buildProtocol = lambda s, c=client: c
        reactor.connectTCP('localhost', self.ftp_port, factory)

        # Issue the command and set the callbacks
        p = BufferingProtocol()
        d = client.nlst('.', p)
        d.addCallbacks(self.callback, self.errback)

        # Wait for the result
        id = reactor.callLater(5, self.errback, "timed out") # timeout so we don't freeze
        while not hasattr(self, 'result') and not hasattr(self, 'error'):
            reactor.iterate()
        try:
            id.cancel()
        except ValueError: pass

        error = getattr(self, 'error', None)
        if error:
            raise error[0], error[1], error[2]

        # Check that the listing contains this file (ftp_crap)
        filenames = p.buf.getvalue().split('\r\n')
        self.failUnless('ftp_crap' in filenames,
                        'ftp_crap not in file listing')

    def testArticleRequest(self):
        d = self.backend.articleRequest(None, None, MESSAGE_ID)
        d.addCallbacks(self.callback, self.errback)

        id = reactor.callLater(5, self.timeout)
        while not hasattr(self, 'result') and not hasattr(self, 'error'):
            reactor.iterate()
        try:
            id.cancel()
        except ValueError: pass

        error = getattr(self, 'error', None)
        if error:
            raise error[0], error[1], error[2]

        self.failUnless(type(self.result) == types.TupleType,
                        'callback result is wrong type: ' + str(self.result))
        self.failUnless(len(self.result) == 3,
                        'callback result list should have three entries: ' +
                        str(self.result))
        self.failUnless(self.result[1] == MESSAGE_ID,
                        "callback result Message-Id doesn't match: %s vs %s" %
                        (MESSAGE_ID, self.result[1]))
        self.failUnless(self.result[2].find('\r\n\r\n'),
                        "Can't find \\r\\n\\r\\n between header and body")

    def testBodyRequest(self):
        self.testArticleRequest()
        index = self.result[0]

        try: del self.result
        except: pass

        try: del self.error
        except: pass

        d = self.backend.bodyRequest("alt.test.nntp", index)
        d.addCallbacks(self.callback, self.errback)

        id = reactor.callLater(5, self.timeout)
        while not hasattr(self, 'result') and not hasattr(self, 'error'):
            reactor.iterate()
        try:
            id.cancel()
        except ValueError: pass

        error = getattr(self, 'error', None)
        if error:
            raise error[0], error[1], error[2]

        self.failUnless(self.result[2][0:4] == 'this', "message body has been altered: " +
                        pformat(self.result[2][0:4]))

    def testHeadRequest(self):
        self.testArticleRequest()
        index = self.result[0]

        try: del self.result
        except: pass

        try: del self.error
        except: pass

        d = self.backend.headRequest("alt.test.nntp", index)
        d.addCallbacks(self.callback, self.errback)

        id = reactor.callLater(5, self.timeout)
        while not hasattr(self, 'result') and not hasattr(self, 'error'):
            reactor.iterate()
        try:
            id.cancel()
        except ValueError: pass

        error = getattr(self, 'error', None)
        if error:
            raise error[0], error[1], error[2]

        self.failUnless(self.result[1] == MESSAGE_ID,
                        "callback result Message-Id doesn't match: %s vs %s" %
                        (MESSAGE_ID, self.result[1]))

        self.failUnless(self.result[2][-2:] == '\r\n',
                        "headers must be \\r\\n terminated.")

    def test_devicemap_as_deferred(self):
        """Return aggregated device map"""

        # safety timeout
        self.timeout = reactor.callLater(10, self.failed, "retrieving devicemap timed out ... failing")

        d = self.aggregator.getDeviceMap()
        assert isinstance(d, defer.Deferred)

        # caution: as this deferred is ready-to-go, the callback is called *immediately*
        d.addCallback(self.succeeded)
        # i.e., self.succeeded has now been called
        
        # idle until code above triggers succeeded or timeout causes failure
        while not self.done:
             reactor.iterate(0.1)
        
        # will arrive here eventually when either succeeded or failed method has fired
        if self.failure:
            self.failed(self.failure)

        devicemap = self.lastargs[0]
        print devicemap
        assert type(devicemap) == types.DictType
        assert len(devicemap) == 1
        assert 'PseudoDevice' in devicemap
        assert len(devicemap['PseudoDevice']) == 2
        assert 'Device1' in devicemap['PseudoDevice']
        assert 'Device2' in devicemap['PseudoDevice']

    def runReactor(self, timesOrSeconds, seconds=False):
        """DEPRECATED: just return a deferred from your test method and
        trial with do the Right Thing. Alternatively, call
        twisted.trial.util.wait to block until the deferred fires.
        
        I'll iterate the reactor for a while.
        
        You probably want to use expectedAssertions with this.
        
        @type timesOrSeconds: int
        @param timesOrSeconds: Either the number of iterations to run,
               or, if `seconds' is True, the number of seconds to run for.

        @type seconds: bool
        @param seconds: If this is True, `timesOrSeconds' will be
               interpreted as seconds, rather than iterations.
        """
        warnings.warn("runReactor is deprecated. return a deferred from "
                      "your test method, and trial will wait for results."
                      "Alternatively, call twisted.trial.util.wait to"
                      "block until the deferred fires.",
                      DeprecationWarning, stacklevel=2)
        from twisted.internet import reactor

        if seconds:
            reactor.callLater(timesOrSeconds, reactor.crash)
            reactor.run()
            return

        for i in xrange(timesOrSeconds):
            reactor.iterate()

 def test_quit(self):
     """command quit"""
     # not using deferredResult here since it swallows timeout
     deferred = self.factory.check("quit")
     deferred.addCallback(self._assert_quit)
     while not self.done:
         reactor.iterate(0.1)

 def testOldAddress(self):
     server = Server()
     p = reactor.listenUDP(0, server, interface="127.0.0.1")
     reactor.iterate()
     addr = p.getHost()
     self.assertEquals(addr, ('INET_UDP', addr.host, addr.port))
     p.stopListening()

 def testDeferredFailure(self):
     def raiseError(): raise TypeError
     d = threads.deferToThread(raiseError)
     d.addErrback(self._resultErrback)
     while not self.gotResult:
         reactor.iterate()
     self.gotResult = 0

 def buildProtocol(self, addr):
     p = TCPListener()
     ah = listenAirhookStream(randrange(10000,12000), UDPOutFactory(p))
     reactor.iterate()
     c = ah.connectionForAddr(self.oaddr)
     #c.noisy= 1 
     return p