diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/__init__.py | 0 | ||||
| -rw-r--r-- | tests/loop.py | 10 | ||||
| -rw-r--r-- | tests/stub_sftp.py | 37 | ||||
| -rw-r--r-- | tests/test_auth.py | 59 | ||||
| -rw-r--r-- | tests/test_buffered_pipe.py | 46 | ||||
| -rw-r--r-- | tests/test_client.py | 158 | ||||
| -rw-r--r-- | tests/test_ecdsa.key | 5 | ||||
| -rw-r--r-- | tests/test_ecdsa_password.key | 8 | ||||
| -rwxr-xr-x | tests/test_file.py | 74 | ||||
| -rw-r--r-- | tests/test_hostkeys.py | 59 | ||||
| -rw-r--r-- | tests/test_kex.py | 156 | ||||
| -rw-r--r-- | tests/test_message.py | 79 | ||||
| -rw-r--r-- | tests/test_packetizer.py | 46 | ||||
| -rw-r--r-- | tests/test_pkey.py | 207 | ||||
| -rwxr-xr-x | tests/test_sftp.py | 523 | ||||
| -rw-r--r-- | tests/test_sftp_big.py | 320 | ||||
| -rw-r--r-- | tests/test_transport.py | 232 | ||||
| -rw-r--r-- | tests/test_util.py | 195 | ||||
| -rw-r--r-- | tests/util.py | 7 |
19 files changed, 1269 insertions, 952 deletions
diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/tests/__init__.py diff --git a/tests/loop.py b/tests/loop.py index ffa8e3c4..4f5dc163 100644 --- a/tests/loop.py +++ b/tests/loop.py @@ -7,7 +7,7 @@ # Software Foundation; either version 2.1 of the License, or (at your option) # any later version. # -# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY +# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR # A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # details. @@ -21,6 +21,7 @@ """ import threading, socket +from paramiko.common import asbytes class LoopSocket (object): @@ -31,7 +32,7 @@ class LoopSocket (object): """ def __init__(self): - self.__in_buffer = '' + self.__in_buffer = bytes() self.__lock = threading.Lock() self.__cv = threading.Condition(self.__lock) self.__timeout = None @@ -41,11 +42,12 @@ class LoopSocket (object): self.__unlink() try: self.__lock.acquire() - self.__in_buffer = '' + self.__in_buffer = bytes() finally: self.__lock.release() def send(self, data): + data = asbytes(data) if self.__mate is None: # EOF raise EOFError() @@ -57,7 +59,7 @@ class LoopSocket (object): try: if self.__mate is None: # EOF - return '' + return bytes() if len(self.__in_buffer) == 0: self.__cv.wait(self.__timeout) if len(self.__in_buffer) == 0: diff --git a/tests/stub_sftp.py b/tests/stub_sftp.py index 7f1ecc7e..47644433 100644 --- a/tests/stub_sftp.py +++ b/tests/stub_sftp.py @@ -7,7 +7,7 @@ # Software Foundation; either version 2.1 of the License, or (at your option) # any later version. # -# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY +# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR # A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # details. @@ -23,6 +23,7 @@ A stub SFTP server for loopback SFTP testing. import os from paramiko import ServerInterface, SFTPServerInterface, SFTPServer, SFTPAttributes, \ SFTPHandle, SFTP_OK, AUTH_SUCCESSFUL, OPEN_SUCCEEDED +from paramiko.common import o666 class StubServer (ServerInterface): @@ -38,7 +39,7 @@ class StubSFTPHandle (SFTPHandle): def stat(self): try: return SFTPAttributes.from_stat(os.fstat(self.readfile.fileno())) - except OSError, e: + except OSError as e: return SFTPServer.convert_errno(e.errno) def chattr(self, attr): @@ -47,7 +48,7 @@ class StubSFTPHandle (SFTPHandle): try: SFTPServer.set_file_attr(self.filename, attr) return SFTP_OK - except OSError, e: + except OSError as e: return SFTPServer.convert_errno(e.errno) @@ -62,34 +63,34 @@ class StubSFTPServer (SFTPServerInterface): def list_folder(self, path): path = self._realpath(path) try: - out = [ ] + out = [] flist = os.listdir(path) for fname in flist: attr = SFTPAttributes.from_stat(os.stat(os.path.join(path, fname))) attr.filename = fname out.append(attr) return out - except OSError, e: + except OSError as e: return SFTPServer.convert_errno(e.errno) def stat(self, path): path = self._realpath(path) try: return SFTPAttributes.from_stat(os.stat(path)) - except OSError, e: + except OSError as e: return SFTPServer.convert_errno(e.errno) def lstat(self, path): path = self._realpath(path) try: return SFTPAttributes.from_stat(os.lstat(path)) - except OSError, e: + except OSError as e: return SFTPServer.convert_errno(e.errno) def open(self, path, flags, attr): path = self._realpath(path) try: - binary_flag = getattr(os, 'O_BINARY', 0) + binary_flag = getattr(os, 'O_BINARY', 0) flags |= binary_flag mode = getattr(attr, 'st_mode', None) if mode is not None: @@ -97,8 +98,8 @@ class StubSFTPServer (SFTPServerInterface): else: # os.open() defaults to 0777 which is # an odd default mode for files - fd = os.open(path, flags, 0666) - except OSError, e: + fd = os.open(path, flags, o666) + except OSError as e: return SFTPServer.convert_errno(e.errno) if (flags & os.O_CREAT) and (attr is not None): attr._flags &= ~attr.FLAG_PERMISSIONS @@ -118,7 +119,7 @@ class StubSFTPServer (SFTPServerInterface): fstr = 'rb' try: f = os.fdopen(fd, fstr) - except OSError, e: + except OSError as e: return SFTPServer.convert_errno(e.errno) fobj = StubSFTPHandle(flags) fobj.filename = path @@ -130,7 +131,7 @@ class StubSFTPServer (SFTPServerInterface): path = self._realpath(path) try: os.remove(path) - except OSError, e: + except OSError as e: return SFTPServer.convert_errno(e.errno) return SFTP_OK @@ -139,7 +140,7 @@ class StubSFTPServer (SFTPServerInterface): newpath = self._realpath(newpath) try: os.rename(oldpath, newpath) - except OSError, e: + except OSError as e: return SFTPServer.convert_errno(e.errno) return SFTP_OK @@ -149,7 +150,7 @@ class StubSFTPServer (SFTPServerInterface): os.mkdir(path) if attr is not None: SFTPServer.set_file_attr(path, attr) - except OSError, e: + except OSError as e: return SFTPServer.convert_errno(e.errno) return SFTP_OK @@ -157,7 +158,7 @@ class StubSFTPServer (SFTPServerInterface): path = self._realpath(path) try: os.rmdir(path) - except OSError, e: + except OSError as e: return SFTPServer.convert_errno(e.errno) return SFTP_OK @@ -165,7 +166,7 @@ class StubSFTPServer (SFTPServerInterface): path = self._realpath(path) try: SFTPServer.set_file_attr(path, attr) - except OSError, e: + except OSError as e: return SFTPServer.convert_errno(e.errno) return SFTP_OK @@ -185,7 +186,7 @@ class StubSFTPServer (SFTPServerInterface): target_path = '<error>' try: os.symlink(target_path, path) - except OSError, e: + except OSError as e: return SFTPServer.convert_errno(e.errno) return SFTP_OK @@ -193,7 +194,7 @@ class StubSFTPServer (SFTPServerInterface): path = self._realpath(path) try: symlink = os.readlink(path) - except OSError, e: + except OSError as e: return SFTPServer.convert_errno(e.errno) # if it's absolute, remove the root if os.path.isabs(symlink): diff --git a/tests/test_auth.py b/tests/test_auth.py index 816e978b..1d972d53 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -7,7 +7,7 @@ # Software Foundation; either version 2.1 of the License, or (at your option) # any later version. # -# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY +# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR # A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # details. @@ -25,18 +25,21 @@ import threading import unittest from paramiko import Transport, ServerInterface, RSAKey, DSSKey, \ - SSHException, BadAuthenticationType, InteractiveQuery, ChannelException, \ + BadAuthenticationType, InteractiveQuery, \ AuthenticationException from paramiko import AUTH_FAILED, AUTH_PARTIALLY_SUCCESSFUL, AUTH_SUCCESSFUL -from paramiko import OPEN_SUCCEEDED, OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED -from loop import LoopSocket +from paramiko.py3compat import u +from tests.loop import LoopSocket +from tests.util import test_path + +_pwd = u('\u2022') class NullServer (ServerInterface): paranoid_did_password = False paranoid_did_public_key = False - paranoid_key = DSSKey.from_private_key_file('tests/test_dss.key') - + paranoid_key = DSSKey.from_private_key_file(test_path('test_dss.key')) + def get_allowed_auths(self, username): if username == 'slowdive': return 'publickey,password' @@ -64,7 +67,7 @@ class NullServer (ServerInterface): if self.paranoid_did_public_key: return AUTH_SUCCESSFUL return AUTH_PARTIALLY_SUCCESSFUL - if (username == 'utf8') and (password == u'\u2022'): + if (username == 'utf8') and (password == _pwd): return AUTH_SUCCESSFUL if (username == 'non-utf8') and (password == '\xff'): return AUTH_SUCCESSFUL @@ -110,18 +113,18 @@ class AuthTest (unittest.TestCase): self.sockc.close() def start_server(self): - host_key = RSAKey.from_private_key_file('tests/test_rsa.key') - self.public_host_key = RSAKey(data=str(host_key)) + host_key = RSAKey.from_private_key_file(test_path('test_rsa.key')) + self.public_host_key = RSAKey(data=host_key.asbytes()) self.ts.add_server_key(host_key) self.event = threading.Event() self.server = NullServer() - self.assert_(not self.event.isSet()) + self.assertTrue(not self.event.isSet()) self.ts.start_server(self.event, self.server) def verify_finished(self): self.event.wait(1.0) - self.assert_(self.event.isSet()) - self.assert_(self.ts.is_active()) + self.assertTrue(self.event.isSet()) + self.assertTrue(self.ts.is_active()) def test_1_bad_auth_type(self): """ @@ -132,11 +135,11 @@ class AuthTest (unittest.TestCase): try: self.tc.connect(hostkey=self.public_host_key, username='unknown', password='error') - self.assert_(False) + self.assertTrue(False) except: etype, evalue, etb = sys.exc_info() - self.assertEquals(BadAuthenticationType, etype) - self.assertEquals(['publickey'], evalue.allowed_types) + self.assertEqual(BadAuthenticationType, etype) + self.assertEqual(['publickey'], evalue.allowed_types) def test_2_bad_password(self): """ @@ -147,10 +150,10 @@ class AuthTest (unittest.TestCase): self.tc.connect(hostkey=self.public_host_key) try: self.tc.auth_password(username='slowdive', password='error') - self.assert_(False) + self.assertTrue(False) except: etype, evalue, etb = sys.exc_info() - self.assert_(issubclass(etype, AuthenticationException)) + self.assertTrue(issubclass(etype, AuthenticationException)) self.tc.auth_password(username='slowdive', password='pygmalion') self.verify_finished() @@ -161,10 +164,10 @@ class AuthTest (unittest.TestCase): self.start_server() self.tc.connect(hostkey=self.public_host_key) remain = self.tc.auth_password(username='paranoid', password='paranoid') - self.assertEquals(['publickey'], remain) - key = DSSKey.from_private_key_file('tests/test_dss.key') + self.assertEqual(['publickey'], remain) + key = DSSKey.from_private_key_file(test_path('test_dss.key')) remain = self.tc.auth_publickey(username='paranoid', key=key) - self.assertEquals([], remain) + self.assertEqual([], remain) self.verify_finished() def test_4_interactive_auth(self): @@ -180,9 +183,9 @@ class AuthTest (unittest.TestCase): self.got_prompts = prompts return ['cat'] remain = self.tc.auth_interactive('commie', handler) - self.assertEquals(self.got_title, 'password') - self.assertEquals(self.got_prompts, [('Password', False)]) - self.assertEquals([], remain) + self.assertEqual(self.got_title, 'password') + self.assertEqual(self.got_prompts, [('Password', False)]) + self.assertEqual([], remain) self.verify_finished() def test_5_interactive_auth_fallback(self): @@ -193,7 +196,7 @@ class AuthTest (unittest.TestCase): self.start_server() self.tc.connect(hostkey=self.public_host_key) remain = self.tc.auth_password('commie', 'cat') - self.assertEquals([], remain) + self.assertEqual([], remain) self.verify_finished() def test_6_auth_utf8(self): @@ -202,8 +205,8 @@ class AuthTest (unittest.TestCase): """ self.start_server() self.tc.connect(hostkey=self.public_host_key) - remain = self.tc.auth_password('utf8', u'\u2022') - self.assertEquals([], remain) + remain = self.tc.auth_password('utf8', _pwd) + self.assertEqual([], remain) self.verify_finished() def test_7_auth_non_utf8(self): @@ -214,7 +217,7 @@ class AuthTest (unittest.TestCase): self.start_server() self.tc.connect(hostkey=self.public_host_key) remain = self.tc.auth_password('non-utf8', '\xff') - self.assertEquals([], remain) + self.assertEqual([], remain) self.verify_finished() def test_8_auth_gets_disconnected(self): @@ -228,4 +231,4 @@ class AuthTest (unittest.TestCase): remain = self.tc.auth_password('bad-server', 'hello') except: etype, evalue, etb = sys.exc_info() - self.assert_(issubclass(etype, AuthenticationException)) + self.assertTrue(issubclass(etype, AuthenticationException)) diff --git a/tests/test_buffered_pipe.py b/tests/test_buffered_pipe.py index b9d91f6a..a53081a9 100644 --- a/tests/test_buffered_pipe.py +++ b/tests/test_buffered_pipe.py @@ -7,7 +7,7 @@ # Software Foundation; either version 2.1 of the License, or (at your option) # any later version. # -# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY +# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR # A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # details. @@ -22,61 +22,60 @@ Some unit tests for BufferedPipe. import threading import time -import unittest from paramiko.buffered_pipe import BufferedPipe, PipeTimeout from paramiko import pipe -from util import ParamikoTest +from tests.util import ParamikoTest -def delay_thread(pipe): - pipe.feed('a') +def delay_thread(p): + p.feed('a') time.sleep(0.5) - pipe.feed('b') - pipe.close() + p.feed('b') + p.close() -def close_thread(pipe): +def close_thread(p): time.sleep(0.2) - pipe.close() + p.close() class BufferedPipeTest(ParamikoTest): def test_1_buffered_pipe(self): p = BufferedPipe() - self.assert_(not p.read_ready()) + self.assertTrue(not p.read_ready()) p.feed('hello.') - self.assert_(p.read_ready()) + self.assertTrue(p.read_ready()) data = p.read(6) - self.assertEquals('hello.', data) + self.assertEqual(b'hello.', data) p.feed('plus/minus') - self.assertEquals('plu', p.read(3)) - self.assertEquals('s/m', p.read(3)) - self.assertEquals('inus', p.read(4)) + self.assertEqual(b'plu', p.read(3)) + self.assertEqual(b's/m', p.read(3)) + self.assertEqual(b'inus', p.read(4)) p.close() - self.assert_(not p.read_ready()) - self.assertEquals('', p.read(1)) + self.assertTrue(not p.read_ready()) + self.assertEqual(b'', p.read(1)) def test_2_delay(self): p = BufferedPipe() - self.assert_(not p.read_ready()) + self.assertTrue(not p.read_ready()) threading.Thread(target=delay_thread, args=(p,)).start() - self.assertEquals('a', p.read(1, 0.1)) + self.assertEqual(b'a', p.read(1, 0.1)) try: p.read(1, 0.1) - self.assert_(False) + self.assertTrue(False) except PipeTimeout: pass - self.assertEquals('b', p.read(1, 1.0)) - self.assertEquals('', p.read(1)) + self.assertEqual(b'b', p.read(1, 1.0)) + self.assertEqual(b'', p.read(1)) def test_3_close_while_reading(self): p = BufferedPipe() threading.Thread(target=close_thread, args=(p,)).start() data = p.read(1, 1.0) - self.assertEquals('', data) + self.assertEqual(b'', data) def test_4_or_pipe(self): p = pipe.make_pipe() @@ -90,4 +89,3 @@ class BufferedPipeTest(ParamikoTest): self.assertTrue(p._set) p2.clear() self.assertFalse(p._set) - diff --git a/tests/test_client.py b/tests/test_client.py index 08ef1f92..7e5c80b4 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -7,7 +7,7 @@ # Software Foundation; either version 2.1 of the License, or (at your option) # any later version. # -# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY +# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR # A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # details. @@ -21,13 +21,15 @@ Some unit tests for SSHClient. """ import socket +from tempfile import mkstemp import threading -import time import unittest import weakref -from binascii import hexlify - +import warnings +import os +from tests.util import test_path import paramiko +from paramiko.common import PY2 class NullServer (paramiko.ServerInterface): @@ -43,7 +45,7 @@ class NullServer (paramiko.ServerInterface): return paramiko.AUTH_FAILED def check_auth_publickey(self, username, key): - if (key.get_name() == 'ssh-dss') and (hexlify(key.get_fingerprint()) == '4478f0b9a23cc5182009ff755bc1d26c'): + if (key.get_name() == 'ssh-dss') and key.get_fingerprint() == b'\x44\x78\xf0\xb9\xa2\x3c\xc5\x18\x20\x09\xff\x75\x5b\xc1\xd2\x6c': return paramiko.AUTH_SUCCESSFUL return paramiko.AUTH_FAILED @@ -64,8 +66,6 @@ class SSHClientTest (unittest.TestCase): self.sockl.listen(1) self.addr, self.port = self.sockl.getsockname() self.event = threading.Event() - thread = threading.Thread(target=self._run) - thread.start() def tearDown(self): for attr in "tc ts socks sockl".split(): @@ -75,28 +75,28 @@ class SSHClientTest (unittest.TestCase): def _run(self): self.socks, addr = self.sockl.accept() self.ts = paramiko.Transport(self.socks) - host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key') + host_key = paramiko.RSAKey.from_private_key_file(test_path('test_rsa.key')) 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)) + threading.Thread(target=self._run).start() + host_key = paramiko.RSAKey.from_private_key_file(test_path('test_rsa.key')) + public_host_key = paramiko.RSAKey(data=host_key.asbytes()) self.tc = paramiko.SSHClient() 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) - self.assert_(self.event.isSet()) - self.assert_(self.ts.is_active()) - self.assertEquals('slowdive', self.ts.get_username()) - self.assertEquals(True, self.ts.is_authenticated()) + self.assertTrue(self.event.isSet()) + self.assertTrue(self.ts.is_active()) + self.assertEqual('slowdive', self.ts.get_username()) + self.assertEqual(True, self.ts.is_authenticated()) stdin, stdout, stderr = self.tc.exec_command('yes') schan = self.ts.accept(1.0) @@ -105,10 +105,10 @@ class SSHClientTest (unittest.TestCase): schan.send_stderr('This is on stderr.\n') schan.close() - self.assertEquals('Hello there.\n', stdout.readline()) - self.assertEquals('', stdout.readline()) - self.assertEquals('This is on stderr.\n', stderr.readline()) - self.assertEquals('', stderr.readline()) + self.assertEqual('Hello there.\n', stdout.readline()) + self.assertEqual('', stdout.readline()) + self.assertEqual('This is on stderr.\n', stderr.readline()) + self.assertEqual('', stderr.readline()) stdin.close() stdout.close() @@ -118,18 +118,19 @@ class SSHClientTest (unittest.TestCase): """ 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)) + threading.Thread(target=self._run).start() + host_key = paramiko.RSAKey.from_private_key_file(test_path('test_rsa.key')) + public_host_key = paramiko.RSAKey(data=host_key.asbytes()) self.tc = paramiko.SSHClient() 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.tc.connect(self.addr, self.port, username='slowdive', key_filename=test_path('test_dss.key')) self.event.wait(1.0) - self.assert_(self.event.isSet()) - self.assert_(self.ts.is_active()) - self.assertEquals('slowdive', self.ts.get_username()) - self.assertEquals(True, self.ts.is_authenticated()) + self.assertTrue(self.event.isSet()) + self.assertTrue(self.ts.is_active()) + self.assertEqual('slowdive', self.ts.get_username()) + self.assertEqual(True, self.ts.is_authenticated()) stdin, stdout, stderr = self.tc.exec_command('yes') schan = self.ts.accept(1.0) @@ -138,10 +139,10 @@ class SSHClientTest (unittest.TestCase): schan.send_stderr('This is on stderr.\n') schan.close() - self.assertEquals('Hello there.\n', stdout.readline()) - self.assertEquals('', stdout.readline()) - self.assertEquals('This is on stderr.\n', stderr.readline()) - self.assertEquals('', stderr.readline()) + self.assertEqual('Hello there.\n', stdout.readline()) + self.assertEqual('', stdout.readline()) + self.assertEqual('This is on stderr.\n', stderr.readline()) + self.assertEqual('', stderr.readline()) stdin.close() stdout.close() @@ -151,62 +152,103 @@ class SSHClientTest (unittest.TestCase): """ 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)) + threading.Thread(target=self._run).start() + host_key = paramiko.RSAKey.from_private_key_file(test_path('test_rsa.key')) + public_host_key = paramiko.RSAKey(data=host_key.asbytes()) self.tc = paramiko.SSHClient() 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.tc.connect(self.addr, self.port, username='slowdive', key_filename=[test_path('test_rsa.key'), test_path('test_dss.key')]) self.event.wait(1.0) - self.assert_(self.event.isSet()) - self.assert_(self.ts.is_active()) - self.assertEquals('slowdive', self.ts.get_username()) - self.assertEquals(True, self.ts.is_authenticated()) + self.assertTrue(self.event.isSet()) + self.assertTrue(self.ts.is_active()) + self.assertEqual('slowdive', self.ts.get_username()) + self.assertEqual(True, self.ts.is_authenticated()) def test_4_auto_add_policy(self): """ verify that SSHClient's AutoAddPolicy works. """ - host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key') - public_host_key = paramiko.RSAKey(data=str(host_key)) + threading.Thread(target=self._run).start() + host_key = paramiko.RSAKey.from_private_key_file(test_path('test_rsa.key')) + public_host_key = paramiko.RSAKey(data=host_key.asbytes()) self.tc = paramiko.SSHClient() self.tc.set_missing_host_key_policy(paramiko.AutoAddPolicy()) - self.assertEquals(0, len(self.tc.get_host_keys())) + self.assertEqual(0, len(self.tc.get_host_keys())) self.tc.connect(self.addr, self.port, username='slowdive', password='pygmalion') self.event.wait(1.0) - self.assert_(self.event.isSet()) - self.assert_(self.ts.is_active()) - 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()['[%s]:%d' % (self.addr, self.port)]['ssh-rsa']) - - def test_5_cleanup(self): + self.assertTrue(self.event.isSet()) + self.assertTrue(self.ts.is_active()) + self.assertEqual('slowdive', self.ts.get_username()) + self.assertEqual(True, self.ts.is_authenticated()) + self.assertEqual(1, len(self.tc.get_host_keys())) + self.assertEqual(public_host_key, self.tc.get_host_keys()['[%s]:%d' % (self.addr, self.port)]['ssh-rsa']) + + def test_5_save_host_keys(self): + """ + verify that SSHClient correctly saves a known_hosts file. + """ + warnings.filterwarnings('ignore', 'tempnam.*') + + host_key = paramiko.RSAKey.from_private_key_file(test_path('test_rsa.key')) + public_host_key = paramiko.RSAKey(data=host_key.asbytes()) + fd, localname = mkstemp() + os.close(fd) + + client = paramiko.SSHClient() + self.assertEquals(0, len(client.get_host_keys())) + + host_id = '[%s]:%d' % (self.addr, self.port) + + client.get_host_keys().add(host_id, 'ssh-rsa', public_host_key) + self.assertEquals(1, len(client.get_host_keys())) + self.assertEquals(public_host_key, client.get_host_keys()[host_id]['ssh-rsa']) + + client.save_host_keys(localname) + + with open(localname) as fd: + assert host_id in fd.read() + + os.unlink(localname) + + def test_6_cleanup(self): """ verify that when an SSHClient is collected, its transport (and the transport's packetizer) is closed. """ - host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key') - public_host_key = paramiko.RSAKey(data=str(host_key)) + # Unclear why this is borked on Py3, but it is, and does not seem worth + # pursuing at the moment. + if not PY2: + return + threading.Thread(target=self._run).start() + host_key = paramiko.RSAKey.from_private_key_file(test_path('test_rsa.key')) + public_host_key = paramiko.RSAKey(data=host_key.asbytes()) self.tc = paramiko.SSHClient() self.tc.set_missing_host_key_policy(paramiko.AutoAddPolicy()) - self.assertEquals(0, len(self.tc.get_host_keys())) + self.assertEqual(0, len(self.tc.get_host_keys())) self.tc.connect(self.addr, self.port, username='slowdive', password='pygmalion') self.event.wait(1.0) - self.assert_(self.event.isSet()) - self.assert_(self.ts.is_active()) + self.assertTrue(self.event.isSet()) + self.assertTrue(self.ts.is_active()) p = weakref.ref(self.tc._transport.packetizer) - self.assert_(p() is not None) + self.assertTrue(p() is not None) + self.tc.close() del self.tc + # hrm, sometimes p isn't cleared right away. why is that? - st = time.time() - while (time.time() - st < 5.0) and (p() is not None): - time.sleep(0.1) - self.assert_(p() is None) + #st = time.time() + #while (time.time() - st < 5.0) and (p() is not None): + # time.sleep(0.1) + + # instead of dumbly waiting for the GC to collect, force a collection + # to see whether the SSHClient object is deallocated correctly + import gc + gc.collect() + self.assertTrue(p() is None) diff --git a/tests/test_ecdsa.key b/tests/test_ecdsa.key new file mode 100644 index 00000000..42d44734 --- /dev/null +++ b/tests/test_ecdsa.key @@ -0,0 +1,5 @@ +-----BEGIN EC PRIVATE KEY----- +MHcCAQEEIKB6ty3yVyKEnfF/zprx0qwC76MsMlHY4HXCnqho2eKioAoGCCqGSM49 +AwEHoUQDQgAElI9mbdlaS+T9nHxY/59lFnn80EEecZDBHq4gLpccY8Mge5ZTMiMD +ADRvOqQ5R98Sxst765CAqXmRtz8vwoD96g== +-----END EC PRIVATE KEY----- diff --git a/tests/test_ecdsa_password.key b/tests/test_ecdsa_password.key new file mode 100644 index 00000000..eb7910ed --- /dev/null +++ b/tests/test_ecdsa_password.key @@ -0,0 +1,8 @@ +-----BEGIN EC PRIVATE KEY----- +Proc-Type: 4,ENCRYPTED +DEK-Info: AES-128-CBC,EEB56BC745EDB2DE04FC3FE1F8DA387E + +wdt7QTCa6ahTJLaEPH7NhHyBcxhzrzf93d4UwQOuAhkM6//jKD4lF9fErHBW0f3B +ExberCU3UxfEF3xX2thXiLw47JgeOCeQUlqRFx92p36k6YmfNGX6W8CsZ3d+XodF +Z+pb6m285CiSX+W95NenFMexXFsIpntiCvTifTKJ8os= +-----END EC PRIVATE KEY----- diff --git a/tests/test_file.py b/tests/test_file.py index c539b221..22a34aca 100755 --- a/tests/test_file.py +++ b/tests/test_file.py @@ -7,7 +7,7 @@ # Software Foundation; either version 2.1 of the License, or (at your option) # any later version. # -# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY +# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR # A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # details. @@ -22,6 +22,8 @@ Some unit tests for the BufferedFile abstraction. import unittest from paramiko.file import BufferedFile +from paramiko.common import linefeed_byte, crlf, cr_byte +import sys class LoopbackFile (BufferedFile): @@ -31,7 +33,7 @@ class LoopbackFile (BufferedFile): def __init__(self, mode='r', bufsize=-1): BufferedFile.__init__(self) self._set_mode(mode, bufsize) - self.buffer = '' + self.buffer = bytes() def _read(self, size): if len(self.buffer) == 0: @@ -52,8 +54,8 @@ class BufferedFileTest (unittest.TestCase): def test_1_simple(self): f = LoopbackFile('r') try: - f.write('hi') - self.assert_(False, 'no exception on write to read-only file') + f.write(b'hi') + self.assertTrue(False, 'no exception on write to read-only file') except: pass f.close() @@ -61,14 +63,14 @@ class BufferedFileTest (unittest.TestCase): f = LoopbackFile('w') try: f.read(1) - self.assert_(False, 'no exception to read from write-only file') + self.assertTrue(False, 'no exception to read from write-only file') except: pass f.close() def test_2_readline(self): f = LoopbackFile('r+U') - f.write('First line.\nSecond line.\r\nThird line.\nFinal line non-terminated.') + f.write(b'First line.\nSecond line.\r\nThird line.\nFinal line non-terminated.') self.assertEqual(f.readline(), 'First line.\n') # universal newline mode should convert this linefeed: self.assertEqual(f.readline(), 'Second line.\n') @@ -80,31 +82,31 @@ class BufferedFileTest (unittest.TestCase): f.close() try: f.readline() - self.assert_(False, 'no exception on readline of closed file') + self.assertTrue(False, 'no exception on readline of closed file') except IOError: pass - self.assert_('\n' in f.newlines) - self.assert_('\r\n' in f.newlines) - self.assert_('\r' not in f.newlines) + self.assertTrue(linefeed_byte in f.newlines) + self.assertTrue(crlf in f.newlines) + self.assertTrue(cr_byte not in f.newlines) def test_3_lf(self): """ try to trick the linefeed detector. """ f = LoopbackFile('r+U') - f.write('First line.\r') + f.write(b'First line.\r') self.assertEqual(f.readline(), 'First line.\n') - f.write('\nSecond.\r\n') + f.write(b'\nSecond.\r\n') self.assertEqual(f.readline(), 'Second.\n') f.close() - self.assertEqual(f.newlines, '\r\n') + self.assertEqual(f.newlines, crlf) def test_4_write(self): """ verify that write buffering is on. """ f = LoopbackFile('r+', 1) - f.write('Complete line.\nIncomplete line.') + f.write(b'Complete line.\nIncomplete line.') self.assertEqual(f.readline(), 'Complete line.\n') self.assertEqual(f.readline(), '') f.write('..\n') @@ -117,12 +119,12 @@ class BufferedFileTest (unittest.TestCase): """ f = LoopbackFile('r+', 512) f.write('Not\nquite\n512 bytes.\n') - self.assertEqual(f.read(1), '') + self.assertEqual(f.read(1), b'') f.flush() - self.assertEqual(f.read(5), 'Not\nq') - self.assertEqual(f.read(10), 'uite\n512 b') - self.assertEqual(f.read(9), 'ytes.\n') - self.assertEqual(f.read(3), '') + self.assertEqual(f.read(5), b'Not\nq') + self.assertEqual(f.read(10), b'uite\n512 b') + self.assertEqual(f.read(9), b'ytes.\n') + self.assertEqual(f.read(3), b'') f.close() def test_6_buffering(self): @@ -130,12 +132,12 @@ class BufferedFileTest (unittest.TestCase): verify that flushing happens automatically on buffer crossing. """ f = LoopbackFile('r+', 16) - f.write('Too small.') - self.assertEqual(f.read(4), '') - f.write(' ') - self.assertEqual(f.read(4), '') - f.write('Enough.') - self.assertEqual(f.read(20), 'Too small. Enough.') + f.write(b'Too small.') + self.assertEqual(f.read(4), b'') + f.write(b' ') + self.assertEqual(f.read(4), b'') + f.write(b'Enough.') + self.assertEqual(f.read(20), b'Too small. Enough.') f.close() def test_7_read_all(self): @@ -143,9 +145,23 @@ class BufferedFileTest (unittest.TestCase): verify that read(-1) returns everything left in the file. """ f = LoopbackFile('r+', 16) - f.write('The first thing you need to do is open your eyes. ') - f.write('Then, you need to close them again.\n') + f.write(b'The first thing you need to do is open your eyes. ') + f.write(b'Then, you need to close them again.\n') s = f.read(-1) - self.assertEqual(s, 'The first thing you need to do is open your eyes. Then, you ' + - 'need to close them again.\n') + self.assertEqual(s, b'The first thing you need to do is open your eyes. Then, you ' + + b'need to close them again.\n') f.close() + + def test_8_buffering(self): + """ + verify that buffered objects can be written + """ + if sys.version_info[0] == 2: + f = LoopbackFile('r+', 16) + f.write(buffer(b'Too small.')) + f.close() + +if __name__ == '__main__': + from unittest import main + main() + diff --git a/tests/test_hostkeys.py b/tests/test_hostkeys.py index e28a41d3..0ee1bbf0 100644 --- a/tests/test_hostkeys.py +++ b/tests/test_hostkeys.py @@ -7,7 +7,7 @@ # Software Foundation; either version 2.1 of the License, or (at your option) # any later version. # -# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY +# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR # A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # details. @@ -20,11 +20,11 @@ Some unit tests for HostKeys. """ -import base64 from binascii import hexlify import os import unittest import paramiko +from paramiko.py3compat import decodebytes test_hosts_file = """\ @@ -36,12 +36,12 @@ BGQ3GQ/Fc7SX6gkpXkwcZryoi4kNFhHu5LvHcZPdxXV1D+uTMfGS1eyd2Yz/DoNWXNAl8TI0cAsW\ 5ymME3bQ4J/k1IKxCtz/bAlAqFgKoc+EolMziDYqWIATtW0rYTJvzGAzTmMj80/QpsFH+Pc2M= """ -keyblob = """\ +keyblob = b"""\ AAAAB3NzaC1yc2EAAAABIwAAAIEA8bP1ZA7DCZDB9J0s50l31MBGQ3GQ/Fc7SX6gkpXkwcZryoi4k\ NFhHu5LvHcZPdxXV1D+uTMfGS1eyd2Yz/DoNWXNAl8TI0cAsW5ymME3bQ4J/k1IKxCtz/bAlAqFgK\ oc+EolMziDYqWIATtW0rYTJvzGAzTmMj80/QpsFH+Pc2M=""" -keyblob_dss = """\ +keyblob_dss = b"""\ AAAAB3NzaC1kc3MAAACBAOeBpgNnfRzr/twmAQRu2XwWAp3CFtrVnug6s6fgwj/oLjYbVtjAy6pl/\ h0EKCWx2rf1IetyNsTxWrniA9I6HeDj65X1FyDkg6g8tvCnaNB8Xp/UUhuzHuGsMIipRxBxw9LF60\ 8EqZcj1E3ytktoW5B5OcjrkEoz3xG7C+rpIjYvAAAAFQDwz4UnmsGiSNu5iqjn3uTzwUpshwAAAIE\ @@ -55,51 +55,50 @@ Ngw3qIch/WgRmMHy4kBq1SsXMjQCte1So6HBMvBPIW5SiMTmjCfZZiw4AYHK+B/JaOwaG9yRg2Ejg\ class HostKeysTest (unittest.TestCase): def setUp(self): - f = open('hostfile.temp', 'w') - f.write(test_hosts_file) - f.close() + with open('hostfile.temp', 'w') as f: + f.write(test_hosts_file) def tearDown(self): os.unlink('hostfile.temp') def test_1_load(self): hostdict = paramiko.HostKeys('hostfile.temp') - self.assertEquals(2, len(hostdict)) - self.assertEquals(1, len(hostdict.values()[0])) - self.assertEquals(1, len(hostdict.values()[1])) + self.assertEqual(2, len(hostdict)) + self.assertEqual(1, len(list(hostdict.values())[0])) + self.assertEqual(1, len(list(hostdict.values())[1])) fp = hexlify(hostdict['secure.example.com']['ssh-rsa'].get_fingerprint()).upper() - self.assertEquals('E6684DB30E109B67B70FF1DC5C7F1363', fp) + self.assertEqual(b'E6684DB30E109B67B70FF1DC5C7F1363', fp) def test_2_add(self): hostdict = paramiko.HostKeys('hostfile.temp') hh = '|1|BMsIC6cUIP2zBuXR3t2LRcJYjzM=|hpkJMysjTk/+zzUUzxQEa2ieq6c=' - key = paramiko.RSAKey(data=base64.decodestring(keyblob)) + key = paramiko.RSAKey(data=decodebytes(keyblob)) hostdict.add(hh, 'ssh-rsa', key) - self.assertEquals(3, len(hostdict)) + self.assertEqual(3, len(list(hostdict))) x = hostdict['foo.example.com'] fp = hexlify(x['ssh-rsa'].get_fingerprint()).upper() - self.assertEquals('7EC91BB336CB6D810B124B1353C32396', fp) - self.assert_(hostdict.check('foo.example.com', key)) + self.assertEqual(b'7EC91BB336CB6D810B124B1353C32396', fp) + self.assertTrue(hostdict.check('foo.example.com', key)) def test_3_dict(self): hostdict = paramiko.HostKeys('hostfile.temp') - self.assert_('secure.example.com' in hostdict) - self.assert_('not.example.com' not in hostdict) - self.assert_(hostdict.has_key('secure.example.com')) - self.assert_(not hostdict.has_key('not.example.com')) + self.assertTrue('secure.example.com' in hostdict) + self.assertTrue('not.example.com' not in hostdict) + self.assertTrue('secure.example.com' in hostdict) + self.assertTrue('not.example.com' not in hostdict) x = hostdict.get('secure.example.com', None) - self.assert_(x is not None) + self.assertTrue(x is not None) fp = hexlify(x['ssh-rsa'].get_fingerprint()).upper() - self.assertEquals('E6684DB30E109B67B70FF1DC5C7F1363', fp) + self.assertEqual(b'E6684DB30E109B67B70FF1DC5C7F1363', fp) i = 0 for key in hostdict: i += 1 - self.assertEquals(2, i) + self.assertEqual(2, i) def test_4_dict_set(self): hostdict = paramiko.HostKeys('hostfile.temp') - key = paramiko.RSAKey(data=base64.decodestring(keyblob)) - key_dss = paramiko.DSSKey(data=base64.decodestring(keyblob_dss)) + key = paramiko.RSAKey(data=decodebytes(keyblob)) + key_dss = paramiko.DSSKey(data=decodebytes(keyblob_dss)) hostdict['secure.example.com'] = { 'ssh-rsa': key, 'ssh-dss': key_dss @@ -107,11 +106,11 @@ class HostKeysTest (unittest.TestCase): hostdict['fake.example.com'] = {} hostdict['fake.example.com']['ssh-rsa'] = key - self.assertEquals(3, len(hostdict)) - self.assertEquals(2, len(hostdict.values()[0])) - self.assertEquals(1, len(hostdict.values()[1])) - self.assertEquals(1, len(hostdict.values()[2])) + self.assertEqual(3, len(hostdict)) + self.assertEqual(2, len(list(hostdict.values())[0])) + self.assertEqual(1, len(list(hostdict.values())[1])) + self.assertEqual(1, len(list(hostdict.values())[2])) fp = hexlify(hostdict['secure.example.com']['ssh-rsa'].get_fingerprint()).upper() - self.assertEquals('7EC91BB336CB6D810B124B1353C32396', fp) + self.assertEqual(b'7EC91BB336CB6D810B124B1353C32396', fp) fp = hexlify(hostdict['secure.example.com']['ssh-dss'].get_fingerprint()).upper() - self.assertEquals('4478F0B9A23CC5182009FF755BC1D26C', fp) + self.assertEqual(b'4478F0B9A23CC5182009FF755BC1D26C', fp) diff --git a/tests/test_kex.py b/tests/test_kex.py index f6e69960..56f1b7c7 100644 --- a/tests/test_kex.py +++ b/tests/test_kex.py @@ -7,7 +7,7 @@ # Software Foundation; either version 2.1 of the License, or (at your option) # any later version. # -# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY +# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR # A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # details. @@ -21,34 +21,40 @@ Some unit tests for the key exchange protocols. """ from binascii import hexlify +import os import unittest + import paramiko.util from paramiko.kex_group1 import KexGroup1 from paramiko.kex_gex import KexGex from paramiko import Message +from paramiko.common import byte_chr -class FakeRng (object): - def read(self, n): - return chr(0xcc) * n +def dummy_urandom(n): + return byte_chr(0xcc) * n class FakeKey (object): def __str__(self): return 'fake-key' - def sign_ssh_data(self, rng, H): - return 'fake-sig' + + def asbytes(self): + return b'fake-key' + + def sign_ssh_data(self, H): + return b'fake-sig' class FakeModulusPack (object): - P = 0xFFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7EDEE386BFB5A899FA5AE9F24117C4B1FE649286651ECE65381FFFFFFFFFFFFFFFFL + P = 0xFFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7EDEE386BFB5A899FA5AE9F24117C4B1FE649286651ECE65381FFFFFFFFFFFFFFFF G = 2 + def get_modulus(self, min, ask, max): return self.G, self.P -class FakeTransport (object): - rng = FakeRng() +class FakeTransport(object): local_version = 'SSH-2.0-paramiko_1.0' remote_version = 'SSH-2.0-lame' local_kex_init = 'local-kex-init' @@ -56,41 +62,49 @@ class FakeTransport (object): def _send_message(self, m): self._message = m + def _expect_packet(self, *t): self._expect = t + def _set_K_H(self, K, H): self._K = K self._H = H + def _verify_key(self, host_key, sig): self._verify = (host_key, sig) + def _activate_outbound(self): self._activated = True + def _log(self, level, s): pass + def get_server_key(self): return FakeKey() + def _get_modulus_pack(self): return FakeModulusPack() class KexTest (unittest.TestCase): - K = 14730343317708716439807310032871972459448364195094179797249681733965528989482751523943515690110179031004049109375612685505881911274101441415545039654102474376472240501616988799699744135291070488314748284283496055223852115360852283821334858541043710301057312858051901453919067023103730011648890038847384890504L + K = 14730343317708716439807310032871972459448364195094179797249681733965528989482751523943515690110179031004049109375612685505881911274101441415545039654102474376472240501616988799699744135291070488314748284283496055223852115360852283821334858541043710301057312858051901453919067023103730011648890038847384890504 def setUp(self): - pass + self._original_urandom = os.urandom + os.urandom = dummy_urandom def tearDown(self): - pass + os.urandom = self._original_urandom def test_1_group1_client(self): transport = FakeTransport() transport.server_mode = False kex = KexGroup1(transport) kex.start_kex() - x = '1E000000807E2DDB1743F3487D6545F04F1C8476092FB912B013626AB5BCEB764257D88BBA64243B9F348DF7B41B8C814A995E00299913503456983FFB9178D3CD79EB6D55522418A8ABF65375872E55938AB99A84A0B5FC8A1ECC66A7C3766E7E0F80B7CE2C9225FC2DD683F4764244B72963BBB383F529DCF0C5D17740B8A2ADBE9208D4' - self.assertEquals(x, hexlify(str(transport._message)).upper()) - self.assertEquals((paramiko.kex_group1._MSG_KEXDH_REPLY,), transport._expect) + x = b'1E000000807E2DDB1743F3487D6545F04F1C8476092FB912B013626AB5BCEB764257D88BBA64243B9F348DF7B41B8C814A995E00299913503456983FFB9178D3CD79EB6D55522418A8ABF65375872E55938AB99A84A0B5FC8A1ECC66A7C3766E7E0F80B7CE2C9225FC2DD683F4764244B72963BBB383F529DCF0C5D17740B8A2ADBE9208D4' + self.assertEqual(x, hexlify(transport._message.asbytes()).upper()) + self.assertEqual((paramiko.kex_group1._MSG_KEXDH_REPLY,), transport._expect) # fake "reply" msg = Message() @@ -99,47 +113,47 @@ class KexTest (unittest.TestCase): msg.add_string('fake-sig') msg.rewind() kex.parse_next(paramiko.kex_group1._MSG_KEXDH_REPLY, msg) - H = '03079780F3D3AD0B3C6DB30C8D21685F367A86D2' - self.assertEquals(self.K, transport._K) - self.assertEquals(H, hexlify(transport._H).upper()) - self.assertEquals(('fake-host-key', 'fake-sig'), transport._verify) - self.assert_(transport._activated) + H = b'03079780F3D3AD0B3C6DB30C8D21685F367A86D2' + self.assertEqual(self.K, transport._K) + self.assertEqual(H, hexlify(transport._H).upper()) + self.assertEqual((b'fake-host-key', b'fake-sig'), transport._verify) + self.assertTrue(transport._activated) def test_2_group1_server(self): transport = FakeTransport() transport.server_mode = True kex = KexGroup1(transport) kex.start_kex() - self.assertEquals((paramiko.kex_group1._MSG_KEXDH_INIT,), transport._expect) + self.assertEqual((paramiko.kex_group1._MSG_KEXDH_INIT,), transport._expect) msg = Message() msg.add_mpint(69) msg.rewind() kex.parse_next(paramiko.kex_group1._MSG_KEXDH_INIT, msg) - H = 'B16BF34DD10945EDE84E9C1EF24A14BFDC843389' - x = '1F0000000866616B652D6B6579000000807E2DDB1743F3487D6545F04F1C8476092FB912B013626AB5BCEB764257D88BBA64243B9F348DF7B41B8C814A995E00299913503456983FFB9178D3CD79EB6D55522418A8ABF65375872E55938AB99A84A0B5FC8A1ECC66A7C3766E7E0F80B7CE2C9225FC2DD683F4764244B72963BBB383F529DCF0C5D17740B8A2ADBE9208D40000000866616B652D736967' - self.assertEquals(self.K, transport._K) - self.assertEquals(H, hexlify(transport._H).upper()) - self.assertEquals(x, hexlify(str(transport._message)).upper()) - self.assert_(transport._activated) + H = b'B16BF34DD10945EDE84E9C1EF24A14BFDC843389' + x = b'1F0000000866616B652D6B6579000000807E2DDB1743F3487D6545F04F1C8476092FB912B013626AB5BCEB764257D88BBA64243B9F348DF7B41B8C814A995E00299913503456983FFB9178D3CD79EB6D55522418A8ABF65375872E55938AB99A84A0B5FC8A1ECC66A7C3766E7E0F80B7CE2C9225FC2DD683F4764244B72963BBB383F529DCF0C5D17740B8A2ADBE9208D40000000866616B652D736967' + self.assertEqual(self.K, transport._K) + self.assertEqual(H, hexlify(transport._H).upper()) + self.assertEqual(x, hexlify(transport._message.asbytes()).upper()) + self.assertTrue(transport._activated) def test_3_gex_client(self): transport = FakeTransport() transport.server_mode = False kex = KexGex(transport) kex.start_kex() - x = '22000004000000080000002000' - self.assertEquals(x, hexlify(str(transport._message)).upper()) - self.assertEquals((paramiko.kex_gex._MSG_KEXDH_GEX_GROUP,), transport._expect) + x = b'22000004000000080000002000' + self.assertEqual(x, hexlify(transport._message.asbytes()).upper()) + self.assertEqual((paramiko.kex_gex._MSG_KEXDH_GEX_GROUP,), transport._expect) msg = Message() msg.add_mpint(FakeModulusPack.P) msg.add_mpint(FakeModulusPack.G) msg.rewind() kex.parse_next(paramiko.kex_gex._MSG_KEXDH_GEX_GROUP, msg) - x = '20000000807E2DDB1743F3487D6545F04F1C8476092FB912B013626AB5BCEB764257D88BBA64243B9F348DF7B41B8C814A995E00299913503456983FFB9178D3CD79EB6D55522418A8ABF65375872E55938AB99A84A0B5FC8A1ECC66A7C3766E7E0F80B7CE2C9225FC2DD683F4764244B72963BBB383F529DCF0C5D17740B8A2ADBE9208D4' - self.assertEquals(x, hexlify(str(transport._message)).upper()) - self.assertEquals((paramiko.kex_gex._MSG_KEXDH_GEX_REPLY,), transport._expect) + x = b'20000000807E2DDB1743F3487D6545F04F1C8476092FB912B013626AB5BCEB764257D88BBA64243B9F348DF7B41B8C814A995E00299913503456983FFB9178D3CD79EB6D55522418A8ABF65375872E55938AB99A84A0B5FC8A1ECC66A7C3766E7E0F80B7CE2C9225FC2DD683F4764244B72963BBB383F529DCF0C5D17740B8A2ADBE9208D4' + self.assertEqual(x, hexlify(transport._message.asbytes()).upper()) + self.assertEqual((paramiko.kex_gex._MSG_KEXDH_GEX_REPLY,), transport._expect) msg = Message() msg.add_string('fake-host-key') @@ -147,29 +161,29 @@ class KexTest (unittest.TestCase): msg.add_string('fake-sig') msg.rewind() kex.parse_next(paramiko.kex_gex._MSG_KEXDH_GEX_REPLY, msg) - H = 'A265563F2FA87F1A89BF007EE90D58BE2E4A4BD0' - self.assertEquals(self.K, transport._K) - self.assertEquals(H, hexlify(transport._H).upper()) - self.assertEquals(('fake-host-key', 'fake-sig'), transport._verify) - self.assert_(transport._activated) + H = b'A265563F2FA87F1A89BF007EE90D58BE2E4A4BD0' + self.assertEqual(self.K, transport._K) + self.assertEqual(H, hexlify(transport._H).upper()) + self.assertEqual((b'fake-host-key', b'fake-sig'), transport._verify) + self.assertTrue(transport._activated) def test_4_gex_old_client(self): transport = FakeTransport() transport.server_mode = False kex = KexGex(transport) kex.start_kex(_test_old_style=True) - x = '1E00000800' - self.assertEquals(x, hexlify(str(transport._message)).upper()) - self.assertEquals((paramiko.kex_gex._MSG_KEXDH_GEX_GROUP,), transport._expect) + x = b'1E00000800' + self.assertEqual(x, hexlify(transport._message.asbytes()).upper()) + self.assertEqual((paramiko.kex_gex._MSG_KEXDH_GEX_GROUP,), transport._expect) msg = Message() msg.add_mpint(FakeModulusPack.P) msg.add_mpint(FakeModulusPack.G) msg.rewind() kex.parse_next(paramiko.kex_gex._MSG_KEXDH_GEX_GROUP, msg) - x = '20000000807E2DDB1743F3487D6545F04F1C8476092FB912B013626AB5BCEB764257D88BBA64243B9F348DF7B41B8C814A995E00299913503456983FFB9178D3CD79EB6D55522418A8ABF65375872E55938AB99A84A0B5FC8A1ECC66A7C3766E7E0F80B7CE2C9225FC2DD683F4764244B72963BBB383F529DCF0C5D17740B8A2ADBE9208D4' - self.assertEquals(x, hexlify(str(transport._message)).upper()) - self.assertEquals((paramiko.kex_gex._MSG_KEXDH_GEX_REPLY,), transport._expect) + x = b'20000000807E2DDB1743F3487D6545F04F1C8476092FB912B013626AB5BCEB764257D88BBA64243B9F348DF7B41B8C814A995E00299913503456983FFB9178D3CD79EB6D55522418A8ABF65375872E55938AB99A84A0B5FC8A1ECC66A7C3766E7E0F80B7CE2C9225FC2DD683F4764244B72963BBB383F529DCF0C5D17740B8A2ADBE9208D4' + self.assertEqual(x, hexlify(transport._message.asbytes()).upper()) + self.assertEqual((paramiko.kex_gex._MSG_KEXDH_GEX_REPLY,), transport._expect) msg = Message() msg.add_string('fake-host-key') @@ -177,18 +191,18 @@ class KexTest (unittest.TestCase): msg.add_string('fake-sig') msg.rewind() kex.parse_next(paramiko.kex_gex._MSG_KEXDH_GEX_REPLY, msg) - H = '807F87B269EF7AC5EC7E75676808776A27D5864C' - self.assertEquals(self.K, transport._K) - self.assertEquals(H, hexlify(transport._H).upper()) - self.assertEquals(('fake-host-key', 'fake-sig'), transport._verify) - self.assert_(transport._activated) + H = b'807F87B269EF7AC5EC7E75676808776A27D5864C' + self.assertEqual(self.K, transport._K) + self.assertEqual(H, hexlify(transport._H).upper()) + self.assertEqual((b'fake-host-key', b'fake-sig'), transport._verify) + self.assertTrue(transport._activated) def test_5_gex_server(self): transport = FakeTransport() transport.server_mode = True kex = KexGex(transport) kex.start_kex() - self.assertEquals((paramiko.kex_gex._MSG_KEXDH_GEX_REQUEST, paramiko.kex_gex._MSG_KEXDH_GEX_REQUEST_OLD), transport._expect) + self.assertEqual((paramiko.kex_gex._MSG_KEXDH_GEX_REQUEST, paramiko.kex_gex._MSG_KEXDH_GEX_REQUEST_OLD), transport._expect) msg = Message() msg.add_int(1024) @@ -196,45 +210,45 @@ class KexTest (unittest.TestCase): msg.add_int(4096) msg.rewind() kex.parse_next(paramiko.kex_gex._MSG_KEXDH_GEX_REQUEST, msg) - x = '1F0000008100FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7EDEE386BFB5A899FA5AE9F24117C4B1FE649286651ECE65381FFFFFFFFFFFFFFFF0000000102' - self.assertEquals(x, hexlify(str(transport._message)).upper()) - self.assertEquals((paramiko.kex_gex._MSG_KEXDH_GEX_INIT,), transport._expect) + x = b'1F0000008100FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7EDEE386BFB5A899FA5AE9F24117C4B1FE649286651ECE65381FFFFFFFFFFFFFFFF0000000102' + self.assertEqual(x, hexlify(transport._message.asbytes()).upper()) + self.assertEqual((paramiko.kex_gex._MSG_KEXDH_GEX_INIT,), transport._expect) msg = Message() msg.add_mpint(12345) msg.rewind() kex.parse_next(paramiko.kex_gex._MSG_KEXDH_GEX_INIT, msg) - K = 67592995013596137876033460028393339951879041140378510871612128162185209509220726296697886624612526735888348020498716482757677848959420073720160491114319163078862905400020959196386947926388406687288901564192071077389283980347784184487280885335302632305026248574716290537036069329724382811853044654824945750581L - H = 'CE754197C21BF3452863B4F44D0B3951F12516EF' - x = '210000000866616B652D6B6579000000807E2DDB1743F3487D6545F04F1C8476092FB912B013626AB5BCEB764257D88BBA64243B9F348DF7B41B8C814A995E00299913503456983FFB9178D3CD79EB6D55522418A8ABF65375872E55938AB99A84A0B5FC8A1ECC66A7C3766E7E0F80B7CE2C9225FC2DD683F4764244B72963BBB383F529DCF0C5D17740B8A2ADBE9208D40000000866616B652D736967' - self.assertEquals(K, transport._K) - self.assertEquals(H, hexlify(transport._H).upper()) - self.assertEquals(x, hexlify(str(transport._message)).upper()) - self.assert_(transport._activated) + K = 67592995013596137876033460028393339951879041140378510871612128162185209509220726296697886624612526735888348020498716482757677848959420073720160491114319163078862905400020959196386947926388406687288901564192071077389283980347784184487280885335302632305026248574716290537036069329724382811853044654824945750581 + H = b'CE754197C21BF3452863B4F44D0B3951F12516EF' + x = b'210000000866616B652D6B6579000000807E2DDB1743F3487D6545F04F1C8476092FB912B013626AB5BCEB764257D88BBA64243B9F348DF7B41B8C814A995E00299913503456983FFB9178D3CD79EB6D55522418A8ABF65375872E55938AB99A84A0B5FC8A1ECC66A7C3766E7E0F80B7CE2C9225FC2DD683F4764244B72963BBB383F529DCF0C5D17740B8A2ADBE9208D40000000866616B652D736967' + self.assertEqual(K, transport._K) + self.assertEqual(H, hexlify(transport._H).upper()) + self.assertEqual(x, hexlify(transport._message.asbytes()).upper()) + self.assertTrue(transport._activated) def test_6_gex_server_with_old_client(self): transport = FakeTransport() transport.server_mode = True kex = KexGex(transport) kex.start_kex() - self.assertEquals((paramiko.kex_gex._MSG_KEXDH_GEX_REQUEST, paramiko.kex_gex._MSG_KEXDH_GEX_REQUEST_OLD), transport._expect) + self.assertEqual((paramiko.kex_gex._MSG_KEXDH_GEX_REQUEST, paramiko.kex_gex._MSG_KEXDH_GEX_REQUEST_OLD), transport._expect) msg = Message() msg.add_int(2048) msg.rewind() kex.parse_next(paramiko.kex_gex._MSG_KEXDH_GEX_REQUEST_OLD, msg) - x = '1F0000008100FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7EDEE386BFB5A899FA5AE9F24117C4B1FE649286651ECE65381FFFFFFFFFFFFFFFF0000000102' - self.assertEquals(x, hexlify(str(transport._message)).upper()) - self.assertEquals((paramiko.kex_gex._MSG_KEXDH_GEX_INIT,), transport._expect) + x = b'1F0000008100FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7EDEE386BFB5A899FA5AE9F24117C4B1FE649286651ECE65381FFFFFFFFFFFFFFFF0000000102' + self.assertEqual(x, hexlify(transport._message.asbytes()).upper()) + self.assertEqual((paramiko.kex_gex._MSG_KEXDH_GEX_INIT,), transport._expect) msg = Message() msg.add_mpint(12345) msg.rewind() kex.parse_next(paramiko.kex_gex._MSG_KEXDH_GEX_INIT, msg) - K = 67592995013596137876033460028393339951879041140378510871612128162185209509220726296697886624612526735888348020498716482757677848959420073720160491114319163078862905400020959196386947926388406687288901564192071077389283980347784184487280885335302632305026248574716290537036069329724382811853044654824945750581L - H = 'B41A06B2E59043CEFC1AE16EC31F1E2D12EC455B' - x = '210000000866616B652D6B6579000000807E2DDB1743F3487D6545F04F1C8476092FB912B013626AB5BCEB764257D88BBA64243B9F348DF7B41B8C814A995E00299913503456983FFB9178D3CD79EB6D55522418A8ABF65375872E55938AB99A84A0B5FC8A1ECC66A7C3766E7E0F80B7CE2C9225FC2DD683F4764244B72963BBB383F529DCF0C5D17740B8A2ADBE9208D40000000866616B652D736967' - self.assertEquals(K, transport._K) - self.assertEquals(H, hexlify(transport._H).upper()) - self.assertEquals(x, hexlify(str(transport._message)).upper()) - self.assert_(transport._activated) + K = 67592995013596137876033460028393339951879041140378510871612128162185209509220726296697886624612526735888348020498716482757677848959420073720160491114319163078862905400020959196386947926388406687288901564192071077389283980347784184487280885335302632305026248574716290537036069329724382811853044654824945750581 + H = b'B41A06B2E59043CEFC1AE16EC31F1E2D12EC455B' + x = b'210000000866616B652D6B6579000000807E2DDB1743F3487D6545F04F1C8476092FB912B013626AB5BCEB764257D88BBA64243B9F348DF7B41B8C814A995E00299913503456983FFB9178D3CD79EB6D55522418A8ABF65375872E55938AB99A84A0B5FC8A1ECC66A7C3766E7E0F80B7CE2C9225FC2DD683F4764244B72963BBB383F529DCF0C5D17740B8A2ADBE9208D40000000866616B652D736967' + self.assertEqual(K, transport._K) + self.assertEqual(H, hexlify(transport._H).upper()) + self.assertEqual(x, hexlify(transport._message.asbytes()).upper()) + self.assertTrue(transport._activated) diff --git a/tests/test_message.py b/tests/test_message.py index 7bfd44df..f308c037 100644 --- a/tests/test_message.py +++ b/tests/test_message.py @@ -7,7 +7,7 @@ # Software Foundation; either version 2.1 of the License, or (at your option) # any later version. # -# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY +# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR # A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # details. @@ -22,14 +22,15 @@ Some unit tests for ssh protocol message blocks. import unittest from paramiko.message import Message +from paramiko.common import byte_chr, zero_byte class MessageTest (unittest.TestCase): - __a = '\x00\x00\x00\x17\x07\x60\xe0\x90\x00\x00\x00\x01q\x00\x00\x00\x05hello\x00\x00\x03\xe8' + ('x' * 1000) - __b = '\x01\x00\xf3\x00\x3f\x00\x00\x00\x10huey,dewey,louie' - __c = '\x00\x00\x00\x00\x00\x00\x00\x05\x00\x00\xf5\xe4\xd3\xc2\xb1\x09\x00\x00\x00\x01\x11\x00\x00\x00\x07\x00\xf5\xe4\xd3\xc2\xb1\x09\x00\x00\x00\x06\x9a\x1b\x2c\x3d\x4e\xf7' - __d = '\x00\x00\x00\x05\x00\x00\x00\x05\x11\x22\x33\x44\x55\x01\x00\x00\x00\x03cat\x00\x00\x00\x03a,b' + __a = b'\x00\x00\x00\x17\x07\x60\xe0\x90\x00\x00\x00\x01\x71\x00\x00\x00\x05\x68\x65\x6c\x6c\x6f\x00\x00\x03\xe8' + b'x' * 1000 + __b = b'\x01\x00\xf3\x00\x3f\x00\x00\x00\x10\x68\x75\x65\x79\x2c\x64\x65\x77\x65\x79\x2c\x6c\x6f\x75\x69\x65' + __c = b'\x00\x00\x00\x00\x00\x00\x00\x05\x00\x00\xf5\xe4\xd3\xc2\xb1\x09\x00\x00\x00\x01\x11\x00\x00\x00\x07\x00\xf5\xe4\xd3\xc2\xb1\x09\x00\x00\x00\x06\x9a\x1b\x2c\x3d\x4e\xf7' + __d = b'\x00\x00\x00\x05\xff\x00\x00\x00\x05\x11\x22\x33\x44\x55\xff\x00\x00\x00\x0a\x00\xf0\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x03\x63\x61\x74\x00\x00\x00\x03\x61\x2c\x62' def test_1_encode(self): msg = Message() @@ -38,63 +39,65 @@ class MessageTest (unittest.TestCase): msg.add_string('q') msg.add_string('hello') msg.add_string('x' * 1000) - self.assertEquals(str(msg), self.__a) + self.assertEqual(msg.asbytes(), self.__a) msg = Message() msg.add_boolean(True) msg.add_boolean(False) - msg.add_byte('\xf3') - msg.add_bytes('\x00\x3f') + msg.add_byte(byte_chr(0xf3)) + + msg.add_bytes(zero_byte + byte_chr(0x3f)) msg.add_list(['huey', 'dewey', 'louie']) - self.assertEquals(str(msg), self.__b) + self.assertEqual(msg.asbytes(), self.__b) msg = Message() msg.add_int64(5) - msg.add_int64(0xf5e4d3c2b109L) + msg.add_int64(0xf5e4d3c2b109) msg.add_mpint(17) - msg.add_mpint(0xf5e4d3c2b109L) - msg.add_mpint(-0x65e4d3c2b109L) - self.assertEquals(str(msg), self.__c) + msg.add_mpint(0xf5e4d3c2b109) + msg.add_mpint(-0x65e4d3c2b109) + self.assertEqual(msg.asbytes(), self.__c) def test_2_decode(self): msg = Message(self.__a) - self.assertEquals(msg.get_int(), 23) - self.assertEquals(msg.get_int(), 123789456) - self.assertEquals(msg.get_string(), 'q') - self.assertEquals(msg.get_string(), 'hello') - self.assertEquals(msg.get_string(), 'x' * 1000) + self.assertEqual(msg.get_int(), 23) + self.assertEqual(msg.get_int(), 123789456) + self.assertEqual(msg.get_text(), 'q') + self.assertEqual(msg.get_text(), 'hello') + self.assertEqual(msg.get_text(), 'x' * 1000) msg = Message(self.__b) - self.assertEquals(msg.get_boolean(), True) - self.assertEquals(msg.get_boolean(), False) - self.assertEquals(msg.get_byte(), '\xf3') - self.assertEquals(msg.get_bytes(2), '\x00\x3f') - self.assertEquals(msg.get_list(), ['huey', 'dewey', 'louie']) + self.assertEqual(msg.get_boolean(), True) + self.assertEqual(msg.get_boolean(), False) + self.assertEqual(msg.get_byte(), byte_chr(0xf3)) + self.assertEqual(msg.get_bytes(2), zero_byte + byte_chr(0x3f)) + self.assertEqual(msg.get_list(), ['huey', 'dewey', 'louie']) msg = Message(self.__c) - self.assertEquals(msg.get_int64(), 5) - self.assertEquals(msg.get_int64(), 0xf5e4d3c2b109L) - self.assertEquals(msg.get_mpint(), 17) - self.assertEquals(msg.get_mpint(), 0xf5e4d3c2b109L) - self.assertEquals(msg.get_mpint(), -0x65e4d3c2b109L) + self.assertEqual(msg.get_int64(), 5) + self.assertEqual(msg.get_int64(), 0xf5e4d3c2b109) + self.assertEqual(msg.get_mpint(), 17) + self.assertEqual(msg.get_mpint(), 0xf5e4d3c2b109) + self.assertEqual(msg.get_mpint(), -0x65e4d3c2b109) def test_3_add(self): msg = Message() msg.add(5) - msg.add(0x1122334455L) + msg.add(0x1122334455) + msg.add(0xf00000000000000000) msg.add(True) msg.add('cat') msg.add(['a', 'b']) - self.assertEquals(str(msg), self.__d) + self.assertEqual(msg.asbytes(), self.__d) def test_4_misc(self): msg = Message(self.__d) - self.assertEquals(msg.get_int(), 5) - self.assertEquals(msg.get_mpint(), 0x1122334455L) - self.assertEquals(msg.get_so_far(), self.__d[:13]) - self.assertEquals(msg.get_remainder(), self.__d[13:]) + self.assertEqual(msg.get_int(), 5) + self.assertEqual(msg.get_int(), 0x1122334455) + self.assertEqual(msg.get_int(), 0xf00000000000000000) + self.assertEqual(msg.get_so_far(), self.__d[:29]) + self.assertEqual(msg.get_remainder(), self.__d[29:]) msg.rewind() - self.assertEquals(msg.get_int(), 5) - self.assertEquals(msg.get_so_far(), self.__d[:4]) - self.assertEquals(msg.get_remainder(), self.__d[4:]) - + self.assertEqual(msg.get_int(), 5) + self.assertEqual(msg.get_so_far(), self.__d[:4]) + self.assertEqual(msg.get_remainder(), self.__d[4:]) diff --git a/tests/test_packetizer.py b/tests/test_packetizer.py index d1eb584a..a8c0f973 100644 --- a/tests/test_packetizer.py +++ b/tests/test_packetizer.py @@ -7,7 +7,7 @@ # Software Foundation; either version 2.1 of the License, or (at your option) # any later version. # -# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY +# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR # A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # details. @@ -21,50 +21,56 @@ Some unit tests for the ssh2 protocol in Transport. """ import unittest -from loop import LoopSocket +from hashlib import sha1 + +from tests.loop import LoopSocket + from Crypto.Cipher import AES -from Crypto.Hash import SHA, HMAC + from paramiko import Message, Packetizer, util +from paramiko.common import byte_chr, zero_byte + +x55 = byte_chr(0x55) +x1f = byte_chr(0x1f) + class PacketizerTest (unittest.TestCase): - def test_1_write (self): + def test_1_write(self): rsock = LoopSocket() wsock = LoopSocket() rsock.link(wsock) p = Packetizer(wsock) p.set_log(util.get_logger('paramiko.transport')) p.set_hexdump(True) - cipher = AES.new('\x00' * 16, AES.MODE_CBC, '\x55' * 16) - p.set_outbound_cipher(cipher, 16, SHA, 12, '\x1f' * 20) + cipher = AES.new(zero_byte * 16, AES.MODE_CBC, x55 * 16) + p.set_outbound_cipher(cipher, 16, sha1, 12, x1f * 20) # message has to be at least 16 bytes long, so we'll have at least one # block of data encrypted that contains zero random padding bytes m = Message() - m.add_byte(chr(100)) + m.add_byte(byte_chr(100)) m.add_int(100) m.add_int(1) m.add_int(900) p.send_message(m) data = rsock.recv(100) # 32 + 12 bytes of MAC = 44 - self.assertEquals(44, len(data)) - self.assertEquals('\x43\x91\x97\xbd\x5b\x50\xac\x25\x87\xc2\xc4\x6b\xc7\xe9\x38\xc0', data[:16]) - - def test_2_read (self): + self.assertEqual(44, len(data)) + self.assertEqual(b'\x43\x91\x97\xbd\x5b\x50\xac\x25\x87\xc2\xc4\x6b\xc7\xe9\x38\xc0', data[:16]) + + def test_2_read(self): rsock = LoopSocket() wsock = LoopSocket() rsock.link(wsock) p = Packetizer(rsock) p.set_log(util.get_logger('paramiko.transport')) p.set_hexdump(True) - cipher = AES.new('\x00' * 16, AES.MODE_CBC, '\x55' * 16) - p.set_inbound_cipher(cipher, 16, SHA, 12, '\x1f' * 20) - - wsock.send('C\x91\x97\xbd[P\xac%\x87\xc2\xc4k\xc7\xe98\xc0' + \ - '\x90\xd2\x16V\rqsa8|L=\xfb\x97}\xe2n\x03\xb1\xa0\xc2\x1c\xd6AAL\xb4Y') + cipher = AES.new(zero_byte * 16, AES.MODE_CBC, x55 * 16) + p.set_inbound_cipher(cipher, 16, sha1, 12, x1f * 20) + wsock.send(b'\x43\x91\x97\xbd\x5b\x50\xac\x25\x87\xc2\xc4\x6b\xc7\xe9\x38\xc0\x90\xd2\x16\x56\x0d\x71\x73\x61\x38\x7c\x4c\x3d\xfb\x97\x7d\xe2\x6e\x03\xb1\xa0\xc2\x1c\xd6\x41\x41\x4c\xb4\x59') cmd, m = p.read_message() - self.assertEquals(100, cmd) - self.assertEquals(100, m.get_int()) - self.assertEquals(1, m.get_int()) - self.assertEquals(900, m.get_int()) + self.assertEqual(100, cmd) + self.assertEqual(100, m.get_int()) + self.assertEqual(1, m.get_int()) + self.assertEqual(900, m.get_int()) diff --git a/tests/test_pkey.py b/tests/test_pkey.py index 89d55803..1468ee27 100644 --- a/tests/test_pkey.py +++ b/tests/test_pkey.py @@ -7,7 +7,7 @@ # Software Foundation; either version 2.1 of the License, or (at your option) # any later version. # -# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY +# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR # A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # details. @@ -20,17 +20,23 @@ Some unit tests for public/private key objects. """ -from binascii import hexlify, unhexlify -import StringIO import unittest -from paramiko import RSAKey, DSSKey, Message, util -from paramiko.common import rng +from binascii import hexlify +from hashlib import md5 + +from paramiko import RSAKey, DSSKey, ECDSAKey, Message, util +from paramiko.py3compat import StringIO, byte_chr, b, bytes + +from tests.util import test_path # from openssh's ssh-keygen PUB_RSA = 'ssh-rsa AAAAB3NzaC1yc2EAAAABIwAAAIEA049W6geFpmsljTwfvI1UmKWWJPNFI74+vNKTk4dmzkQY2yAMs6FhlvhlI8ysU4oj71ZsRYMecHbBbxdN79+JRFVYTKaLqjwGENeTd+yv4q+V2PvZv3fLnzApI3l7EJCqhWwJUHJ1jAkZzqDx0tyOL4uoZpww3nmE0kb3y21tH4c=' PUB_DSS = 'ssh-dss AAAAB3NzaC1kc3MAAACBAOeBpgNnfRzr/twmAQRu2XwWAp3CFtrVnug6s6fgwj/oLjYbVtjAy6pl/h0EKCWx2rf1IetyNsTxWrniA9I6HeDj65X1FyDkg6g8tvCnaNB8Xp/UUhuzHuGsMIipRxBxw9LF608EqZcj1E3ytktoW5B5OcjrkEoz3xG7C+rpIjYvAAAAFQDwz4UnmsGiSNu5iqjn3uTzwUpshwAAAIEAkxfFeY8P2wZpDjX0MimZl5wkoFQDL25cPzGBuB4OnB8NoUk/yjAHIIpEShw8V+LzouMK5CTJQo5+Ngw3qIch/WgRmMHy4kBq1SsXMjQCte1So6HBMvBPIW5SiMTmjCfZZiw4AYHK+B/JaOwaG9yRg2Ejg4Ok10+XFDxlqZo8Y+wAAACARmR7CCPjodxASvRbIyzaVpZoJ/Z6x7dAumV+ysrV1BVYd0lYukmnjO1kKBWApqpH1ve9XDQYN8zgxM4b16L21kpoWQnZtXrY3GZ4/it9kUgyB7+NwacIBlXa8cMDL7Q/69o0d54U0X/NeX5QxuYR6OMJlrkQB7oiW/P/1mwjQgE=' +PUB_ECDSA = 'ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBJSPZm3ZWkvk/Zx8WP+fZRZ5/NBBHnGQwR6uIC6XHGPDIHuWUzIjAwA0bzqkOUffEsbLe+uQgKl5kbc/L8KA/eo=' + FINGER_RSA = '1024 60:73:38:44:cb:51:86:65:7f:de:da:a2:2b:5a:57:d5' FINGER_DSS = '1024 44:78:f0:b9:a2:3c:c5:18:20:09:ff:75:5b:c1:d2:6c' +FINGER_ECDSA = '256 25:19:eb:55:e6:a1:47:ff:4f:38:d2:75:6f:a5:d5:60' SIGNED_RSA = '20:d7:8a:31:21:cb:f7:92:12:f2:a4:89:37:f5:78:af:e6:16:b6:25:b9:97:3d:a2:cd:5f:ca:20:21:73:4c:ad:34:73:8f:20:77:28:e2:94:15:08:d8:91:40:7a:85:83:bf:18:37:95:dc:54:1a:9b:88:29:6c:73:ca:38:b4:04:f1:56:b9:f2:42:9d:52:1b:29:29:b4:4f:fd:c9:2d:af:47:d2:40:76:30:f3:63:45:0c:d9:1d:43:86:0f:1c:70:e2:93:12:34:f3:ac:c5:0a:2f:14:50:66:59:f1:88:ee:c1:4a:e9:d1:9c:4e:46:f0:0e:47:6f:38:74:f1:44:a8' RSA_PRIVATE_OUT = """\ @@ -66,6 +72,16 @@ QPSch9pT9XHqn+1rZ4bK+QGA -----END DSA PRIVATE KEY----- """ +ECDSA_PRIVATE_OUT = """\ +-----BEGIN EC PRIVATE KEY----- +MHcCAQEEIKB6ty3yVyKEnfF/zprx0qwC76MsMlHY4HXCnqho2eKioAoGCCqGSM49 +AwEHoUQDQgAElI9mbdlaS+T9nHxY/59lFnn80EEecZDBHq4gLpccY8Mge5ZTMiMD +ADRvOqQ5R98Sxst765CAqXmRtz8vwoD96g== +-----END EC PRIVATE KEY----- +""" + +x1234 = b'\x01\x02\x03\x04' + class KeyTest (unittest.TestCase): @@ -76,115 +92,164 @@ class KeyTest (unittest.TestCase): pass def test_1_generate_key_bytes(self): - from Crypto.Hash import MD5 - key = util.generate_key_bytes(MD5, '\x01\x02\x03\x04', 'happy birthday', 30) - exp = unhexlify('61E1F272F4C1C4561586BD322498C0E924672780F47BB37DDA7D54019E64') - self.assertEquals(exp, key) + key = util.generate_key_bytes(md5, x1234, 'happy birthday', 30) + exp = b'\x61\xE1\xF2\x72\xF4\xC1\xC4\x56\x15\x86\xBD\x32\x24\x98\xC0\xE9\x24\x67\x27\x80\xF4\x7B\xB3\x7D\xDA\x7D\x54\x01\x9E\x64' + self.assertEqual(exp, key) def test_2_load_rsa(self): - key = RSAKey.from_private_key_file('tests/test_rsa.key') - self.assertEquals('ssh-rsa', key.get_name()) - exp_rsa = FINGER_RSA.split()[1].replace(':', '') + key = RSAKey.from_private_key_file(test_path('test_rsa.key')) + self.assertEqual('ssh-rsa', key.get_name()) + exp_rsa = b(FINGER_RSA.split()[1].replace(':', '')) my_rsa = hexlify(key.get_fingerprint()) - self.assertEquals(exp_rsa, my_rsa) - self.assertEquals(PUB_RSA.split()[1], key.get_base64()) - self.assertEquals(1024, key.get_bits()) + self.assertEqual(exp_rsa, my_rsa) + self.assertEqual(PUB_RSA.split()[1], key.get_base64()) + self.assertEqual(1024, key.get_bits()) - s = StringIO.StringIO() + s = StringIO() key.write_private_key(s) - self.assertEquals(RSA_PRIVATE_OUT, s.getvalue()) + self.assertEqual(RSA_PRIVATE_OUT, s.getvalue()) s.seek(0) key2 = RSAKey.from_private_key(s) - self.assertEquals(key, key2) + self.assertEqual(key, key2) def test_3_load_rsa_password(self): - key = RSAKey.from_private_key_file('tests/test_rsa_password.key', 'television') - self.assertEquals('ssh-rsa', key.get_name()) - exp_rsa = FINGER_RSA.split()[1].replace(':', '') + key = RSAKey.from_private_key_file(test_path('test_rsa_password.key'), 'television') + self.assertEqual('ssh-rsa', key.get_name()) + exp_rsa = b(FINGER_RSA.split()[1].replace(':', '')) my_rsa = hexlify(key.get_fingerprint()) - self.assertEquals(exp_rsa, my_rsa) - self.assertEquals(PUB_RSA.split()[1], key.get_base64()) - self.assertEquals(1024, key.get_bits()) + self.assertEqual(exp_rsa, my_rsa) + self.assertEqual(PUB_RSA.split()[1], key.get_base64()) + self.assertEqual(1024, key.get_bits()) def test_4_load_dss(self): - key = DSSKey.from_private_key_file('tests/test_dss.key') - self.assertEquals('ssh-dss', key.get_name()) - exp_dss = FINGER_DSS.split()[1].replace(':', '') + key = DSSKey.from_private_key_file(test_path('test_dss.key')) + self.assertEqual('ssh-dss', key.get_name()) + exp_dss = b(FINGER_DSS.split()[1].replace(':', '')) my_dss = hexlify(key.get_fingerprint()) - self.assertEquals(exp_dss, my_dss) - self.assertEquals(PUB_DSS.split()[1], key.get_base64()) - self.assertEquals(1024, key.get_bits()) + self.assertEqual(exp_dss, my_dss) + self.assertEqual(PUB_DSS.split()[1], key.get_base64()) + self.assertEqual(1024, key.get_bits()) - s = StringIO.StringIO() + s = StringIO() key.write_private_key(s) - self.assertEquals(DSS_PRIVATE_OUT, s.getvalue()) + self.assertEqual(DSS_PRIVATE_OUT, s.getvalue()) s.seek(0) key2 = DSSKey.from_private_key(s) - self.assertEquals(key, key2) + self.assertEqual(key, key2) def test_5_load_dss_password(self): - key = DSSKey.from_private_key_file('tests/test_dss_password.key', 'television') - self.assertEquals('ssh-dss', key.get_name()) - exp_dss = FINGER_DSS.split()[1].replace(':', '') + key = DSSKey.from_private_key_file(test_path('test_dss_password.key'), 'television') + self.assertEqual('ssh-dss', key.get_name()) + exp_dss = b(FINGER_DSS.split()[1].replace(':', '')) my_dss = hexlify(key.get_fingerprint()) - self.assertEquals(exp_dss, my_dss) - self.assertEquals(PUB_DSS.split()[1], key.get_base64()) - self.assertEquals(1024, key.get_bits()) + self.assertEqual(exp_dss, my_dss) + self.assertEqual(PUB_DSS.split()[1], key.get_base64()) + self.assertEqual(1024, key.get_bits()) def test_6_compare_rsa(self): # verify that the private & public keys compare equal - key = RSAKey.from_private_key_file('tests/test_rsa.key') - self.assertEquals(key, key) - pub = RSAKey(data=str(key)) - self.assert_(key.can_sign()) - self.assert_(not pub.can_sign()) - self.assertEquals(key, pub) + key = RSAKey.from_private_key_file(test_path('test_rsa.key')) + self.assertEqual(key, key) + pub = RSAKey(data=key.asbytes()) + self.assertTrue(key.can_sign()) + self.assertTrue(not pub.can_sign()) + self.assertEqual(key, pub) def test_7_compare_dss(self): # verify that the private & public keys compare equal - key = DSSKey.from_private_key_file('tests/test_dss.key') - self.assertEquals(key, key) - pub = DSSKey(data=str(key)) - self.assert_(key.can_sign()) - self.assert_(not pub.can_sign()) - self.assertEquals(key, pub) + key = DSSKey.from_private_key_file(test_path('test_dss.key')) + self.assertEqual(key, key) + pub = DSSKey(data=key.asbytes()) + self.assertTrue(key.can_sign()) + self.assertTrue(not pub.can_sign()) + self.assertEqual(key, pub) def test_8_sign_rsa(self): # verify that the rsa private key can sign and verify - key = RSAKey.from_private_key_file('tests/test_rsa.key') - msg = key.sign_ssh_data(rng, 'ice weasels') - self.assert_(type(msg) is Message) + key = RSAKey.from_private_key_file(test_path('test_rsa.key')) + msg = key.sign_ssh_data(b'ice weasels') + self.assertTrue(type(msg) is Message) msg.rewind() - self.assertEquals('ssh-rsa', msg.get_string()) - sig = ''.join([chr(int(x, 16)) for x in SIGNED_RSA.split(':')]) - self.assertEquals(sig, msg.get_string()) + self.assertEqual('ssh-rsa', msg.get_text()) + sig = bytes().join([byte_chr(int(x, 16)) for x in SIGNED_RSA.split(':')]) + self.assertEqual(sig, msg.get_binary()) msg.rewind() - pub = RSAKey(data=str(key)) - self.assert_(pub.verify_ssh_sig('ice weasels', msg)) + pub = RSAKey(data=key.asbytes()) + self.assertTrue(pub.verify_ssh_sig(b'ice weasels', msg)) def test_9_sign_dss(self): # verify that the dss private key can sign and verify - key = DSSKey.from_private_key_file('tests/test_dss.key') - msg = key.sign_ssh_data(rng, 'ice weasels') - self.assert_(type(msg) is Message) + key = DSSKey.from_private_key_file(test_path('test_dss.key')) + msg = key.sign_ssh_data(b'ice weasels') + self.assertTrue(type(msg) is Message) msg.rewind() - self.assertEquals('ssh-dss', msg.get_string()) + self.assertEqual('ssh-dss', msg.get_text()) # can't do the same test as we do for RSA, because DSS signatures # are usually different each time. but we can test verification # anyway so it's ok. - self.assertEquals(40, len(msg.get_string())) + self.assertEqual(40, len(msg.get_binary())) msg.rewind() - pub = DSSKey(data=str(key)) - self.assert_(pub.verify_ssh_sig('ice weasels', msg)) - + pub = DSSKey(data=key.asbytes()) + self.assertTrue(pub.verify_ssh_sig(b'ice weasels', msg)) + def test_A_generate_rsa(self): key = RSAKey.generate(1024) - msg = key.sign_ssh_data(rng, 'jerri blank') + msg = key.sign_ssh_data(b'jerri blank') msg.rewind() - self.assert_(key.verify_ssh_sig('jerri blank', msg)) + self.assertTrue(key.verify_ssh_sig(b'jerri blank', msg)) def test_B_generate_dss(self): key = DSSKey.generate(1024) - msg = key.sign_ssh_data(rng, 'jerri blank') + msg = key.sign_ssh_data(b'jerri blank') + msg.rewind() + self.assertTrue(key.verify_ssh_sig(b'jerri blank', msg)) + + def test_10_load_ecdsa(self): + key = ECDSAKey.from_private_key_file(test_path('test_ecdsa.key')) + self.assertEqual('ecdsa-sha2-nistp256', key.get_name()) + exp_ecdsa = b(FINGER_ECDSA.split()[1].replace(':', '')) + my_ecdsa = hexlify(key.get_fingerprint()) + self.assertEqual(exp_ecdsa, my_ecdsa) + self.assertEqual(PUB_ECDSA.split()[1], key.get_base64()) + self.assertEqual(256, key.get_bits()) + + s = StringIO() + key.write_private_key(s) + self.assertEqual(ECDSA_PRIVATE_OUT, s.getvalue()) + s.seek(0) + key2 = ECDSAKey.from_private_key(s) + self.assertEqual(key, key2) + + def test_11_load_ecdsa_password(self): + key = ECDSAKey.from_private_key_file(test_path('test_ecdsa_password.key'), b'television') + self.assertEqual('ecdsa-sha2-nistp256', key.get_name()) + exp_ecdsa = b(FINGER_ECDSA.split()[1].replace(':', '')) + my_ecdsa = hexlify(key.get_fingerprint()) + self.assertEqual(exp_ecdsa, my_ecdsa) + self.assertEqual(PUB_ECDSA.split()[1], key.get_base64()) + self.assertEqual(256, key.get_bits()) + + def test_12_compare_ecdsa(self): + # verify that the private & public keys compare equal + key = ECDSAKey.from_private_key_file(test_path('test_ecdsa.key')) + self.assertEqual(key, key) + pub = ECDSAKey(data=key.asbytes()) + self.assertTrue(key.can_sign()) + self.assertTrue(not pub.can_sign()) + self.assertEqual(key, pub) + + def test_13_sign_ecdsa(self): + # verify that the rsa private key can sign and verify + key = ECDSAKey.from_private_key_file(test_path('test_ecdsa.key')) + msg = key.sign_ssh_data(b'ice weasels') + self.assertTrue(type(msg) is Message) + msg.rewind() + self.assertEqual('ecdsa-sha2-nistp256', msg.get_text()) + # ECDSA signatures, like DSS signatures, tend to be different + # each time, so we can't compare against a "known correct" + # signature. + # Even the length of the signature can change. + msg.rewind() - self.assert_(key.verify_ssh_sig('jerri blank', msg)) + pub = ECDSAKey(data=key.asbytes()) + self.assertTrue(pub.verify_ssh_sig(b'ice weasels', msg)) diff --git a/tests/test_sftp.py b/tests/test_sftp.py index 2eadabcd..2b6aa3b6 100755 --- a/tests/test_sftp.py +++ b/tests/test_sftp.py @@ -7,7 +7,7 @@ # Software Foundation; either version 2.1 of the License, or (at your option) # any later version. # -# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY +# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR # A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # details. @@ -24,18 +24,20 @@ do test file operations in (so no existing files will be harmed). """ from binascii import hexlify -import logging import os -import random -import struct import sys +import warnings import threading -import time import unittest +from tempfile import mkstemp import paramiko -from stub_sftp import StubServer, StubSFTPServer -from loop import LoopSocket +from paramiko.py3compat import PY2, b, u, StringIO +from paramiko.common import o777, o600, o666, o644 +from tests.stub_sftp import StubServer, StubSFTPServer +from tests.loop import LoopSocket +from tests.util import test_path +import paramiko.util from paramiko.sftp_attr import SFTPAttributes ARTICLE = ''' @@ -65,11 +67,27 @@ liver insulin receptors. Their sensitivity to insulin is, however, similarly decreased compared with chicken. ''' + +# Here is how unicode characters are encoded over 1 to 6 bytes in utf-8 +# U-00000000 - U-0000007F: 0xxxxxxx +# U-00000080 - U-000007FF: 110xxxxx 10xxxxxx +# U-00000800 - U-0000FFFF: 1110xxxx 10xxxxxx 10xxxxxx +# U-00010000 - U-001FFFFF: 11110xxx 10xxxxxx 10xxxxxx 10xxxxxx +# U-00200000 - U-03FFFFFF: 111110xx 10xxxxxx 10xxxxxx 10xxxxxx 10xxxxxx +# U-04000000 - U-7FFFFFFF: 1111110x 10xxxxxx 10xxxxxx 10xxxxxx 10xxxxxx 10xxxxxx +# Note that: hex(int('11000011',2)) == '0xc3' +# Thus, the following 2-bytes sequence is not valid utf8: "invalid continuation byte" +NON_UTF8_DATA = b'\xC3\xC3' + FOLDER = os.environ.get('TEST_FOLDER', 'temp-testing000') sftp = None tc = None g_big_file_test = True +# we need to use eval(compile()) here because Py3.2 doesn't support the 'u' marker for unicode +# this test is the only line in the entire program that has to be treated specially to support Py3.2 +unicode_folder = eval(compile(r"u'\u00fcnic\u00f8de'" if PY2 else r"'\u00fcnic\u00f8de'", 'test_sftp.py', 'eval')) +utf8_folder = b'/\xc3\xbcnic\xc3\xb8\x64\x65' def get_sftp(): @@ -121,7 +139,7 @@ class SFTPTest (unittest.TestCase): tc = paramiko.Transport(sockc) ts = paramiko.Transport(socks) - host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key') + host_key = paramiko.RSAKey.from_private_key_file(test_path('test_rsa.key')) ts.add_server_key(host_key) event = threading.Event() server = StubServer() @@ -140,7 +158,7 @@ class SFTPTest (unittest.TestCase): def setUp(self): global FOLDER - for i in xrange(1000): + for i in range(1000): FOLDER = FOLDER[:-3] + '%03d' % i try: sftp.mkdir(FOLDER) @@ -149,6 +167,7 @@ class SFTPTest (unittest.TestCase): pass def tearDown(self): + #sftp.chdir() sftp.rmdir(FOLDER) def test_1_file(self): @@ -158,8 +177,8 @@ class SFTPTest (unittest.TestCase): f = sftp.open(FOLDER + '/test', 'w') try: self.assertEqual(f.stat().st_size, 0) - f.close() finally: + f.close() sftp.remove(FOLDER + '/test') def test_2_close(self): @@ -180,10 +199,20 @@ class SFTPTest (unittest.TestCase): """ verify that a file can be created and written, and the size is correct. """ - f = sftp.open(FOLDER + '/duck.txt', 'w') try: - f.write(ARTICLE) - f.close() + with sftp.open(FOLDER + '/duck.txt', 'w') as f: + f.write(ARTICLE) + self.assertEqual(sftp.stat(FOLDER + '/duck.txt').st_size, 1483) + finally: + sftp.remove(FOLDER + '/duck.txt') + + def test_3_sftp_file_can_be_used_as_context_manager(self): + """ + verify that an opened file can be used as a context manager + """ + try: + with sftp.open(FOLDER + '/duck.txt', 'w') as f: + f.write(ARTICLE) self.assertEqual(sftp.stat(FOLDER + '/duck.txt').st_size, 1483) finally: sftp.remove(FOLDER + '/duck.txt') @@ -192,19 +221,17 @@ class SFTPTest (unittest.TestCase): """ verify that a file can be opened for append, and tell() still works. """ - f = sftp.open(FOLDER + '/append.txt', 'w') try: - f.write('first line\nsecond line\n') - self.assertEqual(f.tell(), 23) - f.close() - - f = sftp.open(FOLDER + '/append.txt', 'a+') - f.write('third line!!!\n') - self.assertEqual(f.tell(), 37) - self.assertEqual(f.stat().st_size, 37) - f.seek(-26, f.SEEK_CUR) - self.assertEqual(f.readline(), 'second line\n') - f.close() + with sftp.open(FOLDER + '/append.txt', 'w') as f: + f.write('first line\nsecond line\n') + self.assertEqual(f.tell(), 23) + + with sftp.open(FOLDER + '/append.txt', 'a+') as f: + f.write('third line!!!\n') + self.assertEqual(f.tell(), 37) + self.assertEqual(f.stat().st_size, 37) + f.seek(-26, f.SEEK_CUR) + self.assertEqual(f.readline(), 'second line\n') finally: sftp.remove(FOLDER + '/append.txt') @@ -212,20 +239,18 @@ class SFTPTest (unittest.TestCase): """ verify that renaming a file works. """ - f = sftp.open(FOLDER + '/first.txt', 'w') try: - f.write('content!\n'); - f.close() + with sftp.open(FOLDER + '/first.txt', 'w') as f: + f.write('content!\n') sftp.rename(FOLDER + '/first.txt', FOLDER + '/second.txt') try: - f = sftp.open(FOLDER + '/first.txt', 'r') - self.assert_(False, 'no exception on reading nonexistent file') + sftp.open(FOLDER + '/first.txt', 'r') + self.assertTrue(False, 'no exception on reading nonexistent file') except IOError: pass - f = sftp.open(FOLDER + '/second.txt', 'r') - f.seek(-6, f.SEEK_END) - self.assertEqual(f.read(4), 'tent') - f.close() + with sftp.open(FOLDER + '/second.txt', 'r') as f: + f.seek(-6, f.SEEK_END) + self.assertEqual(u(f.read(4)), 'tent') finally: try: sftp.remove(FOLDER + '/first.txt') @@ -242,14 +267,13 @@ class SFTPTest (unittest.TestCase): remove the folder and verify that we can't create a file in it anymore. """ sftp.mkdir(FOLDER + '/subfolder') - f = sftp.open(FOLDER + '/subfolder/test', 'w') - f.close() + sftp.open(FOLDER + '/subfolder/test', 'w').close() sftp.remove(FOLDER + '/subfolder/test') sftp.rmdir(FOLDER + '/subfolder') try: - f = sftp.open(FOLDER + '/subfolder/test') + sftp.open(FOLDER + '/subfolder/test') # shouldn't be able to create that file - self.assert_(False, 'no exception at dummy file creation') + self.assertTrue(False, 'no exception at dummy file creation') except IOError: pass @@ -259,21 +283,16 @@ class SFTPTest (unittest.TestCase): and those files show up in sftp.listdir. """ try: - f = sftp.open(FOLDER + '/duck.txt', 'w') - f.close() - - f = sftp.open(FOLDER + '/fish.txt', 'w') - f.close() - - f = sftp.open(FOLDER + '/tertiary.py', 'w') - f.close() + sftp.open(FOLDER + '/duck.txt', 'w').close() + sftp.open(FOLDER + '/fish.txt', 'w').close() + sftp.open(FOLDER + '/tertiary.py', 'w').close() x = sftp.listdir(FOLDER) self.assertEqual(len(x), 3) - self.assert_('duck.txt' in x) - self.assert_('fish.txt' in x) - self.assert_('tertiary.py' in x) - self.assert_('random' not in x) + self.assertTrue('duck.txt' in x) + self.assertTrue('fish.txt' in x) + self.assertTrue('tertiary.py' in x) + self.assertTrue('random' not in x) finally: sftp.remove(FOLDER + '/duck.txt') sftp.remove(FOLDER + '/fish.txt') @@ -283,22 +302,21 @@ class SFTPTest (unittest.TestCase): """ verify that the setstat functions (chown, chmod, utime, truncate) work. """ - f = sftp.open(FOLDER + '/special', 'w') try: - f.write('x' * 1024) - f.close() + with sftp.open(FOLDER + '/special', 'w') as f: + f.write('x' * 1024) stat = sftp.stat(FOLDER + '/special') - sftp.chmod(FOLDER + '/special', (stat.st_mode & ~0777) | 0600) + sftp.chmod(FOLDER + '/special', (stat.st_mode & ~o777) | o600) stat = sftp.stat(FOLDER + '/special') - expected_mode = 0600 + expected_mode = o600 if sys.platform == 'win32': # chmod not really functional on windows - expected_mode = 0666 + expected_mode = o666 if sys.platform == 'cygwin': # even worse. - expected_mode = 0644 - self.assertEqual(stat.st_mode & 0777, expected_mode) + expected_mode = o644 + self.assertEqual(stat.st_mode & o777, expected_mode) self.assertEqual(stat.st_size, 1024) mtime = stat.st_mtime - 3600 @@ -322,40 +340,38 @@ class SFTPTest (unittest.TestCase): verify that the fsetstat functions (chown, chmod, utime, truncate) work on open files. """ - f = sftp.open(FOLDER + '/special', 'w') try: - f.write('x' * 1024) - f.close() - - f = sftp.open(FOLDER + '/special', 'r+') - stat = f.stat() - f.chmod((stat.st_mode & ~0777) | 0600) - stat = f.stat() - - expected_mode = 0600 - if sys.platform == 'win32': - # chmod not really functional on windows - expected_mode = 0666 - if sys.platform == 'cygwin': - # even worse. - expected_mode = 0644 - self.assertEqual(stat.st_mode & 0777, expected_mode) - self.assertEqual(stat.st_size, 1024) - - mtime = stat.st_mtime - 3600 - atime = stat.st_atime - 1800 - f.utime((atime, mtime)) - stat = f.stat() - 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() + with sftp.open(FOLDER + '/special', 'w') as f: + f.write('x' * 1024) + + with sftp.open(FOLDER + '/special', 'r+') as f: + stat = f.stat() + f.chmod((stat.st_mode & ~o777) | o600) + stat = f.stat() + + expected_mode = o600 + if sys.platform == 'win32': + # chmod not really functional on windows + expected_mode = o666 + if sys.platform == 'cygwin': + # even worse. + expected_mode = o644 + self.assertEqual(stat.st_mode & o777, expected_mode) + self.assertEqual(stat.st_size, 1024) + + mtime = stat.st_mtime - 3600 + atime = stat.st_atime - 1800 + f.utime((atime, mtime)) + stat = f.stat() + 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) finally: sftp.remove(FOLDER + '/special') @@ -367,25 +383,23 @@ class SFTPTest (unittest.TestCase): buffering is reset on 'seek'. """ try: - f = sftp.open(FOLDER + '/duck.txt', 'w') - f.write(ARTICLE) - f.close() - - f = sftp.open(FOLDER + '/duck.txt', 'r+') - line_number = 0 - loc = 0 - pos_list = [] - for line in f: - line_number += 1 - pos_list.append(loc) - loc = f.tell() - f.seek(pos_list[6], f.SEEK_SET) - self.assertEqual(f.readline(), 'Nouzilly, France.\n') - f.seek(pos_list[17], f.SEEK_SET) - self.assertEqual(f.readline()[:4], 'duck') - f.seek(pos_list[10], f.SEEK_SET) - self.assertEqual(f.readline(), 'duck types were equally resistant to exogenous insulin compared with chicken.\n') - f.close() + with sftp.open(FOLDER + '/duck.txt', 'w') as f: + f.write(ARTICLE) + + with sftp.open(FOLDER + '/duck.txt', 'r+') as f: + line_number = 0 + loc = 0 + pos_list = [] + for line in f: + line_number += 1 + pos_list.append(loc) + loc = f.tell() + f.seek(pos_list[6], f.SEEK_SET) + self.assertEqual(f.readline(), 'Nouzilly, France.\n') + f.seek(pos_list[17], f.SEEK_SET) + self.assertEqual(f.readline()[:4], 'duck') + f.seek(pos_list[10], f.SEEK_SET) + self.assertEqual(f.readline(), 'duck types were equally resistant to exogenous insulin compared with chicken.\n') finally: sftp.remove(FOLDER + '/duck.txt') @@ -394,18 +408,16 @@ class SFTPTest (unittest.TestCase): create a text file, seek back and change part of it, and verify that the changes worked. """ - f = sftp.open(FOLDER + '/testing.txt', 'w') try: - f.write('hello kitty.\n') - f.seek(-5, f.SEEK_CUR) - f.write('dd') - f.close() + with sftp.open(FOLDER + '/testing.txt', 'w') as f: + f.write('hello kitty.\n') + f.seek(-5, f.SEEK_CUR) + f.write('dd') self.assertEqual(sftp.stat(FOLDER + '/testing.txt').st_size, 13) - f = sftp.open(FOLDER + '/testing.txt', 'r') - data = f.read(20) - f.close() - self.assertEqual(data, 'hello kiddy.\n') + with sftp.open(FOLDER + '/testing.txt', 'r') as f: + data = f.read(20) + self.assertEqual(data, b'hello kiddy.\n') finally: sftp.remove(FOLDER + '/testing.txt') @@ -417,16 +429,14 @@ class SFTPTest (unittest.TestCase): # skip symlink tests on windows return - f = sftp.open(FOLDER + '/original.txt', 'w') try: - f.write('original\n') - f.close() + with sftp.open(FOLDER + '/original.txt', 'w') as f: + f.write('original\n') sftp.symlink('original.txt', FOLDER + '/link.txt') self.assertEqual(sftp.readlink(FOLDER + '/link.txt'), 'original.txt') - f = sftp.open(FOLDER + '/link.txt', 'r') - self.assertEqual(f.readlines(), [ 'original\n' ]) - f.close() + with sftp.open(FOLDER + '/link.txt', 'r') as f: + self.assertEqual(f.readlines(), ['original\n']) cwd = sftp.normalize('.') if cwd[-1] == '/': @@ -439,7 +449,7 @@ class SFTPTest (unittest.TestCase): self.assertEqual(sftp.stat(FOLDER + '/link.txt').st_size, 9) # the sftp server may be hiding extra path members from us, so the # length may be longer than we expect: - self.assert_(sftp.lstat(FOLDER + '/link2.txt').st_size >= len(abs_path)) + self.assertTrue(sftp.lstat(FOLDER + '/link2.txt').st_size >= len(abs_path)) self.assertEqual(sftp.stat(FOLDER + '/link2.txt').st_size, 9) self.assertEqual(sftp.stat(FOLDER + '/original.txt').st_size, 9) finally: @@ -460,18 +470,16 @@ class SFTPTest (unittest.TestCase): """ verify that buffered writes are automatically flushed on seek. """ - f = sftp.open(FOLDER + '/happy.txt', 'w', 1) try: - f.write('full line.\n') - f.write('partial') - f.seek(9, f.SEEK_SET) - f.write('?\n') - f.close() - - f = sftp.open(FOLDER + '/happy.txt', 'r') - self.assertEqual(f.readline(), 'full line?\n') - self.assertEqual(f.read(7), 'partial') - f.close() + with sftp.open(FOLDER + '/happy.txt', 'w', 1) as f: + f.write('full line.\n') + f.write('partial') + f.seek(9, f.SEEK_SET) + f.write('?\n') + + with sftp.open(FOLDER + '/happy.txt', 'r') as f: + self.assertEqual(f.readline(), u('full line?\n')) + self.assertEqual(f.read(7), b'partial') finally: try: sftp.remove(FOLDER + '/happy.txt') @@ -484,10 +492,10 @@ class SFTPTest (unittest.TestCase): error. """ pwd = sftp.normalize('.') - self.assert_(len(pwd) > 0) + self.assertTrue(len(pwd) > 0) f = sftp.normalize('./' + FOLDER) - self.assert_(len(f) > 0) - self.assertEquals(os.path.join(pwd, FOLDER), f) + self.assertTrue(len(f) > 0) + self.assertEqual(os.path.join(pwd, FOLDER), f) def test_F_mkdir(self): """ @@ -496,19 +504,19 @@ class SFTPTest (unittest.TestCase): try: sftp.mkdir(FOLDER + '/subfolder') except: - self.assert_(False, 'exception creating subfolder') + self.assertTrue(False, 'exception creating subfolder') try: sftp.mkdir(FOLDER + '/subfolder') - self.assert_(False, 'no exception overwriting subfolder') + self.assertTrue(False, 'no exception overwriting subfolder') except IOError: pass try: sftp.rmdir(FOLDER + '/subfolder') except: - self.assert_(False, 'exception removing subfolder') + self.assertTrue(False, 'exception removing subfolder') try: sftp.rmdir(FOLDER + '/subfolder') - self.assert_(False, 'no exception removing nonexistent subfolder') + self.assertTrue(False, 'no exception removing nonexistent subfolder') except IOError: pass @@ -523,17 +531,16 @@ class SFTPTest (unittest.TestCase): sftp.mkdir(FOLDER + '/alpha') sftp.chdir(FOLDER + '/alpha') sftp.mkdir('beta') - self.assertEquals(root + FOLDER + '/alpha', sftp.getcwd()) - self.assertEquals(['beta'], sftp.listdir('.')) + self.assertEqual(root + FOLDER + '/alpha', sftp.getcwd()) + self.assertEqual(['beta'], sftp.listdir('.')) sftp.chdir('beta') - f = sftp.open('fish', 'w') - f.write('hello\n') - f.close() + with sftp.open('fish', 'w') as f: + f.write('hello\n') sftp.chdir('..') - self.assertEquals(['fish'], sftp.listdir('beta')) + self.assertEqual(['fish'], sftp.listdir('beta')) sftp.chdir('..') - self.assertEquals(['fish'], sftp.listdir('alpha/beta')) + self.assertEqual(['fish'], sftp.listdir('alpha/beta')) finally: sftp.chdir(root) try: @@ -553,33 +560,32 @@ class SFTPTest (unittest.TestCase): """ verify that get/put work. """ - import os, warnings warnings.filterwarnings('ignore', 'tempnam.*') - localname = os.tempnam() - text = 'All I wanted was a plastic bunny rabbit.\n' - f = open(localname, 'wb') - f.write(text) - f.close() + fd, localname = mkstemp() + os.close(fd) + text = b'All I wanted was a plastic bunny rabbit.\n' + with open(localname, 'wb') as f: + f.write(text) saved_progress = [] + 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]) + with sftp.open(FOLDER + '/bunny.txt', 'rb') as f: + self.assertEqual(text, f.read(128)) + self.assertEqual((41, 41), saved_progress[-1]) os.unlink(localname) - localname = os.tempnam() + fd, localname = mkstemp() + os.close(fd) 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]) + with open(localname, 'rb') as f: + self.assertEqual(text, f.read(128)) + self.assertEqual((41, 41), saved_progress[-1]) os.unlink(localname) sftp.unlink(FOLDER + '/bunny.txt') @@ -590,20 +596,18 @@ class SFTPTest (unittest.TestCase): (it's an sftp extension that we support, and may be the only ones who support it.) """ - f = sftp.open(FOLDER + '/kitty.txt', 'w') - f.write('here kitty kitty' * 64) - f.close() + with sftp.open(FOLDER + '/kitty.txt', 'w') as f: + f.write('here kitty kitty' * 64) try: - f = sftp.open(FOLDER + '/kitty.txt', 'r') - sum = f.check('sha1') - self.assertEquals('91059CFC6615941378D413CB5ADAF4C5EB293402', hexlify(sum).upper()) - sum = f.check('md5', 0, 512) - self.assertEquals('93DE4788FCA28D471516963A1FE3856A', hexlify(sum).upper()) - sum = f.check('md5', 0, 0, 510) - self.assertEquals('EB3B45B8CD55A0707D99B177544A319F373183D241432BB2157AB9E46358C4AC90370B5CADE5D90336FC1716F90B36D6', - hexlify(sum).upper()) - f.close() + with sftp.open(FOLDER + '/kitty.txt', 'r') as f: + sum = f.check('sha1') + self.assertEqual('91059CFC6615941378D413CB5ADAF4C5EB293402', u(hexlify(sum)).upper()) + sum = f.check('md5', 0, 512) + self.assertEqual('93DE4788FCA28D471516963A1FE3856A', u(hexlify(sum)).upper()) + sum = f.check('md5', 0, 0, 510) + self.assertEqual('EB3B45B8CD55A0707D99B177544A319F373183D241432BB2157AB9E46358C4AC90370B5CADE5D90336FC1716F90B36D6', + u(hexlify(sum)).upper()) finally: sftp.unlink(FOLDER + '/kitty.txt') @@ -611,14 +615,13 @@ class SFTPTest (unittest.TestCase): """ verify that the 'x' flag works when opening a file. """ - f = sftp.open(FOLDER + '/unusual.txt', 'wx') - f.close() + sftp.open(FOLDER + '/unusual.txt', 'wx').close() try: try: - f = sftp.open(FOLDER + '/unusual.txt', 'wx') + sftp.open(FOLDER + '/unusual.txt', 'wx') self.fail('expected exception') - except IOError, x: + except IOError: pass finally: sftp.unlink(FOLDER + '/unusual.txt') @@ -627,44 +630,39 @@ class SFTPTest (unittest.TestCase): """ verify that unicode strings are encoded into utf8 correctly. """ - f = sftp.open(FOLDER + '/something', 'w') - f.write('okay') - f.close() + with sftp.open(FOLDER + '/something', 'w') as f: + f.write('okay') try: - sftp.rename(FOLDER + '/something', FOLDER + u'/\u00fcnic\u00f8de') - sftp.open(FOLDER + '/\xc3\xbcnic\xc3\xb8\x64\x65', 'r') - except Exception, e: - self.fail('exception ' + e) - sftp.unlink(FOLDER + '/\xc3\xbcnic\xc3\xb8\x64\x65') + sftp.rename(FOLDER + '/something', FOLDER + '/' + unicode_folder) + sftp.open(b(FOLDER) + utf8_folder, 'r') + except Exception as e: + self.fail('exception ' + str(e)) + sftp.unlink(b(FOLDER) + utf8_folder) def test_L_utf8_chdir(self): - sftp.mkdir(FOLDER + u'\u00fcnic\u00f8de') + sftp.mkdir(FOLDER + '/' + unicode_folder) try: - sftp.chdir(FOLDER + u'\u00fcnic\u00f8de') - f = sftp.open('something', 'w') - f.write('okay') - f.close() + sftp.chdir(FOLDER + '/' + unicode_folder) + with sftp.open('something', 'w') as f: + f.write('okay') sftp.unlink('something') finally: - sftp.chdir(None) - sftp.rmdir(FOLDER + u'\u00fcnic\u00f8de') + sftp.chdir() + sftp.rmdir(FOLDER + '/' + unicode_folder) def test_M_bad_readv(self): """ verify that readv at the end of the file doesn't essplode. """ - f = sftp.open(FOLDER + '/zero', 'w') - f.close() + sftp.open(FOLDER + '/zero', 'w').close() try: - f = sftp.open(FOLDER + '/zero', 'r') - data = f.readv([(0, 12)]) - f.close() + with sftp.open(FOLDER + '/zero', 'r') as f: + f.readv([(0, 12)]) - f = sftp.open(FOLDER + '/zero', 'r') - f.prefetch() - data = f.read(100) - f.close() + with sftp.open(FOLDER + '/zero', 'r') as f: + f.prefetch() + f.read(100) finally: sftp.unlink(FOLDER + '/zero') @@ -672,48 +670,115 @@ class SFTPTest (unittest.TestCase): """ verify that get/put work without confirmation. """ - import os, warnings warnings.filterwarnings('ignore', 'tempnam.*') - localname = os.tempnam() - text = 'All I wanted was a plastic bunny rabbit.\n' - f = open(localname, 'wb') - f.write(text) - f.close() + fd, localname = mkstemp() + os.close(fd) + text = b'All I wanted was a plastic bunny rabbit.\n' + with open(localname, 'wb') as f: + f.write(text) saved_progress = [] + def progress_callback(x, y): saved_progress.append((x, y)) res = sftp.put(localname, FOLDER + '/bunny.txt', progress_callback, False) - - self.assertEquals(SFTPAttributes().attr, res.attr) - f = sftp.open(FOLDER + '/bunny.txt', 'r') - self.assertEquals(text, f.read(128)) - f.close() - self.assertEquals((41, 41), saved_progress[-1]) + self.assertEqual(SFTPAttributes().attr, res.attr) + + with sftp.open(FOLDER + '/bunny.txt', 'r') as f: + self.assertEqual(text, f.read(128)) + self.assertEqual((41, 41), saved_progress[-1]) os.unlink(localname) sftp.unlink(FOLDER + '/bunny.txt') + def test_O_getcwd(self): + """ + verify that chdir/getcwd work. + """ + self.assertEqual(None, sftp.getcwd()) + root = sftp.normalize('.') + if root[-1] != '/': + root += '/' + try: + sftp.mkdir(FOLDER + '/alpha') + sftp.chdir(FOLDER + '/alpha') + self.assertEqual('/' + FOLDER + '/alpha', sftp.getcwd()) + finally: + sftp.chdir(root) + try: + sftp.rmdir(FOLDER + '/alpha') + except: + pass + 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') try: - f.write('first line\nsecond line\n') - f.seek(11, f.SEEK_SET) - f.write('third line\n') - f.close() + with sftp.open(FOLDER + '/append.txt', 'a') as f: + f.write('first line\nsecond line\n') + f.seek(11, f.SEEK_SET) + f.write('third line\n') + + with sftp.open(FOLDER + '/append.txt', 'r') as f: + self.assertEqual(f.stat().st_size, 34) + self.assertEqual(f.readline(), 'first line\n') + self.assertEqual(f.readline(), 'second line\n') + self.assertEqual(f.readline(), 'third line\n') + finally: + sftp.remove(FOLDER + '/append.txt') + + def test_putfo_empty_file(self): + """ + Send an empty file and confirm it is sent. + """ + target = FOLDER + '/empty file.txt' + stream = StringIO() + try: + attrs = sftp.putfo(stream, target) + # the returned attributes should not be null + self.assertNotEqual(attrs, None) + finally: + sftp.remove(target) + - f = sftp.open(FOLDER + '/append.txt', 'r') - self.assertEqual(f.stat().st_size, 34) - self.assertEqual(f.readline(), 'first line\n') - self.assertEqual(f.readline(), 'second line\n') - self.assertEqual(f.readline(), 'third line\n') + def test_N_file_with_percent(self): + """ + verify that we can create a file with a '%' in the filename. + ( it needs to be properly escaped by _log() ) + """ + self.assertTrue( paramiko.util.get_logger("paramiko").handlers, "This unit test requires logging to be enabled" ) + f = sftp.open(FOLDER + '/test%file', 'w') + try: + self.assertEqual(f.stat().st_size, 0) + finally: f.close() + sftp.remove(FOLDER + '/test%file') + + + def test_O_non_utf8_data(self): + """Test write() and read() of non utf8 data""" + try: + with sftp.open('%s/nonutf8data' % FOLDER, 'w') as f: + f.write(NON_UTF8_DATA) + with sftp.open('%s/nonutf8data' % FOLDER, 'r') as f: + data = f.read() + self.assertEqual(data, NON_UTF8_DATA) + with sftp.open('%s/nonutf8data' % FOLDER, 'wb') as f: + f.write(NON_UTF8_DATA) + with sftp.open('%s/nonutf8data' % FOLDER, 'rb') as f: + data = f.read() + self.assertEqual(data, NON_UTF8_DATA) finally: - sftp.remove(FOLDER + '/append.txt') + sftp.remove('%s/nonutf8data' % FOLDER) + +if __name__ == '__main__': + SFTPTest.init_loopback() + # logging is required by test_N_file_with_percent + paramiko.util.log_to_file('test_sftp.log') + from unittest import main + main() diff --git a/tests/test_sftp_big.py b/tests/test_sftp_big.py index a32a7000..abed27b8 100644 --- a/tests/test_sftp_big.py +++ b/tests/test_sftp_big.py @@ -7,7 +7,7 @@ # Software Foundation; either version 2.1 of the License, or (at your option) # any later version. # -# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY +# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR # A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # details. @@ -23,19 +23,15 @@ a real actual sftp server is contacted, and a new folder is created there to do test file operations in (so no existing files will be harmed). """ -import logging import os import random import struct import sys -import threading import time import unittest -import paramiko -from stub_sftp import StubServer, StubSFTPServer -from loop import LoopSocket -from test_sftp import get_sftp +from paramiko.common import o660 +from tests.test_sftp import get_sftp FOLDER = os.environ.get('TEST_FOLDER', 'temp-testing000') @@ -45,7 +41,7 @@ class BigSFTPTest (unittest.TestCase): def setUp(self): global FOLDER sftp = get_sftp() - for i in xrange(1000): + for i in range(1000): FOLDER = FOLDER[:-3] + '%03d' % i try: sftp.mkdir(FOLDER) @@ -65,19 +61,17 @@ class BigSFTPTest (unittest.TestCase): numfiles = 100 try: for i in range(numfiles): - f = sftp.open('%s/file%d.txt' % (FOLDER, i), 'w', 1) - f.write('this is file #%d.\n' % i) - f.close() - sftp.chmod('%s/file%d.txt' % (FOLDER, i), 0660) + with sftp.open('%s/file%d.txt' % (FOLDER, i), 'w', 1) as f: + f.write('this is file #%d.\n' % i) + sftp.chmod('%s/file%d.txt' % (FOLDER, i), o660) # now make sure every file is there, by creating a list of filenmes # and reading them in random order. - numlist = range(numfiles) + numlist = list(range(numfiles)) while len(numlist) > 0: r = numlist[random.randint(0, len(numlist) - 1)] - f = sftp.open('%s/file%d.txt' % (FOLDER, r)) - self.assertEqual(f.readline(), 'this is file #%d.\n' % r) - f.close() + with sftp.open('%s/file%d.txt' % (FOLDER, r)) as f: + self.assertEqual(f.readline(), 'this is file #%d.\n' % r) numlist.remove(r) finally: for i in range(numfiles): @@ -91,15 +85,14 @@ class BigSFTPTest (unittest.TestCase): write a 1MB file with no buffering. """ sftp = get_sftp() - kblob = (1024 * 'x') + kblob = (1024 * b'x') start = time.time() try: - f = sftp.open('%s/hongry.txt' % FOLDER, 'w') - for n in range(1024): - f.write(kblob) - if n % 128 == 0: - sys.stderr.write('.') - f.close() + with sftp.open('%s/hongry.txt' % FOLDER, 'w') as f: + for n in range(1024): + f.write(kblob) + if n % 128 == 0: + sys.stderr.write('.') sys.stderr.write(' ') self.assertEqual(sftp.stat('%s/hongry.txt' % FOLDER).st_size, 1024 * 1024) @@ -107,11 +100,10 @@ class BigSFTPTest (unittest.TestCase): sys.stderr.write('%ds ' % round(end - start)) start = time.time() - f = sftp.open('%s/hongry.txt' % FOLDER, 'r') - for n in range(1024): - data = f.read(1024) - self.assertEqual(data, kblob) - f.close() + with sftp.open('%s/hongry.txt' % FOLDER, 'r') as f: + for n in range(1024): + data = f.read(1024) + self.assertEqual(data, kblob) end = time.time() sys.stderr.write('%ds ' % round(end - start)) @@ -123,16 +115,15 @@ class BigSFTPTest (unittest.TestCase): write a 1MB file, with no linefeeds, using pipelining. """ sftp = get_sftp() - kblob = ''.join([struct.pack('>H', n) for n in xrange(512)]) + kblob = bytes().join([struct.pack('>H', n) for n in range(512)]) start = time.time() try: - f = sftp.open('%s/hongry.txt' % FOLDER, 'w') - f.set_pipelined(True) - for n in range(1024): - f.write(kblob) - if n % 128 == 0: - sys.stderr.write('.') - f.close() + with sftp.open('%s/hongry.txt' % FOLDER, 'wb') as f: + f.set_pipelined(True) + for n in range(1024): + f.write(kblob) + if n % 128 == 0: + sys.stderr.write('.') sys.stderr.write(' ') self.assertEqual(sftp.stat('%s/hongry.txt' % FOLDER).st_size, 1024 * 1024) @@ -140,22 +131,21 @@ class BigSFTPTest (unittest.TestCase): sys.stderr.write('%ds ' % round(end - start)) start = time.time() - f = sftp.open('%s/hongry.txt' % FOLDER, 'r') - f.prefetch() + with sftp.open('%s/hongry.txt' % FOLDER, 'rb') as f: + f.prefetch() - # read on odd boundaries to make sure the bytes aren't getting scrambled - n = 0 - k2blob = kblob + kblob - chunk = 629 - size = 1024 * 1024 - while n < size: - if n + chunk > size: - chunk = size - n - data = f.read(chunk) - offset = n % 1024 - self.assertEqual(data, k2blob[offset:offset + chunk]) - n += chunk - f.close() + # read on odd boundaries to make sure the bytes aren't getting scrambled + n = 0 + k2blob = kblob + kblob + chunk = 629 + size = 1024 * 1024 + while n < size: + if n + chunk > size: + chunk = size - n + data = f.read(chunk) + offset = n % 1024 + self.assertEqual(data, k2blob[offset:offset + chunk]) + n += chunk end = time.time() sys.stderr.write('%ds ' % round(end - start)) @@ -164,15 +154,14 @@ class BigSFTPTest (unittest.TestCase): def test_4_prefetch_seek(self): sftp = get_sftp() - kblob = ''.join([struct.pack('>H', n) for n in xrange(512)]) + kblob = bytes().join([struct.pack('>H', n) for n in range(512)]) try: - f = sftp.open('%s/hongry.txt' % FOLDER, 'w') - f.set_pipelined(True) - for n in range(1024): - f.write(kblob) - if n % 128 == 0: - sys.stderr.write('.') - f.close() + with sftp.open('%s/hongry.txt' % FOLDER, 'wb') as f: + f.set_pipelined(True) + for n in range(1024): + f.write(kblob) + if n % 128 == 0: + sys.stderr.write('.') sys.stderr.write(' ') self.assertEqual(sftp.stat('%s/hongry.txt' % FOLDER).st_size, 1024 * 1024) @@ -180,21 +169,20 @@ class BigSFTPTest (unittest.TestCase): start = time.time() k2blob = kblob + kblob chunk = 793 - for i in xrange(10): - f = sftp.open('%s/hongry.txt' % FOLDER, 'r') - f.prefetch() - base_offset = (512 * 1024) + 17 * random.randint(1000, 2000) - offsets = [base_offset + j * chunk for j in xrange(100)] - # randomly seek around and read them out - for j in xrange(100): - offset = offsets[random.randint(0, len(offsets) - 1)] - offsets.remove(offset) - f.seek(offset) - data = f.read(chunk) - n_offset = offset % 1024 - self.assertEqual(data, k2blob[n_offset:n_offset + chunk]) - offset += chunk - f.close() + for i in range(10): + with sftp.open('%s/hongry.txt' % FOLDER, 'rb') as f: + f.prefetch() + base_offset = (512 * 1024) + 17 * random.randint(1000, 2000) + offsets = [base_offset + j * chunk for j in range(100)] + # randomly seek around and read them out + for j in range(100): + offset = offsets[random.randint(0, len(offsets) - 1)] + offsets.remove(offset) + f.seek(offset) + data = f.read(chunk) + n_offset = offset % 1024 + self.assertEqual(data, k2blob[n_offset:n_offset + chunk]) + offset += chunk end = time.time() sys.stderr.write('%ds ' % round(end - start)) finally: @@ -202,15 +190,14 @@ class BigSFTPTest (unittest.TestCase): def test_5_readv_seek(self): sftp = get_sftp() - kblob = ''.join([struct.pack('>H', n) for n in xrange(512)]) + kblob = bytes().join([struct.pack('>H', n) for n in range(512)]) try: - f = sftp.open('%s/hongry.txt' % FOLDER, 'w') - f.set_pipelined(True) - for n in range(1024): - f.write(kblob) - if n % 128 == 0: - sys.stderr.write('.') - f.close() + with sftp.open('%s/hongry.txt' % FOLDER, 'wb') as f: + f.set_pipelined(True) + for n in range(1024): + f.write(kblob) + if n % 128 == 0: + sys.stderr.write('.') sys.stderr.write(' ') self.assertEqual(sftp.stat('%s/hongry.txt' % FOLDER).st_size, 1024 * 1024) @@ -218,22 +205,21 @@ class BigSFTPTest (unittest.TestCase): start = time.time() k2blob = kblob + kblob chunk = 793 - for i in xrange(10): - f = sftp.open('%s/hongry.txt' % FOLDER, 'r') - base_offset = (512 * 1024) + 17 * random.randint(1000, 2000) - # make a bunch of offsets and put them in random order - offsets = [base_offset + j * chunk for j in xrange(100)] - readv_list = [] - for j in xrange(100): - o = offsets[random.randint(0, len(offsets) - 1)] - offsets.remove(o) - readv_list.append((o, chunk)) - ret = f.readv(readv_list) - for i in xrange(len(readv_list)): - offset = readv_list[i][0] - n_offset = offset % 1024 - self.assertEqual(ret.next(), k2blob[n_offset:n_offset + chunk]) - f.close() + for i in range(10): + with sftp.open('%s/hongry.txt' % FOLDER, 'rb') as f: + base_offset = (512 * 1024) + 17 * random.randint(1000, 2000) + # make a bunch of offsets and put them in random order + offsets = [base_offset + j * chunk for j in range(100)] + readv_list = [] + for j in range(100): + o = offsets[random.randint(0, len(offsets) - 1)] + offsets.remove(o) + readv_list.append((o, chunk)) + ret = f.readv(readv_list) + for i in range(len(readv_list)): + offset = readv_list[i][0] + n_offset = offset % 1024 + self.assertEqual(next(ret), k2blob[n_offset:n_offset + chunk]) end = time.time() sys.stderr.write('%ds ' % round(end - start)) finally: @@ -245,30 +231,28 @@ class BigSFTPTest (unittest.TestCase): without using it, to verify that paramiko doesn't get confused. """ sftp = get_sftp() - kblob = (1024 * 'x') + kblob = (1024 * b'x') try: - f = sftp.open('%s/hongry.txt' % FOLDER, 'w') - f.set_pipelined(True) - for n in range(1024): - f.write(kblob) - if n % 128 == 0: - sys.stderr.write('.') - f.close() + with sftp.open('%s/hongry.txt' % FOLDER, 'w') as f: + f.set_pipelined(True) + for n in range(1024): + f.write(kblob) + if n % 128 == 0: + sys.stderr.write('.') sys.stderr.write(' ') self.assertEqual(sftp.stat('%s/hongry.txt' % FOLDER).st_size, 1024 * 1024) for i in range(10): - f = sftp.open('%s/hongry.txt' % FOLDER, 'r') + with sftp.open('%s/hongry.txt' % FOLDER, 'r') as f: + f.prefetch() + with sftp.open('%s/hongry.txt' % FOLDER, 'r') as f: f.prefetch() - f = sftp.open('%s/hongry.txt' % FOLDER, 'r') - f.prefetch() - for n in range(1024): - data = f.read(1024) - self.assertEqual(data, kblob) - if n % 128 == 0: - sys.stderr.write('.') - f.close() + for n in range(1024): + data = f.read(1024) + self.assertEqual(data, kblob) + if n % 128 == 0: + sys.stderr.write('.') sys.stderr.write(' ') finally: sftp.remove('%s/hongry.txt' % FOLDER) @@ -278,35 +262,33 @@ class BigSFTPTest (unittest.TestCase): verify that prefetch and readv don't conflict with each other. """ sftp = get_sftp() - kblob = ''.join([struct.pack('>H', n) for n in xrange(512)]) + kblob = bytes().join([struct.pack('>H', n) for n in range(512)]) try: - f = sftp.open('%s/hongry.txt' % FOLDER, 'w') - f.set_pipelined(True) - for n in range(1024): - f.write(kblob) - if n % 128 == 0: - sys.stderr.write('.') - f.close() + with sftp.open('%s/hongry.txt' % FOLDER, 'wb') as f: + f.set_pipelined(True) + for n in range(1024): + f.write(kblob) + if n % 128 == 0: + sys.stderr.write('.') sys.stderr.write(' ') self.assertEqual(sftp.stat('%s/hongry.txt' % FOLDER).st_size, 1024 * 1024) - f = sftp.open('%s/hongry.txt' % FOLDER, 'r') - f.prefetch() - data = f.read(1024) - self.assertEqual(data, kblob) - - chunk_size = 793 - base_offset = 512 * 1024 - k2blob = kblob + kblob - chunks = [(base_offset + (chunk_size * i), chunk_size) for i in range(20)] - for data in f.readv(chunks): - offset = base_offset % 1024 - self.assertEqual(chunk_size, len(data)) - self.assertEqual(k2blob[offset:offset + chunk_size], data) - base_offset += chunk_size - - f.close() + with sftp.open('%s/hongry.txt' % FOLDER, 'rb') as f: + f.prefetch() + data = f.read(1024) + self.assertEqual(data, kblob) + + chunk_size = 793 + base_offset = 512 * 1024 + k2blob = kblob + kblob + chunks = [(base_offset + (chunk_size * i), chunk_size) for i in range(20)] + for data in f.readv(chunks): + offset = base_offset % 1024 + self.assertEqual(chunk_size, len(data)) + self.assertEqual(k2blob[offset:offset + chunk_size], data) + base_offset += chunk_size + sys.stderr.write(' ') finally: sftp.remove('%s/hongry.txt' % FOLDER) @@ -317,26 +299,24 @@ class BigSFTPTest (unittest.TestCase): returned as a single blob. """ sftp = get_sftp() - kblob = ''.join([struct.pack('>H', n) for n in xrange(512)]) + kblob = bytes().join([struct.pack('>H', n) for n in range(512)]) try: - f = sftp.open('%s/hongry.txt' % FOLDER, 'w') - f.set_pipelined(True) - for n in range(1024): - f.write(kblob) - if n % 128 == 0: - sys.stderr.write('.') - f.close() + with sftp.open('%s/hongry.txt' % FOLDER, 'wb') as f: + f.set_pipelined(True) + for n in range(1024): + f.write(kblob) + if n % 128 == 0: + sys.stderr.write('.') sys.stderr.write(' ') self.assertEqual(sftp.stat('%s/hongry.txt' % FOLDER).st_size, 1024 * 1024) - f = sftp.open('%s/hongry.txt' % FOLDER, 'r') - data = list(f.readv([(23 * 1024, 128 * 1024)])) - self.assertEqual(1, len(data)) - data = data[0] - self.assertEqual(128 * 1024, len(data)) + with sftp.open('%s/hongry.txt' % FOLDER, 'rb') as f: + data = list(f.readv([(23 * 1024, 128 * 1024)])) + self.assertEqual(1, len(data)) + data = data[0] + self.assertEqual(128 * 1024, len(data)) - f.close() sys.stderr.write(' ') finally: sftp.remove('%s/hongry.txt' % FOLDER) @@ -348,9 +328,8 @@ class BigSFTPTest (unittest.TestCase): sftp = get_sftp() mblob = (1024 * 1024 * 'x') try: - f = sftp.open('%s/hongry.txt' % FOLDER, 'w', 128 * 1024) - f.write(mblob) - f.close() + with sftp.open('%s/hongry.txt' % FOLDER, 'w', 128 * 1024) as f: + f.write(mblob) self.assertEqual(sftp.stat('%s/hongry.txt' % FOLDER).st_size, 1024 * 1024) finally: @@ -365,21 +344,26 @@ class BigSFTPTest (unittest.TestCase): t.packetizer.REKEY_BYTES = 512 * 1024 k32blob = (32 * 1024 * 'x') try: - f = sftp.open('%s/hongry.txt' % FOLDER, 'w', 128 * 1024) - for i in xrange(32): - f.write(k32blob) - f.close() - + with sftp.open('%s/hongry.txt' % FOLDER, 'w', 128 * 1024) as f: + for i in range(32): + f.write(k32blob) + self.assertEqual(sftp.stat('%s/hongry.txt' % FOLDER).st_size, 1024 * 1024) - self.assertNotEquals(t.H, t.session_id) + self.assertNotEqual(t.H, t.session_id) # try to read it too. - f = sftp.open('%s/hongry.txt' % FOLDER, 'r', 128 * 1024) - f.prefetch() - total = 0 - while total < 1024 * 1024: - total += len(f.read(32 * 1024)) - f.close() + with sftp.open('%s/hongry.txt' % FOLDER, 'r', 128 * 1024) as f: + f.prefetch() + total = 0 + while total < 1024 * 1024: + total += len(f.read(32 * 1024)) finally: sftp.remove('%s/hongry.txt' % FOLDER) t.packetizer.REKEY_BYTES = pow(2, 30) + + +if __name__ == '__main__': + from tests.test_sftp import SFTPTest + SFTPTest.init_loopback() + from unittest import main + main() diff --git a/tests/test_transport.py b/tests/test_transport.py index 1c57d18d..485a18e8 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -7,7 +7,7 @@ # Software Foundation; either version 2.1 of the License, or (at your option) # any later version. # -# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY +# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR # A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # details. @@ -20,23 +20,22 @@ Some unit tests for the ssh2 protocol in Transport. """ -from binascii import hexlify, unhexlify +from binascii import hexlify import select import socket -import sys import time import threading -import unittest import random from paramiko import Transport, SecurityOptions, ServerInterface, RSAKey, DSSKey, \ - SSHException, BadAuthenticationType, InteractiveQuery, ChannelException -from paramiko import AUTH_FAILED, AUTH_PARTIALLY_SUCCESSFUL, AUTH_SUCCESSFUL + SSHException, ChannelException +from paramiko import AUTH_FAILED, AUTH_SUCCESSFUL from paramiko import OPEN_SUCCEEDED, OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED -from paramiko.common import MSG_KEXINIT, MSG_CHANNEL_WINDOW_ADJUST +from paramiko.common import MSG_KEXINIT, cMSG_CHANNEL_WINDOW_ADJUST +from paramiko.py3compat import bytes from paramiko.message import Message -from loop import LoopSocket -from util import ParamikoTest +from tests.loop import LoopSocket +from tests.util import ParamikoTest, test_path LONG_BANNER = """\ @@ -55,7 +54,7 @@ Maybe. class NullServer (ServerInterface): paranoid_did_password = False paranoid_did_public_key = False - paranoid_key = DSSKey.from_private_key_file('tests/test_dss.key') + paranoid_key = DSSKey.from_private_key_file(test_path('test_dss.key')) def get_allowed_auths(self, username): if username == 'slowdive': @@ -121,8 +120,8 @@ class TransportTest(ParamikoTest): self.sockc.close() def setup_test_server(self, client_options=None, server_options=None): - host_key = RSAKey.from_private_key_file('tests/test_rsa.key') - public_host_key = RSAKey(data=str(host_key)) + host_key = RSAKey.from_private_key_file(test_path('test_rsa.key')) + public_host_key = RSAKey(data=host_key.asbytes()) self.ts.add_server_key(host_key) if client_options is not None: @@ -132,37 +131,37 @@ class TransportTest(ParamikoTest): event = threading.Event() self.server = NullServer() - self.assert_(not event.isSet()) + self.assertTrue(not event.isSet()) self.ts.start_server(event, self.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()) + self.assertTrue(event.isSet()) + self.assertTrue(self.ts.is_active()) def test_1_security_options(self): o = self.tc.get_security_options() - self.assertEquals(type(o), SecurityOptions) - self.assert_(('aes256-cbc', 'blowfish-cbc') != o.ciphers) + self.assertEqual(type(o), SecurityOptions) + self.assertTrue(('aes256-cbc', 'blowfish-cbc') != o.ciphers) o.ciphers = ('aes256-cbc', 'blowfish-cbc') - self.assertEquals(('aes256-cbc', 'blowfish-cbc'), o.ciphers) + self.assertEqual(('aes256-cbc', 'blowfish-cbc'), o.ciphers) try: o.ciphers = ('aes256-cbc', 'made-up-cipher') - self.assert_(False) + self.assertTrue(False) except ValueError: pass try: o.ciphers = 23 - self.assert_(False) + self.assertTrue(False) except TypeError: pass def test_2_compute_key(self): - self.tc.K = 123281095979686581523377256114209720774539068973101330872763622971399429481072519713536292772709507296759612401802191955568143056534122385270077606457721553469730659233569339356140085284052436697480759510519672848743794433460113118986816826624865291116513647975790797391795651716378444844877749505443714557929L - self.tc.H = unhexlify('0C8307CDE6856FF30BA93684EB0F04C2520E9ED3') + self.tc.K = 123281095979686581523377256114209720774539068973101330872763622971399429481072519713536292772709507296759612401802191955568143056534122385270077606457721553469730659233569339356140085284052436697480759510519672848743794433460113118986816826624865291116513647975790797391795651716378444844877749505443714557929 + self.tc.H = b'\x0C\x83\x07\xCD\xE6\x85\x6F\xF3\x0B\xA9\x36\x84\xEB\x0F\x04\xC2\x52\x0E\x9E\xD3' self.tc.session_id = self.tc.H key = self.tc._compute_key('C', 32) - self.assertEquals('207E66594CA87C44ECCBA3B3CD39FDDB378E6FDB0F97C54B2AA0CFBF900CD995', + self.assertEqual(b'207E66594CA87C44ECCBA3B3CD39FDDB378E6FDB0F97C54B2AA0CFBF900CD995', hexlify(key).upper()) def test_3_simple(self): @@ -171,44 +170,44 @@ class TransportTest(ParamikoTest): loopback sockets. this is hardly "simple" but it's simpler than the later tests. :) """ - host_key = RSAKey.from_private_key_file('tests/test_rsa.key') - public_host_key = RSAKey(data=str(host_key)) + host_key = RSAKey.from_private_key_file(test_path('test_rsa.key')) + public_host_key = RSAKey(data=host_key.asbytes()) self.ts.add_server_key(host_key) event = threading.Event() server = NullServer() - self.assert_(not event.isSet()) - self.assertEquals(None, self.tc.get_username()) - self.assertEquals(None, self.ts.get_username()) - self.assertEquals(False, self.tc.is_authenticated()) - self.assertEquals(False, self.ts.is_authenticated()) + self.assertTrue(not event.isSet()) + self.assertEqual(None, self.tc.get_username()) + self.assertEqual(None, self.ts.get_username()) + self.assertEqual(False, self.tc.is_authenticated()) + self.assertEqual(False, self.ts.is_authenticated()) 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()) - self.assertEquals('slowdive', self.tc.get_username()) - self.assertEquals('slowdive', self.ts.get_username()) - self.assertEquals(True, self.tc.is_authenticated()) - self.assertEquals(True, self.ts.is_authenticated()) + self.assertTrue(event.isSet()) + self.assertTrue(self.ts.is_active()) + self.assertEqual('slowdive', self.tc.get_username()) + self.assertEqual('slowdive', self.ts.get_username()) + self.assertEqual(True, self.tc.is_authenticated()) + self.assertEqual(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)) + host_key = RSAKey.from_private_key_file(test_path('test_rsa.key')) + public_host_key = RSAKey(data=host_key.asbytes()) self.ts.add_server_key(host_key) event = threading.Event() server = NullServer() - self.assert_(not event.isSet()) + self.assertTrue(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()) + self.assertTrue(event.isSet()) + self.assertTrue(self.ts.is_active()) def test_4_special(self): """ @@ -219,10 +218,10 @@ class TransportTest(ParamikoTest): options.ciphers = ('aes256-cbc',) options.digests = ('hmac-md5-96',) self.setup_test_server(client_options=force_algorithms) - self.assertEquals('aes256-cbc', self.tc.local_cipher) - self.assertEquals('aes256-cbc', self.tc.remote_cipher) - self.assertEquals(12, self.tc.packetizer.get_mac_size_out()) - self.assertEquals(12, self.tc.packetizer.get_mac_size_in()) + self.assertEqual('aes256-cbc', self.tc.local_cipher) + self.assertEqual('aes256-cbc', self.tc.remote_cipher) + self.assertEqual(12, self.tc.packetizer.get_mac_size_out()) + self.assertEqual(12, self.tc.packetizer.get_mac_size_in()) self.tc.send_ignore(1024) self.tc.renegotiate_keys() @@ -233,10 +232,10 @@ class TransportTest(ParamikoTest): verify that the keepalive will be sent. """ self.setup_test_server() - self.assertEquals(None, getattr(self.server, '_global_request', None)) + self.assertEqual(None, getattr(self.server, '_global_request', None)) self.tc.set_keepalive(1) time.sleep(2) - self.assertEquals('keepalive@lag.net', self.server._global_request) + self.assertEqual('keepalive@lag.net', self.server._global_request) def test_6_exec_command(self): """ @@ -248,8 +247,8 @@ class TransportTest(ParamikoTest): schan = self.ts.accept(1.0) try: chan.exec_command('no') - self.assert_(False) - except SSHException, x: + self.assertTrue(False) + except SSHException: pass chan = self.tc.open_session() @@ -260,11 +259,11 @@ class TransportTest(ParamikoTest): schan.close() f = chan.makefile() - self.assertEquals('Hello there.\n', f.readline()) - self.assertEquals('', f.readline()) + self.assertEqual('Hello there.\n', f.readline()) + self.assertEqual('', f.readline()) f = chan.makefile_stderr() - self.assertEquals('This is on stderr.\n', f.readline()) - self.assertEquals('', f.readline()) + self.assertEqual('This is on stderr.\n', f.readline()) + self.assertEqual('', f.readline()) # now try it with combined stdout/stderr chan = self.tc.open_session() @@ -276,9 +275,9 @@ class TransportTest(ParamikoTest): chan.set_combine_stderr(True) f = chan.makefile() - self.assertEquals('Hello there.\n', f.readline()) - self.assertEquals('This is on stderr.\n', f.readline()) - self.assertEquals('', f.readline()) + self.assertEqual('Hello there.\n', f.readline()) + self.assertEqual('This is on stderr.\n', f.readline()) + self.assertEqual('', f.readline()) def test_7_invoke_shell(self): """ @@ -290,9 +289,9 @@ class TransportTest(ParamikoTest): schan = self.ts.accept(1.0) chan.send('communist j. cat\n') f = schan.makefile() - self.assertEquals('communist j. cat\n', f.readline()) + self.assertEqual('communist j. cat\n', f.readline()) chan.close() - self.assertEquals('', f.readline()) + self.assertEqual('', f.readline()) def test_8_channel_exception(self): """ @@ -302,8 +301,8 @@ class TransportTest(ParamikoTest): try: chan = self.tc.open_channel('bogus') self.fail('expected exception') - except ChannelException, x: - self.assert_(x.code == OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED) + except ChannelException as e: + self.assertTrue(e.code == OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED) def test_9_exit_status(self): """ @@ -315,7 +314,7 @@ class TransportTest(ParamikoTest): schan = self.ts.accept(1.0) chan.exec_command('yes') schan.send('Hello there.\n') - self.assert_(not chan.exit_status_ready()) + self.assertTrue(not chan.exit_status_ready()) # trigger an EOF schan.shutdown_read() schan.shutdown_write() @@ -323,15 +322,15 @@ class TransportTest(ParamikoTest): schan.close() f = chan.makefile() - self.assertEquals('Hello there.\n', f.readline()) - self.assertEquals('', f.readline()) + self.assertEqual('Hello there.\n', f.readline()) + self.assertEqual('', f.readline()) count = 0 while not chan.exit_status_ready(): time.sleep(0.1) count += 1 if count > 50: raise Exception("timeout") - self.assertEquals(23, chan.recv_exit_status()) + self.assertEqual(23, chan.recv_exit_status()) chan.close() def test_A_select(self): @@ -345,9 +344,9 @@ class TransportTest(ParamikoTest): # nothing should be ready r, w, e = select.select([chan], [], [], 0.1) - self.assertEquals([], r) - self.assertEquals([], w) - self.assertEquals([], e) + self.assertEqual([], r) + self.assertEqual([], w) + self.assertEqual([], e) schan.send('hello\n') @@ -357,17 +356,17 @@ class TransportTest(ParamikoTest): if chan in r: break time.sleep(0.1) - self.assertEquals([chan], r) - self.assertEquals([], w) - self.assertEquals([], e) + self.assertEqual([chan], r) + self.assertEqual([], w) + self.assertEqual([], e) - self.assertEquals('hello\n', chan.recv(6)) + self.assertEqual(b'hello\n', chan.recv(6)) # and, should be dead again now r, w, e = select.select([chan], [], [], 0.1) - self.assertEquals([], r) - self.assertEquals([], w) - self.assertEquals([], e) + self.assertEqual([], r) + self.assertEqual([], w) + self.assertEqual([], e) schan.close() @@ -377,17 +376,17 @@ class TransportTest(ParamikoTest): if chan in r: break time.sleep(0.1) - self.assertEquals([chan], r) - self.assertEquals([], w) - self.assertEquals([], e) - self.assertEquals('', chan.recv(16)) + self.assertEqual([chan], r) + self.assertEqual([], w) + self.assertEqual([], e) + self.assertEqual(bytes(), chan.recv(16)) # make sure the pipe is still open for now... p = chan._pipe - self.assertEquals(False, p._closed) + self.assertEqual(False, p._closed) chan.close() # ...and now is closed. - self.assertEquals(True, p._closed) + self.assertEqual(True, p._closed) def test_B_renegotiate(self): """ @@ -399,17 +398,17 @@ class TransportTest(ParamikoTest): chan.exec_command('yes') schan = self.ts.accept(1.0) - self.assertEquals(self.tc.H, self.tc.session_id) + self.assertEqual(self.tc.H, self.tc.session_id) for i in range(20): chan.send('x' * 1024) chan.close() # allow a few seconds for the rekeying to complete - for i in xrange(50): + for i in range(50): if self.tc.H != self.tc.session_id: break time.sleep(0.1) - self.assertNotEquals(self.tc.H, self.tc.session_id) + self.assertNotEqual(self.tc.H, self.tc.session_id) schan.close() @@ -428,8 +427,8 @@ class TransportTest(ParamikoTest): chan.send('x' * 1024) bytes2 = self.tc.packetizer._Packetizer__sent_bytes # tests show this is actually compressed to *52 bytes*! including packet overhead! nice!! :) - self.assert_(bytes2 - bytes < 1024) - self.assertEquals(52, bytes2 - bytes) + self.assertTrue(bytes2 - bytes < 1024) + self.assertEqual(52, bytes2 - bytes) chan.close() schan.close() @@ -444,24 +443,25 @@ class TransportTest(ParamikoTest): schan = self.ts.accept(1.0) requested = [] - def handler(c, (addr, port)): + def handler(c, addr_port): + addr, port = addr_port requested.append((addr, port)) self.tc._queue_incoming_channel(c) - self.assertEquals(None, getattr(self.server, '_x11_screen_number', None)) + self.assertEqual(None, getattr(self.server, '_x11_screen_number', None)) cookie = chan.request_x11(0, single_connection=True, handler=handler) - self.assertEquals(0, self.server._x11_screen_number) - self.assertEquals('MIT-MAGIC-COOKIE-1', self.server._x11_auth_protocol) - self.assertEquals(cookie, self.server._x11_auth_cookie) - self.assertEquals(True, self.server._x11_single_connection) + self.assertEqual(0, self.server._x11_screen_number) + self.assertEqual('MIT-MAGIC-COOKIE-1', self.server._x11_auth_protocol) + self.assertEqual(cookie, self.server._x11_auth_cookie) + self.assertEqual(True, self.server._x11_single_connection) x11_server = self.ts.open_x11_channel(('localhost', 6093)) x11_client = self.tc.accept() - self.assertEquals('localhost', requested[0][0]) - self.assertEquals(6093, requested[0][1]) + self.assertEqual('localhost', requested[0][0]) + self.assertEqual(6093, requested[0][1]) x11_server.send('hello') - self.assertEquals('hello', x11_client.recv(5)) + self.assertEqual(b'hello', x11_client.recv(5)) x11_server.close() x11_client.close() @@ -479,13 +479,13 @@ class TransportTest(ParamikoTest): schan = self.ts.accept(1.0) requested = [] - def handler(c, (origin_addr, origin_port), (server_addr, server_port)): - requested.append((origin_addr, origin_port)) - requested.append((server_addr, server_port)) + def handler(c, origin_addr_port, server_addr_port): + requested.append(origin_addr_port) + requested.append(server_addr_port) self.tc._queue_incoming_channel(c) port = self.tc.request_port_forward('127.0.0.1', 0, handler) - self.assertEquals(port, self.server._listen.getsockname()[1]) + self.assertEqual(port, self.server._listen.getsockname()[1]) cs = socket.socket() cs.connect(('127.0.0.1', port)) @@ -494,7 +494,7 @@ class TransportTest(ParamikoTest): cch = self.tc.accept() sch.send('hello') - self.assertEquals('hello', cch.recv(5)) + self.assertEqual(b'hello', cch.recv(5)) sch.close() cch.close() ss.close() @@ -526,12 +526,12 @@ class TransportTest(ParamikoTest): cch.connect(self.server._tcpip_dest) ss, _ = greeting_server.accept() - ss.send('Hello!\n') + ss.send(b'Hello!\n') ss.close() sch.send(cch.recv(8192)) sch.close() - self.assertEquals('Hello!\n', cs.recv(7)) + self.assertEqual(b'Hello!\n', cs.recv(7)) cs.close() def test_G_stderr_select(self): @@ -546,9 +546,9 @@ class TransportTest(ParamikoTest): # nothing should be ready r, w, e = select.select([chan], [], [], 0.1) - self.assertEquals([], r) - self.assertEquals([], w) - self.assertEquals([], e) + self.assertEqual([], r) + self.assertEqual([], w) + self.assertEqual([], e) schan.send_stderr('hello\n') @@ -558,17 +558,17 @@ class TransportTest(ParamikoTest): if chan in r: break time.sleep(0.1) - self.assertEquals([chan], r) - self.assertEquals([], w) - self.assertEquals([], e) + self.assertEqual([chan], r) + self.assertEqual([], w) + self.assertEqual([], e) - self.assertEquals('hello\n', chan.recv_stderr(6)) + self.assertEqual(b'hello\n', chan.recv_stderr(6)) # and, should be dead again now r, w, e = select.select([chan], [], [], 0.1) - self.assertEquals([], r) - self.assertEquals([], w) - self.assertEquals([], e) + self.assertEqual([], r) + self.assertEqual([], w) + self.assertEqual([], e) schan.close() chan.close() @@ -582,7 +582,7 @@ class TransportTest(ParamikoTest): chan.invoke_shell() schan = self.ts.accept(1.0) - self.assertEquals(chan.send_ready(), True) + self.assertEqual(chan.send_ready(), True) total = 0 K = '*' * 1024 while total < 1024 * 1024: @@ -590,11 +590,11 @@ class TransportTest(ParamikoTest): total += len(K) if not chan.send_ready(): break - self.assert_(total < 1024 * 1024) + self.assertTrue(total < 1024 * 1024) schan.close() chan.close() - self.assertEquals(chan.send_ready(), True) + self.assertEqual(chan.send_ready(), True) def test_I_rekey_deadlock(self): """ @@ -657,7 +657,7 @@ class TransportTest(ParamikoTest): def run(self): try: - for i in xrange(1, 1+self.iterations): + for i in range(1, 1+self.iterations): if self.done_event.isSet(): break self.watchdog_event.set() @@ -706,7 +706,7 @@ class TransportTest(ParamikoTest): # Simulate in-transit MSG_CHANNEL_WINDOW_ADJUST by sending it # before responding to the incoming MSG_KEXINIT. m2 = Message() - m2.add_byte(chr(MSG_CHANNEL_WINDOW_ADJUST)) + m2.add_byte(cMSG_CHANNEL_WINDOW_ADJUST) m2.add_int(chan.remote_chanid) m2.add_int(1) # bytes to add self._send_message(m2) diff --git a/tests/test_util.py b/tests/test_util.py index 093a2157..44fb8ab5 100644 --- a/tests/test_util.py +++ b/tests/test_util.py @@ -7,7 +7,7 @@ # Software Foundation; either version 2.1 of the License, or (at your option) # any later version. # -# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY +# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR # A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # details. @@ -21,15 +21,15 @@ Some unit tests for utility functions. """ from binascii import hexlify -import cStringIO import errno import os -import unittest -from Crypto.Hash import SHA +from hashlib import sha1 + import paramiko.util from paramiko.util import lookup_ssh_host_config as host_config +from paramiko.py3compat import StringIO, byte_ord -from util import ParamikoTest +from tests.util import ParamikoTest test_config_file = """\ Host * @@ -65,7 +65,7 @@ class UtilTest(ParamikoTest): """ verify that all the classes can be imported from paramiko. """ - symbols = globals().keys() + symbols = list(globals().keys()) self.assertTrue('Transport' in symbols) self.assertTrue('SSHClient' in symbols) self.assertTrue('MissingHostKeyPolicy' in symbols) @@ -101,58 +101,60 @@ class UtilTest(ParamikoTest): def test_2_parse_config(self): global test_config_file - f = cStringIO.StringIO(test_config_file) + f = StringIO(test_config_file) config = paramiko.util.parse_ssh_config(f) - self.assertEquals(config._config, - [ {'identityfile': '~/.ssh/id_rsa', 'host': '*', 'user': 'robey', - 'crazy': 'something dumb '}, - {'host': '*.example.com', 'user': 'bjork', 'port': '3333'}, - {'host': 'spoo.example.com', 'crazy': 'something else'}]) + self.assertEqual(config._config, + [{'host': ['*'], 'config': {}}, {'host': ['*'], 'config': {'identityfile': ['~/.ssh/id_rsa'], 'user': 'robey'}}, + {'host': ['*.example.com'], 'config': {'user': 'bjork', 'port': '3333'}}, + {'host': ['*'], 'config': {'crazy': 'something dumb '}}, + {'host': ['spoo.example.com'], 'config': {'crazy': 'something else'}}]) def test_3_host_config(self): global test_config_file - f = cStringIO.StringIO(test_config_file) + f = StringIO(test_config_file) config = paramiko.util.parse_ssh_config(f) + for host, values in { - 'irc.danger.com': {'user': 'robey', 'crazy': 'something dumb '}, - 'irc.example.com': {'user': 'bjork', 'crazy': 'something dumb ', 'port': '3333'}, - 'spoo.example.com': {'user': 'bjork', 'crazy': 'something else', 'port': '3333'} + 'irc.danger.com': {'crazy': 'something dumb ', + 'hostname': 'irc.danger.com', + 'user': 'robey'}, + 'irc.example.com': {'crazy': 'something dumb ', + 'hostname': 'irc.example.com', + 'user': 'robey', + 'port': '3333'}, + 'spoo.example.com': {'crazy': 'something dumb ', + 'hostname': 'spoo.example.com', + 'user': 'robey', + 'port': '3333'} }.items(): values = dict(values, hostname=host, - identityfile=os.path.expanduser("~/.ssh/id_rsa") + identityfile=[os.path.expanduser("~/.ssh/id_rsa")] ) - self.assertEquals( + self.assertEqual( paramiko.util.lookup_ssh_host_config(host, config), values ) def test_4_generate_key_bytes(self): - x = paramiko.util.generate_key_bytes(SHA, 'ABCDEFGH', 'This is my secret passphrase.', 64) - hex = ''.join(['%02x' % ord(c) for c in x]) - self.assertEquals(hex, '9110e2f6793b69363e58173e9436b13a5a4b339005741d5c680e505f57d871347b4239f14fb5c46e857d5e100424873ba849ac699cea98d729e57b3e84378e8b') + x = paramiko.util.generate_key_bytes(sha1, b'ABCDEFGH', 'This is my secret passphrase.', 64) + hex = ''.join(['%02x' % byte_ord(c) for c in x]) + self.assertEqual(hex, '9110e2f6793b69363e58173e9436b13a5a4b339005741d5c680e505f57d871347b4239f14fb5c46e857d5e100424873ba849ac699cea98d729e57b3e84378e8b') def test_5_host_keys(self): - f = open('hostfile.temp', 'w') - f.write(test_hosts_file) - f.close() + with open('hostfile.temp', 'w') as f: + f.write(test_hosts_file) try: hostdict = paramiko.util.load_host_keys('hostfile.temp') - self.assertEquals(2, len(hostdict)) - self.assertEquals(1, len(hostdict.values()[0])) - self.assertEquals(1, len(hostdict.values()[1])) + self.assertEqual(2, len(hostdict)) + self.assertEqual(1, len(list(hostdict.values())[0])) + self.assertEqual(1, len(list(hostdict.values())[1])) fp = hexlify(hostdict['secure.example.com']['ssh-rsa'].get_fingerprint()).upper() - self.assertEquals('E6684DB30E109B67B70FF1DC5C7F1363', fp) + self.assertEqual(b'E6684DB30E109B67B70FF1DC5C7F1363', fp) finally: os.unlink('hostfile.temp') - def test_6_random(self): - from paramiko.common import rng - # just verify that we can pull out 32 bytes and not get an exception. - x = rng.read(32) - self.assertEquals(len(x), 32) - - def test_7_host_config_expose_ssh_issue_33(self): + def test_7_host_config_expose_issue_33(self): test_config_file = """ Host www13.* Port 22 @@ -163,16 +165,16 @@ Host *.example.com Host * Port 3333 """ - f = cStringIO.StringIO(test_config_file) + f = StringIO(test_config_file) config = paramiko.util.parse_ssh_config(f) host = 'www13.example.com' - self.assertEquals( + self.assertEqual( paramiko.util.lookup_ssh_host_config(host, config), {'hostname': host, 'port': '22'} ) def test_8_eintr_retry(self): - self.assertEquals('foo', paramiko.util.retry_on_signal(lambda: 'foo')) + self.assertEqual('foo', paramiko.util.retry_on_signal(lambda: 'foo')) # Variables that are set by raises_intr intr_errors_remaining = [3] @@ -183,8 +185,8 @@ Host * intr_errors_remaining[0] -= 1 raise IOError(errno.EINTR, 'file', 'interrupted system call') self.assertTrue(paramiko.util.retry_on_signal(raises_intr) is None) - self.assertEquals(0, intr_errors_remaining[0]) - self.assertEquals(4, call_count[0]) + self.assertEqual(0, intr_errors_remaining[0]) + self.assertEqual(4, call_count[0]) def raises_ioerror_not_eintr(): raise IOError(errno.ENOENT, 'file', 'file not found') @@ -207,10 +209,10 @@ Host space-delimited Host equals-delimited ProxyCommand=foo bar=biz baz """ - f = cStringIO.StringIO(conf) + f = StringIO(conf) config = paramiko.util.parse_ssh_config(f) for host in ('space-delimited', 'equals-delimited'): - self.assertEquals( + self.assertEqual( host_config(host, config)['proxycommand'], 'foo bar=biz baz' ) @@ -219,24 +221,121 @@ Host equals-delimited """ ProxyCommand should perform interpolation on the value """ - config = paramiko.util.parse_ssh_config(cStringIO.StringIO(""" -Host * - Port 25 - ProxyCommand host %h port %p - + config = paramiko.util.parse_ssh_config(StringIO(""" Host specific Port 37 ProxyCommand host %h port %p lol Host portonly Port 155 + +Host * + Port 25 + ProxyCommand host %h port %p """)) for host, val in ( ('foo.com', "host foo.com port 25"), ('specific', "host specific port 37 lol"), ('portonly', "host portonly port 155"), ): - self.assertEquals( + self.assertEqual( host_config(host, config)['proxycommand'], val ) + + def test_11_host_config_test_negation(self): + test_config_file = """ +Host www13.* !*.example.com + Port 22 + +Host *.example.com !www13.* + Port 2222 + +Host www13.* + Port 8080 + +Host * + Port 3333 + """ + f = StringIO(test_config_file) + config = paramiko.util.parse_ssh_config(f) + host = 'www13.example.com' + self.assertEqual( + paramiko.util.lookup_ssh_host_config(host, config), + {'hostname': host, 'port': '8080'} + ) + + def test_12_host_config_test_proxycommand(self): + test_config_file = """ +Host proxy-with-equal-divisor-and-space +ProxyCommand = foo=bar + +Host proxy-with-equal-divisor-and-no-space +ProxyCommand=foo=bar + +Host proxy-without-equal-divisor +ProxyCommand foo=bar:%h-%p + """ + for host, values in { + 'proxy-with-equal-divisor-and-space' :{'hostname': 'proxy-with-equal-divisor-and-space', + 'proxycommand': 'foo=bar'}, + 'proxy-with-equal-divisor-and-no-space':{'hostname': 'proxy-with-equal-divisor-and-no-space', + 'proxycommand': 'foo=bar'}, + 'proxy-without-equal-divisor' :{'hostname': 'proxy-without-equal-divisor', + 'proxycommand': + 'foo=bar:proxy-without-equal-divisor-22'} + }.items(): + + f = StringIO(test_config_file) + config = paramiko.util.parse_ssh_config(f) + self.assertEqual( + paramiko.util.lookup_ssh_host_config(host, config), + values + ) + + def test_11_host_config_test_identityfile(self): + test_config_file = """ + +IdentityFile id_dsa0 + +Host * +IdentityFile id_dsa1 + +Host dsa2 +IdentityFile id_dsa2 + +Host dsa2* +IdentityFile id_dsa22 + """ + for host, values in { + 'foo' :{'hostname': 'foo', + 'identityfile': ['id_dsa0', 'id_dsa1']}, + 'dsa2' :{'hostname': 'dsa2', + 'identityfile': ['id_dsa0', 'id_dsa1', 'id_dsa2', 'id_dsa22']}, + 'dsa22' :{'hostname': 'dsa22', + 'identityfile': ['id_dsa0', 'id_dsa1', 'id_dsa22']} + }.items(): + + f = StringIO(test_config_file) + config = paramiko.util.parse_ssh_config(f) + self.assertEqual( + paramiko.util.lookup_ssh_host_config(host, config), + values + ) + + def test_12_config_addressfamily_and_lazy_fqdn(self): + """ + Ensure the code path honoring non-'all' AddressFamily doesn't asplode + """ + test_config = """ +AddressFamily inet +IdentityFile something_%l_using_fqdn +""" + config = paramiko.util.parse_ssh_config(StringIO(test_config)) + assert config.lookup('meh') # will die during lookup() if bug regresses + + def test_13_config_dos_crlf_succeeds(self): + config_file = StringIO("host abcqwerty\r\nHostName 127.0.0.1\r\n") + config = paramiko.SSHConfig() + config.parse(config_file) + self.assertEqual(config.lookup("abcqwerty")["hostname"], "127.0.0.1") diff --git a/tests/util.py b/tests/util.py index 2e0be087..66d2696c 100644 --- a/tests/util.py +++ b/tests/util.py @@ -1,5 +1,8 @@ +import os import unittest +root_path = os.path.dirname(os.path.realpath(__file__)) + class ParamikoTest(unittest.TestCase): # for Python 2.3 and below @@ -8,3 +11,7 @@ class ParamikoTest(unittest.TestCase): if not hasattr(unittest.TestCase, 'assertFalse'): assertFalse = unittest.TestCase.failIf + +def test_path(filename): + return os.path.join(root_path, filename) + |