1 # Copyright (c) Twisted Matrix Laboratories.
2 # See LICENSE for details.
5 Tests for L{twisted.conch.ssh}.
11 import Crypto.Cipher.DES3
20 from twisted.conch.ssh import common, session, forwarding
21 from twisted.conch import avatar, error
22 from twisted.conch.test.keydata import publicRSA_openssh, privateRSA_openssh
23 from twisted.conch.test.keydata import publicDSA_openssh, privateDSA_openssh
24 from twisted.cred import portal
25 from twisted.cred.error import UnauthorizedLogin
26 from twisted.internet import defer, protocol, reactor
27 from twisted.internet.error import ProcessTerminated
28 from twisted.python import failure, log
29 from twisted.trial import unittest
31 from twisted.conch.test.test_recvline import LoopbackRelay
35 class ConchTestRealm(object):
37 A realm which expects a particular avatarId to log in once and creates a
38 L{ConchTestAvatar} for that request.
40 @ivar expectedAvatarID: The only avatarID that this realm will produce an
43 @ivar avatar: A reference to the avatar after it is requested.
47 def __init__(self, expectedAvatarID):
48 self.expectedAvatarID = expectedAvatarID
51 def requestAvatar(self, avatarID, mind, *interfaces):
53 Return a new L{ConchTestAvatar} if the avatarID matches the expected one
54 and this is the first avatar request.
56 if avatarID == self.expectedAvatarID:
57 if self.avatar is not None:
58 raise UnauthorizedLogin("Only one login allowed")
59 self.avatar = ConchTestAvatar()
60 return interfaces[0], self.avatar, self.avatar.logout
61 raise UnauthorizedLogin(
62 "Only %r may log in, not %r" % (self.expectedAvatarID, avatarID))
66 class ConchTestAvatar(avatar.ConchUser):
68 An avatar against which various SSH features can be tested.
70 @ivar loggedOut: A flag indicating whether the avatar logout method has been
76 avatar.ConchUser.__init__(self)
78 self.globalRequests = {}
79 self.channelLookup.update({'session': session.SSHSession,
80 'direct-tcpip':forwarding.openConnectForwardingClient})
81 self.subsystemLookup.update({'crazy': CrazySubsystem})
84 def global_foo(self, data):
85 self.globalRequests['foo'] = data
89 def global_foo_2(self, data):
90 self.globalRequests['foo_2'] = data
94 def global_tcpip_forward(self, data):
95 host, port = forwarding.unpackGlobal_tcpip_forward(data)
97 listener = reactor.listenTCP(
98 port, forwarding.SSHListenForwardingFactory(
99 self.conn, (host, port),
100 forwarding.SSHListenServerForwardingChannel),
103 log.err(None, "something went wrong with remote->local forwarding")
106 self.listeners[(host, port)] = listener
110 def global_cancel_tcpip_forward(self, data):
111 host, port = forwarding.unpackGlobal_tcpip_forward(data)
112 listener = self.listeners.get((host, port), None)
115 del self.listeners[(host, port)]
116 listener.stopListening()
121 self.loggedOut = True
122 for listener in self.listeners.values():
123 log.msg('stopListening %s' % listener)
124 listener.stopListening()
128 class ConchSessionForTestAvatar(object):
130 An ISession adapter for ConchTestAvatar.
132 def __init__(self, avatar):
134 Initialize the session and create a reference to it on the avatar for
138 self.avatar._testSession = self
143 self.onClose = defer.Deferred()
146 def getPty(self, term, windowSize, attrs):
148 self._terminalType = term
149 self._windowSize = windowSize
153 def openShell(self, proto):
154 log.msg('opening shell')
160 def execCommand(self, proto, cmd):
165 t = FalseTransport(proto)
166 # Avoid disconnecting this immediately. If the channel is closed
167 # before execCommand even returns the caller gets confused.
168 reactor.callLater(0, t.loseConnection)
170 t = EchoTransport(proto)
174 t = SuperEchoTransport(proto)
178 t = ErrEchoTransport(proto)
182 raise error.ConchError('bad exec')
183 self.avatar.conn.transport.expectedLoseConnection = 1
186 def eofReceived(self):
191 log.msg('closed cmd "%s"' % self.cmd)
192 self.remoteWindowLeftAtClose = self.proto.session.remoteWindowLeft
193 self.onClose.callback(None)
195 from twisted.python import components
196 components.registerAdapter(ConchSessionForTestAvatar, ConchTestAvatar, session.ISession)
198 class CrazySubsystem(protocol.Protocol):
200 def __init__(self, *args, **kw):
203 def connectionMade(self):
210 class FalseTransport:
212 False transport should act like a /bin/false execution, i.e. just exit with
213 nonzero status, writing nothing to the terminal.
215 @ivar proto: The protocol associated with this transport.
216 @ivar closed: A flag tracking whether C{loseConnection} has been called yet.
219 def __init__(self, p):
221 @type p L{twisted.conch.ssh.session.SSHSessionProcessProtocol} instance
224 p.makeConnection(self)
228 def loseConnection(self):
230 Disconnect the protocol associated with this transport.
235 self.proto.inConnectionLost()
236 self.proto.outConnectionLost()
237 self.proto.errConnectionLost()
238 self.proto.processEnded(failure.Failure(ProcessTerminated(255, None, None)))
244 def __init__(self, p):
246 p.makeConnection(self)
249 def write(self, data):
251 self.proto.outReceived(data)
252 self.proto.outReceived('\r\n')
253 if '\x00' in data: # mimic 'exit' for the shell test
254 self.loseConnection()
256 def loseConnection(self):
257 if self.closed: return
259 self.proto.inConnectionLost()
260 self.proto.outConnectionLost()
261 self.proto.errConnectionLost()
262 self.proto.processEnded(failure.Failure(ProcessTerminated(0, None, None)))
264 class ErrEchoTransport:
266 def __init__(self, p):
268 p.makeConnection(self)
271 def write(self, data):
272 self.proto.errReceived(data)
273 self.proto.errReceived('\r\n')
275 def loseConnection(self):
276 if self.closed: return
278 self.proto.inConnectionLost()
279 self.proto.outConnectionLost()
280 self.proto.errConnectionLost()
281 self.proto.processEnded(failure.Failure(ProcessTerminated(0, None, None)))
283 class SuperEchoTransport:
285 def __init__(self, p):
287 p.makeConnection(self)
290 def write(self, data):
291 self.proto.outReceived(data)
292 self.proto.outReceived('\r\n')
293 self.proto.errReceived(data)
294 self.proto.errReceived('\r\n')
296 def loseConnection(self):
297 if self.closed: return
299 self.proto.inConnectionLost()
300 self.proto.outConnectionLost()
301 self.proto.errConnectionLost()
302 self.proto.processEnded(failure.Failure(ProcessTerminated(0, None, None)))
305 if Crypto is not None and pyasn1 is not None:
306 from twisted.conch import checkers
307 from twisted.conch.ssh import channel, connection, factory, keys
308 from twisted.conch.ssh import transport, userauth
310 class UtilityTestCase(unittest.TestCase):
311 def testCounter(self):
312 c = transport._Counter('\x00\x00', 2)
313 for i in xrange(256 * 256):
314 self.assertEqual(c(), struct.pack('!H', (i + 1) % (2 ** 16)))
315 # It should wrap around, too.
316 for i in xrange(256 * 256):
317 self.assertEqual(c(), struct.pack('!H', (i + 1) % (2 ** 16)))
320 class ConchTestPublicKeyChecker(checkers.SSHPublicKeyDatabase):
321 def checkKey(self, credentials):
322 blob = keys.Key.fromString(publicDSA_openssh).blob()
323 if credentials.username == 'testuser' and credentials.blob == blob:
328 class ConchTestPasswordChecker:
329 credentialInterfaces = checkers.IUsernamePassword,
331 def requestAvatarId(self, credentials):
332 if credentials.username == 'testuser' and credentials.password == 'testpass':
333 return defer.succeed(credentials.username)
334 return defer.fail(Exception("Bad credentials"))
337 class ConchTestSSHChecker(checkers.SSHProtocolChecker):
339 def areDone(self, avatarId):
340 if avatarId != 'testuser' or len(self.successfulCredentials[avatarId]) < 2:
344 class ConchTestServerFactory(factory.SSHFactory):
348 'ssh-userauth':userauth.SSHUserAuthServer,
349 'ssh-connection':connection.SSHConnection
352 def buildProtocol(self, addr):
353 proto = ConchTestServer()
354 proto.supportedPublicKeys = self.privateKeys.keys()
357 if hasattr(self, 'expectedLoseConnection'):
358 proto.expectedLoseConnection = self.expectedLoseConnection
363 def getPublicKeys(self):
365 'ssh-rsa': keys.Key.fromString(publicRSA_openssh),
366 'ssh-dss': keys.Key.fromString(publicDSA_openssh)
369 def getPrivateKeys(self):
371 'ssh-rsa': keys.Key.fromString(privateRSA_openssh),
372 'ssh-dss': keys.Key.fromString(privateDSA_openssh)
377 2048:[(transport.DH_GENERATOR, transport.DH_PRIME)]
380 def getService(self, trans, name):
381 return factory.SSHFactory.getService(self, trans, name)
387 def connectionLost(self, reason):
390 if not hasattr(self,'expectedLoseConnection'):
391 unittest.fail('unexpectedly lost connection %s\n%s' % (self, reason))
394 def receiveError(self, reasonCode, desc):
395 self.expectedLoseConnection = 1
396 # Some versions of OpenSSH (for example, OpenSSH_5.3p1) will
397 # send a DISCONNECT_BY_APPLICATION error before closing the
398 # connection. Other, older versions (for example,
399 # OpenSSH_5.1p1), won't. So accept this particular error here,
401 if reasonCode != transport.DISCONNECT_BY_APPLICATION:
404 'got disconnect for %s: reason %s, desc: %s' % (
405 self, reasonCode, desc)))
406 self.loseConnection()
408 def receiveUnimplemented(self, seqID):
409 unittest.fail('got unimplemented: seqid %s' % seqID)
410 self.expectedLoseConnection = 1
411 self.loseConnection()
413 class ConchTestServer(ConchTestBase, transport.SSHServerTransport):
415 def connectionLost(self, reason):
416 ConchTestBase.connectionLost(self, reason)
417 transport.SSHServerTransport.connectionLost(self, reason)
420 class ConchTestClient(ConchTestBase, transport.SSHClientTransport):
422 @ivar _channelFactory: A callable which accepts an SSH connection and
423 returns a channel which will be attached to a new channel on that
426 def __init__(self, channelFactory):
427 self._channelFactory = channelFactory
429 def connectionLost(self, reason):
430 ConchTestBase.connectionLost(self, reason)
431 transport.SSHClientTransport.connectionLost(self, reason)
433 def verifyHostKey(self, key, fp):
434 keyMatch = key == keys.Key.fromString(publicRSA_openssh).blob()
436 fp == '3d:13:5f:cb:c9:79:8a:93:06:27:65:bc:3d:0b:8f:af')
437 if keyMatch and fingerprintMatch:
438 return defer.succeed(1)
439 return defer.fail(Exception("Key or fingerprint mismatch"))
441 def connectionSecure(self):
442 self.requestService(ConchTestClientAuth('testuser',
443 ConchTestClientConnection(self._channelFactory)))
446 class ConchTestClientAuth(userauth.SSHUserAuthClient):
448 hasTriedNone = 0 # have we tried the 'none' auth yet?
449 canSucceedPublicKey = 0 # can we succed with this yet?
450 canSucceedPassword = 0
452 def ssh_USERAUTH_SUCCESS(self, packet):
453 if not self.canSucceedPassword and self.canSucceedPublicKey:
454 unittest.fail('got USERAUTH_SUCESS before password and publickey')
455 userauth.SSHUserAuthClient.ssh_USERAUTH_SUCCESS(self, packet)
457 def getPassword(self):
458 self.canSucceedPassword = 1
459 return defer.succeed('testpass')
461 def getPrivateKey(self):
462 self.canSucceedPublicKey = 1
463 return defer.succeed(keys.Key.fromString(privateDSA_openssh))
465 def getPublicKey(self):
466 return keys.Key.fromString(publicDSA_openssh)
469 class ConchTestClientConnection(connection.SSHConnection):
471 @ivar _completed: A L{Deferred} which will be fired when the number of
472 results collected reaches C{totalResults}.
474 name = 'ssh-connection'
478 def __init__(self, channelFactory):
479 connection.SSHConnection.__init__(self)
480 self._channelFactory = channelFactory
482 def serviceStarted(self):
483 self.openChannel(self._channelFactory(conn=self))
486 class SSHTestChannel(channel.SSHChannel):
488 def __init__(self, name, opened, *args, **kwargs):
490 self._opened = opened
492 self.receivedExt = []
493 self.onClose = defer.Deferred()
494 channel.SSHChannel.__init__(self, *args, **kwargs)
497 def openFailed(self, reason):
498 self._opened.errback(reason)
501 def channelOpen(self, ignore):
502 self._opened.callback(self)
505 def dataReceived(self, data):
506 self.received.append(data)
509 def extReceived(self, dataType, data):
510 if dataType == connection.EXTENDED_DATA_STDERR:
511 self.receivedExt.append(data)
513 log.msg("Unrecognized extended data: %r" % (dataType,))
516 def request_exit_status(self, status):
517 [self.status] = struct.unpack('>L', status)
520 def eofReceived(self):
521 self.eofCalled = True
525 self.onClose.callback(None)
529 class SSHProtocolTestCase(unittest.TestCase):
531 Tests for communication between L{SSHServerTransport} and
532 L{SSHClientTransport}.
536 skip = "can't run w/o PyCrypto"
539 skip = "Cannot run without PyASN1"
541 def _ourServerOurClientTest(self, name='session', **kwargs):
543 Create a connected SSH client and server protocol pair and return a
544 L{Deferred} which fires with an L{SSHTestChannel} instance connected to
545 a channel on that SSH connection.
547 result = defer.Deferred()
548 self.realm = ConchTestRealm('testuser')
549 p = portal.Portal(self.realm)
550 sshpc = ConchTestSSHChecker()
551 sshpc.registerChecker(ConchTestPasswordChecker())
552 sshpc.registerChecker(ConchTestPublicKeyChecker())
553 p.registerChecker(sshpc)
554 fac = ConchTestServerFactory()
557 self.server = fac.buildProtocol(None)
558 self.clientTransport = LoopbackRelay(self.server)
559 self.client = ConchTestClient(
560 lambda conn: SSHTestChannel(name, result, conn=conn, **kwargs))
562 self.serverTransport = LoopbackRelay(self.client)
564 self.server.makeConnection(self.serverTransport)
565 self.client.makeConnection(self.clientTransport)
569 def test_subsystemsAndGlobalRequests(self):
571 Run the Conch server against the Conch client. Set up several different
572 channels which exercise different behaviors and wait for them to
573 complete. Verify that the channels with errors log them.
575 channel = self._ourServerOurClientTest()
577 def cbSubsystem(channel):
578 self.channel = channel
579 return self.assertFailure(
580 channel.conn.sendRequest(
581 channel, 'subsystem', common.NS('not-crazy'), 1),
583 channel.addCallback(cbSubsystem)
585 def cbNotCrazyFailed(ignored):
586 channel = self.channel
587 return channel.conn.sendRequest(
588 channel, 'subsystem', common.NS('crazy'), 1)
589 channel.addCallback(cbNotCrazyFailed)
591 def cbGlobalRequests(ignored):
592 channel = self.channel
593 d1 = channel.conn.sendGlobalRequest('foo', 'bar', 1)
595 d2 = channel.conn.sendGlobalRequest('foo-2', 'bar2', 1)
596 d2.addCallback(self.assertEqual, 'data')
598 d3 = self.assertFailure(
599 channel.conn.sendGlobalRequest('bar', 'foo', 1),
602 return defer.gatherResults([d1, d2, d3])
603 channel.addCallback(cbGlobalRequests)
605 def disconnect(ignored):
607 self.realm.avatar.globalRequests,
608 {"foo": "bar", "foo_2": "bar2"})
609 channel = self.channel
610 channel.conn.transport.expectedLoseConnection = True
611 channel.conn.serviceStopped()
612 channel.loseConnection()
613 channel.addCallback(disconnect)
618 def test_shell(self):
620 L{SSHChannel.sendRequest} can open a shell with a I{pty-req} request,
621 specifying a terminal type and window size.
623 channel = self._ourServerOurClientTest()
625 data = session.packRequest_pty_req('conch-test-term', (24, 80, 0, 0), '')
626 def cbChannel(channel):
627 self.channel = channel
628 return channel.conn.sendRequest(channel, 'pty-req', data, 1)
629 channel.addCallback(cbChannel)
632 # The server-side object corresponding to our client side channel.
633 session = self.realm.avatar.conn.channels[0].session
634 self.assertIdentical(session.avatar, self.realm.avatar)
635 self.assertEqual(session._terminalType, 'conch-test-term')
636 self.assertEqual(session._windowSize, (24, 80, 0, 0))
637 self.assertTrue(session.ptyReq)
638 channel = self.channel
639 return channel.conn.sendRequest(channel, 'shell', '', 1)
640 channel.addCallback(cbPty)
642 def cbShell(ignored):
643 self.channel.write('testing the shell!\x00')
644 self.channel.conn.sendEOF(self.channel)
645 return defer.gatherResults([
646 self.channel.onClose,
647 self.realm.avatar._testSession.onClose])
648 channel.addCallback(cbShell)
650 def cbExited(ignored):
651 if self.channel.status != 0:
653 'shell exit status was not 0: %i' % (self.channel.status,))
655 "".join(self.channel.received),
656 'testing the shell!\x00\r\n')
657 self.assertTrue(self.channel.eofCalled)
659 self.realm.avatar._testSession.eof)
660 channel.addCallback(cbExited)
664 def test_failedExec(self):
666 If L{SSHChannel.sendRequest} issues an exec which the server responds to
667 with an error, the L{Deferred} it returns fires its errback.
669 channel = self._ourServerOurClientTest()
671 def cbChannel(channel):
672 self.channel = channel
673 return self.assertFailure(
674 channel.conn.sendRequest(
675 channel, 'exec', common.NS('jumboliah'), 1),
677 channel.addCallback(cbChannel)
679 def cbFailed(ignored):
680 # The server logs this exception when it cannot perform the
682 errors = self.flushLoggedErrors(error.ConchError)
683 self.assertEqual(errors[0].value.args, ('bad exec', None))
684 channel.addCallback(cbFailed)
688 def test_falseChannel(self):
690 When the process started by a L{SSHChannel.sendRequest} exec request
691 exits, the exit status is reported to the channel.
693 channel = self._ourServerOurClientTest()
695 def cbChannel(channel):
696 self.channel = channel
697 return channel.conn.sendRequest(
698 channel, 'exec', common.NS('false'), 1)
699 channel.addCallback(cbChannel)
702 return self.channel.onClose
703 channel.addCallback(cbExec)
705 def cbClosed(ignored):
706 # No data is expected
707 self.assertEqual(self.channel.received, [])
708 self.assertNotEquals(self.channel.status, 0)
709 channel.addCallback(cbClosed)
713 def test_errorChannel(self):
715 Bytes sent over the extended channel for stderr data are delivered to
716 the channel's C{extReceived} method.
718 channel = self._ourServerOurClientTest(localWindow=4, localMaxPacket=5)
720 def cbChannel(channel):
721 self.channel = channel
722 return channel.conn.sendRequest(
723 channel, 'exec', common.NS('eecho hello'), 1)
724 channel.addCallback(cbChannel)
727 return defer.gatherResults([
728 self.channel.onClose,
729 self.realm.avatar._testSession.onClose])
730 channel.addCallback(cbExec)
732 def cbClosed(ignored):
733 self.assertEqual(self.channel.received, [])
734 self.assertEqual("".join(self.channel.receivedExt), "hello\r\n")
735 self.assertEqual(self.channel.status, 0)
736 self.assertTrue(self.channel.eofCalled)
737 self.assertEqual(self.channel.localWindowLeft, 4)
739 self.channel.localWindowLeft,
740 self.realm.avatar._testSession.remoteWindowLeftAtClose)
741 channel.addCallback(cbClosed)
745 def test_unknownChannel(self):
747 When an attempt is made to open an unknown channel type, the L{Deferred}
748 returned by L{SSHChannel.sendRequest} fires its errback.
750 d = self.assertFailure(
751 self._ourServerOurClientTest('crazy-unknown-channel'), Exception)
752 def cbFailed(ignored):
753 errors = self.flushLoggedErrors(error.ConchError)
754 self.assertEqual(errors[0].value.args, (3, 'unknown channel'))
755 self.assertEqual(len(errors), 1)
756 d.addCallback(cbFailed)
760 def test_maxPacket(self):
762 An L{SSHChannel} can be configured with a maximum packet size to
765 # localWindow needs to be at least 11 otherwise the assertion about it
766 # in cbClosed is invalid.
767 channel = self._ourServerOurClientTest(
768 localWindow=11, localMaxPacket=1)
770 def cbChannel(channel):
771 self.channel = channel
772 return channel.conn.sendRequest(
773 channel, 'exec', common.NS('secho hello'), 1)
774 channel.addCallback(cbChannel)
777 return self.channel.onClose
778 channel.addCallback(cbExec)
780 def cbClosed(ignored):
781 self.assertEqual(self.channel.status, 0)
782 self.assertEqual("".join(self.channel.received), "hello\r\n")
783 self.assertEqual("".join(self.channel.receivedExt), "hello\r\n")
784 self.assertEqual(self.channel.localWindowLeft, 11)
785 self.assertTrue(self.channel.eofCalled)
786 channel.addCallback(cbClosed)
792 Normal standard out bytes are sent to the channel's C{dataReceived}
795 channel = self._ourServerOurClientTest(localWindow=4, localMaxPacket=5)
797 def cbChannel(channel):
798 self.channel = channel
799 return channel.conn.sendRequest(
800 channel, 'exec', common.NS('echo hello'), 1)
801 channel.addCallback(cbChannel)
804 return defer.gatherResults([
805 self.channel.onClose,
806 self.realm.avatar._testSession.onClose])
807 channel.addCallback(cbEcho)
809 def cbClosed(ignored):
810 self.assertEqual(self.channel.status, 0)
811 self.assertEqual("".join(self.channel.received), "hello\r\n")
812 self.assertEqual(self.channel.localWindowLeft, 4)
813 self.assertTrue(self.channel.eofCalled)
815 self.channel.localWindowLeft,
816 self.realm.avatar._testSession.remoteWindowLeftAtClose)
817 channel.addCallback(cbClosed)
822 class TestSSHFactory(unittest.TestCase):
825 skip = "can't run w/o PyCrypto"
828 skip = "Cannot run without PyASN1"
830 def makeSSHFactory(self, primes=None):
831 sshFactory = factory.SSHFactory()
832 gpk = lambda: {'ssh-rsa' : keys.Key(None)}
833 sshFactory.getPrimes = lambda: primes
834 sshFactory.getPublicKeys = sshFactory.getPrivateKeys = gpk
835 sshFactory.startFactory()
839 def test_buildProtocol(self):
841 By default, buildProtocol() constructs an instance of
844 factory = self.makeSSHFactory()
845 protocol = factory.buildProtocol(None)
846 self.assertIsInstance(protocol, transport.SSHServerTransport)
849 def test_buildProtocolRespectsProtocol(self):
851 buildProtocol() calls 'self.protocol()' to construct a protocol
855 def makeProtocol(*args):
857 return transport.SSHServerTransport()
858 factory = self.makeSSHFactory()
859 factory.protocol = makeProtocol
860 factory.buildProtocol(None)
861 self.assertEqual([()], calls)
864 def test_multipleFactories(self):
865 f1 = self.makeSSHFactory(primes=None)
866 f2 = self.makeSSHFactory(primes={1:(2,3)})
867 p1 = f1.buildProtocol(None)
868 p2 = f2.buildProtocol(None)
869 self.failIf('diffie-hellman-group-exchange-sha1' in p1.supportedKeyExchanges,
870 p1.supportedKeyExchanges)
871 self.failUnless('diffie-hellman-group-exchange-sha1' in p2.supportedKeyExchanges,
872 p2.supportedKeyExchanges)
876 class MPTestCase(unittest.TestCase):
878 Tests for L{common.getMP}.
880 @cvar getMP: a method providing a MP parser.
881 @type getMP: C{callable}
883 getMP = staticmethod(common.getMP)
886 skip = "can't run w/o PyCrypto"
889 skip = "Cannot run without PyASN1"
892 def test_getMP(self):
894 L{common.getMP} should parse the a multiple precision integer from a
895 string: a 4-byte length followed by length bytes of the integer.
898 self.getMP('\x00\x00\x00\x04\x00\x00\x00\x01'),
902 def test_getMPBigInteger(self):
904 L{common.getMP} should be able to parse a big enough integer
905 (that doesn't fit on one byte).
908 self.getMP('\x00\x00\x00\x04\x01\x02\x03\x04'),
912 def test_multipleGetMP(self):
914 L{common.getMP} has the ability to parse multiple integer in the same
918 self.getMP('\x00\x00\x00\x04\x00\x00\x00\x01'
919 '\x00\x00\x00\x04\x00\x00\x00\x02', 2),
923 def test_getMPRemainingData(self):
925 When more data than needed is sent to L{common.getMP}, it should return
929 self.getMP('\x00\x00\x00\x04\x00\x00\x00\x01foo'),
933 def test_notEnoughData(self):
935 When the string passed to L{common.getMP} doesn't even make 5 bytes,
936 it should raise a L{struct.error}.
938 self.assertRaises(struct.error, self.getMP, '\x02\x00')
942 class PyMPTestCase(MPTestCase):
944 Tests for the python implementation of L{common.getMP}.
946 getMP = staticmethod(common.getMP_py)
950 class GMPYMPTestCase(MPTestCase):
952 Tests for the gmpy implementation of L{common.getMP}.
954 getMP = staticmethod(common._fastgetMP)
957 class BuiltinPowHackTestCase(unittest.TestCase):
959 Tests that the builtin pow method is still correct after
960 L{twisted.conch.ssh.common} monkeypatches it to use gmpy.
963 def test_floatBase(self):
965 pow gives the correct result when passed a base of type float with a
968 self.assertEqual(6.25, pow(2.5, 2))
970 def test_intBase(self):
972 pow gives the correct result when passed a base of type int.
974 self.assertEqual(81, pow(3, 4))
976 def test_longBase(self):
978 pow gives the correct result when passed a base of type long.
980 self.assertEqual(81, pow(3, 4))
982 def test_mpzBase(self):
984 pow gives the correct result when passed a base of type gmpy.mpz.
987 raise unittest.SkipTest('gmpy not available')
988 self.assertEqual(81, pow(gmpy.mpz(3), 4))
994 GMPYMPTestCase.skip = "gmpy not available"