#! /usr/bin/env python

import bzrlib.branch
# TODO: When I changed these tests to import bzr2p4 rather than forking, I
# started getting errors about permission problems removing test directories.
# I suspect that bzr2p4 is leaking file handles and that they were formerly
# cleaned up with the forked process exited.
from bzrlib.plugins.bzrp4 import bzr2p4
from bzrlib.plugins.bzrp4.tests import TestCaseForTwoVcs
from bzrlib.plugins.bzrp4.tests.p4_for_test import P4ForTest
import os
import os.path
from P4 import P4, P4Exception
import re
import shutil
import stat
import unittest

class TestBzr2P4Blackbox(TestCaseForTwoVcs):

    def __init__(self, method_name):
        TestCaseForTwoVcs.__init__(self, method_name)
        self.p4client_name = 'TestBzr2P4Blackbox'
        self.bzr_branch_path = os.path.join(
            os.path.dirname(os.path.abspath(__file__)), '..')

    def setUp(self):
        TestCaseForTwoVcs.setUp(self)
        self.p4fortest = P4ForTest()
        self.p4_compatible_cwd = os.getcwd()
        self.p4d_workdir = os.path.join(self.p4_compatible_cwd, 'p4d')
        os.mkdir(self.p4d_workdir)
        self.p4_client_workdir = os.path.join(self.p4_compatible_cwd, 'p4')
        os.mkdir(self.p4_client_workdir)
        self.bzr_client_workdir = os.path.join(self.p4_compatible_cwd, 'bzr')
        # bzr branch will create bzr_client_workdir
        self.p4fortest.p4.port = 'rsh:%s -r %s -L log -vserver=3 -i' % ('p4d', self.p4d_workdir)
        self.p4fortest.create_client(
            self.p4client_name, self.p4_client_workdir)
        self.p4fortest.p4.cwd = self.p4_client_workdir
        

    def tearDown(self):
        self.p4fortest.destroy_client()
        TestCaseForTwoVcs.tearDown(self)


    def test_fixture__client_exists(self):
        self.p4fortest.p4.connect()
        self.assertEquals(
            self.p4client_name, self.p4fortest.p4.run_clients()[0]['client'])
        self.p4fortest.p4.disconnect()


    def test_bzr2p4__syncs_self_to_p4(self):
        os.environ['P4CLIENT'] = self.p4client_name
        self.p4fortest.p4.client = self.p4client_name
        self.p4fortest.p4.connect()
        try:
            os.environ['P4PORT'] = self.p4fortest.p4.port
            bzr2p4.main(['-q', self.bzr_branch_path, self.p4_client_workdir])
            self.run_bzr(
                ['branch', self.bzr_branch_path, self.bzr_client_workdir])
            self.assertBzrBranchAndP4DepotEqual()
        finally:
            self.p4fortest.p4.disconnect()


    def test_bzr2p4__rename_a_to_b_and_add_new_a(self):
        relative_path = os.path.basename(self.bzr_client_workdir)
        tree = self.make_branch_and_tree(relative_path)
        files = self.build_tree([relative_path + '/a'])
        tree.add(['a'])
        tree.commit('message')
        tree.rename_one('a', 'b')
        files2 = self.build_tree([relative_path + '/a'])
        tree.add(['a'])
        tree.commit('message')
        # Exercise bzr2p4.
        os.environ['P4CLIENT'] = self.p4client_name
        self.p4fortest.p4.client = self.p4client_name
        os.environ['P4PORT'] = self.p4fortest.p4.port
        bzr2p4.main(['-q', self.bzr_client_workdir, self.p4_client_workdir])
        # Verify it works.
        self.p4fortest.p4.connect()
        try:
            self.assertBzrBranchAndP4DepotEqual()
        finally:
            self.p4fortest.p4.disconnect()


    def test_bzr2p4__delete_a_and_rename_b_to_a(self):
        relative_path = os.path.basename(self.bzr_client_workdir)
        tree = self.make_branch_and_tree(relative_path)
        files = self.build_tree([relative_path + '/a', relative_path + '/b'])
        tree.add(['a', 'b'])
        tree.commit('message')
        tree.remove(['a'], keep_files=False)
        tree.rename_one('b', 'a')
        tree.commit('message')
        # Exercise bzr2p4.
        os.environ['P4CLIENT'] = self.p4client_name
        self.p4fortest.p4.client = self.p4client_name
        os.environ['P4PORT'] = self.p4fortest.p4.port
        bzr2p4.main(['-q', self.bzr_client_workdir, self.p4_client_workdir])
        # Verify it works.
        self.p4fortest.p4.connect()
        try:
            self.assertBzrBranchAndP4DepotEqual()
        finally:
            self.p4fortest.p4.disconnect()


    def test_bzr2p4__migrate__leading(self):
        relative_path = os.path.basename(self.bzr_client_workdir)
        tree = self.make_branch_and_tree(relative_path)
        rev_1_files = self.build_tree([relative_path + '/a'])
        tree.add(['a'])
        tree.commit('rev 1')
        rev_2_files = self.build_tree([relative_path + '/b'])
        tree.add(['b'])
        tree.commit('rev 2')
        # Exercise bzr2p4.
        os.environ['P4CLIENT'] = self.p4client_name
        self.p4fortest.p4.client = self.p4client_name
        os.environ['P4PORT'] = self.p4fortest.p4.port
        bzr2p4.main(['-r', '1..1',
                     '-q',
                     self.bzr_client_workdir, self.p4_client_workdir])
        # Verify it works.
        self.p4fortest.p4.connect()
        try:
            self.assertBzrBranchAndP4DepotRangeEqual(1, 1)
            self.assertEquals(
                '1',
                self.p4fortest.p4.run_changes('-m', '1')[0]['change'])
        finally:
            self.p4fortest.p4.disconnect()
        

    def test_bzr2p4__migrate__trailing(self):
        relative_path = os.path.basename(self.bzr_client_workdir)
        tree = self.make_branch_and_tree(relative_path)
        rev_1_files = self.build_tree([relative_path + '/a'])
        tree.add(['a'])
        tree.commit('rev 1')
        rev_2_files = self.build_tree([relative_path + '/b'])
        tree.add(['b'])
        tree.commit('rev 2')
        # Exercise bzr2p4.
        os.environ['P4CLIENT'] = self.p4client_name
        self.p4fortest.p4.client = self.p4client_name
        os.environ['P4PORT'] = self.p4fortest.p4.port
        bzr2p4.main(['-r', '1..1',
                     '-q',
                     self.bzr_client_workdir, self.p4_client_workdir])
        # Make it look like the next migration is the first to this client
        # workspace.
        os.remove(self.p4_client_workdir + '/.bzr2p4.revdb')
        bzr2p4.main(['-r', '2..2',
                     '-q',
                     self.bzr_client_workdir, self.p4_client_workdir])
        # Verify it works.
        self.p4fortest.p4.connect()
        try:
            self.assertBzrBranchAndP4DepotRangeEqual(2, 2)
            self.assertEquals(
                '2',
                self.p4fortest.p4.run_changes('-m', '1')[0]['change'])
        finally:
            self.p4fortest.p4.disconnect()


    def test_bzr2p4__migrate__depot_newer_than_parent_of_start(self):
        relative_path = os.path.basename(self.bzr_client_workdir)
        tree = self.make_branch_and_tree(relative_path)
        rev_1_files = self.build_tree([relative_path + '/a'])
        tree.add(['a'])
        tree.commit('rev 1')
        rev_2_files = self.build_tree([relative_path + '/b'])
        tree.add(['b'])
        tree.commit('rev 2')
        # Exercise bzr2p4.
        os.environ['P4CLIENT'] = self.p4client_name
        self.p4fortest.p4.client = self.p4client_name
        os.environ['P4PORT'] = self.p4fortest.p4.port
        bzr2p4.main(['-r', '1..1',
                     '-q',
                     self.bzr_client_workdir, self.p4_client_workdir])
        # Make it look like the next migration is the first to this client
        # workspace.
        os.remove(self.p4_client_workdir + '/.bzr2p4.revdb')
        self.p4fortest.p4.connect()
        try:
            self.p4fortest.add_file('c', 'c contents')
            self.p4fortest.submit('Perforce change 2')
            # Watch out!  bzr2p4 doesn't protect you from submitting
            # revisions when there are newer revisions in the depot.
#             self.assertRaises(
#                 Exception,
#                 bzr2p4.main,
#                 ['-r', '2..2',
#                  '-q',
#                  self.bzr_client_workdir, self.p4_client_workdir])
            bzr2p4.main(
                ['-r', '2..2',
                 '-q',
                 self.bzr_client_workdir, self.p4_client_workdir])
        finally:
            self.p4fortest.p4.disconnect()


    def test_bzr2p4__migrate__parent_of_start_not_in_depot(self):
        relative_path = os.path.basename(self.bzr_client_workdir)
        tree = self.make_branch_and_tree(relative_path)
        rev_1_files = self.build_tree([relative_path + '/a'])
        tree.add(['a'])
        tree.commit('rev 1')
        rev_2_files = self.build_tree([relative_path + '/b'])
        tree.add(['b'])
        tree.commit('rev 2')
        rev_3_files = self.build_tree([relative_path + '/c'])
        tree.add(['c'])
        tree.commit('rev 3')
        # Exercise bzr2p4.
        os.environ['P4CLIENT'] = self.p4client_name
        self.p4fortest.p4.client = self.p4client_name
        os.environ['P4PORT'] = self.p4fortest.p4.port
        bzr2p4.main(['-r', '1..1',
                     '-q',
                     self.bzr_client_workdir, self.p4_client_workdir])
        # Make it look like the next migration is the first to this client
        # workspace.
        os.remove(self.p4_client_workdir + '/.bzr2p4.revdb')
        # Watch out!  bzr2p4 doesn't protect you from submitting a revision
        # whose parent isn't in the depot.
#         self.assertRaises(
#             Exception,
#             bzr2p4.main,
#             ['-r', '3..3',
#              '-q',
#              self.bzr_client_workdir, self.p4_client_workdir])
        bzr2p4.main(
            ['-r', '3..3',
             '-q',
             self.bzr_client_workdir, self.p4_client_workdir])


    def assertBzrBranchAndP4DepotEqual(self):
        self.assertBzrBranchAndP4DepotRangeEqual(None, None)


    def assertBzrBranchAndP4DepotRangeEqual(self, start_revision, end_revision):
        bzr_revision_iter = self._bzr_iter_revisions_range(
            start_revision, end_revision)
        
        p4_change_iter = reversed(self.p4fortest.p4.run_changes(
                *self._p4_changes_args(start_revision, end_revision)))

        # avoid warnings about files already up to date
        self.p4fortest.p4.run_sync('...@0')

        try:
            while True:
                bzr_revision = bzr_revision_iter.next()
                try:
                    p4_change = p4_change_iter.next()
                    while (_is_incomplete_change(p4_change['desc'])):
                        p4_change = p4_change_iter.next()
                except Exception, e:
                    raise e
                self.assertBzrAndP4RevisionEqual(
                    bzr_revision, p4_change['change'])
        except StopIteration, e:
            self.assertRaises(StopIteration, p4_change_iter.next)


    def _p4_changes_args(self, start_revision, end_revision):
        args = ['-l']
        if (None != start_revision and None != end_revision):
            args.append('@%d,%d' % (start_revision, end_revision))
        return args


    def _bzr_iter_revisions(self):
        self._bzr_iter_revisions_range(None, None)


    def _bzr_iter_revisions_range(self, start_revision, end_revision):
        branch = bzrlib.branch.Branch.open(self.bzr_client_workdir)
        branch.lock_read()
        revisions = [tuple[0] for tuple in bzrlib.log.calculate_view_revisions(
            branch, start_revision, end_revision, 'reverse', None, True)]
        revisions.reverse()
        branch.unlock()
        return iter(revisions)


    def assertBzrAndP4RevisionEqual(self, bzr_revision_id, p4_change):
        self.p4fortest.p4.run_sync('...@' + p4_change)
        self.run_bzr(
            ['revert', '-r', 'revid:%s' % bzr_revision_id,
             self.bzr_client_workdir])
        self.assertDirectoriesEqualModIgnored(
            self.bzr_client_workdir, self.p4_client_workdir,
            ['.bzr'], ['.bzr2p4', '.bzr2p4.log', '.bzr2p4.revdb'])
        # TODO: Verify that change descriptions are the same in
        # Perforce and Bazaar.


def _is_incomplete_change(description):
    matches = re.finditer('bzr: .+, part (?P<part>\d+) of (?P<whole>\d+)',
                     description)
    match = None
    # Examine the last match.
    for match in matches:
        pass
    if None == match:
        return False
    return int(match.group('part')) < int(match.group('whole'))


if __name__ == '__main__':
    unittest.main()