Python: Add declarative workflow runtime (#2815)

* Further support for declarative python workflows * Add tests. Clean up for typing and formatting * Improvements and cleanup * Typing cleanup. Improve docstrings * Proper code in docstrings * Fix malformed code-block directive in docstring * Remove dead links * PR feedback * Address PR feedback * Address PR feedback * Remove sl * Update devui frontend * More cleanup * Fix uv lock * Skip Py 3.14 tests as powerfx doesn't support it * Fix mypy error * Fix for tool calls * Removed stale docstring * Fix lint * Standardize on .NET namespaces. Revert DevUI changes (bring in later) * Implement remaining items for Python declarative support to match dotnet
2026-06-16 21:04:09 +08:00 · 2026-01-13 16:11:21 +09:00
parent b2893fbc00
commit 9c094573e8
79 changed files with 18310 additions and 111 deletions
@@ -0,0 +1,286 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for human-in-the-loop action handlers."""
+
+import pytest
+
+from agent_framework_declarative._workflows._handlers import ActionContext, get_action_handler
+from agent_framework_declarative._workflows._human_input import (
+    QuestionRequest,
+    process_external_loop,
+    validate_input_response,
+)
+from agent_framework_declarative._workflows._state import WorkflowState
+
+
+def create_action_context(action: dict, state: WorkflowState | None = None):
+    """Create a minimal action context for testing."""
+    if state is None:
+        state = WorkflowState()
+
+    async def execute_actions(actions, state):
+        for act in actions:
+            handler = get_action_handler(act.get("kind"))
+            if handler:
+                async for event in handler(
+                    ActionContext(
+                        state=state,
+                        action=act,
+                        execute_actions=execute_actions,
+                        agents={},
+                        bindings={},
+                    )
+                ):
+                    yield event
+
+    return ActionContext(
+        state=state,
+        action=action,
+        execute_actions=execute_actions,
+        agents={},
+        bindings={},
+    )
+
+
+class TestQuestionHandler:
+    """Tests for Question action handler."""
+
+    @pytest.mark.asyncio
+    async def test_question_emits_request_info_event(self):
+        """Test that Question handler emits QuestionRequest."""
+        ctx = create_action_context({
+            "kind": "Question",
+            "id": "ask_name",
+            "variable": "Local.userName",
+            "prompt": "What is your name?",
+        })
+
+        handler = get_action_handler("Question")
+        events = [e async for e in handler(ctx)]
+
+        assert len(events) == 1
+        assert isinstance(events[0], QuestionRequest)
+        assert events[0].request_id == "ask_name"
+        assert events[0].prompt == "What is your name?"
+        assert events[0].variable == "Local.userName"
+
+    @pytest.mark.asyncio
+    async def test_question_with_choices(self):
+        """Test Question with multiple choice options."""
+        ctx = create_action_context({
+            "kind": "Question",
+            "id": "ask_choice",
+            "variable": "Local.selection",
+            "prompt": "Select an option:",
+            "choices": ["Option A", "Option B", "Option C"],
+            "default": "Option A",
+        })
+
+        handler = get_action_handler("Question")
+        events = [e async for e in handler(ctx)]
+
+        assert len(events) == 1
+        event = events[0]
+        assert isinstance(event, QuestionRequest)
+        assert event.choices == ["Option A", "Option B", "Option C"]
+        assert event.default_value == "Option A"
+
+    @pytest.mark.asyncio
+    async def test_question_with_validation(self):
+        """Test Question with validation rules."""
+        ctx = create_action_context({
+            "kind": "Question",
+            "id": "ask_email",
+            "variable": "Local.email",
+            "prompt": "Enter your email:",
+            "validation": {
+                "required": True,
+                "pattern": r"^[\w\.-]+@[\w\.-]+\.\w+$",
+            },
+        })
+
+        handler = get_action_handler("Question")
+        events = [e async for e in handler(ctx)]
+
+        assert len(events) == 1
+        event = events[0]
+        assert event.validation == {
+            "required": True,
+            "pattern": r"^[\w\.-]+@[\w\.-]+\.\w+$",
+        }
+
+
+class TestRequestExternalInputHandler:
+    """Tests for RequestExternalInput action handler."""
+
+    @pytest.mark.asyncio
+    async def test_request_external_input(self):
+        """Test RequestExternalInput handler emits event."""
+        ctx = create_action_context({
+            "kind": "RequestExternalInput",
+            "id": "get_approval",
+            "variable": "Local.approval",
+            "prompt": "Please approve or reject",
+            "timeout": 300,
+        })
+
+        handler = get_action_handler("RequestExternalInput")
+        events = [e async for e in handler(ctx)]
+
+        assert len(events) == 1
+        event = events[0]
+        assert isinstance(event, QuestionRequest)
+        assert event.request_id == "get_approval"
+        assert event.variable == "Local.approval"
+        assert event.validation == {"timeout": 300}
+
+
+class TestWaitForInputHandler:
+    """Tests for WaitForInput action handler."""
+
+    @pytest.mark.asyncio
+    async def test_wait_for_input(self):
+        """Test WaitForInput handler."""
+        ctx = create_action_context({
+            "kind": "WaitForInput",
+            "id": "wait",
+            "variable": "Local.response",
+            "message": "Waiting...",
+        })
+
+        handler = get_action_handler("WaitForInput")
+        events = [e async for e in handler(ctx)]
+
+        assert len(events) == 1
+        event = events[0]
+        assert isinstance(event, QuestionRequest)
+        assert event.request_id == "wait"
+        assert event.prompt == "Waiting..."
+
+
+class TestProcessExternalLoop:
+    """Tests for process_external_loop helper function."""
+
+    def test_no_external_loop(self):
+        """Test when no external loop is configured."""
+        state = WorkflowState()
+        result, expr = process_external_loop({}, state)
+
+        assert result is False
+        assert expr is None
+
+    def test_external_loop_true_condition(self):
+        """Test when external loop condition evaluates to true."""
+        state = WorkflowState()
+        state.set("Local.isComplete", False)
+
+        input_config = {
+            "externalLoop": {
+                "when": "=!Local.isComplete",
+            },
+        }
+
+        result, expr = process_external_loop(input_config, state)
+
+        # !False = True, so loop should continue
+        assert result is True
+        assert expr == "=!Local.isComplete"
+
+    def test_external_loop_false_condition(self):
+        """Test when external loop condition evaluates to false."""
+        state = WorkflowState()
+        state.set("Local.isComplete", True)
+
+        input_config = {
+            "externalLoop": {
+                "when": "=!Local.isComplete",
+            },
+        }
+
+        result, expr = process_external_loop(input_config, state)
+
+        # !True = False, so loop should stop
+        assert result is False
+
+
+class TestValidateInputResponse:
+    """Tests for validate_input_response helper function."""
+
+    def test_no_validation(self):
+        """Test with no validation rules."""
+        is_valid, error = validate_input_response("any value", None)
+        assert is_valid is True
+        assert error is None
+
+    def test_required_valid(self):
+        """Test required validation with valid value."""
+        is_valid, error = validate_input_response("value", {"required": True})
+        assert is_valid is True
+        assert error is None
+
+    def test_required_empty_string(self):
+        """Test required validation with empty string."""
+        is_valid, error = validate_input_response("", {"required": True})
+        assert is_valid is False
+        assert "required" in error.lower()
+
+    def test_required_none(self):
+        """Test required validation with None."""
+        is_valid, error = validate_input_response(None, {"required": True})
+        assert is_valid is False
+        assert "required" in error.lower()
+
+    def test_min_length_valid(self):
+        """Test minLength validation with valid value."""
+        is_valid, error = validate_input_response("hello", {"minLength": 3})
+        assert is_valid is True
+
+    def test_min_length_invalid(self):
+        """Test minLength validation with too short value."""
+        is_valid, error = validate_input_response("hi", {"minLength": 3})
+        assert is_valid is False
+        assert "minimum length" in error.lower()
+
+    def test_max_length_valid(self):
+        """Test maxLength validation with valid value."""
+        is_valid, error = validate_input_response("hello", {"maxLength": 10})
+        assert is_valid is True
+
+    def test_max_length_invalid(self):
+        """Test maxLength validation with too long value."""
+        is_valid, error = validate_input_response("hello world", {"maxLength": 5})
+        assert is_valid is False
+        assert "maximum length" in error.lower()
+
+    def test_min_value_valid(self):
+        """Test min validation for numbers."""
+        is_valid, error = validate_input_response(10, {"min": 5})
+        assert is_valid is True
+
+    def test_min_value_invalid(self):
+        """Test min validation with too small number."""
+        is_valid, error = validate_input_response(3, {"min": 5})
+        assert is_valid is False
+        assert "minimum value" in error.lower()
+
+    def test_max_value_valid(self):
+        """Test max validation for numbers."""
+        is_valid, error = validate_input_response(5, {"max": 10})
+        assert is_valid is True
+
+    def test_max_value_invalid(self):
+        """Test max validation with too large number."""
+        is_valid, error = validate_input_response(15, {"max": 10})
+        assert is_valid is False
+        assert "maximum value" in error.lower()
+
+    def test_pattern_valid(self):
+        """Test pattern validation with matching value."""
+        is_valid, error = validate_input_response("test@example.com", {"pattern": r"^[\w\.-]+@[\w\.-]+\.\w+$"})
+        assert is_valid is True
+
+    def test_pattern_invalid(self):
+        """Test pattern validation with non-matching value."""
+        is_valid, error = validate_input_response("not-an-email", {"pattern": r"^[\w\.-]+@[\w\.-]+\.\w+$"})
+        assert is_valid is False
+        assert "pattern" in error.lower()