class FakeEndpointAnswers:
    implements(IStreamClientEndpoint)

    def __init__(self, answers):
        self.answers = answers
        # since we use pop() we need these to be "backwards"
        self.answers.reverse()

    def get_info_raw(self, keys):
        ans = ''
        for k in keys.split():
            if len(self.answers) == 0:
                raise TorProtocolError(551, "ran out of answers")
            ans += '%s=%s\r\n' % (k, self.answers.pop())
        return ans[:-2]                 # don't want trailing \r\n

    def get_info_incremental(self, key, linecb):
        linecb('%s=%s' % (key, self.answers.pop()))
        return defer.succeed('')

    def connect(self, protocol_factory):
        self.proto = TorControlProtocol()
        self.proto.transport = proto_helpers.StringTransport()
        self.proto.get_info_raw = self.get_info_raw
        self.proto.get_info_incremental = self.get_info_incremental
        self.proto._set_valid_events('GUARD STREAM CIRC NS NEWCONSENSUS ORCONN NEWDESC ADDRMAP STATUS_GENERAL')

        return defer.succeed(self.proto)

class ProtocolIntegrationTests(unittest.TestCase):
    """
    Tests which use a real TorControlProtocol objects, not a mock.
    """

    def setUp(self):
        self.protocol = TorControlProtocol(lambda: defer.succeed('foo'))
        self.transport = proto_helpers.StringTransport()

    def send(self, line):
        self.protocol.dataReceived(line.strip() + "\r\n")

    @defer.inlineCallbacks
    def test_with_arg(self):
        info = TorInfo(self.protocol)
        pb = info.post_bootstrap

        ## now we hook up the protocol like it connected to a real Tor
        self.protocol.makeConnection(self.transport)

        ## answer all the requests generated by TorControlProtocol boostrapping etc.
        self.send('250-AUTH METHODS=PASSWORD')
        self.send('250 OK')

        ## response to AUTHENTICATE
        self.send('250 OK')

        ## now we're in _bootstrap() in TorControlProtocol()
        self.send("250-version=foo")
        self.send("250 OK")

        self.send("250-events/names=")
        self.send("250 OK")

        self.send("250 OK")  # for USEFEATURE

        ## do the TorInfo magic
        self.send('250-info/names=')
        self.send('250-multi/path/arg/* a documentation string')
        self.send('250 OK')

        ## we had to save this up above due to the "interesting" way
        ## TorInfo switches to become a possible-nice magic thingy
        ## that does attribute-access for you.
        yield pb

        self.assertTrue(hasattr(info, 'multi'))
        self.assertTrue(hasattr(getattr(info, 'multi'), 'path'))
        self.assertTrue(hasattr(getattr(getattr(info, 'multi'), 'path'), 'arg'))

        ## Finally! The test! We see if we can get this multi-path
        ## value with an argument...
        ## a "real" tor example is "net/listeners/socks" which shows
        ## up in info/names as "net/listeners/*"

        d = info.multi.path.arg('quux')
        d.addCallback(CheckAnswer(self, 'foo'))
        self.send("250-multi/path/arg/quux=foo")
        self.send("250 OK")
        yield d

class FakeTorState(object):

    def __init__(self, routers):
        self.routers = routers
        self.protocol = TorControlProtocol()
        self.protocol.connectionMade = lambda: None
        self.protocol.transport = proto_helpers.StringTransport()
        self.protocol.makeConnection(self.protocol.transport)

    def _find_circuit_after_extend(self, x):
        return defer.succeed(None)

    def build_circuit(self, routers=None, using_guards=True):
        cmd = "EXTENDCIRCUIT 0 "
        first = True
        for router in routers:
            if first:
                first = False
            else:
                cmd += ','
            if isinstance(router, basestring) and len(router) == 40 \
               and hashFromHexId(router):
                cmd += router
            else:
                cmd += router.id_hex[1:]
        d = self.protocol.queue_command(cmd)
        print "d %r" % (d,)
        d.addCallback(self._find_circuit_after_extend)
        return d

 def test_set_conf_wrong_args(self):
     ctl = TorControlProtocol()
     d = ctl.set_conf('a')
     self.assertTrue(d.called)
     self.assertTrue(d.result)
     self.assertTrue('even number' in d.result.getErrorMessage())
     ## ignore the error so trial doesn't get unhappy
     d.addErrback(lambda foo: True)
     return d

class LogicTests(unittest.TestCase):
    def setUp(self):
        self.protocol = TorControlProtocol()
        self.protocol.connectionMade = lambda: None
        self.transport = proto_helpers.StringTransport()
        self.protocol.makeConnection(self.transport)

    def test_set_conf_wrong_args(self):
        ctl = TorControlProtocol()
        d = ctl.set_conf("a")
        self.assertTrue(d.called)
        self.assertTrue(d.result)
        self.assertTrue("even number" in d.result.getErrorMessage())
        ## ignore the error so trial doesn't get unhappy
        d.addErrback(lambda foo: True)
        return d

 def setUp(self):
     self.protocol = TorControlProtocol()
     self.protocol.connectionMade = lambda: None
     self.transport = proto_helpers.StringTransportWithDisconnection()
     self.protocol.makeConnection(self.transport)
     # why doesn't makeConnection do this?
     self.transport.protocol = self.protocol

    def connect(self, protocol_factory):
        self.proto = TorControlProtocol()
        self.proto.transport = proto_helpers.StringTransport()
        self.proto.get_info_raw = self.get_info_raw
        self.proto.get_info_incremental = self.get_info_incremental
        self.proto._set_valid_events('GUARD STREAM CIRC NS NEWCONSENSUS ORCONN NEWDESC ADDRMAP STATUS_GENERAL')

        return defer.succeed(self.proto)

class FakeReactorTcp(FakeReactor):
    implements(IReactorTCP)

    failures = 0
    _port_generator = port_generator()

    def __init__(self, test):
        self.protocol = TorControlProtocol()
        self.protocol.connectionMade = lambda: None
        self.transport = proto_helpers.StringTransport()
        self.transport = FakeProcessTransport()
        self.transport.protocol = self.protocol

        def blam():
            self.protocol.outReceived("Bootstrap")
        self.transport.closeStdin = blam
        self.protocol.makeConnection(self.transport)
        FakeReactor.__init__(self, test, self.transport, lambda x: None)

    def listenTCP(self, port, factory, **kwargs):
        '''returns IListeningPort'''
        if self.failures > 0:
            self.failures -= 1
            raise error.CannotListenError(None, None, None)

        if port == 0:
            port = self._port_generator.next()
        p = FakeListeningPort(port)
        p.factory = factory
        p.startListening()
        return p

    def connectTCP(self, host, port, factory, timeout, bindAddress):
        '''should return IConnector'''
#        print "CONN", host, port, factory
#        print dir(factory)
#        print "XX", factory
        r = tcp.Connector(host, port, factory, timeout, bindAddress, reactor=self)
#        print dir(r)

        def blam(*args):
            print "BLAAAAAM", args
        r.connect = blam
        return r

    def __init__(self, test):
        self.protocol = TorControlProtocol()
        self.protocol.connectionMade = lambda: None
        self.transport = proto_helpers.StringTransport()
        self.transport.protocol = self.protocol

        def blam():
            self.protocol.outReceived(b"Bootstrap")
        self.transport.closeStdin = blam
        self.protocol.makeConnection(self.transport)
        self.test = test

    def __init__(self, test):
        self.protocol = TorControlProtocol()
        self.protocol.connectionMade = lambda: None
        self.transport = proto_helpers.StringTransport()
        self.transport = FakeProcessTransport()
        self.transport.protocol = self.protocol

        def blam():
            self.protocol.outReceived("Bootstrap")
        self.transport.closeStdin = blam
        self.protocol.makeConnection(self.transport)
        FakeReactor.__init__(self, test, self.transport, lambda x: None)

class AuthenticationTests(unittest.TestCase):
    def setUp(self):
        self.protocol = TorControlProtocol()
        self.transport = proto_helpers.StringTransport()

    def send(self, line):
        self.protocol.dataReceived(line.strip() + "\r\n")
        
    def test_authenticate_cookie(self):
        self.protocol.makeConnection(self.transport)
        self.assertEqual(self.transport.value(), 'PROTOCOLINFO 1\r\n')
        self.transport.clear()
        cookie_data = 'cookiedata!cookiedata!cookiedata'
        open('authcookie', 'w').write(cookie_data)
        self.send('250-PROTOCOLINFO 1')
        self.send('250-AUTH METHODS=COOKIE,HASHEDPASSWORD COOKIEFILE="authcookie"')
        self.send('250-VERSION Tor="0.2.2.34"')
        self.send('250 OK')

        self.assertEqual(self.transport.value(), 'AUTHENTICATE %s\r\n' % cookie_data.encode("hex"))

    def test_authenticate_password(self):
        self.protocol.password = 'foo'
        self.protocol.makeConnection(self.transport)
        self.assertEqual(self.transport.value(), 'PROTOCOLINFO 1\r\n')
        self.transport.clear()
        self.send('250-PROTOCOLINFO 1')
        self.send('250-AUTH METHODS=HASHEDPASSWORD')
        self.send('250-VERSION Tor="0.2.2.34"')
        self.send('250 OK')

        self.assertEqual(self.transport.value(), 'AUTHENTICATE %s\r\n' % "foo".encode("hex"))

    def confirmAuthFailed(self, *args):
        self.auth_failed = True
        
    def test_authenticate_no_password(self):
        self.protocol.post_bootstrap.addErrback(self.confirmAuthFailed)
        self.auth_failed = False
        
        self.protocol.makeConnection(self.transport)
        self.assertEqual(self.transport.value(), 'PROTOCOLINFO 1\r\n')
        
        self.send('250-PROTOCOLINFO 1')
        self.send('250-AUTH METHODS=HASHEDPASSWORD')
        self.send('250-VERSION Tor="0.2.2.34"')
        self.send('250 OK')

        self.assertTrue(self.auth_failed)

class TorStatePy3Tests(unittest.TestCase):

    def setUp(self):
        self.protocol = TorControlProtocol()
        self.state = TorState(self.protocol)
        # avoid spew in trial logs; state prints this by default
        self.state._attacher_error = lambda f: f
        self.protocol.connectionMade = lambda: None
        self.transport = proto_helpers.StringTransport()
        self.protocol.makeConnection(self.transport)

    def send(self, line):
        self.protocol.dataReceived(line.strip() + b"\r\n")

    def test_attacher_coroutine(self):
        @implementer(IStreamAttacher)
        class MyAttacher(object):

            def __init__(self, answer):
                self.streams = []
                self.answer = answer

            async def attach_stream(self, stream, circuits):
                self.streams.append(stream)
                x = await defer.succeed(self.answer)
                return x

        self.state.circuits[1] = FakeCircuit(1)
        self.state.circuits[1].state = 'BUILT'
        attacher = MyAttacher(self.state.circuits[1])
        self.state.set_attacher(attacher, FakeReactor(self))

        # boilerplate to finish enough set-up in the protocol so it
        # works
        events = 'GUARD STREAM CIRC NS NEWCONSENSUS ORCONN NEWDESC ADDRMAP STATUS_GENERAL'
        self.protocol._set_valid_events(events)
        self.state._add_events()
        for ignored in self.state.event_map.items():
            self.send(b"250 OK")

        self.send(b"650 STREAM 1 NEW 0 ca.yahoo.com:80 SOURCE_ADDR=127.0.0.1:54327 PURPOSE=USER")
        self.send(b"650 STREAM 1 REMAP 0 87.248.112.181:80 SOURCE=CACHE")
        self.assertEqual(len(attacher.streams), 1)
        self.assertEqual(attacher.streams[0].id, 1)
        self.assertEqual(len(self.protocol.commands), 1)
        self.assertEqual(self.protocol.commands[0][1], b'ATTACHSTREAM 1 1')

class ProtocolTests(unittest.TestCase):
    def setUp(self):
        self.protocol = TorControlProtocol()
        self.protocol.connectionMade = lambda: None
        self.transport = proto_helpers.StringTransport()
        self.protocol.makeConnection(self.transport)

    def tearDown(self):
        self.protocol = None

    def send(self, line):
        self.protocol.dataReceived(line.strip() + "\r\n")

    def test_statemachine_broadcast_no_code(self):
        try:
            self.protocol._broadcast_response("foo")
            self.fail()
        except RuntimeError, e:
            self.assertTrue("No code set yet" in str(e))

class DisconnectionTests(unittest.TestCase):
    def setUp(self):
        self.protocol = TorControlProtocol()
        self.protocol.connectionMade = lambda: None
        self.transport = proto_helpers.StringTransportWithDisconnection()
        self.protocol.makeConnection(self.transport)
        # why doesn't makeConnection do this?
        self.transport.protocol = self.protocol

    def tearDown(self):
        self.protocol = None

    def test_disconnect_callback(self):
        """
        see that we get our callback on_disconnect if the transport
        goes away
        """
        def it_was_called(*args):
            it_was_called.yes = True
            return None
        it_was_called.yes = False
        self.protocol.on_disconnect.addCallback(it_was_called)
        self.protocol.on_disconnect.addErrback(it_was_called)
        f = failure.Failure(error.ConnectionDone("It's all over"))
        self.protocol.connectionLost(f)
        self.assertTrue(it_was_called.yes)

    def test_disconnect_errback(self):
        """
        see that we get our callback on_disconnect if the transport
        goes away
        """
        def it_was_called(*args):
            it_was_called.yes = True
            return None
        it_was_called.yes = False
        self.protocol.on_disconnect.addCallback(it_was_called)
        self.protocol.on_disconnect.addErrback(it_was_called)
        f = failure.Failure(RuntimeError("The thing didn't do the stuff."))
        self.protocol.connectionLost(f)
        self.assertTrue(it_was_called.yes)

 def setUp(self):
     self.protocol = TorControlProtocol()
     self.transport = proto_helpers.StringTransport()

class DisconnectionTests(unittest.TestCase):
    def setUp(self):
        self.protocol = TorControlProtocol()
        self.protocol.connectionMade = lambda: None
        self.transport = proto_helpers.StringTransportWithDisconnection()
        self.protocol.makeConnection(self.transport)
        # why doesn't makeConnection do this?
        self.transport.protocol = self.protocol

    def tearDown(self):
        self.protocol = None

    def test_disconnect_callback(self):
        """
        see that we get our callback on_disconnect if the transport
        goes away
        """
        def it_was_called(*args):
            it_was_called.yes = True
            return None
        it_was_called.yes = False
        self.protocol.on_disconnect.addCallback(it_was_called)
        self.protocol.on_disconnect.addErrback(it_was_called)
        f = failure.Failure(error.ConnectionDone("It's all over"))
        self.protocol.connectionLost(f)
        self.assertTrue(it_was_called.yes)

    def test_when_disconnect(self):
        """
        see that we get our callback for when_disconnected if the
        transport goes away
        """
        def it_was_called(arg):
            it_was_called.yes = True
            return None
        it_was_called.yes = False

        d = self.protocol.when_disconnected()
        d.addCallback(it_was_called)
        f = failure.Failure(error.ConnectionDone("It's all over"))
        self.protocol.connectionLost(f)
        self.assertTrue(it_was_called.yes)

    def test_when_disconnect_error(self):
        """
        see that we get our errback for when_disconnected if the
        transport goes away
        """
        def it_was_called(arg):
            it_was_called.yes = True
            return None
        it_was_called.yes = False

        d = self.protocol.when_disconnected()
        d.addErrback(it_was_called)
        f = failure.Failure(RuntimeError("sadness"))
        self.protocol.connectionLost(f)
        self.assertTrue(it_was_called.yes)

    def test_disconnect_errback(self):
        """
        see that we get our callback on_disconnect if the transport
        goes away
        """
        def it_was_called(*args):
            it_was_called.yes = True
            return None
        it_was_called.yes = False
        self.protocol.on_disconnect.addCallback(it_was_called)
        self.protocol.on_disconnect.addErrback(it_was_called)
        f = failure.Failure(RuntimeError("The thing didn't do the stuff."))
        self.protocol.connectionLost(f)
        self.assertTrue(it_was_called.yes)

    def test_disconnect_outstanding_commands(self):
        """
        outstanding commands should errback on disconnect
        """

        def it_was_called(f):
            str(f)
            it_was_called.count += 1
            return None
        it_was_called.count = 0

        # we want to make sure outstanding commands get errbacks
        d0 = self.protocol.queue_command("some command0")
        d1 = self.protocol.queue_command("some command1")
        d0.addErrback(it_was_called)
        d1.addErrback(it_was_called)
        self.protocol.on_disconnect.addErrback(lambda _: None)

        f = failure.Failure(RuntimeError("The thing didn't do the stuff."))
        self.protocol.connectionLost(f)
        self.assertEqual(it_was_called.count, 2)

class FakeReactorTcp(object):

    failures = 0
    _port_generator = port_generator()

    def __init__(self, test):
        self.protocol = TorControlProtocol()
        self.protocol.connectionMade = lambda: None
        self.transport = proto_helpers.StringTransport()
        self.transport.protocol = self.protocol

        def blam():
            self.protocol.outReceived(b"Bootstrap")
        self.transport.closeStdin = blam
        self.protocol.makeConnection(self.transport)
        self.test = test

    def spawnProcess(self, processprotocol, bin, args, env, path,
                     uid=None, gid=None, usePTY=None, childFDs=None):
        self.protocol = processprotocol
        self.protocol.makeConnection(self.transport)
        self.transport.process_protocol = processprotocol
        return self.transport

    def addSystemEventTrigger(self, *args):
        self.test.assertEqual(args[0], 'before')
        self.test.assertEqual(args[1], 'shutdown')
        # we know this is just for the temporary file cleanup, so we
        # nuke it right away to avoid polluting /tmp by calling the
        # callback now.
        args[2]()

    def listenTCP(self, port, factory, **kwargs):
        '''returns IListeningPort'''
        if self.failures > 0:
            self.failures -= 1
            raise error.CannotListenError(None, None, None)

        if port == 0:
            port = next(self._port_generator)
        p = FakeListeningPort(port)
        p.factory = factory
        p.startListening()
        return p

    def connectTCP(self, host, port, factory, timeout, bindAddress):
        '''should return IConnector'''
        r = tcp.Connector(
            host, port, factory, timeout,
            bindAddress, reactor=self
        )

        def blam(*args):
            print("BLAAAAAM", args)
        r.connect = blam
        return r

    def connectUNIX(self, address, factory, timeout=30, checkPID=0):
        '''should return IConnector'''
        r = unix.Connector(
            address, factory, timeout, self, checkPID,
        )

        def blam(*args):
            print("BLAAAAAM", args)
        r.connect = blam
        return r

class ProtocolTests(unittest.TestCase):

    def setUp(self):
        self.protocol = TorControlProtocol()
        self.protocol.connectionMade = lambda: None
        self.transport = proto_helpers.StringTransport()
        self.protocol.makeConnection(self.transport)

    def tearDown(self):
        self.protocol = None

    def send(self, line):
        assert type(line) == bytes
        self.protocol.dataReceived(line.strip() + b"\r\n")

    def test_statemachine_broadcast_no_code(self):
        try:
            self.protocol._broadcast_response("foo")
            self.fail()
        except RuntimeError as e:
            self.assertTrue('No code set yet' in str(e))

    def test_statemachine_broadcast_unknown_code(self):
        try:
            self.protocol.code = 999
            self.protocol._broadcast_response("foo")
            self.fail()
        except RuntimeError as e:
            self.assertTrue('Unknown code' in str(e))

    def test_statemachine_is_finish(self):
        self.assertTrue(not self.protocol._is_finish_line(''))
        self.assertTrue(self.protocol._is_finish_line('.'))
        self.assertTrue(self.protocol._is_finish_line('300 '))
        self.assertTrue(not self.protocol._is_finish_line('250-'))

    def test_statemachine_singleline(self):
        self.assertTrue(not self.protocol._is_single_line_response('foo'))

    def test_statemachine_continuation(self):
        try:
            self.protocol.code = 250
            self.protocol._is_continuation_line("123 ")
            self.fail()
        except RuntimeError as e:
            self.assertTrue('Unexpected code' in str(e))

    def test_statemachine_multiline(self):
        try:
            self.protocol.code = 250
            self.protocol._is_multi_line("123 ")
            self.fail()
        except RuntimeError as e:
            self.assertTrue('Unexpected code' in str(e))

    def test_response_with_no_request(self):
        with self.assertRaises(RuntimeError) as ctx:
            self.protocol.code = 200
            self.protocol._broadcast_response('200 OK')
        self.assertTrue(
            "didn't issue a command" in str(ctx.exception)
        )

    def auth_failed(self, msg):
        self.assertEqual(str(msg.value), '551 go away')
        self.got_auth_failed = True

    def test_authenticate_fail(self):
        self.got_auth_failed = False
        self.protocol._auth_failed = self.auth_failed

        self.protocol.password_function = lambda: 'foo'
        self.protocol._do_authenticate('''PROTOCOLINFO 1
AUTH METHODS=HASHEDPASSWORD
VERSION Tor="0.2.2.35"
OK''')
        self.send(b'551 go away\r\n')
        self.assertTrue(self.got_auth_failed)

    def test_authenticate_no_auth_line(self):
        try:
            self.protocol._do_authenticate('''PROTOCOLINFO 1
FOOAUTH METHODS=COOKIE,SAFECOOKIE COOKIEFILE="/dev/null"
VERSION Tor="0.2.2.35"
OK''')
            self.assertTrue(False)
        except RuntimeError as e:
            self.assertTrue('find AUTH line' in str(e))

    def test_authenticate_not_enough_cookie_data(self):
        with tempfile.NamedTemporaryFile() as cookietmp:
            cookietmp.write(b'x' * 35)  # too much data
            cookietmp.flush()

            try:
                self.protocol._do_authenticate('''PROTOCOLINFO 1
AUTH METHODS=COOKIE COOKIEFILE="%s"
VERSION Tor="0.2.2.35"
OK''' % cookietmp.name)
                self.assertTrue(False)
#.........这里部分代码省略.........

class LaunchTorTests(unittest.TestCase):
    def setUp(self):
        self.protocol = TorControlProtocol()
        self.protocol.connectionMade = do_nothing
        self.transport = proto_helpers.StringTransport()
        self.protocol.makeConnection(self.transport)

    def setup_complete_no_errors(self, proto):
        todel = proto.to_delete
        self.assertTrue(len(todel) > 0)
        proto.processEnded(Failure(error.ProcessDone(0)))
        self.assertTrue(len(proto.to_delete) == 0)
        for f in todel:
            self.assertTrue(not os.path.exists(f))

    def setup_complete_fails(self, proto):
        todel = proto.to_delete
        self.assertTrue(len(todel) > 0)
        ## the "12" is just arbitrary, we check it later in the error-message
        proto.processEnded(Failure(error.ProcessTerminated(12, None, 'statusFIXME')))
        self.assertTrue(len(proto.to_delete) == 0)
        for f in todel:
            self.assertTrue(not os.path.exists(f))

    def test_basic_launch(self):
        config = TorConfig()
        config.OrPort = 1234
        config.SocksPort = 9999

        def connector(proto, trans):
            proto._set_valid_events('STATUS_CLIENT')
            proto.makeConnection(trans)
            proto.post_bootstrap.callback(proto)
            return proto.post_bootstrap

        class OnProgress:
            def __init__(self, test, expected):
                self.test = test
                self.expected = expected

            def __call__(self, percent, tag, summary):
                self.test.assertTrue(self.expected[0] == (percent, tag, summary))
                self.expected = self.expected[1:]
                self.test.assertTrue('"' not in summary)
                self.test.assertTrue(percent >= 0 and percent <= 100)            
            
        def on_protocol(proto):
            proto.outReceived('Bootstrapped 100%\n')
            proto.progress = OnProgress(self, [(90, 'circuit_create', 'Establishing a Tor circuit'),
                                               (100, 'done', 'Done')])

        trans = FakeProcessTransport()
        trans.protocol = self.protocol
        self.othertrans = trans
        creator = functools.partial(connector, self.protocol, self.transport)
        d = launch_tor(config, FakeReactor(self, trans, on_protocol), connection_creator=creator)
        d.addCallback(self.setup_complete_no_errors)
        return d
        
    def check_setup_failure(self, fail):
        self.assertTrue("with error-code 12" in fail.getErrorMessage())
        ## cancel the errback chain, we wanted this
        return None
                
    def test_launch_tor_fails(self):
        config = TorConfig()
        config.OrPort = 1234
        config.SocksPort = 9999

        def connector(proto, trans):
            proto._set_valid_events('STATUS_CLIENT')
            proto.makeConnection(trans)
            proto.post_bootstrap.callback(proto)
            return proto.post_bootstrap
            
        def on_protocol(proto):
            proto.outReceived('Bootstrapped 100%\n')
            
        trans = FakeProcessTransport()
        trans.protocol = self.protocol
        self.othertrans = trans
        creator = functools.partial(connector, self.protocol, self.transport)
        d = launch_tor(config, FakeReactor(self, trans, on_protocol), connection_creator=creator)
        d.addCallback(self.setup_complete_fails)
        d.addErrback(self.check_setup_failure)
        return d

    def setup_fails_stderr(self, fail):
        self.assertTrue('Something went horribly wrong!' in fail.getErrorMessage())
        ## cancel the errback chain, we wanted this
        return None
        
    def test_tor_produces_stderr_output(self):
        config = TorConfig()
        config.OrPort = 1234
        config.SocksPort = 9999

        def connector(proto, trans):
            proto._set_valid_events('STATUS_CLIENT')
            proto.makeConnection(trans)
#.........这里部分代码省略.........

class StateTests(unittest.TestCase):

    def setUp(self):
        self.protocol = TorControlProtocol()
        self.state = TorState(self.protocol)
        self.protocol.connectionMade = lambda: None
        self.transport = proto_helpers.StringTransport()
        self.protocol.makeConnection(self.transport)

    def test_close_stream_with_attacher(self):
        class MyAttacher(object):
            implements(IStreamAttacher)

            def __init__(self):
                self.streams = []

            def attach_stream(self, stream, circuits):
                self.streams.append(stream)
                return None

        attacher = MyAttacher()
        self.state.set_attacher(attacher, FakeReactor(self))
        self.state._stream_update("76 CLOSED 0 www.example.com:0 REASON=DONE")

    def test_stream_update(self):
        ## we use a circuit ID of 0 so it doesn't try to look anything up but it's
        ## not really correct to have a  SUCCEEDED w/o a valid circuit, I don't think
        self.state._stream_update('1610 SUCCEEDED 0 74.125.224.243:80')
        self.assertTrue(1610 in self.state.streams)

    def test_single_streams(self):
        self.state.circuits[496] = FakeCircuit(496)
        self.state._stream_status('stream-status=123 SUCCEEDED 496 www.example.com:6667\r\nOK')
        self.assertEqual(len(self.state.streams), 1)

    def send(self, line):
        self.protocol.dataReceived(line.strip() + "\r\n")

    def test_bootstrap_callback(self):
        '''
        FIXME: something is still screwy with this; try throwing an
        exception from TorState.bootstrap and we'll just hang...
        '''

        d = self.state.post_bootstrap

        self.protocol._set_valid_events(' '.join(self.state.event_map.keys()))
        self.state._bootstrap()

        self.send("250+ns/all=")
        self.send(".")
        self.send("250 OK")

        self.send("250+circuit-status=")
        self.send(".")
        self.send("250 OK")

        self.send("250-stream-status=")
        self.send("250 OK")

        self.send("250-address-mappings/all=")
        self.send("250 OK")

        for ignored in self.state.event_map.items():
            self.send("250 OK")

        fakerouter = object()
        self.state.routers['$0000000000000000000000000000000000000000'] = fakerouter
        self.state.routers['$9999999999999999999999999999999999999999'] = fakerouter
        self.send("250+entry-guards=")
        self.send("$0000000000000000000000000000000000000000=name up")
        self.send("$1111111111111111111111111111111111111111=foo up")
        self.send("$9999999999999999999999999999999999999999=eman unusable 2012-01-01 22:00:00")
        self.send(".")
        self.send("250 OK")

        ## implicitly created Router object for the $1111...11 lookup
        ## but 0.0.0.0 will have to country, so Router will ask Tor
        ## for one via GETINFO ip-to-country
        self.send("250-ip-to-country/0.0.0.0=??")
        self.send("250 OK")

        self.assertEqual(len(self.state.entry_guards), 2)
        self.assertTrue('$0000000000000000000000000000000000000000' in self.state.entry_guards)
        self.assertEqual(self.state.entry_guards['$0000000000000000000000000000000000000000'], fakerouter)
        self.assertTrue('$1111111111111111111111111111111111111111' in self.state.entry_guards)

        self.assertEqual(len(self.state.unusable_entry_guards), 1)
        self.assertTrue('$9999999999999999999999999999999999999999' in self.state.unusable_entry_guards[0])

        return d

    def test_bootstrap_existing_addresses(self):
        '''
        FIXME: something is still screwy with this; try throwing an
        exception from TorState.bootstrap and we'll just hang...
        '''

        d = self.state.post_bootstrap

#.........这里部分代码省略.........