diff options
Diffstat (limited to 'lib/python2.7/site-packages/Twisted-12.2.0-py2.7-linux-x86_64.egg/twisted/conch/test/test_connection.py')
-rwxr-xr-x | lib/python2.7/site-packages/Twisted-12.2.0-py2.7-linux-x86_64.egg/twisted/conch/test/test_connection.py | 730 |
1 files changed, 0 insertions, 730 deletions
diff --git a/lib/python2.7/site-packages/Twisted-12.2.0-py2.7-linux-x86_64.egg/twisted/conch/test/test_connection.py b/lib/python2.7/site-packages/Twisted-12.2.0-py2.7-linux-x86_64.egg/twisted/conch/test/test_connection.py deleted file mode 100755 index 85a8e6a7..00000000 --- a/lib/python2.7/site-packages/Twisted-12.2.0-py2.7-linux-x86_64.egg/twisted/conch/test/test_connection.py +++ /dev/null @@ -1,730 +0,0 @@ -# Copyright (c) 2007-2010 Twisted Matrix Laboratories. -# See LICENSE for details - -""" -This module tests twisted.conch.ssh.connection. -""" - -import struct - -from twisted.conch import error -from twisted.conch.ssh import channel, common, connection -from twisted.trial import unittest -from twisted.conch.test import test_userauth - - -class TestChannel(channel.SSHChannel): - """ - A mocked-up version of twisted.conch.ssh.channel.SSHChannel. - - @ivar gotOpen: True if channelOpen has been called. - @type gotOpen: C{bool} - @ivar specificData: the specific channel open data passed to channelOpen. - @type specificData: C{str} - @ivar openFailureReason: the reason passed to openFailed. - @type openFailed: C{error.ConchError} - @ivar inBuffer: a C{list} of strings received by the channel. - @type inBuffer: C{list} - @ivar extBuffer: a C{list} of 2-tuples (type, extended data) of received by - the channel. - @type extBuffer: C{list} - @ivar numberRequests: the number of requests that have been made to this - channel. - @type numberRequests: C{int} - @ivar gotEOF: True if the other side sent EOF. - @type gotEOF: C{bool} - @ivar gotOneClose: True if the other side closed the connection. - @type gotOneClose: C{bool} - @ivar gotClosed: True if the channel is closed. - @type gotClosed: C{bool} - """ - name = "TestChannel" - gotOpen = False - - def logPrefix(self): - return "TestChannel %i" % self.id - - def channelOpen(self, specificData): - """ - The channel is open. Set up the instance variables. - """ - self.gotOpen = True - self.specificData = specificData - self.inBuffer = [] - self.extBuffer = [] - self.numberRequests = 0 - self.gotEOF = False - self.gotOneClose = False - self.gotClosed = False - - def openFailed(self, reason): - """ - Opening the channel failed. Store the reason why. - """ - self.openFailureReason = reason - - def request_test(self, data): - """ - A test request. Return True if data is 'data'. - - @type data: C{str} - """ - self.numberRequests += 1 - return data == 'data' - - def dataReceived(self, data): - """ - Data was received. Store it in the buffer. - """ - self.inBuffer.append(data) - - def extReceived(self, code, data): - """ - Extended data was received. Store it in the buffer. - """ - self.extBuffer.append((code, data)) - - def eofReceived(self): - """ - EOF was received. Remember it. - """ - self.gotEOF = True - - def closeReceived(self): - """ - Close was received. Remember it. - """ - self.gotOneClose = True - - def closed(self): - """ - The channel is closed. Rembember it. - """ - self.gotClosed = True - -class TestAvatar: - """ - A mocked-up version of twisted.conch.avatar.ConchUser - """ - _ARGS_ERROR_CODE = 123 - - def lookupChannel(self, channelType, windowSize, maxPacket, data): - """ - The server wants us to return a channel. If the requested channel is - our TestChannel, return it, otherwise return None. - """ - if channelType == TestChannel.name: - return TestChannel(remoteWindow=windowSize, - remoteMaxPacket=maxPacket, - data=data, avatar=self) - elif channelType == "conch-error-args": - # Raise a ConchError with backwards arguments to make sure the - # connection fixes it for us. This case should be deprecated and - # deleted eventually, but only after all of Conch gets the argument - # order right. - raise error.ConchError( - self._ARGS_ERROR_CODE, "error args in wrong order") - - - def gotGlobalRequest(self, requestType, data): - """ - The client has made a global request. If the global request is - 'TestGlobal', return True. If the global request is 'TestData', - return True and the request-specific data we received. Otherwise, - return False. - """ - if requestType == 'TestGlobal': - return True - elif requestType == 'TestData': - return True, data - else: - return False - - - -class TestConnection(connection.SSHConnection): - """ - A subclass of SSHConnection for testing. - - @ivar channel: the current channel. - @type channel. C{TestChannel} - """ - - def logPrefix(self): - return "TestConnection" - - def global_TestGlobal(self, data): - """ - The other side made the 'TestGlobal' global request. Return True. - """ - return True - - def global_Test_Data(self, data): - """ - The other side made the 'Test-Data' global request. Return True and - the data we received. - """ - return True, data - - def channel_TestChannel(self, windowSize, maxPacket, data): - """ - The other side is requesting the TestChannel. Create a C{TestChannel} - instance, store it, and return it. - """ - self.channel = TestChannel(remoteWindow=windowSize, - remoteMaxPacket=maxPacket, data=data) - return self.channel - - def channel_ErrorChannel(self, windowSize, maxPacket, data): - """ - The other side is requesting the ErrorChannel. Raise an exception. - """ - raise AssertionError('no such thing') - - - -class ConnectionTestCase(unittest.TestCase): - - if test_userauth.transport is None: - skip = "Cannot run without both PyCrypto and pyasn1" - - def setUp(self): - self.transport = test_userauth.FakeTransport(None) - self.transport.avatar = TestAvatar() - self.conn = TestConnection() - self.conn.transport = self.transport - self.conn.serviceStarted() - - def _openChannel(self, channel): - """ - Open the channel with the default connection. - """ - self.conn.openChannel(channel) - self.transport.packets = self.transport.packets[:-1] - self.conn.ssh_CHANNEL_OPEN_CONFIRMATION(struct.pack('>2L', - channel.id, 255) + '\x00\x02\x00\x00\x00\x00\x80\x00') - - def tearDown(self): - self.conn.serviceStopped() - - def test_linkAvatar(self): - """ - Test that the connection links itself to the avatar in the - transport. - """ - self.assertIdentical(self.transport.avatar.conn, self.conn) - - def test_serviceStopped(self): - """ - Test that serviceStopped() closes any open channels. - """ - channel1 = TestChannel() - channel2 = TestChannel() - self.conn.openChannel(channel1) - self.conn.openChannel(channel2) - self.conn.ssh_CHANNEL_OPEN_CONFIRMATION('\x00\x00\x00\x00' * 4) - self.assertTrue(channel1.gotOpen) - self.assertFalse(channel2.gotOpen) - self.conn.serviceStopped() - self.assertTrue(channel1.gotClosed) - - def test_GLOBAL_REQUEST(self): - """ - Test that global request packets are dispatched to the global_* - methods and the return values are translated into success or failure - messages. - """ - self.conn.ssh_GLOBAL_REQUEST(common.NS('TestGlobal') + '\xff') - self.assertEqual(self.transport.packets, - [(connection.MSG_REQUEST_SUCCESS, '')]) - self.transport.packets = [] - self.conn.ssh_GLOBAL_REQUEST(common.NS('TestData') + '\xff' + - 'test data') - self.assertEqual(self.transport.packets, - [(connection.MSG_REQUEST_SUCCESS, 'test data')]) - self.transport.packets = [] - self.conn.ssh_GLOBAL_REQUEST(common.NS('TestBad') + '\xff') - self.assertEqual(self.transport.packets, - [(connection.MSG_REQUEST_FAILURE, '')]) - self.transport.packets = [] - self.conn.ssh_GLOBAL_REQUEST(common.NS('TestGlobal') + '\x00') - self.assertEqual(self.transport.packets, []) - - def test_REQUEST_SUCCESS(self): - """ - Test that global request success packets cause the Deferred to be - called back. - """ - d = self.conn.sendGlobalRequest('request', 'data', True) - self.conn.ssh_REQUEST_SUCCESS('data') - def check(data): - self.assertEqual(data, 'data') - d.addCallback(check) - d.addErrback(self.fail) - return d - - def test_REQUEST_FAILURE(self): - """ - Test that global request failure packets cause the Deferred to be - erred back. - """ - d = self.conn.sendGlobalRequest('request', 'data', True) - self.conn.ssh_REQUEST_FAILURE('data') - def check(f): - self.assertEqual(f.value.data, 'data') - d.addCallback(self.fail) - d.addErrback(check) - return d - - def test_CHANNEL_OPEN(self): - """ - Test that open channel packets cause a channel to be created and - opened or a failure message to be returned. - """ - del self.transport.avatar - self.conn.ssh_CHANNEL_OPEN(common.NS('TestChannel') + - '\x00\x00\x00\x01' * 4) - self.assertTrue(self.conn.channel.gotOpen) - self.assertEqual(self.conn.channel.conn, self.conn) - self.assertEqual(self.conn.channel.data, '\x00\x00\x00\x01') - self.assertEqual(self.conn.channel.specificData, '\x00\x00\x00\x01') - self.assertEqual(self.conn.channel.remoteWindowLeft, 1) - self.assertEqual(self.conn.channel.remoteMaxPacket, 1) - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_OPEN_CONFIRMATION, - '\x00\x00\x00\x01\x00\x00\x00\x00\x00\x02\x00\x00' - '\x00\x00\x80\x00')]) - self.transport.packets = [] - self.conn.ssh_CHANNEL_OPEN(common.NS('BadChannel') + - '\x00\x00\x00\x02' * 4) - self.flushLoggedErrors() - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_OPEN_FAILURE, - '\x00\x00\x00\x02\x00\x00\x00\x03' + common.NS( - 'unknown channel') + common.NS(''))]) - self.transport.packets = [] - self.conn.ssh_CHANNEL_OPEN(common.NS('ErrorChannel') + - '\x00\x00\x00\x02' * 4) - self.flushLoggedErrors() - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_OPEN_FAILURE, - '\x00\x00\x00\x02\x00\x00\x00\x02' + common.NS( - 'unknown failure') + common.NS(''))]) - - - def _lookupChannelErrorTest(self, code): - """ - Deliver a request for a channel open which will result in an exception - being raised during channel lookup. Assert that an error response is - delivered as a result. - """ - self.transport.avatar._ARGS_ERROR_CODE = code - self.conn.ssh_CHANNEL_OPEN( - common.NS('conch-error-args') + '\x00\x00\x00\x01' * 4) - errors = self.flushLoggedErrors(error.ConchError) - self.assertEqual( - len(errors), 1, "Expected one error, got: %r" % (errors,)) - self.assertEqual(errors[0].value.args, (123, "error args in wrong order")) - self.assertEqual( - self.transport.packets, - [(connection.MSG_CHANNEL_OPEN_FAILURE, - # The response includes some bytes which identifying the - # associated request, as well as the error code (7b in hex) and - # the error message. - '\x00\x00\x00\x01\x00\x00\x00\x7b' + common.NS( - 'error args in wrong order') + common.NS(''))]) - - - def test_lookupChannelError(self): - """ - If a C{lookupChannel} implementation raises L{error.ConchError} with the - arguments in the wrong order, a C{MSG_CHANNEL_OPEN} failure is still - sent in response to the message. - - This is a temporary work-around until L{error.ConchError} is given - better attributes and all of the Conch code starts constructing - instances of it properly. Eventually this functionality should be - deprecated and then removed. - """ - self._lookupChannelErrorTest(123) - - - def test_lookupChannelErrorLongCode(self): - """ - Like L{test_lookupChannelError}, but for the case where the failure code - is represented as a C{long} instead of a C{int}. - """ - self._lookupChannelErrorTest(123L) - - - def test_CHANNEL_OPEN_CONFIRMATION(self): - """ - Test that channel open confirmation packets cause the channel to be - notified that it's open. - """ - channel = TestChannel() - self.conn.openChannel(channel) - self.conn.ssh_CHANNEL_OPEN_CONFIRMATION('\x00\x00\x00\x00'*5) - self.assertEqual(channel.remoteWindowLeft, 0) - self.assertEqual(channel.remoteMaxPacket, 0) - self.assertEqual(channel.specificData, '\x00\x00\x00\x00') - self.assertEqual(self.conn.channelsToRemoteChannel[channel], - 0) - self.assertEqual(self.conn.localToRemoteChannel[0], 0) - - def test_CHANNEL_OPEN_FAILURE(self): - """ - Test that channel open failure packets cause the channel to be - notified that its opening failed. - """ - channel = TestChannel() - self.conn.openChannel(channel) - self.conn.ssh_CHANNEL_OPEN_FAILURE('\x00\x00\x00\x00\x00\x00\x00' - '\x01' + common.NS('failure!')) - self.assertEqual(channel.openFailureReason.args, ('failure!', 1)) - self.assertEqual(self.conn.channels.get(channel), None) - - - def test_CHANNEL_WINDOW_ADJUST(self): - """ - Test that channel window adjust messages add bytes to the channel - window. - """ - channel = TestChannel() - self._openChannel(channel) - oldWindowSize = channel.remoteWindowLeft - self.conn.ssh_CHANNEL_WINDOW_ADJUST('\x00\x00\x00\x00\x00\x00\x00' - '\x01') - self.assertEqual(channel.remoteWindowLeft, oldWindowSize + 1) - - def test_CHANNEL_DATA(self): - """ - Test that channel data messages are passed up to the channel, or - cause the channel to be closed if the data is too large. - """ - channel = TestChannel(localWindow=6, localMaxPacket=5) - self._openChannel(channel) - self.conn.ssh_CHANNEL_DATA('\x00\x00\x00\x00' + common.NS('data')) - self.assertEqual(channel.inBuffer, ['data']) - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_WINDOW_ADJUST, '\x00\x00\x00\xff' - '\x00\x00\x00\x04')]) - self.transport.packets = [] - longData = 'a' * (channel.localWindowLeft + 1) - self.conn.ssh_CHANNEL_DATA('\x00\x00\x00\x00' + common.NS(longData)) - self.assertEqual(channel.inBuffer, ['data']) - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_CLOSE, '\x00\x00\x00\xff')]) - channel = TestChannel() - self._openChannel(channel) - bigData = 'a' * (channel.localMaxPacket + 1) - self.transport.packets = [] - self.conn.ssh_CHANNEL_DATA('\x00\x00\x00\x01' + common.NS(bigData)) - self.assertEqual(channel.inBuffer, []) - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_CLOSE, '\x00\x00\x00\xff')]) - - def test_CHANNEL_EXTENDED_DATA(self): - """ - Test that channel extended data messages are passed up to the channel, - or cause the channel to be closed if they're too big. - """ - channel = TestChannel(localWindow=6, localMaxPacket=5) - self._openChannel(channel) - self.conn.ssh_CHANNEL_EXTENDED_DATA('\x00\x00\x00\x00\x00\x00\x00' - '\x00' + common.NS('data')) - self.assertEqual(channel.extBuffer, [(0, 'data')]) - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_WINDOW_ADJUST, '\x00\x00\x00\xff' - '\x00\x00\x00\x04')]) - self.transport.packets = [] - longData = 'a' * (channel.localWindowLeft + 1) - self.conn.ssh_CHANNEL_EXTENDED_DATA('\x00\x00\x00\x00\x00\x00\x00' - '\x00' + common.NS(longData)) - self.assertEqual(channel.extBuffer, [(0, 'data')]) - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_CLOSE, '\x00\x00\x00\xff')]) - channel = TestChannel() - self._openChannel(channel) - bigData = 'a' * (channel.localMaxPacket + 1) - self.transport.packets = [] - self.conn.ssh_CHANNEL_EXTENDED_DATA('\x00\x00\x00\x01\x00\x00\x00' - '\x00' + common.NS(bigData)) - self.assertEqual(channel.extBuffer, []) - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_CLOSE, '\x00\x00\x00\xff')]) - - def test_CHANNEL_EOF(self): - """ - Test that channel eof messages are passed up to the channel. - """ - channel = TestChannel() - self._openChannel(channel) - self.conn.ssh_CHANNEL_EOF('\x00\x00\x00\x00') - self.assertTrue(channel.gotEOF) - - def test_CHANNEL_CLOSE(self): - """ - Test that channel close messages are passed up to the channel. Also, - test that channel.close() is called if both sides are closed when this - message is received. - """ - channel = TestChannel() - self._openChannel(channel) - self.conn.sendClose(channel) - self.conn.ssh_CHANNEL_CLOSE('\x00\x00\x00\x00') - self.assertTrue(channel.gotOneClose) - self.assertTrue(channel.gotClosed) - - def test_CHANNEL_REQUEST_success(self): - """ - Test that channel requests that succeed send MSG_CHANNEL_SUCCESS. - """ - channel = TestChannel() - self._openChannel(channel) - self.conn.ssh_CHANNEL_REQUEST('\x00\x00\x00\x00' + common.NS('test') - + '\x00') - self.assertEqual(channel.numberRequests, 1) - d = self.conn.ssh_CHANNEL_REQUEST('\x00\x00\x00\x00' + common.NS( - 'test') + '\xff' + 'data') - def check(result): - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_SUCCESS, '\x00\x00\x00\xff')]) - d.addCallback(check) - return d - - def test_CHANNEL_REQUEST_failure(self): - """ - Test that channel requests that fail send MSG_CHANNEL_FAILURE. - """ - channel = TestChannel() - self._openChannel(channel) - d = self.conn.ssh_CHANNEL_REQUEST('\x00\x00\x00\x00' + common.NS( - 'test') + '\xff') - def check(result): - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_FAILURE, '\x00\x00\x00\xff' - )]) - d.addCallback(self.fail) - d.addErrback(check) - return d - - def test_CHANNEL_REQUEST_SUCCESS(self): - """ - Test that channel request success messages cause the Deferred to be - called back. - """ - channel = TestChannel() - self._openChannel(channel) - d = self.conn.sendRequest(channel, 'test', 'data', True) - self.conn.ssh_CHANNEL_SUCCESS('\x00\x00\x00\x00') - def check(result): - self.assertTrue(result) - return d - - def test_CHANNEL_REQUEST_FAILURE(self): - """ - Test that channel request failure messages cause the Deferred to be - erred back. - """ - channel = TestChannel() - self._openChannel(channel) - d = self.conn.sendRequest(channel, 'test', '', True) - self.conn.ssh_CHANNEL_FAILURE('\x00\x00\x00\x00') - def check(result): - self.assertEqual(result.value.value, 'channel request failed') - d.addCallback(self.fail) - d.addErrback(check) - return d - - def test_sendGlobalRequest(self): - """ - Test that global request messages are sent in the right format. - """ - d = self.conn.sendGlobalRequest('wantReply', 'data', True) - # must be added to prevent errbacking during teardown - d.addErrback(lambda failure: None) - self.conn.sendGlobalRequest('noReply', '', False) - self.assertEqual(self.transport.packets, - [(connection.MSG_GLOBAL_REQUEST, common.NS('wantReply') + - '\xffdata'), - (connection.MSG_GLOBAL_REQUEST, common.NS('noReply') + - '\x00')]) - self.assertEqual(self.conn.deferreds, {'global':[d]}) - - def test_openChannel(self): - """ - Test that open channel messages are sent in the right format. - """ - channel = TestChannel() - self.conn.openChannel(channel, 'aaaa') - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_OPEN, common.NS('TestChannel') + - '\x00\x00\x00\x00\x00\x02\x00\x00\x00\x00\x80\x00aaaa')]) - self.assertEqual(channel.id, 0) - self.assertEqual(self.conn.localChannelID, 1) - - def test_sendRequest(self): - """ - Test that channel request messages are sent in the right format. - """ - channel = TestChannel() - self._openChannel(channel) - d = self.conn.sendRequest(channel, 'test', 'test', True) - # needed to prevent errbacks during teardown. - d.addErrback(lambda failure: None) - self.conn.sendRequest(channel, 'test2', '', False) - channel.localClosed = True # emulate sending a close message - self.conn.sendRequest(channel, 'test3', '', True) - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_REQUEST, '\x00\x00\x00\xff' + - common.NS('test') + '\x01test'), - (connection.MSG_CHANNEL_REQUEST, '\x00\x00\x00\xff' + - common.NS('test2') + '\x00')]) - self.assertEqual(self.conn.deferreds[0], [d]) - - def test_adjustWindow(self): - """ - Test that channel window adjust messages cause bytes to be added - to the window. - """ - channel = TestChannel(localWindow=5) - self._openChannel(channel) - channel.localWindowLeft = 0 - self.conn.adjustWindow(channel, 1) - self.assertEqual(channel.localWindowLeft, 1) - channel.localClosed = True - self.conn.adjustWindow(channel, 2) - self.assertEqual(channel.localWindowLeft, 1) - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_WINDOW_ADJUST, '\x00\x00\x00\xff' - '\x00\x00\x00\x01')]) - - def test_sendData(self): - """ - Test that channel data messages are sent in the right format. - """ - channel = TestChannel() - self._openChannel(channel) - self.conn.sendData(channel, 'a') - channel.localClosed = True - self.conn.sendData(channel, 'b') - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_DATA, '\x00\x00\x00\xff' + - common.NS('a'))]) - - def test_sendExtendedData(self): - """ - Test that channel extended data messages are sent in the right format. - """ - channel = TestChannel() - self._openChannel(channel) - self.conn.sendExtendedData(channel, 1, 'test') - channel.localClosed = True - self.conn.sendExtendedData(channel, 2, 'test2') - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_EXTENDED_DATA, '\x00\x00\x00\xff' + - '\x00\x00\x00\x01' + common.NS('test'))]) - - def test_sendEOF(self): - """ - Test that channel EOF messages are sent in the right format. - """ - channel = TestChannel() - self._openChannel(channel) - self.conn.sendEOF(channel) - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_EOF, '\x00\x00\x00\xff')]) - channel.localClosed = True - self.conn.sendEOF(channel) - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_EOF, '\x00\x00\x00\xff')]) - - def test_sendClose(self): - """ - Test that channel close messages are sent in the right format. - """ - channel = TestChannel() - self._openChannel(channel) - self.conn.sendClose(channel) - self.assertTrue(channel.localClosed) - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_CLOSE, '\x00\x00\x00\xff')]) - self.conn.sendClose(channel) - self.assertEqual(self.transport.packets, - [(connection.MSG_CHANNEL_CLOSE, '\x00\x00\x00\xff')]) - - channel2 = TestChannel() - self._openChannel(channel2) - channel2.remoteClosed = True - self.conn.sendClose(channel2) - self.assertTrue(channel2.gotClosed) - - def test_getChannelWithAvatar(self): - """ - Test that getChannel dispatches to the avatar when an avatar is - present. Correct functioning without the avatar is verified in - test_CHANNEL_OPEN. - """ - channel = self.conn.getChannel('TestChannel', 50, 30, 'data') - self.assertEqual(channel.data, 'data') - self.assertEqual(channel.remoteWindowLeft, 50) - self.assertEqual(channel.remoteMaxPacket, 30) - self.assertRaises(error.ConchError, self.conn.getChannel, - 'BadChannel', 50, 30, 'data') - - def test_gotGlobalRequestWithoutAvatar(self): - """ - Test that gotGlobalRequests dispatches to global_* without an avatar. - """ - del self.transport.avatar - self.assertTrue(self.conn.gotGlobalRequest('TestGlobal', 'data')) - self.assertEqual(self.conn.gotGlobalRequest('Test-Data', 'data'), - (True, 'data')) - self.assertFalse(self.conn.gotGlobalRequest('BadGlobal', 'data')) - - - def test_channelClosedCausesLeftoverChannelDeferredsToErrback(self): - """ - Whenever an SSH channel gets closed any Deferred that was returned by a - sendRequest() on its parent connection must be errbacked. - """ - channel = TestChannel() - self._openChannel(channel) - - d = self.conn.sendRequest( - channel, "dummyrequest", "dummydata", wantReply=1) - d = self.assertFailure(d, error.ConchError) - self.conn.channelClosed(channel) - return d - - - -class TestCleanConnectionShutdown(unittest.TestCase): - """ - Check whether correct cleanup is performed on connection shutdown. - """ - if test_userauth.transport is None: - skip = "Cannot run without both PyCrypto and pyasn1" - - def setUp(self): - self.transport = test_userauth.FakeTransport(None) - self.transport.avatar = TestAvatar() - self.conn = TestConnection() - self.conn.transport = self.transport - - - def test_serviceStoppedCausesLeftoverGlobalDeferredsToErrback(self): - """ - Once the service is stopped any leftover global deferred returned by - a sendGlobalRequest() call must be errbacked. - """ - self.conn.serviceStarted() - - d = self.conn.sendGlobalRequest( - "dummyrequest", "dummydata", wantReply=1) - d = self.assertFailure(d, error.ConchError) - self.conn.serviceStopped() - return d - - |