diff options
Diffstat (limited to 'lib/python2.7/site-packages/Twisted-12.2.0-py2.7-linux-x86_64.egg/twisted/internet/test/test_protocol.py')
| -rwxr-xr-x | lib/python2.7/site-packages/Twisted-12.2.0-py2.7-linux-x86_64.egg/twisted/internet/test/test_protocol.py | 357 |
1 files changed, 0 insertions, 357 deletions
diff --git a/lib/python2.7/site-packages/Twisted-12.2.0-py2.7-linux-x86_64.egg/twisted/internet/test/test_protocol.py b/lib/python2.7/site-packages/Twisted-12.2.0-py2.7-linux-x86_64.egg/twisted/internet/test/test_protocol.py deleted file mode 100755 index 18ef114c..00000000 --- a/lib/python2.7/site-packages/Twisted-12.2.0-py2.7-linux-x86_64.egg/twisted/internet/test/test_protocol.py +++ /dev/null @@ -1,357 +0,0 @@ -# Copyright (c) Twisted Matrix Laboratories. -# See LICENSE for details. - -""" -Tests for L{twisted.internet.protocol}. -""" - -from zope.interface.verify import verifyObject - -from twisted.python.failure import Failure -from twisted.internet.interfaces import IProtocol, ILoggingContext -from twisted.internet.defer import CancelledError -from twisted.internet.protocol import Protocol, ClientCreator -from twisted.internet.task import Clock -from twisted.trial.unittest import TestCase -from twisted.test.proto_helpers import MemoryReactor, StringTransport - - - -class MemoryConnector: - _disconnected = False - - def disconnect(self): - self._disconnected = True - - - -class MemoryReactorWithConnectorsAndTime(MemoryReactor, Clock): - """ - An extension of L{MemoryReactor} which returns L{IConnector} - providers from its C{connectTCP} method. - """ - def __init__(self): - MemoryReactor.__init__(self) - Clock.__init__(self) - self.connectors = [] - - - def connectTCP(self, *a, **kw): - MemoryReactor.connectTCP(self, *a, **kw) - connector = MemoryConnector() - self.connectors.append(connector) - return connector - - - def connectUNIX(self, *a, **kw): - MemoryReactor.connectUNIX(self, *a, **kw) - connector = MemoryConnector() - self.connectors.append(connector) - return connector - - - def connectSSL(self, *a, **kw): - MemoryReactor.connectSSL(self, *a, **kw) - connector = MemoryConnector() - self.connectors.append(connector) - return connector - - - -class ClientCreatorTests(TestCase): - """ - Tests for L{twisted.internet.protocol.ClientCreator}. - """ - def _basicConnectTest(self, check): - """ - Helper for implementing a test to verify that one of the I{connect} - methods of L{ClientCreator} passes the right arguments to the right - reactor method. - - @param check: A function which will be invoked with a reactor and a - L{ClientCreator} instance and which should call one of the - L{ClientCreator}'s I{connect} methods and assert that all of its - arguments except for the factory are passed on as expected to the - reactor. The factory should be returned. - """ - class SomeProtocol(Protocol): - pass - - reactor = MemoryReactorWithConnectorsAndTime() - cc = ClientCreator(reactor, SomeProtocol) - factory = check(reactor, cc) - protocol = factory.buildProtocol(None) - self.assertIsInstance(protocol, SomeProtocol) - - - def test_connectTCP(self): - """ - L{ClientCreator.connectTCP} calls C{reactor.connectTCP} with the host - and port information passed to it, and with a factory which will - construct the protocol passed to L{ClientCreator.__init__}. - """ - def check(reactor, cc): - cc.connectTCP('example.com', 1234, 4321, ('1.2.3.4', 9876)) - host, port, factory, timeout, bindAddress = reactor.tcpClients.pop() - self.assertEqual(host, 'example.com') - self.assertEqual(port, 1234) - self.assertEqual(timeout, 4321) - self.assertEqual(bindAddress, ('1.2.3.4', 9876)) - return factory - self._basicConnectTest(check) - - - def test_connectUNIX(self): - """ - L{ClientCreator.connectUNIX} calls C{reactor.connectUNIX} with the - filename passed to it, and with a factory which will construct the - protocol passed to L{ClientCreator.__init__}. - """ - def check(reactor, cc): - cc.connectUNIX('/foo/bar', 123, True) - address, factory, timeout, checkPID = reactor.unixClients.pop() - self.assertEqual(address, '/foo/bar') - self.assertEqual(timeout, 123) - self.assertEqual(checkPID, True) - return factory - self._basicConnectTest(check) - - - def test_connectSSL(self): - """ - L{ClientCreator.connectSSL} calls C{reactor.connectSSL} with the host, - port, and context factory passed to it, and with a factory which will - construct the protocol passed to L{ClientCreator.__init__}. - """ - def check(reactor, cc): - expectedContextFactory = object() - cc.connectSSL('example.com', 1234, expectedContextFactory, 4321, ('4.3.2.1', 5678)) - host, port, factory, contextFactory, timeout, bindAddress = reactor.sslClients.pop() - self.assertEqual(host, 'example.com') - self.assertEqual(port, 1234) - self.assertIdentical(contextFactory, expectedContextFactory) - self.assertEqual(timeout, 4321) - self.assertEqual(bindAddress, ('4.3.2.1', 5678)) - return factory - self._basicConnectTest(check) - - - def _cancelConnectTest(self, connect): - """ - Helper for implementing a test to verify that cancellation of the - L{Deferred} returned by one of L{ClientCreator}'s I{connect} methods is - implemented to cancel the underlying connector. - - @param connect: A function which will be invoked with a L{ClientCreator} - instance as an argument and which should call one its I{connect} - methods and return the result. - - @return: A L{Deferred} which fires when the test is complete or fails if - there is a problem. - """ - reactor = MemoryReactorWithConnectorsAndTime() - cc = ClientCreator(reactor, Protocol) - d = connect(cc) - connector = reactor.connectors.pop() - self.assertFalse(connector._disconnected) - d.cancel() - self.assertTrue(connector._disconnected) - return self.assertFailure(d, CancelledError) - - - def test_cancelConnectTCP(self): - """ - The L{Deferred} returned by L{ClientCreator.connectTCP} can be cancelled - to abort the connection attempt before it completes. - """ - def connect(cc): - return cc.connectTCP('example.com', 1234) - return self._cancelConnectTest(connect) - - - def test_cancelConnectUNIX(self): - """ - The L{Deferred} returned by L{ClientCreator.connectTCP} can be cancelled - to abort the connection attempt before it completes. - """ - def connect(cc): - return cc.connectUNIX('/foo/bar') - return self._cancelConnectTest(connect) - - - def test_cancelConnectSSL(self): - """ - The L{Deferred} returned by L{ClientCreator.connectTCP} can be cancelled - to abort the connection attempt before it completes. - """ - def connect(cc): - return cc.connectSSL('example.com', 1234, object()) - return self._cancelConnectTest(connect) - - - def _cancelConnectTimeoutTest(self, connect): - """ - Like L{_cancelConnectTest}, but for the case where the L{Deferred} is - cancelled after the connection is set up but before it is fired with the - resulting protocol instance. - """ - reactor = MemoryReactorWithConnectorsAndTime() - cc = ClientCreator(reactor, Protocol) - d = connect(reactor, cc) - connector = reactor.connectors.pop() - # Sanity check - there is an outstanding delayed call to fire the - # Deferred. - self.assertEqual(len(reactor.getDelayedCalls()), 1) - - # Cancel the Deferred, disconnecting the transport just set up and - # cancelling the delayed call. - d.cancel() - - self.assertEqual(reactor.getDelayedCalls(), []) - - # A real connector implementation is responsible for disconnecting the - # transport as well. For our purposes, just check that someone told the - # connector to disconnect. - self.assertTrue(connector._disconnected) - - return self.assertFailure(d, CancelledError) - - - def test_cancelConnectTCPTimeout(self): - """ - L{ClientCreator.connectTCP} inserts a very short delayed call between - the time the connection is established and the time the L{Deferred} - returned from one of its connect methods actually fires. If the - L{Deferred} is cancelled in this interval, the established connection is - closed, the timeout is cancelled, and the L{Deferred} fails with - L{CancelledError}. - """ - def connect(reactor, cc): - d = cc.connectTCP('example.com', 1234) - host, port, factory, timeout, bindAddress = reactor.tcpClients.pop() - protocol = factory.buildProtocol(None) - transport = StringTransport() - protocol.makeConnection(transport) - return d - return self._cancelConnectTimeoutTest(connect) - - - def test_cancelConnectUNIXTimeout(self): - """ - L{ClientCreator.connectUNIX} inserts a very short delayed call between - the time the connection is established and the time the L{Deferred} - returned from one of its connect methods actually fires. If the - L{Deferred} is cancelled in this interval, the established connection is - closed, the timeout is cancelled, and the L{Deferred} fails with - L{CancelledError}. - """ - def connect(reactor, cc): - d = cc.connectUNIX('/foo/bar') - address, factory, timeout, bindAddress = reactor.unixClients.pop() - protocol = factory.buildProtocol(None) - transport = StringTransport() - protocol.makeConnection(transport) - return d - return self._cancelConnectTimeoutTest(connect) - - - def test_cancelConnectSSLTimeout(self): - """ - L{ClientCreator.connectSSL} inserts a very short delayed call between - the time the connection is established and the time the L{Deferred} - returned from one of its connect methods actually fires. If the - L{Deferred} is cancelled in this interval, the established connection is - closed, the timeout is cancelled, and the L{Deferred} fails with - L{CancelledError}. - """ - def connect(reactor, cc): - d = cc.connectSSL('example.com', 1234, object()) - host, port, factory, contextFactory, timeout, bindADdress = reactor.sslClients.pop() - protocol = factory.buildProtocol(None) - transport = StringTransport() - protocol.makeConnection(transport) - return d - return self._cancelConnectTimeoutTest(connect) - - - def _cancelConnectFailedTimeoutTest(self, connect): - """ - Like L{_cancelConnectTest}, but for the case where the L{Deferred} is - cancelled after the connection attempt has failed but before it is fired - with the resulting failure. - """ - reactor = MemoryReactorWithConnectorsAndTime() - cc = ClientCreator(reactor, Protocol) - d, factory = connect(reactor, cc) - connector = reactor.connectors.pop() - factory.clientConnectionFailed( - connector, Failure(Exception("Simulated failure"))) - - # Sanity check - there is an outstanding delayed call to fire the - # Deferred. - self.assertEqual(len(reactor.getDelayedCalls()), 1) - - # Cancel the Deferred, cancelling the delayed call. - d.cancel() - - self.assertEqual(reactor.getDelayedCalls(), []) - - return self.assertFailure(d, CancelledError) - - - def test_cancelConnectTCPFailedTimeout(self): - """ - Similar to L{test_cancelConnectTCPTimeout}, but for the case where the - connection attempt fails. - """ - def connect(reactor, cc): - d = cc.connectTCP('example.com', 1234) - host, port, factory, timeout, bindAddress = reactor.tcpClients.pop() - return d, factory - return self._cancelConnectFailedTimeoutTest(connect) - - - def test_cancelConnectUNIXFailedTimeout(self): - """ - Similar to L{test_cancelConnectUNIXTimeout}, but for the case where the - connection attempt fails. - """ - def connect(reactor, cc): - d = cc.connectUNIX('/foo/bar') - address, factory, timeout, bindAddress = reactor.unixClients.pop() - return d, factory - return self._cancelConnectFailedTimeoutTest(connect) - - - def test_cancelConnectSSLFailedTimeout(self): - """ - Similar to L{test_cancelConnectSSLTimeout}, but for the case where the - connection attempt fails. - """ - def connect(reactor, cc): - d = cc.connectSSL('example.com', 1234, object()) - host, port, factory, contextFactory, timeout, bindADdress = reactor.sslClients.pop() - return d, factory - return self._cancelConnectFailedTimeoutTest(connect) - - -class ProtocolTests(TestCase): - """ - Tests for L{twisted.internet.protocol.Protocol}. - """ - def test_interfaces(self): - """ - L{Protocol} instances provide L{IProtocol} and L{ILoggingContext}. - """ - proto = Protocol() - self.assertTrue(verifyObject(IProtocol, proto)) - self.assertTrue(verifyObject(ILoggingContext, proto)) - - - def test_logPrefix(self): - """ - L{Protocol.logPrefix} returns the protocol class's name. - """ - class SomeThing(Protocol): - pass - self.assertEqual("SomeThing", SomeThing().logPrefix()) |