From e299181a5dda25aed4879ebcbe1359604448b3ae Mon Sep 17 00:00:00 2001 From: "Jeremy T. Bouse" Date: Fri, 27 Nov 2009 16:25:55 -0500 Subject: Imported Upstream version 1.7.6 --- tests/loop.py | 2 +- tests/stub_sftp.py | 2 +- tests/test_auth.py | 2 +- tests/test_buffered_pipe.py | 2 +- tests/test_client.py | 40 +++++++++++++++---------------- tests/test_file.py | 2 +- tests/test_hostkeys.py | 2 +- tests/test_kex.py | 2 +- tests/test_message.py | 2 +- tests/test_packetizer.py | 2 +- tests/test_pkey.py | 2 +- tests/test_sftp.py | 58 +++++++++++++++++++++++++++------------------ tests/test_sftp_big.py | 2 +- tests/test_transport.py | 33 +++++++++++++++++++++++++- tests/test_util.py | 2 +- 15 files changed, 99 insertions(+), 56 deletions(-) (limited to 'tests') diff --git a/tests/loop.py b/tests/loop.py index fb6ffae..bdc2f2d 100644 --- a/tests/loop.py +++ b/tests/loop.py @@ -1,4 +1,4 @@ -# Copyright (C) 2003-2007 Robey Pointer +# Copyright (C) 2003-2009 Robey Pointer # # This file is part of paramiko. # diff --git a/tests/stub_sftp.py b/tests/stub_sftp.py index ac292ff..7f1ecc7 100644 --- a/tests/stub_sftp.py +++ b/tests/stub_sftp.py @@ -1,4 +1,4 @@ -# Copyright (C) 2003-2007 Robey Pointer +# Copyright (C) 2003-2009 Robey Pointer # # This file is part of paramiko. # diff --git a/tests/test_auth.py b/tests/test_auth.py index fadd8ca..816e978 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -1,4 +1,4 @@ -# Copyright (C) 2008 Robey Pointer +# Copyright (C) 2008 Robey Pointer # # This file is part of paramiko. # diff --git a/tests/test_buffered_pipe.py b/tests/test_buffered_pipe.py index f96edb8..f285d05 100644 --- a/tests/test_buffered_pipe.py +++ b/tests/test_buffered_pipe.py @@ -1,4 +1,4 @@ -# Copyright (C) 2006-2007 Robey Pointer +# Copyright (C) 2006-2007 Robey Pointer # # This file is part of paramiko. # diff --git a/tests/test_client.py b/tests/test_client.py index 59cd67c..2f9b9a7 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,4 +1,4 @@ -# Copyright (C) 2003-2007 Robey Pointer +# Copyright (C) 2003-2009 Robey Pointer # # This file is part of paramiko. # @@ -31,7 +31,7 @@ import paramiko class NullServer (paramiko.ServerInterface): - + def get_allowed_auths(self, username): if username == 'slowdive': return 'publickey,password' @@ -46,7 +46,7 @@ class NullServer (paramiko.ServerInterface): if (key.get_name() == 'ssh-dss') and (hexlify(key.get_fingerprint()) == '4478f0b9a23cc5182009ff755bc1d26c'): return paramiko.AUTH_SUCCESSFUL return paramiko.AUTH_FAILED - + def check_channel_request(self, kind, chanid): return paramiko.OPEN_SUCCEEDED @@ -81,17 +81,17 @@ class SSHClientTest (unittest.TestCase): self.ts.add_server_key(host_key) server = NullServer() self.ts.start_server(self.event, server) - - + + def test_1_client(self): """ verify that the SSHClient stuff works too. """ host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key') public_host_key = paramiko.RSAKey(data=str(host_key)) - + self.tc = paramiko.SSHClient() - self.tc.get_host_keys().add(self.addr, 'ssh-rsa', public_host_key) + self.tc.get_host_keys().add('[%s]:%d' % (self.addr, self.port), 'ssh-rsa', public_host_key) self.tc.connect(self.addr, self.port, username='slowdive', password='pygmalion') self.event.wait(1.0) @@ -111,20 +111,20 @@ class SSHClientTest (unittest.TestCase): self.assertEquals('', stdout.readline()) self.assertEquals('This is on stderr.\n', stderr.readline()) self.assertEquals('', stderr.readline()) - + stdin.close() stdout.close() stderr.close() - + def test_2_client_dsa(self): """ verify that SSHClient works with a DSA key. """ host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key') public_host_key = paramiko.RSAKey(data=str(host_key)) - + self.tc = paramiko.SSHClient() - self.tc.get_host_keys().add(self.addr, 'ssh-rsa', public_host_key) + self.tc.get_host_keys().add('[%s]:%d' % (self.addr, self.port), 'ssh-rsa', public_host_key) self.tc.connect(self.addr, self.port, username='slowdive', key_filename='tests/test_dss.key') self.event.wait(1.0) @@ -144,20 +144,20 @@ class SSHClientTest (unittest.TestCase): self.assertEquals('', stdout.readline()) self.assertEquals('This is on stderr.\n', stderr.readline()) self.assertEquals('', stderr.readline()) - + stdin.close() stdout.close() stderr.close() - + def test_3_multiple_key_files(self): """ verify that SSHClient accepts and tries multiple key files. """ host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key') public_host_key = paramiko.RSAKey(data=str(host_key)) - + self.tc = paramiko.SSHClient() - self.tc.get_host_keys().add(self.addr, 'ssh-rsa', public_host_key) + self.tc.get_host_keys().add('[%s]:%d' % (self.addr, self.port), 'ssh-rsa', public_host_key) self.tc.connect(self.addr, self.port, username='slowdive', key_filename=[ 'tests/test_rsa.key', 'tests/test_dss.key' ]) self.event.wait(1.0) @@ -172,7 +172,7 @@ class SSHClientTest (unittest.TestCase): """ host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key') public_host_key = paramiko.RSAKey(data=str(host_key)) - + self.tc = paramiko.SSHClient() self.tc.set_missing_host_key_policy(paramiko.AutoAddPolicy()) self.assertEquals(0, len(self.tc.get_host_keys())) @@ -184,7 +184,7 @@ class SSHClientTest (unittest.TestCase): self.assertEquals('slowdive', self.ts.get_username()) self.assertEquals(True, self.ts.is_authenticated()) self.assertEquals(1, len(self.tc.get_host_keys())) - self.assertEquals(public_host_key, self.tc.get_host_keys()[self.addr]['ssh-rsa']) + self.assertEquals(public_host_key, self.tc.get_host_keys()['[%s]:%d' % (self.addr, self.port)]['ssh-rsa']) def test_5_cleanup(self): """ @@ -193,7 +193,7 @@ class SSHClientTest (unittest.TestCase): """ host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key') public_host_key = paramiko.RSAKey(data=str(host_key)) - + self.tc = paramiko.SSHClient() self.tc.set_missing_host_key_policy(paramiko.AutoAddPolicy()) self.assertEquals(0, len(self.tc.get_host_keys())) @@ -202,7 +202,7 @@ class SSHClientTest (unittest.TestCase): self.event.wait(1.0) self.assert_(self.event.isSet()) self.assert_(self.ts.is_active()) - + p = weakref.ref(self.tc._transport.packetizer) self.assert_(p() is not None) del self.tc @@ -211,4 +211,4 @@ class SSHClientTest (unittest.TestCase): while (time.time() - st < 5.0) and (p() is not None): time.sleep(0.1) self.assert_(p() is None) - + diff --git a/tests/test_file.py b/tests/test_file.py index d66babf..c539b22 100755 --- a/tests/test_file.py +++ b/tests/test_file.py @@ -1,4 +1,4 @@ -# Copyright (C) 2003-2007 Robey Pointer +# Copyright (C) 2003-2009 Robey Pointer # # This file is part of paramiko. # diff --git a/tests/test_hostkeys.py b/tests/test_hostkeys.py index 28521ba..e28a41d 100644 --- a/tests/test_hostkeys.py +++ b/tests/test_hostkeys.py @@ -1,4 +1,4 @@ -# Copyright (C) 2006-2007 Robey Pointer +# Copyright (C) 2006-2007 Robey Pointer # # This file is part of paramiko. # diff --git a/tests/test_kex.py b/tests/test_kex.py index f304275..2ecb757 100644 --- a/tests/test_kex.py +++ b/tests/test_kex.py @@ -1,4 +1,4 @@ -# Copyright (C) 2003-2007 Robey Pointer +# Copyright (C) 2003-2009 Robey Pointer # # This file is part of paramiko. # diff --git a/tests/test_message.py b/tests/test_message.py index e930f71..7bfd44d 100644 --- a/tests/test_message.py +++ b/tests/test_message.py @@ -1,4 +1,4 @@ -# Copyright (C) 2003-2007 Robey Pointer +# Copyright (C) 2003-2009 Robey Pointer # # This file is part of paramiko. # diff --git a/tests/test_packetizer.py b/tests/test_packetizer.py index cb6248f..d1eb584 100644 --- a/tests/test_packetizer.py +++ b/tests/test_packetizer.py @@ -1,4 +1,4 @@ -# Copyright (C) 2003-2007 Robey Pointer +# Copyright (C) 2003-2009 Robey Pointer # # This file is part of paramiko. # diff --git a/tests/test_pkey.py b/tests/test_pkey.py index e591ab1..e40bee1 100644 --- a/tests/test_pkey.py +++ b/tests/test_pkey.py @@ -1,4 +1,4 @@ -# Copyright (C) 2003-2007 Robey Pointer +# Copyright (C) 2003-2009 Robey Pointer # # This file is part of paramiko. # diff --git a/tests/test_sftp.py b/tests/test_sftp.py index edc0599..f9d7270 100755 --- a/tests/test_sftp.py +++ b/tests/test_sftp.py @@ -1,4 +1,4 @@ -# Copyright (C) 2003-2007 Robey Pointer +# Copyright (C) 2003-2009 Robey Pointer # # This file is part of paramiko. # @@ -80,7 +80,7 @@ class SFTPTest (unittest.TestCase): def init(hostname, username, keyfile, passwd): global sftp, tc - + t = paramiko.Transport(hostname) tc = t try: @@ -136,7 +136,7 @@ class SFTPTest (unittest.TestCase): global g_big_file_test g_big_file_test = onoff set_big_file_test = staticmethod(set_big_file_test) - + def setUp(self): global FOLDER for i in xrange(1000): @@ -206,7 +206,7 @@ class SFTPTest (unittest.TestCase): f.close() finally: sftp.remove(FOLDER + '/append.txt') - + def test_5_rename(self): """ verify that renaming a file works. @@ -309,7 +309,7 @@ class SFTPTest (unittest.TestCase): self.assertEqual(stat.st_atime, atime) # can't really test chown, since we'd have to know a valid uid. - + sftp.truncate(FOLDER + '/special', 512) stat = sftp.stat(FOLDER + '/special') self.assertEqual(stat.st_size, 512) @@ -325,7 +325,7 @@ class SFTPTest (unittest.TestCase): try: f.write('x' * 1024) f.close() - + f = sftp.open(FOLDER + '/special', 'r+') stat = f.stat() f.chmod((stat.st_mode & ~0777) | 0600) @@ -348,16 +348,16 @@ class SFTPTest (unittest.TestCase): self.assertEqual(stat.st_mtime, mtime) if sys.platform not in ('win32', 'cygwin'): self.assertEqual(stat.st_atime, atime) - + # can't really test chown, since we'd have to know a valid uid. - + f.truncate(512) stat = f.stat() self.assertEqual(stat.st_size, 512) f.close() finally: sftp.remove(FOLDER + '/special') - + def test_A_readline_seek(self): """ create a text file and write a bunch of text into it. then count the lines @@ -510,7 +510,7 @@ class SFTPTest (unittest.TestCase): self.assert_(False, 'no exception removing nonexistent subfolder') except IOError: pass - + def test_G_chdir(self): """ verify that chdir/getcwd work. @@ -524,7 +524,7 @@ class SFTPTest (unittest.TestCase): sftp.mkdir('beta') self.assertEquals(root + FOLDER + '/alpha', sftp.getcwd()) self.assertEquals(['beta'], sftp.listdir('.')) - + sftp.chdir('beta') f = sftp.open('fish', 'w') f.write('hello\n') @@ -554,7 +554,7 @@ class SFTPTest (unittest.TestCase): """ import os, warnings warnings.filterwarnings('ignore', 'tempnam.*') - + localname = os.tempnam() text = 'All I wanted was a plastic bunny rabbit.\n' f = open(localname, 'wb') @@ -564,22 +564,22 @@ class SFTPTest (unittest.TestCase): def progress_callback(x, y): saved_progress.append((x, y)) sftp.put(localname, FOLDER + '/bunny.txt', progress_callback) - + f = sftp.open(FOLDER + '/bunny.txt', 'r') self.assertEquals(text, f.read(128)) f.close() self.assertEquals((41, 41), saved_progress[-1]) - + os.unlink(localname) localname = os.tempnam() saved_progress = [] sftp.get(FOLDER + '/bunny.txt', localname, progress_callback) - + f = open(localname, 'rb') self.assertEquals(text, f.read(128)) f.close() self.assertEquals((41, 41), saved_progress[-1]) - + os.unlink(localname) sftp.unlink(FOLDER + '/bunny.txt') @@ -592,7 +592,7 @@ class SFTPTest (unittest.TestCase): f = sftp.open(FOLDER + '/kitty.txt', 'w') f.write('here kitty kitty' * 64) f.close() - + try: f = sftp.open(FOLDER + '/kitty.txt', 'r') sum = f.check('sha1') @@ -612,7 +612,7 @@ class SFTPTest (unittest.TestCase): """ f = sftp.open(FOLDER + '/unusual.txt', 'wx') f.close() - + try: try: f = sftp.open(FOLDER + '/unusual.txt', 'wx') @@ -621,7 +621,7 @@ class SFTPTest (unittest.TestCase): pass finally: sftp.unlink(FOLDER + '/unusual.txt') - + def test_K_utf8(self): """ verify that unicode strings are encoded into utf8 correctly. @@ -629,7 +629,7 @@ class SFTPTest (unittest.TestCase): f = sftp.open(FOLDER + '/something', 'w') f.write('okay') f.close() - + try: sftp.rename(FOLDER + '/something', FOLDER + u'/\u00fcnic\u00f8de') sftp.open(FOLDER + '/\xc3\xbcnic\xc3\xb8\x64\x65', 'r') @@ -637,7 +637,19 @@ class SFTPTest (unittest.TestCase): self.fail('exception ' + e) sftp.unlink(FOLDER + '/\xc3\xbcnic\xc3\xb8\x64\x65') - def test_L_bad_readv(self): + def test_L_utf8_chdir(self): + sftp.mkdir(FOLDER + u'\u00fcnic\u00f8de') + try: + sftp.chdir(FOLDER + u'\u00fcnic\u00f8de') + f = sftp.open('something', 'w') + f.write('okay') + f.close() + sftp.unlink('something') + finally: + sftp.chdir(None) + sftp.rmdir(FOLDER + u'\u00fcnic\u00f8de') + + def test_M_bad_readv(self): """ verify that readv at the end of the file doesn't essplode. """ @@ -647,7 +659,7 @@ class SFTPTest (unittest.TestCase): f = sftp.open(FOLDER + '/zero', 'r') data = f.readv([(0, 12)]) f.close() - + f = sftp.open(FOLDER + '/zero', 'r') f.prefetch() data = f.read(100) @@ -658,7 +670,7 @@ class SFTPTest (unittest.TestCase): def XXX_test_M_seek_append(self): """ verify that seek does't affect writes during append. - + does not work except through paramiko. :( openssh fails. """ f = sftp.open(FOLDER + '/append.txt', 'a') diff --git a/tests/test_sftp_big.py b/tests/test_sftp_big.py index c182762..a32a700 100644 --- a/tests/test_sftp_big.py +++ b/tests/test_sftp_big.py @@ -1,4 +1,4 @@ -# Copyright (C) 2003-2007 Robey Pointer +# Copyright (C) 2003-2009 Robey Pointer # # This file is part of paramiko. # diff --git a/tests/test_transport.py b/tests/test_transport.py index 4b52c4f..cea4a1d 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -1,4 +1,4 @@ -# Copyright (C) 2003-2007 Robey Pointer +# Copyright (C) 2003-2009 Robey Pointer # # This file is part of paramiko. # @@ -38,6 +38,19 @@ from paramiko.message import Message from loop import LoopSocket +LONG_BANNER = """\ +Welcome to the super-fun-land BBS, where our MOTD is the primary thing we +provide. All rights reserved. Offer void in Tennessee. Stunt drivers were +used. Do not attempt at home. Some restrictions apply. + +Happy birthday to Commie the cat! + +Note: An SSH banner may eventually appear. + +Maybe. +""" + + class NullServer (ServerInterface): paranoid_did_password = False paranoid_did_public_key = False @@ -182,6 +195,24 @@ class TransportTest (unittest.TestCase): self.assertEquals(True, self.tc.is_authenticated()) self.assertEquals(True, self.ts.is_authenticated()) + def test_3a_long_banner(self): + """ + verify that a long banner doesn't mess up the handshake. + """ + host_key = RSAKey.from_private_key_file('tests/test_rsa.key') + public_host_key = RSAKey(data=str(host_key)) + self.ts.add_server_key(host_key) + event = threading.Event() + server = NullServer() + self.assert_(not event.isSet()) + self.socks.send(LONG_BANNER) + self.ts.start_server(event, server) + self.tc.connect(hostkey=public_host_key, + username='slowdive', password='pygmalion') + event.wait(1.0) + self.assert_(event.isSet()) + self.assert_(self.ts.is_active()) + def test_4_special(self): """ verify that the client can demand odd handshake settings, and can diff --git a/tests/test_util.py b/tests/test_util.py index d385bab..3569abf 100644 --- a/tests/test_util.py +++ b/tests/test_util.py @@ -1,4 +1,4 @@ -# Copyright (C) 2003-2007 Robey Pointer +# Copyright (C) 2003-2009 Robey Pointer # # This file is part of paramiko. # -- cgit v1.2.3