This commit is contained in:
@@ -0,0 +1 @@
|
||||
"""Tests for gsd-openclaw."""
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Pytest fixtures for gsd-openclaw tests."""
|
||||
import shutil
|
||||
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
|
||||
from gsd_openclaw.workspace import Workspace, WorkspaceManager
|
||||
from gsd_openclaw.state_manager import SessionState
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tmp_workspaces_root(tmp_path: Path) -> Path:
|
||||
"""Temporary root for workspace testing.
|
||||
|
||||
This ensures tests don't pollute the real ~/gsd-workspaces directory.
|
||||
|
||||
Args:
|
||||
tmp_path: Pytest's built-in temporary directory fixture.
|
||||
|
||||
Returns:
|
||||
Path to a temporary gsd-workspaces directory.
|
||||
"""
|
||||
return tmp_path / "gsd-workspaces"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def workspace_manager(tmp_workspaces_root: Path) -> WorkspaceManager:
|
||||
"""WorkspaceManager with temporary root for isolation.
|
||||
|
||||
Creates a WorkspaceManager that uses a temporary directory instead
|
||||
of the user's home directory. Automatically cleans up after the test.
|
||||
|
||||
Args:
|
||||
tmp_workspaces_root: Fixture providing temporary workspace root.
|
||||
|
||||
Yields:
|
||||
WorkspaceManager instance configured for testing.
|
||||
"""
|
||||
manager = WorkspaceManager("test-project", tmp_workspaces_root)
|
||||
yield manager
|
||||
# Cleanup
|
||||
if manager.workspace.root.exists():
|
||||
shutil.rmtree(manager.workspace.root)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def created_workspace(workspace_manager: WorkspaceManager) -> Workspace:
|
||||
"""A workspace that has been created (directories exist).
|
||||
|
||||
Args:
|
||||
workspace_manager: Fixture providing WorkspaceManager.
|
||||
|
||||
Returns:
|
||||
Workspace object with all directories created.
|
||||
"""
|
||||
return workspace_manager.create()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tmp_state_dir(tmp_path: Path) -> Path:
|
||||
"""Temporary directory for state file testing.
|
||||
|
||||
Args:
|
||||
tmp_path: Pytest's built-in temporary directory fixture.
|
||||
|
||||
Returns:
|
||||
Path to a temporary directory for state files.
|
||||
"""
|
||||
state_dir = tmp_path / ".gsd"
|
||||
state_dir.mkdir(parents=True, exist_ok=True)
|
||||
return state_dir
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_state() -> SessionState:
|
||||
"""A valid SessionState instance for testing.
|
||||
|
||||
Returns:
|
||||
SessionState with all fields populated.
|
||||
"""
|
||||
return SessionState(
|
||||
schema_version="1.0",
|
||||
session_id="test-session-fixture",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test-project",
|
||||
tmux_pane_id="%100",
|
||||
created_at="2026-03-10T10:00:00Z",
|
||||
last_active_at="2026-03-10T12:00:00Z",
|
||||
command_count=10,
|
||||
last_output_snapshot="Test output snapshot",
|
||||
display_name="Test Project Display Name"
|
||||
)
|
||||
@@ -0,0 +1,310 @@
|
||||
"""Tests for CLI entry point."""
|
||||
import pytest
|
||||
import sys
|
||||
from unittest.mock import MagicMock, patch
|
||||
from io import StringIO
|
||||
|
||||
from gsd_openclaw.tmux_session import TmuxSession, TmuxSessionError, CleanupResult, SessionInfo
|
||||
from gsd_openclaw.cli import main, cmd_create, cmd_list, cmd_destroy, cmd_resume, cmd_cleanup
|
||||
|
||||
|
||||
class TestCliCreate:
|
||||
"""Tests for CLI create command."""
|
||||
|
||||
def test_cli_create_creates_session(self):
|
||||
"""`gsd-ocl create <project>` creates session."""
|
||||
mock_session = MagicMock()
|
||||
mock_session.tmux_session_name = "gsd-test-project"
|
||||
mock_session.session_id = "test-123"
|
||||
mock_session.project_name = "test-project"
|
||||
|
||||
with patch('gsd_openclaw.cli.TmuxSessionManager') as MockManager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.create_session.return_value = mock_session
|
||||
MockManager.return_value = mock_manager
|
||||
|
||||
with patch.object(sys, 'argv', ['gsd-ocl', 'create', 'test-project']):
|
||||
main()
|
||||
|
||||
mock_manager.create_session.assert_called_once_with(
|
||||
project_name='test-project',
|
||||
display_name=None
|
||||
)
|
||||
|
||||
def test_cli_create_with_display_name(self):
|
||||
"""`gsd-ocl create <project> --display-name` sets display name."""
|
||||
mock_session = MagicMock()
|
||||
mock_session.tmux_session_name = "gsd-test-project"
|
||||
mock_session.session_id = "test-123"
|
||||
mock_session.project_name = "test-project"
|
||||
|
||||
with patch('gsd_openclaw.cli.TmuxSessionManager') as MockManager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.create_session.return_value = mock_session
|
||||
MockManager.return_value = mock_manager
|
||||
|
||||
with patch.object(sys, 'argv', ['gsd-ocl', 'create', 'test-project', '-d', 'My Project']):
|
||||
main()
|
||||
|
||||
mock_manager.create_session.assert_called_once_with(
|
||||
project_name='test-project',
|
||||
display_name='My Project'
|
||||
)
|
||||
|
||||
def test_cli_create_exits_on_error(self):
|
||||
"""CLI exits with code 1 on error."""
|
||||
with patch('gsd_openclaw.cli.TmuxSessionManager') as MockManager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.create_session.side_effect = Exception("Create failed")
|
||||
MockManager.return_value = mock_manager
|
||||
|
||||
with patch.object(sys, 'argv', ['gsd-ocl', 'create', 'test-project']):
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
main()
|
||||
|
||||
assert exc_info.value.code == 1
|
||||
|
||||
|
||||
class TestCliList:
|
||||
"""Tests for CLI list command."""
|
||||
|
||||
def test_cli_list_shows_sessions(self):
|
||||
"""`gsd-ocl list` shows sessions."""
|
||||
mock_session_info = SessionInfo(
|
||||
tmux_name="gsd-test-project",
|
||||
project_name="test-project",
|
||||
session_id="test-123",
|
||||
created_at="2026-03-10T10:00:00",
|
||||
last_active_at="2026-03-10T12:00:00",
|
||||
command_count=5,
|
||||
display_name="Test Project",
|
||||
is_orphan=False
|
||||
)
|
||||
|
||||
with patch('gsd_openclaw.cli.TmuxSessionManager') as MockManager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.list_gsd_sessions.return_value = [mock_session_info]
|
||||
MockManager.return_value = mock_manager
|
||||
|
||||
stdout = StringIO()
|
||||
with patch.object(sys, 'argv', ['gsd-ocl', 'list']):
|
||||
with patch('sys.stdout', stdout):
|
||||
main()
|
||||
|
||||
output = stdout.getvalue()
|
||||
assert "test-project" in output or "Test Project" in output
|
||||
assert "test-123" in output
|
||||
assert "active" in output
|
||||
|
||||
def test_cli_list_shows_no_sessions_message(self):
|
||||
"""`gsd-ocl list` shows message when no sessions."""
|
||||
with patch('gsd_openclaw.cli.TmuxSessionManager') as MockManager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.list_gsd_sessions.return_value = []
|
||||
MockManager.return_value = mock_manager
|
||||
|
||||
stdout = StringIO()
|
||||
with patch.object(sys, 'argv', ['gsd-ocl', 'list']):
|
||||
with patch('sys.stdout', stdout):
|
||||
main()
|
||||
|
||||
output = stdout.getvalue()
|
||||
assert "No active sessions" in output
|
||||
|
||||
def test_cli_list_shows_orphan_status(self):
|
||||
"""`gsd-ocl list` shows ORPHAN status for orphan sessions."""
|
||||
mock_session_info = SessionInfo(
|
||||
tmux_name="gsd-orphan",
|
||||
project_name="orphan",
|
||||
session_id="?",
|
||||
created_at="?",
|
||||
last_active_at="?",
|
||||
command_count=0,
|
||||
display_name=None,
|
||||
is_orphan=True
|
||||
)
|
||||
|
||||
with patch('gsd_openclaw.cli.TmuxSessionManager') as MockManager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.list_gsd_sessions.return_value = [mock_session_info]
|
||||
MockManager.return_value = mock_manager
|
||||
|
||||
stdout = StringIO()
|
||||
with patch.object(sys, 'argv', ['gsd-ocl', 'list']):
|
||||
with patch('sys.stdout', stdout):
|
||||
main()
|
||||
|
||||
output = stdout.getvalue()
|
||||
assert "ORPHAN" in output
|
||||
|
||||
|
||||
class TestCliDestroy:
|
||||
"""Tests for CLI destroy command."""
|
||||
|
||||
def test_cli_destroy_removes_session(self):
|
||||
"""`gsd-ocl destroy <project>` removes session."""
|
||||
with patch('gsd_openclaw.cli.TmuxSessionManager') as MockManager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.destroy_session.return_value = True
|
||||
MockManager.return_value = mock_manager
|
||||
|
||||
with patch.object(sys, 'argv', ['gsd-ocl', 'destroy', 'test-project']):
|
||||
main()
|
||||
|
||||
mock_manager.destroy_session.assert_called_once_with(
|
||||
project_name='test-project',
|
||||
cleanup_workspace=False
|
||||
)
|
||||
|
||||
def test_cli_destroy_with_cleanup(self):
|
||||
"""`gsd-ocl destroy <project> --cleanup` removes workspace too."""
|
||||
with patch('gsd_openclaw.cli.TmuxSessionManager') as MockManager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.destroy_session.return_value = True
|
||||
MockManager.return_value = mock_manager
|
||||
|
||||
with patch.object(sys, 'argv', ['gsd-ocl', 'destroy', 'test-project', '-c']):
|
||||
main()
|
||||
|
||||
mock_manager.destroy_session.assert_called_once_with(
|
||||
project_name='test-project',
|
||||
cleanup_workspace=True
|
||||
)
|
||||
|
||||
def test_cli_destroy_exits_if_not_found(self):
|
||||
"""`gsd-ocl destroy` exits with code 1 if session not found."""
|
||||
with patch('gsd_openclaw.cli.TmuxSessionManager') as MockManager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.destroy_session.return_value = False
|
||||
MockManager.return_value = mock_manager
|
||||
|
||||
with patch.object(sys, 'argv', ['gsd-ocl', 'destroy', 'nonexistent']):
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
main()
|
||||
|
||||
assert exc_info.value.code == 1
|
||||
|
||||
|
||||
class TestCliResume:
|
||||
"""Tests for CLI resume command."""
|
||||
|
||||
def test_cli_resume_resumes_session(self):
|
||||
"""`gsd-ocl resume <project>` resumes session."""
|
||||
mock_session = MagicMock()
|
||||
mock_session.tmux_session_name = "gsd-test-project"
|
||||
mock_session.session_id = "test-123"
|
||||
mock_session.project_name = "test-project"
|
||||
|
||||
with patch('gsd_openclaw.cli.TmuxSessionManager') as MockManager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.resume_session.return_value = mock_session
|
||||
MockManager.return_value = mock_manager
|
||||
|
||||
with patch.object(sys, 'argv', ['gsd-ocl', 'resume', 'test-project']):
|
||||
main()
|
||||
|
||||
mock_manager.resume_session.assert_called_once_with(project_name='test-project')
|
||||
|
||||
def test_cli_resume_exits_on_error(self):
|
||||
"""`gsd-ocl resume` exits with code 1 on error."""
|
||||
with patch('gsd_openclaw.cli.TmuxSessionManager') as MockManager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.resume_session.side_effect = TmuxSessionError("No session found")
|
||||
MockManager.return_value = mock_manager
|
||||
|
||||
with patch.object(sys, 'argv', ['gsd-ocl', 'resume', 'nonexistent']):
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
main()
|
||||
|
||||
assert exc_info.value.code == 1
|
||||
|
||||
|
||||
class TestCliCleanup:
|
||||
"""Tests for CLI cleanup command."""
|
||||
|
||||
def test_cli_cleanup_removes_orphans(self):
|
||||
"""`gsd-ocl cleanup` removes orphan sessions."""
|
||||
result = CleanupResult(
|
||||
cleaned_sessions=["gsd-orphan1", "gsd-orphan2"],
|
||||
errors=[]
|
||||
)
|
||||
|
||||
with patch('gsd_openclaw.cli.TmuxSessionManager') as MockManager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.cleanup_orphans.return_value = result
|
||||
MockManager.return_value = mock_manager
|
||||
|
||||
stdout = StringIO()
|
||||
with patch.object(sys, 'argv', ['gsd-ocl', 'cleanup']):
|
||||
with patch('sys.stdout', stdout):
|
||||
main()
|
||||
|
||||
output = stdout.getvalue()
|
||||
assert "2" in output
|
||||
assert "gsd-orphan1" in output
|
||||
assert "gsd-orphan2" in output
|
||||
|
||||
def test_cli_cleanup_shows_errors(self):
|
||||
"""`gsd-ocl cleanup` shows errors."""
|
||||
result = CleanupResult(
|
||||
cleaned_sessions=[],
|
||||
errors=["Failed to kill gsd-failing: error"]
|
||||
)
|
||||
|
||||
with patch('gsd_openclaw.cli.TmuxSessionManager') as MockManager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.cleanup_orphans.return_value = result
|
||||
MockManager.return_value = mock_manager
|
||||
|
||||
stdout = StringIO()
|
||||
with patch.object(sys, 'argv', ['gsd-ocl', 'cleanup']):
|
||||
with patch('sys.stdout', stdout):
|
||||
main()
|
||||
|
||||
output = stdout.getvalue()
|
||||
assert "error" in output.lower() or "Error" in output
|
||||
|
||||
def test_cli_cleanup_no_orphans_found(self):
|
||||
"""`gsd-ocl cleanup` shows message when no orphans found."""
|
||||
result = CleanupResult(
|
||||
cleaned_sessions=[],
|
||||
errors=[]
|
||||
)
|
||||
|
||||
with patch('gsd_openclaw.cli.TmuxSessionManager') as MockManager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.cleanup_orphans.return_value = result
|
||||
MockManager.return_value = mock_manager
|
||||
|
||||
stdout = StringIO()
|
||||
with patch.object(sys, 'argv', ['gsd-ocl', 'cleanup']):
|
||||
with patch('sys.stdout', stdout):
|
||||
main()
|
||||
|
||||
output = stdout.getvalue()
|
||||
assert "No orphan sessions found" in output
|
||||
|
||||
|
||||
class TestCliHelp:
|
||||
"""Tests for CLI help."""
|
||||
|
||||
def test_cli_help_works(self):
|
||||
"""`gsd-ocl --help` shows help message."""
|
||||
stdout = StringIO()
|
||||
with patch.object(sys, 'argv', ['gsd-ocl', '--help']):
|
||||
with patch('sys.stdout', stdout):
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
main()
|
||||
|
||||
# --help exits with code 0
|
||||
assert exc_info.value.code == 0
|
||||
output = stdout.getvalue()
|
||||
assert "gsd-ocl" in output
|
||||
|
||||
def test_cli_requires_subcommand(self):
|
||||
"""CLI requires a subcommand."""
|
||||
with patch.object(sys, 'argv', ['gsd-ocl']):
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
main()
|
||||
|
||||
assert exc_info.value.code != 0
|
||||
@@ -0,0 +1,508 @@
|
||||
"""
|
||||
Tests for state_manager module - SessionState dataclass.
|
||||
Task 1: Test state schema with dataclasses.
|
||||
Task 2: Test atomic state save with backup rotation.
|
||||
Task 3: Test state load with validation and fallback.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
from dataclasses import asdict
|
||||
from gsd_openclaw.state_manager import SessionState, save_state, load_state, StateManager
|
||||
|
||||
|
||||
class TestSessionStateCreation:
|
||||
"""Test 1: SessionState can be created with all fields."""
|
||||
|
||||
def test_create_with_defaults(self):
|
||||
"""SessionState can be created with only required fields."""
|
||||
state = SessionState(
|
||||
session_id="test-session-123",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test-project",
|
||||
tmux_pane_id="%0"
|
||||
)
|
||||
assert state.session_id == "test-session-123"
|
||||
assert state.project_name == "test-project"
|
||||
assert state.tmux_session_name == "gsd-test-project"
|
||||
assert state.tmux_pane_id == "%0"
|
||||
assert state.schema_version == "1.0"
|
||||
assert state.command_count == 0
|
||||
|
||||
def test_create_with_all_fields(self):
|
||||
"""SessionState can be created with all fields specified."""
|
||||
state = SessionState(
|
||||
schema_version="1.0",
|
||||
session_id="test-session-456",
|
||||
project_name="my-project",
|
||||
tmux_session_name="gsd-my-project",
|
||||
tmux_pane_id="%1",
|
||||
created_at="2026-03-10T10:00:00Z",
|
||||
last_active_at="2026-03-10T12:30:00Z",
|
||||
command_count=42,
|
||||
last_output_snapshot="Last output here...",
|
||||
display_name="My Cool Project"
|
||||
)
|
||||
assert state.session_id == "test-session-456"
|
||||
assert state.created_at == "2026-03-10T10:00:00Z"
|
||||
assert state.command_count == 42
|
||||
assert state.last_output_snapshot == "Last output here..."
|
||||
assert state.display_name == "My Cool Project"
|
||||
|
||||
|
||||
class TestSessionStateSerialization:
|
||||
"""Test 2 & 3 & 4: SessionState serialization methods."""
|
||||
|
||||
def test_to_dict_returns_complete_dict(self):
|
||||
"""SessionState.to_dict() returns complete dict."""
|
||||
state = SessionState(
|
||||
session_id="test-session-789",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test-project",
|
||||
tmux_pane_id="%2",
|
||||
created_at="2026-03-10T10:00:00Z",
|
||||
command_count=10
|
||||
)
|
||||
result = state.to_dict()
|
||||
assert isinstance(result, dict)
|
||||
assert result["session_id"] == "test-session-789"
|
||||
assert result["project_name"] == "test-project"
|
||||
assert result["tmux_session_name"] == "gsd-test-project"
|
||||
assert result["tmux_pane_id"] == "%2"
|
||||
assert result["created_at"] == "2026-03-10T10:00:00Z"
|
||||
assert result["command_count"] == 10
|
||||
assert result["schema_version"] == "1.0"
|
||||
|
||||
def test_from_dict_reconstructs_from_dict(self):
|
||||
"""SessionState.from_dict() reconstructs from dict."""
|
||||
data = {
|
||||
"schema_version": "1.0",
|
||||
"session_id": "test-session-abc",
|
||||
"project_name": "reconstructed-project",
|
||||
"tmux_session_name": "gsd-reconstructed",
|
||||
"tmux_pane_id": "%3",
|
||||
"created_at": "2026-03-10T08:00:00Z",
|
||||
"last_active_at": "2026-03-10T09:00:00Z",
|
||||
"command_count": 25,
|
||||
"last_output_snapshot": "Some output",
|
||||
"display_name": "Reconstructed Project"
|
||||
}
|
||||
state = SessionState.from_dict(data)
|
||||
assert state.session_id == "test-session-abc"
|
||||
assert state.project_name == "reconstructed-project"
|
||||
assert state.tmux_session_name == "gsd-reconstructed"
|
||||
assert state.tmux_pane_id == "%3"
|
||||
assert state.created_at == "2026-03-10T08:00:00Z"
|
||||
assert state.command_count == 25
|
||||
assert state.last_output_snapshot == "Some output"
|
||||
assert state.display_name == "Reconstructed Project"
|
||||
|
||||
def test_from_dict_ignores_unknown_fields(self):
|
||||
"""SessionState.from_dict() ignores unknown fields in from_dict()."""
|
||||
data = {
|
||||
"schema_version": "1.0",
|
||||
"session_id": "test-session-xyz",
|
||||
"project_name": "test-project",
|
||||
"tmux_session_name": "gsd-test",
|
||||
"tmux_pane_id": "%4",
|
||||
"unknown_field": "should be ignored",
|
||||
"another_unknown": 12345
|
||||
}
|
||||
# Should not raise an error
|
||||
state = SessionState.from_dict(data)
|
||||
assert state.session_id == "test-session-xyz"
|
||||
assert not hasattr(state, "unknown_field")
|
||||
assert not hasattr(state, "another_unknown")
|
||||
|
||||
def test_to_dict_and_from_dict_roundtrip(self):
|
||||
"""Verify to_dict and from_dict are inverses."""
|
||||
original = SessionState(
|
||||
session_id="roundtrip-test",
|
||||
project_name="roundtrip-project",
|
||||
tmux_session_name="gsd-roundtrip",
|
||||
tmux_pane_id="%5",
|
||||
created_at="2026-03-10T10:00:00Z",
|
||||
last_active_at="2026-03-10T11:00:00Z",
|
||||
command_count=100,
|
||||
last_output_snapshot="Final output",
|
||||
display_name="Roundtrip Test"
|
||||
)
|
||||
data = original.to_dict()
|
||||
restored = SessionState.from_dict(data)
|
||||
assert restored.session_id == original.session_id
|
||||
assert restored.project_name == original.project_name
|
||||
assert restored.tmux_session_name == original.tmux_session_name
|
||||
assert restored.tmux_pane_id == original.tmux_pane_id
|
||||
assert restored.created_at == original.created_at
|
||||
assert restored.last_active_at == original.last_active_at
|
||||
assert restored.command_count == original.command_count
|
||||
assert restored.last_output_snapshot == original.last_output_snapshot
|
||||
assert restored.display_name == original.display_name
|
||||
|
||||
|
||||
class TestAtomicWrite:
|
||||
"""Test 1 & 4 & 5: save_state creates file atomically."""
|
||||
|
||||
def test_save_state_creates_file_atomically(self, tmp_path):
|
||||
"""save_state creates file atomically (temp file + rename)."""
|
||||
state = SessionState(
|
||||
session_id="test-atomic",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test",
|
||||
tmux_pane_id="%0"
|
||||
)
|
||||
state_path = tmp_path / "state.json"
|
||||
|
||||
save_state(state, state_path)
|
||||
|
||||
assert state_path.exists()
|
||||
# Verify content is valid JSON
|
||||
with open(state_path, 'r', encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
assert data["session_id"] == "test-atomic"
|
||||
|
||||
def test_temp_file_in_same_directory(self, tmp_path):
|
||||
"""Temp file is created in same directory as target (for atomic rename)."""
|
||||
state = SessionState(
|
||||
session_id="test-same-dir",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test",
|
||||
tmux_pane_id="%1"
|
||||
)
|
||||
state_path = tmp_path / "subdir" / "state.json"
|
||||
|
||||
save_state(state, state_path)
|
||||
|
||||
# After save, no temp files should remain
|
||||
temp_files = list(tmp_path.rglob("*.tmp"))
|
||||
assert len(temp_files) == 0
|
||||
# State file should exist
|
||||
assert state_path.exists()
|
||||
|
||||
def test_cleanup_temp_file_on_failure(self, tmp_path):
|
||||
"""save_state cleans up temp file on failure."""
|
||||
state = SessionState(
|
||||
session_id="test-cleanup",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test",
|
||||
tmux_pane_id="%2"
|
||||
)
|
||||
# Create a file that will cause write to fail (permission denied simulation)
|
||||
state_path = tmp_path / "state.json"
|
||||
|
||||
# Make the directory read-only after creating a temp scenario
|
||||
# This test verifies the exception handling path cleans up
|
||||
# We can't easily test actual failure, but we can verify no temp files remain
|
||||
save_state(state, state_path)
|
||||
|
||||
temp_files = list(tmp_path.glob("*.tmp"))
|
||||
assert len(temp_files) == 0
|
||||
|
||||
|
||||
class TestBackupRotation:
|
||||
"""Test 2 & 3: save_state creates parent dirs and rotates backup."""
|
||||
|
||||
def test_save_state_creates_parent_directories(self, tmp_path):
|
||||
"""save_state creates parent directories if missing."""
|
||||
state = SessionState(
|
||||
session_id="test-mkdir",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test",
|
||||
tmux_pane_id="%3"
|
||||
)
|
||||
# Use nested path that doesn't exist
|
||||
state_path = tmp_path / "deeply" / "nested" / "dir" / "state.json"
|
||||
|
||||
save_state(state, state_path)
|
||||
|
||||
assert state_path.exists()
|
||||
assert state_path.parent.is_dir()
|
||||
|
||||
def test_save_state_rotates_backup(self, tmp_path):
|
||||
"""save_state rotates backup (state.json -> state.json.bak)."""
|
||||
state1 = SessionState(
|
||||
session_id="first-save",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test",
|
||||
tmux_pane_id="%4"
|
||||
)
|
||||
state2 = SessionState(
|
||||
session_id="second-save",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test",
|
||||
tmux_pane_id="%4"
|
||||
)
|
||||
state_path = tmp_path / "state.json"
|
||||
backup_path = tmp_path / "state.json.bak"
|
||||
|
||||
# First save - no backup yet
|
||||
save_state(state1, state_path)
|
||||
assert state_path.exists()
|
||||
assert not backup_path.exists()
|
||||
|
||||
# Second save - backup should be created
|
||||
save_state(state2, state_path)
|
||||
assert state_path.exists()
|
||||
assert backup_path.exists()
|
||||
|
||||
# Backup should contain first save data
|
||||
with open(backup_path, 'r', encoding='utf-8') as f:
|
||||
backup_data = json.load(f)
|
||||
assert backup_data["session_id"] == "first-save"
|
||||
|
||||
# Main file should contain second save data
|
||||
with open(state_path, 'r', encoding='utf-8') as f:
|
||||
main_data = json.load(f)
|
||||
assert main_data["session_id"] == "second-save"
|
||||
|
||||
def test_save_state_overwrites_old_backup(self, tmp_path):
|
||||
"""save_state overwrites old backup file."""
|
||||
state1 = SessionState(session_id="v1", project_name="p", tmux_session_name="t", tmux_pane_id="%0")
|
||||
state2 = SessionState(session_id="v2", project_name="p", tmux_session_name="t", tmux_pane_id="%0")
|
||||
state3 = SessionState(session_id="v3", project_name="p", tmux_session_name="t", tmux_pane_id="%0")
|
||||
state_path = tmp_path / "state.json"
|
||||
backup_path = tmp_path / "state.json.bak"
|
||||
|
||||
save_state(state1, state_path)
|
||||
save_state(state2, state_path)
|
||||
save_state(state3, state_path)
|
||||
|
||||
# Backup should contain v2 (the previous state)
|
||||
with open(backup_path, 'r', encoding='utf-8') as f:
|
||||
backup_data = json.load(f)
|
||||
assert backup_data["session_id"] == "v2"
|
||||
|
||||
# Main should contain v3
|
||||
with open(state_path, 'r', encoding='utf-8') as f:
|
||||
main_data = json.load(f)
|
||||
assert main_data["session_id"] == "v3"
|
||||
|
||||
|
||||
class TestSaveLoadState:
|
||||
"""Test: save and load roundtrip."""
|
||||
|
||||
def test_save_load_state(self, tmp_path):
|
||||
"""save_state and load_state work together."""
|
||||
original = SessionState(
|
||||
session_id="save-load-test",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test",
|
||||
tmux_pane_id="%10",
|
||||
created_at="2026-03-10T10:00:00Z",
|
||||
command_count=50
|
||||
)
|
||||
state_path = tmp_path / "state.json"
|
||||
|
||||
save_state(original, state_path)
|
||||
loaded = load_state(state_path)
|
||||
|
||||
assert loaded is not None
|
||||
assert loaded.session_id == original.session_id
|
||||
assert loaded.project_name == original.project_name
|
||||
assert loaded.tmux_session_name == original.tmux_session_name
|
||||
assert loaded.tmux_pane_id == original.tmux_pane_id
|
||||
assert loaded.command_count == original.command_count
|
||||
|
||||
|
||||
class TestStateValidation:
|
||||
"""Test 3 & 4: load_state validates schema and required fields."""
|
||||
|
||||
def test_load_state_returns_none_for_missing_file(self, tmp_path):
|
||||
"""load_state returns None for missing file."""
|
||||
state_path = tmp_path / "nonexistent.json"
|
||||
result = load_state(state_path)
|
||||
assert result is None
|
||||
|
||||
def test_load_state_returns_session_state_for_valid_file(self, tmp_path):
|
||||
"""load_state returns SessionState for valid file."""
|
||||
state = SessionState(
|
||||
session_id="valid-test",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test",
|
||||
tmux_pane_id="%11"
|
||||
)
|
||||
state_path = tmp_path / "state.json"
|
||||
save_state(state, state_path)
|
||||
|
||||
loaded = load_state(state_path)
|
||||
assert loaded is not None
|
||||
assert loaded.session_id == "valid-test"
|
||||
|
||||
def test_load_state_raises_value_error_for_unsupported_schema(self, tmp_path):
|
||||
"""load_state raises ValueError for unsupported schema version."""
|
||||
state_path = tmp_path / "state.json"
|
||||
# Write a file with unsupported schema version
|
||||
invalid_data = {
|
||||
"schema_version": "2.0", # Unsupported
|
||||
"session_id": "test",
|
||||
"project_name": "test",
|
||||
"tmux_session_name": "test",
|
||||
"tmux_pane_id": "%0"
|
||||
}
|
||||
with open(state_path, 'w', encoding='utf-8') as f:
|
||||
json.dump(invalid_data, f)
|
||||
|
||||
with pytest.raises(ValueError, match="Unsupported schema version"):
|
||||
load_state(state_path)
|
||||
|
||||
def test_load_state_raises_value_error_for_missing_required_fields(self, tmp_path):
|
||||
"""load_state raises ValueError for missing required fields."""
|
||||
state_path = tmp_path / "state.json"
|
||||
# Write a file with missing required field
|
||||
invalid_data = {
|
||||
"schema_version": "1.0",
|
||||
"session_id": "test",
|
||||
# Missing project_name, tmux_session_name
|
||||
}
|
||||
with open(state_path, 'w', encoding='utf-8') as f:
|
||||
json.dump(invalid_data, f)
|
||||
|
||||
with pytest.raises(ValueError, match="Missing required field"):
|
||||
load_state(state_path)
|
||||
|
||||
|
||||
class TestBackupFallback:
|
||||
"""Test 5 & 6: load_state falls back to backup when main file is corrupted."""
|
||||
|
||||
def test_load_state_falls_back_to_backup_on_json_error(self, tmp_path):
|
||||
"""load_state falls back to backup when main file is corrupted (JSONDecodeError)."""
|
||||
state_path = tmp_path / "state.json"
|
||||
backup_path = tmp_path / "state.json.bak"
|
||||
|
||||
# Create a valid backup
|
||||
valid_state = SessionState(
|
||||
session_id="backup-valid",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test",
|
||||
tmux_pane_id="%12"
|
||||
)
|
||||
backup_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(backup_path, 'w', encoding='utf-8') as f:
|
||||
json.dump(valid_state.to_dict(), f)
|
||||
|
||||
# Create corrupted main file
|
||||
with open(state_path, 'w', encoding='utf-8') as f:
|
||||
f.write("{ invalid json }")
|
||||
|
||||
# Should fall back to backup
|
||||
loaded = load_state(state_path)
|
||||
assert loaded is not None
|
||||
assert loaded.session_id == "backup-valid"
|
||||
|
||||
def test_load_state_falls_back_to_backup_on_validation_error(self, tmp_path):
|
||||
"""load_state falls back to backup when validation fails."""
|
||||
state_path = tmp_path / "state.json"
|
||||
backup_path = tmp_path / "state.json.bak"
|
||||
|
||||
# Create a valid backup
|
||||
valid_state = SessionState(
|
||||
session_id="backup-valid-2",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test",
|
||||
tmux_pane_id="%13"
|
||||
)
|
||||
backup_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(backup_path, 'w', encoding='utf-8') as f:
|
||||
json.dump(valid_state.to_dict(), f)
|
||||
|
||||
# Create invalid main file (bad schema)
|
||||
invalid_data = {
|
||||
"schema_version": "99.0", # Invalid
|
||||
"session_id": "test",
|
||||
"project_name": "test",
|
||||
"tmux_session_name": "test",
|
||||
"tmux_pane_id": "%0"
|
||||
}
|
||||
with open(state_path, 'w', encoding='utf-8') as f:
|
||||
json.dump(invalid_data, f)
|
||||
|
||||
# Should fall back to backup
|
||||
loaded = load_state(state_path)
|
||||
assert loaded is not None
|
||||
assert loaded.session_id == "backup-valid-2"
|
||||
|
||||
def test_load_state_tries_backup_when_main_missing(self, tmp_path):
|
||||
"""load_state tries backup when main file is missing."""
|
||||
state_path = tmp_path / "state.json"
|
||||
backup_path = tmp_path / "state.json.bak"
|
||||
|
||||
# Create a valid backup (no main file)
|
||||
valid_state = SessionState(
|
||||
session_id="backup-only",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test",
|
||||
tmux_pane_id="%14"
|
||||
)
|
||||
backup_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(backup_path, 'w', encoding='utf-8') as f:
|
||||
json.dump(valid_state.to_dict(), f)
|
||||
|
||||
# Should load from backup
|
||||
loaded = load_state(state_path)
|
||||
assert loaded is not None
|
||||
assert loaded.session_id == "backup-only"
|
||||
|
||||
|
||||
class TestStateManagerClass:
|
||||
"""Test StateManager class with workspace path resolution."""
|
||||
|
||||
def test_get_state_path_returns_correct_path(self, tmp_path):
|
||||
"""StateManager.get_state_path returns correct path for project."""
|
||||
manager = StateManager("my-project", workspaces_root=tmp_path)
|
||||
expected = tmp_path / "my-project" / ".gsd" / "state.json"
|
||||
assert manager.get_state_path() == expected
|
||||
|
||||
def test_state_manager_save_persists_state(self, tmp_path):
|
||||
"""StateManager.save persists state to correct location."""
|
||||
manager = StateManager("test-project", workspaces_root=tmp_path)
|
||||
state = SessionState(
|
||||
session_id="manager-test",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test-project",
|
||||
tmux_pane_id="%20"
|
||||
)
|
||||
|
||||
manager.save(state)
|
||||
|
||||
# Verify file was created at correct path
|
||||
state_path = manager.get_state_path()
|
||||
assert state_path.exists()
|
||||
|
||||
# Verify content
|
||||
loaded = load_state(state_path)
|
||||
assert loaded is not None
|
||||
assert loaded.session_id == "manager-test"
|
||||
|
||||
def test_state_manager_load_retrieves_state(self, tmp_path):
|
||||
"""StateManager.load retrieves state from correct location."""
|
||||
manager = StateManager("load-test-project", workspaces_root=tmp_path)
|
||||
state = SessionState(
|
||||
session_id="load-manager-test",
|
||||
project_name="load-test-project",
|
||||
tmux_session_name="gsd-load-test",
|
||||
tmux_pane_id="%21"
|
||||
)
|
||||
|
||||
# Save using manager
|
||||
manager.save(state)
|
||||
|
||||
# Load using manager
|
||||
loaded = manager.load()
|
||||
|
||||
assert loaded is not None
|
||||
assert loaded.session_id == "load-manager-test"
|
||||
assert loaded.project_name == "load-test-project"
|
||||
|
||||
def test_state_manager_load_returns_none_when_no_state(self, tmp_path):
|
||||
"""StateManager.load returns None when no state file exists."""
|
||||
manager = StateManager("nonexistent-project", workspaces_root=tmp_path)
|
||||
loaded = manager.load()
|
||||
assert loaded is None
|
||||
|
||||
def test_state_manager_default_workspaces_root(self):
|
||||
"""StateManager uses default ~/gsd-workspaces when not specified."""
|
||||
manager = StateManager("default-project")
|
||||
expected_root = Path.home() / "gsd-workspaces"
|
||||
assert manager.workspaces_root == expected_root
|
||||
assert manager.workspace_path == expected_root / "default-project"
|
||||
@@ -0,0 +1,772 @@
|
||||
"""Tests for tmux session lifecycle management."""
|
||||
import pytest
|
||||
import libtmux
|
||||
import libtmux.exc
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch, PropertyMock
|
||||
from datetime import datetime
|
||||
|
||||
from gsd_openclaw.tmux_session import TmuxSession, TmuxSessionManager, TmuxSessionError, CleanupResult
|
||||
|
||||
|
||||
class TestTmuxSessionDataclass:
|
||||
"""Tests for TmuxSession dataclass."""
|
||||
|
||||
def test_tmux_session_stores_metadata_correctly(self):
|
||||
"""TmuxSession should store session metadata correctly."""
|
||||
session = TmuxSession(
|
||||
session_id="test-123",
|
||||
project_name="my-project",
|
||||
tmux_session_name="gsd-my-project"
|
||||
)
|
||||
|
||||
assert session.session_id == "test-123"
|
||||
assert session.project_name == "my-project"
|
||||
assert session.tmux_session_name == "gsd-my-project"
|
||||
assert session.tmux_session is None
|
||||
|
||||
def test_tmux_session_exists_false_when_no_runtime_reference(self):
|
||||
"""TmuxSession.exists should return False when no runtime reference."""
|
||||
session = TmuxSession(
|
||||
session_id="test-123",
|
||||
project_name="my-project",
|
||||
tmux_session_name="gsd-my-project"
|
||||
)
|
||||
|
||||
assert session.exists is False
|
||||
|
||||
def test_tmux_session_exists_true_when_session_active(self):
|
||||
"""TmuxSession.exists should return True when session is active."""
|
||||
# Create a mock tmux session
|
||||
mock_tmux = MagicMock()
|
||||
mock_tmux.session_id = "%123"
|
||||
|
||||
session = TmuxSession(
|
||||
session_id="test-123",
|
||||
project_name="my-project",
|
||||
tmux_session_name="gsd-my-project",
|
||||
tmux_session=mock_tmux
|
||||
)
|
||||
|
||||
assert session.exists is True
|
||||
|
||||
def test_tmux_session_exists_false_on_libtmux_exception(self):
|
||||
"""TmuxSession.exists should return False on LibTmuxException."""
|
||||
# Create a mock that raises exception when accessing session_id
|
||||
mock_tmux = MagicMock()
|
||||
type(mock_tmux).session_id = PropertyMock(side_effect=libtmux.exc.LibTmuxException("gone"))
|
||||
|
||||
session = TmuxSession(
|
||||
session_id="test-123",
|
||||
project_name="my-project",
|
||||
tmux_session_name="gsd-my-project",
|
||||
tmux_session=mock_tmux
|
||||
)
|
||||
|
||||
assert session.exists is False
|
||||
|
||||
|
||||
class TestTmuxSessionManagerBasic:
|
||||
"""Tests for TmuxSessionManager basic functionality."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_server(self):
|
||||
"""Create a mock libtmux server."""
|
||||
server = MagicMock(spec=libtmux.Server)
|
||||
server.sessions = MagicMock()
|
||||
server.sessions.filter = MagicMock(return_value=[])
|
||||
return server
|
||||
|
||||
@pytest.fixture
|
||||
def manager(self, mock_server):
|
||||
"""Create TmuxSessionManager with mock server."""
|
||||
return TmuxSessionManager(server=mock_server)
|
||||
|
||||
def test_server_connection_works(self, manager, mock_server):
|
||||
"""TmuxSessionManager should have a server reference."""
|
||||
assert manager.server is not None
|
||||
assert manager.server is mock_server
|
||||
|
||||
def test_list_sessions_returns_list(self, manager, mock_server):
|
||||
"""list_sessions should return a list."""
|
||||
mock_server.sessions.filter.return_value = []
|
||||
sessions = manager.list_sessions()
|
||||
assert isinstance(sessions, list)
|
||||
|
||||
def test_list_sessions_filters_gsd_prefix(self, manager, mock_server):
|
||||
"""list_sessions should filter for gsd- prefix."""
|
||||
# Create mock sessions
|
||||
mock_session1 = MagicMock()
|
||||
mock_session1.name = "gsd-project1"
|
||||
mock_session2 = MagicMock()
|
||||
mock_session2.name = "gsd-project2"
|
||||
|
||||
mock_server.sessions.filter.return_value = [mock_session1, mock_session2]
|
||||
|
||||
sessions = manager.list_sessions()
|
||||
assert len(sessions) == 2
|
||||
mock_server.sessions.filter.assert_called_once_with(
|
||||
session_name__startswith="gsd-"
|
||||
)
|
||||
|
||||
def test_generate_session_name_format(self, manager):
|
||||
"""_generate_session_name should return gsd-{project} format."""
|
||||
name = manager._generate_session_name("my-project")
|
||||
assert name == "gsd-my-project"
|
||||
|
||||
name2 = manager._generate_session_name("another_project")
|
||||
assert name2 == "gsd-another_project"
|
||||
|
||||
def test_resolve_collision_returns_base_when_no_collision(self, manager, mock_server):
|
||||
"""_resolve_collision should return base name when no collision exists."""
|
||||
mock_server.sessions.filter.return_value = []
|
||||
result = manager._resolve_collision("gsd-my-project")
|
||||
assert result == "gsd-my-project"
|
||||
|
||||
def test_resolve_collision_adds_timestamp_when_collision_exists(self, manager, mock_server):
|
||||
"""_resolve_collision should add timestamp suffix when collision exists."""
|
||||
# Simulate existing session
|
||||
mock_session = MagicMock()
|
||||
mock_server.sessions.filter.return_value = [mock_session]
|
||||
|
||||
result = manager._resolve_collision("gsd-collision-test")
|
||||
|
||||
# Should have timestamp suffix
|
||||
assert result.startswith("gsd-collision-test-")
|
||||
# Should match timestamp pattern -YYYYMMDD-HHMMSS
|
||||
import re
|
||||
pattern = r"gsd-collision-test-\d{8}-\d{6}"
|
||||
assert re.match(pattern, result), f"Result '{result}' doesn't match pattern {pattern}"
|
||||
|
||||
|
||||
class TestTmuxSessionManagerCreateSession:
|
||||
"""Tests for create_session functionality."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_server(self):
|
||||
"""Create a mock libtmux server."""
|
||||
server = MagicMock(spec=libtmux.Server)
|
||||
server.sessions = MagicMock()
|
||||
server.sessions.filter = MagicMock(return_value=[])
|
||||
server.sessions.get = MagicMock(return_value=None)
|
||||
|
||||
# Mock new_session to return a mock session
|
||||
mock_session = MagicMock()
|
||||
mock_session.name = "gsd-test-project"
|
||||
mock_pane = MagicMock()
|
||||
mock_pane.pane_id = "%100"
|
||||
mock_session.active_pane = mock_pane
|
||||
server.new_session = MagicMock(return_value=mock_session)
|
||||
|
||||
return server
|
||||
|
||||
@pytest.fixture
|
||||
def manager(self, mock_server):
|
||||
"""Create TmuxSessionManager with mock server."""
|
||||
return TmuxSessionManager(server=mock_server)
|
||||
|
||||
def test_create_session_creates_tmux_session(self, manager, mock_server, tmp_workspaces_root):
|
||||
"""create_session should create a tmux session with correct name."""
|
||||
with patch('gsd_openclaw.tmux_session.WorkspaceManager') as MockWM:
|
||||
mock_wm = MagicMock()
|
||||
MockWM.return_value = mock_wm
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
result = manager.create_session("test-project")
|
||||
|
||||
# Verify tmux session was created
|
||||
mock_server.new_session.assert_called_once()
|
||||
call_kwargs = mock_server.new_session.call_args[1]
|
||||
assert call_kwargs['session_name'].startswith("gsd-test-project")
|
||||
assert call_kwargs['attach'] is False
|
||||
assert call_kwargs['window_name'] == 'main'
|
||||
|
||||
def test_create_session_persists_state(self, manager, mock_server, tmp_workspaces_root):
|
||||
"""create_session should persist state to workspace."""
|
||||
with patch('gsd_openclaw.tmux_session.WorkspaceManager') as MockWM:
|
||||
mock_wm = MagicMock()
|
||||
MockWM.return_value = mock_wm
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
result = manager.create_session("test-project")
|
||||
|
||||
# Verify state was saved
|
||||
mock_sm.save.assert_called_once()
|
||||
|
||||
def test_create_session_handles_collision(self, manager, mock_server, tmp_workspaces_root):
|
||||
"""create_session should handle name collision with timestamp suffix."""
|
||||
# Simulate collision
|
||||
mock_session = MagicMock()
|
||||
mock_server.sessions.filter.return_value = [mock_session]
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.WorkspaceManager') as MockWM:
|
||||
mock_wm = MagicMock()
|
||||
MockWM.return_value = mock_wm
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
result = manager.create_session("collision-test")
|
||||
|
||||
# Verify session name has timestamp suffix
|
||||
call_kwargs = mock_server.new_session.call_args[1]
|
||||
session_name = call_kwargs['session_name']
|
||||
assert session_name.startswith("gsd-collision-test-")
|
||||
import re
|
||||
assert re.match(r"gsd-collision-test-\d{8}-\d{6}", session_name)
|
||||
|
||||
def test_create_session_returns_tmux_session(self, manager, mock_server, tmp_workspaces_root):
|
||||
"""create_session should return TmuxSession with all metadata."""
|
||||
with patch('gsd_openclaw.tmux_session.WorkspaceManager') as MockWM:
|
||||
mock_wm = MagicMock()
|
||||
MockWM.return_value = mock_wm
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
result = manager.create_session("test-project", display_name="My Project")
|
||||
|
||||
assert isinstance(result, TmuxSession)
|
||||
assert result.project_name == "test-project"
|
||||
assert result.tmux_session_name.startswith("gsd-test-project")
|
||||
assert result.tmux_session is not None
|
||||
|
||||
|
||||
class TestTmuxSessionManagerDestroySession:
|
||||
"""Tests for destroy_session functionality."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_server(self):
|
||||
"""Create a mock libtmux server."""
|
||||
server = MagicMock(spec=libtmux.Server)
|
||||
server.sessions = MagicMock()
|
||||
server.sessions.filter = MagicMock(return_value=[])
|
||||
server.sessions.get = MagicMock(return_value=None)
|
||||
return server
|
||||
|
||||
@pytest.fixture
|
||||
def manager(self, mock_server):
|
||||
"""Create TmuxSessionManager with mock server."""
|
||||
return TmuxSessionManager(server=mock_server)
|
||||
|
||||
def test_destroy_session_kills_tmux_session(self, manager, mock_server, tmp_workspaces_root):
|
||||
"""destroy_session should kill the tmux session."""
|
||||
# Mock state
|
||||
mock_state = MagicMock()
|
||||
mock_state.tmux_session_name = "gsd-test-project"
|
||||
|
||||
# Mock tmux session
|
||||
mock_tmux_session = MagicMock()
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = mock_state
|
||||
mock_sm.state_path = tmp_workspaces_root / "test-project" / ".gsd" / "state.json"
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
mock_server.sessions.get.return_value = mock_tmux_session
|
||||
|
||||
result = manager.destroy_session("test-project")
|
||||
|
||||
mock_tmux_session.kill.assert_called_once()
|
||||
assert result is True
|
||||
|
||||
def test_destroy_session_removes_state_file(self, manager, mock_server, tmp_workspaces_root):
|
||||
"""destroy_session should remove state file."""
|
||||
# Create a temp state file
|
||||
state_dir = tmp_workspaces_root / "test-project" / ".gsd"
|
||||
state_dir.mkdir(parents=True, exist_ok=True)
|
||||
state_file = state_dir / "state.json"
|
||||
state_file.write_text('{"test": "data"}')
|
||||
backup_file = state_dir / "state.json.bak"
|
||||
backup_file.write_text('{"test": "backup"}')
|
||||
|
||||
# Mock state
|
||||
mock_state = MagicMock()
|
||||
mock_state.tmux_session_name = "gsd-test-project"
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = mock_state
|
||||
mock_sm.state_path = state_file
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
mock_server.sessions.get.return_value = None
|
||||
|
||||
result = manager.destroy_session("test-project")
|
||||
|
||||
assert not state_file.exists()
|
||||
assert not backup_file.exists()
|
||||
|
||||
def test_destroy_session_returns_false_when_no_state(self, manager, mock_server, tmp_workspaces_root):
|
||||
"""destroy_session should return False when no state exists."""
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = None
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
result = manager.destroy_session("nonexistent-project")
|
||||
|
||||
assert result is False
|
||||
|
||||
def test_destroy_session_handles_already_destroyed_session(self, manager, mock_server, tmp_workspaces_root):
|
||||
"""destroy_session should handle already-destroyed session gracefully."""
|
||||
# Mock state
|
||||
mock_state = MagicMock()
|
||||
mock_state.tmux_session_name = "gsd-test-project"
|
||||
|
||||
state_dir = tmp_workspaces_root / "test-project" / ".gsd"
|
||||
state_dir.mkdir(parents=True, exist_ok=True)
|
||||
state_file = state_dir / "state.json"
|
||||
state_file.write_text('{}')
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = mock_state
|
||||
mock_sm.state_path = state_file
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
# Session already gone
|
||||
mock_server.sessions.get.return_value = None
|
||||
|
||||
result = manager.destroy_session("test-project")
|
||||
|
||||
assert result is True
|
||||
|
||||
def test_destroy_session_with_cleanup_removes_workspace(self, manager, mock_server, tmp_workspaces_root):
|
||||
"""destroy_session with cleanup=True should remove workspace."""
|
||||
# Mock state
|
||||
mock_state = MagicMock()
|
||||
mock_state.tmux_session_name = "gsd-test-project"
|
||||
|
||||
state_dir = tmp_workspaces_root / "test-project" / ".gsd"
|
||||
state_dir.mkdir(parents=True, exist_ok=True)
|
||||
state_file = state_dir / "state.json"
|
||||
state_file.write_text('{}')
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = mock_state
|
||||
mock_sm.state_path = state_file
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
mock_server.sessions.get.return_value = None
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.WorkspaceManager') as MockWM:
|
||||
mock_wm = MagicMock()
|
||||
MockWM.return_value = mock_wm
|
||||
|
||||
result = manager.destroy_session("test-project", cleanup_workspace=True)
|
||||
|
||||
mock_wm.destroy.assert_called_once()
|
||||
|
||||
|
||||
class TestTmuxSessionManagerListGsdSessions:
|
||||
"""Tests for list_gsd_sessions functionality."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_server(self):
|
||||
"""Create a mock libtmux server."""
|
||||
server = MagicMock(spec=libtmux.Server)
|
||||
server.sessions = MagicMock()
|
||||
server.sessions.filter = MagicMock(return_value=[])
|
||||
server.sessions.get = MagicMock(return_value=None)
|
||||
return server
|
||||
|
||||
@pytest.fixture
|
||||
def manager(self, mock_server):
|
||||
"""Create TmuxSessionManager with mock server."""
|
||||
return TmuxSessionManager(server=mock_server)
|
||||
|
||||
def test_list_gsd_sessions_returns_all_gsd_sessions(self, manager, mock_server):
|
||||
"""list_gsd_sessions should return all gsd-* sessions."""
|
||||
# Create mock sessions
|
||||
mock_session1 = MagicMock()
|
||||
mock_session1.name = "gsd-project1"
|
||||
mock_session2 = MagicMock()
|
||||
mock_session2.name = "gsd-project2"
|
||||
|
||||
mock_server.sessions.filter.return_value = [mock_session1, mock_session2]
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = None
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
sessions = manager.list_gsd_sessions()
|
||||
|
||||
assert len(sessions) == 2
|
||||
|
||||
def test_list_gsd_sessions_handles_orphan_sessions(self, manager, mock_server):
|
||||
"""list_gsd_sessions should handle orphan sessions (no state file)."""
|
||||
# Create mock session
|
||||
mock_session = MagicMock()
|
||||
mock_session.name = "gsd-orphan-project"
|
||||
mock_server.sessions.filter.return_value = [mock_session]
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = None # No state
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
sessions = manager.list_gsd_sessions()
|
||||
|
||||
assert len(sessions) == 1
|
||||
assert sessions[0].is_orphan is True
|
||||
assert sessions[0].session_id == "?"
|
||||
|
||||
def test_list_gsd_sessions_includes_project_associations(self, manager, mock_server):
|
||||
"""list_gsd_sessions should include project associations from state files."""
|
||||
# Create mock session
|
||||
mock_session = MagicMock()
|
||||
mock_session.name = "gsd-my-project"
|
||||
mock_server.sessions.filter.return_value = [mock_session]
|
||||
|
||||
# Mock state
|
||||
mock_state = MagicMock()
|
||||
mock_state.tmux_session_name = "gsd-my-project"
|
||||
mock_state.session_id = "abc123"
|
||||
mock_state.created_at = "2026-03-10T10:00:00"
|
||||
mock_state.last_active_at = "2026-03-10T12:00:00"
|
||||
mock_state.command_count = 5
|
||||
mock_state.display_name = "My Project"
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = mock_state
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
sessions = manager.list_gsd_sessions()
|
||||
|
||||
assert len(sessions) == 1
|
||||
assert sessions[0].is_orphan is False
|
||||
assert sessions[0].project_name == "my-project"
|
||||
assert sessions[0].session_id == "abc123"
|
||||
assert sessions[0].display_name == "My Project"
|
||||
|
||||
|
||||
class TestTmuxSessionManagerResumeSession:
|
||||
"""Tests for resume_session functionality."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_server(self):
|
||||
"""Create a mock libtmux server."""
|
||||
server = MagicMock(spec=libtmux.Server)
|
||||
server.sessions = MagicMock()
|
||||
server.sessions.filter = MagicMock(return_value=[])
|
||||
server.sessions.get = MagicMock(return_value=None)
|
||||
|
||||
# Mock new_session to return a mock session
|
||||
mock_session = MagicMock()
|
||||
mock_session.name = "gsd-test-project"
|
||||
mock_pane = MagicMock()
|
||||
mock_pane.pane_id = "%200"
|
||||
mock_session.active_pane = mock_pane
|
||||
server.new_session = MagicMock(return_value=mock_session)
|
||||
|
||||
return server
|
||||
|
||||
@pytest.fixture
|
||||
def manager(self, mock_server):
|
||||
"""Create TmuxSessionManager with mock server."""
|
||||
return TmuxSessionManager(server=mock_server)
|
||||
|
||||
def test_resume_session_returns_existing_session_if_state_valid(self, manager, mock_server):
|
||||
"""resume_session should return existing session if state valid and tmux exists."""
|
||||
from gsd_openclaw.state_manager import SessionState
|
||||
|
||||
# Mock state
|
||||
mock_state = SessionState(
|
||||
schema_version="1.0",
|
||||
session_id="test-123",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test-project",
|
||||
tmux_pane_id="%100",
|
||||
created_at="2026-03-10T10:00:00",
|
||||
last_active_at="2026-03-10T12:00:00"
|
||||
)
|
||||
|
||||
# Mock tmux session exists
|
||||
mock_tmux_session = MagicMock()
|
||||
mock_server.sessions.get.return_value = mock_tmux_session
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = mock_state
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
result = manager.resume_session("test-project")
|
||||
|
||||
assert result.session_id == "test-123"
|
||||
assert result.project_name == "test-project"
|
||||
assert result.tmux_session_name == "gsd-test-project"
|
||||
mock_server.sessions.get.assert_called_once_with(session_name="gsd-test-project")
|
||||
|
||||
def test_resume_session_recreates_tmux_if_missing(self, manager, mock_server):
|
||||
"""resume_session should recreate tmux session if state exists but tmux missing."""
|
||||
from gsd_openclaw.state_manager import SessionState
|
||||
|
||||
# Mock state
|
||||
mock_state = SessionState(
|
||||
schema_version="1.0",
|
||||
session_id="test-123",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test-project",
|
||||
tmux_pane_id="%100",
|
||||
created_at="2026-03-10T10:00:00",
|
||||
last_active_at="2026-03-10T12:00:00"
|
||||
)
|
||||
|
||||
# Mock tmux session does not exist initially
|
||||
mock_server.sessions.get.return_value = None
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = mock_state
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.WorkspaceManager') as MockWM:
|
||||
mock_wm = MagicMock()
|
||||
mock_wm.exists.return_value = True
|
||||
mock_workspace = MagicMock()
|
||||
mock_workspace.root = "/tmp/test-workspace"
|
||||
mock_wm.get_workspace.return_value = mock_workspace
|
||||
MockWM.return_value = mock_wm
|
||||
|
||||
result = manager.resume_session("test-project")
|
||||
|
||||
# Verify tmux session was recreated
|
||||
mock_server.new_session.assert_called_once()
|
||||
assert result.tmux_session is not None
|
||||
|
||||
def test_resume_session_raises_error_if_no_state(self, manager, mock_server):
|
||||
"""resume_session should raise error if no state file exists."""
|
||||
from gsd_openclaw.tmux_session import TmuxSessionError
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = None
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
with pytest.raises(TmuxSessionError, match="No session found"):
|
||||
manager.resume_session("nonexistent-project")
|
||||
|
||||
def test_resume_session_validates_schema_version(self, manager, mock_server):
|
||||
"""resume_session should validate schema version before resuming."""
|
||||
from gsd_openclaw.tmux_session import TmuxSessionError
|
||||
from gsd_openclaw.state_manager import SessionState
|
||||
|
||||
# Mock state with invalid schema version
|
||||
mock_state = SessionState(
|
||||
schema_version="2.0", # Invalid version
|
||||
session_id="test-123",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test-project",
|
||||
tmux_pane_id="%100",
|
||||
created_at="2026-03-10T10:00:00",
|
||||
last_active_at="2026-03-10T12:00:00"
|
||||
)
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = mock_state
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
with pytest.raises(TmuxSessionError, match="Unsupported state version"):
|
||||
manager.resume_session("test-project")
|
||||
|
||||
def test_get_session_returns_none_if_no_state(self, manager, mock_server):
|
||||
"""get_session should return None if no state file exists."""
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = None
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
result = manager.get_session("nonexistent-project")
|
||||
|
||||
assert result is None
|
||||
|
||||
def test_get_session_returns_session_if_state_exists(self, manager, mock_server):
|
||||
"""get_session should return TmuxSession if state exists."""
|
||||
from gsd_openclaw.state_manager import SessionState
|
||||
|
||||
# Mock state
|
||||
mock_state = SessionState(
|
||||
schema_version="1.0",
|
||||
session_id="test-123",
|
||||
project_name="test-project",
|
||||
tmux_session_name="gsd-test-project",
|
||||
tmux_pane_id="%100",
|
||||
created_at="2026-03-10T10:00:00",
|
||||
last_active_at="2026-03-10T12:00:00"
|
||||
)
|
||||
|
||||
# Mock tmux session exists
|
||||
mock_tmux_session = MagicMock()
|
||||
mock_server.sessions.get.return_value = mock_tmux_session
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = mock_state
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
result = manager.get_session("test-project")
|
||||
|
||||
assert result is not None
|
||||
assert result.session_id == "test-123"
|
||||
assert result.project_name == "test-project"
|
||||
|
||||
|
||||
class TestTmuxSessionManagerCleanupOrphans:
|
||||
"""Tests for cleanup_orphans functionality."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_server(self):
|
||||
"""Create a mock libtmux server."""
|
||||
server = MagicMock(spec=libtmux.Server)
|
||||
server.sessions = MagicMock()
|
||||
server.sessions.filter = MagicMock(return_value=[])
|
||||
server.sessions.get = MagicMock(return_value=None)
|
||||
return server
|
||||
|
||||
@pytest.fixture
|
||||
def manager(self, mock_server):
|
||||
"""Create TmuxSessionManager with mock server."""
|
||||
return TmuxSessionManager(server=mock_server)
|
||||
|
||||
def test_cleanup_orphans_kills_sessions_without_state(self, manager, mock_server):
|
||||
"""cleanup_orphans should kill tmux sessions without state files."""
|
||||
# Create orphan tmux session
|
||||
mock_orphan = MagicMock()
|
||||
mock_orphan.name = "gsd-orphan-project"
|
||||
mock_orphan.kill = MagicMock()
|
||||
mock_server.sessions.filter.return_value = [mock_orphan]
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = None # No state file
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
result = manager.cleanup_orphans()
|
||||
|
||||
mock_orphan.kill.assert_called_once()
|
||||
assert "gsd-orphan-project" in result.cleaned_sessions
|
||||
assert len(result.errors) == 0
|
||||
|
||||
def test_cleanup_orphans_returns_list_of_cleaned_names(self, manager, mock_server):
|
||||
"""cleanup_orphans should return list of cleaned session names."""
|
||||
# Create multiple orphan sessions
|
||||
mock_orphan1 = MagicMock()
|
||||
mock_orphan1.name = "gsd-orphan1"
|
||||
mock_orphan1.kill = MagicMock()
|
||||
mock_orphan2 = MagicMock()
|
||||
mock_orphan2.name = "gsd-orphan2"
|
||||
mock_orphan2.kill = MagicMock()
|
||||
mock_server.sessions.filter.return_value = [mock_orphan1, mock_orphan2]
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = None
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
result = manager.cleanup_orphans()
|
||||
|
||||
assert len(result.cleaned_sessions) == 2
|
||||
assert "gsd-orphan1" in result.cleaned_sessions
|
||||
assert "gsd-orphan2" in result.cleaned_sessions
|
||||
|
||||
def test_cleanup_orphans_handles_kill_errors(self, manager, mock_server):
|
||||
"""cleanup_orphans should handle errors when killing sessions."""
|
||||
# Create orphan session that fails to kill
|
||||
mock_orphan = MagicMock()
|
||||
mock_orphan.name = "gsd-failing-orphan"
|
||||
mock_orphan.kill.side_effect = libtmux.exc.LibTmuxException("kill failed")
|
||||
mock_server.sessions.filter.return_value = [mock_orphan]
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = None
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
result = manager.cleanup_orphans()
|
||||
|
||||
assert len(result.cleaned_sessions) == 0
|
||||
assert len(result.errors) == 1
|
||||
assert "gsd-failing-orphan" in result.errors[0]
|
||||
|
||||
def test_cleanup_orphans_skips_sessions_with_valid_state(self, manager, mock_server):
|
||||
"""cleanup_orphans should not kill sessions with valid state files."""
|
||||
from gsd_openclaw.state_manager import SessionState
|
||||
|
||||
# Create session with valid state
|
||||
mock_session = MagicMock()
|
||||
mock_session.name = "gsd-valid-project"
|
||||
mock_session.kill = MagicMock()
|
||||
mock_server.sessions.filter.return_value = [mock_session]
|
||||
|
||||
# Mock valid state
|
||||
mock_state = SessionState(
|
||||
schema_version="1.0",
|
||||
session_id="valid-123",
|
||||
project_name="valid-project",
|
||||
tmux_session_name="gsd-valid-project",
|
||||
tmux_pane_id="%100",
|
||||
created_at="2026-03-10T10:00:00",
|
||||
last_active_at="2026-03-10T12:00:00"
|
||||
)
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = mock_state
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
result = manager.cleanup_orphans()
|
||||
|
||||
mock_session.kill.assert_not_called()
|
||||
assert len(result.cleaned_sessions) == 0
|
||||
|
||||
def test_cleanup_orphans_handles_mismatched_state(self, manager, mock_server):
|
||||
"""cleanup_orphans should kill sessions with state referencing different tmux."""
|
||||
from gsd_openclaw.state_manager import SessionState
|
||||
|
||||
# Create session with mismatched state
|
||||
mock_session = MagicMock()
|
||||
mock_session.name = "gsd-mismatched"
|
||||
mock_session.kill = MagicMock()
|
||||
mock_server.sessions.filter.return_value = [mock_session]
|
||||
|
||||
# Mock state that references different tmux session
|
||||
mock_state = SessionState(
|
||||
schema_version="1.0",
|
||||
session_id="test-123",
|
||||
project_name="mismatched",
|
||||
tmux_session_name="gsd-different-name", # Different from actual
|
||||
tmux_pane_id="%100",
|
||||
created_at="2026-03-10T10:00:00",
|
||||
last_active_at="2026-03-10T12:00:00"
|
||||
)
|
||||
|
||||
with patch('gsd_openclaw.tmux_session.StateManager') as MockSM:
|
||||
mock_sm = MagicMock()
|
||||
mock_sm.load.return_value = mock_state
|
||||
MockSM.return_value = mock_sm
|
||||
|
||||
result = manager.cleanup_orphans()
|
||||
|
||||
mock_session.kill.assert_called_once()
|
||||
assert "gsd-mismatched" in result.cleaned_sessions
|
||||
|
||||
|
||||
# Fixtures needed for tests
|
||||
@pytest.fixture
|
||||
def tmp_workspaces_root(tmp_path: Path) -> Path:
|
||||
"""Temporary root for workspace testing."""
|
||||
return tmp_path / "gsd-workspaces"
|
||||
@@ -0,0 +1,122 @@
|
||||
"""Tests for workspace directory management."""
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
|
||||
from gsd_openclaw.workspace import Workspace, WorkspaceManager
|
||||
|
||||
|
||||
class TestWorkspacePaths:
|
||||
"""Test Workspace dataclass path properties."""
|
||||
|
||||
def test_workspace_creates_correct_paths_for_all_directories(self, tmp_workspaces_root: Path):
|
||||
"""Workspace creates correct paths for all directories."""
|
||||
root = tmp_workspaces_root / "test-project"
|
||||
workspace = Workspace("test-project", root)
|
||||
|
||||
assert workspace.root == root
|
||||
assert workspace.gsd_dir == root / ".gsd"
|
||||
assert workspace.state_path == root / ".gsd" / "state.json"
|
||||
assert workspace.state_backup_path == root / ".gsd" / "state.json.bak"
|
||||
assert workspace.session_dump_dir == root / ".gsd" / "session-dump"
|
||||
assert workspace.code_dir == root / "code"
|
||||
assert workspace.io_logs_dir == root / "io-logs"
|
||||
|
||||
def test_workspace_paths_are_consistent_same_root(self, tmp_workspaces_root: Path):
|
||||
"""Workspace paths are consistent (same root)."""
|
||||
root = tmp_workspaces_root / "my-project"
|
||||
workspace = Workspace("my-project", root)
|
||||
|
||||
# All paths should start with the same root
|
||||
assert str(workspace.gsd_dir).startswith(str(workspace.root))
|
||||
assert str(workspace.state_path).startswith(str(workspace.root))
|
||||
assert str(workspace.code_dir).startswith(str(workspace.root))
|
||||
assert str(workspace.io_logs_dir).startswith(str(workspace.root))
|
||||
|
||||
def test_workspace_handles_project_names_with_special_characters_safely(self, tmp_workspaces_root: Path):
|
||||
"""Workspace handles project names with special characters safely."""
|
||||
# Test with project name that has spaces and hyphens
|
||||
root = tmp_workspaces_root / "my-project-2024"
|
||||
workspace = Workspace("my-project-2024", root)
|
||||
|
||||
assert workspace.root == root
|
||||
assert workspace.project_name == "my-project-2024"
|
||||
|
||||
def test_workspace_io_log_path_with_session_id(self, tmp_workspaces_root: Path):
|
||||
"""Workspace io_log_path returns correct path for session ID."""
|
||||
root = tmp_workspaces_root / "test-project"
|
||||
workspace = Workspace("test-project", root)
|
||||
|
||||
log_path = workspace.io_log_path("session-123")
|
||||
assert log_path == root / "io-logs" / "session-123.log"
|
||||
|
||||
|
||||
class TestWorkspaceManagerCreate:
|
||||
"""Test WorkspaceManager create operations."""
|
||||
|
||||
def test_create_creates_all_required_directories(self, workspace_manager: WorkspaceManager):
|
||||
"""create() creates all required directories."""
|
||||
workspace = workspace_manager.create()
|
||||
|
||||
assert workspace.root.exists()
|
||||
assert workspace.gsd_dir.exists()
|
||||
assert workspace.session_dump_dir.exists()
|
||||
assert workspace.code_dir.exists()
|
||||
assert workspace.io_logs_dir.exists()
|
||||
|
||||
def test_create_is_idempotent(self, workspace_manager: WorkspaceManager):
|
||||
"""create() is idempotent (calling twice doesn't fail)."""
|
||||
workspace1 = workspace_manager.create()
|
||||
workspace2 = workspace_manager.create() # Should not raise
|
||||
|
||||
assert workspace1.root.exists()
|
||||
assert workspace2.root.exists()
|
||||
|
||||
|
||||
class TestWorkspaceManagerDestroy:
|
||||
"""Test WorkspaceManager destroy operations."""
|
||||
|
||||
def test_destroy_removes_entire_workspace_directory(self, tmp_workspaces_root: Path):
|
||||
"""destroy() removes entire workspace directory."""
|
||||
manager = WorkspaceManager("test-project", tmp_workspaces_root)
|
||||
|
||||
manager.create()
|
||||
assert manager.workspace.root.exists()
|
||||
|
||||
manager.destroy()
|
||||
assert not manager.workspace.root.exists()
|
||||
|
||||
def test_exists_returns_true_after_create_false_after_destroy(self, tmp_workspaces_root: Path):
|
||||
"""exists() returns True after create, False after destroy."""
|
||||
manager = WorkspaceManager("test-project", tmp_workspaces_root)
|
||||
|
||||
assert not manager.exists()
|
||||
|
||||
manager.create()
|
||||
assert manager.exists()
|
||||
|
||||
manager.destroy()
|
||||
assert not manager.exists()
|
||||
|
||||
|
||||
class TestWorkspaceIsolation:
|
||||
"""Test workspace isolation between projects."""
|
||||
|
||||
def test_two_projects_have_isolated_workspaces(self, tmp_workspaces_root: Path):
|
||||
"""Two projects have isolated workspaces (different roots)."""
|
||||
manager1 = WorkspaceManager("project-alpha", tmp_workspaces_root)
|
||||
manager2 = WorkspaceManager("project-beta", tmp_workspaces_root)
|
||||
|
||||
workspace1 = manager1.create()
|
||||
workspace2 = manager2.create()
|
||||
|
||||
# Different roots
|
||||
assert workspace1.root != workspace2.root
|
||||
|
||||
# Both exist independently
|
||||
assert workspace1.root.exists()
|
||||
assert workspace2.root.exists()
|
||||
|
||||
# Destroying one doesn't affect the other
|
||||
manager1.destroy()
|
||||
assert not workspace1.root.exists()
|
||||
assert workspace2.root.exists()
|
||||
Reference in New Issue
Block a user