diff --git a/manic/repository_git.py b/manic/repository_git.py index 9a7002d1e5..2c5447ad40 100644 --- a/manic/repository_git.py +++ b/manic/repository_git.py @@ -13,7 +13,7 @@ from .repository import Repository from .externals_status import ExternalStatus from .utils import expand_local_url, split_remote_url, is_remote_url -from .utils import log_process_output +from .utils import log_process_output, fatal_error from .utils import execute_subprocess, check_output @@ -35,14 +35,13 @@ class GitRepository(Repository): """ - GIT_REF_UNKNOWN = 'unknown' - GIT_REF_LOCAL_BRANCH = 'localBranch' - GIT_REF_REMOTE_BRANCH = 'remoteBranch' - GIT_REF_TAG = 'gitTag' - GIT_REF_SHA1 = 'gitSHA1' + # match XYZ of '* (HEAD detached at {XYZ}): + # e.g. * (HEAD detached at origin/feature-2) + RE_DETACHED = re.compile(r'\* \([\w]+[\s]+detached at ([\w\-./]+)\)') - RE_GITHASH = re.compile(r"\A([a-fA-F0-9]+)\Z") - RE_REMOTEBRANCH = re.compile(r"\s*origin/(\S+)") + # match tracking reference info, return XYZ from [XYZ] + # e.g. [origin/master] + RE_TRACKING = re.compile(r'\[([\w\-./]+)\]') def __init__(self, component_name, repo): """ @@ -98,11 +97,31 @@ def _clone_repo(self, base_dir_path, repo_dir_name): self._git_clone(self._url, repo_dir_name) os.chdir(cwd) - @staticmethod - def _current_ref_from_branch_command(git_output): + def _current_ref_from_branch_command(self, git_output): """Parse output of the 'git branch' command to determine the current branch. The line starting with '*' is the current branch. It can be one of: + feature2 36418b4 [origin/feature2] Work on feature2 +* feature3 36418b4 Work on feature2 + master 9b75494 [origin/master] Initialize repository. + +* (HEAD detached at 36418b4) 36418b4 Work on feature2 + feature2 36418b4 [origin/feature2] Work on feature2 + master 9b75494 [origin/master] Initialize repository. + +* (HEAD detached at origin/feature2) 36418b4 Work on feature2 + feature2 36418b4 [origin/feature2] Work on feature2 + feature3 36418b4 Work on feature2 + master 9b75494 [origin/master] Initialize repository. + + Possible head states: + + * detached from remote branch --> ref = remote/branch + * detached from tag --> ref = tag + * detached from sha --> ref = sha + * on local branch --> ref = branch + * on tracking branch --> ref = remote/branch + On a branch: * cm-testing @@ -118,18 +137,36 @@ def _current_ref_from_branch_command(git_output): """ lines = git_output.splitlines() - current_branch = None + ref = '' for line in lines: if line.startswith('*'): - current_branch = line - ref = EMPTY_STR - if current_branch: - if 'detached' in current_branch: - ref = current_branch.split(' ')[-1] - ref = ref.strip(')') - else: - ref = current_branch.split()[-1] - return ref + ref = line + break + current_ref = EMPTY_STR + if not ref: + # not a git repo? some other error? we return so the + # caller can handle. + pass + elif 'detached' in ref: + match = self.RE_DETACHED.search(ref) + try: + current_ref = match.group(1) + except BaseException: + msg = 'DEV_ERROR: regex to detect detached head state failed!' + msg += '\nref:\n{0}\ngit_output\n{1}\n'.format(ref, git_output) + fatal_error(msg) + elif '[' in ref: + match = self.RE_TRACKING.search(ref) + try: + current_ref = match.group(1) + except BaseException: + msg = 'DEV_ERROR: regex to detect tracking branch failed.' + else: + # assumed local branch + current_ref = ref.split()[1] + + current_ref = current_ref.strip() + return current_ref def _check_sync(self, stat, repo_dir_path): """Determine whether a git repository is in-sync with the model @@ -152,23 +189,41 @@ def _check_sync(self, stat, repo_dir_path): # finds the parent repo git dir! stat.sync_state = ExternalStatus.UNKNOWN else: - cwd = os.getcwd() - os.chdir(repo_dir_path) - git_output = self._git_branch() - ref = self._current_ref_from_branch_command(git_output) - if ref == EMPTY_STR: - stat.sync_state = ExternalStatus.UNKNOWN - elif self._tag: - if self._tag == ref: - stat.sync_state = ExternalStatus.STATUS_OK - else: - stat.sync_state = ExternalStatus.MODEL_MODIFIED + self._check_sync_logic(stat, repo_dir_path) + + def _check_sync_logic(self, stat, repo_dir_path): + """Isolate the complicated synce logic so it is not so deeply nested + and a bit easier to understand. + + Sync logic - only reporting on whether we are on the ref + (branch, tag, hash) specified in the externals description. + + + """ + cwd = os.getcwd() + os.chdir(repo_dir_path) + git_output = self._git_branch_vv() + current_ref = self._current_ref_from_branch_command(git_output) + if current_ref == EMPTY_STR: + stat.sync_state = ExternalStatus.UNKNOWN + elif self._branch: + remote_name = self._determine_remote_name() + if not remote_name: + # git doesn't know about this remote. by definition + # this is a modefied state. + stat.sync_state = ExternalStatus.MODEL_MODIFIED + else: + expected_ref = "{0}/{1}".format(remote_name, self._branch) + if current_ref == expected_ref: + stat.sync_state = ExternalStatus.STATUS_OK else: - if self._branch == ref: - stat.sync_state = ExternalStatus.STATUS_OK - else: - stat.sync_state = ExternalStatus.MODEL_MODIFIED - os.chdir(cwd) + stat.sync_state = ExternalStatus.MODEL_MODIFIED + else: + if self._tag == current_ref: + stat.sync_state = ExternalStatus.STATUS_OK + else: + stat.sync_state = ExternalStatus.MODEL_MODIFIED + os.chdir(cwd) def _determine_remote_name(self): """Return the remote name. @@ -181,9 +236,15 @@ def _determine_remote_name(self): git_output = git_output.splitlines() remote_name = '' for line in git_output: - if self._url in line: - data = line.split() - remote_name = data[0].strip() + data = line.strip() + if not data: + continue + data = data.split() + name = data[0].strip() + url = data[1].strip() + if self._url == url: + remote_name = name + break return remote_name def _create_remote_name(self): @@ -222,9 +283,7 @@ def _create_remote_name(self): url = split_remote_url(url) else: url = expand_local_url(url, self._name) - print(url) url = url.split('/') - print(url) repo_name = url[-1] base_name = url[-2] # repo name should nominally already be something that git can @@ -245,12 +304,83 @@ def _checkout_external_ref(self, repo_dir): self._git_remote_add(remote_name, self._url) self._git_fetch(remote_name) if self._tag: + is_unique_tag = self._is_unique_tag(self._tag) + if not is_unique_tag: + msg = ('In repo "{1}": tag "{0}" is either not a valid ' + 'reference or is shadowed by a branch.'.format( + self._tag, self._name)) + fatal_error(msg) ref = self._tag else: ref = '{0}/{1}'.format(remote_name, self._branch) self._git_checkout_ref(ref) os.chdir(cwd) + def _is_unique_tag(self, ref): + """Verify that a reference is a valid tag and is unique (not a branch) + + Tags may be tag names, or SHA id's. It is also possible that a + branch and tag have the some name. + + Note: values returned by git_showref_* and git_revparse are + shell return codes, which are zero for success, non-zero for + error! + + """ + is_tag = self._ref_is_tag(ref) + is_branch = self._ref_is_branch(ref) + is_commit = self._ref_is_commit(ref) + + is_unique_tag = False + if is_tag and not is_branch: + # unique tag + is_unique_tag = True + if not is_branch and is_commit: + # probably a sha1 or HEAD, etc, we call it a tag + is_unique_tag = True + return is_unique_tag + + def _ref_is_tag(self, ref): + """Verify that a reference is a valid tag according to git. + + Note: values returned by git_showref_* and git_revparse are + shell return codes, which are zero for success, non-zero for + error! + """ + is_tag = False + value = self._git_showref_tag(ref) + if value == 0: + is_tag = True + return is_tag + + def _ref_is_branch(self, ref): + """Verify that a reference is a valid branch according to git. + + Note: values returned by git_showref_* and git_revparse are + shell return codes, which are zero for success, non-zero for + error! + """ + is_branch = False + value = self._git_showref_branch(ref) + if value == 0: + is_branch = True + return is_branch + + def _ref_is_commit(self, ref): + """Verify that a reference is a valid commit according to git. + + This could be a tag, branch, sha1 id, HEAD and potentially others... + + Note: values returned by git_showref_* and git_revparse are + shell return codes, which are zero for success, non-zero for + error! + """ + is_commit = False + value = self._git_revparse_commit(ref) + if value == 0: + is_commit = True + return is_commit + def _status_summary(self, stat, repo_dir_path): """Determine the clean/dirty status of a git repository @@ -307,13 +437,47 @@ def _status_v1z_is_dirty(git_output): # # ---------------------------------------------------------------- @staticmethod - def _git_branch(): - """Run git branch to obtain repository information + def _git_branch_vv(): + """Run git branch -vv to obtain verbose branch information, including + upstream tracking and hash. + """ - cmd = ['git', 'branch'] + cmd = ['git', 'branch', '--verbose', '--verbose'] git_output = check_output(cmd) return git_output + @staticmethod + def _git_showref_tag(ref): + """Run git show-ref check if the user supplied ref is a tag. + + could also use git rev-parse --quiet --verify tagname^{tag} + """ + cmd = ['git', 'show-ref', '--quiet', '--verify', + 'refs/tags/{0}'.format(ref), ] + status = execute_subprocess(cmd, status_to_caller=True) + return status + + @staticmethod + def _git_showref_branch(ref): + """Run git show-ref check if the user supplied ref is a branch. + + """ + cmd = ['git', 'show-ref', '--quiet', '--verify', + 'refs/heads/{0}'.format(ref), ] + status = execute_subprocess(cmd, status_to_caller=True) + return status + + @staticmethod + def _git_revparse_commit(ref): + """Run git rev-parse to detect if a reference is a SHA, HEAD or other + valid commit. + + """ + cmd = ['git', 'rev-parse', '--quiet', '--verify', + '{0}^{1}'.format(ref, '{commit}'), ] + status = execute_subprocess(cmd, status_to_caller=True) + return status + @staticmethod def _git_status_porcelain_v1z(): """Run git status to obtain repository information. diff --git a/test/test_sys_checkout.py b/test/test_sys_checkout.py index 2b6bfb39fe..dfd2a30fa9 100644 --- a/test/test_sys_checkout.py +++ b/test/test_sys_checkout.py @@ -32,11 +32,12 @@ import os.path import random import shutil -import subprocess import unittest from manic.externals_description import ExternalsDescription from manic.externals_description import DESCRIPTION_SECTION, VERSION_ITEM +from manic.repository_git import GitRepository +from manic.utils import printlog from manic import checkout # ConfigParser was renamed in python2 to configparser. In python2, @@ -290,18 +291,10 @@ def setup_test_repo(self, parent_repo_name): parent_repo_dir = os.path.join(self._bare_root, parent_repo_name) dest_dir = os.path.join(os.environ[MANIC_TEST_TMP_REPO_ROOT], test_dir_name) - self.clone_repo(parent_repo_dir, dest_dir) + # pylint: disable=W0212 + GitRepository._git_clone(parent_repo_dir, dest_dir) return dest_dir - @staticmethod - def clone_repo(base_repo_dir, dest_dir): - """Call git to clone the repository - - """ - cmd = ['git', 'clone', base_repo_dir, dest_dir] - output = subprocess.check_output(cmd, stderr=subprocess.STDOUT) - return output - @staticmethod def execute_cmd_in_dir(under_test_dir, args): """Extecute the checkout command in the appropriate repo dir with the @@ -313,9 +306,16 @@ def execute_cmd_in_dir(under_test_dir, args): """ cwd = os.getcwd() + checkout_path = os.path.abspath('{0}/../../checkout_externals') os.chdir(under_test_dir) cmdline = ['--externals', CFG_NAME, ] cmdline += args + repo_root = 'MANIC_TEST_BARE_REPO_ROOT={root}'.format( + root=os.environ[MANIC_TEST_BARE_REPO_ROOT]) + manual_cmd = ('Test cmd:\ncd {cwd}; {env} {checkout} {args}'.format( + cwd=under_test_dir, env=repo_root, checkout=checkout_path, + args=' '.join(cmdline))) + printlog(manual_cmd) options = checkout.commandline_arguments(cmdline) status = checkout.main(options) os.chdir(cwd) diff --git a/test/test_unit_repository_git.py b/test/test_unit_repository_git.py index 83831ed9b1..e083e3b1d7 100644 --- a/test/test_unit_repository_git.py +++ b/test/test_unit_repository_git.py @@ -13,6 +13,7 @@ import os import shutil +import string import unittest from manic.repository_git import GitRepository @@ -24,32 +25,62 @@ # pylint: disable=W0212 -class TestGitRepositoryCurrentRefBranch(unittest.TestCase): - """test the current_ref_from_branch_command on a git repository - """ - GIT_BRANCH_OUTPUT_DETACHED_TAG = ''' -* (HEAD detached at rtm1_0_26) - a_feature_branch - master +GIT_BRANCH_OUTPUT_DETACHED_BRANCH = ''' +* (HEAD detached at origin/feature-2) 36418b4 Work on feature-2 + feature-2 36418b4 [origin/feature-2] Work on feature-2 + feature3 36418b4 Work on feature-2 + master 9b75494 [origin/master] Initialize repository. ''' - GIT_BRANCH_OUTPUT_BRANCH = ''' -* great_new_feature_branch - a_feature_branch - master + +GIT_BRANCH_OUTPUT_DETACHED_HASH = ''' +* (HEAD detached at 36418b4) 36418b4 Work on feature-2 + feature-2 36418b4 [origin/feature-2] Work on feature-2 + feature3 36418b4 Work on feature-2 + master 9b75494 [origin/master] Initialize repository. ''' - GIT_BRANCH_OUTPUT_HASH = ''' -* (HEAD detached at 0246874c) - a_feature_branch - master + +GIT_BRANCH_OUTPUT_DETACHED_TAG = ''' +* (HEAD detached at tag1) 9b75494 Initialize repository. + feature-2 36418b4 [origin/feature-2] Work on feature-2 + feature3 36418b4 Work on feature-2 + master 9b75494 [origin/master] Initialize repository. +''' + +GIT_BRANCH_OUTPUT_UNTRACKED_BRANCH = ''' + feature-2 36418b4 [origin/feature-2] Work on feature-2 +* feature3 36418b4 Work on feature-2 + master 9b75494 [origin/master] Initialize repository. ''' +GIT_BRANCH_OUTPUT_TRACKING_BRANCH = ''' +* feature-2 36418b4 [origin/feature-2] Work on feature-2 + feature3 36418b4 Work on feature-2 + master 9b75494 [origin/master] Initialize repository. +''' + +# NOTE(bja, 2017-11) order is important here. origin should be a +# subset of other to trap errors on processing remotes! +GIT_REMOTE_OUTPUT_ORIGIN_UPSTREAM = ''' +upstream /path/to/other/repo (fetch) +upstream /path/to/other/repo (push) +other /path/to/local/repo2 (fetch) +other /path/to/local/repo2 (push) +origin /path/to/local/repo (fetch) +origin /path/to/local/repo (push) +''' + + +class TestGitRepositoryCurrentRefBranch(unittest.TestCase): + """test the current_ref_from_branch_command on a git repository + """ + def setUp(self): self._name = 'component' rdata = {ExternalsDescription.PROTOCOL: 'git', ExternalsDescription.REPO_URL: - 'git@git.github.com:ncar/rtm', + '/path/to/local/repo', ExternalsDescription.TAG: - 'rtm1_0_26', + 'tag1', ExternalsDescription.BRANCH: EMPTY_STR } @@ -69,28 +100,46 @@ def setUp(self): def test_ref_detached_from_tag(self): """Test that we correctly identify that the ref is detached from a tag """ - git_output = self.GIT_BRANCH_OUTPUT_DETACHED_TAG + git_output = GIT_BRANCH_OUTPUT_DETACHED_TAG expected = self._repo.tag() result = self._repo._current_ref_from_branch_command( git_output) self.assertEqual(result, expected) - def test_ref_branch(self): - """Test that we correctly identify we are on a branch + def test_ref_detached_hash(self): + """Test that we can identify ref is detached from a hash + """ - git_output = self.GIT_BRANCH_OUTPUT_BRANCH - expected = 'great_new_feature_branch' + git_output = GIT_BRANCH_OUTPUT_DETACHED_HASH + expected = '36418b4' result = self._repo._current_ref_from_branch_command( git_output) self.assertEqual(result, expected) - def test_ref_detached_hash(self): - """Test that we can handle an empty string for output, e.g. not an git - repo. + def test_ref_detached_branch(self): + """Test that we can identify ref is detached from a remote branch + + """ + git_output = GIT_BRANCH_OUTPUT_DETACHED_BRANCH + expected = 'origin/feature-2' + result = self._repo._current_ref_from_branch_command( + git_output) + self.assertEqual(result, expected) + def test_ref_tracking_branch(self): + """Test that we correctly identify we are on a tracking branch """ - git_output = self.GIT_BRANCH_OUTPUT_HASH - expected = '0246874c' + git_output = GIT_BRANCH_OUTPUT_TRACKING_BRANCH + expected = 'origin/feature-2' + result = self._repo._current_ref_from_branch_command( + git_output) + self.assertEqual(result, expected) + + def test_ref_untracked_branch(self): + """Test that we correctly identify we are on an untracked branch + """ + git_output = GIT_BRANCH_OUTPUT_UNTRACKED_BRANCH + expected = 'feature3' result = self._repo._current_ref_from_branch_command( git_output) self.assertEqual(result, expected) @@ -101,17 +150,47 @@ def test_ref_none(self): """ git_output = EMPTY_STR - expected = EMPTY_STR - result = self._repo._current_ref_from_branch_command( + received = self._repo._current_ref_from_branch_command( git_output) - self.assertEqual(result, expected) + self.assertEqual(received, EMPTY_STR) class TestGitRepositoryCheckSync(unittest.TestCase): - """Test whether the GitRepository git_check_sync functionality is + """Test whether the GitRepository _check_sync_logic functionality is correct. + Note: there are a lot of combinations of state: + + - external description - tag, branch + + - working copy + - doesn't exist (not checked out) + - exists, no git info - incorrect protocol, e.g. svn, or tarball? + - exists, git info + - as expected: + - different from expected: + - detached tag, + - detached hash, + - detached branch (compare remote and branch), + - tracking branch (compare remote and branch), + - same remote + - different remote + - untracked branch + + Test list: + - doesn't exist + - exists no git info + + - num_external * (working copy expected + num_working copy different) + - total tests = 16 + """ + + # NOTE(bja, 2017-11) pylint complains about long method names, but + # it is hard to differentiate tests without making them more + # cryptic. + # pylint: disable=invalid-name + TMP_FAKE_DIR = 'fake' TMP_FAKE_GIT_DIR = os.path.join(TMP_FAKE_DIR, '.git') @@ -121,9 +200,8 @@ def setUp(self): self._name = 'component' rdata = {ExternalsDescription.PROTOCOL: 'git', ExternalsDescription.REPO_URL: - 'git@git.github.com:ncar/rtm', - ExternalsDescription.TAG: - 'rtm1_0_26', + '/path/to/local/repo', + ExternalsDescription.TAG: 'tag1', ExternalsDescription.BRANCH: EMPTY_STR } @@ -139,54 +217,82 @@ def setUp(self): model = ExternalsDescriptionDict(data) repo = model[self._name][ExternalsDescription.REPO] self._repo = GitRepository('test', repo) - self.create_tmp_git_dir() + self._create_tmp_git_dir() def tearDown(self): """Cleanup tmp stuff on the file system """ - self.remove_tmp_git_dir() + self._remove_tmp_git_dir() - def create_tmp_git_dir(self): + def _create_tmp_git_dir(self): """Create a temporary fake git directory for testing purposes. """ if not os.path.exists(self.TMP_FAKE_GIT_DIR): os.makedirs(self.TMP_FAKE_GIT_DIR) - def remove_tmp_git_dir(self): + def _remove_tmp_git_dir(self): """Remove the temporary fake git directory """ if os.path.exists(self.TMP_FAKE_DIR): shutil.rmtree(self.TMP_FAKE_DIR) + # + # mock methods replacing git system calls + # @staticmethod def _git_branch_empty(): - """Return an empty info string. Simulates svn info failing. + """Return an empty info string. Simulates git info failing. """ - return '' + return EMPTY_STR @staticmethod - def _git_branch_synced(): - """Return an info sting that is synced with the setUp data + def _git_branch_detached_tag(): + """Return an info sting that is a checkouted tag """ - git_output = ''' -* (HEAD detached at rtm1_0_26) - a_feature_branch - master -''' - return git_output + return GIT_BRANCH_OUTPUT_DETACHED_TAG @staticmethod - def _git_branch_modified(): - """Return and info string that is modified from the setUp data + def _git_branch_detached_hash(): + """Return an info string that is a checkout hash """ - git_output = ''' -* great_new_feature_branch - a_feature_branch - master -''' - return git_output + return GIT_BRANCH_OUTPUT_DETACHED_HASH + + @staticmethod + def _git_branch_detached_branch(): + """Return an info string that is a checkout hash + """ + return GIT_BRANCH_OUTPUT_DETACHED_BRANCH + + @staticmethod + def _git_branch_untracked_branch(): + """Return an info string that is a checkout branch + """ + return GIT_BRANCH_OUTPUT_UNTRACKED_BRANCH + + @staticmethod + def _git_branch_tracked_branch(): + """Return an info string that is a checkout branch + """ + return GIT_BRANCH_OUTPUT_TRACKING_BRANCH - def test_repo_dir_not_exist(self): + @staticmethod + def _git_remote_origin_upstream(): + """Return an info string that is a checkout hash + """ + return GIT_REMOTE_OUTPUT_ORIGIN_UPSTREAM + + @staticmethod + def _git_remote_none(): + """Return an info string that is a checkout hash + """ + return EMPTY_STR + + # ---------------------------------------------------------------- + # + # Tests where working copy doesn't exist or is invalid + # + # ---------------------------------------------------------------- + def test_sync_dir_not_exist(self): """Test that a directory that doesn't exist returns an error status Note: the Repository classes should be prevented from ever @@ -199,46 +305,371 @@ def test_repo_dir_not_exist(self): # check_dir should only modify the sync_state, not clean_state self.assertEqual(stat.clean_state, ExternalStatus.DEFAULT) - def test_repo_dir_exist_no_git_info(self): + def test_sync_dir_exist_no_git_info(self): """Test that an empty info string returns an unknown status """ stat = ExternalStatus() # Now we over-ride the _git_branch method on the repo to return # a known value without requiring access to git. - self._repo._git_branch = self._git_branch_empty + self._repo._git_remote_verbose = self._git_remote_origin_upstream + self._repo._git_branch_vv = self._git_branch_empty self._repo._check_sync(stat, self.TMP_FAKE_DIR) self.assertEqual(stat.sync_state, ExternalStatus.UNKNOWN) # check_sync should only modify the sync_state, not clean_state self.assertEqual(stat.clean_state, ExternalStatus.DEFAULT) - def test_repo_dir_synced(self): - """Test that a valid info string that is synced to the repo in the - externals description returns an ok status. + # ---------------------------------------------------------------- + # + # Tests where external description specifies a tag + # + # Perturbations of working dir state: on detached + # {tag|branch|hash}, tracking branch, untracked branch. + # + # ---------------------------------------------------------------- + def test_sync_tag_on_detached_tag(self): + """Test expect tag on detached tag --> status ok """ stat = ExternalStatus() - # Now we over-ride the _git_branch method on the repo to return - # a known value without requiring access to svn. - self._repo._git_branch = self._git_branch_synced - self._repo._check_sync(stat, self.TMP_FAKE_DIR) + self._repo._git_remote_verbose = self._git_remote_origin_upstream + self._repo._branch = '' + self._repo._tag = 'tag1' + self._repo._git_branch_vv = self._git_branch_detached_tag + self._repo._check_sync_logic(stat, self.TMP_FAKE_DIR) self.assertEqual(stat.sync_state, ExternalStatus.STATUS_OK) # check_sync should only modify the sync_state, not clean_state self.assertEqual(stat.clean_state, ExternalStatus.DEFAULT) - def test_repo_dir_modified(self): - """Test that a valid svn info string that is out of sync with the - externals description returns a modified status. + def test_sync_tag_on_diff_tag(self): + """Test expect tag on diff tag --> status modified """ stat = ExternalStatus() - # Now we over-ride the _git_branch method on the repo to return - # a known value without requiring access to svn. - self._repo._git_branch = self._git_branch_modified - self._repo._check_sync(stat, self.TMP_FAKE_DIR) + self._repo._git_remote_verbose = self._git_remote_origin_upstream + self._repo._branch = '' + self._repo._tag = 'tag2' + self._repo._git_branch_vv = self._git_branch_detached_tag + self._repo._check_sync_logic(stat, self.TMP_FAKE_DIR) + self.assertEqual(stat.sync_state, ExternalStatus.MODEL_MODIFIED) + # check_sync should only modify the sync_state, not clean_state + self.assertEqual(stat.clean_state, ExternalStatus.DEFAULT) + + def test_sync_tag_on_detached_hash(self): + """Test expect tag on detached hash --> status modified + + """ + stat = ExternalStatus() + self._repo._git_remote_verbose = self._git_remote_origin_upstream + self._repo._branch = '' + self._repo._tag = 'tag1' + self._repo._git_branch_vv = self._git_branch_detached_hash + self._repo._check_sync_logic(stat, self.TMP_FAKE_DIR) + self.assertEqual(stat.sync_state, ExternalStatus.MODEL_MODIFIED) + # check_sync should only modify the sync_state, not clean_state + self.assertEqual(stat.clean_state, ExternalStatus.DEFAULT) + + def test_sync_tag_on_detached_branch(self): + """Test expect tag on detached branch --> status modified + + """ + stat = ExternalStatus() + self._repo._git_remote_verbose = self._git_remote_origin_upstream + self._repo._branch = '' + self._repo._tag = 'tag1' + self._repo._git_branch_vv = self._git_branch_detached_branch + self._repo._check_sync_logic(stat, self.TMP_FAKE_DIR) + self.assertEqual(stat.sync_state, ExternalStatus.MODEL_MODIFIED) + # check_sync should only modify the sync_state, not clean_state + self.assertEqual(stat.clean_state, ExternalStatus.DEFAULT) + + def test_sync_tag_on_tracking_branch(self): + """Test expect tag on tracking branch --> status modified + + """ + stat = ExternalStatus() + self._repo._git_remote_verbose = self._git_remote_origin_upstream + self._repo._branch = '' + self._repo._tag = 'tag1' + self._repo._git_branch_vv = self._git_branch_tracked_branch + self._repo._check_sync_logic(stat, self.TMP_FAKE_DIR) + self.assertEqual(stat.sync_state, ExternalStatus.MODEL_MODIFIED) + # check_sync should only modify the sync_state, not clean_state + self.assertEqual(stat.clean_state, ExternalStatus.DEFAULT) + + def test_sync_tag_on_untracked_branch(self): + """Test expect tag on untracked branch --> status modified + + """ + stat = ExternalStatus() + self._repo._git_remote_verbose = self._git_remote_origin_upstream + self._repo._branch = '' + self._repo._tag = 'tag1' + self._repo._git_branch_vv = self._git_branch_untracked_branch + self._repo._check_sync_logic(stat, self.TMP_FAKE_DIR) + self.assertEqual(stat.sync_state, ExternalStatus.MODEL_MODIFIED) + # check_sync should only modify the sync_state, not clean_state + self.assertEqual(stat.clean_state, ExternalStatus.DEFAULT) + + # ---------------------------------------------------------------- + # + # Tests where external description specifies a branch + # + # Perturbations of working dir state: on detached + # {tag|branch|hash}, tracking branch, untracked branch. + # + # ---------------------------------------------------------------- + def test_sync_branch_on_detached_branch_same_remote(self): + """Test expect branch on detached branch with same remote --> status ok + + """ + stat = ExternalStatus() + self._repo._git_remote_verbose = self._git_remote_origin_upstream + self._repo._branch = 'feature-2' + self._repo._tag = '' + self._repo._git_branch_vv = self._git_branch_detached_branch + self._repo._check_sync_logic(stat, self.TMP_FAKE_DIR) + self.assertEqual(stat.sync_state, ExternalStatus.STATUS_OK) + # check_sync should only modify the sync_state, not clean_state + self.assertEqual(stat.clean_state, ExternalStatus.DEFAULT) + + def test_sync_branch_on_detached_branch_diff_remote(self): + """Test expect branch on detached branch, different remote --> status modified + + """ + stat = ExternalStatus() + self._repo._git_remote_verbose = self._git_remote_origin_upstream + self._repo._branch = 'feature-2' + self._repo._tag = '' + self._repo._url = '/path/to/other/repo' + self._repo._git_branch_vv = self._git_branch_detached_branch + self._repo._check_sync_logic(stat, self.TMP_FAKE_DIR) self.assertEqual(stat.sync_state, ExternalStatus.MODEL_MODIFIED) # check_sync should only modify the sync_state, not clean_state self.assertEqual(stat.clean_state, ExternalStatus.DEFAULT) + def test_sync_branch_on_detached_branch_diff_remote2(self): + """Test expect branch on detached branch, different remote --> status modified + + """ + stat = ExternalStatus() + self._repo._git_remote_verbose = self._git_remote_origin_upstream + self._repo._branch = 'feature-2' + self._repo._tag = '' + self._repo._url = '/path/to/local/repo2' + self._repo._git_branch_vv = self._git_branch_detached_branch + self._repo._check_sync_logic(stat, self.TMP_FAKE_DIR) + self.assertEqual(stat.sync_state, ExternalStatus.MODEL_MODIFIED) + # check_sync should only modify the sync_state, not clean_state + self.assertEqual(stat.clean_state, ExternalStatus.DEFAULT) + + def test_sync_branch_on_diff_branch(self): + """Test expect branch on diff branch --> status modified + + """ + stat = ExternalStatus() + self._repo._git_remote_verbose = self._git_remote_origin_upstream + self._repo._branch = 'nice_new_feature' + self._repo._tag = '' + self._repo._git_branch_vv = self._git_branch_detached_branch + self._repo._check_sync_logic(stat, self.TMP_FAKE_DIR) + self.assertEqual(stat.sync_state, ExternalStatus.MODEL_MODIFIED) + # check_sync should only modify the sync_state, not clean_state + self.assertEqual(stat.clean_state, ExternalStatus.DEFAULT) + + def test_sync_branch_on_detached_hash(self): + """Test expect branch on detached hash --> status modified + + """ + stat = ExternalStatus() + self._repo._git_remote_verbose = self._git_remote_origin_upstream + self._repo._branch = 'feature-2' + self._repo._tag = '' + self._repo._git_branch_vv = self._git_branch_detached_hash + self._repo._check_sync_logic(stat, self.TMP_FAKE_DIR) + self.assertEqual(stat.sync_state, ExternalStatus.MODEL_MODIFIED) + # check_sync should only modify the sync_state, not clean_state + self.assertEqual(stat.clean_state, ExternalStatus.DEFAULT) + + def test_sync_branch_on_detached_tag(self): + """Test expect branch on detached tag --> status modified + + """ + stat = ExternalStatus() + self._repo._git_remote_verbose = self._git_remote_origin_upstream + self._repo._branch = 'feature-2' + self._repo._tag = '' + self._repo._git_branch_vv = self._git_branch_detached_tag + self._repo._check_sync_logic(stat, self.TMP_FAKE_DIR) + self.assertEqual(stat.sync_state, ExternalStatus.MODEL_MODIFIED) + # check_sync should only modify the sync_state, not clean_state + self.assertEqual(stat.clean_state, ExternalStatus.DEFAULT) + + def test_sync_branch_on_tracking_branch_same_remote(self): + """Test expect branch on tracking branch with same remote --> status ok + + """ + stat = ExternalStatus() + self._repo._git_remote_verbose = self._git_remote_origin_upstream + self._repo._branch = 'feature-2' + self._repo._tag = '' + self._repo._git_branch_vv = self._git_branch_tracked_branch + self._repo._check_sync_logic(stat, self.TMP_FAKE_DIR) + self.assertEqual(stat.sync_state, ExternalStatus.STATUS_OK) + # check_sync should only modify the sync_state, not clean_state + self.assertEqual(stat.clean_state, ExternalStatus.DEFAULT) + + def test_sync_branch_on_tracking_branch_diff_remote(self): + """Test expect branch on tracking branch with different remote--> + status modified + + """ + stat = ExternalStatus() + self._repo._git_remote_verbose = self._git_remote_origin_upstream + self._repo._branch = 'feature-2' + self._repo._tag = '' + self._repo._url = '/path/to/other/repo' + self._repo._git_branch_vv = self._git_branch_tracked_branch + self._repo._check_sync_logic(stat, self.TMP_FAKE_DIR) + self.assertEqual(stat.sync_state, ExternalStatus.MODEL_MODIFIED) + # check_sync should only modify the sync_state, not clean_state + self.assertEqual(stat.clean_state, ExternalStatus.DEFAULT) + + def test_sync_branch_on_untracked_branch(self): + """Test expect branch on untracked branch --> status modified + + NOTE(bja, 2017-11) the externals description is always a + remote repository. A local untracked branch only exists + locally, therefore it is always a modified state, even if this + is what the user wants. + + FIME(bja, 2017-11) this brings up an additional case where the + user wants to modify other externals, but keep the current + one fixed. In that case, they would specify '.' for the + repository and their branch.... + + """ + stat = ExternalStatus() + self._repo._git_remote_verbose = self._git_remote_origin_upstream + self._repo._branch = 'feature-2' + self._repo._tag = '' + self._repo._git_branch_vv = self._git_branch_untracked_branch + self._repo._check_sync_logic(stat, self.TMP_FAKE_DIR) + self.assertEqual(stat.sync_state, ExternalStatus.MODEL_MODIFIED) + # check_sync should only modify the sync_state, not clean_state + self.assertEqual(stat.clean_state, ExternalStatus.DEFAULT) + + +class TestGitRegExp(unittest.TestCase): + """Test that the regular expressions in the GitRepository class + capture intended strings + + """ + + def setUp(self): + """Common constans + """ + self._detached_tmpl = string.Template( + '* (HEAD detached at $ref) 36418b4 Work on feature-2') + + self._tracking_tmpl = string.Template( + '* feature-2 36418b4 [$ref] Work on feature-2') + + # + # RE_DETACHED + # + def test_re_detached_alphnum(self): + """Test re correctly matches alphnumeric (basic debugging) + """ + value = 'feature2' + input_str = self._detached_tmpl.substitute(ref=value) + match = GitRepository.RE_DETACHED.search(input_str) + self.assertIsNotNone(match) + self.assertEqual(match.group(1), value) + + def test_re_detached_underscore(self): + """Test re matches with underscore + """ + value = 'feature_2' + input_str = self._detached_tmpl.substitute(ref=value) + match = GitRepository.RE_DETACHED.search(input_str) + self.assertIsNotNone(match) + self.assertEqual(match.group(1), value) + + def test_re_detached_hyphen(self): + """Test re matches - + """ + value = 'feature-2' + input_str = self._detached_tmpl.substitute(ref=value) + match = GitRepository.RE_DETACHED.search(input_str) + self.assertIsNotNone(match) + self.assertEqual(match.group(1), value) + + def test_re_detached_period(self): + """Test re matches . + """ + value = 'feature.2' + input_str = self._detached_tmpl.substitute(ref=value) + match = GitRepository.RE_DETACHED.search(input_str) + self.assertIsNotNone(match) + self.assertEqual(match.group(1), value) + + def test_re_detached_slash(self): + """Test re matches / + """ + value = 'feature/2' + input_str = self._detached_tmpl.substitute(ref=value) + match = GitRepository.RE_DETACHED.search(input_str) + self.assertIsNotNone(match) + self.assertEqual(match.group(1), value) + + # + # RE_TRACKING + # + def test_re_tracking_alphnum(self): + """Test re matches alphanumeric for basic debugging + """ + value = 'feature2' + input_str = self._tracking_tmpl.substitute(ref=value) + match = GitRepository.RE_TRACKING.search(input_str) + self.assertIsNotNone(match) + self.assertEqual(match.group(1), value) + + def test_re_tracking_underscore(self): + """Test re matches _ + """ + value = 'feature_2' + input_str = self._tracking_tmpl.substitute(ref=value) + match = GitRepository.RE_TRACKING.search(input_str) + self.assertIsNotNone(match) + self.assertEqual(match.group(1), value) + + def test_re_tracking_hyphen(self): + """Test re matches - + """ + value = 'feature-2' + input_str = self._tracking_tmpl.substitute(ref=value) + match = GitRepository.RE_TRACKING.search(input_str) + self.assertIsNotNone(match) + self.assertEqual(match.group(1), value) + + def test_re_tracking_period(self): + """Test re match . + """ + value = 'feature.2' + input_str = self._tracking_tmpl.substitute(ref=value) + match = GitRepository.RE_TRACKING.search(input_str) + self.assertIsNotNone(match) + self.assertEqual(match.group(1), value) + + def test_re_tracking_slash(self): + """Test re matches / + """ + value = 'feature/2' + input_str = self._tracking_tmpl.substitute(ref=value) + match = GitRepository.RE_TRACKING.search(input_str) + self.assertIsNotNone(match) + self.assertEqual(match.group(1), value) + class TestGitStatusPorcelain(unittest.TestCase): """Test parsing of output from git status --porcelain=v1 -z @@ -315,5 +746,75 @@ def test_remote_local_rel(self): del os.environ['TEST_VAR'] +class TestVerifyTag(unittest.TestCase): + """Test logic verifying that a tag exists and is unique + + """ + + def setUp(self): + """Setup reusable git repository object + """ + self._name = 'component' + rdata = {ExternalsDescription.PROTOCOL: 'git', + ExternalsDescription.REPO_URL: + '/path/to/local/repo', + ExternalsDescription.TAG: 'tag1', + ExternalsDescription.BRANCH: EMPTY_STR + } + + data = {self._name: + { + ExternalsDescription.REQUIRED: False, + ExternalsDescription.PATH: 'tmp', + ExternalsDescription.EXTERNALS: EMPTY_STR, + ExternalsDescription.REPO: rdata, + }, + } + + model = ExternalsDescriptionDict(data) + repo = model[self._name][ExternalsDescription.REPO] + self._repo = GitRepository('test', repo) + + @staticmethod + def _shell_true(url): + _ = url + return 0 + + @staticmethod + def _shell_false(url): + _ = url + return 1 + + def test_tag_not_tag(self): + """Verify a non-tag returns false + """ + self._repo._git_showref_tag = self._shell_false + self._repo._git_showref_branch = self._shell_false + self._repo._git_revparse_commit = self._shell_false + self._repo._tag = 'tag1' + received = self._repo._is_unique_tag(self._repo._tag) + self.assertFalse(received) + + def test_tag_indeterminant(self): + """Verify an indeterminant tag/branch returns false + """ + self._repo._git_showref_tag = self._shell_true + self._repo._git_showref_branch = self._shell_true + self._repo._git_revparse_commit = self._shell_true + self._repo._tag = 'something' + received = self._repo._is_unique_tag(self._repo._tag) + self.assertFalse(received) + + def test_tag_is_unique(self): + """Verify a unique tag match returns true + """ + self._repo._git_showref_tag = self._shell_true + self._repo._git_showref_branch = self._shell_false + self._repo._git_revparse_commit = self._shell_true + self._repo._tag = 'tag1' + received = self._repo._is_unique_tag(self._repo._tag) + self.assertTrue(received) + + if __name__ == '__main__': unittest.main()