abbede923d
Add _is_system_injected() filter to conversation.py that drops user
messages starting with known system-injected prefixes. These messages
(hook outputs, system reminders, teammate notifications) appear in
JSONL session logs as type: "user" with string content but are not
human-typed input.
Filtered prefixes:
- <system-reminder> — Claude Code system context injection
- <local-command-caveat> — local command hook output
- <available-deferred-tools> — deferred tool discovery messages
- <teammate-message — team agent message delivery (no closing >
because tag has attributes)
This brings Claude parsing to parity with the Codex parser, which
already filters system-injected content via SKIP_PREFIXES (line 222).
The filter uses str.startswith(tuple) with lstrip() to handle leading
whitespace. Applied at line 75 in the existing content-type guard chain.
Affects both chat display and input history navigation — system noise
is removed at the source so all consumers benefit.
tests/test_conversation.py:
- TestIsSystemInjected: 10 unit tests for the filter function covering
each prefix, leading whitespace, normal messages, mid-string tags,
empty strings, slash commands, and multiline content
- TestClaudeSystemInjectedFiltering: 5 integration tests through the
full parser verifying exclusion, sequential ID preservation after
filtering, and all-system-message edge case
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
586 lines
24 KiB
Python
586 lines
24 KiB
Python
"""Tests for mixins/conversation.py edge cases.
|
|
|
|
Unit tests for conversation parsing from Claude Code and Codex JSONL files.
|
|
"""
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from amc_server.mixins.conversation import ConversationMixin, _is_system_injected
|
|
from amc_server.mixins.parsing import SessionParsingMixin
|
|
|
|
|
|
class DummyConversationHandler(ConversationMixin, SessionParsingMixin):
|
|
"""Minimal handler for testing conversation mixin."""
|
|
|
|
def __init__(self):
|
|
self.sent_responses = []
|
|
|
|
def _send_json(self, code, payload):
|
|
self.sent_responses.append((code, payload))
|
|
|
|
|
|
class TestParseCodexArguments(unittest.TestCase):
|
|
"""Tests for _parse_codex_arguments edge cases."""
|
|
|
|
def setUp(self):
|
|
self.handler = DummyConversationHandler()
|
|
|
|
def test_dict_input_returned_as_is(self):
|
|
result = self.handler._parse_codex_arguments({"key": "value"})
|
|
self.assertEqual(result, {"key": "value"})
|
|
|
|
def test_empty_dict_returned_as_is(self):
|
|
result = self.handler._parse_codex_arguments({})
|
|
self.assertEqual(result, {})
|
|
|
|
def test_json_string_parsed(self):
|
|
result = self.handler._parse_codex_arguments('{"key": "value"}')
|
|
self.assertEqual(result, {"key": "value"})
|
|
|
|
def test_invalid_json_string_returns_raw(self):
|
|
result = self.handler._parse_codex_arguments("not valid json")
|
|
self.assertEqual(result, {"raw": "not valid json"})
|
|
|
|
def test_empty_string_returns_raw(self):
|
|
result = self.handler._parse_codex_arguments("")
|
|
self.assertEqual(result, {"raw": ""})
|
|
|
|
def test_none_returns_empty_dict(self):
|
|
result = self.handler._parse_codex_arguments(None)
|
|
self.assertEqual(result, {})
|
|
|
|
def test_int_returns_empty_dict(self):
|
|
result = self.handler._parse_codex_arguments(42)
|
|
self.assertEqual(result, {})
|
|
|
|
def test_list_returns_empty_dict(self):
|
|
result = self.handler._parse_codex_arguments([1, 2, 3])
|
|
self.assertEqual(result, {})
|
|
|
|
|
|
class TestServeEvents(unittest.TestCase):
|
|
"""Tests for _serve_events edge cases."""
|
|
|
|
def setUp(self):
|
|
self.handler = DummyConversationHandler()
|
|
|
|
def test_path_traversal_sanitized(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
events_dir = Path(tmpdir)
|
|
# Create a file that path traversal might try to access (unused - documents intent)
|
|
_secret_file = Path(tmpdir).parent / "secret.jsonl"
|
|
|
|
with patch("amc_server.mixins.conversation.EVENTS_DIR", events_dir):
|
|
# Try path traversal
|
|
self.handler._serve_events("../secret")
|
|
|
|
# Should have served response with sanitized id
|
|
self.assertEqual(len(self.handler.sent_responses), 1)
|
|
code, payload = self.handler.sent_responses[0]
|
|
self.assertEqual(code, 200)
|
|
self.assertEqual(payload["session_id"], "secret")
|
|
self.assertEqual(payload["events"], [])
|
|
|
|
def test_nonexistent_file_returns_empty_events(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
with patch("amc_server.mixins.conversation.EVENTS_DIR", Path(tmpdir)):
|
|
self.handler._serve_events("nonexistent")
|
|
|
|
code, payload = self.handler.sent_responses[0]
|
|
self.assertEqual(payload["events"], [])
|
|
|
|
def test_empty_file_returns_empty_events(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
events_dir = Path(tmpdir)
|
|
event_file = events_dir / "session123.jsonl"
|
|
event_file.write_text("")
|
|
|
|
with patch("amc_server.mixins.conversation.EVENTS_DIR", events_dir):
|
|
self.handler._serve_events("session123")
|
|
|
|
code, payload = self.handler.sent_responses[0]
|
|
self.assertEqual(payload["events"], [])
|
|
|
|
def test_invalid_json_lines_skipped(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
events_dir = Path(tmpdir)
|
|
event_file = events_dir / "session123.jsonl"
|
|
event_file.write_text('{"valid": "event"}\nnot json\n{"another": "event"}\n')
|
|
|
|
with patch("amc_server.mixins.conversation.EVENTS_DIR", events_dir):
|
|
self.handler._serve_events("session123")
|
|
|
|
code, payload = self.handler.sent_responses[0]
|
|
self.assertEqual(len(payload["events"]), 2)
|
|
self.assertEqual(payload["events"][0], {"valid": "event"})
|
|
self.assertEqual(payload["events"][1], {"another": "event"})
|
|
|
|
|
|
class TestParseClaudeConversation(unittest.TestCase):
|
|
"""Tests for _parse_claude_conversation edge cases."""
|
|
|
|
def setUp(self):
|
|
self.handler = DummyConversationHandler()
|
|
|
|
def test_user_message_with_array_content_skipped(self):
|
|
# Array content is tool results, not human messages
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "user",
|
|
"message": {"content": [{"type": "tool_result"}]}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=path):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(messages, [])
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_user_message_with_string_content_included(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "user",
|
|
"timestamp": "2024-01-01T00:00:00Z",
|
|
"message": {"content": "Hello, Claude!"}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=path):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["role"], "user")
|
|
self.assertEqual(messages[0]["content"], "Hello, Claude!")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_assistant_message_with_text_parts(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "assistant",
|
|
"timestamp": "2024-01-01T00:00:00Z",
|
|
"message": {
|
|
"content": [
|
|
{"type": "text", "text": "Part 1"},
|
|
{"type": "text", "text": "Part 2"},
|
|
]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=path):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["content"], "Part 1\nPart 2")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_assistant_message_with_tool_use(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "assistant",
|
|
"message": {
|
|
"content": [
|
|
{"type": "tool_use", "name": "Read", "input": {"file_path": "/test"}},
|
|
]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=path):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["tool_calls"][0]["name"], "Read")
|
|
self.assertEqual(messages[0]["tool_calls"][0]["input"]["file_path"], "/test")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_assistant_message_with_thinking(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "assistant",
|
|
"message": {
|
|
"content": [
|
|
{"type": "thinking", "thinking": "Let me consider..."},
|
|
{"type": "text", "text": "Here's my answer"},
|
|
]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=path):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["thinking"], "Let me consider...")
|
|
self.assertEqual(messages[0]["content"], "Here's my answer")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_assistant_message_content_as_string_parts(self):
|
|
# Some entries might have string content parts instead of dicts
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "assistant",
|
|
"message": {
|
|
"content": ["plain string", {"type": "text", "text": "structured"}]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=path):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(messages[0]["content"], "plain string\nstructured")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_missing_conversation_file_returns_empty(self):
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=None):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(messages, [])
|
|
|
|
def test_non_dict_entry_skipped(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write('"just a string"\n')
|
|
f.write('123\n')
|
|
f.write('{"type": "user", "message": {"content": "valid"}}\n')
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=path):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(len(messages), 1)
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_non_list_content_in_assistant_skipped(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "assistant",
|
|
"message": {"content": "not a list"}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=path):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(messages, [])
|
|
finally:
|
|
path.unlink()
|
|
|
|
|
|
class TestIsSystemInjected(unittest.TestCase):
|
|
"""Tests for _is_system_injected filter."""
|
|
|
|
def test_system_reminder(self):
|
|
self.assertTrue(_is_system_injected("<system-reminder>\nSome reminder text\n</system-reminder>"))
|
|
|
|
def test_local_command_caveat(self):
|
|
self.assertTrue(_is_system_injected("<local-command-caveat>Caveat: The messages below...</local-command-caveat>"))
|
|
|
|
def test_available_deferred_tools(self):
|
|
self.assertTrue(_is_system_injected("<available-deferred-tools>\nAgent\nBash\n</available-deferred-tools>"))
|
|
|
|
def test_teammate_message(self):
|
|
self.assertTrue(_is_system_injected('<teammate-message teammate_id="reviewer" color="yellow">Review complete</teammate-message>'))
|
|
|
|
def test_leading_whitespace_stripped(self):
|
|
self.assertTrue(_is_system_injected(" \n <system-reminder>content</system-reminder>"))
|
|
|
|
def test_normal_user_message(self):
|
|
self.assertFalse(_is_system_injected("Hello, Claude!"))
|
|
|
|
def test_message_containing_tag_not_at_start(self):
|
|
self.assertFalse(_is_system_injected("Please check this <system-reminder> thing"))
|
|
|
|
def test_empty_string(self):
|
|
self.assertFalse(_is_system_injected(""))
|
|
|
|
def test_slash_command(self):
|
|
self.assertFalse(_is_system_injected("/commit"))
|
|
|
|
def test_multiline_user_message(self):
|
|
self.assertFalse(_is_system_injected("Fix this bug\n\nHere's the error:\nTypeError: foo is not a function"))
|
|
|
|
|
|
class TestClaudeSystemInjectedFiltering(unittest.TestCase):
|
|
"""Integration tests: system-injected messages filtered from Claude conversation."""
|
|
|
|
def setUp(self):
|
|
self.handler = DummyConversationHandler()
|
|
|
|
def _parse_with_messages(self, *user_contents):
|
|
"""Helper: write JSONL with user messages, parse, return results."""
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
for content in user_contents:
|
|
f.write(json.dumps({
|
|
"type": "user",
|
|
"timestamp": "2024-01-01T00:00:00Z",
|
|
"message": {"content": content}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=path):
|
|
return self.handler._parse_claude_conversation("session123", "/project")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_system_reminder_excluded(self):
|
|
messages = self._parse_with_messages(
|
|
"real question",
|
|
"<system-reminder>\nHook success\n</system-reminder>",
|
|
)
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["content"], "real question")
|
|
|
|
def test_local_command_caveat_excluded(self):
|
|
messages = self._parse_with_messages(
|
|
"<local-command-caveat>Caveat: generated by local commands</local-command-caveat>",
|
|
"what does this function do?",
|
|
)
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["content"], "what does this function do?")
|
|
|
|
def test_teammate_message_excluded(self):
|
|
messages = self._parse_with_messages(
|
|
'<teammate-message teammate_id="impl" color="green">Task done</teammate-message>',
|
|
"looks good, commit it",
|
|
)
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["content"], "looks good, commit it")
|
|
|
|
def test_all_system_messages_excluded_preserves_ids(self):
|
|
"""Message IDs should be sequential with no gaps from filtering."""
|
|
messages = self._parse_with_messages(
|
|
"first real message",
|
|
"<system-reminder>noise</system-reminder>",
|
|
"<available-deferred-tools>\nAgent\n</available-deferred-tools>",
|
|
"second real message",
|
|
)
|
|
self.assertEqual(len(messages), 2)
|
|
self.assertEqual(messages[0]["content"], "first real message")
|
|
self.assertEqual(messages[1]["content"], "second real message")
|
|
# IDs should be sequential (0, 1) not (0, 3)
|
|
self.assertTrue(messages[0]["id"].endswith("-0"))
|
|
self.assertTrue(messages[1]["id"].endswith("-1"))
|
|
|
|
def test_only_system_messages_returns_empty(self):
|
|
messages = self._parse_with_messages(
|
|
"<system-reminder>reminder</system-reminder>",
|
|
"<local-command-caveat>caveat</local-command-caveat>",
|
|
)
|
|
self.assertEqual(messages, [])
|
|
|
|
|
|
class TestParseCodexConversation(unittest.TestCase):
|
|
"""Tests for _parse_codex_conversation edge cases."""
|
|
|
|
def setUp(self):
|
|
self.handler = DummyConversationHandler()
|
|
|
|
def test_developer_role_skipped(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {
|
|
"type": "message",
|
|
"role": "developer",
|
|
"content": [{"text": "System instructions"}]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_find_codex_transcript_file", return_value=path):
|
|
messages = self.handler._parse_codex_conversation("session123")
|
|
self.assertEqual(messages, [])
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_injected_context_skipped(self):
|
|
skip_prefixes = [
|
|
"<INSTRUCTIONS>",
|
|
"<environment_context>",
|
|
"<permissions instructions>",
|
|
"# AGENTS.md instructions",
|
|
]
|
|
for prefix in skip_prefixes:
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {
|
|
"type": "message",
|
|
"role": "user",
|
|
"content": [{"text": f"{prefix} more content here"}]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_find_codex_transcript_file", return_value=path):
|
|
messages = self.handler._parse_codex_conversation("session123")
|
|
self.assertEqual(messages, [], f"Should skip content starting with {prefix}")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_function_call_accumulated_to_next_assistant(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
# Tool call
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {
|
|
"type": "function_call",
|
|
"name": "shell",
|
|
"arguments": '{"command": "ls"}'
|
|
}
|
|
}) + "\n")
|
|
# Assistant message
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {
|
|
"type": "message",
|
|
"role": "assistant",
|
|
"content": [{"text": "Here are the files"}]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_find_codex_transcript_file", return_value=path):
|
|
messages = self.handler._parse_codex_conversation("session123")
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["tool_calls"][0]["name"], "shell")
|
|
self.assertEqual(messages[0]["content"], "Here are the files")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_function_calls_flushed_before_user_message(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
# Tool call
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {"type": "function_call", "name": "tool1", "arguments": "{}"}
|
|
}) + "\n")
|
|
# User message (tool calls should be flushed first)
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {
|
|
"type": "message",
|
|
"role": "user",
|
|
"content": [{"text": "User response"}]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_find_codex_transcript_file", return_value=path):
|
|
messages = self.handler._parse_codex_conversation("session123")
|
|
# First message should be assistant with tool_calls (flushed)
|
|
# Second should be user
|
|
self.assertEqual(len(messages), 2)
|
|
self.assertEqual(messages[0]["role"], "assistant")
|
|
self.assertEqual(messages[0]["tool_calls"][0]["name"], "tool1")
|
|
self.assertEqual(messages[1]["role"], "user")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_reasoning_creates_thinking_message(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {
|
|
"type": "reasoning",
|
|
"summary": [
|
|
{"type": "summary_text", "text": "Let me think..."},
|
|
{"type": "summary_text", "text": "I'll try this approach."},
|
|
]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_find_codex_transcript_file", return_value=path):
|
|
messages = self.handler._parse_codex_conversation("session123")
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["thinking"], "Let me think...\nI'll try this approach.")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_pending_tool_calls_flushed_at_end(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
# Tool call with no following message
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {"type": "function_call", "name": "final_tool", "arguments": "{}"}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_find_codex_transcript_file", return_value=path):
|
|
messages = self.handler._parse_codex_conversation("session123")
|
|
# Should flush pending tool calls at end
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["tool_calls"][0]["name"], "final_tool")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_non_response_item_types_skipped(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write('{"type": "session_meta"}\n')
|
|
f.write('{"type": "event_msg"}\n')
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {"type": "message", "role": "user", "content": [{"text": "Hello"}]}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_find_codex_transcript_file", return_value=path):
|
|
messages = self.handler._parse_codex_conversation("session123")
|
|
self.assertEqual(len(messages), 1)
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_missing_transcript_file_returns_empty(self):
|
|
with patch.object(self.handler, "_find_codex_transcript_file", return_value=None):
|
|
messages = self.handler._parse_codex_conversation("session123")
|
|
self.assertEqual(messages, [])
|
|
|
|
|
|
class TestServeConversation(unittest.TestCase):
|
|
"""Tests for _serve_conversation routing."""
|
|
|
|
def setUp(self):
|
|
self.handler = DummyConversationHandler()
|
|
|
|
def test_routes_to_codex_parser(self):
|
|
with patch.object(self.handler, "_parse_codex_conversation", return_value=[]) as mock:
|
|
self.handler._serve_conversation("session123", "/project", agent="codex")
|
|
mock.assert_called_once_with("session123")
|
|
|
|
def test_routes_to_claude_parser_by_default(self):
|
|
with patch.object(self.handler, "_parse_claude_conversation", return_value=[]) as mock:
|
|
self.handler._serve_conversation("session123", "/project")
|
|
mock.assert_called_once_with("session123", "/project")
|
|
|
|
def test_sanitizes_session_id(self):
|
|
with patch.object(self.handler, "_parse_claude_conversation", return_value=[]):
|
|
self.handler._serve_conversation("../../../etc/passwd", "/project")
|
|
|
|
code, payload = self.handler.sent_responses[0]
|
|
self.assertEqual(payload["session_id"], "passwd")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|