Files
Ahmed Muhsin bb3d3c2efc Python: Durable Support for Workflows (#3630)
* 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>
2026-02-17 22:11:33 +00:00

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"])