~bzr-pqm/bzr/bzr.dev

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


import os
from cStringIO import StringIO

from bzrlib.errors import (NoSuchFile, FileExists, TransportNotPossible,
                           ConnectionError)
from bzrlib.tests import TestCase, TestCaseInTempDir
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
from bzrlib.transport import memory, urlescape


def _append(fn, txt):
    """Append the given text (file-like object) to the supplied filename."""
    f = open(fn, 'ab')
    f.write(txt)
    f.flush()
    f.close()
    del f

class TestTransport(TestCase):
    """Test the non transport-concrete class functionality."""

    def test_urlescape(self):
        self.assertEqual('%25', urlescape('%'))


class TestTransportMixIn(object):
    """Subclass this, and it will provide a series of tests for a Transport.
    It assumes that the Transport object is connected to the 
    current working directory.  So that whatever is done 
    through the transport, should show up in the working 
    directory, and vice-versa.

    This also tests to make sure that the functions work with both
    generators and lists (assuming iter(list) is effectively a generator)
    """
    readonly = False
    def get_transport(self):
        """Children should override this to return the Transport object.
        """
        raise NotImplementedError

    def test_has(self):
        t = self.get_transport()

        files = ['a', 'b', 'e', 'g', '%']
        self.build_tree(files)
        self.assertEqual(t.has('a'), True)
        self.assertEqual(t.has('c'), False)
        self.assertEqual(t.has(urlescape('%')), True)
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
                [True, True, False, False, True, False, True, False])
        self.assertEqual(t.has_any(['a', 'b', 'c']), True)
        self.assertEqual(t.has_any(['c', 'd', 'f', urlescape('%%')]), False)
        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(t.has_any(['c', 'c', 'c']), False)
        self.assertEqual(t.has_any(['b', 'b', 'b']), True)

    def test_get(self):
        t = self.get_transport()

        files = ['a', 'b', 'e', 'g']
        self.build_tree(files)
        self.assertEqual(t.get('a').read(), open('a').read())
        content_f = t.get_multi(files)
        for path,f in zip(files, content_f):
            self.assertEqual(open(path).read(), f.read())

        content_f = t.get_multi(iter(files))
        for path,f in zip(files, content_f):
            self.assertEqual(open(path).read(), f.read())

        self.assertRaises(NoSuchFile, t.get, 'c')
        try:
            files = list(t.get_multi(['a', 'b', 'c']))
        except NoSuchFile:
            pass
        else:
            self.fail('Failed to raise NoSuchFile for missing file in get_multi')
        try:
            files = list(t.get_multi(iter(['a', 'b', 'c', 'e'])))
        except NoSuchFile:
            pass
        else:
            self.fail('Failed to raise NoSuchFile for missing file in get_multi')

    def test_put(self):
        t = self.get_transport()

        if self.readonly:
            self.assertRaises(TransportNotPossible,
                    t.put, 'a', 'some text for a\n')
            open('a', 'wb').write('some text for a\n')
        else:
            t.put('a', 'some text for a\n')
        self.assert_(os.path.exists('a'))
        self.check_file_contents('a', 'some text for a\n')
        self.assertEqual(t.get('a').read(), 'some text for a\n')
        # Make sure 'has' is updated
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
                [True, False, False, False, False])
        if self.readonly:
            self.assertRaises(TransportNotPossible,
                    t.put_multi,
                    [('a', 'new\ncontents for\na\n'),
                        ('d', 'contents\nfor d\n')])
            open('a', 'wb').write('new\ncontents for\na\n')
            open('d', 'wb').write('contents\nfor d\n')
        else:
            # Put also replaces contents
            self.assertEqual(t.put_multi([('a', 'new\ncontents for\na\n'),
                                          ('d', 'contents\nfor d\n')]),
                             2)
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
                [True, False, False, True, False])
        self.check_file_contents('a', 'new\ncontents for\na\n')
        self.check_file_contents('d', 'contents\nfor d\n')

        if self.readonly:
            self.assertRaises(TransportNotPossible,
                t.put_multi, iter([('a', 'diff\ncontents for\na\n'),
                                  ('d', 'another contents\nfor d\n')]))
            open('a', 'wb').write('diff\ncontents for\na\n')
            open('d', 'wb').write('another contents\nfor d\n')
        else:
            self.assertEqual(
                t.put_multi(iter([('a', 'diff\ncontents for\na\n'),
                                  ('d', 'another contents\nfor d\n')]))
                             , 2)
        self.check_file_contents('a', 'diff\ncontents for\na\n')
        self.check_file_contents('d', 'another contents\nfor d\n')

        if self.readonly:
            self.assertRaises(TransportNotPossible,
                    t.put, 'path/doesnt/exist/c', 'contents')
        else:
            self.assertRaises(NoSuchFile,
                    t.put, 'path/doesnt/exist/c', 'contents')

    def test_put_file(self):
        t = self.get_transport()

        # Test that StringIO can be used as a file-like object with put
        f1 = StringIO('this is a string\nand some more stuff\n')
        if self.readonly:
            open('f1', 'wb').write(f1.read())
        else:
            t.put('f1', f1)

        del f1

        self.check_file_contents('f1', 
                'this is a string\nand some more stuff\n')

        f2 = StringIO('here is some text\nand a bit more\n')
        f3 = StringIO('some text for the\nthird file created\n')

        if self.readonly:
            open('f2', 'wb').write(f2.read())
            open('f3', 'wb').write(f3.read())
        else:
            t.put_multi([('f2', f2), ('f3', f3)])

        del f2, f3

        self.check_file_contents('f2', 'here is some text\nand a bit more\n')
        self.check_file_contents('f3', 'some text for the\nthird file created\n')

        # Test that an actual file object can be used with put
        f4 = open('f1', 'rb')
        if self.readonly:
            open('f4', 'wb').write(f4.read())
        else:
            t.put('f4', f4)

        del f4

        self.check_file_contents('f4', 
                'this is a string\nand some more stuff\n')

        f5 = open('f2', 'rb')
        f6 = open('f3', 'rb')
        if self.readonly:
            open('f5', 'wb').write(f5.read())
            open('f6', 'wb').write(f6.read())
        else:
            t.put_multi([('f5', f5), ('f6', f6)])

        del f5, f6

        self.check_file_contents('f5', 'here is some text\nand a bit more\n')
        self.check_file_contents('f6', 'some text for the\nthird file created\n')



    def test_mkdir(self):
        t = self.get_transport()

        # Test mkdir
        os.mkdir('dir_a')
        self.assertEqual(t.has('dir_a'), True)
        self.assertEqual(t.has('dir_b'), False)

        if self.readonly:
            self.assertRaises(TransportNotPossible,
                    t.mkdir, 'dir_b')
            os.mkdir('dir_b')
        else:
            t.mkdir('dir_b')
        self.assertEqual(t.has('dir_b'), True)
        self.assert_(os.path.isdir('dir_b'))

        if self.readonly:
            self.assertRaises(TransportNotPossible,
                    t.mkdir_multi, ['dir_c', 'dir_d'])
            os.mkdir('dir_c')
            os.mkdir('dir_d')
        else:
            t.mkdir_multi(['dir_c', 'dir_d'])

        if self.readonly:
            self.assertRaises(TransportNotPossible,
                    t.mkdir_multi, iter(['dir_e', 'dir_f']))
            os.mkdir('dir_e')
            os.mkdir('dir_f')
        else:
            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])
        for d in ['dir_a', 'dir_b', 'dir_c', 'dir_d', 'dir_e', 'dir_f']:
            self.assert_(os.path.isdir(d))

        if not self.readonly:
            self.assertRaises(NoSuchFile, t.mkdir, 'path/doesnt/exist')
            self.assertRaises(FileExists, t.mkdir, 'dir_a') # Creating a directory again should fail

        # Make sure the transport recognizes when a
        # directory is created by other means
        # Caching Transports will fail, because dir_e was already seen not
        # to exist. So instead, we will search for a new directory
        #os.mkdir('dir_e')
        #if not self.readonly:
        #    self.assertRaises(FileExists, t.mkdir, 'dir_e')

        os.mkdir('dir_g')
        if not self.readonly:
            self.assertRaises(FileExists, t.mkdir, 'dir_g')

        # Test get/put in sub-directories
        if self.readonly:
            open('dir_a/a', 'wb').write('contents of dir_a/a')
            open('dir_b/b', 'wb').write('contents of dir_b/b')
        else:
            self.assertEqual(
                t.put_multi([('dir_a/a', 'contents of dir_a/a'),
                             ('dir_b/b', 'contents of dir_b/b')])
                          , 2)
        for f in ('dir_a/a', 'dir_b/b'):
            self.assertEqual(t.get(f).read(), open(f).read())

    def test_copy_to(self):
        import tempfile
        from bzrlib.transport.local import LocalTransport

        t = self.get_transport()

        files = ['a', 'b', 'c', 'd']
        self.build_tree(files)

        dtmp = tempfile.mkdtemp(dir=u'.', prefix='test-transport-')
        dtmp_base = os.path.basename(dtmp)
        local_t = LocalTransport(dtmp)

        t.copy_to(files, local_t)
        for f in files:
            self.assertEquals(open(f).read(),
                    open(os.path.join(dtmp_base, f)).read())

        # Test that copying into a missing directory raises
        # NoSuchFile
        os.mkdir('e')
        open('e/f', 'wb').write('contents of e')
        self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], local_t)

        os.mkdir(os.path.join(dtmp_base, 'e'))
        t.copy_to(['e/f'], local_t)

        del dtmp, dtmp_base, local_t

        dtmp = tempfile.mkdtemp(dir=u'.', prefix='test-transport-')
        dtmp_base = os.path.basename(dtmp)
        local_t = LocalTransport(dtmp)

        files = ['a', 'b', 'c', 'd']
        t.copy_to(iter(files), local_t)
        for f in files:
            self.assertEquals(open(f).read(),
                    open(os.path.join(dtmp_base, f)).read())

        del dtmp, dtmp_base, local_t

    def test_append(self):
        t = self.get_transport()

        if self.readonly:
            open('a', 'wb').write('diff\ncontents for\na\n')
            open('b', 'wb').write('contents\nfor b\n')
        else:
            t.put_multi([
                    ('a', 'diff\ncontents for\na\n'),
                    ('b', 'contents\nfor b\n')
                    ])

        if self.readonly:
            self.assertRaises(TransportNotPossible,
                    t.append, 'a', 'add\nsome\nmore\ncontents\n')
            _append('a', 'add\nsome\nmore\ncontents\n')
        else:
            t.append('a', 'add\nsome\nmore\ncontents\n')

        self.check_file_contents('a', 
            'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n')

        if self.readonly:
            self.assertRaises(TransportNotPossible,
                    t.append_multi,
                        [('a', 'and\nthen\nsome\nmore\n'),
                         ('b', 'some\nmore\nfor\nb\n')])
            _append('a', 'and\nthen\nsome\nmore\n')
            _append('b', 'some\nmore\nfor\nb\n')
        else:
            t.append_multi([('a', 'and\nthen\nsome\nmore\n'),
                    ('b', 'some\nmore\nfor\nb\n')])
        self.check_file_contents('a', 
            'diff\ncontents for\na\n'
            'add\nsome\nmore\ncontents\n'
            'and\nthen\nsome\nmore\n')
        self.check_file_contents('b', 
                'contents\nfor b\n'
                'some\nmore\nfor\nb\n')

        if self.readonly:
            _append('a', 'a little bit more\n')
            _append('b', 'from an iterator\n')
        else:
            t.append_multi(iter([('a', 'a little bit more\n'),
                    ('b', 'from an iterator\n')]))
        self.check_file_contents('a', 
            'diff\ncontents for\na\n'
            'add\nsome\nmore\ncontents\n'
            'and\nthen\nsome\nmore\n'
            'a little bit more\n')
        self.check_file_contents('b', 
                'contents\nfor b\n'
                'some\nmore\nfor\nb\n'
                'from an iterator\n')

    def test_append_file(self):
        t = self.get_transport()

        contents = [
            ('f1', 'this is a string\nand some more stuff\n'),
            ('f2', 'here is some text\nand a bit more\n'),
            ('f3', 'some text for the\nthird file created\n'),
            ('f4', 'this is a string\nand some more stuff\n'),
            ('f5', 'here is some text\nand a bit more\n'),
            ('f6', 'some text for the\nthird file created\n')
        ]
        
        if self.readonly:
            for f, val in contents:
                open(f, 'wb').write(val)
        else:
            t.put_multi(contents)

        a1 = StringIO('appending to\none\n')
        if self.readonly:
            _append('f1', a1.read())
        else:
            t.append('f1', a1)

        del a1

        self.check_file_contents('f1', 
                'this is a string\nand some more stuff\n'
                'appending to\none\n')

        a2 = StringIO('adding more\ntext to two\n')
        a3 = StringIO('some garbage\nto put in three\n')

        if self.readonly:
            _append('f2', a2.read())
            _append('f3', a3.read())
        else:
            t.append_multi([('f2', a2), ('f3', a3)])

        del a2, a3

        self.check_file_contents('f2',
                'here is some text\nand a bit more\n'
                'adding more\ntext to two\n')
        self.check_file_contents('f3', 
                'some text for the\nthird file created\n'
                'some garbage\nto put in three\n')

        # Test that an actual file object can be used with put
        a4 = open('f1', 'rb')
        if self.readonly:
            _append('f4', a4.read())
        else:
            t.append('f4', a4)

        del a4

        self.check_file_contents('f4', 
                'this is a string\nand some more stuff\n'
                'this is a string\nand some more stuff\n'
                'appending to\none\n')

        a5 = open('f2', 'rb')
        a6 = open('f3', 'rb')
        if self.readonly:
            _append('f5', a5.read())
            _append('f6', a6.read())
        else:
            t.append_multi([('f5', a5), ('f6', a6)])

        del a5, a6

        self.check_file_contents('f5',
                'here is some text\nand a bit more\n'
                'here is some text\nand a bit more\n'
                'adding more\ntext to two\n')
        self.check_file_contents('f6',
                'some text for the\nthird file created\n'
                'some text for the\nthird file created\n'
                'some garbage\nto put in three\n')

    def test_delete(self):
        # TODO: Test Transport.delete
        t = self.get_transport()

    def test_move(self):
        # TODO: Test Transport.move
        t = self.get_transport()

    def test_copy(self):
        # TODO: Test Transport.move
        t = self.get_transport()

    def test_connection_error(self):
        """ConnectionError is raised when connection is impossible"""
        if not hasattr(self, "get_bogus_transport"):
            return
        t = self.get_bogus_transport()
        try:
            t.get('.bzr/branch')
        except (ConnectionError, NoSuchFile), e:
            pass
        except (Exception), e:
            self.failIf(True, 'Wrong exception thrown: %s' % e)
        else:
            self.failIf(True, 'Did not get the expected exception.')

        
class LocalTransportTest(TestCaseInTempDir, TestTransportMixIn):
    def get_transport(self):
        from bzrlib.transport.local import LocalTransport
        return LocalTransport(u'.')


class HttpTransportTest(TestCaseWithWebserver, TestTransportMixIn):

    readonly = True

    def get_transport(self):
        from bzrlib.transport.http import HttpTransport
        url = self.get_remote_url(u'.')
        return HttpTransport(url)

    def get_bogus_transport(self):
        from bzrlib.transport.http import HttpTransport
        return HttpTransport('http://jasldkjsalkdjalksjdkljasd')


class TestMemoryTransport(TestCase):

    def test_get_transport(self):
        memory.MemoryTransport()

    def test_clone(self):
        transport = memory.MemoryTransport()
        self.failUnless(transport.clone() is transport)

    def test_abspath(self):
        transport = memory.MemoryTransport()
        self.assertEqual("in-memory:relpath", transport.abspath('relpath'))

    def test_relpath(self):
        transport = memory.MemoryTransport()

    def test_append_and_get(self):
        transport = memory.MemoryTransport()
        transport.append('path', StringIO('content'))
        self.assertEqual(transport.get('path').read(), 'content')
        transport.append('path', StringIO('content'))
        self.assertEqual(transport.get('path').read(), 'contentcontent')

    def test_put_and_get(self):
        transport = memory.MemoryTransport()
        transport.put('path', StringIO('content'))
        self.assertEqual(transport.get('path').read(), 'content')
        transport.put('path', StringIO('content'))
        self.assertEqual(transport.get('path').read(), 'content')

    def test_append_without_dir_fails(self):
        transport = memory.MemoryTransport()
        self.assertRaises(NoSuchFile,
                          transport.append, 'dir/path', StringIO('content'))

    def test_put_without_dir_fails(self):
        transport = memory.MemoryTransport()
        self.assertRaises(NoSuchFile,
                          transport.put, 'dir/path', StringIO('content'))

    def test_get_missing(self):
        transport = memory.MemoryTransport()
        self.assertRaises(NoSuchFile, transport.get, 'foo')

    def test_has_missing(self):
        transport = memory.MemoryTransport()
        self.assertEquals(False, transport.has('foo'))

    def test_has_present(self):
        transport = memory.MemoryTransport()
        transport.append('foo', StringIO('content'))
        self.assertEquals(True, transport.has('foo'))

    def test_mkdir(self):
        transport = memory.MemoryTransport()
        transport.mkdir('dir')
        transport.append('dir/path', StringIO('content'))
        self.assertEqual(transport.get('dir/path').read(), 'content')

    def test_mkdir_missing_parent(self):
        transport = memory.MemoryTransport()
        self.assertRaises(NoSuchFile,
                          transport.mkdir, 'dir/dir')

    def test_mkdir_twice(self):
        transport = memory.MemoryTransport()
        transport.mkdir('dir')
        self.assertRaises(FileExists, transport.mkdir, 'dir')

    def test_parameters(self):
        transport = memory.MemoryTransport()
        self.assertEqual(True, transport.listable())
        self.assertEqual(False, transport.should_cache())

    def test_iter_files_recursive(self):
        transport = memory.MemoryTransport()
        transport.mkdir('dir')
        transport.put('dir/foo', StringIO('content'))
        transport.put('dir/bar', StringIO('content'))
        transport.put('bar', StringIO('content'))
        paths = set(transport.iter_files_recursive())
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)

    def test_stat(self):
        transport = memory.MemoryTransport()
        transport.put('foo', StringIO('content'))
        transport.put('bar', StringIO('phowar'))
        self.assertEqual(7, transport.stat('foo').st_size)
        self.assertEqual(6, transport.stat('bar').st_size)