How to use the txtorcon.Circuit function in txtorcon

def test_build_circuit_timeout_after_progress(self):
        """
        Similar to above but we timeout after Tor has ack'd our
        circuit-creation attempt, but before reaching BUILT.
        """
        class FakeRouter:
            def __init__(self, i):
                self.id_hex = i
                self.flags = []

        class FakeCircuit(Circuit):
            def close(self):
                return defer.succeed(None)

        path = []
        for x in range(3):
            path.append(FakeRouter("$%040d" % x))

        def fake_queue(cmd):
            self.assertTrue(cmd.startswith('EXTENDCIRCUIT 0'))
            return defer.succeed("EXTENDED 1234")

        queue_command = patch.object(self.protocol, 'queue_command', fake_queue)
        circuit_factory = patch.object(self.state, 'circuit_factory', FakeCircuit)
        with queue_command, circuit_factory:
            timeout = 10
            clock = task.Clock()

def test_is_built_errback(self):
        tor = FakeTorController()
        a = FakeRouter('$E11D2B2269CC25E67CA6C9FB5843497539A74FD0', 'a')
        tor.routers['$E11D2B2269CC25E67CA6C9FB5843497539A74FD0'] = a

        state = TorState(tor)
        circuit = Circuit(tor)
        circuit.listen(tor)

        circuit.update('123 EXTENDED $E11D2B2269CC25E67CA6C9FB5843497539A74FD0=eris PURPOSE=GENERAL'.split())
        state.circuit_new(circuit)
        d = circuit.when_built()

        called = []

        def err(f):
            called.append(f)
            return None
        d.addErrback(err)

        state.circuit_closed(circuit, REASON='testing')

        self.assertEqual(1, len(called))

def test_unlisten(self):
        tor = FakeTorController()
        tor.routers['$E11D2B2269CC25E67CA6C9FB5843497539A74FD0'] = FakeRouter(
            '$E11D2B2269CC25E67CA6C9FB5843497539A74FD0', 'a'
        )

        circuit = Circuit(tor)
        circuit.listen(tor)
        circuit.listen(tor)
        circuit.update('1 LAUNCHED PURPOSE=GENERAL'.split())
        circuit.unlisten(tor)
        circuit.update('1 EXTENDED $E11D2B2269CC25E67CA6C9FB5843497539A74FD0=eris PURPOSE=GENERAL'.split())
        self.assertEqual(len(tor.circuits), 1)
        self.assertTrue(1 in tor.circuits)
        self.assertEqual(len(tor.extend), 0)
        self.assertEqual(1, len(circuit.path))
        self.assertEqual(0, len(circuit.listeners))

def test_circuit_web_agent(self):
        tor = FakeTorController()
        a = FakeRouter('$E11D2B2269CC25E67CA6C9FB5843497539A74FD0', 'a')
        tor.routers['$E11D2B2269CC25E67CA6C9FB5843497539A74FD0'] = a

        circuit = Circuit(tor)
        reactor = Mock()

        # just testing this doesn't cause an exception
        circuit.web_agent(reactor)

def test_stream_success(self):
        tor = FakeTorController()
        a = FakeRouter('$E11D2B2269CC25E67CA6C9FB5843497539A74FD0', 'a')
        tor.routers['$E11D2B2269CC25E67CA6C9FB5843497539A74FD0'] = a

        circuit = Circuit(tor)
        reactor = Mock()

        circuit.stream_via(
            reactor, 'torproject.org', 443, None,
            use_tls=True,
        )

def test_age_default(self, fake_datetime):
        """
        age() w/ defaults works properly
        """
        from datetime import datetime
        now = datetime.fromtimestamp(60.0)
        fake_datetime.return_value = now
        fake_datetime.utcnow = Mock(return_value=now)
        tor = FakeTorController()

        circuit = Circuit(tor)
        circuit._time_created = datetime.fromtimestamp(0.0)
        self.assertEquals(circuit.age(), 60)
        self.assertTrue(circuit.time_created is not None)

self.test.assertEqual(args[2], self.test.state.undo_attacher)
        return 1

    def removeSystemEventTrigger(self, id):
        self.test.assertEqual(id, 1)

    def connectTCP(self, *args, **kw):
        """for testing build_tor_connection"""
        raise RuntimeError('connectTCP: ' + str(args))

    def connectUNIX(self, *args, **kw):
        """for testing build_tor_connection"""
        raise RuntimeError('connectUNIX: ' + str(args))


class FakeCircuit(Circuit):

    def __init__(self, id=-999):
        self.streams = []
        self.id = id
        self.state = 'BOGUS'


@implementer(IStreamClientEndpoint)
class FakeEndpoint:

    def get_info_raw(self, keys):
        ans = '\r\n'.join(map(lambda k: '%s=' % k, keys.split()))
        return defer.succeed(ans)

    def get_info_incremental(self, key, linecb):
        linecb('%s=' % key)

def test_close_circuit(self):
        tor = FakeTorController()
        a = FakeRouter('$E11D2B2269CC25E67CA6C9FB5843497539A74FD0', 'a')
        b = FakeRouter('$50DD343021E509EB3A5A7FD0D8A4F8364AFBDCB5', 'b')
        c = FakeRouter('$253DFF1838A2B7782BE7735F74E50090D46CA1BC', 'c')
        tor.routers['$E11D2B2269CC25E67CA6C9FB5843497539A74FD0'] = a
        tor.routers['$50DD343021E509EB3A5A7FD0D8A4F8364AFBDCB5'] = b
        tor.routers['$253DFF1838A2B7782BE7735F74E50090D46CA1BC'] = c

        circuit = Circuit(tor)
        circuit.listen(tor)

        circuit.update('123 EXTENDED $E11D2B2269CC25E67CA6C9FB5843497539A74FD0=eris,$50DD343021E509EB3A5A7FD0D8A4F8364AFBDCB5=venus,$253DFF1838A2B7782BE7735F74E50090D46CA1BC=chomsky PURPOSE=GENERAL'.split())

        self.assertEqual(3, len(circuit.path))
        d0 = circuit.close()
        # we already pretended that Tor answered "OK" to the
        # CLOSECIRCUIT call (see close_circuit() in FakeTorController
        # above) however the circuit isn't "really" closed yet...
        self.assertTrue(not d0.called)
        # not unit-test-y? shouldn't probably delve into internals I
        # suppose...
        self.assertTrue(circuit._closing_deferred is not None)

        # if we try to close it again (*before* the actual close has
        # succeeded!) we should also still be waiting.

def test_path_update(self):
        cp = TorControlProtocol()
        state = TorState(cp, False)
        circuit = Circuit(state)
        circuit.update('1 EXTENDED $E11D2B2269CC25E67CA6C9FB5843497539A74FD0=eris PURPOSE=GENERAL'.split())
        self.assertEqual(1, len(circuit.path))
        self.assertEqual(
            '$E11D2B2269CC25E67CA6C9FB5843497539A74FD0',
            circuit.path[0].id_hex
        )
        self.assertEqual('eris', circuit.path[0].name)

def test_wrong_update(self):
        tor = FakeTorController()
        circuit = Circuit(tor)
        circuit.listen(tor)
        circuit.update('1 LAUNCHED PURPOSE=GENERAL'.split())
        self.assertRaises(
            Exception,
            circuit.update,
            '2 LAUNCHED PURPOSE=GENERAL'.split()
        )