test_git_p4.py #28

#! /usr/bin/env python

from bzrlib.plugins.bzrp4 import bzr2p4
from bzrlib.plugins.bzrp4 import git_p4
from bzrlib.plugins.bzrp4.tests import TestCaseForTwoVcs
from bzrlib.plugins.bzrp4.tests.p4_for_test import P4ForTest
from cStringIO import StringIO
import os
import os.path
import shutil
import stat
import subprocess
import sys
import tempfile
import unittest

class TestGitP4(TestCaseForTwoVcs):

    def __init__(self, method_name):
        TestCaseForTwoVcs.__init__(self, method_name)
        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.git_workdir = os.path.join(self.p4_compatible_cwd, 'git')
        self.git_p4_stdout = StringIO()
        self.git_p4_stderr = StringIO()
        os.mkdir(self.git_workdir)
        self.p4fortest.p4.port = 'rsh:%s -r %s -L log -vserver=3 -i' % ('p4d', self.p4d_workdir)
        # TODO: Remove parameters, and add them to initialization of
        # p4fortest?
        self.p4client_name = 'TestGitP4'
        self.p4fortest.create_client(
            self.p4client_name, self.p4_client_workdir)
        

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


    def _git_p4(self, args):
        orig_dir = os.getcwdu()
        orig_stdout = sys.stdout
        orig_stderr = sys.stderr
        sys.stdout = self.git_p4_stdout
        sys.stderr = self.git_p4_stderr
        code = 0
        try:
            os.chdir(self.git_workdir)
            code = git_p4.main(args)
        finally:
            sys.stdout = orig_stdout
            sys.stderr = orig_stderr
            os.chdir(orig_dir)
        return code


    def test_git_p4__gitify_add(self):
        try:
            self.p4fortest.p4.connect()
            self.p4fortest.sync()
            self.p4fortest.add_file('file_1', 'file_1 contents\n')
            self.p4fortest.submit('gitify_add change 1')
            # Runs git-p4.
            os.environ['P4PORT'] = self.p4fortest.p4.port
            self.assertEqual(0, self._git_p4(['clone', '//depot@all']))
            self.assertDirectoriesEqualModIgnored(
                self.p4_client_workdir,
                os.path.join(self.git_workdir, 'depot'),
                [], ['.git'])
        finally:
            self.p4fortest.p4.disconnect()


    def test_git_p4__fails_on_no_args(self):
        self.assertEqual(2, self._git_p4([]))
        self.assertContainsRe(self.git_p4_stdout.getvalue(), 'usage: ')


    def test_git_p4__syncs_branch_from_p4(self):
        # Uses bzr2p4 to migrate this Bazaar branch to Perforce, so
        # that I have a fixture for this test.
        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])
            # Uses git-p4 to migrate the Perforce depot to git, so that I
            # exercise git-p4.
            self.assertEqual(0, self._git_p4(['clone', '//depot@all']))
            # Asserts that the Perforce depot and the git branch are the
            # same at every revision, so that I can verify git-p4 works.
            self.assertP4DepotAndGitRepositoryEqual()
        finally:
            self.p4fortest.p4.disconnect()


    def assertP4DepotAndGitRepositoryEqual(self):
        p4_change_iter = reversed(self.p4fortest.p4.run_changes())
        for commit_line in self._git_iter_revisions():
            commit = commit_line.rstrip('\n\r')
            self.assertP4AndGitRevisionEqual(
                p4_change_iter.next()['change'], commit)
        self.assertRaises(StopIteration, p4_change_iter.next)


    def _git_iter_revisions(self):
        proc = subprocess.Popen(
            ['git', 'rev-list', '--reverse', 'HEAD'],
            stdout=subprocess.PIPE,
            cwd=os.path.join(self.git_workdir, 'depot'))
        proc_stdout = proc.communicate()[0]
        revisions_text = proc_stdout.rstrip('\n')
        return revisions_text.split('\n')


    def assertP4AndGitRevisionEqual(self, p4_change, git_commit):
        self.p4fortest.p4.run_sync('...@' + p4_change)
        self._git_checkout(git_commit)
        self.assertDirectoriesEqualModIgnored(
            self.p4_client_workdir,
            os.path.join(self.git_workdir, 'depot'),
            ['.bzr2p4', '.bzr2p4.log', '.bzr2p4.revdb'], ['.git'])
        # TODO: Verify that change descriptions are the same in
        # Perforce and Git.


    def _git_checkout(self, commit):
        proc = subprocess.Popen(
            ['git', 'checkout', '-q', commit],
            cwd=os.path.join(self.git_workdir, 'depot'))
        proc.wait()


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