mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
bb3d3c2efc
* Add workflow support for Azure Functions * fix compatability with latest framework changes and add integration tests * refactor code * remove white space Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * align help text with actual port used Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * replace instance id with a place holder Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * remove unused import Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * remove redundant typing import and fix SIM115 * fix latest breaking changes * fix mypy issues * clean up imports * define source marker strings as constants * fix json module name * refactor _extract_message_content_from_dict * refactor serialization * add helper method for error response construction and remove _extract_message_content_from_dict since it is not needed * use strict tpe checking for edges * change how duplicate agent registrations are handled * cancel approval_task on HITL timeout * update docstring * fix: align azurefunctions package with core API changes after rebase - State.import_state/export_state are now sync (removed await) - Add State.commit() before export_state() in activity execution - Rename executor parameter shared_state -> state - Rename ctx.set_shared_state/get_shared_state -> set_state/get_state (sync) - WorkflowBuilder now takes start_executor as constructor kwarg - Update WorkflowOutputEvent -> WorkflowEvent with type='output' - Update RequestInfoEvent -> WorkflowEvent[Any] - Update SharedState -> State in test imports - Update duplicate agent name tests to match new warning behavior - Update sample README API references * fix sample check errors * fix mypy issues * fix trailing white spaces * fix test imports * feat: add durable workflow samples and adapt to main branch changes - Add workflow samples 09-12 to 04-hosting/azure_functions/ - Adapt to ChatMessage -> Message rename from main - Adapt to pickle-based checkpoint encoding from main - Simplify _serialization.py to delegate to core encode/decode - Fix Message -> WorkflowMessage disambiguation in _context.py - Remove non-existent _checkpoint_summary import * fix: update create_checkpoint signature to match superclass * fix: correct relative link in HITL sample README * fix: resolve import breakage after rebase (State, DurableAgentThread, get_logger) --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: Dmytro Struk <13853051+dmytrostruk@users.noreply.github.com>
156 lines
4.7 KiB
Python
156 lines
4.7 KiB
Python
# Copyright (c) Microsoft. All rights reserved.
|
|
|
|
"""Unit tests for multi-agent support in AgentFunctionApp."""
|
|
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from agent_framework_azurefunctions import AgentFunctionApp
|
|
|
|
|
|
class TestMultiAgentInit:
|
|
"""Test suite for multi-agent initialization."""
|
|
|
|
def test_init_with_agents_list(self) -> None:
|
|
"""Test initialization with list of agents."""
|
|
agent1 = Mock()
|
|
agent1.name = "Agent1"
|
|
agent2 = Mock()
|
|
agent2.name = "Agent2"
|
|
|
|
app = AgentFunctionApp(agents=[agent1, agent2])
|
|
|
|
assert len(app.agents) == 2
|
|
assert "Agent1" in app.agents
|
|
assert "Agent2" in app.agents
|
|
assert app.agents["Agent1"] == agent1
|
|
assert app.agents["Agent2"] == agent2
|
|
|
|
def test_init_with_empty_agents_list(self) -> None:
|
|
"""Test initialization with empty list of agents."""
|
|
app = AgentFunctionApp(agents=[])
|
|
|
|
assert len(app.agents) == 0
|
|
|
|
def test_init_with_no_agents(self) -> None:
|
|
"""Test initialization without any agents."""
|
|
app = AgentFunctionApp()
|
|
|
|
assert len(app.agents) == 0
|
|
|
|
def test_init_with_duplicate_agent_names(self) -> None:
|
|
"""Test initialization with duplicate agent names deduplicates with warning."""
|
|
agent1 = Mock()
|
|
agent1.name = "TestAgent"
|
|
agent2 = Mock()
|
|
agent2.name = "TestAgent"
|
|
|
|
app = AgentFunctionApp(agents=[agent1, agent2])
|
|
|
|
# Duplicate is skipped, only the first agent is registered
|
|
assert len(app.agents) == 1
|
|
assert "TestAgent" in app.agents
|
|
|
|
def test_init_with_agent_without_name(self) -> None:
|
|
"""Test initialization with agent missing name attribute raises error."""
|
|
agent1 = Mock()
|
|
agent1.name = "Agent1"
|
|
agent2 = Mock(spec=[]) # Mock without name attribute
|
|
|
|
with pytest.raises(ValueError, match="does not have a 'name' attribute"):
|
|
AgentFunctionApp(agents=[agent1, agent2])
|
|
|
|
|
|
class TestAddAgentMethod:
|
|
"""Test suite for add_agent() method."""
|
|
|
|
def test_add_agent_to_empty_app(self) -> None:
|
|
"""Test adding agent to app initialized without agents."""
|
|
app = AgentFunctionApp()
|
|
|
|
agent = Mock()
|
|
agent.name = "NewAgent"
|
|
|
|
app.add_agent(agent)
|
|
|
|
assert len(app.agents) == 1
|
|
assert "NewAgent" in app.agents
|
|
assert app.agents["NewAgent"] == agent
|
|
|
|
def test_add_multiple_agents(self) -> None:
|
|
"""Test adding multiple agents sequentially."""
|
|
app = AgentFunctionApp()
|
|
|
|
agent1 = Mock()
|
|
agent1.name = "Agent1"
|
|
agent2 = Mock()
|
|
agent2.name = "Agent2"
|
|
|
|
app.add_agent(agent1)
|
|
app.add_agent(agent2)
|
|
|
|
assert len(app.agents) == 2
|
|
assert "Agent1" in app.agents
|
|
assert "Agent2" in app.agents
|
|
|
|
def test_add_agent_with_duplicate_name_skips(self) -> None:
|
|
"""Test that adding agent with duplicate name logs warning and skips."""
|
|
agent1 = Mock()
|
|
agent1.name = "MyAgent"
|
|
agent2 = Mock()
|
|
agent2.name = "MyAgent"
|
|
|
|
app = AgentFunctionApp(agents=[agent1])
|
|
|
|
# Duplicate is silently skipped with a warning
|
|
app.add_agent(agent2)
|
|
|
|
# Only the original agent remains
|
|
assert len(app.agents) == 1
|
|
|
|
def test_add_agent_to_app_with_existing_agents(self) -> None:
|
|
"""Test adding agent to app that already has agents."""
|
|
agent1 = Mock()
|
|
agent1.name = "Agent1"
|
|
agent2 = Mock()
|
|
agent2.name = "Agent2"
|
|
|
|
app = AgentFunctionApp(agents=[agent1])
|
|
app.add_agent(agent2)
|
|
|
|
assert len(app.agents) == 2
|
|
assert "Agent1" in app.agents
|
|
assert "Agent2" in app.agents
|
|
|
|
def test_add_agent_without_name_raises_error(self) -> None:
|
|
"""Test that adding agent without name attribute raises error."""
|
|
app = AgentFunctionApp()
|
|
|
|
agent = Mock(spec=[]) # Mock without name attribute
|
|
|
|
with pytest.raises(ValueError, match="does not have a 'name' attribute"):
|
|
app.add_agent(agent)
|
|
|
|
|
|
class TestHealthCheckWithMultipleAgents:
|
|
"""Test suite for health check with multiple agents."""
|
|
|
|
def test_health_check_returns_all_agents(self) -> None:
|
|
"""Test that health check returns information about all agents."""
|
|
agent1 = Mock()
|
|
agent1.name = "Agent1"
|
|
agent2 = Mock()
|
|
agent2.name = "Agent2"
|
|
|
|
app = AgentFunctionApp(agents=[agent1, agent2])
|
|
|
|
# Note: We can't easily test the actual health check endpoint without running the app
|
|
# But we can verify the agents dictionary is properly populated
|
|
assert len(app.agents) == 2
|
|
assert app.enable_health_check is True
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v", "--tb=short"])
|