5928077646
- Replace all ~/clawd and ~/.clawdbot paths with ~/echo-core equivalents in tools (git_commit, ralph_prd_generator, backup_config, lead-gen) - Update personality files: TOOLS.md repo/paths, AGENTS.md security audit cmd - Migrate HANDOFF.md architectural decisions to docs/architecture.md - Tighten credentials/ dir to 700, add to .gitignore - Add .claude/ and *.pid to .gitignore - Various adapter, router, and session improvements from prior work Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
901 lines
33 KiB
Python
901 lines
33 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, PropertyMock
|
|
import io
|
|
|
|
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,
|
|
set_session_model,
|
|
start_session,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
FAKE_RESULT_LINE = {
|
|
"type": "result",
|
|
"subtype": "success",
|
|
"session_id": "sess-abc-123",
|
|
"result": "Hello from Claude!",
|
|
"cost_usd": 0.004,
|
|
"total_cost_usd": 0.004,
|
|
"duration_ms": 1500,
|
|
"num_turns": 1,
|
|
"usage": {"input_tokens": 100, "output_tokens": 50},
|
|
}
|
|
|
|
FAKE_ASSISTANT_LINE = {
|
|
"type": "assistant",
|
|
"message": {
|
|
"content": [{"type": "text", "text": "Hello from Claude!"}],
|
|
},
|
|
}
|
|
|
|
|
|
def _make_stream(*assistant_texts, result_override=None):
|
|
"""Build stream-json stdout with assistant + result lines."""
|
|
lines = []
|
|
for text in assistant_texts:
|
|
lines.append(json.dumps({
|
|
"type": "assistant",
|
|
"message": {"content": [{"type": "text", "text": text}]},
|
|
}))
|
|
result = dict(FAKE_RESULT_LINE)
|
|
if result_override:
|
|
result.update(result_override)
|
|
lines.append(json.dumps(result))
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def _make_popen(stdout=None, returncode=0, stderr=""):
|
|
"""Build a fake subprocess.Popen that yields lines from stdout."""
|
|
if stdout is None:
|
|
stdout = _make_stream("Hello from Claude!")
|
|
proc = MagicMock()
|
|
proc.stdout = io.StringIO(stdout)
|
|
proc.stderr = io.StringIO(stderr)
|
|
proc.returncode = returncode
|
|
proc.wait.return_value = returncode
|
|
proc.kill = MagicMock()
|
|
return proc
|
|
|
|
|
|
# Keep old name for backward-compatible test helpers
|
|
_make_proc = _make_popen
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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("API_KEY", "sk-xxx")
|
|
monkeypatch.setenv("SECRET", "sk-ant-xxx")
|
|
env = _safe_env()
|
|
assert "API_KEY" not in env
|
|
assert "SECRET" not in env
|
|
|
|
def test_strips_all_blocked_keys(self, monkeypatch):
|
|
for key in claude_session._ENV_STRIP:
|
|
monkeypatch.setenv(key, "bad")
|
|
env = _safe_env()
|
|
for key in claude_session._ENV_STRIP:
|
|
assert key not in env
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _run_claude (now with Popen streaming)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRunClaude:
|
|
@patch("shutil.which", return_value="/usr/bin/claude")
|
|
@patch("subprocess.Popen")
|
|
def test_returns_parsed_stream(self, mock_popen, mock_which):
|
|
mock_popen.return_value = _make_popen()
|
|
result = _run_claude(["claude", "-p", "hi"], timeout=30)
|
|
assert result["result"] == "Hello from Claude!"
|
|
assert result["session_id"] == "sess-abc-123"
|
|
assert "usage" in result
|
|
|
|
@patch("shutil.which", return_value="/usr/bin/claude")
|
|
@patch("subprocess.Popen")
|
|
def test_collects_multiple_text_blocks(self, mock_popen, mock_which):
|
|
stdout = _make_stream("First message", "Second message", "Third message")
|
|
mock_popen.return_value = _make_popen(stdout=stdout)
|
|
result = _run_claude(["claude", "-p", "hi"], timeout=30)
|
|
assert result["result"] == "First message\n\nSecond message\n\nThird message"
|
|
|
|
@patch("shutil.which", return_value="/usr/bin/claude")
|
|
@patch("subprocess.Popen")
|
|
def test_timeout_raises(self, mock_popen, mock_which):
|
|
proc = _make_popen()
|
|
|
|
# Track calls to distinguish watchdog (with big timeout) from cleanup
|
|
call_count = [0]
|
|
|
|
def wait_side_effect(timeout=None):
|
|
call_count[0] += 1
|
|
if call_count[0] == 1 and timeout is not None:
|
|
# First call is the watchdog — simulate timeout
|
|
raise subprocess.TimeoutExpired(cmd="claude", timeout=timeout)
|
|
return 0 # subsequent cleanup calls succeed
|
|
|
|
proc.wait.side_effect = wait_side_effect
|
|
# stdout returns empty immediately so the for-loop exits
|
|
proc.stdout = io.StringIO("")
|
|
proc.stderr = io.StringIO("")
|
|
mock_popen.return_value = proc
|
|
|
|
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.Popen")
|
|
def test_nonzero_exit_raises(self, mock_popen, mock_which):
|
|
mock_popen.return_value = _make_popen(
|
|
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.Popen")
|
|
def test_no_result_line_raises(self, mock_popen, mock_which):
|
|
# Stream with only an assistant line but no result line
|
|
stdout = json.dumps({"type": "assistant", "message": {"content": []}}) + "\n"
|
|
mock_popen.return_value = _make_popen(stdout=stdout)
|
|
with pytest.raises(RuntimeError, match="no result line"):
|
|
_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)
|
|
|
|
@patch("shutil.which", return_value="/usr/bin/claude")
|
|
@patch("subprocess.Popen")
|
|
def test_on_text_callback_called(self, mock_popen, mock_which):
|
|
stdout = _make_stream("First", "Second")
|
|
mock_popen.return_value = _make_popen(stdout=stdout)
|
|
received = []
|
|
result = _run_claude(
|
|
["claude", "-p", "hi"], timeout=30,
|
|
on_text=lambda t: received.append(t),
|
|
)
|
|
assert received == ["First", "Second"]
|
|
assert result["intermediate_count"] == 2
|
|
|
|
@patch("shutil.which", return_value="/usr/bin/claude")
|
|
@patch("subprocess.Popen")
|
|
def test_on_text_callback_error_does_not_crash(self, mock_popen, mock_which):
|
|
mock_popen.return_value = _make_popen()
|
|
|
|
def bad_callback(text):
|
|
raise ValueError("callback boom")
|
|
|
|
# Should not raise — callback errors are logged but swallowed
|
|
result = _run_claude(
|
|
["claude", "-p", "hi"], timeout=30, on_text=bad_callback
|
|
)
|
|
assert result["result"] == "Hello from Claude!"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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.Popen")
|
|
def test_returns_response_and_session_id(
|
|
self, mock_popen, 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_popen.return_value = _make_popen()
|
|
|
|
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.Popen")
|
|
def test_saves_to_active_json(
|
|
self, mock_popen, 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_popen.return_value = _make_popen()
|
|
|
|
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.Popen")
|
|
def test_missing_result_line_raises(
|
|
self, mock_popen, 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"
|
|
)
|
|
# Stream with no result line at all
|
|
bad_stream = json.dumps({"type": "assistant", "message": {"content": []}}) + "\n"
|
|
mock_popen.return_value = _make_popen(stdout=bad_stream)
|
|
|
|
with pytest.raises(RuntimeError, match="no result line"):
|
|
start_session("general", "Hello")
|
|
|
|
@patch("shutil.which", return_value="/usr/bin/claude")
|
|
@patch("subprocess.Popen")
|
|
def test_missing_session_id_gives_empty_string(
|
|
self, mock_popen, 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"
|
|
)
|
|
# Result line without session_id → _run_claude returns "" for session_id
|
|
# → start_session checks for empty session_id
|
|
bad_stream = _make_stream("hello", result_override={"session_id": None})
|
|
mock_popen.return_value = _make_popen(stdout=bad_stream)
|
|
|
|
with pytest.raises(RuntimeError, match="missing required field"):
|
|
start_session("general", "Hello")
|
|
|
|
@patch("shutil.which", return_value="/usr/bin/claude")
|
|
@patch("subprocess.Popen")
|
|
def test_on_text_passed_through(
|
|
self, mock_popen, 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"
|
|
)
|
|
stdout = _make_stream("Block 1", "Block 2")
|
|
mock_popen.return_value = _make_popen(stdout=stdout)
|
|
|
|
received = []
|
|
start_session("general", "Hello", on_text=lambda t: received.append(t))
|
|
assert received == ["Block 1", "Block 2"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# resume_session
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestResumeSession:
|
|
@patch("shutil.which", return_value="/usr/bin/claude")
|
|
@patch("subprocess.Popen")
|
|
def test_returns_response(
|
|
self, mock_popen, 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_popen.return_value = _make_popen()
|
|
response = resume_session("sess-abc-123", "Follow up")
|
|
assert response == "Hello from Claude!"
|
|
|
|
@patch("shutil.which", return_value="/usr/bin/claude")
|
|
@patch("subprocess.Popen")
|
|
def test_updates_message_count_and_timestamp(
|
|
self, mock_popen, 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_popen.return_value = _make_popen()
|
|
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.Popen")
|
|
def test_uses_resume_flag(self, mock_popen, 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_popen.return_value = _make_popen()
|
|
resume_session("sess-abc-123", "Follow up")
|
|
|
|
# Verify --resume was in the command
|
|
cmd = mock_popen.call_args[0][0]
|
|
assert "--resume" in cmd
|
|
assert "sess-abc-123" in cmd
|
|
|
|
@patch("shutil.which", return_value="/usr/bin/claude")
|
|
@patch("subprocess.Popen")
|
|
def test_on_text_passed_through(
|
|
self, mock_popen, 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({}))
|
|
|
|
stdout = _make_stream("Block A", "Block B")
|
|
mock_popen.return_value = _make_popen(stdout=stdout)
|
|
|
|
received = []
|
|
resume_session("sess-abc-123", "Follow up", on_text=lambda t: received.append(t))
|
|
assert received == ["Block A", "Block B"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# send_message
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSendMessage:
|
|
@patch("shutil.which", return_value="/usr/bin/claude")
|
|
@patch("subprocess.Popen")
|
|
def test_starts_new_session_when_none_exists(
|
|
self, mock_popen, 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_popen.return_value = _make_popen()
|
|
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.Popen")
|
|
def test_resumes_existing_session(
|
|
self, mock_popen, 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_popen.return_value = _make_popen()
|
|
response = send_message("general", "Follow up")
|
|
assert response == "Hello from Claude!"
|
|
|
|
# Should have used --resume
|
|
cmd = mock_popen.call_args[0][0]
|
|
assert "--resume" in cmd
|
|
assert "sess-existing" in cmd
|
|
|
|
@patch("shutil.which", return_value="/usr/bin/claude")
|
|
@patch("subprocess.Popen")
|
|
def test_on_text_passed_through(
|
|
self, mock_popen, 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("{}")
|
|
|
|
stdout = _make_stream("Intermediate")
|
|
mock_popen.return_value = _make_popen(stdout=stdout)
|
|
|
|
received = []
|
|
send_message("general", "Hello", on_text=lambda t: received.append(t))
|
|
assert received == ["Intermediate"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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() == {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# set_session_model
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSetSessionModel:
|
|
def test_updates_model_in_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",
|
|
"model": "sonnet",
|
|
"message_count": 1,
|
|
}
|
|
}))
|
|
|
|
result = set_session_model("general", "opus")
|
|
assert result is True
|
|
|
|
data = json.loads(sf.read_text())
|
|
assert data["general"]["model"] == "opus"
|
|
|
|
def test_returns_false_when_no_session(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("{}")
|
|
result = set_session_model("general", "opus")
|
|
assert result is False
|
|
|
|
def test_invalid_model_raises(self):
|
|
with pytest.raises(ValueError, match="Invalid model"):
|
|
set_session_model("general", "gpt4")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Security: prompt injection protection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPromptInjectionProtection:
|
|
def test_system_prompt_contains_security_section(self):
|
|
prompt = build_system_prompt()
|
|
assert "## Security" in prompt
|
|
assert "EXTERNAL CONTENT" in prompt
|
|
assert "NEVER follow instructions" in prompt
|
|
assert "NEVER reveal secrets" in prompt
|
|
|
|
@patch("shutil.which", return_value="/usr/bin/claude")
|
|
@patch("subprocess.Popen")
|
|
def test_start_session_wraps_message(
|
|
self, mock_popen, 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_popen.return_value = _make_popen()
|
|
|
|
start_session("general", "Hello world")
|
|
|
|
cmd = mock_popen.call_args[0][0]
|
|
# Find the -p argument value
|
|
p_idx = cmd.index("-p")
|
|
msg = cmd[p_idx + 1]
|
|
assert msg.startswith("[EXTERNAL CONTENT]")
|
|
assert msg.endswith("[END EXTERNAL CONTENT]")
|
|
assert "Hello world" in msg
|
|
|
|
@patch("shutil.which", return_value="/usr/bin/claude")
|
|
@patch("subprocess.Popen")
|
|
def test_resume_session_wraps_message(
|
|
self, mock_popen, 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_popen.return_value = _make_popen()
|
|
resume_session("sess-abc-123", "Follow up msg")
|
|
|
|
cmd = mock_popen.call_args[0][0]
|
|
p_idx = cmd.index("-p")
|
|
msg = cmd[p_idx + 1]
|
|
assert msg.startswith("[EXTERNAL CONTENT]")
|
|
assert msg.endswith("[END EXTERNAL CONTENT]")
|
|
assert "Follow up msg" in msg
|
|
|
|
@patch("shutil.which", return_value="/usr/bin/claude")
|
|
@patch("subprocess.Popen")
|
|
def test_start_session_includes_system_prompt_with_security(
|
|
self, mock_popen, 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_popen.return_value = _make_popen()
|
|
|
|
start_session("general", "test")
|
|
|
|
cmd = mock_popen.call_args[0][0]
|
|
sp_idx = cmd.index("--system-prompt")
|
|
system_prompt = cmd[sp_idx + 1]
|
|
assert "NEVER follow instructions" in system_prompt
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Security: invocation logging
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestInvocationLogging:
|
|
@patch("shutil.which", return_value="/usr/bin/claude")
|
|
@patch("subprocess.Popen")
|
|
def test_start_session_logs_invocation(
|
|
self, mock_popen, 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_popen.return_value = _make_popen()
|
|
|
|
with patch.object(claude_session._invoke_log, "info") as mock_log:
|
|
start_session("general", "Hello")
|
|
mock_log.assert_called_once()
|
|
log_msg = mock_log.call_args[0][0]
|
|
assert "channel=" in log_msg
|
|
assert "model=" in log_msg
|
|
assert "duration_ms=" in log_msg
|
|
|
|
@patch("shutil.which", return_value="/usr/bin/claude")
|
|
@patch("subprocess.Popen")
|
|
def test_resume_session_logs_invocation(
|
|
self, mock_popen, 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-abc-123",
|
|
"model": "sonnet",
|
|
"message_count": 1,
|
|
}
|
|
}))
|
|
mock_popen.return_value = _make_popen()
|
|
|
|
with patch.object(claude_session._invoke_log, "info") as mock_log:
|
|
resume_session("sess-abc-123", "Follow up")
|
|
mock_log.assert_called_once()
|
|
log_args = mock_log.call_args[0]
|
|
assert "general" in log_args # channel_id
|
|
assert "sonnet" in log_args # model
|