# Copyright (C) 2005 Robey Pointer , Canonical Ltd # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # This program is distributed 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 General Public License for more details. # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA import os import socket import threading from bzrlib.tests import TestCaseInTempDir, TestCase from bzrlib.tests.test_transport import TestTransportMixIn import bzrlib.errors as errors try: import paramiko from stub_sftp import StubServer, StubSFTPServer paramiko_loaded = True except ImportError: paramiko_loaded = False # XXX: 20051124 jamesh # The tests currently pop up a password prompt when an external ssh # is used. This forces the use of the paramiko implementation. if paramiko_loaded: import bzrlib.transport.sftp bzrlib.transport.sftp._ssh_vendor = 'none' STUB_SERVER_KEY = """ -----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----- """ class SingleListener (threading.Thread): def __init__(self, callback): threading.Thread.__init__(self) self._callback = callback self._socket = socket.socket() self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self._socket.bind(('localhost', 0)) self._socket.listen(1) self.port = self._socket.getsockname()[1] self.stop_event = threading.Event() def run(self): s, _ = self._socket.accept() # now close the listen socket self._socket.close() self._callback(s, self.stop_event) def stop(self): self.stop_event.set() # We should consider waiting for the other thread # to stop, because otherwise we get spurious # bzr: ERROR: Socket exception: Connection reset by peer (54) # because the test suite finishes before the thread has a chance # to close. (Especially when only running a few tests) class TestCaseWithSFTPServer (TestCaseInTempDir): """ Execute a test case with a stub SFTP server, serving files from the local filesystem over the loopback network. """ def _run_server(self, s, stop_event): ssh_server = paramiko.Transport(s) key_file = os.path.join(self._root, 'test_rsa.key') file(key_file, 'w').write(STUB_SERVER_KEY) host_key = paramiko.RSAKey.from_private_key_file(key_file) ssh_server.add_server_key(host_key) server = StubServer(self) ssh_server.set_subsystem_handler('sftp', paramiko.SFTPServer, StubSFTPServer, root=self._root) event = threading.Event() ssh_server.start_server(event, server) event.wait(5.0) stop_event.wait(30.0) def setUp(self): TestCaseInTempDir.setUp(self) self._root = self.test_dir self._is_setup = False def delayed_setup(self): # some tests are just stubs that call setUp and then immediately call # tearDwon. so don't create the port listener until get_transport is # called and we know we're in an actual test. if self._is_setup: return self._listener = SingleListener(self._run_server) self._listener.setDaemon(True) self._listener.start() self._sftp_url = 'sftp://foo:bar@localhost:%d/' % (self._listener.port,) self._is_setup = True def tearDown(self): try: self._listener.stop() except AttributeError: pass TestCaseInTempDir.tearDown(self) class SFTPTransportTest (TestCaseWithSFTPServer, TestTransportMixIn): readonly = False def setUp(self): TestCaseWithSFTPServer.setUp(self) self.sftplogs = [] def log(self, *args): """Override the default log to grab sftp server messages""" TestCaseWithSFTPServer.log(self, *args) if args and args[0].startswith('sftpserver'): self.sftplogs.append(args[0]) def get_transport(self): self.delayed_setup() from bzrlib.transport.sftp import SFTPTransport url = self._sftp_url return SFTPTransport(url) def test_sftp_locks(self): from bzrlib.errors import LockError t = self.get_transport() l = t.lock_write('bogus') self.failUnlessExists('bogus.write-lock') # Don't wait for the lock, locking an already locked # file should raise an assert self.assertRaises(LockError, t.lock_write, 'bogus') l.unlock() self.failIf(os.path.lexists('bogus.write-lock')) open('something.write-lock', 'wb').write('fake lock\n') self.assertRaises(LockError, t.lock_write, 'something') os.remove('something.write-lock') l = t.lock_write('something') l2 = t.lock_write('bogus') l.unlock() l2.unlock() def test_multiple_connections(self): t = self.get_transport() self.assertEquals(self.sftplogs, ['sftpserver - authorizing: foo' , 'sftpserver - channel request: session, 1']) self.sftplogs = [] # The second request should reuse the first connection # SingleListener only allows for a single connection, # So the next line fails unless the connection is reused t2 = self.get_transport() self.assertEquals(self.sftplogs, []) class FakeSFTPTransport (object): _sftp = object() fake = FakeSFTPTransport() class SFTPNonServerTest(TestCase): def test_parse_url(self): from bzrlib.transport.sftp import SFTPTransport s = SFTPTransport('sftp://simple.example.com/%2fhome/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/%2Fhome/source') s = SFTPTransport('sftp://ro%62ey:h%40t@example.com:2222/relative', clone_from=fake) self.assertEquals(s._host, 'example.com') self.assertEquals(s._port, 2222) self.assertEquals(s._username, 'robey') self.assertEquals(s._password, 'h@t') self.assertEquals(s._path, 'relative') # Base should not keep track of the password self.assertEquals(s.base, 'sftp://robey@example.com:2222/relative') # Double slash should be accepted instead of using %2F s = SFTPTransport('sftp://user@example.com:22//absolute/path', clone_from=fake) self.assertEquals(s._host, 'example.com') self.assertEquals(s._port, 22) self.assertEquals(s._username, 'user') self.assertEquals(s._password, None) self.assertEquals(s._path, '/absolute/path') # Also, don't show the port if it is the default 22 self.assertEquals(s.base, 'sftp://user@example.com:22/%2Fabsolute/path') 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/abs/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), '~janneke: invalid port number') class SFTPBranchTest(TestCaseWithSFTPServer): """Test some stuff when accessing a bzr Branch over sftp""" def test_lock_file(self): """Make sure that a Branch accessed over sftp tries to lock itself.""" from bzrlib.branch import Branch self.delayed_setup() b = Branch.initialize(self._sftp_url) self.failUnlessExists('.bzr/') self.failUnlessExists('.bzr/branch-format') self.failUnlessExists('.bzr/branch-lock') self.failIf(os.path.lexists('.bzr/branch-lock.write-lock')) b.lock_write() self.failUnlessExists('.bzr/branch-lock.write-lock') b.unlock() self.failIf(os.path.lexists('.bzr/branch-lock.write-lock')) def test_no_working_tree(self): from bzrlib.branch import Branch self.delayed_setup() b = Branch.initialize(self._sftp_url) self.assertRaises(errors.NoWorkingTree, b.working_tree) def test_push_support(self): from bzrlib.branch import Branch self.delayed_setup() self.build_tree(['a/', 'a/foo']) b = Branch.initialize('a') t = b.working_tree() t.add('foo') t.commit('foo', rev_id='a1') os.mkdir('b') b2 = Branch.initialize(self._sftp_url + 'b') b2.pull(b) self.assertEquals(b2.revision_history(), ['a1']) if not paramiko_loaded: # TODO: Skip these del SFTPTransportTest del SFTPNonServerTest del SFTPBranchTest