#!/usr/bin/env python3 """utils/prompt_cache.py 테스트 스위트""" import sys from pathlib import Path import pytest sys.path.insert(0, str(Path(__file__).parent.parent.parent)) from utils.prompt_cache import apply_cache_markers class TestApplyCacheMarkersEmpty: """엣지 케이스: 빈 메시지 목록""" def test_empty_messages_returns_empty(self): """빈 리스트 → 빈 리스트""" assert apply_cache_markers([]) == [] def test_returns_deep_copy(self): """원본 메시지 목록을 변경하지 않음 (deep copy)""" messages = [{"role": "user", "content": "hello"}] result = apply_cache_markers(messages) # 원본 변경 없음 assert "cache_control" not in messages[0] assert messages[0].get("content") == "hello" class TestApplyCacheMarkersSystemPrompt: """시스템 프롬프트 캐시 마커 적용""" def test_system_prompt_gets_cache_marker(self): """system 역할 첫 번째 메시지에 cache_control 삽입""" messages = [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hello"}, ] result = apply_cache_markers(messages) sys_msg = result[0] # 시스템 메시지: content가 list로 변환되어 마지막 블록에 cache_control 적용 content = sys_msg.get("content") if isinstance(content, list): assert content[-1].get("cache_control") == {"type": "ephemeral"} else: # 직접 cache_control이 붙는 구현도 허용 assert sys_msg.get("cache_control") == {"type": "ephemeral"} def test_system_string_content_converted_to_list(self): """문자열 content를 가진 system 메시지 → list 형식으로 변환""" messages = [ {"role": "system", "content": "System instruction here."}, ] result = apply_cache_markers(messages) content = result[0].get("content") assert isinstance(content, list) assert content[0]["text"] == "System instruction here." assert content[0]["type"] == "text" class TestApplyCacheMarkersNonSystemMessages: """비시스템 메시지 캐시 마커 적용 (system_and_3 전략)""" def test_single_user_message_gets_marker(self): """사용자 메시지 1개에 마커 적용""" messages = [{"role": "user", "content": "Hello"}] result = apply_cache_markers(messages) content = result[0].get("content") if isinstance(content, list): assert content[-1].get("cache_control") == {"type": "ephemeral"} else: assert result[0].get("cache_control") == {"type": "ephemeral"} def test_last_3_messages_get_markers(self): """마지막 3개 비시스템 메시지에 마커 적용""" messages = [ {"role": "user", "content": "msg1"}, {"role": "assistant", "content": "resp1"}, {"role": "user", "content": "msg2"}, {"role": "assistant", "content": "resp2"}, {"role": "user", "content": "msg3"}, {"role": "assistant", "content": "resp3"}, {"role": "user", "content": "msg4"}, ] result = apply_cache_markers(messages) # 마지막 3개(인덱스 4, 5, 6)에 마커 있어야 함 def has_cache_marker(msg: dict) -> bool: content = msg.get("content") if isinstance(content, list) and content: return content[-1].get("cache_control") == {"type": "ephemeral"} return msg.get("cache_control") == {"type": "ephemeral"} assert has_cache_marker(result[4]) assert has_cache_marker(result[5]) assert has_cache_marker(result[6]) # 인덱스 0~3은 마커 없음 assert not has_cache_marker(result[0]) assert not has_cache_marker(result[1]) def test_system_and_3_total_4_breakpoints(self): """system + 3 메시지 = 최대 4개 breakpoint""" messages = [ {"role": "system", "content": "System prompt."}, {"role": "user", "content": "msg1"}, {"role": "assistant", "content": "resp1"}, {"role": "user", "content": "msg2"}, {"role": "assistant", "content": "resp2"}, {"role": "user", "content": "msg3"}, {"role": "assistant", "content": "resp3"}, ] result = apply_cache_markers(messages) def has_cache_marker(msg: dict) -> bool: content = msg.get("content") if isinstance(content, list) and content: return content[-1].get("cache_control") == {"type": "ephemeral"} return msg.get("cache_control") == {"type": "ephemeral"} # system(0) + 마지막 3개(4, 5, 6) assert has_cache_marker(result[0]) # system assert has_cache_marker(result[4]) # msg2 assert has_cache_marker(result[5]) # resp2 assert has_cache_marker(result[6]) # msg3 # 마커 없는 메시지 assert not has_cache_marker(result[1]) assert not has_cache_marker(result[2]) assert not has_cache_marker(result[3]) def test_fewer_than_3_messages_all_get_markers(self): """3개 미만 메시지: 모두 마커 적용""" messages = [ {"role": "user", "content": "a"}, {"role": "assistant", "content": "b"}, ] result = apply_cache_markers(messages) def has_cache_marker(msg: dict) -> bool: content = msg.get("content") if isinstance(content, list) and content: return content[-1].get("cache_control") == {"type": "ephemeral"} return msg.get("cache_control") == {"type": "ephemeral"} assert has_cache_marker(result[0]) assert has_cache_marker(result[1]) class TestApplyCacheMarkersListContent: """list 형식 content를 가진 메시지 처리""" def test_list_content_marker_on_last_block(self): """list content의 마지막 블록에 cache_control 추가""" messages = [ { "role": "user", "content": [ {"type": "text", "text": "first"}, {"type": "text", "text": "second"}, ], } ] result = apply_cache_markers(messages) content = result[0]["content"] assert isinstance(content, list) assert content[-1].get("cache_control") == {"type": "ephemeral"} # 첫 번째 블록에는 없음 assert "cache_control" not in content[0] def test_original_not_mutated(self): """원본 메시지가 변경되지 않음""" messages = [ { "role": "user", "content": [{"type": "text", "text": "hello"}], } ] apply_cache_markers(messages) # 원본 content에 cache_control 없음 assert "cache_control" not in messages[0]["content"][0] # type: ignore[index] class TestApplyCacheMarkersEphemeralType: """cache_control 형식 검증""" def test_cache_control_type_is_ephemeral(self): """cache_control.type == 'ephemeral'""" messages = [{"role": "user", "content": "test"}] result = apply_cache_markers(messages) content = result[0].get("content") if isinstance(content, list): cc = content[-1].get("cache_control", {}) else: cc = result[0].get("cache_control", {}) assert cc.get("type") == "ephemeral" if __name__ == "__main__": pytest.main([__file__, "-v"])