~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2008, 2009, 2010, 2012, 2016 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

"""Tests for content filtering conformance"""

import os

from bzrlib.controldir import ControlDir

from bzrlib.filters import ContentFilter

from bzrlib.switch import switch

from bzrlib.workingtree import WorkingTree

from bzrlib.tests.per_workingtree import TestCaseWithWorkingTree

def _converter_helper(chunks, fn):

    result = []

    for chunk in chunks:

        result.append(getattr(chunk, fn)())

    return iter(result)

def _swapcase(chunks, context=None):

    """A converter that swaps the case of text."""

    return _converter_helper(chunks, 'swapcase')

def _uppercase(chunks, context=None):

    """A converter that converts text to uppercase."""

    return _converter_helper(chunks, 'upper')

def _lowercase(chunks, context=None):

    """A converter that converts text to lowercase."""

    return _converter_helper(chunks, 'lower')

_trailer_string = '\nend string\n'

def _append_text(chunks, context=None):

    """A content filter that appends a string to the end of the file.

    This tests filters that change the length."""

    return chunks + [_trailer_string]

def _remove_appended_text(chunks, context=None):

    """Remove the appended text."""

    text = ''.join(chunks)

    if text.endswith(_trailer_string):

        text = text[:-len(_trailer_string)]

    return [text]

class TestWorkingTreeWithContentFilters(TestCaseWithWorkingTree):

    def create_cf_tree(self, txt_reader, txt_writer, dir='.'):

        tree = self.make_branch_and_tree(dir)

        def _content_filter_stack(path=None, file_id=None):

            if path.endswith('.txt'):

                return [ContentFilter(txt_reader, txt_writer)]

            else:

                return []

        tree._content_filter_stack = _content_filter_stack

        self.build_tree_contents([

            (dir + '/file1.txt', 'Foo Txt'),

            (dir + '/file2.bin', 'Foo Bin')])

        tree.add(['file1.txt', 'file2.bin'])

        tree.commit('commit raw content')

        txt_fileid = tree.path2id('file1.txt')

        bin_fileid = tree.path2id('file2.bin')

        return tree, txt_fileid, bin_fileid

    def create_cf_tree_with_two_revisions(self, txt_reader, txt_writer,

        dir='.'):

        tree = self.make_branch_and_tree(dir)

        def _content_filter_stack(path=None, file_id=None):

            if path.endswith('.txt'):

                return [ContentFilter(txt_reader, txt_writer)]

            else:

                return []

        tree._content_filter_stack = _content_filter_stack

        self.build_tree_contents([

            (dir + '/file1.txt', 'Foo Txt'),

            (dir + '/file2.bin', 'Foo Bin'),

            (dir + '/file3.txt', 'Bar Txt'),

            ])

        tree.add(['file1.txt', 'file2.bin', 'file3.txt'])

        tree.commit('commit raw content')

        fileid_1 = tree.path2id('file1.txt')

        fileid_2 = tree.path2id('file2.bin')

        fileid_3 = tree.path2id('file3.txt')

        # Commit another revision with various changes. We make sure

        # the change includes a modification, an addition and a deletion.

        # Renames are more complex and need a separate set of tests later.

        self.build_tree_contents([

            (dir + '/file1.txt', 'Foo ROCKS!'),

            (dir + '/file4.txt', 'Hello World'),

            ])

        tree.add(['file4.txt'])

        tree.remove(['file3.txt'], keep_files=False)

        tree.commit("change, add and rename stuff")

        fileid_4 = tree.path2id('file4.txt')

        return tree, fileid_1, fileid_2, fileid_3, fileid_4

    def patch_in_content_filter(self):

        # Patch in a custom, symmetric content filter stack. It's pretty gross

        # that we need to monkey-patch a class method to do this, bit it's

        # the easiest way currently given we don't have easy access to the

        # WorkingTree after it is created but before the filter stack is used

        # to populate content.

        def new_stack(tree, path=None, file_id=None):

            if path.endswith('.txt'):

                return [ContentFilter(_swapcase, _swapcase)]

            else:

                return []

        self.overrideAttr(WorkingTree, '_content_filter_stack', new_stack)

    def assert_basis_content(self, expected_content, branch, file_id):

        # Note: We need to use try/finally here instead of addCleanup()

        # as the latter leaves the read lock in place too long

        basis = branch.basis_tree()

        basis.lock_read()

        try:

            self.assertEqual(expected_content, basis.get_file_text(file_id))

        finally:

            basis.unlock()

    def test_symmetric_content_filtering(self):

        # test handling when read then write gives back the initial content

        tree, txt_fileid, bin_fileid = self.create_cf_tree(

            txt_reader=_swapcase, txt_writer=_swapcase)

        # Check that the basis tree has the expected content

        basis = tree.basis_tree()

        basis.lock_read()

        self.addCleanup(basis.unlock)

        if tree.supports_content_filtering():

            expected = "fOO tXT"

        else:

            expected = "Foo Txt"

        self.assertEqual(expected, basis.get_file_text(txt_fileid))

        self.assertEqual('Foo Bin', basis.get_file_text(bin_fileid))

        # Check that the working tree has the original content

        tree.lock_read()

        self.addCleanup(tree.unlock)

        self.assertEqual('Foo Txt', tree.get_file(txt_fileid,

            filtered=False).read())

        self.assertEqual('Foo Bin', tree.get_file(bin_fileid,

            filtered=False).read())

    def test_readonly_content_filtering(self):

        # test handling with a read filter but no write filter

        tree, txt_fileid, bin_fileid = self.create_cf_tree(

            txt_reader=_uppercase, txt_writer=None)

        # Check that the basis tree has the expected content

        basis = tree.basis_tree()

        basis.lock_read()

        self.addCleanup(basis.unlock)

        if tree.supports_content_filtering():

            expected = "FOO TXT"

        else:

            expected = "Foo Txt"

        self.assertEqual(expected, basis.get_file_text(txt_fileid))

        self.assertEqual('Foo Bin', basis.get_file_text(bin_fileid))

        # We expect the workingtree content to be unchanged (for now at least)

        tree.lock_read()

        self.addCleanup(tree.unlock)

        self.assertEqual('Foo Txt', tree.get_file(txt_fileid,

            filtered=False).read())

        self.assertEqual('Foo Bin', tree.get_file(bin_fileid,

            filtered=False).read())

    def test_branch_source_filtered_target_not(self):

        # Create a source branch with content filtering

        source, txt_fileid, bin_fileid = self.create_cf_tree(

            txt_reader=_uppercase, txt_writer=_lowercase, dir='source')

        if not source.supports_content_filtering():

            return

        self.assertFileEqual("Foo Txt", 'source/file1.txt')

        self.assert_basis_content("FOO TXT", source, txt_fileid)

        # Now branch it

        self.run_bzr('branch source target')

        target = WorkingTree.open('target')

        # Even though the content in source and target are different

        # due to different filters, iter_changes should be clean

        self.assertFileEqual("FOO TXT", 'target/file1.txt')

        changes = target.changes_from(source.basis_tree())

        self.assertFalse(changes.has_changed())

    def test_branch_source_not_filtered_target_is(self):

        # Create a source branch with content filtering

        source, txt_fileid, bin_fileid = self.create_cf_tree(

            txt_reader=None, txt_writer=None, dir='source')

        if not source.supports_content_filtering():

            return

        self.assertFileEqual("Foo Txt", 'source/file1.txt')

        self.assert_basis_content("Foo Txt", source, txt_fileid)

        # Now patch in content filtering and branch the source

        self.patch_in_content_filter()

        self.run_bzr('branch source target')

        target = WorkingTree.open('target')

        # Even though the content in source and target are different

        # due to different filters, iter_changes should be clean

        self.assertFileEqual("fOO tXT", 'target/file1.txt')

        changes = target.changes_from(source.basis_tree())

        self.assertFalse(changes.has_changed())

    def test_path_content_summary(self):

        """path_content_summary should always talk about the canonical form."""

        # see https://bugs.launchpad.net/bzr/+bug/415508

        #

        # set up a tree where the canonical form has a string added to the

        # end

        source, txt_fileid, bin_fileid = self.create_cf_tree(

            txt_reader=_append_text,

            txt_writer=_remove_appended_text,

            dir='source')

        if not source.supports_content_filtering():

            return

        source.lock_read()

        self.addCleanup(source.unlock)

        expected_canonical_form = 'Foo Txt\nend string\n'

        self.assertEqual(source.get_file(txt_fileid, filtered=True).read(),

            expected_canonical_form)

        self.assertEqual(source.get_file(txt_fileid, filtered=False).read(),

            'Foo Txt')

        # results are: kind, size, executable, sha1_or_link_target

        result = source.path_content_summary('file1.txt')

        self.assertEqual(result,

            ('file', None, False, None))

        # we could give back the length of the canonical form, but in general

        # that will be expensive to compute, so it's acceptable to just return

        # None.

    def test_content_filtering_applied_on_pull(self):

        # Create a source branch with two revisions

        source, fileid_1, fileid_2, fileid_3, fileid_4 = \

            self.create_cf_tree_with_two_revisions(txt_reader=None,

            txt_writer=None, dir='source')

        if not source.supports_content_filtering():

            return

        self.assertFileEqual("Foo ROCKS!", 'source/file1.txt')

        self.assert_basis_content("Foo ROCKS!", source, fileid_1)

        # Now patch in content filtering and branch from revision 1

        self.patch_in_content_filter()

        self.run_bzr('branch -r1 source target')

        target = WorkingTree.open('target')

        self.assert_basis_content("Foo Txt", target, fileid_1)

        self.assertFileEqual("fOO tXT", 'target/file1.txt')

        self.assert_basis_content("Foo Bin", target, fileid_2)

        self.assertFileEqual("Foo Bin", 'target/file2.bin')

        self.assert_basis_content("Bar Txt", target, fileid_3)

        self.assertFileEqual("bAR tXT", 'target/file3.txt')

        # Pull the latter change and check the target tree is updated

        self.run_bzr('pull -d target')

        self.assert_basis_content("Foo ROCKS!", target, fileid_1)

        self.assertFileEqual("fOO rocks!", 'target/file1.txt')

        self.assert_basis_content("Foo Bin", target, fileid_2)

        self.assert_basis_content("Hello World", target, fileid_4)

        self.assertFileEqual("hELLO wORLD", 'target/file4.txt')

    def test_content_filtering_applied_on_merge(self):

        # Create a source branch with two revisions

        source, fileid_1, fileid_2, fileid_3, fileid_4 = \

            self.create_cf_tree_with_two_revisions(txt_reader=None,

            txt_writer=None, dir='source')

        if not source.supports_content_filtering():

            return

        self.assert_basis_content("Foo ROCKS!", source, fileid_1)

        self.assertFileEqual("Foo ROCKS!", 'source/file1.txt')

        self.assert_basis_content("Foo Bin", source, fileid_2)

        self.assert_basis_content("Hello World", source, fileid_4)

        self.assertFileEqual("Hello World", 'source/file4.txt')

        # Now patch in content filtering and branch from revision 1

        self.patch_in_content_filter()

        self.run_bzr('branch -r1 source target')

        target = WorkingTree.open('target')

        self.assert_basis_content("Foo Txt", target, fileid_1)

        self.assertFileEqual("fOO tXT", 'target/file1.txt')

        self.assertFileEqual("Foo Bin", 'target/file2.bin')

        self.assertFileEqual("bAR tXT", 'target/file3.txt')

        # Merge the latter change and check the target tree is updated

        self.run_bzr('merge -d target source')

        self.assertFileEqual("fOO rocks!", 'target/file1.txt')

        self.assertFileEqual("hELLO wORLD", 'target/file4.txt')

        # Commit the merge and check the right content is stored

        target.commit("merge file1.txt changes from source")

        self.assert_basis_content("Foo ROCKS!", target, fileid_1)

        self.assert_basis_content("Hello World", target, fileid_4)

    def test_content_filtering_applied_on_switch(self):

        # Create a source branch with two revisions

        source, fileid_1, fileid_2, fileid_3, fileid_4 = \

            self.create_cf_tree_with_two_revisions(txt_reader=None,

            txt_writer=None, dir='branch-a')

        if not source.supports_content_filtering():

            return

        # Now patch in content filtering and branch from revision 1

        self.patch_in_content_filter()

        self.run_bzr('branch -r1 branch-a branch-b')

        # Now create a lightweight checkout referring to branch-b

        self.run_bzr('checkout --lightweight branch-b checkout')

        self.assertFileEqual("fOO tXT", 'checkout/file1.txt')

        # Switch it to branch-b and check the tree is updated

        checkout_control_dir = ControlDir.open_containing('checkout')[0]

        switch(checkout_control_dir, source.branch)

        self.assertFileEqual("fOO rocks!", 'checkout/file1.txt')

        self.assertFileEqual("hELLO wORLD", 'checkout/file4.txt')

    def test_content_filtering_applied_on_revert_delete(self):

        # Create a source branch with content filtering

        source, txt_fileid, bin_fileid = self.create_cf_tree(

            txt_reader=_uppercase, txt_writer=_lowercase, dir='source')

        if not source.supports_content_filtering():

            return

        self.assertFileEqual("Foo Txt", 'source/file1.txt')

        self.assert_basis_content("FOO TXT", source, txt_fileid)

        # Now delete the file, revert it and check the content

        os.unlink('source/file1.txt')

        self.assertFalse(os.path.exists('source/file1.txt'))

        source.revert(['file1.txt'])

        self.assertTrue(os.path.exists('source/file1.txt'))

        # Note: we don't get back exactly what was in the tree

        # previously because lower(upper(text)) is a lossy transformation

        self.assertFileEqual("foo txt", 'source/file1.txt')

    def test_content_filtering_applied_on_revert_rename(self):

        # Create a source branch with content filtering

        source, txt_fileid, bin_fileid = self.create_cf_tree(

            txt_reader=_uppercase, txt_writer=_lowercase, dir='source')

        if not source.supports_content_filtering():

            return

        self.assertFileEqual("Foo Txt", 'source/file1.txt')

        self.assert_basis_content("FOO TXT", source, txt_fileid)

        # Now modify & rename a file, revert it and check the content

        self.build_tree_contents([

            ('source/file1.txt', 'Foo Txt with new content')])

        source.rename_one('file1.txt', 'file1.bin')

        self.assertTrue(os.path.exists('source/file1.bin'))

        self.assertFalse(os.path.exists('source/file1.txt'))

        self.assertFileEqual("Foo Txt with new content", 'source/file1.bin')

        source.revert(['file1.bin'])

        self.assertFalse(os.path.exists('source/file1.bin'))

        self.assertTrue(os.path.exists('source/file1.txt'))

        # Note: we don't get back exactly what was in the tree

        # previously because lower(upper(text)) is a lossy transformation

        self.assertFileEqual("foo txt", 'source/file1.txt')