dorian
/
paramiko
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

[project @ robey@lag.net-20060220031741-3227438ff1b03a29] 

add SFTPClient.truncate.  add chmod, chown, utime, and truncate to SFTPFile.  and of course tests.
This commit is contained in:
Robey Pointer 2006-02-19 19:17:41 -08:00
parent 8446c0bda1
commit 4be91d1c4d
5 changed files with 193 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

109
paramiko/sftp_client.py
View File

@ -59,7 +59,7 @@ class SFTPClient (BaseSFTP):
An alternate way to create an SFTP client context is by using
L{from_transport}.
@param sock: an open L{Channel} using the C{"sftp"} subsystem.
@param sock: an open L{Channel} using the C{"sftp"} subsystem
@type sock: L{Channel}
"""
BaseSFTP.__init__(self)
@ -84,10 +84,10 @@ class SFTPClient (BaseSFTP):
"""
Create an SFTP client channel from an open L{Transport}.
@param t: an open L{Transport} which is already authenticated.
@param t: an open L{Transport} which is already authenticated
@type t: L{Transport}
@return: a new L{SFTPClient} object, referring to an sftp session
(channel) across the transport.
(channel) across the transport
@rtype: L{SFTPClient}
"""
chan = t.open_session()
@ -185,13 +185,13 @@ class SFTPClient (BaseSFTP):
buffering, C{1} uses line buffering, and any number greater than 1
(C{>1}) uses that specific buffer size.
@param filename: name of the file to open.
@type filename: string
@param mode: mode (python-style) to open in.
@type mode: string
@param filename: name of the file to open
@type filename: str
@param mode: mode (python-style) to open in
@type mode: str
@param bufsize: desired buffering (-1 = default buffer size)
@type bufsize: int
@return: a file object representing the open file.
@return: a file object representing the open file
@rtype: SFTPFile
@raise IOError: if the file could not be opened.
@ -223,13 +223,13 @@ class SFTPClient (BaseSFTP):
def remove(self, path):
"""
Remove the file at the given path.
Remove the file at the given path. This only works on files; for
removing folders (directories), use L{rmdir}.
@param path: path (absolute or relative) of the file to remove.
@type path: string
@param path: path (absolute or relative) of the file to remove
@type path: str
@raise IOError: if the path refers to a folder (directory). Use
L{rmdir} to remove a folder.
@raise IOError: if the path refers to a folder (directory)
"""
path = self._adjust_cwd(path)
self._log(DEBUG, 'remove(%r)' % path)
@ -241,13 +241,13 @@ class SFTPClient (BaseSFTP):
"""
Rename a file or folder from C{oldpath} to C{newpath}.
@param oldpath: existing name of the file or folder.
@type oldpath: string
@param newpath: new name for the file or folder.
@type newpath: string
@param oldpath: existing name of the file or folder
@type oldpath: str
@param newpath: new name for the file or folder
@type newpath: str
@raise IOError: if C{newpath} is a folder, or something else goes
wrong.
wrong
"""
oldpath = self._adjust_cwd(oldpath)
newpath = self._adjust_cwd(newpath)
@ -260,9 +260,9 @@ class SFTPClient (BaseSFTP):
The default mode is 0777 (octal). On some systems, mode is ignored.
Where it is used, the current umask value is first masked out.
@param path: name of the folder to create.
@type path: string
@param mode: permissions (posix-style) for the newly-created folder.
@param path: name of the folder to create
@type path: str
@param mode: permissions (posix-style) for the newly-created folder
@type mode: int
"""
path = self._adjust_cwd(path)
@ -275,8 +275,8 @@ class SFTPClient (BaseSFTP):
"""
Remove the folder named C{path}.
@param path: name of the folder to remove.
@type path: string
@param path: name of the folder to remove
@type path: str
"""
path = self._adjust_cwd(path)
self._log(DEBUG, 'rmdir(%r)' % path)
@ -296,9 +296,9 @@ class SFTPClient (BaseSFTP):
The fields supported are: C{st_mode}, C{st_size}, C{st_uid}, C{st_gid},
C{st_atime}, and C{st_mtime}.
@param path: the filename to stat.
@type path: string
@return: an object containing attributes about the given file.
@param path: the filename to stat
@type path: str
@return: an object containing attributes about the given file
@rtype: SFTPAttributes
"""
path = self._adjust_cwd(path)
@ -314,9 +314,9 @@ class SFTPClient (BaseSFTP):
following symbolic links (shortcuts). This otherwise behaves exactly
the same as L{stat}.
@param path: the filename to stat.
@type path: string
@return: an object containing attributes about the given file.
@param path: the filename to stat
@type path: str
@return: an object containing attributes about the given file
@rtype: SFTPAttributes
"""
path = self._adjust_cwd(path)
@ -331,10 +331,10 @@ class SFTPClient (BaseSFTP):
Create a symbolic link (shortcut) of the C{source} path at
C{destination}.
@param source: path of the original file.
@type source: string
@param dest: path of the newly created symlink.
@type dest: string
@param source: path of the original file
@type source: str
@param dest: path of the newly created symlink
@type dest: str
"""
dest = self._adjust_cwd(dest)
self._log(DEBUG, 'symlink(%r, %r)' % (source, dest))
@ -348,9 +348,9 @@ class SFTPClient (BaseSFTP):
unix-style and identical to those used by python's C{os.chmod}
function.
@param path: path of the file to change the permissions of.
@type path: string
@param mode: new permissions.
@param path: path of the file to change the permissions of
@type path: str
@param mode: new permissions
@type mode: int
"""
path = self._adjust_cwd(path)
@ -366,8 +366,8 @@ class SFTPClient (BaseSFTP):
only want to change one, use L{stat} first to retrieve the current
owner and group.
@param path: path of the file to change the owner and group of.
@type path: string
@param path: path of the file to change the owner and group of
@type path: str
@param uid: new owner's uid
@type uid: int
@param gid: new group id
@ -388,11 +388,11 @@ class SFTPClient (BaseSFTP):
modified times, respectively. This bizarre API is mimicked from python
for the sake of consistency -- I apologize.
@param path: path of the file to modify.
@type path: string
@param path: path of the file to modify
@type path: str
@param times: C{None} or a tuple of (access time, modified time) in
standard internet epoch time (seconds since 01 January 1970 GMT).
@type times: tuple of int
standard internet epoch time (seconds since 01 January 1970 GMT)
@type times: tuple(int)
"""
path = self._adjust_cwd(path)
if times is None:
@ -402,15 +402,32 @@ class SFTPClient (BaseSFTP):
attr.st_atime, attr.st_mtime = times
self._request(CMD_SETSTAT, path, attr)
def truncate(self, path, size):
"""
Change the size of the file specified by C{path}. This usually extends
or shrinks the size of the file, just like the C{truncate()} method on
python file objects.
@param path: path of the file to modify
@type path: str
@param size: the new size of the file
@type size: int or long
"""
path = self._adjust_cwd(path)
self._log(DEBUG, 'truncate(%r, %r)' % (path, size))
attr = SFTPAttributes()
attr.st_size = size
self._request(CMD_SETSTAT, path, attr)
def readlink(self, path):
"""
Return the target of a symbolic link (shortcut). You can use
L{symlink} to create these. The result may be either an absolute or
relative pathname.
@param path: path of the symbolic link file.
@param path: path of the symbolic link file
@type path: str
@return: target path.
@return: target path
@rtype: str
"""
path = self._adjust_cwd(path)
@ -432,9 +449,9 @@ class SFTPClient (BaseSFTP):
server is considering to be the "current folder" (by passing C{'.'}
as C{path}).
@param path: path to be normalized.
@param path: path to be normalized
@type path: str
@return: normalized form of the given path.
@return: normalized form of the given path
@rtype: str
@raise IOError: if the path can't be resolved on the server

65
paramiko/sftp_file.py
View File

@ -182,6 +182,71 @@ class SFTPFile (BufferedFile):
raise SFTPError('Expected attributes')
return SFTPAttributes._from_msg(msg)
def chmod(self, mode):
"""
Change the mode (permissions) of this file. The permissions are
unix-style and identical to those used by python's C{os.chmod}
function.
@param mode: new permissions
@type mode: int
"""
self.sftp._log(DEBUG, 'chmod(%s, %r)' % (util.hexify(self.handle), mode))
attr = SFTPAttributes()
attr.st_mode = mode
self.sftp._request(CMD_FSETSTAT, self.handle, attr)
def chown(self, uid, gid):
"""
Change the owner (C{uid}) and group (C{gid}) of this file. As with
python's C{os.chown} function, you must pass both arguments, so if you
only want to change one, use L{stat} first to retrieve the current
owner and group.
@param uid: new owner's uid
@type uid: int
@param gid: new group id
@type gid: int
"""
self.sftp._log(DEBUG, 'chown(%s, %r, %r)' % (util.hexify(self.handle), uid, gid))
attr = SFTPAttributes()
attr.st_uid, attr.st_gid = uid, gid
self.sftp._request(CMD_FSETSTAT, self.handle, attr)
def utime(self, times):
"""
Set the access and modified times of this file. If
C{times} is C{None}, then the file's access and modified times are set
to the current time. Otherwise, C{times} must be a 2-tuple of numbers,
of the form C{(atime, mtime)}, which is used to set the access and
modified times, respectively. This bizarre API is mimicked from python
for the sake of consistency -- I apologize.
@param times: C{None} or a tuple of (access time, modified time) in
standard internet epoch time (seconds since 01 January 1970 GMT)
@type times: tuple(int)
"""
if times is None:
times = (time.time(), time.time())
self.sftp._log(DEBUG, 'utime(%s, %r)' % (util.hexify(self.handle), times))
attr = SFTPAttributes()
attr.st_atime, attr.st_mtime = times
self.sftp._request(CMD_FSETSTAT, self.handle, attr)
def truncate(self, size):
"""
Change the size of this file. This usually extends
or shrinks the size of the file, just like the C{truncate()} method on
python file objects.
@param size: the new size of the file
@type size: int or long
"""
self.sftp._log(DEBUG, 'truncate(%s, %r)' % (util.hexify(self.handle), size))
attr = SFTPAttributes()
attr.st_size = size
self.sftp._request(CMD_FSETSTAT, self.handle, attr)
def check(self, hash_algorithm, offset=0, length=0, block_size=0):
"""
Ask the server for a hash of a section of this file. This can be used

2
paramiko/sftp_server.py
View File

@ -147,6 +147,8 @@ class SFTPServer (BaseSFTP, SubsystemHandler):
os.chown(filename, attr.st_uid, attr.st_gid)
if attr._flags & attr.FLAG_AMTIME:
os.utime(filename, (attr.st_atime, attr.st_mtime))
if attr._flags & attr.FLAG_SIZE:
open(filename, 'w+').truncate(attr.st_size)
set_file_attr = staticmethod(set_file_attr)

1
tests/stub_sftp.py
View File

@ -48,6 +48,7 @@ class StubSFTPHandle (SFTPHandle):
# use the stored filename
try:
SFTPServer.set_file_attr(self.filename, attr)
return SFTP_OK
except OSError, e:
return SFTPServer.convert_errno(e.errno)

84
tests/test_sftp.py
View File

@ -273,28 +273,68 @@ class SFTPTest (unittest.TestCase):
def test_8_setstat(self):
"""
verify that the setstat functions (chown, chmod, utime) work.
verify that the setstat functions (chown, chmod, utime, truncate) work.
"""
f = sftp.open(FOLDER + '/special', 'w')
try:
f.write('x' * 1024)
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)
stat = sftp.stat(FOLDER + '/special')
self.assertEqual(stat.st_mode & 0777, 0600)
self.assertEqual(stat.st_size, 1024)
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)
stat = sftp.stat(FOLDER + '/special')
self.assertEqual(stat.st_mtime, mtime)
self.assertEqual(stat.st_atime, atime)
# can't really test chown, since we'd have to know a valid uid.
sftp.truncate(FOLDER + '/special', 512)
stat = sftp.stat(FOLDER + '/special')
self.assertEqual(stat.st_size, 512)
finally:
sftp.remove(FOLDER + '/special')
def test_9_readline_seek(self):
def test_9_fsetstat(self):
"""
verify that the fsetstat functions (chown, chmod, utime, truncate)
work on open files.
"""
f = sftp.open(FOLDER + '/special', 'w')
try:
f.write('x' * 1024)
f.close()
f = sftp.open(FOLDER + '/special', 'r+')
stat = f.stat()
f.chmod((stat.st_mode & ~0777) | 0600)
stat = f.stat()
self.assertEqual(stat.st_mode & 0777, 0600)
self.assertEqual(stat.st_size, 1024)
mtime = stat.st_mtime - 3600
atime = stat.st_atime - 1800
f.utime((atime, mtime))
stat = f.stat()
self.assertEqual(stat.st_mtime, mtime)
self.assertEqual(stat.st_atime, atime)
# can't really test chown, since we'd have to know a valid uid.
f.truncate(512)
stat = f.stat()
self.assertEqual(stat.st_size, 512)
f.close()
finally:
sftp.remove(FOLDER + '/special')
def test_A_readline_seek(self):
"""
create a text file and write a bunch of text into it. then count the lines
in the file, and seek around to retreive particular lines. this should
@ -324,7 +364,7 @@ class SFTPTest (unittest.TestCase):
finally:
sftp.remove(FOLDER + '/duck.txt')
def test_A_write_seek(self):
def test_B_write_seek(self):
"""
create a text file, seek back and change part of it, and verify that the
changes worked.
@ -344,7 +384,7 @@ class SFTPTest (unittest.TestCase):
finally:
sftp.remove(FOLDER + '/testing.txt')
def test_B_symlink(self):
def test_C_symlink(self):
"""
create a symlink and then check that lstat doesn't follow it.
"""
@ -387,7 +427,7 @@ class SFTPTest (unittest.TestCase):
except:
pass
def test_C_flush_seek(self):
def test_D_flush_seek(self):
"""
verify that buffered writes are automatically flushed on seek.
"""
@ -409,7 +449,7 @@ class SFTPTest (unittest.TestCase):
except:
pass
def test_D_lots_of_files(self):
def test_E_lots_of_files(self):
"""
create a bunch of files over the same session.
"""
@ -440,7 +480,7 @@ class SFTPTest (unittest.TestCase):
except:
pass
def test_E_big_file(self):
def test_F_big_file(self):
"""
write a 1MB file with no buffering.
"""
@ -474,7 +514,7 @@ class SFTPTest (unittest.TestCase):
finally:
sftp.remove('%s/hongry.txt' % FOLDER)
def test_F_big_file_pipelined(self):
def test_G_big_file_pipelined(self):
"""
write a 1MB file, with no linefeeds, using pipelining.
"""
@ -510,7 +550,7 @@ class SFTPTest (unittest.TestCase):
finally:
sftp.remove('%s/hongry.txt' % FOLDER)
def test_G_lots_of_prefetching(self):
def test_H_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.
@ -546,7 +586,7 @@ class SFTPTest (unittest.TestCase):
finally:
sftp.remove('%s/hongry.txt' % FOLDER)
def test_H_big_file_big_buffer(self):
def test_I_big_file_big_buffer(self):
"""
write a 1MB file, with no linefeeds, and a big buffer.
"""
@ -563,7 +603,7 @@ class SFTPTest (unittest.TestCase):
finally:
sftp.remove('%s/hongry.txt' % FOLDER)
def test_I_big_file_renegotiate(self):
def test_J_big_file_renegotiate(self):
"""
write a 1MB file, forcing key renegotiation in the middle.
"""
@ -585,7 +625,7 @@ class SFTPTest (unittest.TestCase):
sftp.remove('%s/hongry.txt' % FOLDER)
t.packetizer.REKEY_BYTES = pow(2, 30)
def test_J_realpath(self):
def test_K_realpath(self):
"""
test that realpath is returning something non-empty and not an
error.
@ -596,7 +636,7 @@ class SFTPTest (unittest.TestCase):
self.assert_(len(f) > 0)
self.assertEquals(os.path.join(pwd, FOLDER), f)
def test_K_mkdir(self):
def test_L_mkdir(self):
"""
verify that mkdir/rmdir work.
"""
@ -619,7 +659,7 @@ class SFTPTest (unittest.TestCase):
except IOError:
pass
def test_L_chdir(self):
def test_M_chdir(self):
"""
verify that chdir/getcwd work.
"""
@ -656,7 +696,7 @@ class SFTPTest (unittest.TestCase):
except:
pass
def test_M_get_put(self):
def test_N_get_put(self):
"""
verify that get/put work.
"""
@ -685,7 +725,7 @@ class SFTPTest (unittest.TestCase):
os.unlink(localname)
sftp.unlink(FOLDER + '/bunny.txt')
def test_N_check(self):
def test_O_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
@ -707,7 +747,7 @@ class SFTPTest (unittest.TestCase):
finally:
sftp.unlink(FOLDER + '/kitty.txt')
def test_O_x_flag(self):
def test_P_x_flag(self):
"""
verify that the 'x' flag works when opening a file.
"""
@ -723,7 +763,7 @@ class SFTPTest (unittest.TestCase):
finally:
sftp.unlink(FOLDER + '/unusual.txt')
def test_P_utf8(self):
def test_Q_utf8(self):
"""
verify that unicode strings are encoded into utf8 correctly.
"""