CM-55207: improve testing and cover some edge cases in commit pre-receive

Author: gotbadger

cycode/cli/files_collector/commit_range_documents.py

@@ -104,19 +104,27 @@ def collect_commit_range_diff_documents(
     return commit_documents_to_scan
 
 
-def calculate_pre_receive_commit_range(branch_update_details: str) -> Optional[str]:
+def calculate_pre_receive_commit_range(repo_path: str, branch_update_details: str) -> Optional[str]:
     end_commit = _get_end_commit_from_branch_update_details(branch_update_details)
 
     # branch is deleted, no need to perform scan
     if end_commit == consts.EMPTY_COMMIT_SHA:
         return None
 
-    start_commit = _get_oldest_unupdated_commit_for_branch(end_commit)
+    repo = git_proxy.get_repo(repo_path)
+    start_commit = _get_oldest_unupdated_commit_for_branch(repo, end_commit)
 
     # no new commit to update found
     if not start_commit:
         return None
 
+    # If the oldest not-yet-updated commit has no parent (root commit or orphaned history),
+    # using '~1' will fail. In that case, scan from the end commit, which effectively
+    # includes the entire history reachable from it (which is exactly what we need here).
+
+    if not bool(repo.commit(start_commit).parents):
+        return f'{end_commit}'
+
     return f'{start_commit}~1...{end_commit}'
 
 
@@ -126,10 +134,10 @@ def _get_end_commit_from_branch_update_details(update_details: str) -> str:
     return end_commit
 
 
-def _get_oldest_unupdated_commit_for_branch(commit: str) -> Optional[str]:
+def _get_oldest_unupdated_commit_for_branch(repo: 'Repo', commit: str) -> Optional[str]:
     # get a list of commits by chronological order that are not in the remote repository yet
     # more info about rev-list command: https://git-scm.com/docs/git-rev-list
-    repo = git_proxy.get_repo(os.getcwd())
+
     not_updated_commits = repo.git.rev_list(commit, '--topo-order', '--reverse', '--not', '--all')
 
     commits = not_updated_commits.splitlines()

tests/cli/files_collector/test_commit_range_documents.py

@@ -12,10 +12,12 @@
 from cycode.cli.files_collector.commit_range_documents import (
     _get_default_branches_for_merge_base,
     calculate_pre_push_commit_range,
+    calculate_pre_receive_commit_range,
     get_diff_file_path,
     get_safe_head_reference_for_diff,
     parse_commit_range,
     parse_pre_push_input,
+    parse_pre_receive_input,
 )
 from cycode.cli.utils.path_utils import get_path_by_os
 
@@ -871,3 +873,174 @@ def test_single_commit_spec(self) -> None:
 
             parsed_from, parsed_to = parse_commit_range(a, temp_dir)
             assert (parsed_from, parsed_to) == (a, c)
+
+
+class TestParsePreReceiveInput:
+    """Test the parse_pre_receive_input function with various pre-receive hook input scenarios."""
+
+    def test_parse_single_update_input(self) -> None:
+        """Test parsing a single branch update input."""
+        pre_receive_input = '1111111111111111111111111111111111111111 2222222222222222222222222222222222222222 refs/heads/main'
+
+        with patch('sys.stdin', StringIO(pre_receive_input)):
+            result = parse_pre_receive_input()
+            assert result == pre_receive_input
+
+    def test_parse_multiple_update_input_returns_first_line(self) -> None:
+        """Test parsing multiple branch updates returns only the first line."""
+        pre_receive_input = """0000000000000000000000000000000000000000 aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa refs/heads/main
+bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb cccccccccccccccccccccccccccccccccccccccc refs/heads/feature"""
+
+        with patch('sys.stdin', StringIO(pre_receive_input)):
+            result = parse_pre_receive_input()
+            assert result == '0000000000000000000000000000000000000000 aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa refs/heads/main'
+
+    def test_parse_empty_input_raises_error(self) -> None:
+        """Test that empty input raises ValueError."""
+        with patch('sys.stdin', StringIO('')), pytest.raises(ValueError, match='Pre receive input was not found'):
+            parse_pre_receive_input()
+
+    def test_parse_whitespace_only_input_raises_error(self) -> None:
+        """Test that whitespace-only input raises ValueError."""
+        with patch('sys.stdin', StringIO('   \n\t  ')), pytest.raises(ValueError, match='Pre receive input was not found'):
+            parse_pre_receive_input()
+
+
+class TestCalculatePreReceiveCommitRange:
+    """Test the calculate_pre_receive_commit_range function with representative scenarios."""
+
+    def test_branch_deletion_returns_none(self) -> None:
+        """When end commit is all zeros (deletion), no scan is needed."""
+        update_details = f'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa {consts.EMPTY_COMMIT_SHA} refs/heads/feature'
+        assert calculate_pre_receive_commit_range(os.getcwd(), update_details) is None
+
+    def test_no_new_commits_returns_none(self) -> None:
+        """When there are no commits not in remote, return None."""
+        with tempfile.TemporaryDirectory() as server_dir:
+            server_repo = Repo.init(server_dir, bare=True)
+            try:
+                with tempfile.TemporaryDirectory() as work_dir:
+                    work_repo = Repo.init(work_dir, b='main')
+                    try:
+                        # Create a single commit and push it to the server as main (end commit is already on a ref)
+                        test_file = os.path.join(work_dir, 'file.txt')
+                        with open(test_file, 'w') as f:
+                            f.write('base')
+                        work_repo.index.add(['file.txt'])
+                        end_commit = work_repo.index.commit('initial')
+
+                        work_repo.create_remote('origin', server_dir)
+                        work_repo.remotes.origin.push('main:main')
+
+                        update_details = f'{"a"*40} {end_commit.hexsha} refs/heads/main'
+                        assert calculate_pre_receive_commit_range(server_dir, update_details) is None
+                    finally:
+                        work_repo.close()
+            finally:
+                server_repo.close()
+
+    def test_returns_triple_dot_range_from_oldest_unupdated(self) -> None:
+        """Returns '<oldest>~1...<end>' when there are new commits to scan."""
+        with tempfile.TemporaryDirectory() as server_dir:
+            server_repo = Repo.init(server_dir, bare=True)
+            try:
+                with tempfile.TemporaryDirectory() as work_dir:
+                    work_repo = Repo.init(work_dir, b='main')
+                    try:
+                        # Create commit A and push it to server as main (server has A on a ref)
+                        a_path = os.path.join(work_dir, 'a.txt')
+                        with open(a_path, 'w') as f:
+                            f.write('A')
+                        work_repo.index.add(['a.txt'])
+                        a_commit = work_repo.index.commit('A')
+
+                        work_repo.create_remote('origin', server_dir)
+                        work_repo.remotes.origin.push('main:main')
+
+                        # Create commits B and C locally (not yet on server ref)
+                        b_path = os.path.join(work_dir, 'b.txt')
+                        with open(b_path, 'w') as f:
+                            f.write('B')
+                        work_repo.index.add(['b.txt'])
+                        b_commit = work_repo.index.commit('B')
+
+                        c_path = os.path.join(work_dir, 'c.txt')
+                        with open(c_path, 'w') as f:
+                            f.write('C')
+                        work_repo.index.add(['c.txt'])
+                        end_commit = work_repo.index.commit('C')
+
+                        # Push the objects to a temporary ref and then delete that ref on server,
+                        # so the objects exist but are not reachable from any ref.
+                        work_repo.remotes.origin.push(f'{end_commit.hexsha}:refs/tmp/hold')
+                        Repo(server_dir).git.update_ref('-d', 'refs/tmp/hold')
+
+                        update_details = f'{"a"*40} {end_commit.hexsha} refs/heads/main'
+                        result = calculate_pre_receive_commit_range(server_dir, update_details)
+                        assert result == f'{b_commit.hexsha}~1...{end_commit.hexsha}'
+                    finally:
+                        work_repo.close()
+            finally:
+                server_repo.close()
+
+    def test_initial_oldest_commit_without_parent_returns_single_commit_range(self) -> None:
+        """If oldest commit has no parent, avoid '~1' and scan from end commit only."""
+        with tempfile.TemporaryDirectory() as server_dir:
+            server_repo = Repo.init(server_dir, bare=True)
+            try:
+                with tempfile.TemporaryDirectory() as work_dir:
+                    work_repo = Repo.init(work_dir, b='main')
+                    try:
+                        # Create a single root commit locally
+                        p = os.path.join(work_dir, 'root.txt')
+                        with open(p, 'w') as f:
+                            f.write('root')
+                        work_repo.index.add(['root.txt'])
+                        end_commit = work_repo.index.commit('root')
+
+                        work_repo.create_remote('origin', server_dir)
+                        # Push objects to a temporary ref and delete it so server has objects but no refs
+                        work_repo.remotes.origin.push(f'{end_commit.hexsha}:refs/tmp/hold')
+                        Repo(server_dir).git.update_ref('-d', 'refs/tmp/hold')
+
+                        update_details = f'{"a"*40} {end_commit.hexsha} refs/heads/main'
+                        result = calculate_pre_receive_commit_range(server_dir, update_details)
+                        assert result == end_commit.hexsha
+                    finally:
+                        work_repo.close()
+            finally:
+                server_repo.close()
+
+    def test_initial_oldest_commit_without_parent_with_two_commits_returns_single_commit_range(self) -> None:
+        """If there are two new commits and the oldest has no parent, avoid '~1' and scan from end commit only."""
+        with tempfile.TemporaryDirectory() as server_dir:
+            server_repo = Repo.init(server_dir, bare=True)
+            try:
+                with tempfile.TemporaryDirectory() as work_dir:
+                    work_repo = Repo.init(work_dir, b='main')
+                    try:
+                        # Create two commits locally: oldest has no parent, second on top
+                        a_path = os.path.join(work_dir, 'a.txt')
+                        with open(a_path, 'w') as f:
+                            f.write('A')
+                        work_repo.index.add(['a.txt'])
+                        _a_commit = work_repo.index.commit('A')
+
+                        d_path = os.path.join(work_dir, 'd.txt')
+                        with open(d_path, 'w') as f:
+                            f.write('D')
+                        work_repo.index.add(['d.txt'])
+                        end_commit = work_repo.index.commit('D')
+
+                        work_repo.create_remote('origin', server_dir)
+                        # Push objects to a temporary ref and delete it so server has objects but no refs
+                        work_repo.remotes.origin.push(f'{end_commit.hexsha}:refs/tmp/hold')
+                        Repo(server_dir).git.update_ref('-d', 'refs/tmp/hold')
+
+                        update_details = f'{consts.EMPTY_COMMIT_SHA} {end_commit.hexsha} refs/heads/main'
+                        result = calculate_pre_receive_commit_range(server_dir, update_details)
+                        assert result == end_commit.hexsha
+                    finally:
+                        work_repo.close()
+            finally:
+                server_repo.close()