~abentley/bzrtools/bzrtools.dev

# Copyright (C) 2007 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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

import os

from bzrlib.bzrdir import BzrDir

from bzrlib.tests import TestCaseWithTransport, TestSkipped

from bzrlib.plugins.bzrtools.bzrtools import (

    history_subset,

    NoRsync,

    NotStandalone,

    rspush,

    rsync,

    RsyncNoFile,

)

class MockRSync(object):

    def __init__(self):

        self.calls = []

    def __call__(self, source, target, **kwargs):

        self.calls.append((source, target, kwargs))

        if target.endswith('revision-history'):

            output = open(target, 'wb')

            output.close()

class TestRSPush(TestCaseWithTransport):

    def test_rsync(self):

        cwd = os.getcwd()

        try:

            self.assertRaises(RsyncNoFile, rsync, "a", "b", silent=True)

        except NoRsync:

            raise TestSkipped('rsync not installed')

        self.assertRaises(RsyncNoFile, rsync, cwd + "/a", cwd + "/b",

                          excludes=("*.py",), silent=True)

        self.assertRaises(NoRsync, rsync, cwd + "/a", cwd + "/b",

                          excludes=("*.py",), silent=True,

                          rsync_name="rsyncc")

    def test_rspush_locking(self):

        tree = self.make_branch_and_tree('tree', format='dirstate-tags')

        tree.commit('some changes')

        mock_rsync = MockRSync()

        rspush(tree, 'example.org:foo', _rsync=mock_rsync)

        rspush(tree, 'example.org:foo', working_tree = False,

               _rsync=mock_rsync)

        tree2 = self.make_branch_and_tree('tree2',

                                          format='pack-0.92')

        rspush(tree2, 'example.org:foo', _rsync=mock_rsync)

    def test_refuse_lightweight_checkout(self):

        branch = self.make_branch('tree')

        checkout = branch.create_checkout('checkout', lightweight=True)

        checkout.commit('some changes')

        mock_rsync = MockRSync()

        e = self.assertRaises(NotStandalone, rspush, checkout,

                              'example.org:foo', _rsync=mock_rsync)

        self.assertContainsRe(str(e), '/checkout/ is not a standalone tree.$')

    def test_refuse_checkout(self):

        branch = self.make_branch('tree')

        checkout = branch.create_checkout('checkout')

        checkout.commit('some changes')

        mock_rsync = MockRSync()

        self.assertRaises(NotStandalone, rspush, checkout, 'example.org:foo',

                          _rsync=mock_rsync)

    def test_refuse_shared(self):

        repo = self.make_repository('repo', shared=True)

        bzrdir = BzrDir.open('repo')

        bzrdir.create_branch()

        bzrdir.create_workingtree()

        tree = bzrdir.open_workingtree()

        tree.commit('some changes')

        mock_rsync = MockRSync()

        self.assertRaises(NotStandalone, rspush, tree, 'example.org:foo',

                          _rsync=mock_rsync)

    def do_history_subset(self, format):

        tree_a = self.make_branch_and_tree('tree_a', format=format)

        tree_a.lock_write()

        self.addCleanup(tree_a.unlock)

        tree_a.commit('foo')

        tree_b = tree_a.bzrdir.sprout('tree_b').open_workingtree()

        tree_a.commit('bar')

        try:

            self.assertTrue(history_subset('tree_b/', tree_a.branch))

        except NoRsync:

            raise TestSkipped('rsync not installed')

        self.assertFalse(history_subset('tree_a/', tree_b.branch))

    def do_history_subset_no_commits(self, format):

        tree_a = self.make_branch_and_tree('tree_a', format=format)

        tree_b = self.make_branch_and_tree('tree_b', format=format)

        try:

            self.assertTrue(history_subset('tree_b/', tree_a.branch))

        except NoRsync:

            raise TestSkipped('rsync not installed')

    def test_history_subset_dirstate(self):

        self.do_history_subset('dirstate')

    def test_history_subset_no_commits_dirstate(self):

        self.do_history_subset_no_commits('dirstate')

    def test_history_subset_pack(self):

        self.do_history_subset('pack-0.92')

    def test_history_subset_no_commits_pack(self):

        self.do_history_subset_no_commits('pack-0.92')