/brz/remove-bazaar

# Copyright (C) 2006-2012 Aaron Bentley

#

# 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

import os

from io import BytesIO

from shutil import rmtree, copy2, copytree

import tarfile

import tempfile

import warnings

from .. import (

    osutils,

    revision as _mod_revision,

    transform

    )

from ..controldir import ControlDir

from ..export import export

from ..upstream_import import (

    common_directory,

    get_archive_type,

    import_archive,

    import_tar,

    import_zip,

    import_dir,

    NotArchiveType,

    top_path,

    ZipFileWrapper,

)

from . import (

    TestCaseInTempDir,

    TestCaseWithTransport,

    )

from .features import UnicodeFilenameFeature

def import_tar_broken(tree, tar_input):

    """

    Import a tarfile with names that that end in //, e.g. Feisty Python 2.5

    """

    tar_file = tarfile.open('lala', 'r', tar_input)

    for member in tar_file.members:

        if member.name.endswith('/'):

            member.name += '/'

    import_archive(tree, tar_file)

class DirFileWriter(object):

    def __init__(self, fileobj, mode):

        # We may be asked to 'append'.  If so, fileobj already has a path.

        # So we copy the existing tree, and overwrite afterward.

        fileobj.seek(0)

        existing = fileobj.read()

        fileobj.seek(0)

        path = tempfile.mkdtemp(dir=os.getcwd())

        if existing != b'':

            # copytree requires the directory not to exist

            os.rmdir(path)

            copytree(existing, path)

        fileobj.write(path.encode('utf-8'))

        self.root = path

    def add(self, path):

        target_path = os.path.join(self.root, path)

        parent = osutils.dirname(target_path)

        if not os.path.exists(parent):

            os.makedirs(parent)

        kind = osutils.file_kind(path)

        if kind == 'file':

            copy2(path, target_path)

        if kind == 'directory':

            os.mkdir(target_path)

    def close(self):

        pass

class TestImport(TestCaseInTempDir):

    def make_tar(self, mode='w'):

        def maker(fileobj):

            return tarfile.open('project-0.1.tar', mode, fileobj)

        return self.make_archive(maker)

    def make_archive(self, maker, subdir=True):

        result = BytesIO()

        archive_file = maker(result)

        try:

            os.mkdir('project-0.1')

            if subdir:

                prefix = 'project-0.1/'

                archive_file.add('project-0.1')

            else:

                prefix = ''

                os.chdir('project-0.1')

            os.mkdir(prefix + 'junk')

            archive_file.add(prefix + 'junk')

            with open(prefix + 'README', 'wb') as f:

                f.write(b'What?')

            archive_file.add(prefix + 'README')

            with open(prefix + 'FEEDME', 'wb') as f:

                f.write(b'Hungry!!')

            archive_file.add(prefix + 'FEEDME')

            archive_file.close()

        finally:

            if not subdir:

                os.chdir('..')

            rmtree('project-0.1')

        result.seek(0)

        return result

    def make_archive2(self, builder, subdir):

        result = BytesIO()

        archive_file = builder(result)

        os.mkdir('project-0.2')

        try:

            if subdir:

                prefix = 'project-0.2/'

                archive_file.add('project-0.2')

            else:

                prefix = ''

                os.chdir('project-0.2')

            os.mkdir(prefix + 'junk')

            archive_file.add(prefix + 'junk')

            with open(prefix + 'README', 'wb') as f:

                f.write(b'Now?')

            archive_file.add(prefix + 'README')

            with open(prefix + 'README', 'wb') as f:

                f.write(b'Wow?')

            # Add a second entry for README with different contents.

            archive_file.add(prefix + 'README')

            archive_file.close()

        finally:

            if not subdir:

                os.chdir('..')

        result.seek(0)

        return result

    def make_messed_tar(self):

        result = BytesIO()

        with tarfile.open('project-0.1.tar', 'w', result) as tar_file:

            os.mkdir('project-0.1')

            tar_file.add('project-0.1')

            os.mkdir('project-0.2')

            tar_file.add('project-0.2')

            with open('project-0.1/README', 'wb') as f:

                f.write(b'What?')

            tar_file.add('project-0.1/README')

        rmtree('project-0.1')

        result.seek(0)

        return result

    def make_zip(self):

        def maker(fileobj):

            return ZipFileWrapper(fileobj, 'w')

        return self.make_archive(maker)

    def make_tar_with_bzrdir(self):

        result = BytesIO()

        with tarfile.open('tar-with-bzrdir.tar', 'w', result) as tar_file:

            os.mkdir('toplevel-dir')

            tar_file.add('toplevel-dir')

            os.mkdir('toplevel-dir/.bzr')

            tar_file.add('toplevel-dir/.bzr')

        rmtree('toplevel-dir')

        result.seek(0)

        return result

    def test_top_path(self):

        self.assertEqual(top_path('ab/b/c'), 'ab')

        self.assertEqual(top_path('etc'), 'etc')

        self.assertEqual(top_path('project-0.1'), 'project-0.1')

    def test_common_directory(self):

        self.assertEqual(common_directory(['ab/c/d', 'ab/c/e']), 'ab')

        self.assertIs(common_directory(['ab/c/d', 'ac/c/e']), None)

        self.assertEqual('FEEDME', common_directory(['FEEDME']))

    def test_untar(self):

        def builder(fileobj, mode='w'):

            return tarfile.open('project-0.1.tar', mode, fileobj)

        self.archive_test(builder, import_tar)

    def test_broken_tar(self):

        def builder(fileobj, mode='w'):

            return tarfile.open('project-0.1.tar', mode, fileobj)

        self.archive_test(builder, import_tar_broken, subdir=True)

    def test_unzip(self):

        def builder(fileobj, mode='w'):

            return ZipFileWrapper(fileobj, mode)

        self.archive_test(builder, import_zip)

    def test_copydir_nosub(self):

        def builder(fileobj, mode='w'):

            return DirFileWriter(fileobj, mode)

        # It would be bogus to test with the result in a subdirectory,

        # because for directories, the input root is always the output root.

        self.archive_test(builder, import_dir)

    def archive_test(self, builder, importer, subdir=False):

        archive_file = self.make_archive(builder, subdir)

        tree = ControlDir.create_standalone_workingtree('tree')

        with tree.lock_write():

            importer(tree, archive_file)

            self.assertTrue(tree.is_versioned('README'))

            self.assertTrue(tree.is_versioned('FEEDME'))

            self.assertTrue(os.path.isfile(tree.abspath('README')))

            self.assertEqual(tree.stored_kind('README'), 'file')

            self.assertEqual(tree.stored_kind('FEEDME'), 'file')

            with open(tree.abspath('junk/food'), 'wb') as f:

                f.write(b'I like food\n')

            with warnings.catch_warnings():

                warnings.simplefilter('ignore')

                archive_file = self.make_archive2(builder, subdir)

                importer(tree, archive_file)

            self.assertTrue(tree.is_versioned('README'))

            # Ensure the second version of the file is used.

            self.assertEqual(tree.get_file_text('README'), b'Wow?')

            self.assertTrue(not os.path.exists(tree.abspath('FEEDME')))

    def test_untar2(self):

        tar_file = self.make_messed_tar()

        tree = ControlDir.create_standalone_workingtree('tree')

        import_tar(tree, tar_file)

        self.assertTrue(tree.is_versioned('project-0.1/README'))

    def test_untar_gzip(self):

        tar_file = self.make_tar(mode='w:gz')

        tree = ControlDir.create_standalone_workingtree('tree')

        import_tar(tree, tar_file)

        self.assertTrue(tree.is_versioned('README'))

    def test_no_crash_with_bzrdir(self):

        tar_file = self.make_tar_with_bzrdir()

        tree = ControlDir.create_standalone_workingtree('tree')

        import_tar(tree, tar_file)

        # So long as it did not crash, that should be ok

    def test_get_archive_type(self):

        self.assertEqual(('tar', None), get_archive_type('foo.tar'))

        self.assertEqual(('zip', None), get_archive_type('foo.zip'))

        self.assertRaises(NotArchiveType, get_archive_type, 'foo.gif')

        self.assertEqual(('tar', 'gz'), get_archive_type('foo.tar.gz'))

        self.assertRaises(NotArchiveType, get_archive_type,

                          'foo.zip.gz')

        self.assertEqual(('tar', 'gz'), get_archive_type('foo.tgz'))

        self.assertEqual(('tar', 'lzma'), get_archive_type('foo.tar.lzma'))

        self.assertEqual(('tar', 'lzma'), get_archive_type('foo.tar.xz'))

        self.assertEqual(('tar', 'bz2'), get_archive_type('foo.tar.bz2'))

class TestWithStuff(TestCaseWithTransport):

    def transform_to_tar(self, tt):

        stream = BytesIO()

        export(tt.get_preview_tree(), root='', fileobj=stream, format='tar',

               dest=None)

        return stream

    def get_empty_tt(self):

        b = self.make_repository('foo')

        null_tree = b.revision_tree(_mod_revision.NULL_REVISION)

        tt = null_tree.preview_transform()

        tt.new_directory('', transform.ROOT_PARENT, b'tree-root')

        tt.fixup_new_roots()

        self.addCleanup(tt.finalize)

        return tt

    def test_nonascii_paths(self):

        self.requireFeature(UnicodeFilenameFeature)

        tt = self.get_empty_tt()

        tt.new_file(u'\u1234file', tt.root, [b'contents'], b'new-file')

        tt.new_file('other', tt.root, [b'contents'], b'other-file')

        tarfile = self.transform_to_tar(tt)

        tarfile.seek(0)

        tree = self.make_branch_and_tree('bar')

        import_tar(tree, tarfile)

        self.assertPathExists(u'bar/\u1234file')