test_base_vcs_mercurial.py (13964B)
1 import os 2 import platform 3 import shutil 4 import tempfile 5 import unittest 6 7 import mozunit 8 from mozharness.base.vcs import mercurial 9 10 test_string = """foo 11 bar 12 baz""" 13 14 HG = ["hg"] + mercurial.HG_OPTIONS 15 16 # Known default .hgrc 17 os.environ["HGRCPATH"] = os.path.abspath( 18 os.path.join(os.path.dirname(__file__), "helper_files", ".hgrc") 19 ) 20 21 22 def cleanup(): 23 if os.path.exists("test_logs"): 24 shutil.rmtree("test_logs") 25 if os.path.exists("test_dir"): 26 if os.path.isdir("test_dir"): 27 shutil.rmtree("test_dir") 28 else: 29 os.remove("test_dir") 30 for filename in ("localconfig.json", "localconfig.json.bak"): 31 if os.path.exists(filename): 32 os.remove(filename) 33 34 35 def get_mercurial_vcs_obj(): 36 m = mercurial.MercurialVCS() 37 m.config = {} 38 return m 39 40 41 def get_revisions(dest): 42 m = get_mercurial_vcs_obj() 43 retval = [] 44 command = HG + ["log", "-R", dest, "--template", "{node}\n"] 45 for rev in m.get_output_from_command(command).split("\n"): 46 rev = rev.strip() 47 if not rev: 48 continue 49 retval.append(rev) 50 return retval 51 52 53 class TestMakeAbsolute(unittest.TestCase): 54 # _make_absolute() doesn't play nicely with windows/msys paths. 55 # TODO: fix _make_absolute, write it out of the picture, or determine 56 # that it's not needed on windows. 57 if platform.system() not in ("Windows",): 58 59 def test_absolute_path(self): 60 m = get_mercurial_vcs_obj() 61 self.assertEqual(m._make_absolute("/foo/bar"), "/foo/bar") 62 63 def test_relative_path(self): 64 m = get_mercurial_vcs_obj() 65 self.assertEqual(m._make_absolute("foo/bar"), os.path.abspath("foo/bar")) 66 67 def test_HTTP_paths(self): 68 m = get_mercurial_vcs_obj() 69 self.assertEqual(m._make_absolute("http://foo/bar"), "http://foo/bar") 70 71 def test_absolute_file_path(self): 72 m = get_mercurial_vcs_obj() 73 self.assertEqual(m._make_absolute("file:///foo/bar"), "file:///foo/bar") 74 75 def test_relative_file_path(self): 76 m = get_mercurial_vcs_obj() 77 self.assertEqual( 78 m._make_absolute("file://foo/bar"), "file://%s/foo/bar" % os.getcwd() 79 ) 80 81 82 class TestHg(unittest.TestCase): 83 def _init_hg_repo(self, hg_obj, repodir): 84 hg_obj.run_command([ 85 "bash", 86 os.path.join(os.path.dirname(__file__), "helper_files", "init_hgrepo.sh"), 87 repodir, 88 ]) 89 90 def setUp(self): 91 self.tmpdir = tempfile.mkdtemp() 92 self.repodir = os.path.join(self.tmpdir, "repo") 93 m = get_mercurial_vcs_obj() 94 self._init_hg_repo(m, self.repodir) 95 self.revisions = get_revisions(self.repodir) 96 self.wc = os.path.join(self.tmpdir, "wc") 97 self.pwd = os.getcwd() 98 99 def tearDown(self): 100 shutil.rmtree(self.tmpdir) 101 os.chdir(self.pwd) 102 103 def test_get_branch(self): 104 m = get_mercurial_vcs_obj() 105 m.clone(self.repodir, self.wc) 106 b = m.get_branch_from_path(self.wc) 107 self.assertEqual(b, "default") 108 109 def test_get_branches(self): 110 m = get_mercurial_vcs_obj() 111 m.clone(self.repodir, self.wc) 112 branches = m.get_branches_from_path(self.wc) 113 self.assertEqual(sorted(branches), sorted(["branch2", "default"])) 114 115 def test_clone(self): 116 m = get_mercurial_vcs_obj() 117 rev = m.clone(self.repodir, self.wc, update_dest=False) 118 self.assertEqual(rev, None) 119 self.assertEqual(self.revisions, get_revisions(self.wc)) 120 self.assertEqual(sorted(os.listdir(self.wc)), [".hg"]) 121 122 def test_clone_into_non_empty_dir(self): 123 m = get_mercurial_vcs_obj() 124 m.mkdir_p(self.wc) 125 open(os.path.join(self.wc, "test.txt"), "w").write("hello") 126 m.clone(self.repodir, self.wc, update_dest=False) 127 self.assertTrue(not os.path.exists(os.path.join(self.wc, "test.txt"))) 128 129 def test_clone_update(self): 130 m = get_mercurial_vcs_obj() 131 rev = m.clone(self.repodir, self.wc, update_dest=True) 132 self.assertEqual(rev, self.revisions[0]) 133 134 def test_clone_branch(self): 135 m = get_mercurial_vcs_obj() 136 m.clone(self.repodir, self.wc, branch="branch2", update_dest=False) 137 # On hg 1.6, we should only have a subset of the revisions 138 if m.hg_ver() >= (1, 6, 0): 139 self.assertEqual(self.revisions[1:], get_revisions(self.wc)) 140 else: 141 self.assertEqual(self.revisions, get_revisions(self.wc)) 142 143 def test_clone_update_branch(self): 144 m = get_mercurial_vcs_obj() 145 rev = m.clone( 146 self.repodir, 147 os.path.join(self.tmpdir, "wc"), 148 branch="branch2", 149 update_dest=True, 150 ) 151 self.assertEqual(rev, self.revisions[1], self.revisions) 152 153 def test_clone_revision(self): 154 m = get_mercurial_vcs_obj() 155 m.clone(self.repodir, self.wc, revision=self.revisions[0], update_dest=False) 156 # We'll only get a subset of the revisions 157 self.assertEqual( 158 self.revisions[:1] + self.revisions[2:], get_revisions(self.wc) 159 ) 160 161 def test_update_revision(self): 162 m = get_mercurial_vcs_obj() 163 rev = m.clone(self.repodir, self.wc, update_dest=False) 164 self.assertEqual(rev, None) 165 166 rev = m.update(self.wc, revision=self.revisions[1]) 167 self.assertEqual(rev, self.revisions[1]) 168 169 def test_pull(self): 170 m = get_mercurial_vcs_obj() 171 # Clone just the first rev 172 m.clone(self.repodir, self.wc, revision=self.revisions[-1], update_dest=False) 173 self.assertEqual(get_revisions(self.wc), self.revisions[-1:]) 174 175 # Now pull in new changes 176 rev = m.pull(self.repodir, self.wc, update_dest=False) 177 self.assertEqual(rev, None) 178 self.assertEqual(get_revisions(self.wc), self.revisions) 179 180 def test_pull_revision(self): 181 m = get_mercurial_vcs_obj() 182 # Clone just the first rev 183 m.clone(self.repodir, self.wc, revision=self.revisions[-1], update_dest=False) 184 self.assertEqual(get_revisions(self.wc), self.revisions[-1:]) 185 186 # Now pull in just the last revision 187 rev = m.pull( 188 self.repodir, self.wc, revision=self.revisions[0], update_dest=False 189 ) 190 self.assertEqual(rev, None) 191 192 # We'll be missing the middle revision (on another branch) 193 self.assertEqual( 194 get_revisions(self.wc), self.revisions[:1] + self.revisions[2:] 195 ) 196 197 def test_pull_branch(self): 198 m = get_mercurial_vcs_obj() 199 # Clone just the first rev 200 m.clone(self.repodir, self.wc, revision=self.revisions[-1], update_dest=False) 201 self.assertEqual(get_revisions(self.wc), self.revisions[-1:]) 202 203 # Now pull in the other branch 204 rev = m.pull(self.repodir, self.wc, branch="branch2", update_dest=False) 205 self.assertEqual(rev, None) 206 207 # On hg 1.6, we'll be missing the last revision (on another branch) 208 if m.hg_ver() >= (1, 6, 0): 209 self.assertEqual(get_revisions(self.wc), self.revisions[1:]) 210 else: 211 self.assertEqual(get_revisions(self.wc), self.revisions) 212 213 def test_pull_unrelated(self): 214 m = get_mercurial_vcs_obj() 215 # Create a new repo 216 repo2 = os.path.join(self.tmpdir, "repo2") 217 self._init_hg_repo(m, repo2) 218 219 self.assertNotEqual(self.revisions, get_revisions(repo2)) 220 221 # Clone the original repo 222 m.clone(self.repodir, self.wc, update_dest=False) 223 # Hide the wanted error 224 m.config = {"log_to_console": False} 225 # Try and pull in changes from the new repo 226 self.assertRaises( 227 mercurial.VCSException, m.pull, repo2, self.wc, update_dest=False 228 ) 229 230 def test_push(self): 231 m = get_mercurial_vcs_obj() 232 m.clone(self.repodir, self.wc, revision=self.revisions[-2]) 233 m.push(src=self.repodir, remote=self.wc) 234 self.assertEqual(get_revisions(self.wc), self.revisions) 235 236 def test_push_with_branch(self): 237 m = get_mercurial_vcs_obj() 238 if m.hg_ver() >= (1, 6, 0): 239 m.clone(self.repodir, self.wc, revision=self.revisions[-1]) 240 m.push(src=self.repodir, remote=self.wc, branch="branch2") 241 m.push(src=self.repodir, remote=self.wc, branch="default") 242 self.assertEqual(get_revisions(self.wc), self.revisions) 243 244 def test_push_with_revision(self): 245 m = get_mercurial_vcs_obj() 246 m.clone(self.repodir, self.wc, revision=self.revisions[-2]) 247 m.push(src=self.repodir, remote=self.wc, revision=self.revisions[-1]) 248 self.assertEqual(get_revisions(self.wc), self.revisions[-2:]) 249 250 def test_mercurial(self): 251 m = get_mercurial_vcs_obj() 252 m.vcs_config = { 253 "repo": self.repodir, 254 "dest": self.wc, 255 "vcs_share_base": os.path.join(self.tmpdir, "share"), 256 } 257 m.ensure_repo_and_revision() 258 rev = m.ensure_repo_and_revision() 259 self.assertEqual(rev, self.revisions[0]) 260 261 def test_push_new_branches_not_allowed(self): 262 m = get_mercurial_vcs_obj() 263 m.clone(self.repodir, self.wc, revision=self.revisions[0]) 264 # Hide the wanted error 265 m.config = {"log_to_console": False} 266 self.assertRaises( 267 Exception, m.push, self.repodir, self.wc, push_new_branches=False 268 ) 269 270 def test_mercurial_relative_dir(self): 271 m = get_mercurial_vcs_obj() 272 repo = os.path.basename(self.repodir) 273 wc = os.path.basename(self.wc) 274 m.vcs_config = { 275 "repo": repo, 276 "dest": wc, 277 "revision": self.revisions[-1], 278 "vcs_share_base": os.path.join(self.tmpdir, "share"), 279 } 280 m.chdir(os.path.dirname(self.repodir)) 281 try: 282 rev = m.ensure_repo_and_revision() 283 self.assertEqual(rev, self.revisions[-1]) 284 m.info("Creating test.txt") 285 open(os.path.join(self.wc, "test.txt"), "w").write("hello!") 286 287 m = get_mercurial_vcs_obj() 288 m.vcs_config = { 289 "repo": repo, 290 "dest": wc, 291 "revision": self.revisions[0], 292 "vcs_share_base": os.path.join(self.tmpdir, "share"), 293 } 294 rev = m.ensure_repo_and_revision() 295 self.assertEqual(rev, self.revisions[0]) 296 # Make sure our local file didn't go away 297 self.assertTrue(os.path.exists(os.path.join(self.wc, "test.txt"))) 298 finally: 299 m.chdir(self.pwd) 300 301 def test_mercurial_update_tip(self): 302 m = get_mercurial_vcs_obj() 303 m.vcs_config = { 304 "repo": self.repodir, 305 "dest": self.wc, 306 "revision": self.revisions[-1], 307 "vcs_share_base": os.path.join(self.tmpdir, "share"), 308 } 309 rev = m.ensure_repo_and_revision() 310 self.assertEqual(rev, self.revisions[-1]) 311 open(os.path.join(self.wc, "test.txt"), "w").write("hello!") 312 313 m = get_mercurial_vcs_obj() 314 m.vcs_config = { 315 "repo": self.repodir, 316 "dest": self.wc, 317 "vcs_share_base": os.path.join(self.tmpdir, "share"), 318 } 319 rev = m.ensure_repo_and_revision() 320 self.assertEqual(rev, self.revisions[0]) 321 # Make sure our local file didn't go away 322 self.assertTrue(os.path.exists(os.path.join(self.wc, "test.txt"))) 323 324 def test_mercurial_update_rev(self): 325 m = get_mercurial_vcs_obj() 326 m.vcs_config = { 327 "repo": self.repodir, 328 "dest": self.wc, 329 "revision": self.revisions[-1], 330 "vcs_share_base": os.path.join(self.tmpdir, "share"), 331 } 332 rev = m.ensure_repo_and_revision() 333 self.assertEqual(rev, self.revisions[-1]) 334 open(os.path.join(self.wc, "test.txt"), "w").write("hello!") 335 336 m = get_mercurial_vcs_obj() 337 m.vcs_config = { 338 "repo": self.repodir, 339 "dest": self.wc, 340 "revision": self.revisions[0], 341 "vcs_share_base": os.path.join(self.tmpdir, "share"), 342 } 343 rev = m.ensure_repo_and_revision() 344 self.assertEqual(rev, self.revisions[0]) 345 # Make sure our local file didn't go away 346 self.assertTrue(os.path.exists(os.path.join(self.wc, "test.txt"))) 347 348 def test_make_hg_url(self): 349 # construct an hg url specific to revision, branch and filename and try to pull it down 350 file_url = mercurial.make_hg_url( 351 "hg.mozilla.org", 352 "//build/tools/", 353 revision="FIREFOX_3_6_12_RELEASE", 354 filename="/lib/python/util/hg.py", 355 protocol="https", 356 ) 357 expected_url = ( 358 "https://hg.mozilla.org/build/tools/raw-file/" 359 "FIREFOX_3_6_12_RELEASE/lib/python/util/hg.py" 360 ) 361 self.assertEqual(file_url, expected_url) 362 363 def test_make_hg_url_no_filename(self): 364 file_url = mercurial.make_hg_url( 365 "hg.mozilla.org", 366 "/build/tools", 367 revision="default", 368 protocol="https", 369 ) 370 expected_url = "https://hg.mozilla.org/build/tools/rev/default" 371 self.assertEqual(file_url, expected_url) 372 373 def test_make_hg_url_no_revision_no_filename(self): 374 repo_url = mercurial.make_hg_url( 375 "hg.mozilla.org", 376 "/build/tools", 377 protocol="https", 378 ) 379 expected_url = "https://hg.mozilla.org/build/tools" 380 self.assertEqual(repo_url, expected_url) 381 382 def test_make_hg_url_different_protocol(self): 383 repo_url = mercurial.make_hg_url( 384 "hg.mozilla.org", 385 "/build/tools", 386 protocol="ssh", 387 ) 388 expected_url = "ssh://hg.mozilla.org/build/tools" 389 self.assertEqual(repo_url, expected_url) 390 391 392 if __name__ == "__main__": 393 mozunit.main()