"""Tests for reset_context tool."""
import pytest
import json
import time
from pathlib import Path
import tempfile
import sys
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from amicus.core import read_with_lock, write_with_lock, get_state_file
def reset_context_impl(
state_file,
preserve_cluster_config: bool = False,
preserve_messages: bool = False,
confirm: bool = False
) -> str:
"""Implementation of reset_context for testing"""
if not confirm:
return "ERROR: This will clear all shared context. Call with confirm=True to proceed."
state_data = read_with_lock(state_file)
# Preserve selected data if requested
preserved_metadata = {}
preserved_messages = []
if preserve_cluster_config and "cluster_metadata" in state_data:
preserved_metadata = {
"max_agents": state_data["cluster_metadata"].get("max_agents"),
"active_agent_count": 0,
}
if preserve_messages and "messages" in state_data:
preserved_messages = state_data["messages"]
# Create fresh state
new_state = {
"summary": "",
"next_steps": [],
"active_files": [],
"messages": preserved_messages,
"last_heartbeat": time.time(),
"cluster_nodes": {},
"cluster_metadata": preserved_metadata if preserved_metadata else {
"max_agents": 5,
"active_agent_count": 0
},
"work_distribution": {},
"code_audits": {},
"file_intents": {}
}
write_with_lock(state_file, new_state)
preserved_items = []
if preserve_cluster_config:
preserved_items.append("cluster_config")
if preserve_messages:
preserved_items.append("messages")
preserved_note = f" (preserved: {', '.join(preserved_items)})" if preserved_items else ""
return f"✅ Context reset. All state cleared{preserved_note}."
@pytest.fixture
def temp_state_file(monkeypatch):
"""Create temporary state file for testing"""
with tempfile.TemporaryDirectory() as tmpdir:
state_file = Path(tmpdir) / "state.json"
# Mock get_state_file to return temp path
monkeypatch.setattr("amicus.server.get_state_file", lambda: state_file)
# Create initial state
initial_state = {
"summary": "Test summary",
"next_steps": [
{"task": "Task 1", "status": "pending"},
{"task": "Task 2", "status": "completed"}
],
"active_files": ["file1.py", "file2.py"],
"messages": ["Message 1", "Message 2"],
"last_heartbeat": time.time(),
"cluster_nodes": {
"node-1": {"role": "developer", "last_heartbeat": time.time()}
},
"cluster_metadata": {
"max_agents": 10,
"active_agent_count": 1
},
"work_distribution": {"node-1": ["task-1"]},
"code_audits": {"audit-1": {"status": "pending"}},
"file_intents": {"intent-1": {"files": ["test.py"]}}
}
write_with_lock(state_file, initial_state)
yield state_file
def test_reset_context_requires_confirmation(temp_state_file):
"""Test that reset requires confirm=True"""
result = reset_context_impl(temp_state_file)
assert "ERROR" in result
assert "confirm=True" in result
# State should be unchanged
state = read_with_lock(temp_state_file)
assert state["summary"] == "Test summary"
def test_reset_context_clears_all_state(temp_state_file):
"""Test that reset clears all state by default"""
result = reset_context_impl(temp_state_file, confirm=True)
assert "✅" in result
assert "cleared" in result
state = read_with_lock(temp_state_file)
# All fields should be cleared
assert state["summary"] == ""
assert state["next_steps"] == []
assert state["active_files"] == []
assert state["messages"] == []
assert state["cluster_nodes"] == {}
assert state["work_distribution"] == {}
assert state["code_audits"] == {}
assert state["file_intents"] == {}
# Cluster metadata should be reset to defaults
assert state["cluster_metadata"]["max_agents"] == 5
assert state["cluster_metadata"]["active_agent_count"] == 0
# Heartbeat should be recent
assert state["last_heartbeat"] > 0
def test_reset_context_preserve_cluster_config(temp_state_file):
"""Test preserving cluster configuration"""
result = reset_context_impl(temp_state_file, preserve_cluster_config=True, confirm=True)
assert "✅" in result
assert "cluster_config" in result
state = read_with_lock(temp_state_file)
# Cluster metadata should be preserved (except count)
assert state["cluster_metadata"]["max_agents"] == 10
assert state["cluster_metadata"]["active_agent_count"] == 0
# Other fields should be cleared
assert state["summary"] == ""
assert state["next_steps"] == []
def test_reset_context_preserve_messages(temp_state_file):
"""Test preserving message history"""
result = reset_context_impl(temp_state_file, preserve_messages=True, confirm=True)
assert "✅" in result
assert "messages" in result
state = read_with_lock(temp_state_file)
# Messages should be preserved
assert state["messages"] == ["Message 1", "Message 2"]
# Other fields should be cleared
assert state["summary"] == ""
assert state["next_steps"] == []
def test_reset_context_preserve_both(temp_state_file):
"""Test preserving both cluster config and messages"""
result = reset_context_impl(
temp_state_file,
preserve_cluster_config=True,
preserve_messages=True,
confirm=True
)
assert "✅" in result
assert "cluster_config" in result
assert "messages" in result
state = read_with_lock(temp_state_file)
# Both should be preserved
assert state["cluster_metadata"]["max_agents"] == 10
assert state["messages"] == ["Message 1", "Message 2"]
# Other fields should be cleared
assert state["summary"] == ""
assert state["next_steps"] == []
assert state["cluster_nodes"] == {}
def test_reset_context_creates_valid_state(temp_state_file):
"""Test that reset creates a valid, usable state"""
reset_context_impl(temp_state_file, confirm=True)
state = read_with_lock(temp_state_file)
# Verify all expected fields exist
assert "summary" in state
assert "next_steps" in state
assert "active_files" in state
assert "messages" in state
assert "last_heartbeat" in state
assert "cluster_nodes" in state
assert "cluster_metadata" in state
assert "work_distribution" in state
assert "code_audits" in state
assert "file_intents" in state
# Verify types
assert isinstance(state["summary"], str)
assert isinstance(state["next_steps"], list)
assert isinstance(state["active_files"], list)
assert isinstance(state["messages"], list)
assert isinstance(state["cluster_nodes"], dict)
assert isinstance(state["cluster_metadata"], dict)
if __name__ == '__main__':
pytest.main([__file__, '-v'])
