"""Tests for advanced_features module.
Tests advanced MCP tools for multi-project coordination, natural scheduling,
interruption management, and enhanced search.
Phase: Week 5 Day 2 - Advanced Features Coverage
"""
from __future__ import annotations
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
class TestAdvancedFeaturesHub:
"""Test AdvancedFeaturesHub coordinator class."""
def test_hub_initialization(self) -> None:
"""Should initialize with logger and feature flags."""
from session_buddy.advanced_features import AdvancedFeaturesHub
mock_logger = MagicMock()
hub = AdvancedFeaturesHub(mock_logger)
assert hub.logger == mock_logger
assert hub._multi_project_initialized is False
assert hub._advanced_search_initialized is False
assert hub._app_monitor_initialized is False
@pytest.mark.asyncio
async def test_initialize_multi_project_not_implemented(self) -> None:
"""Should raise NotImplementedError for multi-project init."""
from session_buddy.advanced_features import AdvancedFeaturesHub
hub = AdvancedFeaturesHub(MagicMock())
with pytest.raises(NotImplementedError):
await hub.initialize_multi_project()
@pytest.mark.asyncio
async def test_initialize_advanced_search_not_implemented(self) -> None:
"""Should raise NotImplementedError for advanced search init."""
from session_buddy.advanced_features import AdvancedFeaturesHub
hub = AdvancedFeaturesHub(MagicMock())
with pytest.raises(NotImplementedError):
await hub.initialize_advanced_search()
@pytest.mark.asyncio
async def test_initialize_app_monitor_not_implemented(self) -> None:
"""Should raise NotImplementedError for app monitor init."""
from session_buddy.advanced_features import AdvancedFeaturesHub
hub = AdvancedFeaturesHub(MagicMock())
with pytest.raises(NotImplementedError):
await hub.initialize_app_monitor()
class TestNaturalReminderTools:
"""Test natural language reminder MCP tools."""
@pytest.mark.asyncio
async def test_create_natural_reminder_success(self) -> None:
"""Should create reminder and return formatted output."""
from session_buddy.advanced_features import create_natural_reminder
with patch(
"session_buddy.natural_scheduler.create_natural_reminder"
) as mock_create:
mock_create.return_value = "reminder-123"
result = await create_natural_reminder(
title="Test reminder",
time_expression="in 30 minutes",
description="Test description",
)
assert isinstance(result, str)
assert "successfully" in result or "✅" in result or "⏰" in result
assert "reminder-123" in result
@pytest.mark.asyncio
async def test_create_natural_reminder_handles_import_error(self) -> None:
"""Should handle missing dependencies gracefully."""
from session_buddy.advanced_features import create_natural_reminder
# Mock the import to raise ImportError
with patch("builtins.__import__", side_effect=ImportError):
result = await create_natural_reminder(
title="Test", time_expression="in 1 hour"
)
assert "not available" in result or "❌" in result
@pytest.mark.asyncio
async def test_list_user_reminders_with_reminders(self) -> None:
"""Should list user reminders."""
from session_buddy.advanced_features import list_user_reminders
with patch("session_buddy.natural_scheduler.list_user_reminders") as mock_list:
mock_list.return_value = [{"id": "1", "title": "Reminder 1"}]
with patch(
"session_buddy.utils.server_helpers._format_reminders_list"
) as mock_format:
mock_format.return_value = ["Formatted output"]
result = await list_user_reminders(user_id="test-user")
assert isinstance(result, str)
@pytest.mark.asyncio
async def test_cancel_user_reminder_success(self) -> None:
"""Should cancel reminder and return confirmation."""
from session_buddy.advanced_features import cancel_user_reminder
with patch(
"session_buddy.natural_scheduler.cancel_user_reminder"
) as mock_cancel:
mock_cancel.return_value = True
result = await cancel_user_reminder(reminder_id="reminder-123")
assert isinstance(result, str)
assert "cancelled" in result or "❌" in result
@pytest.mark.asyncio
async def test_start_reminder_service_success(self) -> None:
"""Should start reminder service."""
from session_buddy.advanced_features import start_reminder_service
with patch("session_buddy.natural_scheduler.register_session_notifications"):
with patch("session_buddy.natural_scheduler.start_reminder_service"):
result = await start_reminder_service()
assert isinstance(result, str)
assert "started" in result or "🚀" in result
@pytest.mark.asyncio
async def test_stop_reminder_service_success(self) -> None:
"""Should stop reminder service."""
from session_buddy.advanced_features import stop_reminder_service
with patch("session_buddy.natural_scheduler.stop_reminder_service"):
result = await stop_reminder_service()
assert isinstance(result, str)
assert "stopped" in result or "🛑" in result
class TestInterruptionManagement:
"""Test interruption management tools."""
@pytest.mark.asyncio
async def test_get_interruption_statistics_with_data(self) -> None:
"""Should return formatted interruption statistics."""
from session_buddy.advanced_features import get_interruption_statistics
# Test the import error path since interruption_manager is optional
with patch("builtins.__import__", side_effect=ImportError):
result = await get_interruption_statistics(user_id="test-user")
assert isinstance(result, str)
assert "not available" in result or "❌" in result
class TestMultiProjectCoordination:
"""Test multi-project coordination tools."""
@pytest.mark.asyncio
async def test_create_project_group_success(self) -> None:
"""Should create project group and return formatted output."""
from session_buddy.advanced_features import create_project_group
mock_group = MagicMock()
mock_group.name = "Test Group"
mock_group.projects = ["project1", "project2"]
mock_group.description = "Test description"
mock_group.id = "group-123"
with patch(
"session_buddy.advanced_features._get_multi_project_coordinator"
) as mock_get:
mock_coordinator = AsyncMock()
mock_coordinator.create_project_group = AsyncMock(return_value=mock_group)
mock_get.return_value = mock_coordinator
result = await create_project_group(
name="Test Group", projects=["project1", "project2"]
)
assert isinstance(result, str)
assert "Created" in result or "✅" in result
assert "Test Group" in result
@pytest.mark.asyncio
async def test_add_project_dependency_success(self) -> None:
"""Should add project dependency."""
from session_buddy.advanced_features import add_project_dependency
mock_dependency = MagicMock()
mock_dependency.source_project = "project1"
mock_dependency.target_project = "project2"
mock_dependency.dependency_type = "uses"
mock_dependency.description = "Uses API"
with patch(
"session_buddy.advanced_features._get_multi_project_coordinator"
) as mock_get:
mock_coordinator = AsyncMock()
mock_coordinator.add_project_dependency = AsyncMock(
return_value=mock_dependency
)
mock_get.return_value = mock_coordinator
result = await add_project_dependency(
source_project="project1",
target_project="project2",
dependency_type="uses",
)
assert isinstance(result, str)
assert "Added" in result or "✅" in result
@pytest.mark.asyncio
async def test_search_across_projects_with_results(self) -> None:
"""Should search across related projects."""
from session_buddy.advanced_features import search_across_projects
mock_results = [
{
"content": "Test content",
"score": 0.95,
"is_current_project": True,
"source_project": "project1",
"timestamp": "2025-01-01",
}
]
with patch(
"session_buddy.advanced_features._get_multi_project_coordinator"
) as mock_get:
mock_coordinator = AsyncMock()
mock_coordinator.find_related_conversations = AsyncMock(
return_value=mock_results
)
mock_get.return_value = mock_coordinator
result = await search_across_projects(
query="test", current_project="project1"
)
assert isinstance(result, str)
assert "Test content" in result
@pytest.mark.asyncio
async def test_get_project_insights_success(self) -> None:
"""Should get project insights."""
from session_buddy.advanced_features import get_project_insights
# Test unavailable coordinator path
with patch(
"session_buddy.advanced_features._get_multi_project_coordinator"
) as mock_get:
mock_get.return_value = None
result = await get_project_insights(projects=["project1", "project2"])
assert isinstance(result, str)
assert "not available" in result or "❌" in result
class TestAdvancedSearch:
"""Test advanced search tools."""
@pytest.mark.asyncio
async def test_advanced_search_with_results(self) -> None:
"""Should perform advanced search with filters."""
from session_buddy.advanced_features import advanced_search
# Test unavailable search engine path
with patch(
"session_buddy.advanced_features._get_advanced_search_engine"
) as mock_get:
mock_get.return_value = None
result = await advanced_search(query="test")
assert isinstance(result, str)
assert "not available" in result or "❌" in result
@pytest.mark.asyncio
async def test_search_suggestions_success(self) -> None:
"""Should return search suggestions."""
from session_buddy.advanced_features import search_suggestions
mock_suggestions = ["suggestion1", "suggestion2"]
with patch(
"session_buddy.advanced_features._get_advanced_search_engine"
) as mock_get:
mock_engine = AsyncMock()
mock_engine.get_suggestions = AsyncMock(return_value=mock_suggestions)
mock_get.return_value = mock_engine
result = await search_suggestions(query="test")
assert isinstance(result, str)
@pytest.mark.asyncio
async def test_get_search_metrics_success(self) -> None:
"""Should return search metrics."""
from session_buddy.advanced_features import get_search_metrics
mock_metrics = {"total_searches": 100, "avg_results": 5}
with patch(
"session_buddy.advanced_features._get_advanced_search_engine"
) as mock_get:
mock_engine = AsyncMock()
mock_engine.get_metrics = AsyncMock(return_value=mock_metrics)
mock_get.return_value = mock_engine
result = await get_search_metrics(metric_type="searches")
assert isinstance(result, str)
class TestGitWorktreeManagement:
"""Test git worktree management tools."""
@pytest.mark.asyncio
async def test_git_worktree_add_success(self) -> None:
"""Should add git worktree."""
from session_buddy.advanced_features import git_worktree_add
# Mock WorktreeManager where it's imported from
with patch(
"session_buddy.worktree_manager.WorktreeManager"
) as mock_manager_cls:
mock_manager = AsyncMock()
mock_manager.create_worktree = AsyncMock(
return_value={
"success": True,
"branch": "feature",
"worktree_path": "/tmp/worktree",
"output": "Created worktree",
}
)
mock_manager_cls.return_value = mock_manager
result = await git_worktree_add(branch="feature", path="/tmp/worktree")
assert isinstance(result, str)
assert "🎉" in result or "Created" in result
@pytest.mark.asyncio
async def test_git_worktree_remove_success(self) -> None:
"""Should remove git worktree."""
from session_buddy.advanced_features import git_worktree_remove
# Mock WorktreeManager where it's imported from
with patch(
"session_buddy.worktree_manager.WorktreeManager"
) as mock_manager_cls:
mock_manager = AsyncMock()
mock_manager.remove_worktree = AsyncMock(
return_value={
"success": True,
"removed_path": "/tmp/worktree",
"output": "Removed worktree",
}
)
mock_manager_cls.return_value = mock_manager
result = await git_worktree_remove(path="/tmp/worktree")
assert isinstance(result, str)
assert "🗑️" in result or "Removed" in result
@pytest.mark.asyncio
async def test_git_worktree_switch_success(self) -> None:
"""Should switch between worktrees."""
from session_buddy.advanced_features import git_worktree_switch
# Mock WorktreeManager where it's imported from
with patch(
"session_buddy.worktree_manager.WorktreeManager"
) as mock_manager_cls:
mock_manager = AsyncMock()
mock_manager.switch_worktree_context = AsyncMock(
return_value={
"success": True,
"from_worktree": {"branch": "main", "path": "/tmp/wt1"},
"to_worktree": {"branch": "feature", "path": "/tmp/wt2"},
"context_preserved": True,
}
)
mock_manager_cls.return_value = mock_manager
result = await git_worktree_switch(from_path="/tmp/wt1", to_path="/tmp/wt2")
assert isinstance(result, str)
assert "Switch" in result or "Complete" in result
class TestSessionWelcome:
"""Test session welcome tool."""
@pytest.mark.asyncio
async def test_session_welcome_returns_formatted_message(self) -> None:
"""Should return session welcome message."""
from session_buddy.advanced_features import session_welcome
# Test basic functionality without setting connection info
result = await session_welcome()
assert isinstance(result, str)
# Should return some welcome-related content
assert len(result) > 0
class TestHelperFunctions:
"""Test utility and helper functions."""
def test_calculate_overdue_time_for_overdue_reminder(self) -> None:
"""Should calculate overdue time."""
from session_buddy.advanced_features import _calculate_overdue_time
# Use a past timestamp
past_time = "2020-01-01T12:00:00"
result = _calculate_overdue_time(past_time)
assert isinstance(result, str)
assert "Overdue" in result or "⏱️" in result
def test_format_session_statistics_with_data(self) -> None:
"""Should format session statistics."""
from session_buddy.advanced_features import _format_session_statistics
sessions = {
"total_sessions": 10,
"active_sessions": 3,
"avg_duration": "1h 30m",
}
result = _format_session_statistics(sessions)
assert isinstance(result, list)
assert len(result) > 0
def test_has_statistics_data_with_data(self) -> None:
"""Should detect presence of statistics data."""
from session_buddy.advanced_features import _has_statistics_data
assert _has_statistics_data({"total": 1}, {}, {}) is True
assert _has_statistics_data({}, {"count": 1}, {}) is True
assert _has_statistics_data({}, {}, {"snapshots": 1}) is True
def test_has_statistics_data_without_data(self) -> None:
"""Should detect absence of statistics data."""
from session_buddy.advanced_features import _has_statistics_data
assert _has_statistics_data({}, {}, {}) is False
assert _has_statistics_data(None, None, None) is False
def test_build_advanced_search_filters_creates_dict(self) -> None:
"""Should build search filters dictionary."""
# _build_advanced_search_filters is defined inside advanced_search function
# Test that it exists and can be called
from session_buddy import advanced_features
# Verify the module has the expected structure
assert hasattr(advanced_features, "advanced_search")
assert callable(advanced_features.advanced_search)
