Test session management functionality.
| 408 | |
| 409 | |
| 410 | class TestSessionManagement: |
| 411 | """Test session management functionality.""" |
| 412 | |
| 413 | @pytest.fixture |
| 414 | def mock_session_environment(self): |
| 415 | """Fixture providing mocked session management environment.""" |
| 416 | client = MockMsfRpcClient() |
| 417 | session = Mock() |
| 418 | session.run_with_output = Mock(return_value="command output") |
| 419 | session.read = Mock(return_value="session data") |
| 420 | session.write = Mock() |
| 421 | session.stop = Mock() |
| 422 | |
| 423 | # Override the default Mock with actual dict return values |
| 424 | client.sessions.list = Mock(return_value={ |
| 425 | "1": {"type": "meterpreter", "info": "Windows session"}, |
| 426 | "2": {"type": "shell", "info": "Linux session"} |
| 427 | }) |
| 428 | client.sessions.session = Mock(return_value=session) |
| 429 | |
| 430 | with patch('MetasploitMCP.get_msf_client', return_value=client): |
| 431 | yield client, session |
| 432 | |
| 433 | @pytest.mark.asyncio |
| 434 | async def test_list_active_sessions(self, mock_session_environment): |
| 435 | """Test listing active sessions.""" |
| 436 | client, session = mock_session_environment |
| 437 | |
| 438 | result = await list_active_sessions() |
| 439 | |
| 440 | assert result["status"] == "success" |
| 441 | assert result["count"] == 2 |
| 442 | assert "1" in result["sessions"] |
| 443 | assert "2" in result["sessions"] |
| 444 | |
| 445 | @pytest.mark.asyncio |
| 446 | async def test_send_session_command_meterpreter(self, mock_session_environment): |
| 447 | """Test sending command to Meterpreter session.""" |
| 448 | client, session = mock_session_environment |
| 449 | |
| 450 | result = await send_session_command(1, "sysinfo") |
| 451 | |
| 452 | assert result["status"] == "success" |
| 453 | session.run_with_output.assert_called_once_with("sysinfo") |
| 454 | |
| 455 | @pytest.mark.asyncio |
| 456 | async def test_send_session_command_nonexistent(self, mock_session_environment): |
| 457 | """Test sending command to non-existent session.""" |
| 458 | client, session = mock_session_environment |
| 459 | client.sessions.list.return_value = {} # No sessions |
| 460 | |
| 461 | result = await send_session_command(999, "whoami") |
| 462 | |
| 463 | assert result["status"] == "error" |
| 464 | assert "not found" in result["message"] |
| 465 | |
| 466 | @pytest.mark.asyncio |
| 467 | async def test_terminate_session(self, mock_session_environment): |
nothing calls this directly
no outgoing calls
no test coverage detected