diff --git a/src/vcspull/cli/sync.py b/src/vcspull/cli/sync.py index 1f754887..d7094036 100644 --- a/src/vcspull/cli/sync.py +++ b/src/vcspull/cli/sync.py @@ -3,10 +3,12 @@ from __future__ import annotations import logging +import subprocess import sys import typing as t from copy import deepcopy +from libvcs import exc as libvcs_exc from libvcs._internal.shortcuts import create_project from libvcs.url import registry as url_tools @@ -23,6 +25,22 @@ log = logging.getLogger(__name__) +BRANCH_ERROR_MSG = """ +⚠️ Error syncing '{repo_name}': Branch issue detected + +The repository appears to be on a local branch with no remote tracking. + +To fix this, add a 'rev' field to your vcspull configuration: + + {repo_name}: + repo: {repo_url} + rev: main # or specify your desired branch + +Alternatively, switch to a branch that exists remotely: + cd {repo_path} + git checkout main # or another remote branch{available_branches} +""" + def clamp(n: int, _min: int, _max: int) -> int: """Clamp a number between a min and max value.""" @@ -142,6 +160,130 @@ def __init__(self, repo_url: str, *args: object, **kwargs: object) -> None: return super().__init__(f"Could not automatically determine VCS for {repo_url}") +def check_branch_state(repo_dict: dict[str, t.Any]) -> bool: + """Check if repository is in a problematic branch state. + + Returns True if there are potential branch issues that should be addressed + before attempting to sync. + + Parameters + ---------- + repo_dict : dict + Repository configuration dictionary + + Returns + ------- + bool + True if branch issues detected, False otherwise + """ + repo_path = repo_dict.get("path") + if not repo_path: + return False + + try: + # Check if we're in a git repository + git_dir = subprocess.run( + ["git", "-C", str(repo_path), "rev-parse", "--git-dir"], + capture_output=True, + text=True, + timeout=5, + check=False, + ) + if git_dir.returncode != 0: + return False # Not a git repo or git not available + + # Get current branch + current_branch = subprocess.run( + ["git", "-C", str(repo_path), "branch", "--show-current"], + capture_output=True, + text=True, + timeout=5, + check=False, + ) + if current_branch.returncode != 0: + return False # Can't determine current branch + + branch_name = current_branch.stdout.strip() + if not branch_name: + return False # Detached HEAD or other edge case + + # Check if current branch has upstream tracking + upstream_ref = f"{branch_name}@{{upstream}}" + upstream = subprocess.run( + ["git", "-C", str(repo_path), "rev-parse", "--abbrev-ref", upstream_ref], + capture_output=True, + text=True, + timeout=5, + check=False, + ) + + if upstream.returncode != 0: + # No upstream tracking - this is potentially problematic + # Even if the remote branch exists, the lack of upstream tracking + # can cause libvcs to construct invalid remote references + # Check if there's a remote branch with the same name + remote_ref = f"origin/{branch_name}" + remote_branch = subprocess.run( + ["git", "-C", str(repo_path), "rev-parse", "--verify", remote_ref], + capture_output=True, + text=True, + timeout=5, + check=False, + ) + if remote_branch.returncode != 0: + # Current branch doesn't exist on remote - this will cause issues + return True + # Branch exists on remote but no upstream tracking + # This can still cause issues with libvcs's remote name detection + return True + else: + return False # Branch state looks good + + except (subprocess.TimeoutExpired, FileNotFoundError): + return False # Git command failed or not available + + +def handle_branch_error(repo_dict: dict[str, t.Any]) -> None: + """Handle branch-related errors by showing helpful instructions. + + Parameters + ---------- + repo_dict : dict + Repository configuration dictionary + """ + repo_name = repo_dict.get("name", "repository") + repo_url = repo_dict.get("url", "YOUR_REPO_URL") + repo_path = repo_dict.get("path", "REPO_PATH") + + # Try to get available remote branches + available_branches = "" + try: + cmd = ["git", "-C", str(repo_path), "branch", "-r"] + result = subprocess.run( + cmd, capture_output=True, text=True, timeout=5, check=False + ) + if result.returncode == 0 and result.stdout: + branches = [ + line.strip().replace("origin/", "") + for line in result.stdout.splitlines() + if "origin/" in line and "->" not in line + ][:5] # Show first 5 branches + if branches: + branch_list = ", ".join(branches) + available_branches = f"\n\nAvailable remote branches: {branch_list}" + except (subprocess.TimeoutExpired, FileNotFoundError): + pass # Git command failed or not available + + error_msg = BRANCH_ERROR_MSG.format( + repo_name=repo_name, + repo_url=repo_url, + repo_path=repo_path, + available_branches=available_branches, + ) + + print(error_msg, file=sys.stderr) + + def update_repo( repo_dict: t.Any, # repo_dict: Dict[str, Union[str, Dict[str, GitRemote], pathlib.Path]] @@ -162,7 +304,21 @@ def update_repo( repo_dict["vcs"] = vcs r = create_project(**repo_dict) # Creates the repo object - r.update_repo(set_remotes=True) # Creates repo if not exists and fetches + + # Check for branch issues before attempting to sync + if check_branch_state(repo_dict): + handle_branch_error(repo_dict) + # Still attempt the sync but user has been warned + + try: + r.update_repo(set_remotes=True) # Creates repo if not exists and fetches + except libvcs_exc.CommandError as e: + error_msg = str(e).lower() + # Check for branch-related errors (fallback in case pre-check missed something) + if "invalid upstream" in error_msg or "ambiguous argument" in error_msg: + handle_branch_error(repo_dict) + raise + raise # TODO: Fix this return r # type:ignore diff --git a/tests/test_sync.py b/tests/test_sync.py index e7a379ed..e99af84c 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -2,15 +2,18 @@ from __future__ import annotations +import subprocess import textwrap import typing as t +from unittest.mock import patch import pytest +from libvcs import exc as libvcs_exc from libvcs._internal.shortcuts import create_project from libvcs.sync.git import GitRemote, GitSync from vcspull._internal.config_reader import ConfigReader -from vcspull.cli.sync import update_repo +from vcspull.cli.sync import check_branch_state, handle_branch_error, update_repo from vcspull.config import extract_repos, filter_repos, load_configs from vcspull.validator import is_valid_config @@ -314,3 +317,226 @@ def test_updating_remote( expected_config["remotes"]["origin"].fetch_url.replace("git+", "") == current_remote_url ) + + +def test_check_branch_state(tmp_path: pathlib.Path) -> None: + """Test that check_branch_state detects problematic branches.""" + repo_dict = { + "name": "test-repo", + "url": "git@github.com:user/repo.git", + "path": tmp_path / "test-repo", + } + + # Create a mock git directory for testing + (tmp_path / "test-repo" / ".git").mkdir(parents=True, exist_ok=True) + + # Mock successful git commands that indicate problematic state + with patch("subprocess.run") as mock_run: + + def git_side_effect(cmd: list[str], **kwargs: t.Any) -> t.Any: + from unittest.mock import MagicMock + + mock_result = MagicMock() + + if "--git-dir" in cmd: + # Git directory check - success + mock_result.returncode = 0 + return mock_result + elif "--show-current" in cmd: + # Current branch - return 'docs' + mock_result.returncode = 0 + mock_result.stdout = "docs\n" + return mock_result + elif "@{upstream}" in " ".join(cmd): + # No upstream tracking + mock_result.returncode = 1 # No upstream + return mock_result + elif "origin/docs" in cmd: + # No origin/docs branch + mock_result.returncode = 1 # Branch doesn't exist on remote + return mock_result + else: + mock_result.returncode = 0 + return mock_result + + mock_run.side_effect = git_side_effect + + # Should detect branch issue + assert check_branch_state(repo_dict) is True + + +def test_check_branch_state_good_branch(tmp_path: pathlib.Path) -> None: + """Test that check_branch_state doesn't flag good branches.""" + repo_dict = { + "name": "test-repo", + "url": "git@github.com:user/repo.git", + "path": tmp_path / "test-repo", + } + + # Create a mock git directory for testing + (tmp_path / "test-repo" / ".git").mkdir(parents=True, exist_ok=True) + + # Mock git commands that indicate good state + with patch("subprocess.run") as mock_run: + + def git_side_effect(cmd: list[str], **kwargs: t.Any) -> t.Any: + from unittest.mock import MagicMock + + mock_result = MagicMock() + + if "--git-dir" in cmd: + # Git directory check - success + mock_result.returncode = 0 + return mock_result + elif "--show-current" in cmd: + # Current branch - return 'main' + mock_result.returncode = 0 + mock_result.stdout = "main\n" + return mock_result + elif "@{upstream}" in " ".join(cmd): + # Has upstream tracking + mock_result.returncode = 0 # Has upstream + mock_result.stdout = "origin/main\n" + return mock_result + else: + mock_result.returncode = 0 + return mock_result + + mock_run.side_effect = git_side_effect + + # Should NOT detect branch issue + assert check_branch_state(repo_dict) is False + + +def test_handle_branch_error( + tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str] +) -> None: + """Test that handle_branch_error shows helpful error message.""" + repo_dict = { + "name": "test-repo", + "url": "git@github.com:user/repo.git", + "path": tmp_path / "test-repo", + } + + # Create a mock git directory for testing + (tmp_path / "test-repo" / ".git").mkdir(parents=True, exist_ok=True) + + # Mock subprocess.run to simulate git branch output + with patch("subprocess.run") as mock_run: + mock_run.return_value.returncode = 0 + mock_run.return_value.stdout = """ origin/HEAD -> origin/main + origin/main + origin/develop + origin/feature/test + origin/v1.0.0 + origin/v2.0.0 + origin/docs""" + + handle_branch_error(repo_dict) + + captured = capsys.readouterr() + + # Check that the error message contains expected elements + assert "Error syncing 'test-repo': Branch issue detected" in captured.err + assert "add a 'rev' field to your vcspull configuration" in captured.err + assert "test-repo:" in captured.err + assert "repo: git@github.com:user/repo.git" in captured.err + assert f"cd {tmp_path / 'test-repo'}" in captured.err + expected_branches = "main, develop, feature/test, v1.0.0, v2.0.0" + assert f"Available remote branches: {expected_branches}" in captured.err + + +def test_handle_branch_error_no_git( + tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str] +) -> None: + """Test handle_branch_error when git command fails.""" + repo_dict = { + "name": "test-repo", + "url": "git@github.com:user/repo.git", + "path": tmp_path / "test-repo", + } + + # Mock subprocess.run to simulate git command failure + with patch("subprocess.run") as mock_run: + mock_run.side_effect = FileNotFoundError("git not found") + + handle_branch_error(repo_dict) + + captured = capsys.readouterr() + + # Check that the error message still works without branches list + assert "Error syncing 'test-repo': Branch issue detected" in captured.err + assert "Available remote branches" not in captured.err + + +def test_update_repo_branch_error_precheck( + tmp_path: pathlib.Path, + git_remote_repo: pathlib.Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Test that update_repo detects branch errors with pre-validation.""" + repo_config = { + "name": "test-repo", + "url": f"git+file://{git_remote_repo}", + "path": str(tmp_path / "test-repo"), + "vcs": "git", + } + + # First, create the repository normally + update_repo(repo_config) + + # Create and checkout a local branch with no remote + repo_path = tmp_path / "test-repo" + subprocess.run( + ["git", "-C", str(repo_path), "checkout", "-b", "local-only"], check=True + ) + + # Mock check_branch_state to return True (problematic branch detected) + with patch("vcspull.cli.sync.check_branch_state", return_value=True): + # This should trigger the pre-validation warning + update_repo(repo_config) + + captured = capsys.readouterr() + + # Verify error handler was called during pre-validation + assert "Error syncing 'test-repo': Branch issue detected" in captured.err + assert "add a 'rev' field to your vcspull configuration" in captured.err + + +def test_update_repo_branch_error_fallback( + tmp_path: pathlib.Path, + git_remote_repo: pathlib.Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Test that update_repo still catches CommandError as fallback.""" + repo_config = { + "name": "test-repo", + "url": f"git+file://{git_remote_repo}", + "path": str(tmp_path / "test-repo"), + "vcs": "git", + } + + # First, create the repository normally + update_repo(repo_config) + + # Mock check_branch_state to return False (no issue detected in pre-check) + # But mock update_repo to raise CommandError (fallback case) + with ( + patch("vcspull.cli.sync.check_branch_state", return_value=False), + patch.object(GitSync, "update_repo") as mock_update, + ): + mock_update.side_effect = libvcs_exc.CommandError( + output="fatal: invalid upstream 'origin/local-only'", + returncode=1, + cmd="git rebase origin/local-only", + ) + + # Try to update, should trigger fallback error handling + with pytest.raises(libvcs_exc.CommandError): + update_repo(repo_config) + + captured = capsys.readouterr() + + # Verify error handler was called as fallback + assert "Error syncing 'test-repo': Branch issue detected" in captured.err + assert "add a 'rev' field to your vcspull configuration" in captured.err