From 0d15787c7a6b35c5381e2db49ee201fb628b4b67 Mon Sep 17 00:00:00 2001 From: teernisse Date: Thu, 26 Feb 2026 15:23:42 -0500 Subject: [PATCH] test(server): add unit tests for conversation mtime tracking Comprehensive test coverage for _get_conversation_mtime() and its integration with _collect_sessions(). Test cases: - Claude sessions: file exists, file missing, OSError on stat, missing project_dir, missing session_id - Codex sessions: transcript_path exists, transcript_path missing, discovery fallback, discovery returns None, OSError on stat - Edge cases: unknown agent type, missing agent key - Integration: mtime included when file exists, omitted when missing - Change detection: mtime changes trigger payload hash changes Uses mock patching to isolate file system interactions and test error handling paths without requiring actual conversation files. Co-Authored-By: Claude Opus 4.5 --- tests/test_conversation_mtime.py | 357 +++++++++++++++++++++++++++++++ 1 file changed, 357 insertions(+) create mode 100644 tests/test_conversation_mtime.py diff --git a/tests/test_conversation_mtime.py b/tests/test_conversation_mtime.py new file mode 100644 index 0000000..a345cca --- /dev/null +++ b/tests/test_conversation_mtime.py @@ -0,0 +1,357 @@ +"""Tests for conversation_mtime_ns feature in state.py. + +This feature enables real-time dashboard updates by tracking the conversation +file's modification time, which changes on every write (tool call, message, etc.), +rather than relying solely on hook events which only fire at specific moments. +""" + +import json +import tempfile +import time +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from amc_server.mixins.state import StateMixin +from amc_server.mixins.parsing import SessionParsingMixin +from amc_server.mixins.discovery import SessionDiscoveryMixin + + +class CombinedMixin(StateMixin, SessionParsingMixin, SessionDiscoveryMixin): + """Combined mixin for testing - mirrors AMCHandler's inheritance.""" + pass + + +class TestGetConversationMtime(unittest.TestCase): + """Tests for _get_conversation_mtime method.""" + + def setUp(self): + self.handler = CombinedMixin() + self.temp_dir = tempfile.mkdtemp() + + def tearDown(self): + import shutil + shutil.rmtree(self.temp_dir, ignore_errors=True) + + def test_claude_session_with_existing_file(self): + """When conversation file exists, returns its mtime_ns.""" + # Create a temp conversation file + conv_file = Path(self.temp_dir) / "test-session.jsonl" + conv_file.write_text('{"type": "user"}\n') + expected_mtime = conv_file.stat().st_mtime_ns + + session_data = { + "agent": "claude", + "session_id": "test-session", + "project_dir": "/some/project", + } + + with patch.object( + self.handler, "_get_claude_conversation_file", return_value=conv_file + ): + result = self.handler._get_conversation_mtime(session_data) + + self.assertEqual(result, expected_mtime) + + def test_claude_session_file_not_found(self): + """When _get_claude_conversation_file returns None, returns None.""" + session_data = { + "agent": "claude", + "session_id": "nonexistent", + "project_dir": "/some/project", + } + + with patch.object( + self.handler, "_get_claude_conversation_file", return_value=None + ): + result = self.handler._get_conversation_mtime(session_data) + + self.assertIsNone(result) + + def test_claude_session_oserror_on_stat(self): + """When stat() raises OSError, returns None gracefully.""" + mock_file = MagicMock() + mock_file.stat.side_effect = OSError("Permission denied") + + session_data = { + "agent": "claude", + "session_id": "test-session", + "project_dir": "/some/project", + } + + with patch.object( + self.handler, "_get_claude_conversation_file", return_value=mock_file + ): + result = self.handler._get_conversation_mtime(session_data) + + self.assertIsNone(result) + + def test_claude_session_missing_project_dir(self): + """When project_dir is empty, _get_claude_conversation_file returns None.""" + session_data = { + "agent": "claude", + "session_id": "test-session", + "project_dir": "", + } + + # Real method will return None for empty project_dir + result = self.handler._get_conversation_mtime(session_data) + self.assertIsNone(result) + + def test_claude_session_missing_session_id(self): + """When session_id is empty, returns None.""" + session_data = { + "agent": "claude", + "session_id": "", + "project_dir": "/some/project", + } + + # _get_claude_conversation_file needs both session_id and project_dir + with patch.object( + self.handler, "_get_claude_conversation_file", return_value=None + ): + result = self.handler._get_conversation_mtime(session_data) + + self.assertIsNone(result) + + def test_codex_session_with_transcript_path(self): + """When transcript_path is provided and exists, returns its mtime_ns.""" + transcript_file = Path(self.temp_dir) / "codex-transcript.jsonl" + transcript_file.write_text('{"type": "response_item"}\n') + expected_mtime = transcript_file.stat().st_mtime_ns + + session_data = { + "agent": "codex", + "session_id": "codex-123", + "transcript_path": str(transcript_file), + } + + result = self.handler._get_conversation_mtime(session_data) + self.assertEqual(result, expected_mtime) + + def test_codex_session_transcript_path_missing_file(self): + """When transcript_path points to nonexistent file, falls back to discovery.""" + session_data = { + "agent": "codex", + "session_id": "codex-123", + "transcript_path": "/nonexistent/path.jsonl", + } + + # Mock the discovery fallback + with patch.object( + self.handler, "_find_codex_transcript_file", return_value=None + ): + result = self.handler._get_conversation_mtime(session_data) + + self.assertIsNone(result) + + def test_codex_session_discovery_fallback(self): + """When transcript_path not provided, uses _find_codex_transcript_file.""" + transcript_file = Path(self.temp_dir) / "discovered-transcript.jsonl" + transcript_file.write_text('{"type": "response_item"}\n') + expected_mtime = transcript_file.stat().st_mtime_ns + + session_data = { + "agent": "codex", + "session_id": "codex-456", + # No transcript_path + } + + with patch.object( + self.handler, "_find_codex_transcript_file", return_value=transcript_file + ): + result = self.handler._get_conversation_mtime(session_data) + + self.assertEqual(result, expected_mtime) + + def test_codex_session_discovery_returns_none(self): + """When discovery finds nothing, returns None.""" + session_data = { + "agent": "codex", + "session_id": "codex-789", + } + + with patch.object( + self.handler, "_find_codex_transcript_file", return_value=None + ): + result = self.handler._get_conversation_mtime(session_data) + + self.assertIsNone(result) + + def test_codex_session_oserror_on_transcript_stat(self): + """When stat() on discovered transcript raises OSError, returns None.""" + mock_file = MagicMock() + mock_file.stat.side_effect = OSError("I/O error") + + session_data = { + "agent": "codex", + "session_id": "codex-err", + } + + with patch.object( + self.handler, "_find_codex_transcript_file", return_value=mock_file + ): + result = self.handler._get_conversation_mtime(session_data) + + self.assertIsNone(result) + + def test_unknown_agent_returns_none(self): + """When agent is neither 'claude' nor 'codex', returns None.""" + session_data = { + "agent": "unknown_agent", + "session_id": "test-123", + "project_dir": "/some/project", + } + + result = self.handler._get_conversation_mtime(session_data) + self.assertIsNone(result) + + def test_missing_agent_returns_none(self): + """When agent key is missing, returns None.""" + session_data = { + "session_id": "test-123", + "project_dir": "/some/project", + } + + result = self.handler._get_conversation_mtime(session_data) + self.assertIsNone(result) + + def test_mtime_changes_on_file_modification(self): + """Verify mtime actually changes when file is modified.""" + conv_file = Path(self.temp_dir) / "changing-file.jsonl" + conv_file.write_text('{"type": "user"}\n') + mtime_1 = conv_file.stat().st_mtime_ns + + # Small delay to ensure filesystem mtime granularity is captured + time.sleep(0.01) + conv_file.write_text('{"type": "user"}\n{"type": "assistant"}\n') + mtime_2 = conv_file.stat().st_mtime_ns + + session_data = { + "agent": "claude", + "session_id": "test-session", + "project_dir": "/some/project", + } + + with patch.object( + self.handler, "_get_claude_conversation_file", return_value=conv_file + ): + result = self.handler._get_conversation_mtime(session_data) + + self.assertEqual(result, mtime_2) + self.assertNotEqual(mtime_1, mtime_2) + + +class TestCollectSessionsIntegration(unittest.TestCase): + """Integration tests verifying conversation_mtime_ns is included in session data.""" + + def setUp(self): + self.temp_dir = tempfile.mkdtemp() + self.sessions_dir = Path(self.temp_dir) / "sessions" + self.sessions_dir.mkdir() + + def tearDown(self): + import shutil + shutil.rmtree(self.temp_dir, ignore_errors=True) + + def test_collect_sessions_includes_mtime_when_available(self): + """_collect_sessions adds conversation_mtime_ns when file exists.""" + handler = CombinedMixin() + + # Create a session file + session_file = self.sessions_dir / "test-session.json" + session_data = { + "session_id": "test-session", + "agent": "claude", + "project_dir": "/test/project", + "status": "active", + "last_event_at": "2024-01-01T00:00:00Z", + } + session_file.write_text(json.dumps(session_data)) + + # Create a conversation file + conv_file = Path(self.temp_dir) / "conversation.jsonl" + conv_file.write_text('{"type": "user"}\n') + expected_mtime = conv_file.stat().st_mtime_ns + + with patch("amc_server.mixins.state.SESSIONS_DIR", self.sessions_dir), \ + patch("amc_server.mixins.state.EVENTS_DIR", Path(self.temp_dir) / "events"), \ + patch.object(handler, "_discover_active_codex_sessions"), \ + patch.object(handler, "_get_active_zellij_sessions", return_value=None), \ + patch.object(handler, "_get_context_usage_for_session", return_value=None), \ + patch.object(handler, "_get_claude_conversation_file", return_value=conv_file): + + sessions = handler._collect_sessions() + + self.assertEqual(len(sessions), 1) + self.assertEqual(sessions[0]["session_id"], "test-session") + self.assertEqual(sessions[0]["conversation_mtime_ns"], expected_mtime) + + def test_collect_sessions_omits_mtime_when_file_missing(self): + """_collect_sessions does not add conversation_mtime_ns when file doesn't exist.""" + handler = CombinedMixin() + + session_file = self.sessions_dir / "no-conv-session.json" + session_data = { + "session_id": "no-conv-session", + "agent": "claude", + "project_dir": "/test/project", + "status": "active", + "last_event_at": "2024-01-01T00:00:00Z", + } + session_file.write_text(json.dumps(session_data)) + + with patch("amc_server.mixins.state.SESSIONS_DIR", self.sessions_dir), \ + patch("amc_server.mixins.state.EVENTS_DIR", Path(self.temp_dir) / "events"), \ + patch.object(handler, "_discover_active_codex_sessions"), \ + patch.object(handler, "_get_active_zellij_sessions", return_value=None), \ + patch.object(handler, "_get_context_usage_for_session", return_value=None), \ + patch.object(handler, "_get_claude_conversation_file", return_value=None): + + sessions = handler._collect_sessions() + + self.assertEqual(len(sessions), 1) + self.assertNotIn("conversation_mtime_ns", sessions[0]) + + +class TestDashboardChangeDetection(unittest.TestCase): + """Tests verifying the dashboard uses mtime for change detection.""" + + def test_mtime_triggers_state_hash_change(self): + """When conversation_mtime_ns changes, payload hash should change.""" + # This is implicitly tested by the SSE mechanism: + # state.py builds payload with conversation_mtime_ns + # _serve_stream hashes payload and sends on change + + # Simulate two state payloads with different mtimes + payload_1 = { + "sessions": [{"session_id": "s1", "conversation_mtime_ns": 1000}], + "server_time": "2024-01-01T00:00:00Z", + } + payload_2 = { + "sessions": [{"session_id": "s1", "conversation_mtime_ns": 2000}], + "server_time": "2024-01-01T00:00:00Z", + } + + import hashlib + hash_1 = hashlib.sha1(json.dumps(payload_1).encode()).hexdigest() + hash_2 = hashlib.sha1(json.dumps(payload_2).encode()).hexdigest() + + self.assertNotEqual(hash_1, hash_2) + + def test_same_mtime_same_hash(self): + """When mtime hasn't changed, hash should be stable.""" + payload = { + "sessions": [{"session_id": "s1", "conversation_mtime_ns": 1000}], + "server_time": "2024-01-01T00:00:00Z", + } + + import hashlib + hash_1 = hashlib.sha1(json.dumps(payload).encode()).hexdigest() + hash_2 = hashlib.sha1(json.dumps(payload).encode()).hexdigest() + + self.assertEqual(hash_1, hash_2) + + +if __name__ == "__main__": + unittest.main()