diff options
author | Jeff Forcier <jeff@bitprophet.org> | 2017-10-23 16:05:30 -0700 |
---|---|---|
committer | Jeff Forcier <jeff@bitprophet.org> | 2018-09-17 14:44:07 -0700 |
commit | 8868305a76e84e2ee00e35b5131b7b98e45f92a1 (patch) | |
tree | 9caa303d099bef32f7a7e9dbdf6e4688f34e3ecb /tests/test_sftp.py | |
parent | 903f21d99d9e47dfc519c1521f8dbb8b4ff5724a (diff) |
Start overhauling regular SFTP suite.
Includes attempt to split out the longer-lived server component from the
client component, in fixtures
Also starts tweaking early tests so they're less bad. E.g. test_2_close
had its docstring actively disagreeing with its code (which was super
confusing since both were committed at the same time in 2005...)
Diffstat (limited to 'tests/test_sftp.py')
-rwxr-xr-x | tests/test_sftp.py | 479 |
1 files changed, 220 insertions, 259 deletions
diff --git a/tests/test_sftp.py b/tests/test_sftp.py index 9154a795..c1d4cfe0 100755 --- a/tests/test_sftp.py +++ b/tests/test_sftp.py @@ -84,143 +84,102 @@ decreased compared with chicken. # Thus, the following 2-bytes sequence is not valid utf8: "invalid continuation byte" NON_UTF8_DATA = b'\xC3\xC3' -FOLDER = os.environ.get('TEST_FOLDER', 'temp-testing000') - -sftp = None -tc = None -g_big_file_test = True unicode_folder = u'\u00fcnic\u00f8de' if PY2 else '\u00fcnic\u00f8de' utf8_folder = b'/\xc3\xbcnic\xc3\xb8\x64\x65' -class SFTPTest(unittest.TestCase): - @staticmethod - def set_big_file_test(onoff): - global g_big_file_test - g_big_file_test = onoff - - def setUp(self): - # TODO: I think this was created to allow working around litter from - # broken previous test runs, and/or to allow concurrent testing? Either - # way I hate it and it should die ASAP in favor of pytest fixtures or - # similar things that do try/finally. - # TODO: also, for god's sake remove the global in favor of self. - global FOLDER - for i in range(1000): - FOLDER = FOLDER[:-3] + '%03d' % i - try: - sftp.mkdir(FOLDER) - break - except (IOError, OSError): - pass - - def tearDown(self): - #sftp.chdir() - sftp.rmdir(FOLDER) - - def test_1_file(self): +class TestSFTP(object): + def test_1_file(self, sftp): """ verify that we can create a file. """ - f = sftp.open(FOLDER + '/test', 'w') + f = sftp.open(sftp.FOLDER + '/test', 'w') try: - self.assertEqual(f.stat().st_size, 0) + assert f.stat().st_size == 0 finally: f.close() - sftp.remove(FOLDER + '/test') + sftp.remove(sftp.FOLDER + '/test') - def test_2_close(self): + def test_2_close(self, sftp): """ - verify that closing the sftp session doesn't do anything bad, and that - a new one can be opened. + Verify that SFTP session close() causes a socket error on next action. """ - global sftp sftp.close() - try: - sftp.open(FOLDER + '/test2', 'w') - self.fail('expected exception') - except: - pass - sftp = paramiko.SFTP.from_transport(tc) + with pytest.raises(socket.error, match='Socket is closed'): + sftp.open(sftp.FOLDER + '/test2', 'w') - def test_2_sftp_can_be_used_as_context_manager(self): + def test_2_sftp_can_be_used_as_context_manager(self, sftp): """ verify that the sftp session is closed when exiting the context manager """ - global sftp with sftp: pass - try: - sftp.open(FOLDER + '/test2', 'w') - self.fail('expected exception') - except (EOFError, socket.error): - pass - finally: - sftp = paramiko.SFTP.from_transport(tc) + with pytest.raises(socket.error, match='Socket is closed'): + sftp.open(sftp.FOLDER + '/test2', 'w') - def test_3_write(self): + def test_3_write(self, sftp): """ verify that a file can be created and written, and the size is correct. """ try: - with sftp.open(FOLDER + '/duck.txt', 'w') as f: + with sftp.open(sftp.FOLDER + '/duck.txt', 'w') as f: f.write(ARTICLE) - self.assertEqual(sftp.stat(FOLDER + '/duck.txt').st_size, 1483) + assert sftp.stat(sftp.FOLDER + '/duck.txt').st_size == 1483 finally: - sftp.remove(FOLDER + '/duck.txt') + sftp.remove(sftp.FOLDER + '/duck.txt') - def test_3_sftp_file_can_be_used_as_context_manager(self): + def test_3_sftp_file_can_be_used_as_context_manager(self, sftp): """ verify that an opened file can be used as a context manager """ try: - with sftp.open(FOLDER + '/duck.txt', 'w') as f: + with sftp.open(sftp.FOLDER + '/duck.txt', 'w') as f: f.write(ARTICLE) - self.assertEqual(sftp.stat(FOLDER + '/duck.txt').st_size, 1483) + assert sftp.stat(sftp.FOLDER + '/duck.txt').st_size == 1483 finally: - sftp.remove(FOLDER + '/duck.txt') + sftp.remove(sftp.FOLDER + '/duck.txt') - def test_4_append(self): + def test_4_append(self, sftp): """ verify that a file can be opened for append, and tell() still works. """ try: - with sftp.open(FOLDER + '/append.txt', 'w') as f: + with sftp.open(sftp.FOLDER + '/append.txt', 'w') as f: f.write('first line\nsecond line\n') - self.assertEqual(f.tell(), 23) + assert f.tell() == 23 - with sftp.open(FOLDER + '/append.txt', 'a+') as f: + with sftp.open(sftp.FOLDER + '/append.txt', 'a+') as f: f.write('third line!!!\n') - self.assertEqual(f.tell(), 37) - self.assertEqual(f.stat().st_size, 37) + assert f.tell() == 37 + assert f.stat().st_size == 37 f.seek(-26, f.SEEK_CUR) - self.assertEqual(f.readline(), 'second line\n') + assert f.readline() == 'second line\n' finally: - sftp.remove(FOLDER + '/append.txt') + sftp.remove(sftp.FOLDER + '/append.txt') - def test_5_rename(self): + def test_5_rename(self, sftp): """ verify that renaming a file works. """ try: - with sftp.open(FOLDER + '/first.txt', 'w') as f: + with sftp.open(sftp.FOLDER + '/first.txt', 'w') as f: f.write('content!\n') - sftp.rename(FOLDER + '/first.txt', FOLDER + '/second.txt') + sftp.rename(sftp.FOLDER + '/first.txt', sftp.FOLDER + '/second.txt') try: - sftp.open(FOLDER + '/first.txt', 'r') + sftp.open(sftp.FOLDER + '/first.txt', 'r') self.assertTrue(False, 'no exception on reading nonexistent file') except IOError: pass - with sftp.open(FOLDER + '/second.txt', 'r') as f: + with sftp.open(sftp.FOLDER + '/second.txt', 'r') as f: f.seek(-6, f.SEEK_END) - self.assertEqual(u(f.read(4)), 'tent') + assert u(f.read(4)) == 'tent' finally: try: - sftp.remove(FOLDER + '/first.txt') + sftp.remove(sftp.FOLDER + '/first.txt') except: pass try: - sftp.remove(FOLDER + '/second.txt') + sftp.remove(sftp.FOLDER + '/second.txt') except: pass @@ -229,69 +188,69 @@ class SFTPTest(unittest.TestCase): 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') - sftp.open(FOLDER + '/subfolder/test', 'w').close() - sftp.remove(FOLDER + '/subfolder/test') - sftp.rmdir(FOLDER + '/subfolder') + sftp.mkdir(sftp.FOLDER + '/subfolder') + sftp.open(sftp.FOLDER + '/subfolder/test', 'w').close() + sftp.remove(sftp.FOLDER + '/subfolder/test') + sftp.rmdir(sftp.FOLDER + '/subfolder') try: - sftp.open(FOLDER + '/subfolder/test') + sftp.open(sftp.FOLDER + '/subfolder/test') # shouldn't be able to create that file self.assertTrue(False, 'no exception at dummy file creation') except IOError: pass - def test_7_listdir(self): + def test_7_listdir(self, sftp): """ 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: - sftp.open(FOLDER + '/duck.txt', 'w').close() - sftp.open(FOLDER + '/fish.txt', 'w').close() - sftp.open(FOLDER + '/tertiary.py', 'w').close() + sftp.open(sftp.FOLDER + '/duck.txt', 'w').close() + sftp.open(sftp.FOLDER + '/fish.txt', 'w').close() + sftp.open(sftp.FOLDER + '/tertiary.py', 'w').close() - x = sftp.listdir(FOLDER) - self.assertEqual(len(x), 3) + x = sftp.listdir(sftp.FOLDER) + assert len(x) == 3 self.assertTrue('duck.txt' in x) self.assertTrue('fish.txt' in x) self.assertTrue('tertiary.py' in x) self.assertTrue('random' not in x) finally: - sftp.remove(FOLDER + '/duck.txt') - sftp.remove(FOLDER + '/fish.txt') - sftp.remove(FOLDER + '/tertiary.py') + sftp.remove(sftp.FOLDER + '/duck.txt') + sftp.remove(sftp.FOLDER + '/fish.txt') + sftp.remove(sftp.FOLDER + '/tertiary.py') - def test_7_5_listdir_iter(self): + def test_7_5_listdir_iter(self, sftp): """ listdir_iter version of above test """ try: - sftp.open(FOLDER + '/duck.txt', 'w').close() - sftp.open(FOLDER + '/fish.txt', 'w').close() - sftp.open(FOLDER + '/tertiary.py', 'w').close() + sftp.open(sftp.FOLDER + '/duck.txt', 'w').close() + sftp.open(sftp.FOLDER + '/fish.txt', 'w').close() + sftp.open(sftp.FOLDER + '/tertiary.py', 'w').close() - x = [x.filename for x in sftp.listdir_iter(FOLDER)] - self.assertEqual(len(x), 3) + x = [x.filename for x in sftp.listdir_iter(sftp.FOLDER)] + assert len(x) == 3 self.assertTrue('duck.txt' in x) self.assertTrue('fish.txt' in x) self.assertTrue('tertiary.py' in x) self.assertTrue('random' not in x) finally: - sftp.remove(FOLDER + '/duck.txt') - sftp.remove(FOLDER + '/fish.txt') - sftp.remove(FOLDER + '/tertiary.py') + sftp.remove(sftp.FOLDER + '/duck.txt') + sftp.remove(sftp.FOLDER + '/fish.txt') + sftp.remove(sftp.FOLDER + '/tertiary.py') - def test_8_setstat(self): + def test_8_setstat(self, sftp): """ verify that the setstat functions (chown, chmod, utime, truncate) work. """ try: - with sftp.open(FOLDER + '/special', 'w') as f: + with sftp.open(sftp.FOLDER + '/special', 'w') as f: f.write('x' * 1024) - stat = sftp.stat(FOLDER + '/special') - sftp.chmod(FOLDER + '/special', (stat.st_mode & ~o777) | o600) - stat = sftp.stat(FOLDER + '/special') + stat = sftp.stat(sftp.FOLDER + '/special') + sftp.chmod(sftp.FOLDER + '/special', (stat.st_mode & ~o777) | o600) + stat = sftp.stat(sftp.FOLDER + '/special') expected_mode = o600 if sys.platform == 'win32': # chmod not really functional on windows @@ -299,35 +258,35 @@ class SFTPTest(unittest.TestCase): if sys.platform == 'cygwin': # even worse. expected_mode = o644 - self.assertEqual(stat.st_mode & o777, expected_mode) - self.assertEqual(stat.st_size, 1024) + assert stat.st_mode & o777 == expected_mode + assert stat.st_size == 1024 mtime = stat.st_mtime - 3600 atime = stat.st_atime - 1800 - sftp.utime(FOLDER + '/special', (atime, mtime)) - stat = sftp.stat(FOLDER + '/special') - self.assertEqual(stat.st_mtime, mtime) + sftp.utime(sftp.FOLDER + '/special', (atime, mtime)) + stat = sftp.stat(sftp.FOLDER + '/special') + assert stat.st_mtime == mtime if sys.platform not in ('win32', 'cygwin'): - self.assertEqual(stat.st_atime, atime) + assert stat.st_atime == atime # can't really test chown, since we'd have to know a valid uid. - sftp.truncate(FOLDER + '/special', 512) - stat = sftp.stat(FOLDER + '/special') - self.assertEqual(stat.st_size, 512) + sftp.truncate(sftp.FOLDER + '/special', 512) + stat = sftp.stat(sftp.FOLDER + '/special') + assert stat.st_size == 512 finally: - sftp.remove(FOLDER + '/special') + sftp.remove(sftp.FOLDER + '/special') - def test_9_fsetstat(self): + def test_9_fsetstat(self, sftp): """ verify that the fsetstat functions (chown, chmod, utime, truncate) work on open files. """ try: - with sftp.open(FOLDER + '/special', 'w') as f: + with sftp.open(sftp.FOLDER + '/special', 'w') as f: f.write('x' * 1024) - with sftp.open(FOLDER + '/special', 'r+') as f: + with sftp.open(sftp.FOLDER + '/special', 'r+') as f: stat = f.stat() f.chmod((stat.st_mode & ~o777) | o600) stat = f.stat() @@ -339,26 +298,26 @@ class SFTPTest(unittest.TestCase): if sys.platform == 'cygwin': # even worse. expected_mode = o644 - self.assertEqual(stat.st_mode & o777, expected_mode) - self.assertEqual(stat.st_size, 1024) + assert stat.st_mode & o777 == expected_mode + assert stat.st_size == 1024 mtime = stat.st_mtime - 3600 atime = stat.st_atime - 1800 f.utime((atime, mtime)) stat = f.stat() - self.assertEqual(stat.st_mtime, mtime) + assert stat.st_mtime == mtime if sys.platform not in ('win32', 'cygwin'): - self.assertEqual(stat.st_atime, atime) + assert stat.st_atime == atime # can't really test chown, since we'd have to know a valid uid. f.truncate(512) stat = f.stat() - self.assertEqual(stat.st_size, 512) + assert stat.st_size == 512 finally: - sftp.remove(FOLDER + '/special') + sftp.remove(sftp.FOLDER + '/special') - def test_A_readline_seek(self): + def test_A_readline_seek(self, sftp): """ create a text file and write a bunch of text into it. then count the lines in the file, and seek around to retrieve particular lines. this should @@ -366,10 +325,10 @@ class SFTPTest(unittest.TestCase): buffering is reset on 'seek'. """ try: - with sftp.open(FOLDER + '/duck.txt', 'w') as f: + with sftp.open(sftp.FOLDER + '/duck.txt', 'w') as f: f.write(ARTICLE) - with sftp.open(FOLDER + '/duck.txt', 'r+') as f: + with sftp.open(sftp.FOLDER + '/duck.txt', 'r+') as f: line_number = 0 loc = 0 pos_list = [] @@ -379,33 +338,33 @@ class SFTPTest(unittest.TestCase): loc = f.tell() self.assertTrue(f.seekable()) f.seek(pos_list[6], f.SEEK_SET) - self.assertEqual(f.readline(), 'Nouzilly, France.\n') + assert f.readline(), 'Nouzilly == France.\n' f.seek(pos_list[17], f.SEEK_SET) - self.assertEqual(f.readline()[:4], 'duck') + assert 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') + assert f.readline() == 'duck types were equally resistant to exogenous insulin compared with chicken.\n' finally: - sftp.remove(FOLDER + '/duck.txt') + sftp.remove(sftp.FOLDER + '/duck.txt') - def test_B_write_seek(self): + def test_B_write_seek(self, sftp): """ create a text file, seek back and change part of it, and verify that the changes worked. """ try: - with sftp.open(FOLDER + '/testing.txt', 'w') as f: + with sftp.open(sftp.FOLDER + '/testing.txt', 'w') as f: f.write('hello kitty.\n') f.seek(-5, f.SEEK_CUR) f.write('dd') - self.assertEqual(sftp.stat(FOLDER + '/testing.txt').st_size, 13) - with sftp.open(FOLDER + '/testing.txt', 'r') as f: + assert sftp.stat(sftp.FOLDER + '/testing.txt').st_size == 13 + with sftp.open(sftp.FOLDER + '/testing.txt', 'r') as f: data = f.read(20) - self.assertEqual(data, b'hello kiddy.\n') + assert data == b'hello kiddy.\n' finally: - sftp.remove(FOLDER + '/testing.txt') + sftp.remove(sftp.FOLDER + '/testing.txt') - def test_C_symlink(self): + def test_C_symlink(self, sftp): """ create a symlink and then check that lstat doesn't follow it. """ @@ -414,97 +373,97 @@ class SFTPTest(unittest.TestCase): return try: - with sftp.open(FOLDER + '/original.txt', 'w') as f: + with sftp.open(sftp.FOLDER + '/original.txt', 'w') as f: f.write('original\n') - sftp.symlink('original.txt', FOLDER + '/link.txt') - self.assertEqual(sftp.readlink(FOLDER + '/link.txt'), 'original.txt') + sftp.symlink('original.txt', sftp.FOLDER + '/link.txt') + assert sftp.readlink(sftp.FOLDER + '/link.txt') == 'original.txt' - with sftp.open(FOLDER + '/link.txt', 'r') as f: - self.assertEqual(f.readlines(), ['original\n']) + with sftp.open(sftp.FOLDER + '/link.txt', 'r') as f: + assert f.readlines() == ['original\n'] 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')) + abs_path = cwd + '/' + sftp.FOLDER + '/original.txt' + sftp.symlink(abs_path, sftp.FOLDER + '/link2.txt') + assert abs_path == sftp.readlink(sftp.FOLDER + '/link2.txt') - self.assertEqual(sftp.lstat(FOLDER + '/link.txt').st_size, 12) - self.assertEqual(sftp.stat(FOLDER + '/link.txt').st_size, 9) + assert sftp.lstat(sftp.FOLDER + '/link.txt').st_size == 12 + assert sftp.stat(sftp.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.assertTrue(sftp.lstat(FOLDER + '/link2.txt').st_size >= len(abs_path)) - self.assertEqual(sftp.stat(FOLDER + '/link2.txt').st_size, 9) - self.assertEqual(sftp.stat(FOLDER + '/original.txt').st_size, 9) + self.assertTrue(sftp.lstat(sftp.FOLDER + '/link2.txt').st_size >= len(abs_path)) + assert sftp.stat(sftp.FOLDER + '/link2.txt').st_size == 9 + assert sftp.stat(sftp.FOLDER + '/original.txt').st_size == 9 finally: try: - sftp.remove(FOLDER + '/link.txt') + sftp.remove(sftp.FOLDER + '/link.txt') except: pass try: - sftp.remove(FOLDER + '/link2.txt') + sftp.remove(sftp.FOLDER + '/link2.txt') except: pass try: - sftp.remove(FOLDER + '/original.txt') + sftp.remove(sftp.FOLDER + '/original.txt') except: pass - def test_D_flush_seek(self): + def test_D_flush_seek(self, sftp): """ verify that buffered writes are automatically flushed on seek. """ try: - with sftp.open(FOLDER + '/happy.txt', 'w', 1) as f: + with sftp.open(sftp.FOLDER + '/happy.txt', 'w', 1) as f: f.write('full line.\n') f.write('partial') f.seek(9, f.SEEK_SET) f.write('?\n') - with sftp.open(FOLDER + '/happy.txt', 'r') as f: - self.assertEqual(f.readline(), u('full line?\n')) - self.assertEqual(f.read(7), b'partial') + with sftp.open(sftp.FOLDER + '/happy.txt', 'r') as f: + assert f.readline() == u('full line?\n') + assert f.read(7) == b'partial' finally: try: - sftp.remove(FOLDER + '/happy.txt') + sftp.remove(sftp.FOLDER + '/happy.txt') except: pass - def test_E_realpath(self): + def test_E_realpath(self, sftp): """ test that realpath is returning something non-empty and not an error. """ pwd = sftp.normalize('.') self.assertTrue(len(pwd) > 0) - f = sftp.normalize('./' + FOLDER) + f = sftp.normalize('./' + sftp.FOLDER) self.assertTrue(len(f) > 0) - self.assertEqual(os.path.join(pwd, FOLDER), f) + assert os.path.join(pwd, sftp.FOLDER) == f - def test_F_mkdir(self): + def test_F_mkdir(self, sftp): """ verify that mkdir/rmdir work. """ try: - sftp.mkdir(FOLDER + '/subfolder') + sftp.mkdir(sftp.FOLDER + '/subfolder') except: self.assertTrue(False, 'exception creating subfolder') try: - sftp.mkdir(FOLDER + '/subfolder') + sftp.mkdir(sftp.FOLDER + '/subfolder') self.assertTrue(False, 'no exception overwriting subfolder') except IOError: pass try: - sftp.rmdir(FOLDER + '/subfolder') + sftp.rmdir(sftp.FOLDER + '/subfolder') except: self.assertTrue(False, 'exception removing subfolder') try: - sftp.rmdir(FOLDER + '/subfolder') + sftp.rmdir(sftp.FOLDER + '/subfolder') self.assertTrue(False, 'no exception removing nonexistent subfolder') except IOError: pass - def test_G_chdir(self): + def test_G_chdir(self, sftp): """ verify that chdir/getcwd work. """ @@ -512,35 +471,35 @@ class SFTPTest(unittest.TestCase): if root[-1] != '/': root += '/' try: - sftp.mkdir(FOLDER + '/alpha') - sftp.chdir(FOLDER + '/alpha') + sftp.mkdir(sftp.FOLDER + '/alpha') + sftp.chdir(sftp.FOLDER + '/alpha') sftp.mkdir('beta') - self.assertEqual(root + FOLDER + '/alpha', sftp.getcwd()) - self.assertEqual(['beta'], sftp.listdir('.')) + assert root + sftp.FOLDER + '/alpha' == sftp.getcwd() + assert ['beta'] == sftp.listdir('.') sftp.chdir('beta') with sftp.open('fish', 'w') as f: f.write('hello\n') sftp.chdir('..') - self.assertEqual(['fish'], sftp.listdir('beta')) + assert ['fish'] == sftp.listdir('beta') sftp.chdir('..') - self.assertEqual(['fish'], sftp.listdir('alpha/beta')) + assert ['fish'] == sftp.listdir('alpha/beta') finally: sftp.chdir(root) try: - sftp.unlink(FOLDER + '/alpha/beta/fish') + sftp.unlink(sftp.FOLDER + '/alpha/beta/fish') except: pass try: - sftp.rmdir(FOLDER + '/alpha/beta') + sftp.rmdir(sftp.FOLDER + '/alpha/beta') except: pass try: - sftp.rmdir(FOLDER + '/alpha') + sftp.rmdir(sftp.FOLDER + '/alpha') except: pass - def test_H_get_put(self): + def test_H_get_put(self, sftp): """ verify that get/put work. """ @@ -555,103 +514,103 @@ class SFTPTest(unittest.TestCase): def progress_callback(x, y): saved_progress.append((x, y)) - sftp.put(localname, FOLDER + '/bunny.txt', progress_callback) + sftp.put(localname, sftp.FOLDER + '/bunny.txt', progress_callback) - with sftp.open(FOLDER + '/bunny.txt', 'rb') as f: - self.assertEqual(text, f.read(128)) - self.assertEqual([(41, 41)], saved_progress) + with sftp.open(sftp.FOLDER + '/bunny.txt', 'rb') as f: + assert text == f.read(128) + assert [(41, 41)] == saved_progress os.unlink(localname) fd, localname = mkstemp() os.close(fd) saved_progress = [] - sftp.get(FOLDER + '/bunny.txt', localname, progress_callback) + sftp.get(sftp.FOLDER + '/bunny.txt', localname, progress_callback) with open(localname, 'rb') as f: - self.assertEqual(text, f.read(128)) - self.assertEqual([(41, 41)], saved_progress) + assert text == f.read(128) + assert [(41, 41)] == saved_progress os.unlink(localname) - sftp.unlink(FOLDER + '/bunny.txt') + sftp.unlink(sftp.FOLDER + '/bunny.txt') - def test_I_check(self): + def test_I_check(self, sftp): """ 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.) """ - with sftp.open(FOLDER + '/kitty.txt', 'w') as f: + with sftp.open(sftp.FOLDER + '/kitty.txt', 'w') as f: f.write('here kitty kitty' * 64) try: - with sftp.open(FOLDER + '/kitty.txt', 'r') as f: + with sftp.open(sftp.FOLDER + '/kitty.txt', 'r') as f: sum = f.check('sha1') - self.assertEqual('91059CFC6615941378D413CB5ADAF4C5EB293402', u(hexlify(sum)).upper()) + assert '91059CFC6615941378D413CB5ADAF4C5EB293402' == u(hexlify(sum)).upper() sum = f.check('md5', 0, 512) - self.assertEqual('93DE4788FCA28D471516963A1FE3856A', u(hexlify(sum)).upper()) + assert '93DE4788FCA28D471516963A1FE3856A' == u(hexlify(sum)).upper() sum = f.check('md5', 0, 0, 510) self.assertEqual('EB3B45B8CD55A0707D99B177544A319F373183D241432BB2157AB9E46358C4AC90370B5CADE5D90336FC1716F90B36D6', u(hexlify(sum)).upper()) finally: - sftp.unlink(FOLDER + '/kitty.txt') + sftp.unlink(sftp.FOLDER + '/kitty.txt') - def test_J_x_flag(self): + def test_J_x_flag(self, sftp): """ verify that the 'x' flag works when opening a file. """ - sftp.open(FOLDER + '/unusual.txt', 'wx').close() + sftp.open(sftp.FOLDER + '/unusual.txt', 'wx').close() try: try: - sftp.open(FOLDER + '/unusual.txt', 'wx') + sftp.open(sftp.FOLDER + '/unusual.txt', 'wx') self.fail('expected exception') except IOError: pass finally: - sftp.unlink(FOLDER + '/unusual.txt') + sftp.unlink(sftp.FOLDER + '/unusual.txt') - def test_K_utf8(self): + def test_K_utf8(self, sftp): """ verify that unicode strings are encoded into utf8 correctly. """ - with sftp.open(FOLDER + '/something', 'w') as f: + with sftp.open(sftp.FOLDER + '/something', 'w') as f: f.write('okay') try: - sftp.rename(FOLDER + '/something', FOLDER + '/' + unicode_folder) - sftp.open(b(FOLDER) + utf8_folder, 'r') + sftp.rename(sftp.FOLDER + '/something', sftp.FOLDER + '/' + unicode_folder) + sftp.open(b(sftp.FOLDER) + utf8_folder, 'r') except Exception as e: self.fail('exception ' + str(e)) - sftp.unlink(b(FOLDER) + utf8_folder) + sftp.unlink(b(sftp.FOLDER) + utf8_folder) - def test_L_utf8_chdir(self): - sftp.mkdir(FOLDER + '/' + unicode_folder) + def test_L_utf8_chdir(self, sftp): + sftp.mkdir(sftp.FOLDER + '/' + unicode_folder) try: - sftp.chdir(FOLDER + '/' + unicode_folder) + sftp.chdir(sftp.FOLDER + '/' + unicode_folder) with sftp.open('something', 'w') as f: f.write('okay') sftp.unlink('something') finally: sftp.chdir() - sftp.rmdir(FOLDER + '/' + unicode_folder) + sftp.rmdir(sftp.FOLDER + '/' + unicode_folder) - def test_M_bad_readv(self): + def test_M_bad_readv(self, sftp): """ verify that readv at the end of the file doesn't essplode. """ - sftp.open(FOLDER + '/zero', 'w').close() + sftp.open(sftp.FOLDER + '/zero', 'w').close() try: - with sftp.open(FOLDER + '/zero', 'r') as f: + with sftp.open(sftp.FOLDER + '/zero', 'r') as f: f.readv([(0, 12)]) - with sftp.open(FOLDER + '/zero', 'r') as f: + with sftp.open(sftp.FOLDER + '/zero', 'r') as f: file_size = f.stat().st_size f.prefetch(file_size) f.read(100) finally: - sftp.unlink(FOLDER + '/zero') + sftp.unlink(sftp.FOLDER + '/zero') - def test_N_put_without_confirm(self): + def test_N_put_without_confirm(self, sftp): """ verify that get/put work without confirmation. """ @@ -666,61 +625,61 @@ class SFTPTest(unittest.TestCase): def progress_callback(x, y): saved_progress.append((x, y)) - res = sftp.put(localname, FOLDER + '/bunny.txt', progress_callback, False) + res = sftp.put(localname, sftp.FOLDER + '/bunny.txt', progress_callback, False) - self.assertEqual(SFTPAttributes().attr, res.attr) + assert SFTPAttributes().attr == res.attr - with sftp.open(FOLDER + '/bunny.txt', 'r') as f: - self.assertEqual(text, f.read(128)) - self.assertEqual((41, 41), saved_progress[-1]) + with sftp.open(sftp.FOLDER + '/bunny.txt', 'r') as f: + assert text == f.read(128) + assert (41, 41) == saved_progress[-1] os.unlink(localname) - sftp.unlink(FOLDER + '/bunny.txt') + sftp.unlink(sftp.FOLDER + '/bunny.txt') - def test_O_getcwd(self): + def test_O_getcwd(self, sftp): """ verify that chdir/getcwd work. """ - self.assertEqual(None, sftp.getcwd()) + assert sftp.getcwd() == None root = sftp.normalize('.') if root[-1] != '/': root += '/' try: - sftp.mkdir(FOLDER + '/alpha') - sftp.chdir(FOLDER + '/alpha') - self.assertEqual('/' + FOLDER + '/alpha', sftp.getcwd()) + sftp.mkdir(sftp.FOLDER + '/alpha') + sftp.chdir(sftp.FOLDER + '/alpha') + assert sftp.getcwd() == '/' + sftp.FOLDER + '/alpha' finally: sftp.chdir(root) try: - sftp.rmdir(FOLDER + '/alpha') + sftp.rmdir(sftp.FOLDER + '/alpha') except: pass - def XXX_test_M_seek_append(self): + def XXX_test_M_seek_append(self, sftp): """ verify that seek does't affect writes during append. does not work except through paramiko. :( openssh fails. """ try: - with sftp.open(FOLDER + '/append.txt', 'a') as f: + with sftp.open(sftp.FOLDER + '/append.txt', 'a') as f: f.write('first line\nsecond line\n') f.seek(11, f.SEEK_SET) f.write('third line\n') - with sftp.open(FOLDER + '/append.txt', 'r') as f: - self.assertEqual(f.stat().st_size, 34) - self.assertEqual(f.readline(), 'first line\n') - self.assertEqual(f.readline(), 'second line\n') - self.assertEqual(f.readline(), 'third line\n') + with sftp.open(sftp.FOLDER + '/append.txt', 'r') as f: + assert f.stat().st_size == 34 + assert f.readline() == 'first line\n' + assert f.readline() == 'second line\n' + assert f.readline() == 'third line\n' finally: - sftp.remove(FOLDER + '/append.txt') + sftp.remove(sftp.FOLDER + '/append.txt') - def test_putfo_empty_file(self): + def test_putfo_empty_file(self, sftp): """ Send an empty file and confirm it is sent. """ - target = FOLDER + '/empty file.txt' + target = sftp.FOLDER + '/empty file.txt' stream = StringIO() try: attrs = sftp.putfo(stream, target) @@ -730,69 +689,71 @@ class SFTPTest(unittest.TestCase): sftp.remove(target) - def test_N_file_with_percent(self): + def test_N_file_with_percent(self, sftp): """ verify that we can create a file with a '%' in the filename. ( it needs to be properly escaped by _log() ) """ - self.assertTrue( paramiko.util.get_logger("paramiko").handlers, "This unit test requires logging to be enabled" ) - f = sftp.open(FOLDER + '/test%file', 'w') + # TODO: how best to enable this only for the one test? & how to make it + # not log to actual file? lol?? + paramiko.util.log_to_file('test_sftp.log') + f = sftp.open(sftp.FOLDER + '/test%file', 'w') try: - self.assertEqual(f.stat().st_size, 0) + assert f.stat().st_size == 0 finally: f.close() - sftp.remove(FOLDER + '/test%file') + sftp.remove(sftp.FOLDER + '/test%file') - def test_O_non_utf8_data(self): + def test_O_non_utf8_data(self, sftp): """Test write() and read() of non utf8 data""" try: - with sftp.open('%s/nonutf8data' % FOLDER, 'w') as f: + with sftp.open('%s/nonutf8data' % sftp.FOLDER, 'w') as f: f.write(NON_UTF8_DATA) - with sftp.open('%s/nonutf8data' % FOLDER, 'r') as f: + with sftp.open('%s/nonutf8data' % sftp.FOLDER, 'r') as f: data = f.read() - self.assertEqual(data, NON_UTF8_DATA) - with sftp.open('%s/nonutf8data' % FOLDER, 'wb') as f: + assert data == NON_UTF8_DATA + with sftp.open('%s/nonutf8data' % sftp.FOLDER, 'wb') as f: f.write(NON_UTF8_DATA) - with sftp.open('%s/nonutf8data' % FOLDER, 'rb') as f: + with sftp.open('%s/nonutf8data' % sftp.FOLDER, 'rb') as f: data = f.read() - self.assertEqual(data, NON_UTF8_DATA) + assert data == NON_UTF8_DATA finally: - sftp.remove('%s/nonutf8data' % FOLDER) + sftp.remove('%s/nonutf8data' % sftp.FOLDER) - def test_sftp_attributes_empty_str(self): + def test_sftp_attributes_empty_str(self, sftp): sftp_attributes = SFTPAttributes() - self.assertEqual(str(sftp_attributes), "?--------- 1 0 0 0 (unknown date) ?") + assert str(sftp_attributes) == "?--------- 1 0 0 0 (unknown date) ?" @needs_builtin('buffer') - def test_write_buffer(self): + def test_write_buffer(self, sftp): """Test write() using a buffer instance.""" data = 3 * b'A potentially large block of data to chunk up.\n' try: - with sftp.open('%s/write_buffer' % FOLDER, 'wb') as f: + with sftp.open('%s/write_buffer' % sftp.FOLDER, 'wb') as f: for offset in range(0, len(data), 8): f.write(buffer(data, offset, 8)) - with sftp.open('%s/write_buffer' % FOLDER, 'rb') as f: - self.assertEqual(f.read(), data) + with sftp.open('%s/write_buffer' % sftp.FOLDER, 'rb') as f: + assert f.read() == data finally: - sftp.remove('%s/write_buffer' % FOLDER) + sftp.remove('%s/write_buffer' % sftp.FOLDER) @needs_builtin('memoryview') - def test_write_memoryview(self): + def test_write_memoryview(self, sftp): """Test write() using a memoryview instance.""" data = 3 * b'A potentially large block of data to chunk up.\n' try: - with sftp.open('%s/write_memoryview' % FOLDER, 'wb') as f: + with sftp.open('%s/write_memoryview' % sftp.FOLDER, 'wb') as f: view = memoryview(data) for offset in range(0, len(data), 8): f.write(view[offset:offset+8]) - with sftp.open('%s/write_memoryview' % FOLDER, 'rb') as f: - self.assertEqual(f.read(), data) + with sftp.open('%s/write_memoryview' % sftp.FOLDER, 'rb') as f: + assert f.read() == data finally: - sftp.remove('%s/write_memoryview' % FOLDER) + sftp.remove('%s/write_memoryview' % sftp.FOLDER) if __name__ == '__main__': |