diff --git a/src/forge/workspace/manager.py b/src/forge/workspace/manager.py index a9644785..9d635414 100644 --- a/src/forge/workspace/manager.py +++ b/src/forge/workspace/manager.py @@ -2,6 +2,7 @@ import logging import shutil +import subprocess import tempfile from dataclasses import dataclass from pathlib import Path @@ -102,18 +103,37 @@ def get_workspace( def destroy_workspace(self, workspace: Workspace) -> None: """Destroy a workspace and clean up files. - Args: - workspace: Workspace to destroy. + Uses podman unshare rm -rf to handle root-owned files written by + containers. Falls back to shutil.rmtree when podman is unavailable + or the unshare command fails. """ workspace_id = f"{workspace.ticket_key}:{workspace.repo_name}" - try: - if workspace.path.exists(): + if workspace.path.exists(): + podman = shutil.which("podman") + if podman: + try: + result = subprocess.run( + [podman, "unshare", "rm", "-rf", str(workspace.path)], + check=False, + ) + except OSError as exc: + logger.warning( + "podman unshare failed for %s, falling back to shutil: %s", + workspace.path, + exc, + ) + shutil.rmtree(workspace.path) + else: + if result.returncode != 0: + logger.warning( + "podman unshare failed for %s, falling back to shutil", + workspace.path, + ) + shutil.rmtree(workspace.path) + else: shutil.rmtree(workspace.path) - logger.info(f"Destroyed workspace: {workspace}") - except Exception as e: - logger.error(f"Failed to destroy workspace {workspace}: {e}") - raise + logger.info(f"Destroyed workspace: {workspace}") workspace.is_active = False if workspace_id in self._workspaces: diff --git a/tests/unit/workspace/test_manager_cleanup.py b/tests/unit/workspace/test_manager_cleanup.py new file mode 100644 index 00000000..e016e716 --- /dev/null +++ b/tests/unit/workspace/test_manager_cleanup.py @@ -0,0 +1,92 @@ +"""Tests for WorkspaceManager.destroy_workspace with podman unshare fallback.""" + +from pathlib import Path +from unittest.mock import MagicMock, patch + +from forge.workspace.manager import Workspace, WorkspaceManager + + +def _workspace(tmp_path: Path) -> Workspace: + ws = tmp_path / "ws" + ws.mkdir() + (ws / "file.txt").write_text("content") + return Workspace(path=ws, repo_name="org/repo", branch_name="forge/x", ticket_key="T-1") + + +class TestDestroyWorkspace: + def test_uses_podman_unshare_when_available(self, tmp_path): + ws = _workspace(tmp_path) + manager = WorkspaceManager() + manager._workspaces["T-1:org/repo"] = ws + + with ( + patch("forge.workspace.manager.shutil.which", return_value="/usr/bin/podman"), + patch("forge.workspace.manager.subprocess.run") as mock_run, + ): + mock_run.return_value = MagicMock(returncode=0) + manager.destroy_workspace(ws) + + mock_run.assert_called_once_with( + ["/usr/bin/podman", "unshare", "rm", "-rf", str(ws.path)], + check=False, + ) + assert not ws.is_active + + def test_falls_back_to_shutil_when_podman_unavailable(self, tmp_path): + ws = _workspace(tmp_path) + manager = WorkspaceManager() + manager._workspaces["T-1:org/repo"] = ws + + with ( + patch("forge.workspace.manager.shutil.which", return_value=None), + patch("forge.workspace.manager.shutil.rmtree") as mock_rmtree, + ): + manager.destroy_workspace(ws) + + mock_rmtree.assert_called_once_with(ws.path) + assert not ws.is_active + + def test_falls_back_to_shutil_when_podman_fails(self, tmp_path): + ws = _workspace(tmp_path) + manager = WorkspaceManager() + manager._workspaces["T-1:org/repo"] = ws + + with ( + patch("forge.workspace.manager.shutil.which", return_value="/usr/bin/podman"), + patch("forge.workspace.manager.subprocess.run") as mock_run, + patch("forge.workspace.manager.shutil.rmtree") as mock_rmtree, + ): + mock_run.return_value = MagicMock(returncode=1) + manager.destroy_workspace(ws) + + mock_rmtree.assert_called_once_with(ws.path) + + def test_falls_back_to_shutil_when_podman_cannot_start(self, tmp_path): + ws = _workspace(tmp_path) + manager = WorkspaceManager() + + with ( + patch("forge.workspace.manager.shutil.which", return_value="/usr/bin/podman"), + patch( + "forge.workspace.manager.subprocess.run", + side_effect=OSError("podman disappeared"), + ), + patch("forge.workspace.manager.shutil.rmtree") as mock_rmtree, + ): + manager.destroy_workspace(ws) + + mock_rmtree.assert_called_once_with(ws.path) + + def test_workspace_marked_inactive_and_removed_from_registry(self, tmp_path): + ws = _workspace(tmp_path) + manager = WorkspaceManager() + manager._workspaces["T-1:org/repo"] = ws + + with ( + patch("forge.workspace.manager.shutil.which", return_value=None), + patch("forge.workspace.manager.shutil.rmtree"), + ): + manager.destroy_workspace(ws) + + assert not ws.is_active + assert "T-1:org/repo" not in manager._workspaces