Gecko [ gandhischool.edu.np ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
__init__.py	475 B	-rw-r--r--
cert.pem.no_trailing_newline	1.38 KB	-rw-r--r--
crash_test_dummy.py	549 B	-rw-r--r--
ignore_test_failure.py	30.72 KB	-rw-r--r--
iosim.py	17.74 KB	-rw-r--r--
key.pem.no_trailing_newline	1.67 KB	-rw-r--r--
mock_win32process.py	1.27 KB	-rw-r--r--
myrebuilder1.py	151 B	-rw-r--r--
myrebuilder2.py	151 B	-rw-r--r--
plugin_basic.py	901 B	-rw-r--r--
plugin_extra1.py	392 B	-rw-r--r--
plugin_extra2.py	550 B	-rw-r--r--
process_cmdline.py	123 B	-rw-r--r--
process_echoer.py	214 B	-rw-r--r--
process_fds.py	984 B	-rw-r--r--
process_getargv.py	234 B	-rw-r--r--
process_getenv.py	268 B	-rw-r--r--
process_linger.py	297 B	-rw-r--r--
process_reader.py	178 B	-rw-r--r--
process_signal.py	220 B	-rw-r--r--
process_stdinreader.py	739 B	-rw-r--r--
process_tester.py	779 B	-rw-r--r--
process_tty.py	130 B	-rw-r--r--
process_twisted.py	1.15 KB	-rw-r--r--
proto_helpers.py	1.34 KB	-rw-r--r--
reflect_helper_IE.py	60 B	-rw-r--r--
reflect_helper_VE.py	81 B	-rw-r--r--
reflect_helper_ZDE.py	48 B	-rw-r--r--
server.pem	4.34 KB	-rw-r--r--
ssl_helpers.py	1.61 KB	-rw-r--r--
stdio_test_consumer.py	1.14 KB	-rw-r--r--
stdio_test_halfclose.py	2 KB	-rw-r--r--
stdio_test_hostpeer.py	1.06 KB	-rw-r--r--
stdio_test_lastwrite.py	1.13 KB	-rw-r--r--
stdio_test_loseconn.py	1.55 KB	-rw-r--r--
stdio_test_producer.py	1.45 KB	-rw-r--r--
stdio_test_write.py	902 B	-rw-r--r--
stdio_test_writeseq.py	894 B	-rw-r--r--
test_abstract.py	3.35 KB	-rw-r--r--
test_adbapi.py	25.56 KB	-rw-r--r--
test_amp.py	107.92 KB	-rw-r--r--
test_application.py	33.34 KB	-rw-r--r--
test_compat.py	17.75 KB	-rw-r--r--
test_context.py	1.43 KB	-rw-r--r--
test_cooperator.py	20.84 KB	-rw-r--r--
test_defer.py	121.86 KB	-rw-r--r--
test_defgen.py	10.38 KB	-rw-r--r--
test_dict.py	1.49 KB	-rw-r--r--
test_dirdbm.py	6.69 KB	-rw-r--r--
test_error.py	8.72 KB	-rw-r--r--
test_factories.py	4.46 KB	-rw-r--r--
test_fdesc.py	7.28 KB	-rw-r--r--
test_finger.py	1.85 KB	-rw-r--r--
test_formmethod.py	3.87 KB	-rw-r--r--
test_ftp.py	126.96 KB	-rw-r--r--
test_ftp_options.py	2.63 KB	-rw-r--r--
test_htb.py	3.08 KB	-rw-r--r--
test_ident.py	6.56 KB	-rw-r--r--
test_internet.py	45.38 KB	-rw-r--r--
test_iosim.py	8.7 KB	-rw-r--r--
test_iutils.py	13.31 KB	-rw-r--r--
test_lockfile.py	14.98 KB	-rw-r--r--
test_log.py	35.23 KB	-rw-r--r--
test_logfile.py	17.47 KB	-rw-r--r--
test_loopback.py	13.99 KB	-rw-r--r--
test_main.py	2.12 KB	-rw-r--r--
test_memcache.py	24.69 KB	-rw-r--r--
test_modules.py	17.02 KB	-rw-r--r--
test_monkey.py	5.39 KB	-rw-r--r--
test_paths.py	69.31 KB	-rw-r--r--
test_pcp.py	12.23 KB	-rw-r--r--
test_persisted.py	13.09 KB	-rw-r--r--
test_plugin.py	24.79 KB	-rw-r--r--
test_policies.py	32.29 KB	-rw-r--r--
test_postfix.py	4.32 KB	-rw-r--r--
test_process.py	85.16 KB	-rw-r--r--
test_protocols.py	7.16 KB	-rw-r--r--
test_randbytes.py	3.19 KB	-rw-r--r--
test_rebuild.py	7.15 KB	-rw-r--r--
test_reflect.py	23.9 KB	-rw-r--r--
test_roots.py	1.6 KB	-rw-r--r--
test_shortcut.py	1.92 KB	-rw-r--r--
test_sip.py	24.9 KB	-rw-r--r--
test_sob.py	5.53 KB	-rw-r--r--
test_socks.py	17.09 KB	-rw-r--r--
test_ssl.py	22.75 KB	-rw-r--r--
test_sslverify.py	113.89 KB	-rw-r--r--
test_stateful.py	1.97 KB	-rw-r--r--
test_stdio.py	12.43 KB	-rw-r--r--
test_strerror.py	5.12 KB	-rw-r--r--
test_strports.py	1.67 KB	-rw-r--r--
test_task.py	45.26 KB	-rw-r--r--
test_tcp.py	64.27 KB	-rw-r--r--
test_tcp_internals.py	12.73 KB	-rw-r--r--
test_text.py	6.22 KB	-rw-r--r--
test_threadable.py	3.26 KB	-rw-r--r--
test_threadpool.py	21.44 KB	-rw-r--r--
test_threads.py	12.9 KB	-rw-r--r--
test_tpfile.py	1.51 KB	-rw-r--r--
test_twistd.py	72.04 KB	-rw-r--r--
test_twisted.py	6.13 KB	-rw-r--r--
test_udp.py	24.43 KB	-rw-r--r--
test_unix.py	13.26 KB	-rw-r--r--
test_usage.py	22.76 KB	-rw-r--r--
testutils.py	5.06 KB	-rw-r--r--

Code Editor : test_protocols.py

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Test cases for twisted.protocols package.
"""

from twisted.internet import address, defer, protocol, reactor
from twisted.protocols import portforward, wire
from twisted.python.compat import iterbytes
from twisted.test import proto_helpers
from twisted.trial import unittest

class WireTests(unittest.TestCase):
    """
    Test wire protocols.
    """

def test_echo(self):
        """
        Test wire.Echo protocol: send some data and check it send it back.
        """
        t = proto_helpers.StringTransport()
        a = wire.Echo()
        a.makeConnection(t)
        a.dataReceived(b"hello")
        a.dataReceived(b"world")
        a.dataReceived(b"how")
        a.dataReceived(b"are")
        a.dataReceived(b"you")
        self.assertEqual(t.value(), b"helloworldhowareyou")

def test_who(self):
        """
        Test wire.Who protocol.
        """
        t = proto_helpers.StringTransport()
        a = wire.Who()
        a.makeConnection(t)
        self.assertEqual(t.value(), b"root\r\n")

def test_QOTD(self):
        """
        Test wire.QOTD protocol.
        """
        t = proto_helpers.StringTransport()
        a = wire.QOTD()
        a.makeConnection(t)
        self.assertEqual(t.value(), b"An apple a day keeps the doctor away.\r\n")

def test_discard(self):
        """
        Test wire.Discard protocol.
        """
        t = proto_helpers.StringTransport()
        a = wire.Discard()
        a.makeConnection(t)
        a.dataReceived(b"hello")
        a.dataReceived(b"world")
        a.dataReceived(b"how")
        a.dataReceived(b"are")
        a.dataReceived(b"you")
        self.assertEqual(t.value(), b"")

class TestableProxyClientFactory(portforward.ProxyClientFactory):
    """
    Test proxy client factory that keeps the last created protocol instance.

@ivar protoInstance: the last instance of the protocol.
    @type protoInstance: L{portforward.ProxyClient}
    """

def buildProtocol(self, addr):
        """
        Create the protocol instance and keeps track of it.
        """
        proto = portforward.ProxyClientFactory.buildProtocol(self, addr)
        self.protoInstance = proto
        return proto

class TestableProxyFactory(portforward.ProxyFactory):
    """
    Test proxy factory that keeps the last created protocol instance.

@ivar protoInstance: the last instance of the protocol.
    @type protoInstance: L{portforward.ProxyServer}

@ivar clientFactoryInstance: client factory used by C{protoInstance} to
        create forward connections.
    @type clientFactoryInstance: L{TestableProxyClientFactory}
    """

def buildProtocol(self, addr):
        """
        Create the protocol instance, keeps track of it, and makes it use
        C{clientFactoryInstance} as client factory.
        """
        proto = portforward.ProxyFactory.buildProtocol(self, addr)
        self.clientFactoryInstance = TestableProxyClientFactory()
        # Force the use of this specific instance
        proto.clientProtocolFactory = lambda: self.clientFactoryInstance
        self.protoInstance = proto
        return proto

class PortforwardingTests(unittest.TestCase):
    """
    Test port forwarding.
    """

def setUp(self):
        self.serverProtocol = wire.Echo()
        self.clientProtocol = protocol.Protocol()
        self.openPorts = []

def tearDown(self):
        try:
            self.proxyServerFactory.protoInstance.transport.loseConnection()
        except AttributeError:
            pass
        try:
            pi = self.proxyServerFactory.clientFactoryInstance.protoInstance
            pi.transport.loseConnection()
        except AttributeError:
            pass
        try:
            self.clientProtocol.transport.loseConnection()
        except AttributeError:
            pass
        try:
            self.serverProtocol.transport.loseConnection()
        except AttributeError:
            pass
        return defer.gatherResults(
            [defer.maybeDeferred(p.stopListening) for p in self.openPorts]
        )

def test_portforward(self):
        """
        Test port forwarding through Echo protocol.
        """
        realServerFactory = protocol.ServerFactory()
        realServerFactory.protocol = lambda: self.serverProtocol
        realServerPort = reactor.listenTCP(0, realServerFactory, interface="127.0.0.1")
        self.openPorts.append(realServerPort)
        self.proxyServerFactory = TestableProxyFactory(
            "127.0.0.1", realServerPort.getHost().port
        )
        proxyServerPort = reactor.listenTCP(
            0, self.proxyServerFactory, interface="127.0.0.1"
        )
        self.openPorts.append(proxyServerPort)

nBytes = 1000
        received = []
        d = defer.Deferred()

def testDataReceived(data):
            received.extend(iterbytes(data))
            if len(received) >= nBytes:
                self.assertEqual(b"".join(received), b"x" * nBytes)
                d.callback(None)

self.clientProtocol.dataReceived = testDataReceived

def testConnectionMade():
            self.clientProtocol.transport.write(b"x" * nBytes)

self.clientProtocol.connectionMade = testConnectionMade

clientFactory = protocol.ClientFactory()
        clientFactory.protocol = lambda: self.clientProtocol

reactor.connectTCP("127.0.0.1", proxyServerPort.getHost().port, clientFactory)

return d

def test_registerProducers(self):
        """
        The proxy client registers itself as a producer of the proxy server and
        vice versa.
        """
        # create a ProxyServer instance
        addr = address.IPv4Address("TCP", "127.0.0.1", 0)
        server = portforward.ProxyFactory("127.0.0.1", 0).buildProtocol(addr)

# set the reactor for this test
        reactor = proto_helpers.MemoryReactor()
        server.reactor = reactor

# make the connection
        serverTransport = proto_helpers.StringTransport()
        server.makeConnection(serverTransport)

# check that the ProxyClientFactory is connecting to the backend
        self.assertEqual(len(reactor.tcpClients), 1)
        # get the factory instance and check it's the one we expect
        host, port, clientFactory, timeout, _ = reactor.tcpClients[0]
        self.assertIsInstance(clientFactory, portforward.ProxyClientFactory)

# Connect it
        client = clientFactory.buildProtocol(addr)
        clientTransport = proto_helpers.StringTransport()
        client.makeConnection(clientTransport)

# check that the producers are registered
        self.assertIs(clientTransport.producer, serverTransport)
        self.assertIs(serverTransport.producer, clientTransport)
        # check the streaming attribute in both transports
        self.assertTrue(clientTransport.streaming)
        self.assertTrue(serverTransport.streaming)

class StringTransportTests(unittest.TestCase):
    """
    Test L{proto_helpers.StringTransport} helper behaviour.
    """

def test_noUnicode(self):
        """
        Test that L{proto_helpers.StringTransport} doesn't accept unicode data.
        """
        s = proto_helpers.StringTransport()
        self.assertRaises(TypeError, s.write, "foo")

${this.title}

Code Editor : test_protocols.py