| import unittest |
| import tempfile |
| import shutil |
| import os |
| import subprocess |
| from util.commands import run_cmd, get_output |
| |
| import util.git as git |
| from util.file import touch |
| |
| |
| def getRevisions(dest, branches=None): |
| retval = [] |
| cmd = ['git', 'log', '--pretty=oneline'] |
| if branches: |
| cmd.extend(branches) |
| for line in get_output(cmd, cwd=dest).split('\n'): |
| line = line.strip() |
| if not line: |
| continue |
| rev, ref = line.split(" ", 1) |
| rev = rev.strip() |
| if not rev: |
| continue |
| if rev not in retval: |
| retval.append(rev) |
| retval.reverse() |
| return retval |
| |
| |
| class TestGitFunctions(unittest.TestCase): |
| def test_get_repo_name(self): |
| self.assertEquals(git.get_repo_name("https://git.mozilla.org/releases/gecko.git"), |
| "git.mozilla.org/releases%2Fgecko.git") |
| self.assertEquals(git.get_repo_name("git://git.mozilla.org/releases/gecko.git"), |
| "git.mozilla.org/releases%2Fgecko.git") |
| self.assertEquals(git.get_repo_name("http://git.mozilla.org/releases/gecko.git"), |
| "git.mozilla.org/releases%2Fgecko.git") |
| |
| |
| class TestGit(unittest.TestCase): |
| def setUp(self): |
| self.tmpdir = tempfile.mkdtemp() |
| self.repodir = os.path.join(self.tmpdir, 'repo') |
| run_cmd(['%s/init_gitrepo.sh' % os.path.dirname(__file__), |
| self.repodir]) |
| |
| self.revisions = getRevisions( |
| self.repodir, branches=['master', 'branch2']) |
| self.wc = os.path.join(self.tmpdir, 'wc') |
| self.pwd = os.getcwd() |
| |
| # Make sure we're not setting GIT_SHARE_BASE_DIR |
| if 'GIT_SHARE_BASE_DIR' in os.environ: |
| del os.environ['GIT_SHARE_BASE_DIR'] |
| |
| def tearDown(self): |
| shutil.rmtree(self.tmpdir) |
| os.chdir(self.pwd) |
| |
| def testHasRevision(self): |
| self.assertTrue(git.has_revision(self.repodir, self.revisions[0])) |
| self.assertFalse(git.has_revision(self.repodir, "foooooooooooooooo")) |
| |
| def testClean(self): |
| git.git(self.repodir, self.wc) |
| try: |
| f = open(os.path.join(self.wc, 'blaaah'), 'w') |
| f.write('blah') |
| finally: |
| if not f.closed: |
| f.close() |
| git.clean(self.wc) |
| self.assertFalse(os.path.exists(os.path.join(self.wc, 'blaaah'))) |
| |
| def testCleanFail(self): |
| self.assertRaises(subprocess.CalledProcessError, git.clean, '/tmp') |
| |
| def testClone(self): |
| rev = git.clone(self.repodir, self.wc, update_dest=False) |
| self.assertEquals(rev, None) |
| # This clones the master branch by default |
| # This has the 1st and 3rd commit |
| self.assertEquals( |
| [self.revisions[0], self.revisions[2]], getRevisions(self.wc)) |
| self.assertEquals(sorted(os.listdir(self.wc)), ['.git']) |
| |
| def testCloneIntoNonEmptyDir(self): |
| os.mkdir(self.wc) |
| open(os.path.join(self.wc, 'test.txt'), 'w').write('hello') |
| git.clone(self.repodir, self.wc, update_dest=False) |
| self.failUnless(not os.path.exists(os.path.join(self.wc, 'test.txt'))) |
| |
| def testCloneUpdate(self): |
| rev = git.clone(self.repodir, self.wc, update_dest=True) |
| self.assertEquals(rev, self.revisions[-1]) |
| |
| def testCloneBranch(self): |
| git.clone(self.repodir, self.wc, refname='branch2', update_dest=False) |
| # This should have just the 1st and 2nd commit |
| self.assertEquals( |
| [self.revisions[0], self.revisions[1]], getRevisions(self.wc)) |
| # We should also have no working copy |
| self.failUnless(not os.path.exists(os.path.join(self.wc, 'hello.txt'))) |
| |
| def testUpdateRevision(self): |
| rev = git.clone(self.repodir, self.wc, update_dest=False) |
| self.assertEquals(rev, None) |
| |
| rev = git.update(self.wc, revision=self.revisions[2]) |
| self.assertEquals(rev, self.revisions[2]) |
| |
| def testUpdateBranch(self): |
| rev = git.clone(self.repodir, self.wc, update_dest=False) |
| self.assertEquals(rev, None) |
| |
| rev = git.update(self.wc, refname='branch2') |
| self.assertEquals(rev, self.revisions[1]) |
| |
| def testUpdateBranchDefault(self): |
| rev = git.clone(self.repodir, self.wc, update_dest=False) |
| self.assertEquals(rev, None) |
| |
| rev = git.update(self.wc) |
| self.assertEquals(rev, self.revisions[2]) |
| |
| def testFetchBranch(self): |
| # Clone just the main branch |
| git.clone(self.repodir, self.wc, refname='master', update_dest=False) |
| self.assertEquals( |
| [self.revisions[0], self.revisions[2]], getRevisions(self.wc)) |
| |
| # Now pull in branch2 |
| git.fetch(self.repodir, self.wc, refname='branch2') |
| |
| self.assertEquals(getRevisions(self.wc, branches=[ |
| 'origin/master', 'origin/branch2']), self.revisions) |
| |
| def testFetchAll(self): |
| # Clone just the main branch |
| git.clone(self.repodir, self.wc, update_dest=False) |
| # Now pull in branch2 |
| git.fetch(self.repodir, self.wc, refname='branch2') |
| |
| # Change the original repo |
| newfile = os.path.join(self.repodir, 'newfile') |
| touch(newfile) |
| run_cmd(['git', 'add', 'newfile'], cwd=self.repodir) |
| run_cmd(['git', 'commit', '-q', '-m', 'add newfile'], cwd=self.repodir) |
| |
| # Now pull in everything from master branch |
| git.fetch(self.repodir, self.wc, refname='master') |
| |
| for branch in 'master', 'branch2': |
| self.assertEquals(getRevisions(self.wc, branches=['origin/%s' % branch]), getRevisions(self.repodir, branches=[branch])) |
| |
| # Make sure we actually changed something |
| self.assertNotEqual(getRevisions( |
| self.repodir, branches=['master', 'branch2']), self.revisions) |
| |
| def testGit(self): |
| rev = git.git(self.repodir, self.wc) |
| self.assertEquals(rev, self.revisions[-1]) |
| self.assertEquals(getRevisions(self.wc, branches=[ |
| 'origin/master', 'origin/branch2']), self.revisions) |
| self.assertTrue(os.path.exists(os.path.join(self.wc, 'hello.txt'))) |
| self.assertTrue("Is this thing on" in open( |
| os.path.join(self.wc, 'hello.txt')).read()) |
| |
| # Switch branches, make sure we get the right content |
| rev = git.git(self.repodir, self.wc, refname='branch2') |
| self.assertEquals(rev, self.revisions[1]) |
| self.assertFalse("Is this thing on" in open( |
| os.path.join(self.wc, 'hello.txt')).read()) |
| |
| def testGitCleanDest(self): |
| git.git(self.repodir, self.wc) |
| try: |
| f = open(os.path.join(self.wc, 'blaaah'), 'w') |
| f.write('blah') |
| finally: |
| if not f.closed: |
| f.close() |
| git.git(self.repodir, self.wc, clean_dest=True) |
| self.assertFalse(os.path.exists(os.path.join(self.wc, 'blaaah'))) |
| |
| def testGitRev(self): |
| rev = git.git(self.repodir, self.wc) |
| self.assertEquals(rev, self.revisions[-1]) |
| |
| # Update to a different rev |
| rev = git.git(self.repodir, self.wc, revision=self.revisions[1]) |
| self.assertEquals(rev, self.revisions[1]) |
| |
| # Make a new commit in repodir |
| newfile = os.path.join(self.repodir, 'newfile') |
| touch(newfile) |
| run_cmd(['git', 'add', 'newfile'], cwd=self.repodir) |
| run_cmd(['git', 'commit', '-q', '-m', 'add newfile'], cwd=self.repodir) |
| new_rev = getRevisions(self.repodir)[-1] |
| |
| rev = git.git(self.repodir, self.wc, revision=new_rev) |
| |
| self.assertEquals(new_rev, rev) |
| |
| def testGitBranch(self): |
| rev = git.git(self.repodir, self.wc, refname='branch2') |
| self.assertEquals(rev, self.revisions[1]) |
| self.assertEquals(getRevisions(self.wc, branches=['origin/branch2'] |
| ), getRevisions(self.repodir, branches=['branch2'])) |
| self.assertTrue(os.path.exists(os.path.join(self.wc, 'hello.txt'))) |
| self.assertFalse("Is this thing on" in open( |
| os.path.join(self.wc, 'hello.txt')).read()) |
| |
| # Switch branches, make sure we get all the revisions |
| rev = git.git(self.repodir, self.wc, refname='master') |
| self.assertEquals(rev, self.revisions[-1]) |
| self.assertEquals(getRevisions(self.wc, branches=[ |
| 'origin/master', 'origin/branch2']), self.revisions) |
| self.assertTrue("Is this thing on" in open( |
| os.path.join(self.wc, 'hello.txt')).read()) |
| |
| def testGitShare(self): |
| shareBase = os.path.join(self.tmpdir, 'git-repos') |
| rev = git.git(self.repodir, self.wc, shareBase=shareBase) |
| shareDir = os.path.join(shareBase, git.get_repo_name(self.repodir)) |
| self.assertEquals(rev, self.revisions[-1]) |
| |
| # We should see all the revisions |
| revs = getRevisions( |
| self.wc, branches=['origin/master', 'origin/branch2']) |
| shared_revs = getRevisions( |
| shareDir, branches=['origin/master', 'origin/branch2']) |
| |
| self.assertEquals(revs, shared_revs) |
| self.assertEquals(revs, self.revisions) |
| |
| # Update to a different rev |
| rev = git.git(self.repodir, self.wc, |
| revision=self.revisions[0], shareBase=shareBase) |
| self.assertEquals(rev, self.revisions[0]) |
| self.assertFalse(os.path.exists(os.path.join(self.wc, 'newfile'))) |
| |
| # Add a commit to the original repo |
| newfile = os.path.join(self.repodir, 'newfile') |
| touch(newfile) |
| run_cmd(['git', 'add', 'newfile'], cwd=self.repodir) |
| run_cmd(['git', 'commit', '-q', '-m', 'add newfile'], cwd=self.repodir) |
| new_rev = getRevisions(self.repodir)[-1] |
| |
| # Update to the new rev |
| rev = git.git( |
| self.repodir, self.wc, revision=new_rev, shareBase=shareBase) |
| self.assertEquals(rev, new_rev) |
| self.assertTrue(os.path.exists(os.path.join(self.wc, 'newfile'))) |
| |
| def testGitBadDest(self): |
| # Create self.wc without .git |
| os.makedirs(self.wc) |
| self.assertFalse(git.is_git_repo(self.wc)) |
| |
| rev = git.git(self.repodir, self.wc) |
| self.assertEquals(rev, self.revisions[-1]) |
| self.assertTrue(os.path.exists(os.path.join(self.wc, 'hello.txt'))) |
| self.assertTrue("Is this thing on" in open( |
| os.path.join(self.wc, 'hello.txt')).read()) |
| |
| def testGitBadShare(self): |
| shareBase = os.path.join(self.tmpdir, 'git-repos') |
| open(shareBase, 'w').write("hello!") |
| rev = git.git(self.repodir, self.wc, shareBase=shareBase) |
| self.assertEquals(rev, self.revisions[-1]) |
| |
| def testGitMirrors(self): |
| # Create a bad mirror and a good mirror |
| mirror1 = os.path.join(self.tmpdir, 'mirror1') |
| mirror2 = os.path.join(self.tmpdir, 'mirror2') |
| git.git(self.repodir, mirror2, update_dest=False) |
| |
| rev = git.git(self.repodir, self.wc, mirrors=[mirror1, mirror2]) |
| self.assertEquals(rev, self.revisions[-1]) |
| |
| # Add a commit to the mirror |
| newfile = os.path.join(mirror2, 'newfile') |
| touch(newfile) |
| run_cmd(['git', 'add', 'newfile'], cwd=mirror2) |
| run_cmd(['git', 'commit', '-q', '-m', 'add newfile'], cwd=mirror2) |
| new_rev = getRevisions(mirror2)[-1] |
| |
| # Now clone using the mirror. We should get the new commit that's in |
| # the mirror |
| rev = git.git(self.repodir, self.wc, mirrors=[mirror1, mirror2]) |
| self.assertEquals(rev, new_rev) |
| |
| def testGitBadRev(self): |
| rev = git.git(self.repodir, self.wc) |
| self.assertEquals(rev, self.revisions[-1]) |
| |
| # Now try and update to a bad revision |
| self.assertRaises(subprocess.CalledProcessError, git.git, |
| self.repodir, self.wc, revision="badrev") |
| |
| def testGitTag(self): |
| rev = git.git(self.repodir, self.wc) |
| self.assertEquals(rev, self.revisions[-1]) |
| |
| # Now try and update to a tag |
| rev = git.git(self.repodir, self.wc, revision="TAG1") |
| self.assertEquals(rev, self.revisions[0]) |
| |
| # TODO: Test blowing away shared repo and then updating working copy |
