"""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, set_session_model, 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("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 # --------------------------------------------------------------------------- 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() == {} # --------------------------------------------------------------------------- # 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")