# Copyright (C) 2009 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


from cStringIO import StringIO
import errno

from bzrlib import branch, osutils, transform, workingtree
from bzrlib.plugins.pipeline.pipeline import PipeManager, PipeStorage
from bzrlib.plugins.pipeline.tests import TestCaseWithPipes


class TestBlackbox(TestCaseWithPipes):

    def test_add_pipe(self):
        tree = self.make_branch_and_checkout('first')
        self.run_bzr('add-pipe last -d checkout')
        manager = PipeManager(branch.Branch.open('checkout'))
        self.assertContainsRe(manager.storage.branch.base, 'last/$')
        self.assertContainsRe(manager.storage.get_prev(), 'first/$')
        top_manager = PipeManager(branch.Branch.open('first'))
        self.assertContainsRe(top_manager.storage.get_next(), 'last/$')

    def test_add_pipe_no_switch(self):
        tree = self.make_branch_and_checkout('first')
        self.run_bzr('add-pipe last --no-switch', working_dir='checkout')
        manager = PipeManager(branch.Branch.open('checkout'))
        self.assertContainsRe(manager.storage.branch.base, 'first/$')
        self.assertContainsRe(manager.storage.get_next(), 'last/$')
        top_manager = PipeManager(branch.Branch.open('last'))
        self.assertContainsRe(top_manager.storage.get_prev(), 'first/$')

    def test_add_pipe_after(self):
        tree = self.make_branch_and_checkout('first')
        branch2 = self.make_branch('middle')
        PipeStorage.connect(tree.branch, branch2)
        self.run_bzr('add-pipe last --after middle', working_dir='checkout')
        manager = PipeManager(branch2)
        self.assertContainsRe(manager.storage.get_next(), 'last/$')
        top_manager = PipeManager(branch.Branch.open('last'))
        self.assertContainsRe(top_manager.storage.get_prev(), 'middle/$')

    def test_add_pipe_before(self):
        tree = self.make_branch_and_checkout('first')
        tree.commit('first commit', rev_id='rev1')
        last = PipeManager(tree.branch).insert_pipe('last')
        last.lock_write()
        try:
            tt = transform.TransformPreview(last.basis_tree())
            tt.commit(last, 'empty commit')
        finally:
            last.unlock()
        self.run_bzr('add-pipe middle --before last', working_dir='checkout')
        manager = PipeManager(last)
        self.assertContainsRe(manager.storage.get_prev(), 'middle/$')
        middle_manager = PipeManager(branch.Branch.open('middle'))
        self.assertContainsRe(middle_manager.storage.get_next(), 'last/$')
        self.assertEqual('rev1', middle_manager.storage.branch.last_revision())

    def test_add_pipe_before_first(self):
        """Test the LCA codepath."""
        submit = self.make_branch_and_tree('submit')
        submit.commit('rev1')
        tree = self.make_branch_and_checkout('first')
        tree.commit('foo')
        tree.branch.set_parent(submit.branch.base)
        self.run_bzr('add-pipe -d checkout new-first --before')

    def test_add_pipe_before_no_neighbour(self):
        tree = self.make_branch_and_checkout('last')
        branch2 = self.make_branch('middle')
        PipeStorage.connect(tree.branch, branch2)
        self.run_bzr('add-pipe first --before', working_dir='checkout')
        manager = PipeManager(tree.branch)
        self.assertContainsRe(manager.storage.get_prev(), 'first/$')
        first_manager = PipeManager(branch.Branch.open('first'))
        self.assertContainsRe(first_manager.storage.get_next(), 'last/$')

    def test_add_pipe_revision(self):
        tree = self.make_branch_and_checkout('first')
        rev1 = tree.commit('rev1')
        tree.commit('rev2')
        self.run_bzr('add-pipe -r 1 next', working_dir='checkout')
        self.assertEqual(rev1, branch.Branch.open('next').last_revision())

    def test_add_pipe_two_revision(self):
        tree = self.make_branch_and_checkout('first')
        rev1 = tree.commit('rev1')
        rev2 = tree.commit('rev2')
        out, err = self.run_bzr('add-pipe -r 1..2 next',
                                working_dir='checkout', retcode=3)
        self.assertContainsRe(err, 'Only one revision may be supplied.')

    def test_add_pipe_requires_lightweight_checkout(self):
        tree = self.make_branch_and_tree('first')
        out, err = self.run_bzr('add-pipe new', working_dir='first',
                                retcode=3)
        self.assertContainsRe(err, 'add-pipe should be run in a lightweight'
                              ' checkout.  See bzr help pipeline for details.')

    def do_add(self, no_switch=False):
        branch = self.make_branch('first')
        tree = branch.create_checkout('checkout', lightweight=True)
        self.build_tree(['checkout/file1'])
        tree.add('file1')
        rev1 = tree.commit('rev1')
        self.build_tree(['checkout/file2'])
        tree.add('file2')
        cmdline = ['add-pipe', 'next']
        if no_switch:
            cmdline.append('--no-switch')
        self.run_bzr(cmdline, working_dir='checkout')

    def test_add_pipe_no_switch_stores_uncommitted(self):
        self.do_add(no_switch=True)
        self.failUnlessExists('checkout/file1')
        self.failIfExists('checkout/file2')

    def test_add_pipe_keeps_uncommitted(self):
        self.do_add(no_switch=False)
        self.failUnlessExists('checkout/file1')
        self.failUnlessExists('checkout/file2')

    def test_add_pipe_accepts_branch(self):
        self.make_branch_and_checkout('first')
        self.make_branch('second')
        self.run_bzr('add-pipe -d checkout second')

    def test_remove_pipe(self):
        foo = PipeManager(self.make_branch('foo'))
        bar = PipeManager(foo.insert_pipe('bar'))
        self.run_bzr('remove-pipe foo')
        self.assertIs(None, foo.get_next_pipe())
        self.assertIs(None, bar.get_prev_pipe())

    def test_remove_switches_checkout_next(self):
        foo = PipeManager(self.make_branch('foo'))
        bar = PipeManager(foo.insert_pipe('bar'))
        checkout = foo.storage.branch.create_checkout('checkout',
                                                      lightweight=True)
        self.run_bzr('remove-pipe', working_dir='checkout')
        self.assertEqual(branch.Branch.open('checkout').base,
                         bar.storage.branch.base)

    def test_remove_switches_checkout_prev(self):
        foo = PipeManager(self.make_branch('foo'))
        bar = PipeManager(foo.insert_pipe('bar'))
        checkout = bar.storage.branch.create_checkout('checkout',
                                                      lightweight=True)
        self.run_bzr('remove-pipe', working_dir='checkout')
        self.assertEqual(branch.Branch.open('checkout').base,
                         foo.storage.branch.base)

    def test_remove_ignores_non_checkout_tree(self):
        foo = self.make_branch_and_tree('foo')
        PipeManager(foo.branch).insert_pipe('bar')
        self.run_bzr('remove-pipe', working_dir='foo')

    def test_remove_unconnected(self):
        foo = self.make_branch('foo')
        checkout = foo.create_checkout('checkout', lightweight=True)
        out, err = self.run_bzr('remove-pipe', working_dir='checkout',
                                retcode=3)
        self.assertContainsRe(err, 'Branch is not connected to a pipeline.')

    def test_remove_specified_pipe(self):
        foo = self.make_branch('foo')
        manager = PipeManager(foo)
        checkout = foo.create_checkout('checkout', lightweight=True)
        bar = PipeManager(manager.insert_pipe('bar'))
        baz = manager.insert_pipe('baz')
        self.run_bzr('remove-pipe bar', working_dir='checkout')
        self.assertIs(None, bar.get_next_pipe())
        self.assertIs(None, bar.get_prev_pipe())
        self.assertEqual('../baz/', manager.storage.get_next())
        checkout = workingtree.WorkingTree.open('checkout')
        self.assertEqual(foo.base, checkout.branch.base)

    def test_remove_pipe_keeps_branch(self):
        tree = self.make_branch_and_checkout('first')
        PipeManager(tree.branch).insert_pipe('second')
        self.run_bzr('remove-pipe second', working_dir='checkout')
        self.failUnlessExists('second')

    def test_remove_pipe_branch(self):
        tree = self.make_branch_and_checkout('first')
        PipeManager(tree.branch).insert_pipe('second')
        self.run_bzr('remove-pipe second --branch', working_dir='checkout')
        self.failIfExists('second')

    def test_switch_pipe(self):
        self.do_add(no_switch=True)
        self.build_tree(['checkout/file3'])
        workingtree.WorkingTree.open('checkout').add('file3')
        out, err = self.run_bzr('switch-pipe next -d checkout')
        self.failUnlessExists('checkout/file1')
        self.failUnlessExists('checkout/file2')
        self.failIfExists('checkout/file3')
        self.assertEqual('Uncommitted changes stored in pipe "first".\n'
                         'Switched from "first" to "next".\n', err)

    def test_switch_not_checkout_keeps_changes(self):
        foo = self.make_branch_and_tree('foo')
        bar = PipeManager(foo.branch).insert_pipe('bar')
        self.build_tree(['foo/file'])
        foo.add('file')
        out, err = self.run_bzr('switch-pipe bar', working_dir='foo',
                                retcode=3)
        self.failUnlessExists('foo/file')
        self.assertContainsRe(err, 'Directory is not a lightweight checkout.')

    def test_switch_pipe_accepts_aliases(self):
        checkout = self.make_branch_and_checkout('foo')
        checkout.commit('lala')
        PipeManager(checkout.branch).insert_pipe('bar')
        self.run_bzr('switch-pipe -d checkout :next')
        checkout = workingtree.WorkingTree.open('checkout')

    def test_show_pipeline(self):
        first = self.make_branch('first')
        last = self.make_branch('last')
        PipeStorage.connect(first, last)
        PipeStorage(first)._put_transform(StringIO('lala'))
        tree = last.create_checkout('checkout', lightweight=True)
        out, err = self.run_bzr('show-pipeline checkout')
        self.assertEqual(' U first\n*  last\n', out)

    def test_show_pipeline_subdir(self):
        first = self.make_branch('first')
        last = self.make_branch('last')
        PipeStorage.connect(first, last)
        PipeStorage(first)._put_transform(StringIO('lala'))
        tree = last.create_checkout('checkout', lightweight=True)
        self.build_tree(['checkout/subdir/'])
        out, err = self.run_bzr('show-pipeline', working_dir='checkout/subdir')
        self.assertEqual(' U first\n*  last\n', out)

    def test_pump(self):
        first = self.make_branch('first')
        checkout = first.create_checkout('checkout', lightweight=True)
        checkout.commit('first 1', rev_id='rev1-id')
        first_manager = PipeManager(first)
        next = first_manager.insert_pipe('next')
        checkout.commit('first 2', rev_id='rev2-id')
        first_manager.switch_to_pipe(checkout, next)
        checkout = workingtree.WorkingTree.open('checkout')
        checkout.commit('next', rev_id='rev3-id')
        PipeManager(next).switch_to_pipe(checkout, first)
        self.run_bzr('pump -d checkout')
        rev = next.repository.get_revision(next.last_revision())
        self.assertEqual(['rev3-id', 'rev2-id'], rev.parent_ids)

    def test_pump_requires_light_checkout(self):
        tree = self.make_branch_and_tree('foo')
        out, err = self.run_bzr('pump', working_dir='foo', retcode=3)
        self.assertContainsRe(err, 'Directory is not a lightweight checkout.')

    def test_pump_from_submit(self):
        submit = self.make_branch_and_tree('submit')
        submit.commit('empty')
        one = submit.bzrdir.sprout('one').open_branch()
        one.set_submit_branch(submit.branch.base)
        self.build_tree(['submit/submit-file'])
        submit.add('submit-file')
        submit.commit('added submit-file')
        two = PipeManager(one).insert_pipe('two')
        checkout = two.create_checkout('checkout', lightweight=True)
        self.run_bzr('pump -d checkout --from-submit')
        self.failUnlessExists('checkout/submit-file')

    def test_merge_uncommitted_from_pipe(self):
        foo = self.make_branch('foo')
        checkout = foo.create_checkout('checkout', lightweight=True)
        checkout.commit('Empty commit.')
        self.build_tree(['checkout/file'])
        checkout.add('file')
        bar = PipeManager(PipeManager(foo).insert_pipe('bar'))
        bar.store_uncommitted(checkout)
        self.run_bzr('merge -d checkout --uncommitted bar')
        self.failUnlessExists('checkout/file')
        self.assertTrue(bar.has_stored_changes())

    def test_merge_uncommitted_from_pipe_alias(self):
        foo = self.make_branch('foo')
        checkout = foo.create_checkout('checkout', lightweight=True)
        checkout.commit('Empty commit.')
        self.build_tree(['checkout/file'])
        checkout.add('file')
        bar = PipeManager(PipeManager(foo).insert_pipe('bar'))
        bar.store_uncommitted(checkout)
        self.run_bzr('merge --uncommitted :next', working_dir='checkout')
        self.failUnlessExists('checkout/file')
        self.assertTrue(bar.has_stored_changes())

    def test_merge_uncommitted_no_location_uses_submit_branch(self):
        foo = self.make_branch_and_checkout('foo')
        foo.commit('Empty commit.')
        bar = foo.bzrdir.sprout('bar').open_workingtree()
        self.build_tree(['bar/file'])
        bar.add('file')
        foo.branch.set_submit_branch(bar.branch.base)
        self.run_bzr('merge --uncommitted', working_dir='checkout')
        self.failUnlessExists('checkout/file')

    def test_sync_pipeline(self):
        self.build_tree(['local/', 'remote/'])
        foo = PipeManager(self.make_branch('local/foo'))
        foo.insert_pipe('bar')
        self.run_bzr('sync-pipeline ../../remote/foo', working_dir='local/foo')
        self.failUnlessExists('remote/foo')
        self.failUnlessExists('remote/bar')

    def test_sync_pipeline_default(self):
        self.build_tree(['local/', 'remote/'])
        foo = PipeManager(self.make_branch('local/foo'))
        foo.insert_pipe('bar')
        foo.storage.branch.set_push_location(osutils.abspath('remote/foo'))
        self.run_bzr('sync-pipeline', working_dir='local/foo')
        self.failUnlessExists('remote/foo')
        self.failUnlessExists('remote/bar')

    def test_sync_pipeline_errors_if_default_none(self):
        self.build_tree(['local/', 'remote/'])
        foo = PipeManager(self.make_branch('local/foo'))
        foo.insert_pipe('bar')
        out, err = self.run_bzr('sync-pipeline', working_dir='local/foo',
            retcode=3)
        self.assertContainsRe(err,
            'No location specified and none remembered.')

    def test_sync_pipeline_uses_first_pipe(self):
        self.build_tree(['local/', 'remote/'])
        foo = PipeManager(self.make_branch('local/foo'))
        bar = foo.insert_pipe('bar')
        foo.storage.branch.set_push_location(osutils.abspath('remote/foo'))
        bar.set_push_location(osutils.abspath('nowhere/bar'))
        self.run_bzr('sync-pipeline', working_dir='local/bar')
        self.failUnlessExists('remote/foo')
        self.failUnlessExists('remote/bar')

    def test_sync_pipeline_falls_back_to_other_pipes(self):
        self.build_tree(['local/', 'remote/'])
        foo = PipeManager(self.make_branch('local/foo'))
        bar = foo.insert_pipe('bar')
        bar.set_push_location(osutils.abspath('remote/bar'))
        self.run_bzr('sync-pipeline', working_dir='local/foo')
        self.failUnlessExists('remote/foo')
        self.failUnlessExists('remote/bar')

    def test_sync_pipeline_previous_pipe_added_with_push_loc(self):
        self.build_tree(['local/', 'remote/'])
        foo = PipeManager(self.make_branch('local/foo'))
        self.run_bzr('sync-pipeline ../../remote/foo', working_dir='local/foo')
        bar = foo.insert_pipe('bar', before=True)
        bar.set_push_location(osutils.abspath('remote/bar'))
        self.run_bzr('sync-pipeline', working_dir='local/bar')

    def test_reconfigure_pipeline(self):
        self.make_branch_and_tree('pipeline')
        self.run_bzr('reconfigure-pipeline', working_dir='pipeline')
        b = branch.Branch.open('pipeline/.bzr/pipes/pipeline')

    def assertFileContainsRe(self, regex, path):
        try:
            my_file = open(path, 'rb')
        except IOError, e:
            if e.errno == errno.ENOENT:
                self.fail('%s does not exist' % path)
            else:
                raise
        try:
            self.assertContainsRe(my_file.read(), regex)
        finally:
            my_file.close()

    def test_pipe_patches(self):
        """Pipe-patches should export a series of patches."""
        upstream = self.make_branch_and_tree('upstream')
        self.build_tree_contents([('upstream/file', 'A\n')])
        upstream.add('file')
        upstream.commit('commit')
        b_pipe = upstream.bzrdir.sprout('b-pipe').open_workingtree()
        self.build_tree_contents([('b-pipe/file', 'B\n')])
        b_pipe.commit('commit')
        c_pipe = PipeManager(b_pipe.branch).insert_pipe('c-pipe')
        self.build_tree_contents([('c-pipe/file', 'C\n')])
        c_pipe_tree = c_pipe.bzrdir.open_workingtree().commit('commit')
        self.build_tree(['patches/'])
        self.run_bzr('pipe-patches -d b-pipe patches')
        self.assertFileContainsRe('-A\n\\+B', 'patches/00-b-pipe.patch')
        self.assertFileContainsRe('-B\n\\+C', 'patches/01-c-pipe.patch')