Spaces:
Running
Running
File size: 25,986 Bytes
9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 142a6dd 308f1f9 142a6dd 308f1f9 142a6dd 308f1f9 142a6dd 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 | """Tests for the parser module.
Tests all parsing and analysis functions:
- compute_hash: Content hashing
- detect_waste_signals: Waste signal detection
- is_rag_content: RAG content detection
- parse_message_to_blocks: Single message parsing
- parse_messages: Multi-message parsing
- find_tool_units: Tool call/response pairing
- get_message_content_text: Content extraction
"""
from unittest.mock import Mock
import pytest
from headroom.parser import (
compute_hash,
detect_waste_signals,
find_tool_units,
get_message_content_text,
is_rag_content,
parse_message_to_blocks,
parse_messages,
)
# --- Fixtures ---
@pytest.fixture
def mock_tokenizer():
"""Mock tokenizer that returns predictable token counts."""
tokenizer = Mock()
# Simple mock: 1 token per 4 characters
tokenizer.count_text = Mock(side_effect=lambda text: len(text) // 4 + 1)
return tokenizer
@pytest.fixture
def system_message():
"""Basic system message."""
return {"role": "system", "content": "You are a helpful assistant."}
@pytest.fixture
def user_message():
"""Basic user message."""
return {"role": "user", "content": "Hello, how are you?"}
@pytest.fixture
def assistant_message():
"""Basic assistant message."""
return {"role": "assistant", "content": "I'm doing well, thank you!"}
@pytest.fixture
def tool_call_message():
"""Assistant message with tool calls."""
return {
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_abc123",
"type": "function",
"function": {"name": "search_user", "arguments": '{"user_id": "12345"}'},
}
],
}
@pytest.fixture
def tool_result_message():
"""Tool result message."""
return {
"role": "tool",
"tool_call_id": "call_abc123",
"content": '{"id": "12345", "name": "Alice", "email": "alice@example.com"}',
}
@pytest.fixture
def multimodal_message():
"""User message with multimodal content (list format)."""
return {
"role": "user",
"content": [
{"type": "text", "text": "Analyze this image:"},
{"type": "image", "source": {"type": "base64", "data": "..."}},
{"type": "text", "text": "What do you see?"},
],
}
@pytest.fixture
def rag_user_message():
"""User message containing RAG content markers."""
return {
"role": "user",
"content": "[Document 1] Here is the relevant context from our knowledge base. [Source: docs/manual.md]",
}
@pytest.fixture
def html_waste_text():
"""Text containing HTML noise."""
return "<div class='container'><p>Hello</p><!-- comment --></div>"
@pytest.fixture
def base64_waste_text():
"""Text containing base64 encoded data."""
return "Data: " + "A" * 60 + "=="
@pytest.fixture
def whitespace_waste_text():
"""Text with excessive whitespace."""
return "Line 1\n\n\n\nLine 2 extra spaces"
@pytest.fixture
def json_bloat_text():
"""Text containing large JSON block (>500 chars).
Uses spaces and punctuation to avoid base64 pattern matching.
"""
# Use content that won't match base64 pattern (needs non-base64 chars)
content = "This is a long text value. " * 25 # ~675 chars
return '{"data": "' + content + '"}'
# --- TestComputeHash ---
class TestComputeHash:
"""Tests for compute_hash function."""
def test_consistent_hash(self):
"""Same text produces same hash."""
text = "Hello, world!"
hash1 = compute_hash(text)
hash2 = compute_hash(text)
assert hash1 == hash2
def test_different_texts_different_hashes(self):
"""Different texts produce different hashes."""
hash1 = compute_hash("Hello")
hash2 = compute_hash("World")
assert hash1 != hash2
def test_hash_length_16(self):
"""Hash is truncated to 16 characters."""
text = "Any text content"
hash_result = compute_hash(text)
assert len(hash_result) == 16
def test_empty_string_hash(self):
"""Empty string produces valid hash."""
hash_result = compute_hash("")
assert len(hash_result) == 16
assert hash_result.isalnum()
def test_unicode_text_hash(self):
"""Unicode text produces valid hash."""
hash_result = compute_hash("Hello \\u4e16\\u754c")
assert len(hash_result) == 16
# --- TestDetectWasteSignals ---
class TestDetectWasteSignals:
"""Tests for detect_waste_signals function."""
def test_detect_html_tags(self, mock_tokenizer, html_waste_text):
"""Detects HTML tags as waste."""
signals = detect_waste_signals(html_waste_text, mock_tokenizer)
assert signals.html_noise_tokens > 0
def test_detect_html_comments(self, mock_tokenizer):
"""Detects HTML comments as waste."""
text = "Some text <!-- this is a comment --> more text"
signals = detect_waste_signals(text, mock_tokenizer)
assert signals.html_noise_tokens > 0
def test_detect_base64(self, mock_tokenizer, base64_waste_text):
"""Detects base64 encoded content as waste."""
signals = detect_waste_signals(base64_waste_text, mock_tokenizer)
assert signals.base64_tokens > 0
def test_detect_excessive_whitespace(self, mock_tokenizer, whitespace_waste_text):
"""Detects excessive whitespace as waste."""
signals = detect_waste_signals(whitespace_waste_text, mock_tokenizer)
assert signals.whitespace_tokens >= 0 # May be 0 if normalized tokens <= matches
def test_detect_json_bloat(self, mock_tokenizer, json_bloat_text):
"""Detects large JSON blocks as bloat."""
# Need to ensure the mock returns >500 tokens for JSON bloat
# The JSON pattern requires the matched block to have >500 tokens
mock_tokenizer.count_text = Mock(side_effect=lambda text: len(text))
signals = detect_waste_signals(json_bloat_text, mock_tokenizer)
assert signals.json_bloat_tokens > 0
def test_empty_text_no_waste(self, mock_tokenizer):
"""Empty text returns zero waste signals."""
signals = detect_waste_signals("", mock_tokenizer)
assert signals.total() == 0
def test_combined_waste_signals(self, mock_tokenizer):
"""Multiple waste types are detected together."""
text = "<div>Hello</div> " + "B" * 60 + "== and <!-- comment -->"
signals = detect_waste_signals(text, mock_tokenizer)
assert signals.html_noise_tokens > 0
assert signals.base64_tokens > 0
def test_clean_text_no_waste(self, mock_tokenizer):
"""Clean text produces minimal waste signals."""
text = "This is a normal sentence without any waste."
signals = detect_waste_signals(text, mock_tokenizer)
assert signals.html_noise_tokens == 0
assert signals.base64_tokens == 0
assert signals.json_bloat_tokens == 0
# --- TestIsRagContent ---
class TestIsRagContent:
"""Tests for is_rag_content function."""
def test_document_markers(self):
"""Detects [Document N] markers."""
text = "[Document 1] This is the first document. [Document 2] Second document."
assert is_rag_content(text) is True
def test_source_markers(self):
"""Detects [Source: ...] markers."""
text = "[Source: knowledge_base/docs.md] Here is the information."
assert is_rag_content(text) is True
def test_context_tags(self):
"""Detects <context> and <document> tags."""
assert is_rag_content("<context>Retrieved content here</context>") is True
assert is_rag_content("<document>Document content</document>") is True
def test_retrieved_from_marker(self):
"""Detects 'Retrieved from:' marker."""
text = "Retrieved from: https://example.com/docs\nHere is the content."
assert is_rag_content(text) is True
def test_knowledge_base_marker(self):
"""Detects 'From the knowledge base:' marker."""
text = "From the knowledge base: This is relevant information."
assert is_rag_content(text) is True
def test_not_rag_content(self):
"""Regular text is not detected as RAG content."""
text = "Hello, how can I help you today?"
assert is_rag_content(text) is False
def test_case_insensitive(self):
"""RAG detection is case insensitive."""
assert is_rag_content("[DOCUMENT 1] Content") is True
assert is_rag_content("retrieved FROM: somewhere") is True
# --- TestParseMessageToBlocks ---
class TestParseMessageToBlocks:
"""Tests for parse_message_to_blocks function."""
def test_system_message_block(self, mock_tokenizer, system_message):
"""System message creates system block."""
blocks = parse_message_to_blocks(system_message, 0, mock_tokenizer)
assert len(blocks) == 1
assert blocks[0].kind == "system"
assert blocks[0].text == "You are a helpful assistant."
assert blocks[0].source_index == 0
def test_user_message_block(self, mock_tokenizer, user_message):
"""User message creates user block."""
blocks = parse_message_to_blocks(user_message, 1, mock_tokenizer)
assert len(blocks) == 1
assert blocks[0].kind == "user"
assert blocks[0].text == "Hello, how are you?"
assert blocks[0].source_index == 1
def test_assistant_message_block(self, mock_tokenizer, assistant_message):
"""Assistant message creates assistant block."""
blocks = parse_message_to_blocks(assistant_message, 2, mock_tokenizer)
assert len(blocks) == 1
assert blocks[0].kind == "assistant"
assert blocks[0].text == "I'm doing well, thank you!"
def test_tool_result_block(self, mock_tokenizer, tool_result_message):
"""Tool result creates tool_result block with tool_call_id."""
blocks = parse_message_to_blocks(tool_result_message, 3, mock_tokenizer)
assert len(blocks) == 1
assert blocks[0].kind == "tool_result"
assert blocks[0].flags.get("tool_call_id") == "call_abc123"
def test_rag_detection_in_user_message(self, mock_tokenizer, rag_user_message):
"""User message with RAG markers creates rag block."""
blocks = parse_message_to_blocks(rag_user_message, 0, mock_tokenizer)
assert len(blocks) == 1
assert blocks[0].kind == "rag"
def test_multimodal_content(self, mock_tokenizer, multimodal_message):
"""Multimodal content (list with text parts) is extracted."""
blocks = parse_message_to_blocks(multimodal_message, 0, mock_tokenizer)
assert len(blocks) == 1
assert "Analyze this image:" in blocks[0].text
assert "What do you see?" in blocks[0].text
def test_tool_calls_create_separate_blocks(self, mock_tokenizer, tool_call_message):
"""Tool calls create separate tool_call blocks."""
blocks = parse_message_to_blocks(tool_call_message, 0, mock_tokenizer)
# Should have tool_call blocks (no content block since content is None)
tool_call_blocks = [b for b in blocks if b.kind == "tool_call"]
assert len(tool_call_blocks) == 1
assert tool_call_blocks[0].flags.get("tool_call_id") == "call_abc123"
assert tool_call_blocks[0].flags.get("function_name") == "search_user"
assert "search_user" in tool_call_blocks[0].text
def test_empty_message_creates_block(self, mock_tokenizer):
"""Empty message (no content or tool_calls) creates minimal block."""
empty_msg = {"role": "assistant"}
blocks = parse_message_to_blocks(empty_msg, 0, mock_tokenizer)
assert len(blocks) == 1
assert blocks[0].kind == "unknown"
assert blocks[0].text == ""
def test_message_with_content_and_tool_calls(self, mock_tokenizer):
"""Message with both content and tool_calls creates multiple blocks."""
msg = {
"role": "assistant",
"content": "Let me search for that.",
"tool_calls": [{"id": "call_xyz", "function": {"name": "search", "arguments": "{}"}}],
}
blocks = parse_message_to_blocks(msg, 0, mock_tokenizer)
kinds = [b.kind for b in blocks]
assert "assistant" in kinds
assert "tool_call" in kinds
def test_waste_signals_in_flags(self, mock_tokenizer, html_waste_text):
"""Waste signals are added to block flags."""
msg = {"role": "user", "content": html_waste_text}
blocks = parse_message_to_blocks(msg, 0, mock_tokenizer)
assert "waste_signals" in blocks[0].flags
assert blocks[0].flags["waste_signals"]["html_noise"] > 0
def test_content_hash_generated(self, mock_tokenizer, user_message):
"""Content hash is generated for blocks."""
blocks = parse_message_to_blocks(user_message, 0, mock_tokenizer)
assert len(blocks[0].content_hash) == 16
def test_tokens_estimated(self, mock_tokenizer, user_message):
"""Token count is estimated."""
blocks = parse_message_to_blocks(user_message, 0, mock_tokenizer)
assert blocks[0].tokens_est > 0
# --- TestParseMessages ---
class TestParseMessages:
"""Tests for parse_messages function."""
def test_parse_all_messages(self, mock_tokenizer, sample_messages):
"""All messages are parsed into blocks."""
blocks, breakdown, waste = parse_messages(sample_messages, mock_tokenizer)
assert len(blocks) >= len(sample_messages)
def test_block_breakdown(self, mock_tokenizer, sample_messages):
"""Block breakdown counts tokens per kind."""
blocks, breakdown, waste = parse_messages(sample_messages, mock_tokenizer)
assert "system" in breakdown
assert "user" in breakdown
assert "assistant" in breakdown
assert all(v > 0 for v in breakdown.values())
def test_waste_signals_accumulated(self, mock_tokenizer):
"""Waste signals are accumulated across messages."""
messages = [
{"role": "user", "content": "<div>HTML here</div>"},
{"role": "assistant", "content": "More <span>HTML</span>"},
]
blocks, breakdown, waste = parse_messages(messages, mock_tokenizer)
assert waste.html_noise_tokens > 0
def test_empty_messages(self, mock_tokenizer):
"""Empty message list returns empty results."""
blocks, breakdown, waste = parse_messages([], mock_tokenizer)
assert blocks == []
assert breakdown == {}
assert waste.total() == 0
def test_multiple_tool_calls_parsed(self, mock_tokenizer, sample_messages_with_tools):
"""Messages with tool calls are parsed correctly."""
blocks, breakdown, waste = parse_messages(sample_messages_with_tools, mock_tokenizer)
tool_call_blocks = [b for b in blocks if b.kind == "tool_call"]
tool_result_blocks = [b for b in blocks if b.kind == "tool_result"]
assert len(tool_call_blocks) >= 1
assert len(tool_result_blocks) >= 1
# --- TestFindToolUnits ---
class TestFindToolUnits:
"""Tests for find_tool_units function."""
def test_finds_tool_call_and_responses(self, sample_messages_with_tools):
"""Finds matching tool call and response pairs."""
units = find_tool_units(sample_messages_with_tools)
assert len(units) >= 1
# Each unit is (assistant_index, [tool_response_indices])
assistant_idx, response_indices = units[0]
assert response_indices # Should have at least one response
def test_multiple_tool_calls_same_assistant(self):
"""Multiple tool calls from same assistant are grouped."""
messages = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Search both"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{"id": "call_1", "function": {"name": "search", "arguments": "{}"}},
{"id": "call_2", "function": {"name": "fetch", "arguments": "{}"}},
],
},
{"role": "tool", "tool_call_id": "call_1", "content": "result 1"},
{"role": "tool", "tool_call_id": "call_2", "content": "result 2"},
]
units = find_tool_units(messages)
assert len(units) == 1
assistant_idx, response_indices = units[0]
assert len(response_indices) == 2
def test_no_tool_units(self):
"""Returns empty list when no tool calls present."""
messages = [
{"role": "system", "content": "Hello"},
{"role": "user", "content": "Hi"},
{"role": "assistant", "content": "Hello!"},
]
units = find_tool_units(messages)
assert units == []
def test_orphaned_tool_response(self):
"""Tool response without matching assistant is not included."""
messages = [
{"role": "system", "content": "Hello"},
{"role": "user", "content": "Hi"},
# Orphaned tool response - no assistant with tool_calls
{"role": "tool", "tool_call_id": "orphan_call", "content": "orphaned"},
{"role": "assistant", "content": "I don't have tools."},
]
units = find_tool_units(messages)
assert units == []
def test_tool_response_order_sorted(self):
"""Tool response indices are sorted."""
messages = [
{"role": "user", "content": "Do two things"},
{
"role": "assistant",
"tool_calls": [
{"id": "call_a", "function": {"name": "first", "arguments": "{}"}},
{"id": "call_b", "function": {"name": "second", "arguments": "{}"}},
],
},
{"role": "tool", "tool_call_id": "call_b", "content": "second result"},
{"role": "tool", "tool_call_id": "call_a", "content": "first result"},
]
units = find_tool_units(messages)
assert len(units) == 1
_, response_indices = units[0]
assert response_indices == sorted(response_indices)
def test_anthropic_format_tool_use_and_result(self):
"""Finds Anthropic format tool_use/tool_result pairs in content blocks."""
messages = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Take a screenshot"},
{
"role": "assistant",
"content": [
{"type": "text", "text": "Let me take a screenshot."},
{
"type": "tool_use",
"id": "toolu_123",
"name": "browser_screenshot",
"input": {},
},
],
},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "toolu_123",
"content": "Screenshot taken successfully",
}
],
},
{"role": "user", "content": "Thanks!"},
]
units = find_tool_units(messages)
assert len(units) == 1
assistant_idx, response_indices = units[0]
assert assistant_idx == 2
assert response_indices == [3]
def test_anthropic_format_multiple_tool_uses(self):
"""Finds multiple Anthropic format tool_use blocks from same assistant."""
messages = [
{"role": "user", "content": "Do two things"},
{
"role": "assistant",
"content": [
{"type": "tool_use", "id": "toolu_a", "name": "first", "input": {}},
{"type": "tool_use", "id": "toolu_b", "name": "second", "input": {}},
],
},
{
"role": "user",
"content": [
{"type": "tool_result", "tool_use_id": "toolu_a", "content": "first done"},
{"type": "tool_result", "tool_use_id": "toolu_b", "content": "second done"},
],
},
]
units = find_tool_units(messages)
assert len(units) == 1
assistant_idx, response_indices = units[0]
assert assistant_idx == 1
assert response_indices == [2]
def test_anthropic_format_orphaned_tool_result(self):
"""Anthropic tool_result without matching tool_use is not included."""
messages = [
{"role": "user", "content": "Hi"},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "orphan_toolu",
"content": "orphaned result",
}
],
},
{"role": "assistant", "content": "Hello!"},
]
units = find_tool_units(messages)
assert units == []
def test_mixed_openai_and_anthropic_formats(self):
"""Both OpenAI and Anthropic formats can coexist (edge case)."""
messages = [
{"role": "user", "content": "Do things"},
# OpenAI format
{
"role": "assistant",
"tool_calls": [
{"id": "call_1", "function": {"name": "openai_tool", "arguments": "{}"}}
],
},
{"role": "tool", "tool_call_id": "call_1", "content": "openai result"},
# Anthropic format
{
"role": "assistant",
"content": [
{"type": "tool_use", "id": "toolu_2", "name": "anthropic_tool", "input": {}}
],
},
{
"role": "user",
"content": [
{"type": "tool_result", "tool_use_id": "toolu_2", "content": "anthropic result"}
],
},
]
units = find_tool_units(messages)
assert len(units) == 2
# First unit: OpenAI format (assistant at 1, tool response at 2)
assert units[0] == (1, [2])
# Second unit: Anthropic format (assistant at 3, user with tool_result at 4)
assert units[1] == (3, [4])
# --- TestGetMessageContentText ---
class TestGetMessageContentText:
"""Tests for get_message_content_text function."""
def test_string_content(self):
"""Extracts string content directly."""
msg = {"role": "user", "content": "Hello, world!"}
text = get_message_content_text(msg)
assert text == "Hello, world!"
def test_list_content(self):
"""Extracts text from list content (multimodal)."""
msg = {
"role": "user",
"content": [
{"type": "text", "text": "First part"},
{"type": "image", "source": {}},
{"type": "text", "text": "Second part"},
],
}
text = get_message_content_text(msg)
assert "First part" in text
assert "Second part" in text
def test_none_content(self):
"""Returns empty string for None content."""
msg = {"role": "assistant", "content": None}
text = get_message_content_text(msg)
assert text == ""
def test_mixed_content_list(self):
"""Handles list with both dict and string items."""
msg = {
"role": "user",
"content": [
{"type": "text", "text": "Dict text"},
"Plain string",
],
}
text = get_message_content_text(msg)
assert "Dict text" in text
assert "Plain string" in text
def test_missing_content_key(self):
"""Returns empty string when content key is missing."""
msg = {"role": "user"}
text = get_message_content_text(msg)
assert text == ""
def test_non_text_type_skipped(self):
"""Non-text types in list are skipped."""
msg = {
"role": "user",
"content": [
{"type": "image", "data": "..."},
{"type": "text", "text": "Only this"},
],
}
text = get_message_content_text(msg)
assert text == "Only this"
def test_empty_list_content(self):
"""Empty list content returns empty string."""
msg = {"role": "user", "content": []}
text = get_message_content_text(msg)
assert text == ""
# --- Additional fixtures for complex tests ---
@pytest.fixture
def sample_messages():
"""Basic conversation messages."""
return [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello, how are you?"},
{"role": "assistant", "content": "I'm doing well, thank you!"},
]
@pytest.fixture
def sample_messages_with_tools():
"""Conversation with tool calls and responses."""
return [
{"role": "system", "content": "You are a helpful assistant with tools."},
{"role": "user", "content": "Search for user 12345"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_123",
"type": "function",
"function": {"name": "search_user", "arguments": '{"user_id": "12345"}'},
}
],
},
{
"role": "tool",
"tool_call_id": "call_123",
"content": '{"id": "12345", "name": "Alice", "email": "alice@example.com"}',
},
{"role": "assistant", "content": "I found user Alice with ID 12345."},
]
|