azad.prompts.dialects.native.test_native_parser Module

azad.prompts.dialects.native.test_native_parser

Test suite for the Native dialect parser.

Attributes

Classes

MockPromptData

MockPromptData()

Mock prompt data for testing.

Source code in azad/prompts/dialects/native/test_native_parser.py

def __init__(self):
    self.task = None
    self.task_config = None
    self.tool_metadata = []
    self.current_tool_call_id = None
    self.current_assistant_id = None

Attributes

task instance-attribute

task = None

task_config instance-attribute

task_config = None

tool_metadata instance-attribute

tool_metadata = []

current_tool_call_id instance-attribute

current_tool_call_id = None

current_assistant_id instance-attribute

current_assistant_id = None

TestNativeParser

Bases: TestCase

Test suite for the Native dialect parser.

Functions

setUp

setUp()

Set up test case.

Source code in azad/prompts/dialects/native/test_native_parser.py

def setUp(self):
    """Set up test case."""
    schema = {
        "tools": {
            "attempt_completion": {
                "parameters": {
                    "result": {"type": "string"}
                }
            }
        }
    }
    self.prompt_data = MockPromptData()
    self.config = NativeDialectConfig()
    self.parser = NativeDialectParser(schema, self.prompt_data, self.config) # type: ignore

create_tool_call_delta

create_tool_call_delta(tool_id: Optional[str] = None, tool_name: Optional[str] = None, arguments: str = '', tool_type: str = 'function', index: int = 0) -> ChatCompletionDeltaToolCall

Helper to create a tool call delta with the specified properties.

Source code in azad/prompts/dialects/native/test_native_parser.py

def create_tool_call_delta(
    self,
    tool_id: Optional[str] = None,
    tool_name: Optional[str] = None,
    arguments: str = "",
    tool_type: str = "function",
    index: int = 0
) -> ChatCompletionDeltaToolCall:
    """Helper to create a tool call delta with the specified properties."""
    return ChatCompletionDeltaToolCall(
        id=tool_id,
        function=Function(
            arguments=arguments,
            name=tool_name
        ),
        type=tool_type,
        index=index
    )

test_native_parser_with_realistic_stream

test_native_parser_with_realistic_stream()

Test parsing a realistic stream of tool call deltas.

Source code in azad/prompts/dialects/native/test_native_parser.py

def test_native_parser_with_realistic_stream(self):
    """Test parsing a realistic stream of tool call deltas."""
    events = []

    # Simulate chunks as seen in the LiteLLM stream example
    # First chunk with tool ID and name
    tool_call = self.create_tool_call_delta(
        tool_id='toolu_01T8rmwSEpbXrr5fL8qZY5jt',
        tool_name='attempt_completion',
        arguments=''
    )
    events.extend(self.parser.feed_tool_call_delta(tool_call))

    # Subsequent chunks with argument pieces
    argument_pieces = [
        '{"',
        'result": "T',
        'he ',
        'sum o',
        'f 25 +',
        ' 25 eq',
        'uals 50."}'
    ]

    for piece in argument_pieces:
        tool_call = self.create_tool_call_delta(arguments=piece)
        events.extend(self.parser.feed_tool_call_delta(tool_call))

    # Validate events
    self._validate_events(events)

test_stream_with_long_parameter

test_stream_with_long_parameter()

Test parsing a stream with a longer parameter value that comes in many pieces.

Source code in azad/prompts/dialects/native/test_native_parser.py

def test_stream_with_long_parameter(self):
    """Test parsing a stream with a longer parameter value that comes in many pieces."""
    events = []

    # First chunk with tool ID and name
    tool_call = self.create_tool_call_delta(
        tool_id='toolu_01T8rmwSEpbXrr5fL8qZY5jt',
        tool_name='attempt_completion',
        arguments=''
    )
    events.extend(self.parser.feed_tool_call_delta(tool_call))

    # Start JSON
    tool_call = self.create_tool_call_delta(arguments='{"result": "')
    events.extend(self.parser.feed_tool_call_delta(tool_call))

    # Send a longer text in small chunks (simulating letter-by-letter streaming)
    long_text = "I have completed the task of analyzing the data. The results show that 42% of users preferred the new UI design, while 58% preferred the original. Key feedback points included: better color contrast in the new design, more intuitive navigation in the original, and concerns about button placement in both designs."
    chunk_size = 3  # Really small chunks to test buffering

    for i in range(0, len(long_text), chunk_size):
        chunk = long_text[i:i+chunk_size]
        tool_call = self.create_tool_call_delta(arguments=chunk)
        events.extend(self.parser.feed_tool_call_delta(tool_call))

    # Close JSON
    tool_call = self.create_tool_call_delta(arguments='"}\n')
    events.extend(self.parser.feed_tool_call_delta(tool_call))

    # Validate that we didn't get an event for every single chunk
    param_chunks = [e for e in events if isinstance(e, AINetworkEventParameterChunk)]

    # We should have fewer parameter chunk events than individual chunks we sent
    # This validates that the parser is bundling small chunks together
    expected_max_chunks = (len(long_text) // chunk_size) + 2  # +2 for start/end pieces
    self.assertLess(len(param_chunks), expected_max_chunks, 
                    f"Expected fewer than {expected_max_chunks} parameter chunks, got {len(param_chunks)}")

    # But we should still have the complete parameter value when combined
    combined = "".join([chunk.content for chunk in param_chunks])
    self.assertIn(long_text, combined, "Combined parameter chunks should contain the full text")