Rev 2505: Refactor SFTPTransport. Test suite passes. in file:///v/home/vila/src/experimental/reuse.transports/
Vincent Ladeuil
v.ladeuil+lp at free.fr
Thu May 31 19:12:13 BST 2007
At file:///v/home/vila/src/experimental/reuse.transports/
------------------------------------------------------------
revno: 2505
revision-id: v.ladeuil+lp at free.fr-20070531181210-jcwx0x2t1wqathbv
parent: v.ladeuil+lp at free.fr-20070531180111-xef34xmn8l3hjyz0
committer: Vincent Ladeuil <v.ladeuil+lp at free.fr>
branch nick: reuse.transports
timestamp: Thu 2007-05-31 20:12:10 +0200
message:
Refactor SFTPTransport. Test suite passes.
* bzrlib/transport/sftp.py:
(SFTPUrlHandling): Simplified.
(SFTPTransport._remote_path): Deleted. SFTPUrlHandling now has a
simpler version.
* bzrlib/tests/test_sftp_transport.py:
Refactor the tests, most of them have gone to TestConnectedTransport.
(SFTPNonServerTest.test_parse_url_with_home_dir): Renamed from
test_parse_url and simplified to keep only specific tests.
modified:
bzrlib/tests/test_sftp_transport.py testsftp.py-20051027032739-247570325fec7e7e
bzrlib/transport/sftp.py sftp.py-20051019050329-ab48ce71b7e32dfe
-------------- next part --------------
=== modified file 'bzrlib/tests/test_sftp_transport.py'
--- a/bzrlib/tests/test_sftp_transport.py 2007-03-28 08:49:06 +0000
+++ b/bzrlib/tests/test_sftp_transport.py 2007-05-31 18:12:10 +0000
@@ -21,14 +21,28 @@
import threading
import time
-import bzrlib.bzrdir as bzrdir
-import bzrlib.errors as errors
-from bzrlib.osutils import pathjoin, lexists, set_or_unset_env
-from bzrlib.tests import TestCaseWithTransport, TestCase, TestSkipped
+from bzrlib import (
+ bzrdir,
+ errors,
+ )
+from bzrlib.osutils import (
+ pathjoin,
+ lexists,
+ set_or_unset_env,
+ )
+from bzrlib.tests import (
+ TestCaseWithTransport,
+ TestCase,
+ TestSkipped,
+ )
from bzrlib.tests.HttpServer import HttpServer
-import bzrlib.transport
from bzrlib.transport import get_transport
import bzrlib.transport.http
+from bzrlib.transport.sftp import (
+ SFTPAbsoluteServer,
+ SFTPHomeDirServer,
+ SFTPTransport,
+ )
from bzrlib.workingtree import WorkingTree
try:
@@ -40,7 +54,6 @@
def set_test_transport_to_sftp(testcase):
"""A helper to set transports on test case instances."""
- from bzrlib.transport.sftp import SFTPAbsoluteServer, SFTPHomeDirServer
if getattr(testcase, '_get_remote_is_absolute', None) is None:
testcase._get_remote_is_absolute = True
if testcase._get_remote_is_absolute:
@@ -143,7 +156,7 @@
class FakeSFTPTransport (object):
_sftp = object()
-fake = FakeSFTPTransport()
+avoid_sftp_connection = FakeSFTPTransport()
class SFTPNonServerTest(TestCase):
@@ -152,63 +165,20 @@
if not paramiko_loaded:
raise TestSkipped('you must have paramiko to run this test')
- def test_parse_url(self):
- from bzrlib.transport.sftp import SFTPTransport
- s = SFTPTransport('sftp://simple.example.com/home/source', clone_from=fake)
- self.assertEquals(s._host, 'simple.example.com')
- self.assertEquals(s._port, None)
- self.assertEquals(s._path, '/home/source')
- self.failUnless(s._password is None)
-
- self.assertEquals(s.base, 'sftp://simple.example.com/home/source/')
-
- s = SFTPTransport('sftp://ro%62ey:h%40t@example.com:2222/~/relative', clone_from=fake)
+ def test_parse_url_with_home_dir(self):
+ s = SFTPTransport('sftp://ro%62ey:h%40t@example.com:2222/~/relative',
+ clone_from=avoid_sftp_connection)
self.assertEquals(s._host, 'example.com')
self.assertEquals(s._port, 2222)
- self.assertEquals(s._username, 'robey')
+ self.assertEquals(s._user, 'robey')
self.assertEquals(s._password, 'h at t')
- self.assertEquals(s._path, 'relative')
-
- # Base should not keep track of the password
- self.assertEquals(s.base, 'sftp://robey@example.com:2222/~/relative/')
+ self.assertEquals(s._path, 'relative/')
def test_relpath(self):
- from bzrlib.transport.sftp import SFTPTransport
- from bzrlib.errors import PathNotChild
-
- s = SFTPTransport('sftp://user@host.com/abs/path', clone_from=fake)
- self.assertEquals(s.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
- # Can't test this one, because we actually get an AssertionError
- # TODO: Consider raising an exception rather than an assert
- #self.assertRaises(PathNotChild, s.relpath, 'http://user@host.com/abs/path/sub')
- self.assertRaises(PathNotChild, s.relpath, 'sftp://user2@host.com/abs/path/sub')
- self.assertRaises(PathNotChild, s.relpath, 'sftp://user@otherhost.com/abs/path/sub')
- self.assertRaises(PathNotChild, s.relpath, 'sftp://user@host.com:33/abs/path/sub')
- self.assertRaises(PathNotChild, s.relpath, 'sftp://user@host.com/~/rel/path/sub')
-
- # Make sure it works when we don't supply a username
- s = SFTPTransport('sftp://host.com/abs/path', clone_from=fake)
- self.assertEquals(s.relpath('sftp://host.com/abs/path/sub'), 'sub')
-
- # Make sure it works when parts of the path will be url encoded
- # TODO: These may be incorrect, we might need to urllib.urlencode() before
- # we pass the paths into the SFTPTransport constructor
- s = SFTPTransport('sftp://host.com/dev/,path', clone_from=fake)
- self.assertEquals(s.relpath('sftp://host.com/dev/,path/sub'), 'sub')
- s = SFTPTransport('sftp://host.com/dev/%path', clone_from=fake)
- self.assertEquals(s.relpath('sftp://host.com/dev/%path/sub'), 'sub')
-
- def test_parse_invalid_url(self):
- from bzrlib.transport.sftp import SFTPTransport, TransportError
- try:
- s = SFTPTransport('sftp://lilypond.org:~janneke/public_html/bzr/gub',
- clone_from=fake)
- self.fail('expected exception not raised')
- except TransportError, e:
- self.assertEquals(str(e),
- 'Transport error: '
- 'invalid port number ~janneke in url:\n'
- 'sftp://lilypond.org:~janneke/public_html/bzr/gub ')
+ s = SFTPTransport('sftp://user@host.com/abs/path',
+ clone_from=avoid_sftp_connection)
+ self.assertRaises(errors.PathNotChild, s.relpath,
+ 'sftp://user@host.com/~/rel/path/sub')
def test_get_paramiko_vendor(self):
"""Test that if no 'ssh' is available we get builtin paramiko"""
=== modified file 'bzrlib/transport/sftp.py'
--- a/bzrlib/transport/sftp.py 2007-05-30 07:15:16 +0000
+++ b/bzrlib/transport/sftp.py 2007-05-31 18:12:10 +0000
@@ -53,7 +53,6 @@
local,
register_urlparse_netloc_protocol,
Server,
- split_url,
ssh,
ConnectedTransport,
)
@@ -138,64 +137,45 @@
class SFTPUrlHandling(ConnectedTransport):
"""Mix-in that does common handling of SSH/SFTP URLs."""
- def __init__(self, base):
- self._parse_url(base)
- base = self._unparse_url(self._path)
- if base[-1] != '/':
- base += '/'
- super(SFTPUrlHandling, self).__init__(base)
-
- def _parse_url(self, url):
- (self._scheme,
- self._username, self._password,
- self._host, self._port, self._path) = self._split_url(url)
-
- def _unparse_url(self, path):
- """Return a URL for a path relative to this transport.
- """
- path = urllib.quote(path)
+ def _urlencode_abspath(self, abspath):
+ abspath = super(SFTPUrlHandling, self)._urlencode_abspath(abspath)
# handle homedir paths
- if not path.startswith('/'):
- path = "/~/" + path
- netloc = urllib.quote(self._host)
- if self._username is not None:
- netloc = '%s@%s' % (urllib.quote(self._username), netloc)
- if self._port is not None:
- netloc = '%s:%d' % (netloc, self._port)
- return urlparse.urlunparse((self._scheme, netloc, path, '', '', ''))
-
- def _split_url(self, url):
- (scheme, username, password, host, port, path) = split_url(url)
- ## assert scheme == 'sftp'
-
+ if not abspath.startswith('/'):
+ abspath = "/~/" + abspath
+ return abspath
+
+ def _urldecode_abspath(self, abspath):
+ abspath = super(SFTPUrlHandling, self)._urldecode_abspath(abspath)
# the initial slash should be removed from the path, and treated
# as a homedir relative path (the path begins with a double slash
# if it is absolute).
# see draft-ietf-secsh-scp-sftp-ssh-uri-03.txt
# RBC 20060118 we are not using this as its too user hostile. instead
# we are following lftp and using /~/foo to mean '~/foo'.
+
# handle homedir paths
- if path.startswith('/~/'):
- path = path[3:]
- elif path == '/~':
- path = ''
- return (scheme, username, password, host, port, path)
+ if abspath.startswith('/~/'):
+ abspath = abspath[3:]
+ elif abspath == '/~':
+ abspath = ''
+ return abspath
- def abspath(self, relpath):
- """Return the full url to the given relative path.
-
- @param relpath: the relative path or path components
- @type relpath: str or list
- """
- return self._unparse_url(self._remote_path(relpath))
-
def _remote_path(self, relpath):
"""Return the path to be passed along the sftp protocol for relpath.
:param relpath: is a urlencoded string.
"""
- return self._combine_paths(self._path, relpath)
-
+ relative = urlutils.unescape(relpath).encode('utf-8')
+ if not self._path.startswith('/'):
+ # handle homedir paths
+ remote_path = self._combine_paths('/~/' + self._path, relative)
+ if remote_path.startswith('/~/'):
+ remote_path = remote_path[3:]
+ elif remote_path == '/~':
+ remote_path = ''
+ else:
+ remote_path = self._combine_paths(self._path, relative)
+ return remote_path
class SFTPTransport(SFTPUrlHandling):
"""Transport implementation for SFTP access."""
@@ -244,132 +224,6 @@
else:
return SFTPTransport(self.abspath(offset), self)
- def _remote_path(self, relpath):
- """Return the path to be passed along the sftp protocol for relpath.
-
- relpath is a urlencoded string.
-
- :return: a path prefixed with / for regular abspath-based urls, or a
- path that does not begin with / for urls which begin with /~/.
- """
- # how does this work?
- # it processes relpath with respect to
- # our state:
- # firstly we create a path to evaluate:
- # if relpath is an abspath or homedir path, its the entire thing
- # otherwise we join our base with relpath
- # then we eliminate all empty segments (double //'s) outside the first
- # two elements of the list. This avoids problems with trailing
- # slashes, or other abnormalities.
- # finally we evaluate the entire path in a single pass
- # '.'s are stripped,
- # '..' result in popping the left most already
- # processed path (which can never be empty because of the check for
- # abspath and homedir meaning that its not, or that we've used our
- # path. If the pop would pop the root, we ignore it.
-
- # Specific case examinations:
- # remove the special casefor ~: if the current root is ~/ popping of it
- # = / thus our seed for a ~ based path is ['', '~']
- # and if we end up with [''] then we had basically ('', '..') (which is
- # '/..' so we append '' if the length is one, and assert that the first
- # element is still ''. Lastly, if we end with ['', '~'] as a prefix for
- # the output, we've got a homedir path, so we strip that prefix before
- # '/' joining the resulting list.
- #
- # case one: '/' -> ['', ''] cannot shrink
- # case two: '/' + '../foo' -> ['', 'foo'] (take '', '', '..', 'foo')
- # and pop the second '' for the '..', append 'foo'
- # case three: '/~/' -> ['', '~', '']
- # case four: '/~/' + '../foo' -> ['', '~', '', '..', 'foo'],
- # and we want to get '/foo' - the empty path in the middle
- # needs to be stripped, then normal path manipulation will
- # work.
- # case five: '/..' ['', '..'], we want ['', '']
- # stripping '' outside the first two is ok
- # ignore .. if its too high up
- #
- # lastly this code is possibly reusable by FTP, but not reusable by
- # local paths: ~ is resolvable correctly, nor by HTTP or the smart
- # server: ~ is resolved remotely.
- #
- # however, a version of this that acts on self.base is possible to be
- # written which manipulates the URL in canonical form, and would be
- # reusable for all transports, if a flag for allowing ~/ at all was
- # provided.
- assert isinstance(relpath, basestring)
- relpath = urlutils.unescape(relpath)
-
- # case 1)
- if relpath.startswith('/'):
- # abspath - normal split is fine.
- current_path = relpath.split('/')
- elif relpath.startswith('~/'):
- # root is homedir based: normal split and prefix '' to remote the
- # special case
- current_path = [''].extend(relpath.split('/'))
- else:
- # root is from the current directory:
- if self._path.startswith('/'):
- # abspath, take the regular split
- current_path = []
- else:
- # homedir based, add the '', '~' not present in self._path
- current_path = ['', '~']
- # add our current dir
- current_path.extend(self._path.split('/'))
- # add the users relpath
- current_path.extend(relpath.split('/'))
- # strip '' segments that are not in the first one - the leading /.
- to_process = current_path[:1]
- for segment in current_path[1:]:
- if segment != '':
- to_process.append(segment)
-
- # process '.' and '..' segments into output_path.
- output_path = []
- for segment in to_process:
- if segment == '..':
- # directory pop. Remove a directory
- # as long as we are not at the root
- if len(output_path) > 1:
- output_path.pop()
- # else: pass
- # cannot pop beyond the root, so do nothing
- elif segment == '.':
- continue # strip the '.' from the output.
- else:
- # this will append '' to output_path for the root elements,
- # which is appropriate: its why we strip '' in the first pass.
- output_path.append(segment)
-
- # check output special cases:
- if output_path == ['']:
- # [''] -> ['', '']
- output_path = ['', '']
- elif output_path[:2] == ['', '~']:
- # ['', '~', ...] -> ...
- output_path = output_path[2:]
- path = '/'.join(output_path)
- return path
-
- def relpath(self, abspath):
- scheme, username, password, host, port, path = self._split_url(abspath)
- error = []
- if (username != self._username):
- error.append('username mismatch')
- if (host != self._host):
- error.append('host mismatch')
- if (port != self._port):
- error.append('port mismatch')
- if (not path.startswith(self._path)):
- error.append('path mismatch')
- if error:
- extra = ': ' + ', '.join(error)
- raise PathNotChild(abspath, self.base, extra=extra)
- pl = len(self._path)
- return path[pl:].strip('/')
-
def has(self, relpath):
"""
Does the target location exist?
@@ -832,8 +686,9 @@
TODO: Raise a more reasonable ConnectionFailed exception
"""
- self._sftp = _sftp_connect(self._host, self._port, self._username,
- self._password)
+ self._sftp = _sftp_connect(self._host, self._port,
+ self._user,
+ self._password)
def _sftp_open_exclusive(self, abspath, mode=None):
"""Open a remote path exclusively.
More information about the bazaar-commits
mailing list