Files
echo-core/tests/test_claude_session.py
MoltBot Service 339866baa1 stage-3: claude CLI wrapper with session management
Subprocess wrapper for Claude CLI with start/resume/clear sessions, personality system prompt, atomic session tracking. 38 new tests (89 total).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-13 12:12:07 +00:00

577 lines
20 KiB
Python

"""Comprehensive tests for src/claude_session.py."""
import json
import os
import subprocess
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from src import claude_session
from src.claude_session import (
_load_sessions,
_run_claude,
_safe_env,
_save_sessions,
build_system_prompt,
clear_session,
get_active_session,
list_sessions,
resume_session,
send_message,
start_session,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
FAKE_CLI_RESPONSE = {
"type": "result",
"subtype": "success",
"session_id": "sess-abc-123",
"result": "Hello from Claude!",
"cost_usd": 0.004,
"duration_ms": 1500,
"num_turns": 1,
}
def _make_proc(stdout=None, returncode=0, stderr=""):
"""Build a fake subprocess.CompletedProcess."""
if stdout is None:
stdout = json.dumps(FAKE_CLI_RESPONSE)
proc = MagicMock(spec=subprocess.CompletedProcess)
proc.stdout = stdout
proc.stderr = stderr
proc.returncode = returncode
return proc
# ---------------------------------------------------------------------------
# build_system_prompt
# ---------------------------------------------------------------------------
class TestBuildSystemPrompt:
def test_returns_non_empty_string(self):
prompt = build_system_prompt()
assert isinstance(prompt, str)
assert len(prompt) > 0
def test_contains_personality_content(self):
prompt = build_system_prompt()
# IDENTITY.md is first; should appear in the prompt
identity_path = claude_session.PERSONALITY_DIR / "IDENTITY.md"
identity_content = identity_path.read_text(encoding="utf-8")
assert identity_content in prompt
def test_correct_order(self):
prompt = build_system_prompt()
expected_order = [
"IDENTITY.md",
"SOUL.md",
"USER.md",
"AGENTS.md",
"TOOLS.md",
"HEARTBEAT.md",
]
# Read each file and find its position in the prompt
positions = []
for filename in expected_order:
filepath = claude_session.PERSONALITY_DIR / filename
if filepath.is_file():
content = filepath.read_text(encoding="utf-8")
pos = prompt.find(content)
assert pos >= 0, f"{filename} not found in prompt"
positions.append(pos)
# Positions must be strictly increasing
for i in range(1, len(positions)):
assert positions[i] > positions[i - 1], (
f"Order violation: {expected_order[i]} appears before "
f"{expected_order[i - 1]}"
)
def test_separator_between_files(self):
prompt = build_system_prompt()
assert "\n\n---\n\n" in prompt
def test_missing_personality_dir_raises(self, tmp_path, monkeypatch):
monkeypatch.setattr(
claude_session, "PERSONALITY_DIR", tmp_path / "nonexistent"
)
with pytest.raises(FileNotFoundError):
build_system_prompt()
# ---------------------------------------------------------------------------
# _safe_env
# ---------------------------------------------------------------------------
class TestSafeEnv:
def test_includes_path_and_home(self, monkeypatch):
monkeypatch.setenv("PATH", "/usr/bin")
monkeypatch.setenv("HOME", "/home/test")
env = _safe_env()
assert "PATH" in env
assert "HOME" in env
def test_excludes_discord_token(self, monkeypatch):
monkeypatch.setenv("DISCORD_TOKEN", "secret-token")
env = _safe_env()
assert "DISCORD_TOKEN" not in env
def test_excludes_claudecode(self, monkeypatch):
monkeypatch.setenv("CLAUDECODE", "1")
env = _safe_env()
assert "CLAUDECODE" not in env
def test_excludes_api_keys(self, monkeypatch):
monkeypatch.setenv("OPENAI_API_KEY", "sk-xxx")
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-xxx")
env = _safe_env()
assert "OPENAI_API_KEY" not in env
assert "ANTHROPIC_API_KEY" not in env
def test_only_passthrough_keys(self, monkeypatch):
monkeypatch.setenv("RANDOM_SECRET", "bad")
env = _safe_env()
for key in env:
assert key in claude_session._ENV_PASSTHROUGH
# ---------------------------------------------------------------------------
# _run_claude
# ---------------------------------------------------------------------------
class TestRunClaude:
@patch("shutil.which", return_value="/usr/bin/claude")
@patch("subprocess.run")
def test_returns_parsed_json(self, mock_run, mock_which):
mock_run.return_value = _make_proc()
result = _run_claude(["claude", "-p", "hi"], timeout=30)
assert result == FAKE_CLI_RESPONSE
@patch("shutil.which", return_value="/usr/bin/claude")
@patch("subprocess.run")
def test_timeout_raises(self, mock_run, mock_which):
mock_run.side_effect = subprocess.TimeoutExpired(cmd="claude", timeout=30)
with pytest.raises(TimeoutError, match="timed out after 30s"):
_run_claude(["claude", "-p", "hi"], timeout=30)
@patch("shutil.which", return_value="/usr/bin/claude")
@patch("subprocess.run")
def test_nonzero_exit_raises(self, mock_run, mock_which):
mock_run.return_value = _make_proc(
stdout="", returncode=1, stderr="something went wrong"
)
with pytest.raises(RuntimeError, match="exit 1"):
_run_claude(["claude", "-p", "hi"], timeout=30)
@patch("shutil.which", return_value="/usr/bin/claude")
@patch("subprocess.run")
def test_invalid_json_raises(self, mock_run, mock_which):
mock_run.return_value = _make_proc(stdout="not json {{{")
with pytest.raises(RuntimeError, match="Failed to parse"):
_run_claude(["claude", "-p", "hi"], timeout=30)
@patch("shutil.which", return_value=None)
def test_missing_binary_raises(self, mock_which):
with pytest.raises(FileNotFoundError, match="Claude CLI not found"):
_run_claude(["claude", "-p", "hi"], timeout=30)
# ---------------------------------------------------------------------------
# Session file helpers (_load_sessions / _save_sessions)
# ---------------------------------------------------------------------------
class TestSessionFileOps:
def test_load_missing_file_returns_empty(self, tmp_path, monkeypatch):
monkeypatch.setattr(
claude_session, "_SESSIONS_FILE", tmp_path / "no_such.json"
)
assert _load_sessions() == {}
def test_load_empty_file_returns_empty(self, tmp_path, monkeypatch):
f = tmp_path / "active.json"
f.write_text("")
monkeypatch.setattr(claude_session, "_SESSIONS_FILE", f)
assert _load_sessions() == {}
def test_save_and_load_roundtrip(self, tmp_path, monkeypatch):
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir()
sf = sessions_dir / "active.json"
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(claude_session, "_SESSIONS_FILE", sf)
data = {"general": {"session_id": "abc", "message_count": 1}}
_save_sessions(data)
loaded = _load_sessions()
assert loaded == data
def test_save_creates_directory(self, tmp_path, monkeypatch):
sessions_dir = tmp_path / "deep" / "sessions"
sf = sessions_dir / "active.json"
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(claude_session, "_SESSIONS_FILE", sf)
_save_sessions({"test": "data"})
assert sf.exists()
assert json.loads(sf.read_text()) == {"test": "data"}
def test_atomic_write_no_corruption_on_error(self, tmp_path, monkeypatch):
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir()
sf = sessions_dir / "active.json"
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(claude_session, "_SESSIONS_FILE", sf)
# Write initial good data
_save_sessions({"good": "data"})
# Attempt to write non-serializable data → should raise
class Unserializable:
pass
with pytest.raises(TypeError):
_save_sessions({"bad": Unserializable()})
# Original file should still be intact
assert json.loads(sf.read_text()) == {"good": "data"}
# ---------------------------------------------------------------------------
# start_session
# ---------------------------------------------------------------------------
class TestStartSession:
@patch("shutil.which", return_value="/usr/bin/claude")
@patch("subprocess.run")
def test_returns_response_and_session_id(
self, mock_run, mock_which, tmp_path, monkeypatch
):
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir()
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(
claude_session, "_SESSIONS_FILE", sessions_dir / "active.json"
)
mock_run.return_value = _make_proc()
response, sid = start_session("general", "Hello")
assert response == "Hello from Claude!"
assert sid == "sess-abc-123"
@patch("shutil.which", return_value="/usr/bin/claude")
@patch("subprocess.run")
def test_saves_to_active_json(
self, mock_run, mock_which, tmp_path, monkeypatch
):
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir()
sf = sessions_dir / "active.json"
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(claude_session, "_SESSIONS_FILE", sf)
mock_run.return_value = _make_proc()
start_session("general", "Hello")
data = json.loads(sf.read_text())
assert "general" in data
assert data["general"]["session_id"] == "sess-abc-123"
assert data["general"]["model"] == "sonnet"
assert data["general"]["message_count"] == 1
assert "created_at" in data["general"]
assert "last_message_at" in data["general"]
def test_invalid_model_raises(self):
with pytest.raises(ValueError, match="Invalid model"):
start_session("general", "Hello", model="gpt-4")
@patch("shutil.which", return_value="/usr/bin/claude")
@patch("subprocess.run")
def test_missing_result_field_raises(
self, mock_run, mock_which, tmp_path, monkeypatch
):
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir()
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(
claude_session, "_SESSIONS_FILE", sessions_dir / "active.json"
)
bad_response = {"session_id": "abc"} # missing "result"
mock_run.return_value = _make_proc(stdout=json.dumps(bad_response))
with pytest.raises(RuntimeError, match="missing required field"):
start_session("general", "Hello")
@patch("shutil.which", return_value="/usr/bin/claude")
@patch("subprocess.run")
def test_missing_session_id_field_raises(
self, mock_run, mock_which, tmp_path, monkeypatch
):
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir()
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(
claude_session, "_SESSIONS_FILE", sessions_dir / "active.json"
)
bad_response = {"result": "hello"} # missing "session_id"
mock_run.return_value = _make_proc(stdout=json.dumps(bad_response))
with pytest.raises(RuntimeError, match="missing required field"):
start_session("general", "Hello")
# ---------------------------------------------------------------------------
# resume_session
# ---------------------------------------------------------------------------
class TestResumeSession:
@patch("shutil.which", return_value="/usr/bin/claude")
@patch("subprocess.run")
def test_returns_response(
self, mock_run, mock_which, tmp_path, monkeypatch
):
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir()
sf = sessions_dir / "active.json"
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(claude_session, "_SESSIONS_FILE", sf)
# Pre-populate active.json
sf.write_text(json.dumps({
"general": {
"session_id": "sess-abc-123",
"model": "sonnet",
"created_at": "2026-01-01T00:00:00+00:00",
"last_message_at": "2026-01-01T00:00:00+00:00",
"message_count": 1,
}
}))
mock_run.return_value = _make_proc()
response = resume_session("sess-abc-123", "Follow up")
assert response == "Hello from Claude!"
@patch("shutil.which", return_value="/usr/bin/claude")
@patch("subprocess.run")
def test_updates_message_count_and_timestamp(
self, mock_run, mock_which, tmp_path, monkeypatch
):
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir()
sf = sessions_dir / "active.json"
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(claude_session, "_SESSIONS_FILE", sf)
old_ts = "2026-01-01T00:00:00+00:00"
sf.write_text(json.dumps({
"general": {
"session_id": "sess-abc-123",
"model": "sonnet",
"created_at": old_ts,
"last_message_at": old_ts,
"message_count": 3,
}
}))
mock_run.return_value = _make_proc()
resume_session("sess-abc-123", "Follow up")
data = json.loads(sf.read_text())
assert data["general"]["message_count"] == 4
assert data["general"]["last_message_at"] != old_ts
@patch("shutil.which", return_value="/usr/bin/claude")
@patch("subprocess.run")
def test_uses_resume_flag(self, mock_run, mock_which, tmp_path, monkeypatch):
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir()
sf = sessions_dir / "active.json"
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(claude_session, "_SESSIONS_FILE", sf)
sf.write_text(json.dumps({}))
mock_run.return_value = _make_proc()
resume_session("sess-abc-123", "Follow up")
# Verify --resume was in the command
cmd = mock_run.call_args[0][0]
assert "--resume" in cmd
assert "sess-abc-123" in cmd
# ---------------------------------------------------------------------------
# send_message
# ---------------------------------------------------------------------------
class TestSendMessage:
@patch("shutil.which", return_value="/usr/bin/claude")
@patch("subprocess.run")
def test_starts_new_session_when_none_exists(
self, mock_run, mock_which, tmp_path, monkeypatch
):
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir()
sf = sessions_dir / "active.json"
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(claude_session, "_SESSIONS_FILE", sf)
sf.write_text("{}")
mock_run.return_value = _make_proc()
response = send_message("general", "Hello")
assert response == "Hello from Claude!"
# Should have created a session
data = json.loads(sf.read_text())
assert "general" in data
@patch("shutil.which", return_value="/usr/bin/claude")
@patch("subprocess.run")
def test_resumes_existing_session(
self, mock_run, mock_which, tmp_path, monkeypatch
):
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir()
sf = sessions_dir / "active.json"
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(claude_session, "_SESSIONS_FILE", sf)
sf.write_text(json.dumps({
"general": {
"session_id": "sess-existing",
"model": "sonnet",
"created_at": "2026-01-01T00:00:00+00:00",
"last_message_at": "2026-01-01T00:00:00+00:00",
"message_count": 1,
}
}))
mock_run.return_value = _make_proc()
response = send_message("general", "Follow up")
assert response == "Hello from Claude!"
# Should have used --resume
cmd = mock_run.call_args[0][0]
assert "--resume" in cmd
assert "sess-existing" in cmd
# ---------------------------------------------------------------------------
# clear_session
# ---------------------------------------------------------------------------
class TestClearSession:
def test_returns_true_when_existed(self, tmp_path, monkeypatch):
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir()
sf = sessions_dir / "active.json"
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(claude_session, "_SESSIONS_FILE", sf)
sf.write_text(json.dumps({"general": {"session_id": "abc"}}))
assert clear_session("general") is True
def test_returns_false_when_not_existed(self, tmp_path, monkeypatch):
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir()
sf = sessions_dir / "active.json"
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(claude_session, "_SESSIONS_FILE", sf)
sf.write_text("{}")
assert clear_session("general") is False
def test_removes_from_active_json(self, tmp_path, monkeypatch):
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir()
sf = sessions_dir / "active.json"
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(claude_session, "_SESSIONS_FILE", sf)
sf.write_text(json.dumps({
"general": {"session_id": "abc"},
"other": {"session_id": "def"},
}))
clear_session("general")
data = json.loads(sf.read_text())
assert "general" not in data
assert "other" in data
# ---------------------------------------------------------------------------
# get_active_session
# ---------------------------------------------------------------------------
class TestGetActiveSession:
def test_returns_dict_when_exists(self, tmp_path, monkeypatch):
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir()
sf = sessions_dir / "active.json"
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(claude_session, "_SESSIONS_FILE", sf)
session_data = {"session_id": "abc", "model": "sonnet"}
sf.write_text(json.dumps({"general": session_data}))
result = get_active_session("general")
assert result == session_data
def test_returns_none_when_not_exists(self, tmp_path, monkeypatch):
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir()
sf = sessions_dir / "active.json"
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(claude_session, "_SESSIONS_FILE", sf)
sf.write_text("{}")
assert get_active_session("general") is None
def test_returns_none_when_file_missing(self, tmp_path, monkeypatch):
monkeypatch.setattr(
claude_session, "_SESSIONS_FILE", tmp_path / "nonexistent.json"
)
assert get_active_session("general") is None
# ---------------------------------------------------------------------------
# list_sessions
# ---------------------------------------------------------------------------
class TestListSessions:
def test_returns_all_sessions(self, tmp_path, monkeypatch):
sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir()
sf = sessions_dir / "active.json"
monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
monkeypatch.setattr(claude_session, "_SESSIONS_FILE", sf)
data = {
"general": {"session_id": "abc"},
"dev": {"session_id": "def"},
}
sf.write_text(json.dumps(data))
result = list_sessions()
assert result == data
def test_returns_empty_when_none(self, tmp_path, monkeypatch):
monkeypatch.setattr(
claude_session, "_SESSIONS_FILE", tmp_path / "nonexistent.json"
)
assert list_sessions() == {}