[loop-cycle-52] test: comprehensive session_logger.py coverage (#175) (#187)

tests/timmy/test_session_logger.py

@@ -0,0 +1,511 @@
+"""Comprehensive unit tests for src/timmy/session_logger.py."""
+
+import json
+from datetime import date
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from timmy.session_logger import (
+    SessionLogger,
+    flush_session_logs,
+    get_session_logger,
+    get_session_summary,
+)
+
+
+@pytest.fixture(autouse=True)
+def reset_global_singleton(monkeypatch):
+    """Reset global singleton between tests."""
+    monkeypatch.setattr("timmy.session_logger._session_logger", None)
+
+
+@pytest.fixture
+def mock_settings(tmp_path):
+    """Mock config.settings.repo_root to use tmp_path."""
+    mock_settings = MagicMock()
+    mock_settings.repo_root = str(tmp_path)
+    # Patch at the source where settings is imported from
+    with patch("config.settings", mock_settings):
+        yield mock_settings
+
+
+class TestSessionLoggerInit:
+    """Tests for SessionLogger initialization."""
+
+    def test_init_with_custom_logs_dir(self, tmp_path):
+        """Should initialize with custom logs_dir using tmp_path."""
+        logs_dir = tmp_path / "custom_logs"
+        logger = SessionLogger(logs_dir=logs_dir)
+
+        assert logger.logs_dir == logs_dir
+        assert logs_dir.exists()
+
+    def test_init_default_logs_dir(self, mock_settings, tmp_path):
+        """Should default to /logs in repo root when logs_dir is None."""
+        logger = SessionLogger()
+
+        assert logger.logs_dir == tmp_path / "logs"
+        assert logger.logs_dir.exists()
+
+    def test_init_creates_session_file_path(self, tmp_path):
+        """Should create session file path with today's date."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        today_str = date.today().isoformat()
+        expected_path = tmp_path / f"session_{today_str}.jsonl"
+        assert logger.session_file == expected_path
+
+    def test_init_initializes_empty_buffer(self, tmp_path):
+        """Should initialize with empty buffer."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        assert logger._buffer == []
+
+
+class TestRecordMessage:
+    """Tests for record_message method."""
+
+    def test_record_message_without_confidence(self, tmp_path):
+        """Should record message without confidence."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        logger.record_message("user", "Hello Timmy")
+
+        assert len(logger._buffer) == 1
+        entry = logger._buffer[0]
+        assert entry["type"] == "message"
+        assert entry["role"] == "user"
+        assert entry["content"] == "Hello Timmy"
+        assert "timestamp" in entry
+        assert "confidence" not in entry
+
+    def test_record_message_with_confidence(self, tmp_path):
+        """Should record message with confidence."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        logger.record_message("timmy", "Hello user", confidence=0.95)
+
+        assert len(logger._buffer) == 1
+        entry = logger._buffer[0]
+        assert entry["type"] == "message"
+        assert entry["role"] == "timmy"
+        assert entry["content"] == "Hello user"
+        assert entry["confidence"] == 0.95
+
+    def test_record_message_multiple(self, tmp_path):
+        """Should record multiple messages."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        logger.record_message("user", "First message")
+        logger.record_message("timmy", "Second message", confidence=0.8)
+        logger.record_message("user", "Third message")
+
+        assert len(logger._buffer) == 3
+        assert logger._buffer[0]["role"] == "user"
+        assert logger._buffer[1]["role"] == "timmy"
+        assert logger._buffer[2]["role"] == "user"
+
+
+class TestRecordToolCall:
+    """Tests for record_tool_call method."""
+
+    def test_record_tool_call_basic(self, tmp_path):
+        """Should record basic tool call."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        logger.record_tool_call("read_file", {"path": "test.py"}, "file content")
+
+        assert len(logger._buffer) == 1
+        entry = logger._buffer[0]
+        assert entry["type"] == "tool_call"
+        assert entry["tool"] == "read_file"
+        assert entry["args"] == {"path": "test.py"}
+        assert entry["result"] == "file content"
+        assert "timestamp" in entry
+
+    def test_record_tool_call_result_truncation_over_500_chars(self, tmp_path):
+        """Should truncate result when over 500 characters."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        long_result = "a" * 600
+        logger.record_tool_call("search", {"query": "test"}, long_result)
+
+        entry = logger._buffer[0]
+        assert len(entry["result"]) == 500
+        assert entry["result"] == "a" * 500
+
+    def test_record_tool_call_result_exactly_500_chars(self, tmp_path):
+        """Should not truncate result when exactly 500 characters."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        exact_result = "b" * 500
+        logger.record_tool_call("read_file", {"path": "file.txt"}, exact_result)
+
+        entry = logger._buffer[0]
+        assert len(entry["result"]) == 500
+        assert entry["result"] == exact_result
+
+    def test_record_tool_call_result_under_500_chars(self, tmp_path):
+        """Should not truncate result when under 500 characters."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        short_result = "short result"
+        logger.record_tool_call("calc", {"expr": "1+1"}, short_result)
+
+        entry = logger._buffer[0]
+        assert entry["result"] == short_result
+
+    def test_record_tool_call_non_string_result(self, tmp_path):
+        """Should handle non-string result."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        logger.record_tool_call("calc", {"expr": "1+1"}, 42)
+
+        entry = logger._buffer[0]
+        assert entry["result"] == "42"
+
+    def test_record_tool_call_non_string_result_truncation(self, tmp_path):
+        """Should truncate non-string result over 500 chars."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        long_list = list(range(200))  # Will be longer than 500 chars as string
+        logger.record_tool_call("search", {"query": "test"}, long_list)
+
+        entry = logger._buffer[0]
+        assert len(entry["result"]) == 500
+
+
+class TestRecordError:
+    """Tests for record_error method."""
+
+    def test_record_error_without_context(self, tmp_path):
+        """Should record error without context."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        logger.record_error("File not found")
+
+        assert len(logger._buffer) == 1
+        entry = logger._buffer[0]
+        assert entry["type"] == "error"
+        assert entry["error"] == "File not found"
+        assert entry["context"] is None
+        assert "timestamp" in entry
+
+    def test_record_error_with_context(self, tmp_path):
+        """Should record error with context."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        logger.record_error("Connection failed", "Reading config file")
+
+        assert len(logger._buffer) == 1
+        entry = logger._buffer[0]
+        assert entry["type"] == "error"
+        assert entry["error"] == "Connection failed"
+        assert entry["context"] == "Reading config file"
+
+    def test_record_error_multiple(self, tmp_path):
+        """Should record multiple errors."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        logger.record_error("Error 1", "Context 1")
+        logger.record_error("Error 2")
+        logger.record_error("Error 3", "Context 3")
+
+        assert len(logger._buffer) == 3
+        assert logger._buffer[0]["error"] == "Error 1"
+        assert logger._buffer[1]["context"] is None
+        assert logger._buffer[2]["error"] == "Error 3"
+
+
+class TestRecordDecision:
+    """Tests for record_decision method."""
+
+    def test_record_decision_without_rationale(self, tmp_path):
+        """Should record decision without rationale."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        logger.record_decision("Use OOP pattern")
+
+        assert len(logger._buffer) == 1
+        entry = logger._buffer[0]
+        assert entry["type"] == "decision"
+        assert entry["decision"] == "Use OOP pattern"
+        assert entry["rationale"] is None
+        assert "timestamp" in entry
+
+    def test_record_decision_with_rationale(self, tmp_path):
+        """Should record decision with rationale."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        logger.record_decision("Refactor to classes", "More maintainable code")
+
+        assert len(logger._buffer) == 1
+        entry = logger._buffer[0]
+        assert entry["type"] == "decision"
+        assert entry["decision"] == "Refactor to classes"
+        assert entry["rationale"] == "More maintainable code"
+
+    def test_record_decision_multiple(self, tmp_path):
+        """Should record multiple decisions."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        logger.record_decision("Decision 1")
+        logger.record_decision("Decision 2", "Reason 2")
+        logger.record_decision("Decision 3", "Reason 3")
+
+        assert len(logger._buffer) == 3
+        assert logger._buffer[0]["rationale"] is None
+        assert logger._buffer[1]["rationale"] == "Reason 2"
+
+
+class TestFlush:
+    """Tests for flush method."""
+
+    def test_flush_writes_jsonl(self, tmp_path):
+        """Should write buffer to JSONL file."""
+        logger = SessionLogger(logs_dir=tmp_path)
+        logger.record_message("user", "Hello")
+        logger.record_tool_call("read", {"path": "test.txt"}, "content")
+
+        result_path = logger.flush()
+
+        assert result_path == logger.session_file
+        assert result_path.exists()
+
+        # Verify JSONL format
+        lines = result_path.read_text().strip().split("\n")
+        assert len(lines) == 2
+
+        # Verify each line is valid JSON
+        entry1 = json.loads(lines[0])
+        assert entry1["type"] == "message"
+        assert entry1["content"] == "Hello"
+
+        entry2 = json.loads(lines[1])
+        assert entry2["type"] == "tool_call"
+        assert entry2["tool"] == "read"
+
+    def test_flush_clears_buffer(self, tmp_path):
+        """Should clear buffer after flush."""
+        logger = SessionLogger(logs_dir=tmp_path)
+        logger.record_message("user", "Hello")
+        logger.record_error("Test error")
+
+        assert len(logger._buffer) == 2
+
+        logger.flush()
+
+        assert len(logger._buffer) == 0
+        assert logger._buffer == []
+
+    def test_flush_returns_path(self, tmp_path):
+        """Should return path to session file."""
+        logger = SessionLogger(logs_dir=tmp_path)
+        logger.record_message("user", "Hello")
+
+        result_path = logger.flush()
+
+        assert isinstance(result_path, Path)
+        assert result_path == logger.session_file
+
+    def test_flush_empty_buffer_returns_path_without_writing(self, tmp_path):
+        """Should return path without writing when buffer is empty."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        result_path = logger.flush()
+
+        assert isinstance(result_path, Path)
+        assert result_path == logger.session_file
+        # File should not exist since buffer was empty
+        assert not result_path.exists()
+
+    def test_flush_appends_to_existing_file(self, tmp_path):
+        """Should append to existing file on subsequent flushes."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        logger.record_message("user", "First")
+        logger.flush()
+
+        logger.record_message("user", "Second")
+        logger.record_error("Error")
+        logger.flush()
+
+        lines = logger.session_file.read_text().strip().split("\n")
+        assert len(lines) == 3
+
+        entries = [json.loads(line) for line in lines]
+        assert entries[0]["content"] == "First"
+        assert entries[1]["content"] == "Second"
+        assert entries[2]["type"] == "error"
+
+
+class TestGetSessionSummary:
+    """Tests for get_session_summary method."""
+
+    def test_get_session_summary_file_not_exist(self, tmp_path):
+        """Should return exists=False when file does not exist."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        summary = logger.get_session_summary()
+
+        assert summary["exists"] is False
+        assert summary["entries"] == 0
+
+    def test_get_session_summary_after_flush(self, tmp_path):
+        """Should return summary after flushing entries."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        logger.record_message("user", "Hello")
+        logger.record_message("timmy", "Hi")
+        logger.record_tool_call("read", {"path": "test.txt"}, "content")
+        logger.record_error("Test error")
+        logger.record_decision("Use pattern")
+
+        logger.flush()
+
+        summary = logger.get_session_summary()
+
+        assert summary["exists"] is True
+        assert summary["file"] == str(logger.session_file)
+        assert summary["entries"] == 5
+        assert summary["messages"] == 2
+        assert summary["tool_calls"] == 1
+        assert summary["errors"] == 1
+        assert summary["decisions"] == 1
+
+    def test_get_session_summary_empty_file(self, tmp_path):
+        """Should handle empty file."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        # Create empty file
+        logger.session_file.touch()
+
+        summary = logger.get_session_summary()
+
+        assert summary["exists"] is True
+        assert summary["entries"] == 0
+
+    def test_get_session_summary_with_blank_lines(self, tmp_path):
+        """Should handle file with blank lines."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        # Create file with blank lines
+        logger.session_file.write_text(
+            '{"type": "message", "content": "Hello"}\n\n{"type": "error", "error": "Oops"}\n'
+        )
+
+        summary = logger.get_session_summary()
+
+        assert summary["exists"] is True
+        assert summary["entries"] == 2
+
+
+class TestModuleLevelFunctions:
+    """Tests for module-level functions."""
+
+    def test_get_session_logger_returns_singleton(self, mock_settings, tmp_path):
+        """Should return same singleton instance."""
+        logger1 = get_session_logger()
+        logger2 = get_session_logger()
+
+        assert logger1 is logger2
+        assert isinstance(logger1, SessionLogger)
+
+    def test_get_session_logger_creates_new_when_none(self, mock_settings, tmp_path):
+        """Should create new logger when singleton is None."""
+        logger = get_session_logger()
+
+        assert isinstance(logger, SessionLogger)
+        assert logger.logs_dir == tmp_path / "logs"
+
+    def test_get_session_summary_module_function(self, mock_settings, tmp_path):
+        """Should get summary via module-level function."""
+        # First, create and flush some logs
+        logger = get_session_logger()
+        logger.record_message("user", "Test")
+        logger.flush()
+
+        # Call module-level function
+        summary = get_session_summary()
+
+        assert summary["exists"] is True
+        assert summary["entries"] >= 1
+        assert summary["messages"] >= 1
+
+    def test_flush_session_logs_module_function(self, mock_settings, tmp_path):
+        """Should flush logs via module-level function."""
+        logger = get_session_logger()
+        logger.record_message("user", "Test message")
+
+        # Call module-level function
+        result_path_str = flush_session_logs()
+
+        assert isinstance(result_path_str, str)
+        assert result_path_str == str(logger.session_file)
+        assert logger._buffer == []
+
+    def test_flush_session_logs_returns_string_path(self, mock_settings, tmp_path):
+        """Should return string path, not Path object."""
+        get_session_logger()
+
+        result = flush_session_logs()
+
+        assert isinstance(result, str)
+
+
+class TestIntegration:
+    """Integration tests for SessionLogger."""
+
+    def test_full_session_workflow(self, tmp_path):
+        """Should handle full session workflow."""
+        logger = SessionLogger(logs_dir=tmp_path)
+
+        # Simulate a session
+        logger.record_message("user", "What is 2+2?")
+        logger.record_decision("Use calculator tool", "User needs calculation")
+        logger.record_tool_call("calculate", {"expr": "2+2"}, "4")
+        logger.record_message("timmy", "2+2 is 4", confidence=0.99)
+
+        # Verify buffer state
+        assert len(logger._buffer) == 4
+
+        # Flush and verify
+        logger.flush()
+        assert len(logger._buffer) == 0
+
+        # Verify summary
+        summary = logger.get_session_summary()
+        assert summary["entries"] == 4
+        assert summary["messages"] == 2
+        assert summary["decisions"] == 1
+        assert summary["tool_calls"] == 1
+        assert summary["errors"] == 0
+
+        # Verify file content
+        lines = logger.session_file.read_text().strip().split("\n")
+        assert len(lines) == 4
+
+        # Load and verify structure
+        entries = [json.loads(line) for line in lines]
+        assert entries[0]["type"] == "message"
+        assert entries[1]["type"] == "decision"
+        assert entries[2]["type"] == "tool_call"
+        assert entries[3]["type"] == "message"
+
+    def test_logs_dir_created_as_string(self, tmp_path):
+        """Should accept logs_dir as string."""
+        logs_dir_str = str(tmp_path / "string_logs")
+        logger = SessionLogger(logs_dir=logs_dir_str)
+
+        assert logger.logs_dir == Path(logs_dir_str)
+        assert logger.logs_dir.exists()
+
+    def test_logs_dir_created_with_parents(self, tmp_path):
+        """Should create parent directories if needed."""
+        deep_logs_dir = tmp_path / "very" / "deep" / "logs"
+        logger = SessionLogger(logs_dir=deep_logs_dir)
+
+        assert logger.logs_dir == deep_logs_dir
+        assert deep_logs_dir.exists()