| import unittest |
| import tempfile |
| import shutil |
| import os |
| import subprocess |
| import time |
| import util.hg as hg |
| from util.hg import clone, pull, update, hg_ver, mercurial, _make_absolute, \ |
| share, push, apply_and_push, HgUtilError, make_hg_url, get_branch, purge, \ |
| get_branches, path, init, unbundle, adjust_paths, is_hg_cset, commit, tag, \ |
| get_hg_output, has_rev |
| from util.file import touch |
| from util.commands import run_cmd, get_output |
| |
| from mock import patch |
| |
| |
| def getRevisions(dest): |
| retval = [] |
| for rev in get_output(['hg', 'log', '-R', dest, '--template', '{node|short}\n']).split('\n'): |
| rev = rev.strip() |
| if not rev: |
| continue |
| retval.append(rev) |
| return retval |
| |
| |
| def getRevInfo(dest, rev): |
| output = get_output(['hg', 'log', '-R', dest, '-r', rev, '--template', |
| '{author}\n{desc}\n{tags}']).splitlines() |
| info = { |
| 'user': output[0], |
| 'msg': output[1], |
| 'tags': [] |
| } |
| if len(output) > 2: |
| info['tags'] = output[2].split() |
| return info |
| |
| |
| def getTags(dest): |
| tags = [] |
| for t in get_output(['hg', 'tags', '-R', dest]).splitlines(): |
| tags.append(t.split()[0]) |
| return tags |
| |
| |
| def yesterday_timestamp(): |
| """returns a valid yesterday timestamp for touch""" |
| yesterday = time.time() - 86400 |
| return (yesterday, yesterday) |
| |
| |
| class TestMakeAbsolute(unittest.TestCase): |
| def testAbsolutePath(self): |
| self.assertEquals(_make_absolute("/foo/bar"), "/foo/bar") |
| |
| def testStripTrailingSlash(self): |
| self.assertEquals(_make_absolute("/foo/bar/"), "/foo/bar") |
| |
| def testRelativePath(self): |
| self.assertEquals( |
| _make_absolute("foo/bar"), os.path.abspath("foo/bar")) |
| |
| def testHTTPPaths(self): |
| self.assertEquals(_make_absolute("http://foo/bar"), "http://foo/bar") |
| |
| def testAbsoluteFilePath(self): |
| self.assertEquals(_make_absolute("file:///foo/bar"), "file:///foo/bar") |
| |
| def testRelativeFilePath(self): |
| self.assertEquals(_make_absolute( |
| "file://foo/bar"), "file://%s/foo/bar" % os.getcwd()) |
| |
| |
| class TestIsHgCset(unittest.TestCase): |
| |
| def testShortCset(self): |
| self.assertTrue(is_hg_cset('fd06332733e5')) |
| |
| def testLongCset(self): |
| self.assertTrue(is_hg_cset('bf37aabfd9367aec573487ebe1f784108bbef73f')) |
| |
| def testMediumCset(self): |
| self.assertTrue(is_hg_cset('1e3391794bac9d0e707a7681de3')) |
| |
| def testInt(self): |
| self.assertFalse(is_hg_cset(1234567890)) |
| |
| def testTag(self): |
| self.assertFalse(is_hg_cset('FIREFOX_50_0_RELEASE')) |
| |
| def testDefault(self): |
| self.assertFalse(is_hg_cset('default')) |
| |
| def testTip(self): |
| self.assertFalse(is_hg_cset('tip')) |
| |
| def testBranch(self): |
| self.assertFalse(is_hg_cset('GECKO77_203512230833_RELBRANCH')) |
| |
| |
| class TestHg(unittest.TestCase): |
| def setUp(self): |
| tmpdir = None |
| if os.path.isdir("/dev/shm"): |
| tmpdir = "/dev/shm" |
| self.tmpdir = tempfile.mkdtemp(dir=tmpdir) |
| self.pwd = os.getcwd() |
| os.chdir(self.tmpdir) |
| self.repodir = os.path.join(self.tmpdir, 'repo') |
| # Have a stable hgrc to test with |
| os.environ['HGRCPATH'] = os.path.join(os.path.dirname(__file__), "hgrc") |
| run_cmd(['%s/init_hgrepo.sh' % os.path.dirname(__file__), |
| self.repodir]) |
| |
| self.revisions = getRevisions(self.repodir) |
| self.wc = os.path.join(self.tmpdir, 'wc') |
| self.sleep_patcher = patch('time.sleep') |
| self.sleep_patcher.start() |
| hg.RETRY_ATTEMPTS = 2 |
| |
| def tearDown(self): |
| shutil.rmtree(self.tmpdir) |
| os.chdir(self.pwd) |
| self.sleep_patcher.stop() |
| |
| def testGetBranch(self): |
| clone(self.repodir, self.wc) |
| b = get_branch(self.wc) |
| self.assertEquals(b, 'default') |
| |
| def testGetBranches(self): |
| clone(self.repodir, self.wc) |
| branches = get_branches(self.wc) |
| self.assertEquals(sorted(branches), sorted(["branch2", "default"])) |
| |
| def testClone(self): |
| rev = clone(self.repodir, self.wc, update_dest=False) |
| self.assertEquals(rev, None) |
| self.assertEquals(self.revisions, getRevisions(self.wc)) |
| self.assertEquals(sorted(os.listdir(self.wc)), ['.hg']) |
| |
| def testCloneIntoNonEmptyDir(self): |
| os.mkdir(self.wc) |
| open(os.path.join(self.wc, 'test.txt'), 'w').write('hello') |
| 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 = clone(self.repodir, self.wc, update_dest=True) |
| self.assertEquals(rev, self.revisions[0]) |
| |
| def testCloneBranch(self): |
| clone(self.repodir, self.wc, branch='branch2', |
| update_dest=False, clone_by_rev=True) |
| # On hg 1.6, we should only have a subset of the revisions |
| if hg_ver() >= (1, 6, 0): |
| self.assertEquals(self.revisions[1:], |
| getRevisions(self.wc)) |
| else: |
| self.assertEquals(self.revisions, |
| getRevisions(self.wc)) |
| |
| def testCloneUpdateBranch(self): |
| rev = clone(self.repodir, os.path.join(self.tmpdir, 'wc'), |
| branch="branch2", update_dest=True, clone_by_rev=True) |
| self.assertEquals(rev, self.revisions[1], self.revisions) |
| |
| def testCloneRevision(self): |
| clone(self.repodir, self.wc, |
| revision=self.revisions[0], update_dest=False, |
| clone_by_rev=True) |
| # We'll only get a subset of the revisions |
| self.assertEquals(self.revisions[:1] + self.revisions[2:], |
| getRevisions(self.wc)) |
| |
| def testUpdateRevision(self): |
| rev = clone(self.repodir, self.wc, update_dest=False) |
| self.assertEquals(rev, None) |
| |
| rev = update(self.wc, revision=self.revisions[1]) |
| self.assertEquals(rev, self.revisions[1]) |
| |
| def testPull(self): |
| # Clone just the first rev |
| clone(self.repodir, self.wc, revision=self.revisions[-1], |
| update_dest=False, clone_by_rev=True) |
| self.assertEquals(getRevisions(self.wc), self.revisions[-1:]) |
| |
| # Now pull in new changes |
| rev = pull(self.repodir, self.wc, update_dest=False) |
| self.assertEquals(rev, None) |
| self.assertEquals(getRevisions(self.wc), self.revisions) |
| |
| def testPullRevision(self): |
| # Clone just the first rev |
| clone(self.repodir, self.wc, revision=self.revisions[-1], |
| update_dest=False, clone_by_rev=True) |
| self.assertEquals(getRevisions(self.wc), self.revisions[-1:]) |
| |
| # Now pull in just the last revision |
| rev = pull(self.repodir, self.wc, revision=self.revisions[ |
| 0], update_dest=False) |
| self.assertEquals(rev, None) |
| |
| # We'll be missing the middle revision (on another branch) |
| self.assertEquals( |
| getRevisions(self.wc), self.revisions[:1] + self.revisions[2:]) |
| |
| def testPullBranch(self): |
| # Clone just the first rev |
| clone(self.repodir, self.wc, revision=self.revisions[-1], |
| update_dest=False, clone_by_rev=True) |
| self.assertEquals(getRevisions(self.wc), self.revisions[-1:]) |
| |
| # Now pull in the other branch |
| rev = pull(self.repodir, self.wc, branch="branch2", update_dest=False) |
| self.assertEquals(rev, None) |
| |
| # On hg 1.6, we'll be missing the last revision (on another branch) |
| if hg_ver() >= (1, 6, 0): |
| self.assertEquals(getRevisions(self.wc), self.revisions[1:]) |
| else: |
| self.assertEquals(getRevisions(self.wc), self.revisions) |
| |
| def testPullTag(self): |
| clone(self.repodir, self.wc) |
| run_cmd(['hg', 'tag', '-f', 'TAG1'], cwd=self.repodir) |
| revisions = getRevisions(self.repodir) |
| |
| rev = pull(self.repodir, self.wc, revision='TAG1') |
| self.assertEquals(rev, revisions[1]) |
| self.assertEquals(getRevisions(self.wc), revisions) |
| |
| def testPullTip(self): |
| clone(self.repodir, self.wc) |
| run_cmd(['hg', 'tag', '-f', 'TAG1'], cwd=self.repodir) |
| revisions = getRevisions(self.repodir) |
| |
| rev = pull(self.repodir, self.wc, revision='tip') |
| self.assertEquals(rev, revisions[0]) |
| self.assertEquals(getRevisions(self.wc), revisions) |
| |
| def testPullDefault(self): |
| clone(self.repodir, self.wc) |
| run_cmd(['hg', 'tag', '-f', 'TAG1'], cwd=self.repodir) |
| revisions = getRevisions(self.repodir) |
| |
| rev = pull(self.repodir, self.wc, revision='default') |
| self.assertEquals(rev, revisions[0]) |
| self.assertEquals(getRevisions(self.wc), revisions) |
| |
| def testPullUnrelated(self): |
| # Create a new repo |
| repo2 = os.path.join(self.tmpdir, 'repo2') |
| run_cmd(['%s/init_hgrepo.sh' % os.path.dirname(__file__), repo2], |
| env={'extra': 'unrelated'}) |
| |
| self.assertNotEqual(self.revisions, getRevisions(repo2)) |
| |
| # Clone the original repo |
| clone(self.repodir, self.wc, update_dest=False) |
| |
| # Try and pull in changes from the new repo |
| self.assertRaises(subprocess.CalledProcessError, pull, |
| repo2, self.wc, update_dest=False) |
| |
| def testShareUnrelated(self): |
| # Create a new repo |
| repo2 = os.path.join(self.tmpdir, 'repo2') |
| run_cmd(['%s/init_hgrepo.sh' % os.path.dirname(__file__), repo2], |
| env={'extra': 'share_unrelated'}) |
| |
| self.assertNotEqual(self.revisions, getRevisions(repo2)) |
| |
| shareBase = os.path.join(self.tmpdir, 'share') |
| |
| # Clone the original repo |
| mercurial(self.repodir, self.wc, shareBase=shareBase) |
| |
| # Clone the new repo |
| mercurial(repo2, self.wc, shareBase=shareBase) |
| |
| self.assertEquals(getRevisions(self.wc), getRevisions(repo2)) |
| |
| def testShareExtraFiles(self): |
| shareBase = os.path.join(self.tmpdir, 'share') |
| backup = os.path.join(self.tmpdir, 'backup') |
| |
| # Clone the original repo |
| mercurial(self.repodir, self.wc, shareBase=shareBase) |
| |
| clone(self.repodir, backup) |
| |
| # Make the working repo have a new file. We need it to have an earlier |
| # timestamp (yesterday) to trigger the odd behavior in hg |
| newfile = os.path.join(self.wc, 'newfile') |
| touch(newfile, timestamp=yesterday_timestamp()) |
| run_cmd(['hg', 'add', 'newfile'], cwd=self.wc) |
| run_cmd(['hg', 'commit', '-m', '"add newfile"'], cwd=self.wc) |
| |
| # Reset the share base to remove the 'add newfile' commit. We |
| # overwrite repodir with the backup that doesn't have the commit, |
| # then clone the repodir to a throwaway dir to create the new |
| # shareBase. Now self.wc still points to shareBase, but the |
| # changeset that self.wc was on is lost. |
| shutil.rmtree(self.repodir) |
| shutil.rmtree(shareBase) |
| clone(backup, self.repodir) |
| throwaway = os.path.join(self.tmpdir, 'throwaway') |
| mercurial(self.repodir, throwaway, shareBase=shareBase) |
| |
| # Try and update our working copy |
| mercurial(self.repodir, self.wc, shareBase=shareBase) |
| |
| self.assertFalse(os.path.exists(os.path.join(self.wc, 'newfile'))) |
| |
| def testShareExtraFilesReset(self): |
| shareBase = os.path.join(self.tmpdir, 'share') |
| |
| # Clone the original repo |
| mercurial(self.repodir, self.wc, shareBase=shareBase) |
| |
| # Reset the repo |
| run_cmd( |
| ['%s/init_hgrepo.sh' % os.path.dirname(__file__), self.repodir], |
| env={'extra': 'extra_files_reset'}) |
| |
| # Make the working repo have a new file. We need it to have an earlier |
| # timestamp (yesterday) to trigger the odd behavior in hg |
| newfile = os.path.join(self.wc, 'newfile') |
| touch(newfile, timestamp=yesterday_timestamp()) |
| run_cmd(['hg', 'add', 'newfile'], cwd=self.wc) |
| run_cmd(['hg', 'commit', '-m', '"add newfile"'], cwd=self.wc) |
| |
| # Try and update our working copy |
| mercurial(self.repodir, self.wc, shareBase=shareBase) |
| |
| self.assertFalse(os.path.exists(os.path.join(self.wc, 'newfile'))) |
| |
| def testShareReset(self): |
| shareBase = os.path.join(self.tmpdir, 'share') |
| |
| # Clone the original repo |
| mercurial(self.repodir, self.wc, shareBase=shareBase) |
| |
| old_revs = self.revisions[:] |
| |
| # Reset the repo |
| run_cmd( |
| ['%s/init_hgrepo.sh' % os.path.dirname(__file__), self.repodir], |
| env={'extra': 'share_reset'}) |
| |
| self.assertNotEqual(old_revs, getRevisions(self.repodir)) |
| |
| # Try and update our working copy |
| mercurial(self.repodir, self.wc, shareBase=shareBase) |
| |
| self.assertEquals(getRevisions(self.repodir), getRevisions(self.wc)) |
| self.assertNotEqual(old_revs, getRevisions(self.wc)) |
| |
| def testPush(self): |
| clone(self.repodir, self.wc, revision=self.revisions[-2], |
| clone_by_rev=True) |
| push(src=self.repodir, remote=self.wc) |
| self.assertEquals(getRevisions(self.wc), self.revisions) |
| |
| def testPushWithBranch(self): |
| clone(self.repodir, self.wc, revision=self.revisions[-1], |
| clone_by_rev=True) |
| push(src=self.repodir, remote=self.wc, branch='branch2') |
| push(src=self.repodir, remote=self.wc, branch='default') |
| self.assertEquals(getRevisions(self.wc), self.revisions) |
| |
| def testPushWithRevision(self): |
| clone(self.repodir, self.wc, revision=self.revisions[-1], |
| clone_by_rev=True) |
| push(src=self.repodir, remote=self.wc, revision=self.revisions[-2]) |
| self.assertEquals(getRevisions(self.wc), self.revisions[-2:]) |
| |
| def testPurgeUntrackedFile(self): |
| clone(self.repodir, self.wc) |
| fileToPurge = os.path.join(self.wc, 'fileToPurge') |
| with file(fileToPurge, 'a') as f: |
| f.write('purgeme') |
| purge(self.wc) |
| self.assertFalse(os.path.exists(fileToPurge)) |
| |
| def testPurgeUntrackedDirectory(self): |
| clone(self.repodir, self.wc) |
| directoryToPurge = os.path.join(self.wc, 'directoryTopPurge') |
| os.makedirs(directoryToPurge) |
| purge(directoryToPurge) |
| self.assertFalse(os.path.isdir(directoryToPurge)) |
| |
| def testPurgeVeryLongPath(self): |
| clone(self.repodir, self.wc) |
| # now create a very long path name |
| longPath = self.wc |
| for new_dir in xrange(1, 64): |
| longPath = os.path.join(longPath, str(new_dir)) |
| os.makedirs(longPath) |
| self.assertTrue(os.path.isdir(longPath)) |
| purge(self.wc) |
| self.assertFalse(os.path.isdir(longPath)) |
| |
| def testPurgeAFreshClone(self): |
| clone(self.repodir, self.wc) |
| purge(self.wc) |
| self.assertTrue(os.path.exists(os.path.join(self.wc, 'hello.txt'))) |
| |
| def testPurgeTrackedFile(self): |
| clone(self.repodir, self.wc) |
| fileToModify = os.path.join(self.wc, 'hello.txt') |
| with open(fileToModify, 'w') as f: |
| f.write('hello!') |
| purge(self.wc) |
| with open(fileToModify, 'r') as f: |
| content = f.read() |
| self.assertEqual(content, 'hello!') |
| |
| def testMercurial(self): |
| rev = mercurial(self.repodir, self.wc) |
| self.assertEquals(rev, self.revisions[0]) |
| |
| def testPushNewBranchesNotAllowed(self): |
| clone(self.repodir, self.wc, revision=self.revisions[0], |
| clone_by_rev=True) |
| self.assertRaises(Exception, push, self.repodir, self.wc, |
| push_new_branches=False) |
| |
| def testPushWithForce(self): |
| clone(self.repodir, self.wc, revision=self.revisions[0], |
| clone_by_rev=True) |
| newfile = os.path.join(self.wc, 'newfile') |
| touch(newfile) |
| run_cmd(['hg', 'add', 'newfile'], cwd=self.wc) |
| run_cmd(['hg', 'commit', '-m', '"re-add newfile"'], cwd=self.wc) |
| push(self.repodir, self.wc, push_new_branches=False, force=True) |
| |
| def testPushForceFail(self): |
| clone(self.repodir, self.wc, revision=self.revisions[0], |
| clone_by_rev=True) |
| newfile = os.path.join(self.wc, 'newfile') |
| touch(newfile) |
| run_cmd(['hg', 'add', 'newfile'], cwd=self.wc) |
| run_cmd(['hg', 'commit', '-m', '"add newfile"'], cwd=self.wc) |
| self.assertRaises(Exception, push, self.repodir, self.wc, |
| push_new_branches=False, force=False) |
| |
| def testMercurialWithNewShare(self): |
| shareBase = os.path.join(self.tmpdir, 'share') |
| sharerepo = os.path.join(shareBase, self.repodir.lstrip("/")) |
| os.mkdir(shareBase) |
| mercurial(self.repodir, self.wc, shareBase=shareBase) |
| self.assertEquals(getRevisions(self.repodir), getRevisions(self.wc)) |
| self.assertEquals(getRevisions(self.repodir), getRevisions(sharerepo)) |
| |
| def testMercurialWithShareBaseInEnv(self): |
| shareBase = os.path.join(self.tmpdir, 'share') |
| sharerepo = os.path.join(shareBase, self.repodir.lstrip("/")) |
| os.mkdir(shareBase) |
| try: |
| os.environ['HG_SHARE_BASE_DIR'] = shareBase |
| mercurial(self.repodir, self.wc) |
| self.assertEquals( |
| getRevisions(self.repodir), getRevisions(self.wc)) |
| self.assertEquals( |
| getRevisions(self.repodir), getRevisions(sharerepo)) |
| finally: |
| del os.environ['HG_SHARE_BASE_DIR'] |
| |
| def testMercurialWithExistingShare(self): |
| shareBase = os.path.join(self.tmpdir, 'share') |
| sharerepo = os.path.join(shareBase, self.repodir.lstrip("/")) |
| os.mkdir(shareBase) |
| mercurial(self.repodir, sharerepo) |
| open(os.path.join(self.repodir, 'test.txt'), 'w').write('hello!') |
| run_cmd(['hg', 'add', 'test.txt'], cwd=self.repodir) |
| run_cmd(['hg', 'commit', '-m', 'adding changeset'], cwd=self.repodir) |
| mercurial(self.repodir, self.wc, shareBase=shareBase) |
| self.assertEquals(getRevisions(self.repodir), getRevisions(self.wc)) |
| self.assertEquals(getRevisions(self.repodir), getRevisions(sharerepo)) |
| |
| def testMercurialRelativeDir(self): |
| os.chdir(os.path.dirname(self.repodir)) |
| |
| repo = os.path.basename(self.repodir) |
| wc = os.path.basename(self.wc) |
| |
| rev = mercurial(repo, wc, revision=self.revisions[-1]) |
| self.assertEquals(rev, self.revisions[-1]) |
| open(os.path.join(self.wc, 'test.txt'), 'w').write("hello!") |
| |
| rev = mercurial(repo, wc) |
| self.assertEquals(rev, self.revisions[0]) |
| # Make sure our local file didn't go away |
| self.failUnless(os.path.exists(os.path.join(self.wc, 'test.txt'))) |
| |
| def testMercurialUpdateTip(self): |
| rev = mercurial(self.repodir, self.wc, revision=self.revisions[-1]) |
| self.assertEquals(rev, self.revisions[-1]) |
| open(os.path.join(self.wc, 'test.txt'), 'w').write("hello!") |
| |
| rev = mercurial(self.repodir, self.wc) |
| self.assertEquals(rev, self.revisions[0]) |
| # Make sure our local file didn't go away |
| self.failUnless(os.path.exists(os.path.join(self.wc, 'test.txt'))) |
| |
| def testMercurialUpdateRev(self): |
| rev = mercurial(self.repodir, self.wc, revision=self.revisions[-1]) |
| self.assertEquals(rev, self.revisions[-1]) |
| open(os.path.join(self.wc, 'test.txt'), 'w').write("hello!") |
| |
| rev = mercurial(self.repodir, self.wc, revision=self.revisions[0]) |
| self.assertEquals(rev, self.revisions[0]) |
| # Make sure our local file didn't go away |
| self.failUnless(os.path.exists(os.path.join(self.wc, 'test.txt'))) |
| |
| # TODO: this test doesn't seem to be compatible with mercurial()'s |
| # share() usage, and fails when HG_SHARE_BASE_DIR is set |
| def testMercurialChangeRepo(self): |
| # Create a new repo |
| old_env = os.environ.copy() |
| if 'HG_SHARE_BASE_DIR' in os.environ: |
| del os.environ['HG_SHARE_BASE_DIR'] |
| |
| try: |
| repo2 = os.path.join(self.tmpdir, 'repo2') |
| run_cmd(['%s/init_hgrepo.sh' % os.path.dirname(__file__), repo2], |
| env={'extra': 'change_repo'}) |
| |
| self.assertNotEqual(self.revisions, getRevisions(repo2)) |
| |
| # Clone the original repo |
| mercurial(self.repodir, self.wc) |
| self.assertEquals(getRevisions(self.wc), self.revisions) |
| open(os.path.join(self.wc, 'test.txt'), 'w').write("hello!") |
| |
| # Clone the new one |
| mercurial(repo2, self.wc) |
| self.assertEquals(getRevisions(self.wc), getRevisions(repo2)) |
| # Make sure our local file went away |
| self.failUnless( |
| not os.path.exists(os.path.join(self.wc, 'test.txt'))) |
| finally: |
| os.environ.clear() |
| os.environ.update(old_env) |
| |
| def testMakeHGUrl(self): |
| # construct an hg url specific to revision, branch and filename and try |
| # to pull it down |
| file_url = make_hg_url( |
| "hg.mozilla.org", |
| '//build/tools/', |
| revision='FIREFOX_3_6_12_RELEASE', |
| filename="/lib/python/util/hg.py" |
| ) |
| expected_url = "https://hg.mozilla.org/build/tools/raw-file/FIREFOX_3_6_12_RELEASE/lib/python/util/hg.py" |
| self.assertEquals(file_url, expected_url) |
| |
| def testMakeHGUrlNoFilename(self): |
| file_url = make_hg_url( |
| "hg.mozilla.org", |
| "/build/tools", |
| revision="default" |
| ) |
| expected_url = "https://hg.mozilla.org/build/tools/rev/default" |
| self.assertEquals(file_url, expected_url) |
| |
| def testMakeHGUrlNoRevisionNoFilename(self): |
| repo_url = make_hg_url( |
| "hg.mozilla.org", |
| "/build/tools" |
| ) |
| expected_url = "https://hg.mozilla.org/build/tools" |
| self.assertEquals(repo_url, expected_url) |
| |
| def testMakeHGUrlDifferentProtocol(self): |
| repo_url = make_hg_url( |
| "hg.mozilla.org", |
| "/build/tools", |
| protocol='ssh' |
| ) |
| expected_url = "ssh://hg.mozilla.org/build/tools" |
| self.assertEquals(repo_url, expected_url) |
| |
| def testShareRepo(self): |
| repo3 = os.path.join(self.tmpdir, 'repo3') |
| share(self.repodir, repo3) |
| # make sure shared history is identical |
| self.assertEquals(self.revisions, getRevisions(repo3)) |
| |
| def testMercurialShareOutgoing(self): |
| # ensure that outgoing changesets in a shared clone affect the shared |
| # history |
| repo5 = os.path.join(self.tmpdir, 'repo5') |
| repo6 = os.path.join(self.tmpdir, 'repo6') |
| mercurial(self.repodir, repo5) |
| share(repo5, repo6) |
| open(os.path.join(repo6, 'test.txt'), 'w').write("hello!") |
| # modify the history of the new clone |
| run_cmd(['hg', 'add', 'test.txt'], cwd=repo6) |
| run_cmd(['hg', 'commit', '-m', 'adding changeset'], cwd=repo6) |
| self.assertNotEquals(self.revisions, getRevisions(repo6)) |
| self.assertNotEquals(self.revisions, getRevisions(repo5)) |
| self.assertEquals(getRevisions(repo5), getRevisions(repo6)) |
| |
| def testApplyAndPush(self): |
| clone(self.repodir, self.wc) |
| |
| def c(repo, attempt): |
| run_cmd(['hg', 'tag', '-f', 'TEST'], cwd=repo) |
| apply_and_push(self.wc, self.repodir, c) |
| self.assertEquals(getRevisions(self.wc), getRevisions(self.repodir)) |
| |
| def testApplyAndPushFail(self): |
| clone(self.repodir, self.wc) |
| |
| def c(repo, attempt, remote): |
| run_cmd(['hg', 'tag', '-f', 'TEST'], cwd=repo) |
| run_cmd(['hg', 'tag', '-f', 'CONFLICTING_TAG'], cwd=remote) |
| self.assertRaises(HgUtilError, apply_and_push, self.wc, self.repodir, |
| lambda r, a: c(r, a, self.repodir), max_attempts=2) |
| |
| def testApplyAndPushWithRebase(self): |
| clone(self.repodir, self.wc) |
| |
| def c(repo, attempt, remote): |
| run_cmd(['hg', 'tag', '-f', 'TEST'], cwd=repo) |
| if attempt == 1: |
| run_cmd(['hg', 'rm', 'hello.txt'], cwd=remote) |
| run_cmd(['hg', 'commit', '-m', 'test'], cwd=remote) |
| apply_and_push(self.wc, self.repodir, |
| lambda r, a: c(r, a, self.repodir), max_attempts=2) |
| self.assertEquals(getRevisions(self.wc), getRevisions(self.repodir)) |
| |
| def testApplyAndPushRebaseFails(self): |
| clone(self.repodir, self.wc) |
| |
| def c(repo, attempt, remote): |
| run_cmd(['hg', 'tag', '-f', 'TEST'], cwd=repo) |
| if attempt in (1, 2): |
| run_cmd(['hg', 'tag', '-f', 'CONFLICTING_TAG'], cwd=remote) |
| apply_and_push(self.wc, self.repodir, |
| lambda r, a: c(r, a, self.repodir), max_attempts=3) |
| self.assertEquals(getRevisions(self.wc), getRevisions(self.repodir)) |
| |
| def testApplyAndPushOnBranch(self): |
| clone(self.repodir, self.wc) |
| |
| def c(repo, attempt): |
| run_cmd(['hg', 'branch', 'branch3'], cwd=repo) |
| run_cmd(['hg', 'tag', '-f', 'TEST'], cwd=repo) |
| apply_and_push(self.wc, self.repodir, c) |
| self.assertEquals(getRevisions(self.wc), getRevisions(self.repodir)) |
| |
| def testApplyAndPushWithNoChange(self): |
| clone(self.repodir, self.wc) |
| |
| def c(r, a): |
| pass |
| self.assertRaises( |
| HgUtilError, apply_and_push, self.wc, self.repodir, c) |
| |
| def testApplyAndPushForce(self): |
| clone(self.repodir, self.wc) |
| |
| def c(repo, attempt, remote, local): |
| newfile_remote = os.path.join(remote, 'newfile') |
| newfile_local = os.path.join(local, 'newfile') |
| touch(newfile_remote) |
| run_cmd(['hg', 'add', 'newfile'], cwd=remote) |
| run_cmd(['hg', 'commit', '-m', '"add newfile"'], cwd=remote) |
| touch(newfile_local) |
| run_cmd(['hg', 'add', 'newfile'], cwd=local) |
| run_cmd(['hg', 'commit', '-m', '"re-add newfile"'], cwd=local) |
| apply_and_push(self.wc, self.repodir, |
| (lambda r, a: c(r, a, self.repodir, self.wc)), force=True) |
| |
| def testApplyAndPushForceFail(self): |
| clone(self.repodir, self.wc) |
| |
| def c(repo, attempt, remote, local): |
| newfile_remote = os.path.join(remote, 'newfile') |
| newfile_local = os.path.join(local, 'newfile') |
| touch(newfile_remote) |
| run_cmd(['hg', 'add', 'newfile'], cwd=remote) |
| run_cmd(['hg', 'commit', '-m', '"add newfile"'], cwd=remote) |
| touch(newfile_local) |
| run_cmd(['hg', 'add', 'newfile'], cwd=local) |
| run_cmd(['hg', 'commit', '-m', '"re-add newfile"'], cwd=local) |
| self.assertRaises(HgUtilError, |
| apply_and_push, self.wc, self.repodir, |
| (lambda r, a: c(r, a, self.repodir, self.wc)), force=False) |
| |
| def testPath(self): |
| clone(self.repodir, self.wc) |
| p = path(self.wc) |
| self.assertEquals(p, self.repodir) |
| |
| def testBustedHgrcWithShare(self): |
| # Test that we can recover from hgrc being lost |
| shareBase = os.path.join(self.tmpdir, 'share') |
| sharerepo = os.path.join(shareBase, self.repodir.lstrip("/")) |
| os.mkdir(shareBase) |
| mercurial(self.repodir, self.wc, shareBase=shareBase) |
| |
| # Delete .hg/hgrc |
| for d in sharerepo, self.wc: |
| f = os.path.join(d, '.hg', 'hgrc') |
| os.unlink(f) |
| |
| # path is busted now |
| p = path(self.wc) |
| self.assertEquals(p, None) |
| |
| # cloning again should fix this up |
| mercurial(self.repodir, self.wc, shareBase=shareBase) |
| |
| p = path(self.wc) |
| self.assertEquals(p, self.repodir) |
| |
| def testBustedHgrc(self): |
| # Test that we can recover from hgrc being lost |
| mercurial(self.repodir, self.wc) |
| |
| # Delete .hg/hgrc |
| os.unlink(os.path.join(self.wc, '.hg', 'hgrc')) |
| |
| # path is busted now |
| p = path(self.wc) |
| self.assertEquals(p, None) |
| |
| # cloning again should fix this up |
| mercurial(self.repodir, self.wc) |
| |
| p = path(self.wc) |
| self.assertEquals(p, self.repodir) |
| |
| def testInit(self): |
| tmpdir = os.path.join(self.tmpdir, 'new') |
| self.assertEquals(False, os.path.exists(tmpdir)) |
| init(tmpdir) |
| self.assertEquals(True, os.path.exists(tmpdir)) |
| self.assertEquals(True, os.path.exists(os.path.join(tmpdir, '.hg'))) |
| |
| def testUnbundle(self): |
| # First create the bundle |
| bundle = os.path.join(self.tmpdir, 'bundle') |
| run_cmd(['hg', 'bundle', '-a', bundle], cwd=self.repodir) |
| |
| # Now unbundle it in a new place |
| newdir = os.path.join(self.tmpdir, 'new') |
| init(newdir) |
| unbundle(bundle, newdir) |
| |
| self.assertEquals(self.revisions, getRevisions(newdir)) |
| |
| def testCloneWithBundle(self): |
| # First create the bundle |
| bundle = os.path.join(self.tmpdir, 'bundle') |
| run_cmd(['hg', 'bundle', '-a', bundle], cwd=self.repodir) |
| |
| # Wrap unbundle so we can tell if it got called |
| orig_unbundle = unbundle |
| try: |
| called = [] |
| |
| def new_unbundle(*args, **kwargs): |
| called.append(True) |
| return orig_unbundle(*args, **kwargs) |
| hg.unbundle = new_unbundle |
| |
| # Now clone it using the bundle |
| clone(self.repodir, self.wc, bundles=[bundle]) |
| self.assertEquals(self.revisions, getRevisions(self.wc)) |
| self.assertEquals(called, [True]) |
| self.assertEquals(path(self.wc), self.repodir) |
| finally: |
| hg.unbundle = orig_unbundle |
| |
| def testCloneWithUnrelatedBundle(self): |
| # First create the bundle |
| bundle = os.path.join(self.tmpdir, 'bundle') |
| run_cmd(['hg', 'bundle', '-a', bundle], cwd=self.repodir) |
| |
| # Create an unrelated repo |
| repo2 = os.path.join(self.tmpdir, 'repo2') |
| run_cmd(['%s/init_hgrepo.sh' % os.path.dirname(__file__), |
| repo2], env={'extra': 'clone_unrelated'}) |
| |
| self.assertNotEqual(self.revisions, getRevisions(repo2)) |
| |
| # Clone repo2 using the unrelated bundle |
| clone(repo2, self.wc, bundles=[bundle]) |
| |
| # Make sure we don't have unrelated revisions |
| self.assertEquals(getRevisions(repo2), getRevisions(self.wc)) |
| self.assertEquals(set(), |
| set(getRevisions(self.repodir)).intersection(set(getRevisions(self.wc)))) |
| |
| def testCloneWithBadBundle(self): |
| # First create the bad bundle |
| bundle = os.path.join(self.tmpdir, 'bundle') |
| open(bundle, 'w').write('ruh oh!') |
| |
| # Wrap unbundle so we can tell if it got called |
| orig_unbundle = unbundle |
| try: |
| called = [] |
| |
| def new_unbundle(*args, **kwargs): |
| called.append(True) |
| return orig_unbundle(*args, **kwargs) |
| hg.unbundle = new_unbundle |
| |
| # Now clone it using the bundle |
| clone(self.repodir, self.wc, bundles=[bundle]) |
| self.assertEquals(self.revisions, getRevisions(self.wc)) |
| self.assertEquals(called, [True]) |
| finally: |
| hg.unbundle = orig_unbundle |
| |
| def testCloneWithBundleMissingRevs(self): |
| # First create the bundle |
| bundle = os.path.join(self.tmpdir, 'bundle') |
| run_cmd(['hg', 'bundle', '-a', bundle], cwd=self.repodir) |
| |
| # Create a commit |
| open(os.path.join(self.repodir, 'test.txt'), 'w').write('hello!') |
| run_cmd(['hg', 'add', 'test.txt'], cwd=self.repodir) |
| run_cmd(['hg', 'commit', '-m', 'adding changeset'], cwd=self.repodir) |
| |
| # Wrap unbundle so we can tell if it got called |
| orig_unbundle = unbundle |
| try: |
| called = [] |
| |
| def new_unbundle(*args, **kwargs): |
| called.append(True) |
| return orig_unbundle(*args, **kwargs) |
| hg.unbundle = new_unbundle |
| |
| # Now clone it using the bundle |
| clone(self.repodir, self.wc, bundles=[bundle]) |
| self.assertEquals( |
| getRevisions(self.repodir), getRevisions(self.wc)) |
| self.assertEquals(called, [True]) |
| finally: |
| hg.unbundle = orig_unbundle |
| |
| def testCloneWithMirror(self): |
| mirror = os.path.join(self.tmpdir, 'repo2') |
| clone(self.repodir, mirror) |
| |
| # Create a commit in the original repo |
| open(os.path.join(self.repodir, 'test.txt'), 'w').write('hello!') |
| run_cmd(['hg', 'add', 'test.txt'], cwd=self.repodir) |
| run_cmd(['hg', 'commit', '-m', 'adding changeset'], cwd=self.repodir) |
| |
| # Now clone from the mirror |
| clone(self.repodir, self.wc, mirrors=[mirror]) |
| |
| # We'll be missing the new revision from repodir |
| self.assertNotEquals(getRevisions(self.repodir), getRevisions(self.wc)) |
| # But we should have everything from the mirror |
| self.assertEquals(getRevisions(mirror), getRevisions(self.wc)) |
| # Our default path should point to the original repo though. |
| self.assertEquals(self.repodir, path(self.wc)) |
| |
| def testCloneWithBadMirror(self): |
| mirror = os.path.join(self.tmpdir, 'repo2') |
| |
| # Now clone from the mirror |
| clone(self.repodir, self.wc, mirrors=[mirror]) |
| |
| # We still end up with a valid repo |
| self.assertEquals(self.revisions, getRevisions(self.wc)) |
| self.assertEquals(self.repodir, path(self.wc)) |
| |
| def testCloneWithOneBadMirror(self): |
| mirror1 = os.path.join(self.tmpdir, 'mirror1') |
| mirror2 = os.path.join(self.tmpdir, 'mirror2') |
| |
| # Mirror 2 is ok |
| clone(self.repodir, mirror2) |
| |
| # Create a commit in the original repo |
| open(os.path.join(self.repodir, 'test.txt'), 'w').write('hello!') |
| run_cmd(['hg', 'add', 'test.txt'], cwd=self.repodir) |
| run_cmd(['hg', 'commit', '-m', 'adding changeset'], cwd=self.repodir) |
| |
| # Now clone from the mirror |
| clone(self.repodir, self.wc, mirrors=[mirror1, mirror2]) |
| |
| # We'll be missing the new revision from repodir |
| self.assertNotEquals(getRevisions(self.repodir), getRevisions(self.wc)) |
| # But we should have everything from the mirror |
| self.assertEquals(getRevisions(mirror2), getRevisions(self.wc)) |
| # Our default path should point to the original repo though. |
| self.assertEquals(self.repodir, path(self.wc)) |
| |
| def testCloneWithRevAndMirror(self): |
| mirror = os.path.join(self.tmpdir, 'repo2') |
| clone(self.repodir, mirror) |
| |
| # Create a commit in the original repo |
| open(os.path.join(self.repodir, 'test.txt'), 'w').write('hello!') |
| run_cmd(['hg', 'add', 'test.txt'], cwd=self.repodir) |
| run_cmd(['hg', 'commit', '-m', 'adding changeset'], cwd=self.repodir) |
| |
| # Now clone from the mirror |
| revisions = getRevisions(self.repodir) |
| clone(self.repodir, self.wc, revision=revisions[0], |
| mirrors=[mirror], clone_by_rev=True) |
| |
| # We'll be missing the middle revision (on another branch) |
| self.assertEquals(revisions[:2] + revisions[3:], getRevisions(self.wc)) |
| # But not from the mirror |
| self.assertNotEquals(getRevisions(mirror), getRevisions(self.wc)) |
| # Our default path should point to the original repo though. |
| self.assertEquals(self.repodir, path(self.wc)) |
| |
| def testPullWithMirror(self): |
| mirror = os.path.join(self.tmpdir, 'repo2') |
| clone(self.repodir, mirror) |
| |
| # Create a new commit in the mirror repo |
| open(os.path.join(mirror, 'test.txt'), 'w').write('hello!') |
| run_cmd(['hg', 'add', 'test.txt'], cwd=mirror) |
| run_cmd(['hg', 'commit', '-m', 'adding changeset'], cwd=mirror) |
| |
| # Now clone from the original |
| clone(self.repodir, self.wc) |
| |
| # Pull using the mirror |
| pull(self.repodir, self.wc, mirrors=[mirror]) |
| |
| self.assertEquals(getRevisions(self.wc), getRevisions(mirror)) |
| |
| # Our default path should point to the original repo |
| self.assertEquals(self.repodir, path(self.wc)) |
| |
| def testPullWithBadMirror(self): |
| mirror = os.path.join(self.tmpdir, 'repo2') |
| |
| # Now clone from the original |
| clone(self.repodir, self.wc) |
| |
| # Create a new commit in the original repo |
| open(os.path.join(self.repodir, 'test.txt'), 'w').write('hello!') |
| run_cmd(['hg', 'add', 'test.txt'], cwd=self.repodir) |
| run_cmd(['hg', 'commit', '-m', 'adding changeset'], cwd=self.repodir) |
| |
| # Pull using the mirror |
| pull(self.repodir, self.wc, mirrors=[mirror]) |
| |
| self.assertEquals(getRevisions(self.wc), getRevisions(self.repodir)) |
| |
| # Our default path should point to the original repo |
| self.assertEquals(self.repodir, path(self.wc)) |
| |
| def testPullWithUnrelatedMirror(self): |
| mirror = os.path.join(self.tmpdir, 'repo2') |
| run_cmd(['%s/init_hgrepo.sh' % os.path.dirname(__file__), mirror], |
| env={'extra': 'pull_unrealted'}) |
| |
| # Now clone from the original |
| clone(self.repodir, self.wc) |
| |
| # Create a new commit in the original repo |
| open(os.path.join(self.repodir, 'test.txt'), 'w').write('hello!') |
| run_cmd(['hg', 'add', 'test.txt'], cwd=self.repodir) |
| run_cmd(['hg', 'commit', '-m', 'adding changeset'], cwd=self.repodir) |
| |
| # Pull using the mirror |
| pull(self.repodir, self.wc, mirrors=[mirror]) |
| |
| self.assertEquals(getRevisions(self.wc), getRevisions(self.repodir)) |
| # We shouldn't have anything from the unrelated mirror |
| self.assertEquals(set(), |
| set(getRevisions(mirror)).intersection(set(getRevisions(self.wc)))) |
| |
| # Our default path should point to the original repo |
| self.assertEquals(self.repodir, path(self.wc)) |
| |
| def testMercurialWithShareAndBundle(self): |
| # First create the bundle |
| bundle = os.path.join(self.tmpdir, 'bundle') |
| run_cmd(['hg', 'bundle', '-a', bundle], cwd=self.repodir) |
| |
| # Create a commit |
| open(os.path.join(self.repodir, 'test.txt'), 'w').write('hello!') |
| run_cmd(['hg', 'add', 'test.txt'], cwd=self.repodir) |
| run_cmd(['hg', 'commit', '-m', 'adding changeset'], cwd=self.repodir) |
| |
| # Wrap unbundle so we can tell if it got called |
| orig_unbundle = unbundle |
| try: |
| called = [] |
| |
| def new_unbundle(*args, **kwargs): |
| called.append(True) |
| return orig_unbundle(*args, **kwargs) |
| hg.unbundle = new_unbundle |
| |
| shareBase = os.path.join(self.tmpdir, 'share') |
| sharerepo = os.path.join(shareBase, self.repodir.lstrip("/")) |
| os.mkdir(shareBase) |
| mercurial( |
| self.repodir, self.wc, shareBase=shareBase, bundles=[bundle]) |
| |
| self.assertEquals(called, [True]) |
| self.assertEquals( |
| getRevisions(self.repodir), getRevisions(self.wc)) |
| self.assertEquals( |
| getRevisions(self.repodir), getRevisions(sharerepo)) |
| finally: |
| hg.unbundle = orig_unbundle |
| |
| def testMercurialWithShareAndMirror(self): |
| # First create the mirror |
| mirror = os.path.join(self.tmpdir, 'repo2') |
| clone(self.repodir, mirror) |
| |
| # Create a commit |
| open(os.path.join(self.repodir, 'test.txt'), 'w').write('hello!') |
| run_cmd(['hg', 'add', 'test.txt'], cwd=self.repodir) |
| run_cmd(['hg', 'commit', '-m', 'adding changeset'], cwd=self.repodir) |
| |
| shareBase = os.path.join(self.tmpdir, 'share') |
| sharerepo = os.path.join(shareBase, self.repodir.lstrip("/")) |
| os.mkdir(shareBase) |
| mercurial(self.repodir, self.wc, shareBase=shareBase, mirrors=[mirror]) |
| |
| # Since we used the mirror, we should be missing a commit |
| self.assertNotEquals(getRevisions(self.repodir), getRevisions(self.wc)) |
| self.assertNotEquals( |
| getRevisions(self.repodir), getRevisions(sharerepo)) |
| self.assertEquals(getRevisions(mirror), getRevisions(self.wc)) |
| |
| def testMercurialByRevWithShareAndMirror(self): |
| # First create the mirror |
| mirror = os.path.join(self.tmpdir, 'repo2') |
| clone(self.repodir, mirror) |
| |
| shareBase = os.path.join(self.tmpdir, 'share') |
| sharerepo = os.path.join(shareBase, self.repodir.lstrip("/")) |
| os.mkdir(shareBase) |
| mercurial(self.repodir, self.wc, shareBase=shareBase, mirrors=[ |
| mirror], clone_by_rev=True, revision=self.revisions[-1]) |
| |
| # We should only have the one revision |
| self.assertEquals(getRevisions(sharerepo), self.revisions[-1:]) |
| self.assertEquals(getRevisions(self.wc), self.revisions[-1:]) |
| |
| def testMercurialSkipPull(self): |
| # Clone once into our working copy |
| mercurial(self.repodir, self.wc) |
| |
| # The second clone should avoid calling pull() |
| with patch('util.hg.pull') as patched_pull: |
| mercurial(self.repodir, self.wc, revision=self.revisions[-1]) |
| self.assertEquals(patched_pull.call_count, 0) |
| |
| def testAdjustPaths(self): |
| mercurial(self.repodir, self.wc) |
| |
| # Make sure our default path is correct |
| self.assertEquals(path(self.wc), self.repodir) |
| |
| # Add a comment, make sure it's still there if we don't change |
| # anything |
| hgrc = os.path.join(self.wc, '.hg', 'hgrc') |
| open(hgrc, 'a').write("# Hello world") |
| adjust_paths(self.wc, default=self.repodir) |
| self.assert_("Hello world" in open(hgrc).read()) |
| |
| # Add a path, and the comment goes away |
| adjust_paths(self.wc, test=self.repodir) |
| self.assert_("Hello world" not in open(hgrc).read()) |
| |
| # Make sure our paths are correct |
| self.assertEquals(path(self.wc), self.repodir) |
| self.assertEquals(path(self.wc, 'test'), self.repodir) |
| |
| def testOutofSyncMirrorFailingMaster(self): |
| # First create the mirror |
| mirror = os.path.join(self.tmpdir, 'repo2') |
| clone(self.repodir, mirror) |
| |
| shareBase = os.path.join(self.tmpdir, 'share') |
| os.mkdir(shareBase) |
| mercurial(self.repodir, self.wc, shareBase=shareBase, mirrors=[mirror]) |
| |
| # Create a bundle |
| bundle = os.path.join(self.tmpdir, 'bundle') |
| run_cmd(['hg', 'bundle', '-a', bundle], cwd=self.repodir) |
| |
| # Move our repodir out of the way so that pulling/cloning from it fails |
| os.rename(self.repodir, self.repodir + "-bad") |
| |
| # Try and update to a non-existent revision using our mirror and |
| # bundle, with the master failing. We should fail |
| self.assertRaises(subprocess.CalledProcessError, mercurial, |
| self.repodir, self.wc, shareBase=shareBase, |
| mirrors=[mirror], bundles=[bundle], |
| revision="1234567890") |
| |
| def testCommit(self): |
| newfile = os.path.join(self.repodir, 'newfile') |
| touch(newfile) |
| run_cmd(['hg', 'add', 'newfile'], cwd=self.repodir) |
| rev = commit(self.repodir, user='unittest', msg='gooooood') |
| info = getRevInfo(self.repodir, rev) |
| self.assertEquals(info['msg'], 'gooooood') |
| # can't test for user, because it depends on local hg configs. |
| |
| def testCommitWithUser(self): |
| newfile = os.path.join(self.repodir, 'newfile') |
| touch(newfile) |
| run_cmd(['hg', 'add', 'newfile'], cwd=self.repodir) |
| rev = commit(self.repodir, user='unittest', msg='new stuff!') |
| info = getRevInfo(self.repodir, rev) |
| self.assertEquals(info['user'], 'unittest') |
| self.assertEquals(info['msg'], 'new stuff!') |
| |
| def testTag(self): |
| tag(self.repodir, ['test_tag']) |
| self.assertTrue('test_tag' in getTags(self.repodir)) |
| |
| def testMultitag(self): |
| tag(self.repodir, ['tag1', 'tag2']) |
| tags = getTags(self.repodir) |
| self.assertTrue('tag1' in tags) |
| self.assertTrue('tag2' in tags) |
| |
| def testMultitagSet(self): |
| tag(self.repodir, set(['tag1', 'tag5'])) |
| tags = getTags(self.repodir) |
| self.assertTrue('tag1' in tags) |
| self.assertTrue('tag5' in tags) |
| |
| def testTagWithMsg(self): |
| rev = tag(self.repodir, ['tag'], msg='I made a tag!') |
| info = getRevInfo(self.repodir, rev) |
| self.assertEquals('I made a tag!', info['msg']) |
| |
| def testTagWithUser(self): |
| rev = tag(self.repodir, ['taggy'], user='tagger') |
| info = getRevInfo(self.repodir, rev) |
| self.assertEquals('tagger', info['user']) |
| |
| def testTagWithRevision(self): |
| tag(self.repodir, ['tag'], rev='1') |
| info = getRevInfo(self.repodir, '1') |
| self.assertEquals(['tag'], info['tags']) |
| |
| def testMultitagWithRevision(self): |
| tag(self.repodir, ['tag1', 'tag2'], rev='1') |
| info = getRevInfo(self.repodir, '1') |
| self.assertEquals(['tag1', 'tag2'], info['tags']) |
| |
| def testTagFailsIfExists(self): |
| run_cmd(['hg', 'tag', '-R', self.repodir, 'tagg']) |
| self.assertRaises( |
| subprocess.CalledProcessError, tag, self.repodir, 'tagg') |
| |
| def testForcedTag(self): |
| run_cmd(['hg', 'tag', '-R', self.repodir, 'tag']) |
| tag(self.repodir, ['tag'], force=True) |
| self.assertTrue('tag' in getTags(self.repodir)) |
| |
| def testFailingBackend(self): |
| # Test that remote failures get retried |
| num_calls = [0] |
| |
| def _my_get_hg_output(cmd, **kwargs): |
| num_calls[0] += 1 |
| if cmd[0] == 'clone': |
| e = subprocess.CalledProcessError(-1, cmd) |
| e.output = "error: Name or service not known" |
| raise e |
| else: |
| return get_hg_output(cmd, **kwargs) |
| |
| # Two retries are forced in setUp() above |
| with patch('util.hg.get_hg_output', new=_my_get_hg_output): |
| self.assertRaises(subprocess.CalledProcessError, |
| clone, "http://nxdomain.nxnx", self.wc) |
| self.assertEquals(num_calls, [2]) |
| |
| @patch("util.commands.get_output") |
| def testGetHgOutputAlwaysLogsException(self, get_output): |
| get_output.side_effect = subprocess.CalledProcessError(255, "kaboom!") |
| |
| with patch("util.hg.log") as mocked_log: |
| try: |
| get_hg_output(["status"]) |
| self.fail("CalledProcessError not raised") |
| except subprocess.CalledProcessError: |
| self.assertTrue(mocked_log.exception.called, "log.exception not called") |
| |
| def testHasRev(self): |
| self.assertTrue(has_rev(self.repodir, self.revisions[0])) |
| self.assertFalse(has_rev(self.repodir, self.revisions[0] + 'g')) |
