"""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"