/brz/remove-bazaar

# Copyright (C) 2004, 2005, 2006 by 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

"""Tests for Transport implementations.

Transport implementations tested here are supplied by

TransportTestProviderAdapter.

"""

import os

from cStringIO import StringIO

import stat

import sys

from bzrlib import (

    osutils,

    urlutils,

    )

from bzrlib.errors import (DirectoryNotEmpty, NoSuchFile, FileExists,

                           LockError, PathError,

                           TransportNotPossible, ConnectionError,

                           InvalidURL)

from bzrlib.osutils import getcwd

from bzrlib.tests import TestCaseInTempDir, TestSkipped

from bzrlib.tests.test_transport import TestTransportImplementation

from bzrlib.transport import memory

import bzrlib.transport

def _append(fn, txt):

    """Append the given text (file-like object) to the supplied filename."""

    f = open(fn, 'ab')

    try:

        f.write(txt.read())

    finally:

        f.close()

class TransportTests(TestTransportImplementation):

    def check_transport_contents(self, content, transport, relpath):

        """Check that transport.get(relpath).read() == content."""

        self.assertEqualDiff(content, transport.get(relpath).read())

    def assertListRaises(self, excClass, func, *args, **kwargs):

        """Fail unless excClass is raised when the iterator from func is used.

        Many transport functions can return generators this makes sure

        to wrap them in a list() call to make sure the whole generator

        is run, and that the proper exception is raised.

        """

        try:

            list(func(*args, **kwargs))

        except excClass:

            return

        else:

            if hasattr(excClass,'__name__'): excName = excClass.__name__

            else: excName = str(excClass)

            raise self.failureException, "%s not raised" % excName

    def test_has(self):

        t = self.get_transport()

        files = ['a', 'b', 'e', 'g', '%']

        self.build_tree(files, transport=t)

        self.assertEqual(True, t.has('a'))

        self.assertEqual(False, t.has('c'))

        self.assertEqual(True, t.has(urlutils.escape('%')))

        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),

                [True, True, False, False, True, False, True, False])

        self.assertEqual(True, t.has_any(['a', 'b', 'c']))

        self.assertEqual(False, t.has_any(['c', 'd', 'f', urlutils.escape('%%')]))

        self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),

                [True, True, False, False, True, False, True, False])

        self.assertEqual(False, t.has_any(['c', 'c', 'c']))

        self.assertEqual(True, t.has_any(['b', 'b', 'b']))

    def test_get(self):

        t = self.get_transport()

        files = ['a', 'b', 'e', 'g']

        contents = ['contents of a\n',

                    'contents of b\n',

                    'contents of e\n',

                    'contents of g\n',

                    ]

        self.build_tree(files, transport=t, line_endings='binary')

        self.check_transport_contents('contents of a\n', t, 'a')

        content_f = t.get_multi(files)

        for content, f in zip(contents, content_f):

            self.assertEqual(content, f.read())

        content_f = t.get_multi(iter(files))

        for content, f in zip(contents, content_f):

            self.assertEqual(content, f.read())

        self.assertRaises(NoSuchFile, t.get, 'c')

        self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])

        self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))

    def test_put(self):

        t = self.get_transport()

        if t.is_readonly():

            self.assertRaises(TransportNotPossible,

                    t.put, 'a', StringIO('some text for a\n'))

            return

        t.put('a', StringIO('some text for a\n'))

        self.failUnless(t.has('a'))

        self.check_transport_contents('some text for a\n', t, 'a')

        # Make sure 'has' is updated

        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),

                [True, False, False, False, False])

        # Put also replaces contents

        self.assertEqual(t.put_multi([('a', StringIO('new\ncontents for\na\n')),

                                      ('d', StringIO('contents\nfor d\n'))]),

                         2)

        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),

                [True, False, False, True, False])

        self.check_transport_contents('new\ncontents for\na\n', t, 'a')

        self.check_transport_contents('contents\nfor d\n', t, 'd')

        self.assertEqual(

            t.put_multi(iter([('a', StringIO('diff\ncontents for\na\n')),

                              ('d', StringIO('another contents\nfor d\n'))])),

                        2)

        self.check_transport_contents('diff\ncontents for\na\n', t, 'a')

        self.check_transport_contents('another contents\nfor d\n', t, 'd')

        self.assertRaises(NoSuchFile,

                          t.put, 'path/doesnt/exist/c', StringIO('contents'))

    def test_put_bytes(self):

        t = self.get_transport()

        if t.is_readonly():

            self.assertRaises(TransportNotPossible,

                    t.put_bytes, 'a', 'some text for a\n')

            return

        t.put_bytes('a', 'some text for a\n')

        self.failUnless(t.has('a'))

        self.check_transport_contents('some text for a\n', t, 'a')

        # The contents should be overwritten

        t.put_bytes('a', 'new text for a\n')

        self.check_transport_contents('new text for a\n', t, 'a')

        self.assertRaises(NoSuchFile,

                          t.put_bytes, 'path/doesnt/exist/c', 'contents')

    def test_put_permissions(self):

        t = self.get_transport()

        if t.is_readonly():

            return

        if not t._can_roundtrip_unix_modebits():

            # Can't roundtrip, so no need to run this test

            return

        t.put('mode644', StringIO('test text\n'), mode=0644)

        self.assertTransportMode(t, 'mode644', 0644)

        t.put('mode666', StringIO('test text\n'), mode=0666)

        self.assertTransportMode(t, 'mode666', 0666)

        t.put('mode600', StringIO('test text\n'), mode=0600)

        self.assertTransportMode(t, 'mode600', 0600)

        # Yes, you can put a file such that it becomes readonly

        t.put('mode400', StringIO('test text\n'), mode=0400)

        self.assertTransportMode(t, 'mode400', 0400)

        t.put_multi([('mmode644', StringIO('text\n'))], mode=0644)

        self.assertTransportMode(t, 'mmode644', 0644)

        # The default permissions should be based on the current umask

        umask = osutils.get_umask()

        t.put('nomode', StringIO('test text\n'), mode=None)

        self.assertTransportMode(t, 'nomode', 0666 & ~umask)

    def test_put_bytes_permissions(self):

        t = self.get_transport()

        if t.is_readonly():

            return

        if not t._can_roundtrip_unix_modebits():

            # Can't roundtrip, so no need to run this test

            return

        t.put_bytes('mode644', 'test text\n', mode=0644)

        self.assertTransportMode(t, 'mode644', 0644)

        t.put_bytes('mode666', 'test text\n', mode=0666)

        self.assertTransportMode(t, 'mode666', 0666)

        t.put_bytes('mode600', 'test text\n', mode=0600)

        self.assertTransportMode(t, 'mode600', 0600)

        # Yes, you can put_bytes a file such that it becomes readonly

        t.put_bytes('mode400', 'test text\n', mode=0400)

        self.assertTransportMode(t, 'mode400', 0400)

        # The default permissions should be based on the current umask

        umask = osutils.get_umask()

        t.put_bytes('nomode', 'test text\n', mode=None)

        self.assertTransportMode(t, 'nomode', 0666 & ~umask)

    def test_mkdir(self):

        t = self.get_transport()

        if t.is_readonly():

            # cannot mkdir on readonly transports. We're not testing for 

            # cache coherency because cache behaviour is not currently

            # defined for the transport interface.

            self.assertRaises(TransportNotPossible, t.mkdir, '.')

            self.assertRaises(TransportNotPossible, t.mkdir, 'new_dir')

            self.assertRaises(TransportNotPossible, t.mkdir_multi, ['new_dir'])

            self.assertRaises(TransportNotPossible, t.mkdir, 'path/doesnt/exist')

            return

        # Test mkdir

        t.mkdir('dir_a')

        self.assertEqual(t.has('dir_a'), True)

        self.assertEqual(t.has('dir_b'), False)

        t.mkdir('dir_b')

        self.assertEqual(t.has('dir_b'), True)

        t.mkdir_multi(['dir_c', 'dir_d'])

        t.mkdir_multi(iter(['dir_e', 'dir_f']))

        self.assertEqual(list(t.has_multi(

            ['dir_a', 'dir_b', 'dir_c', 'dir_q',

             'dir_d', 'dir_e', 'dir_f', 'dir_b'])),

            [True, True, True, False,

             True, True, True, True])

        # we were testing that a local mkdir followed by a transport

        # mkdir failed thusly, but given that we * in one process * do not

        # concurrently fiddle with disk dirs and then use transport to do 

        # things, the win here seems marginal compared to the constraint on

        # the interface. RBC 20051227

        t.mkdir('dir_g')

        self.assertRaises(FileExists, t.mkdir, 'dir_g')

        # Test get/put in sub-directories

        self.assertEqual(2, 

            t.put_multi([('dir_a/a', StringIO('contents of dir_a/a')),

                         ('dir_b/b', StringIO('contents of dir_b/b'))]))

        self.check_transport_contents('contents of dir_a/a', t, 'dir_a/a')

        self.check_transport_contents('contents of dir_b/b', t, 'dir_b/b')

        # mkdir of a dir with an absent parent

        self.assertRaises(NoSuchFile, t.mkdir, 'missing/dir')

    def test_mkdir_permissions(self):

        t = self.get_transport()

        if t.is_readonly():

            return

        if not t._can_roundtrip_unix_modebits():

            # no sense testing on this transport

            return

        # Test mkdir with a mode

        t.mkdir('dmode755', mode=0755)

        self.assertTransportMode(t, 'dmode755', 0755)

        t.mkdir('dmode555', mode=0555)

        self.assertTransportMode(t, 'dmode555', 0555)

        t.mkdir('dmode777', mode=0777)

        self.assertTransportMode(t, 'dmode777', 0777)

        t.mkdir('dmode700', mode=0700)

        self.assertTransportMode(t, 'dmode700', 0700)

        t.mkdir_multi(['mdmode755'], mode=0755)

        self.assertTransportMode(t, 'mdmode755', 0755)

        # Default mode should be based on umask

        umask = osutils.get_umask()

        t.mkdir('dnomode', mode=None)

        self.assertTransportMode(t, 'dnomode', 0777 & ~umask)

    def test_copy_to(self):

        # FIXME: test:   same server to same server (partly done)

        # same protocol two servers

        # and    different protocols (done for now except for MemoryTransport.

        # - RBC 20060122

        from bzrlib.transport.memory import MemoryTransport

        def simple_copy_files(transport_from, transport_to):

            files = ['a', 'b', 'c', 'd']

            self.build_tree(files, transport=transport_from)

            self.assertEqual(4, transport_from.copy_to(files, transport_to))

            for f in files:

                self.check_transport_contents(transport_to.get(f).read(),

                                              transport_from, f)

        t = self.get_transport()

        temp_transport = MemoryTransport('memory:///')

        simple_copy_files(t, temp_transport)

        if not t.is_readonly():

            t.mkdir('copy_to_simple')

            t2 = t.clone('copy_to_simple')

            simple_copy_files(t, t2)

        # Test that copying into a missing directory raises

        # NoSuchFile

        if t.is_readonly():

            self.build_tree(['e/', 'e/f'])

        else:

            t.mkdir('e')

            t.put('e/f', StringIO('contents of e'))

        self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], temp_transport)

        temp_transport.mkdir('e')

        t.copy_to(['e/f'], temp_transport)

        del temp_transport

        temp_transport = MemoryTransport('memory:///')

        files = ['a', 'b', 'c', 'd']

        t.copy_to(iter(files), temp_transport)

        for f in files:

            self.check_transport_contents(temp_transport.get(f).read(),

                                          t, f)

        del temp_transport

        for mode in (0666, 0644, 0600, 0400):

            temp_transport = MemoryTransport("memory:///")

            t.copy_to(files, temp_transport, mode=mode)

            for f in files:

                self.assertTransportMode(temp_transport, f, mode)

    def test_append(self):

        t = self.get_transport()

        if t.is_readonly():

            self.assertRaises(TransportNotPossible,

                    t.append, 'a', 'add\nsome\nmore\ncontents\n')

            return

        t.put_multi([

                ('a', StringIO('diff\ncontents for\na\n')),

                ('b', StringIO('contents\nfor b\n'))

                ])

        self.assertEqual(20,

            t.append('a', StringIO('add\nsome\nmore\ncontents\n')))

        self.check_transport_contents(

            'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',

            t, 'a')

        self.assertEqual((43, 15),

            t.append_multi([('a', StringIO('and\nthen\nsome\nmore\n')),

                            ('b', StringIO('some\nmore\nfor\nb\n'))]))

        self.check_transport_contents(

            'diff\ncontents for\na\n'

            'add\nsome\nmore\ncontents\n'

            'and\nthen\nsome\nmore\n',

            t, 'a')

        self.check_transport_contents(

                'contents\nfor b\n'

                'some\nmore\nfor\nb\n',

                t, 'b')

        self.assertEqual((62, 31),

            t.append_multi(iter([('a', StringIO('a little bit more\n')),

                                 ('b', StringIO('from an iterator\n'))])))

        self.check_transport_contents(

            'diff\ncontents for\na\n'

            'add\nsome\nmore\ncontents\n'

            'and\nthen\nsome\nmore\n'

            'a little bit more\n',

            t, 'a')

        self.check_transport_contents(

                'contents\nfor b\n'

                'some\nmore\nfor\nb\n'

                'from an iterator\n',

                t, 'b')

        self.assertEqual(0,

            t.append('c', StringIO('some text\nfor a missing file\n')))

        self.assertEqual((80, 0),

            t.append_multi([('a', StringIO('some text in a\n')),

                            ('d', StringIO('missing file r\n'))]))

        self.check_transport_contents(

            'diff\ncontents for\na\n'

            'add\nsome\nmore\ncontents\n'

            'and\nthen\nsome\nmore\n'

            'a little bit more\n'

            'some text in a\n',

            t, 'a')

        self.check_transport_contents('some text\nfor a missing file\n',

                                      t, 'c')

        self.check_transport_contents('missing file r\n', t, 'd')

        # a file with no parent should fail..

        self.assertRaises(NoSuchFile,

                          t.append, 'missing/path',

                          StringIO('content'))

    def test_append_bytes(self):

        t = self.get_transport()

        if t.is_readonly():

            self.assertRaises(TransportNotPossible,

                    t.append, 'a', 'add\nsome\nmore\ncontents\n')

            return

        self.assertEqual(0, t.append_bytes('a', 'diff\ncontents for\na\n'))

        self.assertEqual(0, t.append_bytes('b', 'contents\nfor b\n'))

        self.assertEqual(20,

            t.append_bytes('a', 'add\nsome\nmore\ncontents\n'))

        self.check_transport_contents(

            'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',

            t, 'a')

        # a file with no parent should fail..

        self.assertRaises(NoSuchFile,

                          t.append_bytes, 'missing/path', 'content')

    def test_append_mode(self):

        # check append accepts a mode

        t = self.get_transport()

        if t.is_readonly():

            return

        t.append('f', StringIO('f'), mode=None)

    def test_append_bytes_mode(self):

        # check append_bytes accepts a mode

        t = self.get_transport()

        if t.is_readonly():

            return

        t.append('f', StringIO('f'), mode=None)

    def test_delete(self):

        # TODO: Test Transport.delete

        t = self.get_transport()

        # Not much to do with a readonly transport

        if t.is_readonly():

            self.assertRaises(TransportNotPossible, t.delete, 'missing')

            return

        t.put('a', StringIO('a little bit of text\n'))

        self.failUnless(t.has('a'))

        t.delete('a')

        self.failIf(t.has('a'))

        self.assertRaises(NoSuchFile, t.delete, 'a')

        t.put('a', StringIO('a text\n'))

        t.put('b', StringIO('b text\n'))

        t.put('c', StringIO('c text\n'))

        self.assertEqual([True, True, True],

                list(t.has_multi(['a', 'b', 'c'])))

        t.delete_multi(['a', 'c'])

        self.assertEqual([False, True, False],

                list(t.has_multi(['a', 'b', 'c'])))

        self.failIf(t.has('a'))

        self.failUnless(t.has('b'))

        self.failIf(t.has('c'))

        self.assertRaises(NoSuchFile,

                t.delete_multi, ['a', 'b', 'c'])

        self.assertRaises(NoSuchFile,

                t.delete_multi, iter(['a', 'b', 'c']))

        t.put('a', StringIO('another a text\n'))

        t.put('c', StringIO('another c text\n'))

        t.delete_multi(iter(['a', 'b', 'c']))

        # We should have deleted everything

        # SftpServer creates control files in the

        # working directory, so we can just do a

        # plain "listdir".

        # self.assertEqual([], os.listdir('.'))

    def test_rmdir(self):

        t = self.get_transport()

        # Not much to do with a readonly transport

        if t.is_readonly():

            self.assertRaises(TransportNotPossible, t.rmdir, 'missing')

            return

        t.mkdir('adir')

        t.mkdir('adir/bdir')

        t.rmdir('adir/bdir')

        self.assertRaises(NoSuchFile, t.stat, 'adir/bdir')

        t.rmdir('adir')

        self.assertRaises(NoSuchFile, t.stat, 'adir')

    def test_rmdir_not_empty(self):

        """Deleting a non-empty directory raises an exception

        sftp (and possibly others) don't give us a specific "directory not

        empty" exception -- we can just see that the operation failed.

        """

        t = self.get_transport()

        if t.is_readonly():

            return

        t.mkdir('adir')

        t.mkdir('adir/bdir')

        self.assertRaises(PathError, t.rmdir, 'adir')

    def test_rename_dir_succeeds(self):

        t = self.get_transport()

        if t.is_readonly():

            raise TestSkipped("transport is readonly")

        t.mkdir('adir')

        t.mkdir('adir/asubdir')

        t.rename('adir', 'bdir')

        self.assertTrue(t.has('bdir/asubdir'))

        self.assertFalse(t.has('adir'))

    def test_rename_dir_nonempty(self):

        """Attempting to replace a nonemtpy directory should fail"""

        t = self.get_transport()

        if t.is_readonly():

            raise TestSkipped("transport is readonly")

        t.mkdir('adir')

        t.mkdir('adir/asubdir')

        t.mkdir('bdir')

        t.mkdir('bdir/bsubdir')

        self.assertRaises(PathError, t.rename, 'bdir', 'adir')

        # nothing was changed so it should still be as before

        self.assertTrue(t.has('bdir/bsubdir'))

        self.assertFalse(t.has('adir/bdir'))

        self.assertFalse(t.has('adir/bsubdir'))

    def test_delete_tree(self):

        t = self.get_transport()

        # Not much to do with a readonly transport

        if t.is_readonly():

            self.assertRaises(TransportNotPossible, t.delete_tree, 'missing')

            return

        # and does it like listing ?

        t.mkdir('adir')

        try:

            t.delete_tree('adir')

        except TransportNotPossible:

            # ok, this transport does not support delete_tree

            return

        # did it delete that trivial case?

        self.assertRaises(NoSuchFile, t.stat, 'adir')

        self.build_tree(['adir/',

                         'adir/file', 

                         'adir/subdir/', 

                         'adir/subdir/file', 

                         'adir/subdir2/',

                         'adir/subdir2/file',

                         ], transport=t)

        t.delete_tree('adir')

        # adir should be gone now.

        self.assertRaises(NoSuchFile, t.stat, 'adir')

    def test_move(self):

        t = self.get_transport()

        if t.is_readonly():

            return

        # TODO: I would like to use os.listdir() to

        # make sure there are no extra files, but SftpServer

        # creates control files in the working directory

        # perhaps all of this could be done in a subdirectory

        t.put('a', StringIO('a first file\n'))

        self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))

        t.move('a', 'b')

        self.failUnless(t.has('b'))

        self.failIf(t.has('a'))

        self.check_transport_contents('a first file\n', t, 'b')

        self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))

        # Overwrite a file

        t.put('c', StringIO('c this file\n'))

        t.move('c', 'b')

        self.failIf(t.has('c'))

        self.check_transport_contents('c this file\n', t, 'b')

        # TODO: Try to write a test for atomicity

        # TODO: Test moving into a non-existant subdirectory

        # TODO: Test Transport.move_multi

    def test_copy(self):

        t = self.get_transport()

        if t.is_readonly():

            return

        t.put('a', StringIO('a file\n'))

        t.copy('a', 'b')

        self.check_transport_contents('a file\n', t, 'b')

        self.assertRaises(NoSuchFile, t.copy, 'c', 'd')

        os.mkdir('c')

        # What should the assert be if you try to copy a

        # file over a directory?

        #self.assertRaises(Something, t.copy, 'a', 'c')

        t.put('d', StringIO('text in d\n'))

        t.copy('d', 'b')

        self.check_transport_contents('text in d\n', t, 'b')

        # TODO: test copy_multi

    def test_connection_error(self):

        """ConnectionError is raised when connection is impossible"""

        try:

            url = self._server.get_bogus_url()

        except NotImplementedError: