Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 28 additions & 8 deletions src/forge/workspace/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import logging
import shutil
import subprocess
import tempfile
from dataclasses import dataclass
from pathlib import Path
Expand Down Expand Up @@ -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:
Expand Down
92 changes: 92 additions & 0 deletions tests/unit/workspace/test_manager_cleanup.py
Original file line number Diff line number Diff line change
@@ -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
Loading