(monkeypatch)
| 100 | |
| 101 | @pytest.mark.asyncio |
| 102 | async def test_single_agent_workflow(monkeypatch) -> None: |
| 103 | model = FakeStreamingModel() |
| 104 | model.add_multiple_turn_outputs( |
| 105 | [ |
| 106 | # First turn: a message and a tool call |
| 107 | [ |
| 108 | get_function_tool_call("some_function", json.dumps({"a": "b"})), |
| 109 | get_text_message("a_message"), |
| 110 | ], |
| 111 | # Second turn: text message |
| 112 | [get_text_message("done")], |
| 113 | ] |
| 114 | ) |
| 115 | |
| 116 | agent = Agent( |
| 117 | "initial_agent", |
| 118 | model=model, |
| 119 | tools=[get_function_tool("some_function", "tool_result")], |
| 120 | ) |
| 121 | |
| 122 | workflow = SingleAgentVoiceWorkflow(agent) |
| 123 | output = [] |
| 124 | async for chunk in workflow.run("transcription_1"): |
| 125 | output.append(chunk) |
| 126 | |
| 127 | # Validate that the text yielded matches our fake events |
| 128 | assert output == ["a_message", "done"] |
| 129 | # Validate that internal state was updated |
| 130 | assert workflow._input_history == snapshot( |
| 131 | [ |
| 132 | {"content": "transcription_1", "role": "user"}, |
| 133 | { |
| 134 | "arguments": '{"a": "b"}', |
| 135 | "call_id": "2", |
| 136 | "name": "some_function", |
| 137 | "type": "function_call", |
| 138 | "id": "1", |
| 139 | }, |
| 140 | { |
| 141 | "id": "1", |
| 142 | "content": [ |
| 143 | {"annotations": [], "logprobs": [], "text": "a_message", "type": "output_text"} |
| 144 | ], |
| 145 | "role": "assistant", |
| 146 | "status": "completed", |
| 147 | "type": "message", |
| 148 | }, |
| 149 | { |
| 150 | "call_id": "2", |
| 151 | "output": "tool_result", |
| 152 | "type": "function_call_output", |
| 153 | }, |
| 154 | { |
| 155 | "id": "1", |
| 156 | "content": [ |
| 157 | {"annotations": [], "logprobs": [], "text": "done", "type": "output_text"} |
| 158 | ], |
| 159 | "role": "assistant", |
nothing calls this directly
no test coverage detected