From 24ca1c2898043ab36b2732a68e398a21d92f9ddb Mon Sep 17 00:00:00 2001 From: meeks Date: Thu, 16 Jul 2026 12:46:27 +0200 Subject: [PATCH] Refactor WorkspaceManager.sanitize_repo to stash changes and raise errors on failure, add unit tests --- gitea/workspace.py | 16 +++++- tests/test_workspace.py | 109 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 124 insertions(+), 1 deletion(-) diff --git a/gitea/workspace.py b/gitea/workspace.py index 1699ee9..b729899 100644 --- a/gitea/workspace.py +++ b/gitea/workspace.py @@ -67,6 +67,19 @@ class WorkspaceManager: check=True, capture_output=True, ) self._configure_repo_user(repo_path) + + # Check for any uncommitted changes or untracked files + status_res = subprocess.run( + ["git", "-C", str(repo_path), "status", "--porcelain"], + check=True, capture_output=True, text=True + ) + if status_res.stdout.strip(): + logger.info(f"Uncommitted changes detected in {repo_path}. Stashing before sanitization.") + subprocess.run( + ["git", "-C", str(repo_path), "stash", "push", "-u", "-m", "Auto-backup before agent sanitization"], + check=True, capture_output=True + ) + subprocess.run( ["git", "-C", str(repo_path), "reset", "--hard", "HEAD"], check=True, capture_output=True, @@ -96,7 +109,8 @@ class WorkspaceManager: check=True, capture_output=True, ) except Exception as e: - logger.error(f"Error during sanitization: {e}") + logger.error(f"Error during sanitization: {e}", exc_info=True) + raise RuntimeError(f"Failed to sanitize repository {repo_full_name} at {repo_path}: {e}") from e def clone_repo(self, repo_full_name: str, clone_url: str | None = None) -> Path: repo_path: Path = self.get_repo_path(repo_full_name) diff --git a/tests/test_workspace.py b/tests/test_workspace.py index c216aec..8e5c707 100644 --- a/tests/test_workspace.py +++ b/tests/test_workspace.py @@ -1,4 +1,5 @@ from pathlib import Path +from typing import Any from unittest.mock import MagicMock, patch import pytest @@ -95,3 +96,111 @@ def test_workspace_manager_fails_if_authenticated_user_has_no_login( with pytest.raises(RuntimeError, match="No authenticated user found."): workspace._configure_repo_user(Path("/tmp/mock-repo")) + + +@patch("gitea.workspace.subprocess.run") +@patch("gitea.workspace.GiteaClient") +def test_workspace_manager_sanitize_repo_no_changes( + mock_client_class: MagicMock, + mock_run: MagicMock +) -> None: + # Setup Gitea client mock + mock_client = MagicMock() + mock_client_class.return_value = mock_client + mock_user = MagicMock() + mock_user.full_name = "Agent Tester" + mock_user.login = "agent-test" + mock_user.email = "agent-test@example.com" + mock_client.get_authenticated_user.return_value = mock_user + + # Mock subprocess.run for status check and others + def mock_run_side_effect(args: list[str], **kwargs: Any) -> MagicMock: + result = MagicMock() + result.returncode = 0 + if "status" in args: + result.stdout = "" + else: + result.stdout = "some output" + return result + + mock_run.side_effect = mock_run_side_effect + + workspace = WorkspaceManager() + repo_path = Path("/tmp/mock-repo") + + workspace.sanitize_repo("meeks/repo1", repo_path) + + # Verify that stash was not called + stash_calls = [call for call in mock_run.call_args_list if "stash" in call[0][0]] + assert len(stash_calls) == 0 + + # Verify other expected git calls + reset_calls = [call for call in mock_run.call_args_list if "reset" in call[0][0]] + clean_calls = [call for call in mock_run.call_args_list if "clean" in call[0][0]] + assert len(reset_calls) > 0 + assert len(clean_calls) > 0 + + +@patch("gitea.workspace.subprocess.run") +@patch("gitea.workspace.GiteaClient") +def test_workspace_manager_sanitize_repo_with_changes( + mock_client_class: MagicMock, + mock_run: MagicMock +) -> None: + # Setup Gitea client mock + mock_client = MagicMock() + mock_client_class.return_value = mock_client + mock_user = MagicMock() + mock_user.full_name = "Agent Tester" + mock_user.login = "agent-test" + mock_user.email = "agent-test@example.com" + mock_client.get_authenticated_user.return_value = mock_user + + # Mock subprocess.run to show modified files + def mock_run_side_effect(args: list[str], **kwargs: Any) -> MagicMock: + result = MagicMock() + result.returncode = 0 + if "status" in args: + result.stdout = " M file.py\n?? untracked.py\n" + else: + result.stdout = "" + return result + + mock_run.side_effect = mock_run_side_effect + + workspace = WorkspaceManager() + repo_path = Path("/tmp/mock-repo") + + workspace.sanitize_repo("meeks/repo1", repo_path) + + # Verify stash push was called + stash_calls = [call for call in mock_run.call_args_list if "stash" in call[0][0]] + assert len(stash_calls) == 1 + assert "push" in stash_calls[0][0][0] + assert "-u" in stash_calls[0][0][0] + + +@patch("gitea.workspace.subprocess.run") +@patch("gitea.workspace.GiteaClient") +def test_workspace_manager_sanitize_repo_fails( + mock_client_class: MagicMock, + mock_run: MagicMock +) -> None: + # Setup Gitea client mock + mock_client = MagicMock() + mock_client_class.return_value = mock_client + mock_user = MagicMock() + mock_user.full_name = "Agent Tester" + mock_user.login = "agent-test" + mock_user.email = "agent-test@example.com" + mock_client.get_authenticated_user.return_value = mock_user + + # Mock remote set-url to fail + import subprocess + mock_run.side_effect = subprocess.CalledProcessError(1, "git remote set-url") + + workspace = WorkspaceManager() + repo_path = Path("/tmp/mock-repo") + + with pytest.raises(RuntimeError, match="Failed to sanitize repository"): + workspace.sanitize_repo("meeks/repo1", repo_path)