wehub-resource-sync
db620d33df
CodeQL / Analyze (csharp) (push) Waiting to run
CodeQL / Analyze (python) (push) Waiting to run
dotnet-build-and-test / dotnet-test-functions (push) Has been cancelled
dotnet-build-and-test / paths-filter (push) Has been cancelled
dotnet-build-and-test / dotnet-build (Debug, windows-latest, net9.0) (push) Has been cancelled
dotnet-build-and-test / dotnet-build (Release, ubuntu-latest, net10.0) (push) Has been cancelled
dotnet-build-and-test / dotnet-build (Release, ubuntu-latest, net8.0) (push) Has been cancelled
dotnet-build-and-test / dotnet-build (Release, windows-latest, net472) (push) Has been cancelled
dotnet-build-and-test / dotnet-test (Release, integration, true, ubuntu-latest, net10.0) (push) Has been cancelled
dotnet-build-and-test / dotnet-test (Release, integration, true, windows-latest, net472) (push) Has been cancelled
dotnet-build-and-test / dotnet-foundry-hosted-it (push) Has been cancelled
dotnet-build-and-test / dotnet-build-and-test-check (push) Has been cancelled
dotnet-build-and-test / Integration Test Report (push) Has been cancelled
474 lines
19 KiB
Python
474 lines
19 KiB
Python
# Copyright (c) Microsoft. All rights reserved.
|
|
|
|
"""Unit tests for DurableAIAgentWorker.
|
|
|
|
Focuses on critical worker flows: agent registration, validation, callbacks, and lifecycle.
|
|
"""
|
|
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from agent_framework_durabletask import DurableAIAgentWorker
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_grpc_worker() -> Mock:
|
|
"""Create a mock TaskHubGrpcWorker for testing."""
|
|
mock = Mock()
|
|
mock.add_entity = Mock(return_value="dafx-test_agent")
|
|
mock.start = Mock()
|
|
mock.stop = Mock()
|
|
return mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_agent() -> Mock:
|
|
"""Create a mock agent for testing."""
|
|
agent = Mock()
|
|
agent.name = "test_agent"
|
|
return agent
|
|
|
|
|
|
@pytest.fixture
|
|
def agent_worker(mock_grpc_worker: Mock) -> DurableAIAgentWorker:
|
|
"""Create a DurableAIAgentWorker with mock worker."""
|
|
return DurableAIAgentWorker(mock_grpc_worker)
|
|
|
|
|
|
class TestDurableAIAgentWorkerRegistration:
|
|
"""Test agent registration behavior."""
|
|
|
|
def test_add_agent_accepts_agent_with_name(
|
|
self, agent_worker: DurableAIAgentWorker, mock_agent: Mock, mock_grpc_worker: Mock
|
|
) -> None:
|
|
"""Verify that agents with names can be registered."""
|
|
agent_worker.add_agent(mock_agent)
|
|
|
|
# Verify entity was registered with underlying worker
|
|
mock_grpc_worker.add_entity.assert_called_once()
|
|
# Verify agent name is tracked
|
|
assert "test_agent" in agent_worker.registered_agent_names
|
|
|
|
def test_add_agent_rejects_agent_without_name(self, agent_worker: DurableAIAgentWorker) -> None:
|
|
"""Verify that agents without names are rejected."""
|
|
agent_no_name = Mock()
|
|
agent_no_name.name = None
|
|
|
|
with pytest.raises(ValueError, match="Agent must have a name"):
|
|
agent_worker.add_agent(agent_no_name)
|
|
|
|
def test_add_agent_rejects_empty_name(self, agent_worker: DurableAIAgentWorker) -> None:
|
|
"""Verify that agents with empty names are rejected."""
|
|
agent_empty_name = Mock()
|
|
agent_empty_name.name = ""
|
|
|
|
with pytest.raises(ValueError, match="Agent must have a name"):
|
|
agent_worker.add_agent(agent_empty_name)
|
|
|
|
def test_add_agent_rejects_duplicate_names(self, agent_worker: DurableAIAgentWorker, mock_agent: Mock) -> None:
|
|
"""Verify duplicate agent names are not allowed."""
|
|
agent_worker.add_agent(mock_agent)
|
|
|
|
# Try to register another agent with the same name
|
|
duplicate_agent = Mock()
|
|
duplicate_agent.name = "test_agent"
|
|
|
|
with pytest.raises(ValueError, match="already registered"):
|
|
agent_worker.add_agent(duplicate_agent)
|
|
|
|
def test_registered_agent_names_tracks_multiple_agents(self, agent_worker: DurableAIAgentWorker) -> None:
|
|
"""Verify registered_agent_names tracks all registered agents."""
|
|
agent1 = Mock()
|
|
agent1.name = "agent1"
|
|
agent2 = Mock()
|
|
agent2.name = "agent2"
|
|
agent3 = Mock()
|
|
agent3.name = "agent3"
|
|
|
|
agent_worker.add_agent(agent1)
|
|
agent_worker.add_agent(agent2)
|
|
agent_worker.add_agent(agent3)
|
|
|
|
registered = agent_worker.registered_agent_names
|
|
assert "agent1" in registered
|
|
assert "agent2" in registered
|
|
assert "agent3" in registered
|
|
assert len(registered) == 3
|
|
|
|
|
|
class TestDurableAIAgentWorkerCallbacks:
|
|
"""Test callback configuration behavior."""
|
|
|
|
def test_worker_level_callback_accepted(self, mock_grpc_worker: Mock) -> None:
|
|
"""Verify worker-level callback can be set."""
|
|
mock_callback = Mock()
|
|
agent_worker = DurableAIAgentWorker(mock_grpc_worker, callback=mock_callback)
|
|
|
|
assert agent_worker is not None
|
|
|
|
def test_agent_level_callback_accepted(self, agent_worker: DurableAIAgentWorker, mock_agent: Mock) -> None:
|
|
"""Verify agent-level callback can be set during registration."""
|
|
mock_callback = Mock()
|
|
|
|
# Should not raise exception
|
|
agent_worker.add_agent(mock_agent, callback=mock_callback)
|
|
|
|
assert "test_agent" in agent_worker.registered_agent_names
|
|
|
|
def test_none_callback_accepted(self, mock_grpc_worker: Mock, mock_agent: Mock) -> None:
|
|
"""Verify None callback is valid (no callbacks required)."""
|
|
agent_worker = DurableAIAgentWorker(mock_grpc_worker, callback=None)
|
|
agent_worker.add_agent(mock_agent, callback=None)
|
|
|
|
assert "test_agent" in agent_worker.registered_agent_names
|
|
|
|
|
|
class TestDurableAIAgentWorkerLifecycle:
|
|
"""Test worker lifecycle behavior."""
|
|
|
|
def test_start_delegates_to_underlying_worker(
|
|
self, agent_worker: DurableAIAgentWorker, mock_grpc_worker: Mock
|
|
) -> None:
|
|
"""Verify start() delegates to wrapped worker."""
|
|
agent_worker.start()
|
|
|
|
mock_grpc_worker.start.assert_called_once()
|
|
|
|
def test_stop_delegates_to_underlying_worker(
|
|
self, agent_worker: DurableAIAgentWorker, mock_grpc_worker: Mock
|
|
) -> None:
|
|
"""Verify stop() delegates to wrapped worker."""
|
|
agent_worker.stop()
|
|
|
|
mock_grpc_worker.stop.assert_called_once()
|
|
|
|
def test_start_works_with_no_agents(self, agent_worker: DurableAIAgentWorker, mock_grpc_worker: Mock) -> None:
|
|
"""Verify worker can start even with no agents registered."""
|
|
agent_worker.start()
|
|
|
|
mock_grpc_worker.start.assert_called_once()
|
|
|
|
def test_start_works_with_multiple_agents(self, agent_worker: DurableAIAgentWorker, mock_grpc_worker: Mock) -> None:
|
|
"""Verify worker can start with multiple agents registered."""
|
|
agent1 = Mock()
|
|
agent1.name = "agent1"
|
|
agent2 = Mock()
|
|
agent2.name = "agent2"
|
|
|
|
agent_worker.add_agent(agent1)
|
|
agent_worker.add_agent(agent2)
|
|
agent_worker.start()
|
|
|
|
mock_grpc_worker.start.assert_called_once()
|
|
assert len(agent_worker.registered_agent_names) == 2
|
|
|
|
|
|
class TestDurableAIAgentWorkerWorkflow:
|
|
"""Test workflow registration, including the agent-executor identity fix."""
|
|
|
|
def test_add_agent_with_entity_id_registers_under_override(
|
|
self, agent_worker: DurableAIAgentWorker, mock_agent: Mock
|
|
) -> None:
|
|
"""An explicit entity_id overrides the agent name as the entity identity."""
|
|
agent_worker.add_agent(mock_agent, entity_id="node-7")
|
|
|
|
assert "node-7" in agent_worker.registered_agent_names
|
|
assert "test_agent" not in agent_worker.registered_agent_names
|
|
|
|
def test_configure_workflow_registers_agent_entity_by_executor_id(
|
|
self, agent_worker: DurableAIAgentWorker, mock_grpc_worker: Mock
|
|
) -> None:
|
|
"""Workflow agent executors register entities keyed by the workflow-scoped id.
|
|
|
|
The orchestrator dispatches by the scoped identity
|
|
``{workflow}-{executorId}``, so an ``AgentExecutor(agent, id=...)`` whose id
|
|
differs from the agent name must still be reachable under that scoped id.
|
|
"""
|
|
from agent_framework import AgentExecutor
|
|
|
|
agent = Mock()
|
|
agent.name = "Reviewer"
|
|
agent_executor = Mock(spec=AgentExecutor)
|
|
agent_executor.id = "custom-executor-id"
|
|
agent_executor.agent = agent
|
|
|
|
workflow = Mock()
|
|
workflow.name = "review"
|
|
workflow.executors = {"custom-executor-id": agent_executor}
|
|
|
|
agent_worker.configure_workflow(workflow)
|
|
|
|
assert "review-custom-executor-id" in agent_worker.registered_agent_names
|
|
assert "Reviewer" not in agent_worker.registered_agent_names
|
|
assert "custom-executor-id" not in agent_worker.registered_agent_names
|
|
mock_grpc_worker.add_orchestrator.assert_called_once()
|
|
|
|
def test_configure_workflow_registers_non_agent_executor_as_activity(
|
|
self, agent_worker: DurableAIAgentWorker, mock_grpc_worker: Mock
|
|
) -> None:
|
|
"""Non-agent executors are registered as activities, not entities."""
|
|
from agent_framework import Executor
|
|
|
|
activity_executor = Mock(spec=Executor)
|
|
activity_executor.id = "router-node"
|
|
|
|
workflow = Mock()
|
|
workflow.name = "route"
|
|
workflow.executors = {"router-node": activity_executor}
|
|
|
|
agent_worker.configure_workflow(workflow)
|
|
|
|
assert agent_worker.registered_agent_names == []
|
|
mock_grpc_worker.add_activity.assert_called_once()
|
|
mock_grpc_worker.add_orchestrator.assert_called_once()
|
|
# The activity is registered under the workflow-scoped name.
|
|
registered_activity = mock_grpc_worker.add_activity.call_args[0][0]
|
|
assert registered_activity.__name__ == "dafx-route-router-node"
|
|
|
|
|
|
class TestMultiWorkflowRegistration:
|
|
"""Test hosting multiple workflows on one worker with scoped names."""
|
|
|
|
def _agent_workflow(self, name: str, executor_id: str) -> Mock:
|
|
from agent_framework import AgentExecutor
|
|
|
|
agent = Mock()
|
|
agent.name = "Assistant"
|
|
agent_executor = Mock(spec=AgentExecutor)
|
|
agent_executor.id = executor_id
|
|
agent_executor.agent = agent
|
|
|
|
workflow = Mock()
|
|
workflow.name = name
|
|
workflow.executors = {executor_id: agent_executor}
|
|
return workflow
|
|
|
|
def test_two_workflows_reusing_executor_id_do_not_collide(self, agent_worker: DurableAIAgentWorker) -> None:
|
|
"""Two workflows that reuse an executor id register distinct scoped entities."""
|
|
agent_worker.configure_workflow(self._agent_workflow("orders", "assistant"))
|
|
agent_worker.configure_workflow(self._agent_workflow("billing", "assistant"))
|
|
|
|
assert "orders-assistant" in agent_worker.registered_agent_names
|
|
assert "billing-assistant" in agent_worker.registered_agent_names
|
|
assert set(agent_worker.registered_workflow_names) == {"orders", "billing"}
|
|
|
|
def test_registers_one_orchestrator_per_workflow(
|
|
self, agent_worker: DurableAIAgentWorker, mock_grpc_worker: Mock
|
|
) -> None:
|
|
"""Each configured workflow registers its own orchestrator."""
|
|
agent_worker.configure_workflow(self._agent_workflow("orders", "a"))
|
|
agent_worker.configure_workflow(self._agent_workflow("billing", "b"))
|
|
|
|
assert mock_grpc_worker.add_orchestrator.call_count == 2
|
|
registered_names = {call.args[0].__name__ for call in mock_grpc_worker.add_orchestrator.call_args_list}
|
|
assert registered_names == {"dafx-orders", "dafx-billing"}
|
|
|
|
def test_rejects_duplicate_workflow_name(self, agent_worker: DurableAIAgentWorker) -> None:
|
|
"""Configuring two workflows with the same name is rejected."""
|
|
agent_worker.configure_workflow(self._agent_workflow("orders", "a"))
|
|
|
|
with pytest.raises(ValueError, match="already registered"):
|
|
agent_worker.configure_workflow(self._agent_workflow("orders", "b"))
|
|
|
|
def test_rejects_case_insensitive_duplicate_workflow_name(self, agent_worker: DurableAIAgentWorker) -> None:
|
|
"""Workflow names that differ only by case collide and are rejected.
|
|
|
|
The route ownership guard folds case, so allowing both ``orders`` and
|
|
``Orders`` would let one workflow's surface reach the other's instances.
|
|
"""
|
|
agent_worker.configure_workflow(self._agent_workflow("orders", "a"))
|
|
|
|
with pytest.raises(ValueError, match="case-insensitively"):
|
|
agent_worker.configure_workflow(self._agent_workflow("Orders", "b"))
|
|
|
|
def test_rejects_auto_generated_workflow_name(self, agent_worker: DurableAIAgentWorker) -> None:
|
|
"""A workflow with an auto-generated WorkflowBuilder name is rejected."""
|
|
import uuid
|
|
|
|
workflow = self._agent_workflow(f"WorkflowBuilder-{uuid.uuid4()}", "a")
|
|
|
|
with pytest.raises(ValueError, match="auto-generated"):
|
|
agent_worker.configure_workflow(workflow)
|
|
|
|
def test_rejects_invalid_workflow_name(self, agent_worker: DurableAIAgentWorker) -> None:
|
|
"""A workflow with an invalid name is rejected."""
|
|
workflow = self._agent_workflow("has space", "a")
|
|
|
|
with pytest.raises(ValueError, match="invalid"):
|
|
agent_worker.configure_workflow(workflow)
|
|
|
|
|
|
class TestSubworkflowRegistration:
|
|
"""Test recursive registration of nested sub-workflows on one worker."""
|
|
|
|
def _inner_agent_workflow(self, name: str, executor_id: str) -> Mock:
|
|
from agent_framework import AgentExecutor
|
|
|
|
agent = Mock()
|
|
agent.name = "InnerAssistant"
|
|
agent_executor = Mock(spec=AgentExecutor)
|
|
agent_executor.id = executor_id
|
|
agent_executor.agent = agent
|
|
|
|
workflow = Mock()
|
|
workflow.name = name
|
|
workflow.executors = {executor_id: agent_executor}
|
|
return workflow
|
|
|
|
def _outer_workflow(self, name: str, inner: Mock, *, sub_ids: tuple[str, ...] = ("sub",)) -> Mock:
|
|
from agent_framework import Executor, WorkflowExecutor
|
|
|
|
executors: dict[str, Mock] = {}
|
|
for sub_id in sub_ids:
|
|
sub = Mock(spec=WorkflowExecutor)
|
|
sub.id = sub_id
|
|
sub.workflow = inner
|
|
sub.allow_direct_output = False
|
|
executors[sub_id] = sub
|
|
|
|
router = Mock(spec=Executor)
|
|
router.id = "router"
|
|
executors["router"] = router
|
|
|
|
workflow = Mock()
|
|
workflow.name = name
|
|
workflow.executors = executors
|
|
return workflow
|
|
|
|
def test_nested_workflow_registers_both_orchestrations(
|
|
self, agent_worker: DurableAIAgentWorker, mock_grpc_worker: Mock
|
|
) -> None:
|
|
"""Configuring an outer workflow registers the inner workflow's orchestration too."""
|
|
inner = self._inner_agent_workflow("inner", "agent_node")
|
|
outer = self._outer_workflow("outer", inner)
|
|
|
|
agent_worker.configure_workflow(outer)
|
|
|
|
registered = {call.args[0].__name__ for call in mock_grpc_worker.add_orchestrator.call_args_list}
|
|
assert registered == {"dafx-outer", "dafx-inner"}
|
|
|
|
def test_nested_workflow_registers_inner_agent_scoped(self, agent_worker: DurableAIAgentWorker) -> None:
|
|
"""The inner workflow's agent is registered under the inner-scoped id."""
|
|
inner = self._inner_agent_workflow("inner", "agent_node")
|
|
outer = self._outer_workflow("outer", inner)
|
|
|
|
agent_worker.configure_workflow(outer)
|
|
|
|
assert "inner-agent_node" in agent_worker.registered_agent_names
|
|
|
|
def test_subworkflow_node_not_registered_as_activity(
|
|
self, agent_worker: DurableAIAgentWorker, mock_grpc_worker: Mock
|
|
) -> None:
|
|
"""A WorkflowExecutor node is driven as a child orchestration, not an activity."""
|
|
inner = self._inner_agent_workflow("inner", "agent_node")
|
|
outer = self._outer_workflow("outer", inner)
|
|
|
|
agent_worker.configure_workflow(outer)
|
|
|
|
# Only the outer 'router' non-agent executor becomes an activity.
|
|
registered_activities = {call.args[0].__name__ for call in mock_grpc_worker.add_activity.call_args_list}
|
|
assert registered_activities == {"dafx-outer-router"}
|
|
|
|
def test_top_level_names_exclude_nested_workflows(self, agent_worker: DurableAIAgentWorker) -> None:
|
|
"""``registered_workflow_names`` reports only top-level workflows."""
|
|
inner = self._inner_agent_workflow("inner", "agent_node")
|
|
outer = self._outer_workflow("outer", inner)
|
|
|
|
agent_worker.configure_workflow(outer)
|
|
|
|
assert agent_worker.registered_workflow_names == ["outer"]
|
|
|
|
def test_shared_subworkflow_registered_once(
|
|
self, agent_worker: DurableAIAgentWorker, mock_grpc_worker: Mock
|
|
) -> None:
|
|
"""A sub-workflow reused by two nodes registers its orchestration only once."""
|
|
inner = self._inner_agent_workflow("inner", "agent_node")
|
|
outer = self._outer_workflow("outer", inner, sub_ids=("sub_a", "sub_b"))
|
|
|
|
agent_worker.configure_workflow(outer)
|
|
|
|
registered = [call.args[0].__name__ for call in mock_grpc_worker.add_orchestrator.call_args_list]
|
|
assert sorted(registered) == ["dafx-inner", "dafx-outer"]
|
|
|
|
def test_nested_workflow_with_invalid_name_is_rejected(self, agent_worker: DurableAIAgentWorker) -> None:
|
|
"""A nested sub-workflow must also have a valid, stable name."""
|
|
inner = self._inner_agent_workflow("has space", "agent_node")
|
|
outer = self._outer_workflow("outer", inner)
|
|
|
|
with pytest.raises(ValueError, match="invalid"):
|
|
agent_worker.configure_workflow(outer)
|
|
|
|
def test_different_subworkflow_sharing_a_name_is_rejected(self, agent_worker: DurableAIAgentWorker) -> None:
|
|
"""Two different sub-workflow instances that share a name collide and are rejected."""
|
|
from agent_framework import Executor, WorkflowExecutor
|
|
|
|
inner_a = self._inner_agent_workflow("shared", "agent_node")
|
|
inner_b = self._inner_agent_workflow("shared", "other_node") # different instance, same name
|
|
|
|
sub_a = Mock(spec=WorkflowExecutor)
|
|
sub_a.id = "a"
|
|
sub_a.workflow = inner_a
|
|
sub_b = Mock(spec=WorkflowExecutor)
|
|
sub_b.id = "b"
|
|
sub_b.workflow = inner_b
|
|
router = Mock(spec=Executor)
|
|
router.id = "router"
|
|
outer = Mock()
|
|
outer.name = "outer"
|
|
outer.executors = {"a": sub_a, "b": sub_b, "router": router}
|
|
|
|
with pytest.raises(ValueError, match="different workflow|different workflows"):
|
|
agent_worker.configure_workflow(outer)
|
|
|
|
def test_cross_registration_nested_collision_is_atomic(
|
|
self, agent_worker: DurableAIAgentWorker, mock_grpc_worker: Mock
|
|
) -> None:
|
|
"""A later configure_workflow whose nested child collides leaves the worker unchanged.
|
|
|
|
Reproduces the partial-registration path: configure one workflow, then configure
|
|
a second whose nested sub-workflow reuses the first's child name. The second call
|
|
must raise *before* mutating any state, so the second top-level workflow is not
|
|
left half-registered (which would also wedge a corrected retry on the duplicate
|
|
guard).
|
|
"""
|
|
shared_a = self._inner_agent_workflow("shared", "agent_node")
|
|
agent_worker.configure_workflow(self._outer_workflow("first", shared_a))
|
|
|
|
orchestrators_before = mock_grpc_worker.add_orchestrator.call_count
|
|
|
|
# A *different* 'shared' instance nested under a new top-level workflow collides.
|
|
shared_b = self._inner_agent_workflow("shared", "other_node")
|
|
with pytest.raises(ValueError, match="collides"):
|
|
agent_worker.configure_workflow(self._outer_workflow("second", shared_b))
|
|
|
|
# The worker is not partially configured: 'second' was never added, and no new
|
|
# orchestration was registered.
|
|
assert agent_worker.registered_workflow_names == ["first"]
|
|
assert mock_grpc_worker.add_orchestrator.call_count == orchestrators_before
|
|
|
|
def test_executor_id_with_reserved_separator_is_rejected(self, agent_worker: DurableAIAgentWorker) -> None:
|
|
"""An executor id containing the nested-HITL separator is rejected at registration."""
|
|
workflow = self._agent_workflow_with_executor_id("orders", "bad~id")
|
|
|
|
with pytest.raises(ValueError, match="reserved sub-workflow request separator"):
|
|
agent_worker.configure_workflow(workflow)
|
|
|
|
@staticmethod
|
|
def _agent_workflow_with_executor_id(name: str, executor_id: str) -> Mock:
|
|
from agent_framework import AgentExecutor
|
|
|
|
agent = Mock()
|
|
agent.name = "Assistant"
|
|
agent_executor = Mock(spec=AgentExecutor)
|
|
agent_executor.id = executor_id
|
|
agent_executor.agent = agent
|
|
workflow = Mock()
|
|
workflow.name = name
|
|
workflow.executors = {executor_id: agent_executor}
|
|
return workflow
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v", "--tb=short"])
|