[project @ Arch-1:robey@lag.net--2003-public%secsh--dev--1.0--patch-136]

loopback sftp test
add ability to turn off more tests, and a secret (for now) -X option to do
the sftp tests via loopback socket.  added another symlink sftp test to see
what happens with absolute symlinks.

2 files changed, 66 insertions, 4 deletions

diff --git a/test.py b/test.py
index b8b76d42..ad2f9f51 100755
--- a/test.py
+++ b/test.py
@@ -57,6 +57,11 @@ parser.add_option('-P', '--sftp-passwd', dest='password', type='string', default
                   help='(optional) password to unlock the private key for sftp tests')
 parser.add_option('--no-pkey', action='store_false', dest='use_pkey', default=True,
                   help='skip RSA/DSS private key tests (which can take a while)')
+parser.add_option('--no-transport', action='store_false', dest='use_transport', default=True,
+                  help='skip transport tests (which can take a while)')
+parser.add_option('--no-big-file', action='store_false', dest='use_big_file', default=True,
+                  help='skip big file SFTP tests, which are slow as molasses')
+parser.add_option('-X', action='store_true', dest='use_loopback_sftp', default=False)
 
 options, args = parser.parse_args()
 if len(args) > 0:
@@ -64,6 +69,10 @@ if len(args) > 0:
 
 if options.use_sftp:
     SFTPTest.init(options.hostname, options.username, options.keyfile, options.password)
+if options.use_loopback_sftp:
+    SFTPTest.init_loopback()
+if not options.use_big_file:
+    SFTPTest.set_big_file_test(False)
 
 # setup logging
 paramiko.util.log_to_file('test.log')
@@ -74,7 +83,8 @@ suite.addTest(unittest.makeSuite(BufferedFileTest))
 if options.use_pkey:
     suite.addTest(unittest.makeSuite(KeyTest))
 suite.addTest(unittest.makeSuite(KexTest))
-suite.addTest(unittest.makeSuite(TransportTest))
-if options.use_sftp:
+if options.use_transport:
+    suite.addTest(unittest.makeSuite(TransportTest))
+if options.use_sftp or options.use_loopback_sftp:
     suite.addTest(unittest.makeSuite(SFTPTest))
 unittest.TextTestRunner(verbosity=2).run(suite)
diff --git a/tests/test_sftp.py b/tests/test_sftp.py
index eebb2184..ef8adbc3 100755
--- a/tests/test_sftp.py
+++ b/tests/test_sftp.py
@@ -27,9 +27,11 @@ do test file operations in (so no existing files will be harmed).
 
 import sys, os
 import random
-import logging
+import logging, threading
 
 import paramiko, unittest
+from stub_sftp import StubServer, StubSFTPServer
+from loop import LoopSocket
 
 ARTICLE = '''
 Insulin sensitivity and liver insulin receptor structure in ducks from two
@@ -61,6 +63,7 @@ decreased compared with chicken.
 FOLDER = os.environ.get('TEST_FOLDER', 'temp-testing')
 
 sftp = None
+g_big_file_test = True
 
 
 class SFTPTest (unittest.TestCase):
@@ -97,6 +100,33 @@ class SFTPTest (unittest.TestCase):
         sftp = paramiko.SFTP.from_transport(t)
     init = staticmethod(init)
 
+    def init_loopback():
+        global sftp
+
+        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)
+#        self.assert_(self.ts.is_active())
+
+        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):
         sftp.mkdir(FOLDER)
 
@@ -294,8 +324,20 @@ class SFTPTest (unittest.TestCase):
             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:
@@ -303,6 +345,10 @@ class SFTPTest (unittest.TestCase):
             except:
                 pass
             try:
+                sftp.remove(FOLDER + '/link2.txt')
+            except:
+                pass
+            try:
                 sftp.remove(FOLDER + '/original.txt')
             except:
                 pass
@@ -333,6 +379,9 @@ class SFTPTest (unittest.TestCase):
         """
         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):
@@ -362,6 +411,9 @@ class SFTPTest (unittest.TestCase):
         write a 1MB file, with no linefeeds, using line buffering.
         FIXME: this is slow!  what causes the slowness?
         """
+        global g_big_file_test
+        if not g_big_file_test:
+            return
         kblob = (1024 * 'x')
         try:
             f = sftp.open('%s/hongry.txt' % FOLDER, 'w', 1)
@@ -385,7 +437,7 @@ class SFTPTest (unittest.TestCase):
         self.assert_(len(pwd) > 0)
         f = sftp.normalize('./' + FOLDER)
         self.assert_(len(f) > 0)
-        self.assert_(f == pwd + '/' + FOLDER)
+        self.assertEquals(os.path.join(pwd, FOLDER), f)
 
     def test_E_mkdir(self):
         """