1""" 2Tests for the Git state 3""" 4 5import functools 6import inspect 7import logging 8import os 9import shutil 10import socket 11import string 12import tempfile 13import urllib.parse 14 15import pytest 16import salt.utils.files 17import salt.utils.path 18from salt.utils.versions import LooseVersion as _LooseVersion 19from tests.support.case import ModuleCase 20from tests.support.helpers import TstSuiteLoggingHandler, with_tempdir 21from tests.support.mixins import SaltReturnAssertsMixin 22from tests.support.runtests import RUNTIME_VARS 23 24TEST_REPO = "https://github.com/saltstack/salt-test-repo.git" 25 26 27def __check_git_version(caller, min_version, skip_msg): 28 """ 29 Common logic for version check 30 """ 31 if inspect.isclass(caller): 32 actual_setup = getattr(caller, "setUp", None) 33 34 def setUp(self, *args, **kwargs): 35 if not salt.utils.path.which("git"): 36 self.skipTest("git is not installed") 37 git_version = self.run_function("git.version") 38 if _LooseVersion(git_version) < _LooseVersion(min_version): 39 self.skipTest(skip_msg.format(min_version, git_version)) 40 if actual_setup is not None: 41 actual_setup(self, *args, **kwargs) 42 43 caller.setUp = setUp 44 return caller 45 46 @functools.wraps(caller) 47 def wrapper(self, *args, **kwargs): 48 if not salt.utils.path.which("git"): 49 self.skipTest("git is not installed") 50 git_version = self.run_function("git.version") 51 if _LooseVersion(git_version) < _LooseVersion(min_version): 52 self.skipTest(skip_msg.format(min_version, git_version)) 53 return caller(self, *args, **kwargs) 54 55 return wrapper 56 57 58def ensure_min_git(caller=None, min_version="1.6.5"): 59 """ 60 Skip test if minimum supported git version is not installed 61 """ 62 if caller is None: 63 return functools.partial(ensure_min_git, min_version=min_version) 64 65 return __check_git_version( 66 caller, min_version, "git {0} or newer required to run this test (detected {1})" 67 ) 68 69 70def uses_git_opts(caller): 71 """ 72 Skip test if git_opts is not supported 73 74 IMPORTANT! This decorator should be at the bottom of any decorators added 75 to a given function. 76 """ 77 min_version = "1.7.2" 78 return __check_git_version( 79 caller, 80 min_version, 81 "git_opts only supported in git {0} and newer (detected {1})", 82 ) 83 84 85class WithGitMirror: 86 def __init__(self, repo_url, **kwargs): 87 self.repo_url = repo_url 88 if "dir" not in kwargs: 89 kwargs["dir"] = RUNTIME_VARS.TMP 90 self.kwargs = kwargs 91 92 def __call__(self, func): 93 self.func = func 94 return functools.wraps(func)( 95 # pylint: disable=unnecessary-lambda 96 lambda testcase, *args, **kwargs: self.wrap(testcase, *args, **kwargs) 97 # pylint: enable=unnecessary-lambda 98 ) 99 100 def wrap(self, testcase, *args, **kwargs): 101 # Get temp dir paths 102 mirror_dir = tempfile.mkdtemp(**self.kwargs) 103 admin_dir = tempfile.mkdtemp(**self.kwargs) 104 clone_dir = tempfile.mkdtemp(**self.kwargs) 105 # Clean up the directories, we want git to actually create them 106 os.rmdir(mirror_dir) 107 os.rmdir(admin_dir) 108 os.rmdir(clone_dir) 109 # Create a URL to clone 110 mirror_url = "file://" + mirror_dir 111 # Mirror the repo 112 testcase.run_function("git.clone", [mirror_dir], url=TEST_REPO, opts="--mirror") 113 # Make sure the directory for the mirror now exists 114 assert os.path.exists(mirror_dir) 115 # Clone to the admin dir 116 ret = testcase.run_state("git.latest", name=mirror_url, target=admin_dir) 117 ret = ret[next(iter(ret))] 118 assert os.path.exists(admin_dir) 119 120 try: 121 # Run the actual function with three arguments added: 122 # 1. URL for the test to use to clone 123 # 2. Cloned admin dir for making/pushing changes to the mirror 124 # 3. Yet-nonexistent clone_dir for the test function to use as a 125 # destination for cloning. 126 return self.func( 127 testcase, mirror_url, admin_dir, clone_dir, *args, **kwargs 128 ) 129 finally: 130 shutil.rmtree(mirror_dir, ignore_errors=True) 131 shutil.rmtree(admin_dir, ignore_errors=True) 132 shutil.rmtree(clone_dir, ignore_errors=True) 133 134 135with_git_mirror = WithGitMirror 136 137 138@ensure_min_git 139class GitTest(ModuleCase, SaltReturnAssertsMixin): 140 """ 141 Validate the git state 142 """ 143 144 def setUp(self): 145 domain = urllib.parse.urlparse(TEST_REPO).netloc 146 try: 147 if hasattr(socket, "setdefaulttimeout"): 148 # 10 second dns timeout 149 socket.setdefaulttimeout(10) 150 socket.gethostbyname(domain) 151 except OSError: 152 msg = "error resolving {0}, possible network issue?" 153 self.skipTest(msg.format(domain)) 154 155 def tearDown(self): 156 # Reset the dns timeout after the test is over 157 socket.setdefaulttimeout(None) 158 159 def _head(self, cwd): 160 return self.run_function("git.rev_parse", [cwd, "HEAD"]) 161 162 @with_tempdir(create=False) 163 @pytest.mark.slow_test 164 def test_latest(self, target): 165 """ 166 git.latest 167 """ 168 ret = self.run_state("git.latest", name=TEST_REPO, target=target) 169 self.assertSaltTrueReturn(ret) 170 self.assertTrue(os.path.isdir(os.path.join(target, ".git"))) 171 172 @with_tempdir(create=False) 173 @pytest.mark.slow_test 174 def test_latest_config_get_regexp_retcode(self, target): 175 """ 176 git.latest 177 """ 178 179 log_format = "[%(levelname)-8s] %(jid)s %(message)s" 180 self.handler = TstSuiteLoggingHandler(format=log_format, level=logging.DEBUG) 181 ret_code_err = "failed with return code: 1" 182 with self.handler: 183 ret = self.run_state("git.latest", name=TEST_REPO, target=target) 184 self.assertSaltTrueReturn(ret) 185 self.assertTrue(os.path.isdir(os.path.join(target, ".git"))) 186 assert any(ret_code_err in s for s in self.handler.messages) is False, False 187 188 @with_tempdir(create=False) 189 @pytest.mark.slow_test 190 def test_latest_with_rev_and_submodules(self, target): 191 """ 192 git.latest 193 """ 194 ret = self.run_state( 195 "git.latest", name=TEST_REPO, rev="develop", target=target, submodules=True 196 ) 197 self.assertSaltTrueReturn(ret) 198 self.assertTrue(os.path.isdir(os.path.join(target, ".git"))) 199 200 @with_tempdir(create=False) 201 @pytest.mark.slow_test 202 def test_latest_failure(self, target): 203 """ 204 git.latest 205 """ 206 ret = self.run_state( 207 "git.latest", 208 name="https://youSpelledGitHubWrong.com/saltstack/salt-test-repo.git", 209 rev="develop", 210 target=target, 211 submodules=True, 212 ) 213 self.assertSaltFalseReturn(ret) 214 self.assertFalse(os.path.isdir(os.path.join(target, ".git"))) 215 216 @with_tempdir() 217 @pytest.mark.slow_test 218 def test_latest_empty_dir(self, target): 219 """ 220 git.latest 221 """ 222 ret = self.run_state( 223 "git.latest", name=TEST_REPO, rev="develop", target=target, submodules=True 224 ) 225 self.assertSaltTrueReturn(ret) 226 self.assertTrue(os.path.isdir(os.path.join(target, ".git"))) 227 228 @with_tempdir(create=False) 229 @pytest.mark.slow_test 230 def test_latest_unless_no_cwd_issue_6800(self, target): 231 """ 232 cwd=target was being passed to _run_check which blew up if 233 target dir did not already exist. 234 """ 235 ret = self.run_state( 236 "git.latest", 237 name=TEST_REPO, 238 rev="develop", 239 target=target, 240 unless="test -e {}".format(target), 241 submodules=True, 242 ) 243 self.assertSaltTrueReturn(ret) 244 self.assertTrue(os.path.isdir(os.path.join(target, ".git"))) 245 246 @with_tempdir(create=False) 247 @pytest.mark.slow_test 248 def test_numeric_rev(self, target): 249 """ 250 git.latest with numeric revision 251 """ 252 ret = self.run_state( 253 "git.latest", 254 name=TEST_REPO, 255 rev=0.11, 256 target=target, 257 submodules=True, 258 timeout=120, 259 ) 260 self.assertSaltTrueReturn(ret) 261 self.assertTrue(os.path.isdir(os.path.join(target, ".git"))) 262 263 @with_tempdir(create=False) 264 @pytest.mark.slow_test 265 def test_latest_with_local_changes(self, target): 266 """ 267 Ensure that we fail the state when there are local changes and succeed 268 when force_reset is True. 269 """ 270 # Clone repo 271 ret = self.run_state("git.latest", name=TEST_REPO, target=target) 272 self.assertSaltTrueReturn(ret) 273 self.assertTrue(os.path.isdir(os.path.join(target, ".git"))) 274 275 # Make change to LICENSE file. 276 with salt.utils.files.fopen(os.path.join(target, "LICENSE"), "a") as fp_: 277 fp_.write("Lorem ipsum dolor blah blah blah....\n") 278 279 # Make sure that we now have uncommitted changes 280 self.assertTrue(self.run_function("git.diff", [target, "HEAD"])) 281 282 # Re-run state with force_reset=False 283 ret = self.run_state( 284 "git.latest", name=TEST_REPO, target=target, force_reset=False 285 ) 286 self.assertSaltTrueReturn(ret) 287 self.assertEqual( 288 ret[next(iter(ret))]["comment"], 289 "Repository {} is up-to-date, but with uncommitted changes. " 290 "Set 'force_reset' to True to purge uncommitted changes.".format(target), 291 ) 292 293 # Now run the state with force_reset=True 294 ret = self.run_state( 295 "git.latest", name=TEST_REPO, target=target, force_reset=True 296 ) 297 self.assertSaltTrueReturn(ret) 298 299 # Make sure that we no longer have uncommitted changes 300 self.assertFalse(self.run_function("git.diff", [target, "HEAD"])) 301 302 @with_git_mirror(TEST_REPO) 303 @uses_git_opts 304 @pytest.mark.slow_test 305 def test_latest_fast_forward(self, mirror_url, admin_dir, clone_dir): 306 """ 307 Test running git.latest state a second time after changes have been 308 made to the remote repo. 309 """ 310 # Clone the repo 311 ret = self.run_state("git.latest", name=mirror_url, target=clone_dir) 312 ret = ret[next(iter(ret))] 313 assert ret["result"] 314 315 # Make a change to the repo by editing the file in the admin copy 316 # of the repo and committing. 317 head_pre = self._head(admin_dir) 318 with salt.utils.files.fopen(os.path.join(admin_dir, "LICENSE"), "a") as fp_: 319 fp_.write("Hello world!") 320 self.run_function( 321 "git.commit", 322 [admin_dir, "added a line"], 323 git_opts='-c user.name="Foo Bar" -c user.email=foo@bar.com', 324 opts="-a", 325 ) 326 # Make sure HEAD is pointing to a new SHA so we know we properly 327 # committed our change. 328 head_post = self._head(admin_dir) 329 assert head_pre != head_post 330 331 # Push the change to the mirror 332 # NOTE: the test will fail if the salt-test-repo's default branch 333 # is changed. 334 self.run_function("git.push", [admin_dir, "origin", "develop"]) 335 336 # Re-run the git.latest state on the clone_dir 337 ret = self.run_state("git.latest", name=mirror_url, target=clone_dir) 338 ret = ret[next(iter(ret))] 339 assert ret["result"] 340 341 # Make sure that the clone_dir now has the correct SHA 342 assert head_post == self._head(clone_dir) 343 344 @with_tempdir(create=False) 345 def _changed_local_branch_helper(self, target, rev, hint): 346 """ 347 We're testing two almost identical cases, the only thing that differs 348 is the rev used for the git.latest state. 349 """ 350 # Clone repo 351 ret = self.run_state("git.latest", name=TEST_REPO, rev=rev, target=target) 352 self.assertSaltTrueReturn(ret) 353 354 # Check out a new branch in the clone and make a commit, to ensure 355 # that when we re-run the state, it is not a fast-forward change 356 self.run_function("git.checkout", [target, "new_branch"], opts="-b") 357 with salt.utils.files.fopen(os.path.join(target, "foo"), "w"): 358 pass 359 self.run_function("git.add", [target, "."]) 360 self.run_function( 361 "git.commit", 362 [target, "add file"], 363 git_opts='-c user.name="Foo Bar" -c user.email=foo@bar.com', 364 ) 365 366 # Re-run the state, this should fail with a specific hint in the 367 # comment field. 368 ret = self.run_state("git.latest", name=TEST_REPO, rev=rev, target=target) 369 self.assertSaltFalseReturn(ret) 370 371 comment = ret[next(iter(ret))]["comment"] 372 self.assertTrue(hint in comment) 373 374 @uses_git_opts 375 @pytest.mark.slow_test 376 def test_latest_changed_local_branch_rev_head(self): 377 """ 378 Test for presence of hint in failure message when the local branch has 379 been changed and a the rev is set to HEAD 380 381 This test will fail if the default branch for the salt-test-repo is 382 ever changed. 383 """ 384 self._changed_local_branch_helper( # pylint: disable=no-value-for-parameter 385 "HEAD", 386 "The default remote branch (develop) differs from the local " 387 "branch (new_branch)", 388 ) 389 390 @uses_git_opts 391 @pytest.mark.slow_test 392 def test_latest_changed_local_branch_rev_develop(self): 393 """ 394 Test for presence of hint in failure message when the local branch has 395 been changed and a non-HEAD rev is specified 396 """ 397 self._changed_local_branch_helper( # pylint: disable=no-value-for-parameter 398 "develop", 399 "The desired rev (develop) differs from the name of the local " 400 "branch (new_branch)", 401 ) 402 403 @uses_git_opts 404 @with_tempdir(create=False) 405 @with_tempdir() 406 @pytest.mark.slow_test 407 def test_latest_updated_remote_rev(self, name, target): 408 """ 409 Ensure that we don't exit early when checking for a fast-forward 410 """ 411 # Initialize a new git repository 412 self.run_function("git.init", [name]) 413 414 # Add and commit a file 415 with salt.utils.files.fopen(os.path.join(name, "foo.txt"), "w") as fp_: 416 fp_.write("Hello world\n") 417 self.run_function("git.add", [name, "."]) 418 self.run_function( 419 "git.commit", 420 [name, "initial commit"], 421 git_opts='-c user.name="Foo Bar" -c user.email=foo@bar.com', 422 ) 423 424 # Run the state to clone the repo we just created 425 ret = self.run_state( 426 "git.latest", 427 name=name, 428 target=target, 429 ) 430 self.assertSaltTrueReturn(ret) 431 432 # Add another commit 433 with salt.utils.files.fopen(os.path.join(name, "foo.txt"), "w") as fp_: 434 fp_.write("Added a line\n") 435 self.run_function( 436 "git.commit", 437 [name, "added a line"], 438 git_opts='-c user.name="Foo Bar" -c user.email=foo@bar.com', 439 opts="-a", 440 ) 441 442 # Run the state again. It should pass, if it doesn't then there was 443 # a problem checking whether or not the change is a fast-forward. 444 ret = self.run_state( 445 "git.latest", 446 name=name, 447 target=target, 448 ) 449 self.assertSaltTrueReturn(ret) 450 451 @with_tempdir(create=False) 452 @pytest.mark.slow_test 453 def test_latest_depth(self, target): 454 """ 455 Test running git.latest state using the "depth" argument to limit the 456 history. See #45394. 457 """ 458 ret = self.run_state( 459 "git.latest", name=TEST_REPO, rev="HEAD", target=target, depth=1 460 ) 461 # HEAD is not a branch, this should fail 462 self.assertSaltFalseReturn(ret) 463 self.assertIn( 464 "must be set to the name of a branch", ret[next(iter(ret))]["comment"] 465 ) 466 467 ret = self.run_state( 468 "git.latest", 469 name=TEST_REPO, 470 rev="non-default-branch", 471 target=target, 472 depth=1, 473 ) 474 self.assertSaltTrueReturn(ret) 475 self.assertTrue(os.path.isdir(os.path.join(target, ".git"))) 476 477 @with_git_mirror(TEST_REPO) 478 @uses_git_opts 479 @pytest.mark.slow_test 480 def test_latest_sync_tags(self, mirror_url, admin_dir, clone_dir): 481 """ 482 Test that a removed tag is properly reported as such and removed in the 483 local clone, and that new tags are reported as new. 484 """ 485 tag1 = "mytag1" 486 tag2 = "mytag2" 487 488 # Add and push a tag 489 self.run_function("git.tag", [admin_dir, tag1]) 490 self.run_function("git.push", [admin_dir, "origin", tag1]) 491 492 # Clone the repo 493 ret = self.run_state("git.latest", name=mirror_url, target=clone_dir) 494 ret = ret[next(iter(ret))] 495 assert ret["result"] 496 497 # Now remove the tag 498 self.run_function("git.push", [admin_dir, "origin", ":{}".format(tag1)]) 499 # Add and push another tag 500 self.run_function("git.tag", [admin_dir, tag2]) 501 self.run_function("git.push", [admin_dir, "origin", tag2]) 502 503 # Re-run the state with sync_tags=False. This should NOT delete the tag 504 # from the local clone, but should report that a tag has been added. 505 ret = self.run_state( 506 "git.latest", name=mirror_url, target=clone_dir, sync_tags=False 507 ) 508 ret = ret[next(iter(ret))] 509 assert ret["result"] 510 # Make ABSOLUTELY SURE both tags are present, since we shouldn't have 511 # removed tag1. 512 all_tags = self.run_function("git.list_tags", [clone_dir]) 513 assert tag1 in all_tags 514 assert tag2 in all_tags 515 # Make sure the reported changes are correct 516 expected_changes = {"new_tags": [tag2]} 517 assert ret["changes"] == expected_changes, ret["changes"] 518 519 # Re-run the state with sync_tags=True. This should remove the local 520 # tag, since it doesn't exist in the remote repository. 521 ret = self.run_state( 522 "git.latest", name=mirror_url, target=clone_dir, sync_tags=True 523 ) 524 ret = ret[next(iter(ret))] 525 assert ret["result"] 526 # Make ABSOLUTELY SURE the expected tags are present/gone 527 all_tags = self.run_function("git.list_tags", [clone_dir]) 528 assert tag1 not in all_tags 529 assert tag2 in all_tags 530 # Make sure the reported changes are correct 531 expected_changes = {"deleted_tags": [tag1]} 532 assert ret["changes"] == expected_changes, ret["changes"] 533 534 @with_tempdir(create=False) 535 @pytest.mark.slow_test 536 def test_cloned(self, target): 537 """ 538 Test git.cloned state 539 """ 540 # Test mode 541 ret = self.run_state("git.cloned", name=TEST_REPO, target=target, test=True) 542 ret = ret[next(iter(ret))] 543 assert ret["result"] is None 544 assert ret["changes"] == {"new": "{} => {}".format(TEST_REPO, target)} 545 assert ret["comment"] == "{} would be cloned to {}".format(TEST_REPO, target) 546 547 # Now actually run the state 548 ret = self.run_state("git.cloned", name=TEST_REPO, target=target) 549 ret = ret[next(iter(ret))] 550 assert ret["result"] is True 551 assert ret["changes"] == {"new": "{} => {}".format(TEST_REPO, target)} 552 assert ret["comment"] == "{} cloned to {}".format(TEST_REPO, target) 553 554 # Run the state again to test idempotence 555 ret = self.run_state("git.cloned", name=TEST_REPO, target=target) 556 ret = ret[next(iter(ret))] 557 assert ret["result"] is True 558 assert not ret["changes"] 559 assert ret["comment"] == "Repository already exists at {}".format(target) 560 561 # Run the state again to test idempotence (test mode) 562 ret = self.run_state("git.cloned", name=TEST_REPO, target=target, test=True) 563 ret = ret[next(iter(ret))] 564 assert not ret["changes"] 565 assert ret["result"] is True 566 assert ret["comment"] == "Repository already exists at {}".format(target) 567 568 @with_tempdir(create=False) 569 @pytest.mark.slow_test 570 def test_cloned_with_branch(self, target): 571 """ 572 Test git.cloned state with branch provided 573 """ 574 old_branch = "master" 575 new_branch = "develop" 576 bad_branch = "thisbranchdoesnotexist" 577 578 # Test mode 579 ret = self.run_state( 580 "git.cloned", name=TEST_REPO, target=target, branch=old_branch, test=True 581 ) 582 ret = ret[next(iter(ret))] 583 assert ret["result"] is None 584 assert ret["changes"] == {"new": "{} => {}".format(TEST_REPO, target)} 585 assert ret["comment"] == "{} would be cloned to {} with branch '{}'".format( 586 TEST_REPO, target, old_branch 587 ) 588 589 # Now actually run the state 590 ret = self.run_state( 591 "git.cloned", name=TEST_REPO, target=target, branch=old_branch 592 ) 593 ret = ret[next(iter(ret))] 594 assert ret["result"] is True 595 assert ret["changes"] == {"new": "{} => {}".format(TEST_REPO, target)} 596 assert ret["comment"] == "{} cloned to {} with branch '{}'".format( 597 TEST_REPO, target, old_branch 598 ) 599 600 # Run the state again to test idempotence 601 ret = self.run_state( 602 "git.cloned", name=TEST_REPO, target=target, branch=old_branch 603 ) 604 ret = ret[next(iter(ret))] 605 assert ret["result"] is True 606 assert not ret["changes"] 607 assert ret[ 608 "comment" 609 ] == "Repository already exists at {} and is checked out to branch '{}'".format( 610 target, old_branch 611 ) 612 613 # Run the state again to test idempotence (test mode) 614 ret = self.run_state( 615 "git.cloned", name=TEST_REPO, target=target, test=True, branch=old_branch 616 ) 617 ret = ret[next(iter(ret))] 618 assert ret["result"] is True 619 assert not ret["changes"] 620 assert ret[ 621 "comment" 622 ] == "Repository already exists at {} and is checked out to branch '{}'".format( 623 target, old_branch 624 ) 625 626 # Change branch (test mode) 627 ret = self.run_state( 628 "git.cloned", name=TEST_REPO, target=target, branch=new_branch, test=True 629 ) 630 ret = ret[next(iter(ret))] 631 assert ret["result"] is None 632 assert ret["changes"] == {"branch": {"old": old_branch, "new": new_branch}} 633 assert ret["comment"] == "Branch would be changed to '{}'".format(new_branch) 634 635 # Now really change the branch 636 ret = self.run_state( 637 "git.cloned", name=TEST_REPO, target=target, branch=new_branch 638 ) 639 ret = ret[next(iter(ret))] 640 assert ret["result"] is True 641 assert ret["changes"] == {"branch": {"old": old_branch, "new": new_branch}} 642 assert ret["comment"] == "Branch changed to '{}'".format(new_branch) 643 644 # Change back to original branch. This tests that we don't attempt to 645 # checkout a new branch (i.e. git checkout -b) for a branch that exists 646 # locally, as that would fail. 647 ret = self.run_state( 648 "git.cloned", name=TEST_REPO, target=target, branch=old_branch 649 ) 650 ret = ret[next(iter(ret))] 651 assert ret["result"] is True 652 assert ret["changes"] == {"branch": {"old": new_branch, "new": old_branch}} 653 assert ret["comment"] == "Branch changed to '{}'".format(old_branch) 654 655 # Test switching to a nonexistent branch. This should fail. 656 ret = self.run_state( 657 "git.cloned", name=TEST_REPO, target=target, branch=bad_branch 658 ) 659 ret = ret[next(iter(ret))] 660 assert ret["result"] is False 661 assert not ret["changes"] 662 assert ret["comment"].startswith( 663 "Failed to change branch to '{}':".format(bad_branch) 664 ) 665 666 @with_tempdir(create=False) 667 @ensure_min_git(min_version="1.7.10") 668 @pytest.mark.slow_test 669 def test_cloned_with_nonexistant_branch(self, target): 670 """ 671 Test git.cloned state with a nonexistent branch provided 672 """ 673 branch = "thisbranchdoesnotexist" 674 675 # Test mode 676 ret = self.run_state( 677 "git.cloned", name=TEST_REPO, target=target, branch=branch, test=True 678 ) 679 ret = ret[next(iter(ret))] 680 assert ret["result"] is None 681 assert ret["changes"] 682 assert ret["comment"] == "{} would be cloned to {} with branch '{}'".format( 683 TEST_REPO, target, branch 684 ) 685 686 # Now actually run the state 687 ret = self.run_state("git.cloned", name=TEST_REPO, target=target, branch=branch) 688 ret = ret[next(iter(ret))] 689 assert ret["result"] is False 690 assert not ret["changes"] 691 assert ret["comment"].startswith("Clone failed:") 692 assert "not found in upstream origin" in ret["comment"] 693 694 @with_tempdir(create=False) 695 @pytest.mark.slow_test 696 def test_present(self, name): 697 """ 698 git.present 699 """ 700 ret = self.run_state("git.present", name=name, bare=True) 701 self.assertSaltTrueReturn(ret) 702 self.assertTrue(os.path.isfile(os.path.join(name, "HEAD"))) 703 704 @with_tempdir() 705 @pytest.mark.slow_test 706 def test_present_failure(self, name): 707 """ 708 git.present 709 """ 710 fname = os.path.join(name, "stoptheprocess") 711 712 with salt.utils.files.fopen(fname, "a"): 713 pass 714 715 ret = self.run_state("git.present", name=name, bare=True) 716 self.assertSaltFalseReturn(ret) 717 self.assertFalse(os.path.isfile(os.path.join(name, "HEAD"))) 718 719 @with_tempdir() 720 @pytest.mark.slow_test 721 def test_present_empty_dir(self, name): 722 """ 723 git.present 724 """ 725 ret = self.run_state("git.present", name=name, bare=True) 726 self.assertSaltTrueReturn(ret) 727 self.assertTrue(os.path.isfile(os.path.join(name, "HEAD"))) 728 729 @with_tempdir() 730 @pytest.mark.slow_test 731 def test_config_set_value_with_space_character(self, name): 732 """ 733 git.config 734 """ 735 self.run_function("git.init", [name]) 736 737 ret = self.run_state( 738 "git.config_set", 739 name="user.name", 740 value="foo bar", 741 repo=name, 742 **{"global": False} 743 ) 744 self.assertSaltTrueReturn(ret) 745 746 747@ensure_min_git 748@uses_git_opts 749class LocalRepoGitTest(ModuleCase, SaltReturnAssertsMixin): 750 """ 751 Tests which do no require connectivity to github.com 752 """ 753 754 def setUp(self): 755 self.repo = tempfile.mkdtemp(dir=RUNTIME_VARS.TMP) 756 self.admin = tempfile.mkdtemp(dir=RUNTIME_VARS.TMP) 757 self.target = tempfile.mkdtemp(dir=RUNTIME_VARS.TMP) 758 for dirname in (self.repo, self.admin, self.target): 759 self.addCleanup(shutil.rmtree, dirname, ignore_errors=True) 760 761 # Create bare repo 762 self.run_function("git.init", [self.repo], bare=True) 763 # Clone bare repo 764 self.run_function("git.clone", [self.admin], url=self.repo) 765 self._commit(self.admin, "", message="initial commit") 766 self._push(self.admin) 767 768 def _commit(self, repo_path, content, message): 769 with salt.utils.files.fopen(os.path.join(repo_path, "foo"), "a") as fp_: 770 fp_.write(content) 771 self.run_function("git.add", [repo_path, "."]) 772 self.run_function( 773 "git.commit", 774 [repo_path, message], 775 git_opts='-c user.name="Foo Bar" -c user.email=foo@bar.com', 776 ) 777 778 def _push(self, repo_path, remote="origin", ref="master"): 779 self.run_function("git.push", [repo_path], remote=remote, ref=ref) 780 781 def _test_latest_force_reset_setup(self): 782 # Perform the initial clone 783 ret = self.run_state("git.latest", name=self.repo, target=self.target) 784 self.assertSaltTrueReturn(ret) 785 786 # Make and push changes to remote repo 787 self._commit(self.admin, content="Hello world!\n", message="added a line") 788 self._push(self.admin) 789 790 # Make local changes to clone, but don't commit them 791 with salt.utils.files.fopen(os.path.join(self.target, "foo"), "a") as fp_: 792 fp_.write("Local changes!\n") 793 794 @pytest.mark.slow_test 795 def test_latest_force_reset_remote_changes(self): 796 """ 797 This tests that an otherwise fast-forward change with local chanegs 798 will not reset local changes when force_reset='remote_changes' 799 """ 800 self._test_latest_force_reset_setup() 801 802 # This should fail because of the local changes 803 ret = self.run_state("git.latest", name=self.repo, target=self.target) 804 self.assertSaltFalseReturn(ret) 805 ret = ret[next(iter(ret))] 806 self.assertIn("there are uncommitted changes", ret["comment"]) 807 self.assertIn("Set 'force_reset' to True (or 'remote-changes')", ret["comment"]) 808 self.assertEqual(ret["changes"], {}) 809 810 # Now run again with force_reset='remote_changes', the state should 811 # succeed and discard the local changes 812 ret = self.run_state( 813 "git.latest", 814 name=self.repo, 815 target=self.target, 816 force_reset="remote-changes", 817 ) 818 self.assertSaltTrueReturn(ret) 819 ret = ret[next(iter(ret))] 820 self.assertIn("Uncommitted changes were discarded", ret["comment"]) 821 self.assertIn("Repository was fast-forwarded", ret["comment"]) 822 self.assertNotIn("forced update", ret["changes"]) 823 self.assertIn("revision", ret["changes"]) 824 825 # Add new local changes, but don't commit them 826 with salt.utils.files.fopen(os.path.join(self.target, "foo"), "a") as fp_: 827 fp_.write("More local changes!\n") 828 829 # Now run again with force_reset='remote_changes', the state should 830 # succeed with an up-to-date message and mention that there are local 831 # changes, telling the user how to discard them. 832 ret = self.run_state( 833 "git.latest", 834 name=self.repo, 835 target=self.target, 836 force_reset="remote-changes", 837 ) 838 self.assertSaltTrueReturn(ret) 839 ret = ret[next(iter(ret))] 840 self.assertIn("up-to-date, but with uncommitted changes", ret["comment"]) 841 self.assertIn( 842 "Set 'force_reset' to True to purge uncommitted changes", ret["comment"] 843 ) 844 self.assertEqual(ret["changes"], {}) 845 846 @pytest.mark.slow_test 847 def test_latest_force_reset_true_fast_forward(self): 848 """ 849 This tests that an otherwise fast-forward change with local chanegs 850 does reset local changes when force_reset=True 851 """ 852 self._test_latest_force_reset_setup() 853 854 # Test that local changes are discarded and that we fast-forward 855 ret = self.run_state( 856 "git.latest", name=self.repo, target=self.target, force_reset=True 857 ) 858 self.assertSaltTrueReturn(ret) 859 ret = ret[next(iter(ret))] 860 self.assertIn("Uncommitted changes were discarded", ret["comment"]) 861 self.assertIn("Repository was fast-forwarded", ret["comment"]) 862 863 # Add new local changes 864 with salt.utils.files.fopen(os.path.join(self.target, "foo"), "a") as fp_: 865 fp_.write("More local changes!\n") 866 867 # Running without setting force_reset should mention uncommitted changes 868 ret = self.run_state("git.latest", name=self.repo, target=self.target) 869 self.assertSaltTrueReturn(ret) 870 ret = ret[next(iter(ret))] 871 self.assertIn("up-to-date, but with uncommitted changes", ret["comment"]) 872 self.assertIn( 873 "Set 'force_reset' to True to purge uncommitted changes", ret["comment"] 874 ) 875 self.assertEqual(ret["changes"], {}) 876 877 # Test that local changes are discarded 878 ret = self.run_state( 879 "git.latest", name=TEST_REPO, target=self.target, force_reset=True 880 ) 881 self.assertSaltTrueReturn(ret) 882 ret = ret[next(iter(ret))] 883 assert "Uncommitted changes were discarded" in ret["comment"] 884 assert "Repository was hard-reset" in ret["comment"] 885 assert "forced update" in ret["changes"] 886 887 @pytest.mark.slow_test 888 def test_latest_force_reset_true_non_fast_forward(self): 889 """ 890 This tests that a non fast-forward change with divergent commits fails 891 unless force_reset=True. 892 """ 893 self._test_latest_force_reset_setup() 894 895 # Reset to remote HEAD 896 ret = self.run_state( 897 "git.latest", name=self.repo, target=self.target, force_reset=True 898 ) 899 self.assertSaltTrueReturn(ret) 900 ret = ret[next(iter(ret))] 901 self.assertIn("Uncommitted changes were discarded", ret["comment"]) 902 self.assertIn("Repository was fast-forwarded", ret["comment"]) 903 904 # Make and push changes to remote repo 905 self._commit(self.admin, content="New line\n", message="added another line") 906 self._push(self.admin) 907 908 # Make different changes to local file and commit locally 909 self._commit( 910 self.target, 911 content="Different new line\n", 912 message="added a different line", 913 ) 914 915 # This should fail since the local clone has diverged and cannot 916 # fast-forward to the remote rev 917 ret = self.run_state("git.latest", name=self.repo, target=self.target) 918 self.assertSaltFalseReturn(ret) 919 ret = ret[next(iter(ret))] 920 self.assertIn("this is not a fast-forward merge", ret["comment"]) 921 self.assertIn("Set 'force_reset' to True to force this update", ret["comment"]) 922 self.assertEqual(ret["changes"], {}) 923 924 # Repeat the state with force_reset=True and confirm that the hard 925 # reset was performed 926 ret = self.run_state( 927 "git.latest", name=self.repo, target=self.target, force_reset=True 928 ) 929 self.assertSaltTrueReturn(ret) 930 ret = ret[next(iter(ret))] 931 self.assertIn("Repository was hard-reset", ret["comment"]) 932 self.assertIn("forced update", ret["changes"]) 933 self.assertIn("revision", ret["changes"]) 934 935 @pytest.mark.slow_test 936 def test_renamed_default_branch(self): 937 """ 938 Test the case where the remote branch has been removed 939 https://github.com/saltstack/salt/issues/36242 940 """ 941 # Rename remote 'master' branch to 'develop' 942 os.rename( 943 os.path.join(self.repo, "refs", "heads", "master"), 944 os.path.join(self.repo, "refs", "heads", "develop"), 945 ) 946 947 # Run git.latest state. This should successfully clone and fail with a 948 # specific error in the comment field. 949 ret = self.run_state( 950 "git.latest", 951 name=self.repo, 952 target=self.target, 953 rev="develop", 954 ) 955 self.assertSaltFalseReturn(ret) 956 self.assertEqual( 957 ret[next(iter(ret))]["comment"], 958 "Remote HEAD refers to a ref that does not exist. " 959 "This can happen when the default branch on the " 960 "remote repository is renamed or deleted. If you " 961 "are unable to fix the remote repository, you can " 962 "work around this by setting the 'branch' argument " 963 "(which will ensure that the named branch is created " 964 "if it does not already exist).\n\n" 965 "Changes already made: {} cloned to {}".format(self.repo, self.target), 966 ) 967 self.assertEqual( 968 ret[next(iter(ret))]["changes"], 969 {"new": "{} => {}".format(self.repo, self.target)}, 970 ) 971 972 # Run git.latest state again. This should fail again, with a different 973 # error in the comment field, and should not change anything. 974 ret = self.run_state( 975 "git.latest", 976 name=self.repo, 977 target=self.target, 978 rev="develop", 979 ) 980 self.assertSaltFalseReturn(ret) 981 self.assertEqual( 982 ret[next(iter(ret))]["comment"], 983 "Cannot set/unset upstream tracking branch, local " 984 "HEAD refers to nonexistent branch. This may have " 985 "been caused by cloning a remote repository for which " 986 "the default branch was renamed or deleted. If you " 987 "are unable to fix the remote repository, you can " 988 "work around this by setting the 'branch' argument " 989 "(which will ensure that the named branch is created " 990 "if it does not already exist).", 991 ) 992 self.assertEqual(ret[next(iter(ret))]["changes"], {}) 993 994 # Run git.latest state again with a branch manually set. This should 995 # checkout a new branch and the state should pass. 996 ret = self.run_state( 997 "git.latest", 998 name=self.repo, 999 target=self.target, 1000 rev="develop", 1001 branch="develop", 1002 ) 1003 # State should succeed 1004 self.assertSaltTrueReturn(ret) 1005 self.assertSaltCommentRegexpMatches( 1006 ret, 1007 "New branch 'develop' was checked out, with origin/develop " 1008 r"\([0-9a-f]{7}\) as a starting point", 1009 ) 1010 # Only the revision should be in the changes dict. 1011 self.assertEqual(list(ret[next(iter(ret))]["changes"].keys()), ["revision"]) 1012 # Since the remote repo was incorrectly set up, the local head should 1013 # not exist (therefore the old revision should be None). 1014 self.assertEqual(ret[next(iter(ret))]["changes"]["revision"]["old"], None) 1015 # Make sure the new revision is a SHA (40 chars, all hex) 1016 self.assertTrue(len(ret[next(iter(ret))]["changes"]["revision"]["new"]) == 40) 1017 self.assertTrue( 1018 all( 1019 [ 1020 x in string.hexdigits 1021 for x in ret[next(iter(ret))]["changes"]["revision"]["new"] 1022 ] 1023 ) 1024 ) 1025