Imported Upstream version 1.5.2upstream/1.5.2

14 files changed, 2397 insertions, 0 deletions

diff --git a/tests/loop.py b/tests/loop.py
new file mode 100644
index 0000000..ad5f7ca
--- /dev/null
+++ b/tests/loop.py
@@ -0,0 +1,104 @@
+#!/usr/bin/python
+
+# Copyright (C) 2003-2005 Robey Pointer <robey@lag.net>
+#
+# This file is part of paramiko.
+#
+# Paramiko is free software; you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# 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
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+# A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.
+
+"""
+...
+"""
+
+import threading, socket
+
+
+class LoopSocket (object):
+    """
+    A LoopSocket looks like a normal socket, but all data written to it is
+    delivered on the read-end of another LoopSocket, and vice versa.  It's
+    like a software "socketpair".
+    """
+    
+    def __init__(self):
+        self.__in_buffer = ''
+        self.__lock = threading.Lock()
+        self.__cv = threading.Condition(self.__lock)
+        self.__timeout = None
+        self.__mate = None
+
+    def close(self):
+        self.__unlink()
+        try:
+            self.__lock.acquire()
+            self.__in_buffer = ''
+        finally:
+            self.__lock.release()
+
+    def send(self, data):
+        if self.__mate is None:
+            # EOF
+            raise EOFError()
+        self.__mate.__feed(data)
+        return len(data)
+
+    def recv(self, n):
+        self.__lock.acquire()
+        try:
+            if self.__mate is None:
+                # EOF
+                return ''
+            if len(self.__in_buffer) == 0:
+                self.__cv.wait(self.__timeout)
+            if len(self.__in_buffer) == 0:
+                raise socket.timeout
+            if n < self.__in_buffer:
+                out = self.__in_buffer[:n]
+                self.__in_buffer = self.__in_buffer[n:]
+            else:
+                out = self.__in_buffer
+                self.__in_buffer = ''
+            return out
+        finally:
+            self.__lock.release()
+
+    def settimeout(self, n):
+        self.__timeout = n
+
+    def link(self, other):
+        self.__mate = other
+        self.__mate.__mate = self
+
+    def __feed(self, data):
+        self.__lock.acquire()
+        try:
+            self.__in_buffer += data
+            self.__cv.notifyAll()
+        finally:
+            self.__lock.release()
+            
+    def __unlink(self):
+        m = None
+        self.__lock.acquire()
+        try:
+            if self.__mate is not None:
+                m = self.__mate
+                self.__mate = None
+        finally:
+            self.__lock.release()
+        if m is not None:
+            m.__unlink()
+
+
diff --git a/tests/stub_sftp.py b/tests/stub_sftp.py
new file mode 100644
index 0000000..4b8b9c3
--- /dev/null
+++ b/tests/stub_sftp.py
@@ -0,0 +1,192 @@
+#!/usr/bin/python
+
+# Copyright (C) 2003-2005 Robey Pointer <robey@lag.net>
+#
+# This file is part of paramiko.
+#
+# Paramiko is free software; you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# 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
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+# A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.
+
+"""
+A stub SFTP server for loopback SFTP testing.
+"""
+
+import os
+from paramiko import ServerInterface, SFTPServerInterface, SFTPServer, SFTPAttributes, \
+    SFTPHandle, SFTP_OK, AUTH_SUCCESSFUL, OPEN_SUCCEEDED
+
+
+class StubServer (ServerInterface):
+    def check_auth_password(self, username, password):
+        # all are allowed
+        return AUTH_SUCCESSFUL
+
+    def check_channel_request(self, kind, chanid):
+        return OPEN_SUCCEEDED
+
+
+class StubSFTPHandle (SFTPHandle):
+    def stat(self):
+        try:
+            return SFTPAttributes.from_stat(os.fstat(self.readfile.fileno()))
+        except OSError, e:
+            return SFTPServer.convert_errno(e.errno)
+
+    def chattr(self, attr):
+        # python doesn't have equivalents to fchown or fchmod, so we have to
+        # use the stored filename
+        try:
+            SFTPServer.set_file_attr(self.filename, attr)
+        except OSError, e:
+            return SFTPServer.convert_errno(e.errno)
+
+
+class StubSFTPServer (SFTPServerInterface):
+    # assume current folder is a fine root
+    # (the tests always create and eventualy delete a subfolder, so there shouldn't be any mess)
+    ROOT = os.getcwd()
+        
+    def _realpath(self, path):
+        return self.ROOT + self.canonicalize(path)
+
+    def list_folder(self, path):
+        path = self._realpath(path)
+        try:
+            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:
+            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:
+            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:
+            return SFTPServer.convert_errno(e.errno)
+
+    def open(self, path, flags, attr):
+        path = self._realpath(path)
+        try:
+            fd = os.open(path, flags)
+        except OSError, e:
+            return SFTPServer.convert_errno(e.errno)
+        if (flags & os.O_CREAT) and (attr is not None):
+            SFTPServer.set_file_attr(path, attr)
+        if flags & os.O_WRONLY:
+            fstr = 'w'
+        elif flags & os.O_RDWR:
+            fstr = 'r+'
+        else:
+            # O_RDONLY (== 0)
+            fstr = 'r'
+        try:
+            f = os.fdopen(fd, fstr)
+        except OSError, e:
+            return SFTPServer.convert_errno(e.errno)
+        fobj = StubSFTPHandle()
+        fobj.filename = path
+        fobj.readfile = f
+        fobj.writefile = f
+        return fobj
+
+    def remove(self, path):
+        path = self._realpath(path)
+        try:
+            os.remove(path)
+        except OSError, e:
+            return SFTPServer.convert_errno(e.errno)
+        return SFTP_OK
+
+    def rename(self, oldpath, newpath):
+        oldpath = self._realpath(oldpath)
+        newpath = self._realpath(newpath)
+        try:
+            os.rename(oldpath, newpath)
+        except OSError, e:
+            return SFTPServer.convert_errno(e.errno)
+        return SFTP_OK
+
+    def mkdir(self, path, attr):
+        path = self._realpath(path)
+        try:
+            os.mkdir(path)
+            if attr is not None:
+                SFTPServer.set_file_attr(path, attr)
+        except OSError, e:
+            return SFTPServer.convert_errno(e.errno)
+        return SFTP_OK
+
+    def rmdir(self, path):
+        path = self._realpath(path)
+        try:
+            os.rmdir(path)
+        except OSError, e:
+            return SFTPServer.convert_errno(e.errno)
+        return SFTP_OK
+
+    def chattr(self, path, attr):
+        path = self._realpath(path)
+        try:
+            SFTPServer.set_file_attr(path, attr)
+        except OSError, e:
+            return SFTPServer.convert_errno(e.errno)
+        return SFTP_OK
+
+    def symlink(self, target_path, path):
+        path = self._realpath(path)
+        if (len(target_path) > 0) and (target_path[0] == '/'):
+            # absolute symlink
+            target_path = os.path.join(self.ROOT, target_path[1:])
+            if target_path[:2] == '//':
+                # bug in os.path.join
+                target_path = target_path[1:]
+        else:
+            # compute relative to path
+            abspath = os.path.join(os.path.dirname(path), target_path)
+            if abspath[:len(self.ROOT)] != self.ROOT:
+                # this symlink isn't going to work anyway -- just break it immediately
+                target_path = '<error>'
+        try:
+            os.symlink(target_path, path)
+        except:
+            return SFTPServer.convert_errno(e.errno)
+        return SFTP_OK
+
+    def readlink(self, path):
+        path = self._realpath(path)
+        try:
+            symlink = os.readlink(path)
+        except OSError, e:
+            return SFTPServer.convert_errno(e.errno)
+        # if it's absolute, remove the root
+        if os.path.isabs(symlink):
+            if symlink[:len(self.ROOT)] == self.ROOT:
+                symlink = symlink[len(self.ROOT):]
+                if (len(symlink) == 0) or (symlink[0] != '/'):
+                    symlink = '/' + symlink
+            else:
+                symlink = '<error>'
+        return symlink
diff --git a/tests/test_dss.key b/tests/test_dss.key
new file mode 100644
index 0000000..e10807f
--- /dev/null
+++ b/tests/test_dss.key
@@ -0,0 +1,12 @@
+-----BEGIN DSA PRIVATE KEY-----
+MIIBuwIBAAKBgQDngaYDZ30c6/7cJgEEbtl8FgKdwhba1Z7oOrOn4MI/6C42G1bY
+wMuqZf4dBCglsdq39SHrcjbE8Vq54gPSOh3g4+uV9Rcg5IOoPLbwp2jQfF6f1FIb
+sx7hrDCIqUcQccPSxetPBKmXI9RN8rZLaFuQeTnI65BKM98Ruwvq6SI2LwIVAPDP
+hSeawaJI27mKqOfe5PPBSmyHAoGBAJMXxXmPD9sGaQ419DIpmZecJKBUAy9uXD8x
+gbgeDpwfDaFJP8owByCKREocPFfi86LjCuQkyUKOfjYMN6iHIf1oEZjB8uJAatUr
+FzI0ArXtUqOhwTLwTyFuUojE5own2WYsOAGByvgfyWjsGhvckYNhI4ODpNdPlxQ8
+ZamaPGPsAoGARmR7CCPjodxASvRbIyzaVpZoJ/Z6x7dAumV+ysrV1BVYd0lYukmn
+jO1kKBWApqpH1ve9XDQYN8zgxM4b16L21kpoWQnZtXrY3GZ4/it9kUgyB7+NwacI
+BlXa8cMDL7Q/69o0d54U0X/NeX5QxuYR6OMJlrkQB7oiW/P/1mwjQgECFGI9QPSc
+h9pT9XHqn+1rZ4bK+QGA
+-----END DSA PRIVATE KEY-----
diff --git a/tests/test_dss_password.key b/tests/test_dss_password.key
new file mode 100644
index 0000000..e2a9bc5
--- /dev/null
+++ b/tests/test_dss_password.key
@@ -0,0 +1,15 @@
+-----BEGIN DSA PRIVATE KEY-----
+Proc-Type: 4,ENCRYPTED
+DEK-Info: DES-EDE3-CBC,78DAEB836ED0A646
+
+ldWkq9OMlXqWmjIqppNnmNPIUj5uVT12LkBosTApTbibTme3kIJb1uDeG2BShVfY
++vDOTUE9koGPDLsxW1t5At+EVyIDK8aIO0uHteXM5AbBX20LLUWRbRVqZhsMxqQh
+3H3XlHiN+QhaWcb4fFuu18a8SkimTFpDnZuffoCDl/zh/B7XieARTLA805K/ZgVB
+BBwflkR2BE053XHrJAIx9BEUlLP76Fo18rvjLZOSeu3s+VnnhqUb5FCt5h50a46u
+YXQBbo2r9Zo1ilGMNEXJO0gk5hwGVmTySz53NkPA5HmWt8NIzv5jQHMDy7N+ZykF
+uwpP1R5M/ZIFY4Y5h/lvn6IJjQ7VySRPIbpN8o2YJv2OD1Ja80n3tU8Mg77o3o4d
+NwKm7cCjlq+FuIBdOsSgsB8FPQRUhW+jpFDxmWN64DM2cEg6RUdptby7WmMp0HwK
+1qyEfxHjLMuDVlD7lASIDBrRlUjPtXEH1DzIYQuYaRZaixFoZ7EY+X73TwmrKFEU
+US9ZnQZtRtroRqGwR4fz4wQQsjTl/AmOijlBmi29taJccJsT/THrLQ5plOEd8OMv
+9FsaPJXBU85gaRKo3JZtrw==
+-----END DSA PRIVATE KEY-----
diff --git a/tests/test_file.py b/tests/test_file.py
new file mode 100644
index 0000000..250821c
--- /dev/null
+++ b/tests/test_file.py
@@ -0,0 +1,153 @@
+#!/usr/bin/python
+
+# Copyright (C) 2003-2005 Robey Pointer <robey@lag.net>
+#
+# This file is part of paramiko.
+#
+# Paramiko is free software; you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# 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
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+# A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.
+
+"""
+Some unit tests for the BufferedFile abstraction.
+"""
+
+import unittest
+from paramiko.file import BufferedFile
+
+
+class LoopbackFile (BufferedFile):
+    """
+    BufferedFile object that you can write data into, and then read it back.
+    """
+    def __init__(self, mode='r', bufsize=-1):
+        BufferedFile.__init__(self)
+        self._set_mode(mode, bufsize)
+        self.buffer = ''
+
+    def _read(self, size):
+        if len(self.buffer) == 0:
+            return None
+        if size > len(self.buffer):
+            size = len(self.buffer)
+        data = self.buffer[:size]
+        self.buffer = self.buffer[size:]
+        return data
+
+    def _write(self, data):
+        self.buffer += data
+        return len(data)
+
+
+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')
+        except:
+            pass
+        f.close()
+
+        f = LoopbackFile('w')
+        try:
+            f.read(1)
+            self.assert_(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.')
+        self.assertEqual(f.readline(), 'First line.\n')
+        # universal newline mode should convert this linefeed:
+        self.assertEqual(f.readline(), 'Second line.\n')
+        # truncated line:
+        self.assertEqual(f.readline(7), 'Third l')
+        self.assertEqual(f.readline(), 'ine.\n')
+        self.assertEqual(f.readline(), 'Final line non-terminated.')
+        self.assertEqual(f.readline(), '')
+        f.close()
+        try:
+            f.readline()
+            self.assert_(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)
+
+    def test_3_lf(self):
+        """
+        try to trick the linefeed detector.
+        """
+        f = LoopbackFile('r+U')
+        f.write('First line.\r')
+        self.assertEqual(f.readline(), 'First line.\n')
+        f.write('\nSecond.\r\n')
+        self.assertEqual(f.readline(), 'Second.\n')
+        f.close()
+        self.assertEqual(f.newlines, '\r\n')
+
+    def test_4_write(self):
+        """
+        verify that write buffering is on.
+        """
+        f = LoopbackFile('r+', 1)
+        f.write('Complete line.\nIncomplete line.')
+        self.assertEqual(f.readline(), 'Complete line.\n')
+        self.assertEqual(f.readline(), '')
+        f.write('..\n')
+        self.assertEqual(f.readline(), 'Incomplete line...\n')
+        f.close()
+
+    def test_5_flush(self):
+        """
+        verify that flush will force a write.
+        """
+        f = LoopbackFile('r+', 512)
+        f.write('Not\nquite\n512 bytes.\n')
+        self.assertEqual(f.read(1), '')
+        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), '')
+        f.close()
+
+    def test_6_buffering(self):
+        """
+        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.close()
+
+    def test_7_read_all(self):
+        """
+        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')
+        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')
+        f.close()
diff --git a/tests/test_kex.py b/tests/test_kex.py
new file mode 100644
index 0000000..2680853
--- /dev/null
+++ b/tests/test_kex.py
@@ -0,0 +1,183 @@
+#!/usr/bin/python
+
+# Copyright (C) 2003-2005 Robey Pointer <robey@lag.net>
+#
+# This file is part of paramiko.
+#
+# Paramiko is free software; you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# 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
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+# A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.
+
+"""
+Some unit tests for the key exchange protocols.
+"""
+
+import unittest
+import paramiko.util
+from paramiko.kex_group1 import KexGroup1
+from paramiko.kex_gex import KexGex
+from paramiko import Message
+
+
+class FakeRandpool (object):
+    def stir(self):
+        pass
+    def get_bytes(self, n):
+        return chr(0xcc) * n
+
+class FakeKey (object):
+    def __str__(self):
+        return 'fake-key'
+    def sign_ssh_data(self, randpool, H):
+        return 'fake-sig'
+
+class FakeModulusPack (object):
+    P = 0xFFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7EDEE386BFB5A899FA5AE9F24117C4B1FE649286651ECE65381FFFFFFFFFFFFFFFFL
+    G = 2
+    def get_modulus(self, min, ask, max):
+        return self.G, self.P
+
+class FakeTransport (object):
+    randpool = FakeRandpool()
+    local_version = 'SSH-2.0-paramiko_1.0'
+    remote_version = 'SSH-2.0-lame'
+    local_kex_init = 'local-kex-init'
+    remote_kex_init = 'remote-kex-init'
+
+    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
+
+    def setUp(self):
+        pass
+
+    def tearDown(self):
+        pass
+
+    def test_1_group1_client(self):
+        transport = FakeTransport()
+        transport.server_mode = False
+        kex = KexGroup1(transport)
+        kex.start_kex()
+        x = '1E000000807E2DDB1743F3487D6545F04F1C8476092FB912B013626AB5BCEB764257D88BBA64243B9F348DF7B41B8C814A995E00299913503456983FFB9178D3CD79EB6D55522418A8ABF65375872E55938AB99A84A0B5FC8A1ECC66A7C3766E7E0F80B7CE2C9225FC2DD683F4764244B72963BBB383F529DCF0C5D17740B8A2ADBE9208D4'
+        self.assertEquals(x, paramiko.util.hexify(str(transport._message)))
+        self.assertEquals(paramiko.kex_group1._MSG_KEXDH_REPLY, transport._expect)
+
+        # fake "reply"
+        msg = Message()
+        msg.add_string('fake-host-key')
+        msg.add_mpint(69)
+        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, paramiko.util.hexify(transport._H))
+        self.assertEquals(('fake-host-key', 'fake-sig'), transport._verify)
+        self.assert_(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)
+
+        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, paramiko.util.hexify(transport._H))
+        self.assertEquals(x, paramiko.util.hexify(str(transport._message)))
+        self.assert_(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, paramiko.util.hexify(str(transport._message)))
+        self.assertEquals(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, paramiko.util.hexify(str(transport._message)))
+        self.assertEquals(paramiko.kex_gex._MSG_KEXDH_GEX_REPLY, transport._expect)
+
+        msg = Message()
+        msg.add_string('fake-host-key')
+        msg.add_mpint(69)
+        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, paramiko.util.hexify(transport._H))
+        self.assertEquals(('fake-host-key', 'fake-sig'), transport._verify)
+        self.assert_(transport._activated)
+
+    def test_4_gex_server(self):
+        transport = FakeTransport()
+        transport.server_mode = True
+        kex = KexGex(transport)
+        kex.start_kex()
+        self.assertEquals(paramiko.kex_gex._MSG_KEXDH_GEX_REQUEST, transport._expect)
+
+        msg = Message()
+        msg.add_int(1024)
+        msg.add_int(2048)
+        msg.add_int(4096)
+        msg.rewind()
+        kex.parse_next(paramiko.kex_gex._MSG_KEXDH_GEX_REQUEST, msg)
+        x = '1F0000008100FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7EDEE386BFB5A899FA5AE9F24117C4B1FE649286651ECE65381FFFFFFFFFFFFFFFF0000000102'
+        self.assertEquals(x, paramiko.util.hexify(str(transport._message)))
+        self.assertEquals(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, paramiko.util.hexify(transport._H))
+        self.assertEquals(x, paramiko.util.hexify(str(transport._message)))
+        self.assert_(transport._activated)
diff --git a/tests/test_message.py b/tests/test_message.py
new file mode 100644
index 0000000..441e3ce
--- /dev/null
+++ b/tests/test_message.py
@@ -0,0 +1,102 @@
+#!/usr/bin/python
+
+# Copyright (C) 2003-2005 Robey Pointer <robey@lag.net>
+#
+# This file is part of paramiko.
+#
+# Paramiko is free software; you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# 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
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+# A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.
+
+"""
+Some unit tests for ssh protocol message blocks.
+"""
+
+import unittest
+from paramiko.message import Message
+
+
+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'
+
+    def test_1_encode(self):
+        msg = Message()
+        msg.add_int(23)
+        msg.add_int(123789456)
+        msg.add_string('q')
+        msg.add_string('hello')
+        msg.add_string('x' * 1000)
+        self.assertEquals(str(msg), self.__a)
+
+        msg = Message()
+        msg.add_boolean(True)
+        msg.add_boolean(False)
+        msg.add_byte('\xf3')
+        msg.add_bytes('\x00\x3f')
+        msg.add_list(['huey', 'dewey', 'louie'])
+        self.assertEquals(str(msg), self.__b)
+
+        msg = Message()
+        msg.add_int64(5)
+        msg.add_int64(0xf5e4d3c2b109L)
+        msg.add_mpint(17)
+        msg.add_mpint(0xf5e4d3c2b109L)
+        msg.add_mpint(-0x65e4d3c2b109L)
+        self.assertEquals(str(msg), 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)
+
+        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'])
+
+        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)
+
+    def test_3_add(self):
+        msg = Message()
+        msg.add(5)
+        msg.add(0x1122334455L)
+        msg.add(True)
+        msg.add('cat')
+        msg.add(['a', 'b'])
+        self.assertEquals(str(msg), 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:])
+        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:])
+
diff --git a/tests/test_packetizer.py b/tests/test_packetizer.py
new file mode 100644
index 0000000..8c992bd
--- /dev/null
+++ b/tests/test_packetizer.py
@@ -0,0 +1,70 @@
+# Copyright (C) 2003-2005 Robey Pointer <robey@lag.net>
+#
+# This file is part of paramiko.
+#
+# Paramiko is free software; you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# 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
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+# A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.
+
+"""
+Some unit tests for the ssh2 protocol in Transport.
+"""
+
+import unittest
+from loop import LoopSocket
+from Crypto.Cipher import AES
+from Crypto.Hash import SHA, HMAC
+from paramiko import Message, Packetizer, util
+
+class PacketizerTest (unittest.TestCase):
+
+    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)
+
+        # 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_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):
+        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')
+        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())
diff --git a/tests/test_pkey.py b/tests/test_pkey.py
new file mode 100644
index 0000000..e56edb1
--- /dev/null
+++ b/tests/test_pkey.py
@@ -0,0 +1,140 @@
+# Copyright (C) 2003-2005 Robey Pointer <robey@lag.net>
+#
+# This file is part of paramiko.
+#
+# Paramiko is free software; you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# 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
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+# A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.
+
+"""
+Some unit tests for public/private key objects.
+"""
+
+import unittest
+from paramiko import RSAKey, DSSKey, Message, util, randpool
+
+# 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='
+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'
+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'
+
+
+class KeyTest (unittest.TestCase):
+
+    def setUp(self):
+        pass
+
+    def tearDown(self):
+        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 = util.unhexify('61E1F272F4C1C4561586BD322498C0E924672780F47BB37DDA7D54019E64')
+        self.assertEquals(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(':', '')
+        my_rsa = util.hexify(key.get_fingerprint()).lower()
+        self.assertEquals(exp_rsa, my_rsa)
+        self.assertEquals(PUB_RSA.split()[1], key.get_base64())
+        self.assertEquals(1024, key.get_bits())
+
+    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(':', '')
+        my_rsa = util.hexify(key.get_fingerprint()).lower()
+        self.assertEquals(exp_rsa, my_rsa)
+        self.assertEquals(PUB_RSA.split()[1], key.get_base64())
+        self.assertEquals(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(':', '')
+        my_dss = util.hexify(key.get_fingerprint()).lower()
+        self.assertEquals(exp_dss, my_dss)
+        self.assertEquals(PUB_DSS.split()[1], key.get_base64())
+        self.assertEquals(1024, key.get_bits())
+
+    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(':', '')
+        my_dss = util.hexify(key.get_fingerprint()).lower()
+        self.assertEquals(exp_dss, my_dss)
+        self.assertEquals(PUB_DSS.split()[1], key.get_base64())
+        self.assertEquals(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)
+
+    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)
+
+    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(randpool, 'ice weasels')
+        self.assert_(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())
+        msg.rewind()
+        pub = RSAKey(data=str(key))
+        self.assert_(pub.verify_ssh_sig('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(randpool, 'ice weasels')
+        self.assert_(type(msg) is Message)
+        msg.rewind()
+        self.assertEquals('ssh-dss', msg.get_string())
+        # 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()))
+        msg.rewind()
+        pub = DSSKey(data=str(key))
+        self.assert_(pub.verify_ssh_sig('ice weasels', msg))
+    
+    def test_A_generate_rsa(self):
+        key = RSAKey.generate(1024)
+        msg = key.sign_ssh_data(randpool, 'jerri blank')
+        msg.rewind()
+        self.assert_(key.verify_ssh_sig('jerri blank', msg))
+
+    def test_B_generate_dss(self):
+        key = DSSKey.generate(1024)
+        msg = key.sign_ssh_data(randpool, 'jerri blank')
+        msg.rewind()
+        self.assert_(key.verify_ssh_sig('jerri blank', msg))
diff --git a/tests/test_rsa.key b/tests/test_rsa.key
new file mode 100644
index 0000000..f50e9c5
--- /dev/null
+++ b/tests/test_rsa.key
@@ -0,0 +1,15 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIICWgIBAAKBgQDTj1bqB4WmayWNPB+8jVSYpZYk80Ujvj680pOTh2bORBjbIAyz
+oWGW+GUjzKxTiiPvVmxFgx5wdsFvF03v34lEVVhMpouqPAYQ15N37K/ir5XY+9m/
+d8ufMCkjeXsQkKqFbAlQcnWMCRnOoPHS3I4vi6hmnDDeeYTSRvfLbW0fhwIBIwKB
+gBIiOqZYaoqbeD9OS9z2K9KR2atlTxGxOJPXiP4ESqP3NVScWNwyZ3NXHpyrJLa0
+EbVtzsQhLn6rF+TzXnOlcipFvjsem3iYzCpuChfGQ6SovTcOjHV9z+hnpXvQ/fon
+soVRZY65wKnF7IAoUwTmJS9opqgrN6kRgCd3DASAMd1bAkEA96SBVWFt/fJBNJ9H
+tYnBKZGw0VeHOYmVYbvMSstssn8un+pQpUm9vlG/bp7Oxd/m+b9KWEh2xPfv6zqU
+avNwHwJBANqzGZa/EpzF4J8pGti7oIAPUIDGMtfIcmqNXVMckrmzQ2vTfqtkEZsA
+4rE1IERRyiJQx6EJsz21wJmGV9WJQ5kCQQDwkS0uXqVdFzgHO6S++tjmjYcxwr3g
+H0CoFYSgbddOT6miqRskOQF3DZVkJT3kyuBgU2zKygz52ukQZMqxCb1fAkASvuTv
+qfpH87Qq5kQhNKdbbwbmd2NxlNabazPijWuphGTdW0VfJdWfklyS2Kr+iqrs/5wV
+HhathJt636Eg7oIjAkA8ht3MQ+XSl9yIJIS8gVpbPxSw5OMfw0PjVE7tBdQruiSc
+nvuQES5C9BMHjF39LZiGH1iLQy7FgdHyoP+eodI7
+-----END RSA PRIVATE KEY-----
diff --git a/tests/test_rsa_password.key b/tests/test_rsa_password.key
new file mode 100644
index 0000000..7713049
--- /dev/null
+++ b/tests/test_rsa_password.key
@@ -0,0 +1,18 @@
+-----BEGIN RSA PRIVATE KEY-----
+Proc-Type: 4,ENCRYPTED
+DEK-Info: DES-EDE3-CBC,DAA422E8A5A8EFB7
+
++nssHGmWl91IcmGiE6DdCIqGvAP04tuLh60wLjWBvdjtF9CjztPnF57xe+6pBk7o
+YgF/Ry3ik9ZV9rHNcRXifDKM9crxtYlpUlkM2C0SP89sXaO0P1Q1yCnrtZUwDIKO
+BNV8et5X7+AGMFsy/nmv0NFMrbpoG03Dppsloecd29NTRlIXwxHRFyHxy6BdEib/
+Dn0mEVbwg3dTvKrd/sODWR9hRwpDGM9nkEbUNJCh7vMwFKkIZZF8yqFvmGckuO5C
+HZkDJ6RkEDYrSZJAavQaiOPF5bu3cHughRfnrIKVrQuTTDiWjwX9Ny8e4p4k7dy7
+rLpbPhtxUOUbpOF7T1QxljDi1Tcq3Ebk3kN/ZLPRFnDrJfyUx+m9BXmAa78Wxs/l
+KaS8DTkYykd3+EGOeJFjZg2bvgqil4V+5JIt/+MQ5pZ/ui7i4GcH2bvZyGAbrXzP
+3LipSAdN5RG+fViLe3HUtfCx4ZAgtU78TWJrLk2FwKQGglFxKLnswp+IKZb09rZV
+uxmG4pPLUnH+mMYdiy5ugzj+5C8iZ0/IstpHVmO6GWROfedpJ82eMztTOtdhfMep
+8Z3HwAwkDtksL7Gq9klb0Wq5+uRlBWetixddAvnmqXNzYhaANWcAF/2a2Hz06Rb0
+e6pe/g0Ek5KV+6YI+D+oEblG0Sr+d4NtxtDTmIJKNVkmzlhI2s53bHp6txCb5JWJ
+S8mKLPBBBzaNXYd3odDvGXguuxUntWSsD11KyR6B9DXMIfWQW5dT7hp5kTMGlXWJ
+lD2hYab13DCCuAkwVTdpzhHYLZyxLYoSu05W6z8SAOs=
+-----END RSA PRIVATE KEY-----
diff --git a/tests/test_sftp.py b/tests/test_sftp.py
new file mode 100644
index 0000000..993899a
--- /dev/null
+++ b/tests/test_sftp.py
@@ -0,0 +1,740 @@
+# Copyright (C) 2003-2005 Robey Pointer <robey@lag.net>
+#
+# This file is part of paramiko.
+#
+# Paramiko is free software; you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# 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
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+# A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.
+
+"""
+some unit tests to make sure sftp works.
+
+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 sys
+import threading
+import time
+import unittest
+
+import paramiko
+from stub_sftp import StubServer, StubSFTPServer
+from loop import LoopSocket
+
+ARTICLE = '''
+Insulin sensitivity and liver insulin receptor structure in ducks from two
+genera
+
+T. Constans, B. Chevalier, M. Derouet and J. Simon
+Station de Recherches Avicoles, Institut National de la Recherche Agronomique,
+Nouzilly, France.
+
+Insulin sensitivity and liver insulin receptor structure were studied in
+5-wk-old ducks from two genera (Muscovy and Pekin). In the fasting state, both
+duck types were equally resistant to exogenous insulin compared with chicken.
+Despite the low potency of duck insulin, the number of insulin receptors was
+lower in Muscovy duck and similar in Pekin duck and chicken liver membranes.
+After 125I-insulin cross-linking, the size of the alpha-subunit of the
+receptors from the three species was 135,000. Wheat germ agglutinin-purified
+receptors from the three species were contaminated by an active and unusual
+adenosinetriphosphatase (ATPase) contaminant (highest activity in Muscovy
+duck). Sequential purification of solubilized receptor from both duck types on
+lentil and then wheat germ agglutinin lectins led to a fraction of receptors
+very poor in ATPase activity that exhibited a beta-subunit size (95,000) and
+tyrosine kinase activity similar to those of ATPase-free chicken insulin
+receptors. Therefore the ducks from the two genera exhibit an alpha-beta-
+structure for liver insulin receptors and a clear difference in the number of
+liver insulin receptors. Their sensitivity to insulin is, however, similarly
+decreased compared with chicken.
+'''
+
+FOLDER = os.environ.get('TEST_FOLDER', 'temp-testing000')
+
+sftp = None
+tc = None
+g_big_file_test = True
+
+
+class SFTPTest (unittest.TestCase):
+
+    def init(hostname, username, keyfile, passwd):
+        global sftp, tc
+        
+        t = paramiko.Transport(hostname)
+        tc = t
+        try:
+            key = paramiko.RSAKey.from_private_key_file(keyfile, passwd)
+        except paramiko.PasswordRequiredException:
+            sys.stderr.write('\n\nparamiko.RSAKey.from_private_key_file REQUIRES PASSWORD.\n')
+            sys.stderr.write('You have two options:\n')
+            sys.stderr.write('* Use the "-K" option to point to a different (non-password-protected)\n')
+            sys.stderr.write('  private key file.\n')
+            sys.stderr.write('* Use the "-P" option to provide the password needed to unlock this private\n')
+            sys.stderr.write('  key.\n')
+            sys.stderr.write('\n')
+            sys.exit(1)
+        try:
+            t.connect(username=username, pkey=key)
+        except paramiko.SSHException:
+            t.close()
+            sys.stderr.write('\n\nparamiko.Transport.connect FAILED.\n')
+            sys.stderr.write('There are several possible reasons why it might fail so quickly:\n\n')
+            sys.stderr.write('* The host to connect to (%s) is not a valid SSH server.\n' % hostname)
+            sys.stderr.write('  (Use the "-H" option to change the host.)\n')
+            sys.stderr.write('* The username to auth as (%s) is invalid.\n' % username)
+            sys.stderr.write('  (Use the "-U" option to change the username.)\n')
+            sys.stderr.write('* The private key given (%s) is not accepted by the server.\n' % keyfile)
+            sys.stderr.write('  (Use the "-K" option to provide a different key file.)\n')
+            sys.stderr.write('\n')
+            sys.exit(1)
+        sftp = paramiko.SFTP.from_transport(t)
+    init = staticmethod(init)
+
+    def init_loopback():
+        global sftp, tc
+
+        socks = LoopSocket()
+        sockc = LoopSocket()
+        sockc.link(socks)
+        tc = paramiko.Transport(sockc)
+        ts = paramiko.Transport(socks)
+
+        host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key')
+        ts.add_server_key(host_key)
+        event = threading.Event()
+        server = StubServer()
+        ts.set_subsystem_handler('sftp', paramiko.SFTPServer, StubSFTPServer)
+        ts.start_server(event, server)
+        tc.connect(username='slowdive', password='pygmalion')
+        event.wait(1.0)
+
+        sftp = paramiko.SFTP.from_transport(tc)
+    init_loopback = staticmethod(init_loopback)
+
+    def set_big_file_test(onoff):
+        global g_big_file_test
+        g_big_file_test = onoff
+    set_big_file_test = staticmethod(set_big_file_test)
+        
+    def setUp(self):
+        global FOLDER
+        for i in xrange(1000):
+            FOLDER = FOLDER[:-3] + '%03d' % i
+            try:
+                sftp.mkdir(FOLDER)
+                break
+            except (IOError, OSError):
+                pass
+
+    def tearDown(self):
+        sftp.rmdir(FOLDER)
+
+    def test_1_file(self):
+        """
+        verify that we can create a file.
+        """
+        f = sftp.open(FOLDER + '/test', 'w')
+        try:
+            self.assertEqual(f.stat().st_size, 0)
+            f.close()
+        finally:
+            sftp.remove(FOLDER + '/test')
+
+    def test_2_close(self):
+        """
+        verify that closing the sftp session doesn't do anything bad, and that
+        a new one can be opened.
+        """
+        global sftp
+        sftp.close()
+        try:
+            sftp.open(FOLDER + '/test2', 'w')
+            self.fail('expected exception')
+        except:
+            pass
+        sftp = paramiko.SFTP.from_transport(tc)
+
+    def test_3_write(self):
+        """
+        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()
+            self.assertEqual(sftp.stat(FOLDER + '/duck.txt').st_size, 1483)
+        finally:
+            sftp.remove(FOLDER + '/duck.txt')
+
+    def test_4_append(self):
+        """
+        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()
+        finally:
+            sftp.remove(FOLDER + '/append.txt')
+        
+    def test_5_rename(self):
+        """
+        verify that renaming a file works.
+        """
+        f = sftp.open(FOLDER + '/first.txt', 'w')
+        try:
+            f.write('content!\n');
+            f.close()
+            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')
+            except IOError:
+                pass
+            f = sftp.open(FOLDER + '/second.txt', 'r')
+            f.seek(-6, f.SEEK_END)
+            self.assertEqual(f.read(4), 'tent')
+            f.close()
+        finally:
+            try:
+                sftp.remove(FOLDER + '/first.txt')
+            except:
+                pass
+            try:
+                sftp.remove(FOLDER + '/second.txt')
+            except:
+                pass
+
+    def test_6_folder(self):
+        """
+        create a temporary folder, verify that we can create a file in it, then
+        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.remove(FOLDER + '/subfolder/test')
+        sftp.rmdir(FOLDER + '/subfolder')
+        try:
+            f = sftp.open(FOLDER + '/subfolder/test')
+            # shouldn't be able to create that file
+            self.assert_(False, 'no exception at dummy file creation')
+        except IOError:
+            pass
+
+    def test_7_listdir(self):
+        """
+        verify that a folder can be created, a bunch of files can be placed in it,
+        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()
+
+            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)
+        finally:
+            sftp.remove(FOLDER + '/duck.txt')
+            sftp.remove(FOLDER + '/fish.txt')
+            sftp.remove(FOLDER + '/tertiary.py')
+
+    def test_8_setstat(self):
+        """
+        verify that the setstat functions (chown, chmod, utime) work.
+        """
+        f = sftp.open(FOLDER + '/special', 'w')
+        try:
+            f.close()
+
+            stat = sftp.stat(FOLDER + '/special')
+            sftp.chmod(FOLDER + '/special', (stat.st_mode & ~0777) | 0600)
+            self.assertEqual(sftp.stat(FOLDER + '/special').st_mode & 0777, 0600)
+
+            mtime = stat.st_mtime - 3600
+            atime = stat.st_atime - 1800
+            sftp.utime(FOLDER + '/special', (atime, mtime))
+            nstat = sftp.stat(FOLDER + '/special')
+            self.assertEqual(nstat.st_mtime, mtime)
+            self.assertEqual(nstat.st_atime, atime)
+
+            # can't really test chown, since we'd have to know a valid uid.
+        finally:
+            sftp.remove(FOLDER + '/special')
+
+    def test_9_readline_seek(self):
+        """
+        create a text file and write a bunch of text into it.  then count the lines
+        in the file, and seek around to retreive particular lines.  this should
+        verify that read buffering and 'tell' work well together, and that read
+        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()
+        finally:
+            sftp.remove(FOLDER + '/duck.txt')
+
+    def test_A_write_seek(self):
+        """
+        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()
+
+            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')
+        finally:
+            sftp.remove(FOLDER + '/testing.txt')
+
+    def test_B_symlink(self):
+        """
+        create a symlink and then check that lstat doesn't follow it.
+        """
+        f = sftp.open(FOLDER + '/original.txt', 'w')
+        try:
+            f.write('original\n')
+            f.close()
+            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()
+
+            cwd = sftp.normalize('.')
+            if cwd[-1] == '/':
+                cwd = cwd[:-1]
+            abs_path = cwd + '/' + FOLDER + '/original.txt'
+            sftp.symlink(abs_path, FOLDER + '/link2.txt')
+            self.assertEqual(abs_path, sftp.readlink(FOLDER + '/link2.txt'))
+
+            self.assertEqual(sftp.lstat(FOLDER + '/link.txt').st_size, 12)
+            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.assertEqual(sftp.stat(FOLDER + '/link2.txt').st_size, 9)
+            self.assertEqual(sftp.stat(FOLDER + '/original.txt').st_size, 9)
+        finally:
+            try:
+                sftp.remove(FOLDER + '/link.txt')
+            except:
+                pass
+            try:
+                sftp.remove(FOLDER + '/link2.txt')
+            except:
+                pass
+            try:
+                sftp.remove(FOLDER + '/original.txt')
+            except:
+                pass
+
+    def test_C_flush_seek(self):
+        """
+        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()
+        finally:
+            try:
+                sftp.remove(FOLDER + '/happy.txt')
+            except:
+                pass
+
+    def test_D_lots_of_files(self):
+        """
+        create a bunch of files over the same session.
+        """
+        global g_big_file_test
+        if not g_big_file_test:
+            return
+        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)
+
+            # now make sure every file is there, by creating a list of filenmes
+            # and reading them in random order.
+            numlist = 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()
+                numlist.remove(r)
+        finally:
+            for i in range(numfiles):
+                try:
+                    sftp.remove('%s/file%d.txt' % (FOLDER, i))
+                except:
+                    pass
+
+    def test_E_big_file(self):
+        """
+        write a 1MB file with no buffering.
+        """
+        global g_big_file_test
+        if not g_big_file_test:
+            return
+        kblob = (1024 * '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()
+            sys.stderr.write(' ')
+
+            self.assertEqual(sftp.stat('%s/hongry.txt' % FOLDER).st_size, 1024 * 1024)
+            end = time.time()
+            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()
+
+            end = time.time()
+            sys.stderr.write('%ds ' % round(end - start))
+        finally:
+            sftp.remove('%s/hongry.txt' % FOLDER)
+
+    def test_F_big_file_pipelined(self):
+        """
+        write a 1MB file, with no linefeeds, using pipelining.
+        """
+        global g_big_file_test
+        if not g_big_file_test:
+            return
+        kblob = (1024 * 'x')
+        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()
+            sys.stderr.write(' ')
+
+            self.assertEqual(sftp.stat('%s/hongry.txt' % FOLDER).st_size, 1024 * 1024)
+            end = time.time()
+            sys.stderr.write('%ds ' % round(end - start))
+            
+            start = time.time()
+            f = sftp.open('%s/hongry.txt' % FOLDER, 'r')
+            f.prefetch()
+            for n in range(1024):
+                data = f.read(1024)
+                self.assertEqual(data, kblob)
+            f.close()
+
+            end = time.time()
+            sys.stderr.write('%ds ' % round(end - start))
+        finally:
+            sftp.remove('%s/hongry.txt' % FOLDER)
+
+    def test_G_lots_of_prefetching(self):
+        """
+        prefetch a 1MB file a bunch of times, discarding the file object
+        without using it, to verify that paramiko doesn't get confused.
+        """
+        global g_big_file_test
+        if not g_big_file_test:
+            return
+        kblob = (1024 * '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()
+            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')
+                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()
+            sys.stderr.write(' ')
+        finally:
+            sftp.remove('%s/hongry.txt' % FOLDER)
+        
+    def test_H_big_file_big_buffer(self):
+        """
+        write a 1MB file, with no linefeeds, and a big buffer.
+        """
+        global g_big_file_test
+        if not g_big_file_test:
+            return
+        mblob = (1024 * 1024 * 'x')
+        try:
+            f = sftp.open('%s/hongry.txt' % FOLDER, 'w', 128 * 1024)
+            f.write(mblob)
+            f.close()
+
+            self.assertEqual(sftp.stat('%s/hongry.txt' % FOLDER).st_size, 1024 * 1024)
+        finally:
+            sftp.remove('%s/hongry.txt' % FOLDER)
+    
+    def test_I_big_file_renegotiate(self):
+        """
+        write a 1MB file, forcing key renegotiation in the middle.
+        """
+        global g_big_file_test
+        if not g_big_file_test:
+            return
+        t = sftp.sock.get_transport()
+        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()
+            
+            self.assertEqual(sftp.stat('%s/hongry.txt' % FOLDER).st_size, 1024 * 1024)
+            self.assertNotEquals(t.H, t.session_id)
+        finally:
+            sftp.remove('%s/hongry.txt' % FOLDER)
+            t.packetizer.REKEY_BYTES = pow(2, 30)
+
+    def test_J_realpath(self):
+        """
+        test that realpath is returning something non-empty and not an
+        error.
+        """
+        pwd = sftp.normalize('.')
+        self.assert_(len(pwd) > 0)
+        f = sftp.normalize('./' + FOLDER)
+        self.assert_(len(f) > 0)
+        self.assertEquals(os.path.join(pwd, FOLDER), f)
+
+    def test_K_mkdir(self):
+        """
+        verify that mkdir/rmdir work.
+        """
+        try:
+            sftp.mkdir(FOLDER + '/subfolder')
+        except:
+            self.assert_(False, 'exception creating subfolder')
+        try:
+            sftp.mkdir(FOLDER + '/subfolder')
+            self.assert_(False, 'no exception overwriting subfolder')
+        except IOError:
+            pass
+        try:
+            sftp.rmdir(FOLDER + '/subfolder')
+        except:
+            self.assert_(False, 'exception removing subfolder')
+        try:
+            sftp.rmdir(FOLDER + '/subfolder')
+            self.assert_(False, 'no exception removing nonexistent subfolder')
+        except IOError:
+            pass
+    
+    def test_L_chdir(self):
+        """
+        verify that chdir/getcwd work.
+        """
+        root = sftp.normalize('.')
+        if root[-1] != '/':
+            root += '/'
+        try:
+            sftp.mkdir(FOLDER + '/alpha')
+            sftp.chdir(FOLDER + '/alpha')
+            sftp.mkdir('beta')
+            self.assertEquals(root + FOLDER + '/alpha', sftp.getcwd())
+            self.assertEquals(['beta'], sftp.listdir('.'))
+        
+            sftp.chdir('beta')
+            f = sftp.open('fish', 'w')
+            f.write('hello\n')
+            f.close()
+            sftp.chdir('..')
+            self.assertEquals(['fish'], sftp.listdir('beta'))
+            sftp.chdir('..')
+            self.assertEquals(['fish'], sftp.listdir('alpha/beta'))
+        finally:
+            sftp.chdir(root)
+            try:
+                sftp.unlink(FOLDER + '/alpha/beta/fish')
+            except:
+                pass
+            try:
+                sftp.rmdir(FOLDER + '/alpha/beta')
+            except:
+                pass
+            try:
+                sftp.rmdir(FOLDER + '/alpha')
+            except:
+                pass
+
+    def test_M_get_put(self):
+        """
+        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, 'w')
+        f.write(text)
+        f.close()
+        sftp.put(localname, FOLDER + '/bunny.txt')
+        
+        f = sftp.open(FOLDER + '/bunny.txt', 'r')
+        self.assertEquals(text, f.read(128))
+        f.close()
+        
+        os.unlink(localname)
+        localname = os.tempnam()
+        sftp.get(FOLDER + '/bunny.txt', localname)
+        
+        f = open(localname, 'r')
+        self.assertEquals(text, f.read(128))
+        f.close()
+    
+        os.unlink(localname)
+        sftp.unlink(FOLDER + '/bunny.txt')
+
+    def test_N_check(self):
+        """
+        verify that file.check() works against our own server.
+        (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()
+        
+        try:
+            f = sftp.open(FOLDER + '/kitty.txt', 'r')
+            sum = f.check('sha1')
+            self.assertEquals('91059CFC6615941378D413CB5ADAF4C5EB293402', paramiko.util.hexify(sum))
+            sum = f.check('md5', 0, 512)
+            self.assertEquals('93DE4788FCA28D471516963A1FE3856A', paramiko.util.hexify(sum))
+            sum = f.check('md5', 0, 0, 510)
+            self.assertEquals('EB3B45B8CD55A0707D99B177544A319F373183D241432BB2157AB9E46358C4AC90370B5CADE5D90336FC1716F90B36D6',
+                              paramiko.util.hexify(sum))
+        finally:
+            sftp.unlink(FOLDER + '/kitty.txt')
+
+    def test_O_x_flag(self):
+        """
+        verify that the 'x' flag works when opening a file.
+        """
+        f = sftp.open(FOLDER + '/unusual.txt', 'wx')
+        f.close()
+        
+        try:
+            try:
+                f = sftp.open(FOLDER + '/unusual.txt', 'wx')
+                self.fail('expected exception')
+            except IOError, x:
+                pass
+        finally:
+            sftp.unlink(FOLDER + '/unusual.txt')
+    
+    def test_P_utf8(self):
+        """
+        verify that unicode strings are encoded into utf8 correctly.
+        """
+        f = sftp.open(FOLDER + '/something', 'w')
+        f.write('okay')
+        f.close()
+        
+        try:
+            sftp.rename(FOLDER + '/something', FOLDER + u'/\u00fcnic\u00f8de')
+            sftp.open(FOLDER + '/\xc3\xbcnic\xc3\xb8\x64\x65', 'r')
+        except Exception, e:
+            self.fail('exception ' + e)
+        sftp.unlink(FOLDER + '/\xc3\xbcnic\xc3\xb8\x64\x65')
+
diff --git a/tests/test_transport.py b/tests/test_transport.py
new file mode 100644
index 0000000..5fcc786
--- /dev/null
+++ b/tests/test_transport.py
@@ -0,0 +1,573 @@
+# Copyright (C) 2003-2005 Robey Pointer <robey@lag.net>
+#
+# This file is part of paramiko.
+#
+# Paramiko is free software; you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# 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
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+# A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.
+
+"""
+Some unit tests for the ssh2 protocol in Transport.
+"""
+
+import sys, time, threading, unittest
+import select
+from paramiko import Transport, SecurityOptions, ServerInterface, RSAKey, DSSKey, \
+    SSHException, BadAuthenticationType, InteractiveQuery, util
+from paramiko import AUTH_FAILED, AUTH_PARTIALLY_SUCCESSFUL, AUTH_SUCCESSFUL
+from paramiko import OPEN_SUCCEEDED
+from loop import LoopSocket
+
+
+class NullServer (ServerInterface):
+    paranoid_did_password = False
+    paranoid_did_public_key = False
+    paranoid_key = DSSKey.from_private_key_file('tests/test_dss.key')
+    
+    def get_allowed_auths(self, username):
+        if username == 'slowdive':
+            return 'publickey,password'
+        if username == 'paranoid':
+            if not self.paranoid_did_password and not self.paranoid_did_public_key:
+                return 'publickey,password'
+            elif self.paranoid_did_password:
+                return 'publickey'
+            else:
+                return 'password'
+        if username == 'commie':
+            return 'keyboard-interactive'
+        return 'publickey'
+
+    def check_auth_password(self, username, password):
+        if (username == 'slowdive') and (password == 'pygmalion'):
+            return AUTH_SUCCESSFUL
+        if (username == 'paranoid') and (password == 'paranoid'):
+            # 2-part auth (even openssh doesn't support this)
+            self.paranoid_did_password = True
+            if self.paranoid_did_public_key:
+                return AUTH_SUCCESSFUL
+            return AUTH_PARTIALLY_SUCCESSFUL
+        return AUTH_FAILED
+
+    def check_auth_publickey(self, username, key):
+        if (username == 'paranoid') and (key == self.paranoid_key):
+            # 2-part auth
+            self.paranoid_did_public_key = True
+            if self.paranoid_did_password:
+                return AUTH_SUCCESSFUL
+            return AUTH_PARTIALLY_SUCCESSFUL
+        return AUTH_FAILED
+    
+    def check_auth_interactive(self, username, submethods):
+        if username == 'commie':
+            self.username = username
+            return InteractiveQuery('password', 'Please enter a password.', ('Password', False))
+        return AUTH_FAILED
+    
+    def check_auth_interactive_response(self, responses):
+        if self.username == 'commie':
+            if (len(responses) == 1) and (responses[0] == 'cat'):
+                return AUTH_SUCCESSFUL
+        return AUTH_FAILED
+
+    def check_channel_request(self, kind, chanid):
+        return OPEN_SUCCEEDED
+
+    def check_channel_exec_request(self, channel, command):
+        if command != 'yes':
+            return False
+        return True
+
+    def check_channel_shell_request(self, channel):
+        return True
+    
+    def check_global_request(self, kind, msg):
+        self._global_request = kind
+        return False
+
+
+class TransportTest (unittest.TestCase):
+
+    def setUp(self):
+        self.socks = LoopSocket()
+        self.sockc = LoopSocket()
+        self.sockc.link(self.socks)
+        self.tc = Transport(self.sockc)
+        self.ts = Transport(self.socks)
+
+    def tearDown(self):
+        self.tc.close()
+        self.ts.close()
+        self.socks.close()
+        self.sockc.close()
+
+    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)
+        o.ciphers = ('aes256-cbc', 'blowfish-cbc')
+        self.assertEquals(('aes256-cbc', 'blowfish-cbc'), o.ciphers)
+        try:
+            o.ciphers = ('aes256-cbc', 'made-up-cipher')
+            self.assert_(False)
+        except ValueError:
+            pass
+        try:
+            o.ciphers = 23
+            self.assert_(False)
+        except TypeError:
+            pass
+            
+    def test_2_compute_key(self):
+        self.tc.K = 123281095979686581523377256114209720774539068973101330872763622971399429481072519713536292772709507296759612401802191955568143056534122385270077606457721553469730659233569339356140085284052436697480759510519672848743794433460113118986816826624865291116513647975790797391795651716378444844877749505443714557929L
+        self.tc.H = util.unhexify('0C8307CDE6856FF30BA93684EB0F04C2520E9ED3')
+        self.tc.session_id = self.tc.H
+        key = self.tc._compute_key('C', 32)
+        self.assertEquals('207E66594CA87C44ECCBA3B3CD39FDDB378E6FDB0F97C54B2AA0CFBF900CD995',
+            util.hexify(key))
+
+    def test_3_simple(self):
+        """
+        verify that we can establish an ssh link with ourselves across the
+        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))
+        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.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())
+
+    def test_4_special(self):
+        """
+        verify that the client can demand odd handshake settings, and can
+        renegotiate keys in mid-stream.
+        """
+        host_key = RSAKey.from_private_key_file('tests/test_rsa.key')
+        public_host_key = RSAKey(data=str(host_key))
+        self.ts.add_server_key(host_key)
+        event = threading.Event()
+        server = NullServer()
+        self.assert_(not event.isSet())
+        self.ts.start_server(event, server)
+        options = self.tc.get_security_options()
+        options.ciphers = ('aes256-cbc',)
+        options.digests = ('hmac-md5-96',)
+        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('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.tc.send_ignore(1024)
+        self.assert_(self.tc.renegotiate_keys())
+        self.ts.send_ignore(1024)
+
+    def test_5_keepalive(self):
+        """
+        verify that the keepalive will be sent.
+        """
+        self.tc.set_hexdump(True)
+        
+        host_key = RSAKey.from_private_key_file('tests/test_rsa.key')
+        public_host_key = RSAKey(data=str(host_key))
+        self.ts.add_server_key(host_key)
+        event = threading.Event()
+        server = NullServer()
+        self.assert_(not event.isSet())
+        self.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(None, getattr(server, '_global_request', None))
+        self.tc.set_keepalive(1)
+        time.sleep(2)
+        self.assertEquals('keepalive@lag.net', server._global_request)
+        
+    def test_6_bad_auth_type(self):
+        """
+        verify that we get the right exception when an unsupported auth
+        type is requested.
+        """
+        host_key = RSAKey.from_private_key_file('tests/test_rsa.key')
+        public_host_key = RSAKey(data=str(host_key))
+        self.ts.add_server_key(host_key)
+        event = threading.Event()
+        server = NullServer()
+        self.assert_(not event.isSet())
+        self.ts.start_server(event, server)
+        try:
+            self.tc.connect(hostkey=public_host_key,
+                            username='unknown', password='error')
+            self.assert_(False)
+        except:
+            etype, evalue, etb = sys.exc_info()
+            self.assertEquals(BadAuthenticationType, etype)
+            self.assertEquals(['publickey'], evalue.allowed_types)
+
+    def test_7_bad_password(self):
+        """
+        verify that a bad password gets the right exception, and that a retry
+        with the right password works.
+        """
+        host_key = RSAKey.from_private_key_file('tests/test_rsa.key')
+        public_host_key = RSAKey(data=str(host_key))
+        self.ts.add_server_key(host_key)
+        event = threading.Event()
+        server = NullServer()
+        self.assert_(not event.isSet())
+        self.ts.start_server(event, server)
+        self.tc.ultra_debug = True
+        self.tc.connect(hostkey=public_host_key)
+        try:
+            self.tc.auth_password(username='slowdive', password='error')
+            self.assert_(False)
+        except:
+            etype, evalue, etb = sys.exc_info()
+            self.assertEquals(SSHException, etype)
+        self.tc.auth_password(username='slowdive', password='pygmalion')
+        event.wait(1.0)
+        self.assert_(event.isSet())
+        self.assert_(self.ts.is_active())
+    
+    def test_8_multipart_auth(self):
+        """
+        verify that multipart auth works.
+        """
+        host_key = RSAKey.from_private_key_file('tests/test_rsa.key')
+        public_host_key = RSAKey(data=str(host_key))
+        self.ts.add_server_key(host_key)
+        event = threading.Event()
+        server = NullServer()
+        self.assert_(not event.isSet())
+        self.ts.start_server(event, server)
+        self.tc.ultra_debug = True
+        self.tc.connect(hostkey=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')
+        remain = self.tc.auth_publickey(username='paranoid', key=key)
+        self.assertEquals([], remain)
+        event.wait(1.0)
+        self.assert_(event.isSet())
+        self.assert_(self.ts.is_active())
+
+    def test_9_interactive_auth(self):
+        """
+        verify keyboard-interactive auth works.
+        """
+        host_key = RSAKey.from_private_key_file('tests/test_rsa.key')
+        public_host_key = RSAKey(data=str(host_key))
+        self.ts.add_server_key(host_key)
+        event = threading.Event()
+        server = NullServer()
+        self.assert_(not event.isSet())
+        self.ts.start_server(event, server)
+        self.tc.ultra_debug = True
+        self.tc.connect(hostkey=public_host_key)
+
+        def handler(title, instructions, prompts):
+            self.got_title = title
+            self.got_instructions = instructions
+            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)
+        event.wait(1.0)
+        self.assert_(event.isSet())
+        self.assert_(self.ts.is_active())
+        
+    def test_A_interactive_auth_fallback(self):
+        """
+        verify that a password auth attempt will fallback to "interactive"
+        if password auth isn't supported but interactive is.
+        """
+        host_key = RSAKey.from_private_key_file('tests/test_rsa.key')
+        public_host_key = RSAKey(data=str(host_key))
+        self.ts.add_server_key(host_key)
+        event = threading.Event()
+        server = NullServer()
+        self.assert_(not event.isSet())
+        self.ts.start_server(event, server)
+        self.tc.ultra_debug = True
+        self.tc.connect(hostkey=public_host_key)
+        remain = self.tc.auth_password('commie', 'cat')
+        self.assertEquals([], remain)
+        event.wait(1.0)
+        self.assert_(event.isSet())
+        self.assert_(self.ts.is_active())
+            
+    def test_B_exec_command(self):
+        """
+        verify that exec_command() does something reasonable.
+        """
+        host_key = RSAKey.from_private_key_file('tests/test_rsa.key')
+        public_host_key = RSAKey(data=str(host_key))
+        self.ts.add_server_key(host_key)
+        event = threading.Event()
+        server = NullServer()
+        self.assert_(not event.isSet())
+        self.ts.start_server(event, server)
+        self.tc.ultra_debug = True
+        self.tc.connect(hostkey=public_host_key)
+        self.tc.auth_password(username='slowdive', password='pygmalion')
+        event.wait(1.0)
+        self.assert_(event.isSet())
+        self.assert_(self.ts.is_active())
+
+        chan = self.tc.open_session()
+        schan = self.ts.accept(1.0)
+        self.assert_(not chan.exec_command('no'))
+        
+        chan = self.tc.open_session()
+        self.assert_(chan.exec_command('yes'))
+        schan = self.ts.accept(1.0)
+        schan.send('Hello there.\n')
+        schan.send_stderr('This is on stderr.\n')
+        schan.close()
+
+        f = chan.makefile()
+        self.assertEquals('Hello there.\n', f.readline())
+        self.assertEquals('', f.readline())
+        f = chan.makefile_stderr()
+        self.assertEquals('This is on stderr.\n', f.readline())
+        self.assertEquals('', f.readline())
+        
+        # now try it with combined stdout/stderr
+        chan = self.tc.open_session()
+        self.assert_(chan.exec_command('yes'))
+        schan = self.ts.accept(1.0)
+        schan.send('Hello there.\n')
+        schan.send_stderr('This is on stderr.\n')
+        schan.close()
+
+        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())
+
+    def test_C_invoke_shell(self):
+        """
+        verify that invoke_shell() does something reasonable.
+        """
+        host_key = RSAKey.from_private_key_file('tests/test_rsa.key')
+        public_host_key = RSAKey(data=str(host_key))
+        self.ts.add_server_key(host_key)
+        event = threading.Event()
+        server = NullServer()
+        self.assert_(not event.isSet())
+        self.ts.start_server(event, server)
+        self.tc.ultra_debug = True
+        self.tc.connect(hostkey=public_host_key)
+        self.tc.auth_password(username='slowdive', password='pygmalion')
+        event.wait(1.0)
+        self.assert_(event.isSet())
+        self.assert_(self.ts.is_active())
+
+        chan = self.tc.open_session()
+        self.assert_(chan.invoke_shell())
+        schan = self.ts.accept(1.0)
+        chan.send('communist j. cat\n')
+        f = schan.makefile()
+        self.assertEquals('communist j. cat\n', f.readline())
+        chan.close()
+        self.assertEquals('', f.readline())
+
+    def test_D_exit_status(self):
+        """
+        verify that get_exit_status() works.
+        """
+        host_key = RSAKey.from_private_key_file('tests/test_rsa.key')
+        public_host_key = RSAKey(data=str(host_key))
+        self.ts.add_server_key(host_key)
+        event = threading.Event()
+        server = NullServer()
+        self.assert_(not event.isSet())
+        self.ts.start_server(event, server)
+        self.tc.ultra_debug = True
+        self.tc.connect(hostkey=public_host_key)
+        self.tc.auth_password(username='slowdive', password='pygmalion')
+        event.wait(1.0)
+        self.assert_(event.isSet())
+        self.assert_(self.ts.is_active())
+
+        chan = self.tc.open_session()
+        schan = self.ts.accept(1.0)
+        self.assert_(chan.exec_command('yes'))
+        schan.send('Hello there.\n')
+        # trigger an EOF
+        schan.shutdown_read()
+        schan.shutdown_write()
+        schan.send_exit_status(23)
+        schan.close()
+        
+        f = chan.makefile()
+        self.assertEquals('Hello there.\n', f.readline())
+        self.assertEquals('', f.readline())
+        self.assertEquals(23, chan.recv_exit_status())
+        chan.close()
+
+    def test_E_select(self):
+        """
+        verify that select() on a channel works.
+        """
+        host_key = RSAKey.from_private_key_file('tests/test_rsa.key')
+        public_host_key = RSAKey(data=str(host_key))
+        self.ts.add_server_key(host_key)
+        event = threading.Event()
+        server = NullServer()
+        self.assert_(not event.isSet())
+        self.ts.start_server(event, server)
+        self.tc.ultra_debug = True
+        self.tc.connect(hostkey=public_host_key)
+        self.tc.auth_password(username='slowdive', password='pygmalion')
+        event.wait(1.0)
+        self.assert_(event.isSet())
+        self.assert_(self.ts.is_active())
+
+        chan = self.tc.open_session()
+        self.assert_(chan.invoke_shell())
+        schan = self.ts.accept(1.0)
+
+        # nothing should be ready        
+        r, w, e = select.select([chan], [], [], 0.1)
+        self.assertEquals([], r)
+        self.assertEquals([], w)
+        self.assertEquals([], e)
+        
+        schan.send('hello\n')
+        
+        # something should be ready now (give it 1 second to appear)
+        for i in range(10):
+            r, w, e = select.select([chan], [], [], 0.1)
+            if chan in r:
+                break
+            time.sleep(0.1)
+        self.assertEquals([chan], r)
+        self.assertEquals([], w)
+        self.assertEquals([], e)
+
+        self.assertEquals('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)
+
+        schan.close()
+        
+        # detect eof?
+        for i in range(10):
+            r, w, e = select.select([chan], [], [], 0.1)
+            if chan in r:
+                break
+            time.sleep(0.1)
+        self.assertEquals([chan], r)
+        self.assertEquals([], w)
+        self.assertEquals([], e)
+        self.assertEquals('', chan.recv(16))
+        
+        chan.close()
+   
+    def test_F_renegotiate(self):
+        """
+        verify that a transport can correctly renegotiate mid-stream.
+        """
+        host_key = RSAKey.from_private_key_file('tests/test_rsa.key')
+        public_host_key = RSAKey(data=str(host_key))
+        self.ts.add_server_key(host_key)
+        event = threading.Event()
+        server = NullServer()
+        self.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.tc.packetizer.REKEY_BYTES = 16384
+        
+        chan = self.tc.open_session()
+        self.assert_(chan.exec_command('yes'))
+        schan = self.ts.accept(1.0)
+
+        self.assertEquals(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):
+            if self.tc.H != self.tc.session_id:
+                break
+            time.sleep(0.1)
+        self.assertNotEquals(self.tc.H, self.tc.session_id)
+
+        schan.close()
+
+    def test_G_compression(self):
+        """
+        verify that zlib compression is basically working.
+        """
+        host_key = RSAKey.from_private_key_file('tests/test_rsa.key')
+        public_host_key = RSAKey(data=str(host_key))
+        self.ts.add_server_key(host_key)
+        self.ts.get_security_options().compression = ('zlib',)
+        self.tc.get_security_options().compression = ('zlib',)
+        event = threading.Event()
+        server = NullServer()
+        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())
+
+        chan = self.tc.open_session()
+        self.assert_(chan.exec_command('yes'))
+        schan = self.ts.accept(1.0)
+
+        bytes = self.tc.packetizer._Packetizer__sent_bytes
+        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)
+
+        chan.close()
+        schan.close()
diff --git a/tests/test_util.py b/tests/test_util.py
new file mode 100644
index 0000000..fa8c029
--- /dev/null
+++ b/tests/test_util.py
@@ -0,0 +1,80 @@
+#!/usr/bin/python
+
+# Copyright (C) 2003-2005 Robey Pointer <robey@lag.net>
+#
+# This file is part of paramiko.
+#
+# Paramiko is free software; you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# 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
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+# A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.
+
+"""
+Some unit tests for utility functions.
+"""
+
+import cStringIO
+import unittest
+from Crypto.Hash import SHA
+import paramiko.util
+
+
+test_config_file = """\
+Host *
+    User robey
+    IdentityFile    =~/.ssh/id_rsa
+
+# comment
+Host *.example.com
+    \tUser bjork
+Port=3333
+Host *
+ \t  \t Crazy something dumb  
+Host spoo.example.com
+Crazy something else
+"""
+
+
+class UtilTest (unittest.TestCase):
+
+    K = 14730343317708716439807310032871972459448364195094179797249681733965528989482751523943515690110179031004049109375612685505881911274101441415545039654102474376472240501616988799699744135291070488314748284283496055223852115360852283821334858541043710301057312858051901453919067023103730011648890038847384890504L
+
+    def setUp(self):
+        pass
+
+    def tearDown(self):
+        pass
+
+    def test_1_parse_config(self):
+        global test_config_file
+        f = cStringIO.StringIO(test_config_file)
+        config = paramiko.util.parse_ssh_config(f)
+        self.assertEquals(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'}])
+
+    def test_2_host_config(self):
+        global test_config_file
+        f = cStringIO.StringIO(test_config_file)
+        config = paramiko.util.parse_ssh_config(f)
+        c = paramiko.util.lookup_ssh_host_config('irc.danger.com', config)
+        self.assertEquals(c, {'identityfile': '~/.ssh/id_rsa', 'user': 'robey', 'crazy': 'something dumb  '})
+        c = paramiko.util.lookup_ssh_host_config('irc.example.com', config)
+        self.assertEquals(c, {'identityfile': '~/.ssh/id_rsa', 'user': 'bjork', 'crazy': 'something dumb  ', 'port': '3333'})
+        c = paramiko.util.lookup_ssh_host_config('spoo.example.com', config)
+        self.assertEquals(c, {'identityfile': '~/.ssh/id_rsa', 'user': 'bjork', 'crazy': 'something else', 'port': '3333'})
+
+    def test_3_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')