agent-framework/python/packages/declarative/tests/test_external_input.py

# 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()