"""Integration tests for MCP tools."""
import json
from pathlib import Path
from unittest.mock import patch
import pytest
from ccsession.server import (
_check_context_budget,
_get_session_state,
_get_session_history,
_sync_planning_doc,
_should_reset_context,
)
@pytest.mark.asyncio
async def test_check_context_budget(sample_transcript_path, monkeypatch):
"""Test check_context_budget tool."""
from ccsession.parsers.transcript import TranscriptParser
monkeypatch.setattr(
TranscriptParser,
"_discover_transcript",
lambda self: sample_transcript_path,
)
result = await _check_context_budget({"context_limit": 20000})
assert "tokens_used" in result
assert "tokens_remaining" in result
assert "percentage_used" in result
assert "context_limit" in result
assert "status" in result
assert result["tokens_used"] == 8600
assert result["tokens_remaining"] == 11400
assert result["context_limit"] == 20000
assert result["status"] in ["sufficient", "low", "critical"]
@pytest.mark.asyncio
async def test_get_session_state(
sample_transcript_path,
sample_todos_path,
mock_git_repo_with_changes,
sample_plan_path,
monkeypatch,
):
"""Test get_session_state tool."""
from ccsession.parsers.transcript import TranscriptParser
from ccsession.parsers.todos import TodoParser
todos_path, session_id = sample_todos_path
monkeypatch.setattr(
TranscriptParser,
"_discover_transcript",
lambda self: sample_transcript_path,
)
monkeypatch.setattr(
TodoParser,
"__init__",
lambda self, todos_dir=None: setattr(self, "todos_dir", todos_path.parent) or None,
)
result = await _get_session_state({"working_directory": str(mock_git_repo_with_changes)})
# Check structure
assert "todos" in result
assert "git" in result
assert "context_files" in result
assert "session" in result
# Check todos
assert "pending" in result["todos"]
assert "in_progress" in result["todos"]
assert "completed" in result["todos"]
# Check git
assert result["git"]["branch"] == "feat-auth"
assert result["git"]["has_uncommitted_changes"] is True
assert result["git"]["uncommitted_file_count"] == 1
# Check context files
assert result["context_files"]["exists"] is True
assert "feat-auth-detailed-plan.md" in result["context_files"]["plan_path"]
# Check session
assert result["session"]["session_id"] == session_id
assert result["session"]["start_time"] is not None
@pytest.mark.asyncio
async def test_get_session_history(
sample_transcript_path,
sample_todos_path,
mock_git_repo,
monkeypatch,
):
"""Test get_session_history tool."""
from ccsession.parsers.transcript import TranscriptParser
from ccsession.parsers.todos import TodoParser
todos_path, session_id = sample_todos_path
monkeypatch.setattr(
TranscriptParser,
"_discover_transcript",
lambda self: sample_transcript_path,
)
monkeypatch.setattr(
TodoParser,
"__init__",
lambda self, todos_dir=None: setattr(self, "todos_dir", todos_path.parent) or None,
)
result = await _get_session_history({"working_directory": str(mock_git_repo)})
# Check structure
assert "completed_todos" in result
assert "files_modified" in result
assert "tool_calls" in result
assert "git_commits" in result
# Check completed todos
assert len(result["completed_todos"]) == 2
assert "Implement authentication feature" in result["completed_todos"]
# Check files modified
assert "/tmp/test.py" in result["files_modified"]["created"]
assert "/tmp/test.py" in result["files_modified"]["edited"]
# Check tool calls
assert result["tool_calls"]["files_read"] >= 0
assert result["tool_calls"]["files_written"] >= 1
assert "Explore" in result["tool_calls"]["agents_spawned"]
assert any("git commit" in cmd for cmd in result["tool_calls"]["bash_commands"])
@pytest.mark.asyncio
async def test_sync_planning_doc_append_progress(mock_git_repo, sample_plan_path):
"""Test sync_planning_doc tool with append_progress_log mode."""
result = await _sync_planning_doc(
{
"mode": "append_progress_log",
"completed_tasks": ["Task A", "Task B"],
"decisions": ["Use bcrypt for hashing"],
"working_directory": str(mock_git_repo),
}
)
assert result["success"] is True
assert "Progress Log" in result["sections_updated"]
assert "feat-auth-detailed-plan.md" in result["plan_path"]
# Verify content was updated
plan_content = Path(result["plan_path"]).read_text()
assert "Task A" in plan_content
assert "Task B" in plan_content
assert "Use bcrypt for hashing" in plan_content
@pytest.mark.asyncio
async def test_sync_planning_doc_update_active_work(mock_git_repo, sample_plan_path):
"""Test sync_planning_doc tool with update_active_work mode."""
result = await _sync_planning_doc(
{
"mode": "update_active_work",
"in_progress": "Implementing user authentication",
"next_steps": ["Write unit tests", "Deploy to staging"],
"blockers": ["Waiting for API keys"],
"working_directory": str(mock_git_repo),
}
)
assert result["success"] is True
assert "Active Work" in result["sections_updated"]
# Verify content was updated
plan_content = Path(result["plan_path"]).read_text()
assert "Implementing user authentication" in plan_content
assert "Write unit tests" in plan_content
assert "Waiting for API keys" in plan_content
@pytest.mark.asyncio
async def test_sync_planning_doc_mark_tasks_complete(mock_git_repo, sample_plan_path):
"""Test sync_planning_doc tool with mark_tasks_complete mode."""
# First, ensure the plan has unchecked tasks
plan_path = mock_git_repo / ".context" / "dev" / "feat-auth" / "feat-auth-detailed-plan.md"
original_content = plan_path.read_text()
assert "- [ ] Implement authentication" in original_content
result = await _sync_planning_doc(
{
"mode": "mark_tasks_complete",
"completed_tasks": ["Implement authentication"],
"working_directory": str(mock_git_repo),
}
)
assert result["success"] is True
assert "Implementation Plan" in result["sections_updated"]
# Verify content was updated
plan_content = plan_path.read_text()
assert "- [x] Implement authentication" in plan_content
@pytest.mark.asyncio
async def test_sync_planning_doc_no_file(mock_git_repo):
"""Test sync_planning_doc when plan file doesn't exist."""
# Remove .context directory
result = await _sync_planning_doc(
{
"mode": "append_progress_log",
"completed_tasks": ["Task A"],
"working_directory": str(mock_git_repo),
}
)
assert result["success"] is False
assert "not found" in result["error"].lower()
@pytest.mark.asyncio
async def test_sync_planning_doc_missing_mode(mock_git_repo, sample_plan_path):
"""Test sync_planning_doc without mode parameter."""
result = await _sync_planning_doc({"working_directory": str(mock_git_repo)})
assert result["success"] is False
assert "mode is required" in result["error"]
@pytest.mark.asyncio
async def test_should_reset_context_critical(
sample_transcript_path,
sample_todos_path,
mock_git_repo,
monkeypatch,
):
"""Test should_reset_context with critical context and clean state."""
from ccsession.parsers.transcript import TranscriptParser, ContextBudget
from ccsession.parsers.todos import TodoParser, TodoList, TodoItem
todos_path, session_id = sample_todos_path
# Mock critical context usage
def mock_get_context_budget(self, context_limit=156000):
return ContextBudget(
tokens_used=130000,
tokens_remaining=26000,
percentage_used=83.3,
context_limit=156000,
status="critical",
)
# Mock all todos completed
def mock_get_todos_for_session(self, session_id):
return TodoList(
pending=[],
in_progress=[],
completed=[
TodoItem("Task 1", "completed", "Doing task 1"),
TodoItem("Task 2", "completed", "Doing task 2"),
],
)
monkeypatch.setattr(
TranscriptParser,
"_discover_transcript",
lambda self: sample_transcript_path,
)
monkeypatch.setattr(TranscriptParser, "get_context_budget", mock_get_context_budget)
monkeypatch.setattr(
TodoParser,
"__init__",
lambda self, todos_dir=None: setattr(self, "todos_dir", todos_path.parent) or None,
)
monkeypatch.setattr(TodoParser, "get_todos_for_session", mock_get_todos_for_session)
result = await _should_reset_context({"working_directory": str(mock_git_repo)})
assert result["should_reset"] is True
assert result["confidence"] in ["high", "medium"]
assert result["safe_to_reset"] is True
assert len(result["blockers"]) == 0
assert "critical" in " ".join(result["reasoning"]).lower()
@pytest.mark.asyncio
async def test_should_reset_context_with_uncommitted_changes(
sample_transcript_path,
sample_todos_path,
mock_git_repo_with_changes,
monkeypatch,
):
"""Test should_reset_context with uncommitted changes."""
from ccsession.parsers.transcript import TranscriptParser
from ccsession.parsers.todos import TodoParser
todos_path, session_id = sample_todos_path
monkeypatch.setattr(
TranscriptParser,
"_discover_transcript",
lambda self: sample_transcript_path,
)
monkeypatch.setattr(
TodoParser,
"__init__",
lambda self, todos_dir=None: setattr(self, "todos_dir", todos_path.parent) or None,
)
result = await _should_reset_context({"working_directory": str(mock_git_repo_with_changes)})
assert result["safe_to_reset"] is False
assert len(result["blockers"]) > 0
assert "uncommitted" in " ".join(result["blockers"]).lower()
@pytest.mark.asyncio
async def test_should_reset_context_low_usage(
sample_transcript_path,
sample_todos_path,
mock_git_repo,
monkeypatch,
):
"""Test should_reset_context with low context usage."""
from ccsession.parsers.transcript import TranscriptParser
from ccsession.parsers.todos import TodoParser
todos_path, session_id = sample_todos_path
monkeypatch.setattr(
TranscriptParser,
"_discover_transcript",
lambda self: sample_transcript_path,
)
monkeypatch.setattr(
TodoParser,
"__init__",
lambda self, todos_dir=None: setattr(self, "todos_dir", todos_path.parent) or None,
)
# With default fixtures, context usage is low (~40%)
result = await _should_reset_context({"working_directory": str(mock_git_repo)})
assert result["should_reset"] is False
assert result["confidence"] == "low"
assert "sufficient" in " ".join(result["reasoning"]).lower()
