leocrm/tests/test_phase_g_workflows.py

"""Tests for Phase G — Workflow MVP: step handlers, wait/resume, retry, SSRF, triggers.

All tests use mocks — no real DB/LLM/Redis/HTTP needed.
"""

from __future__ import annotations

import asyncio
import uuid
from datetime import UTC, datetime, timedelta
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from app.workflows.step_handlers import (
    StepResult,
    get_step_handler,
    get_available_step_types,
    _is_url_safe,
)


# ─── Step Handler Registry ────────────────────────────────────────────────────


class TestStepHandlerRegistry:
    """Test the step handler registry."""

    def test_all_step_types_registered(self):
        """All 10 built-in step types are registered."""
        types = get_available_step_types()
        expected = {"agent", "calendar", "crm", "dms", "event", "http", "mail", "search", "wait", "webhook"}
        assert set(types) == expected

    def test_get_step_handler_returns_callable(self):
        """get_step_handler returns a callable for each registered type."""
        for step_type in get_available_step_types():
            handler = get_step_handler(step_type)
            assert handler is not None
            assert callable(handler)

    def test_get_step_handler_unknown_returns_none(self):
        """get_step_handler returns None for unknown step type."""
        assert get_step_handler("nonexistent") is None


# ─── Wait Step (G-WAIT) ───────────────────────────────────────────────────────


class TestWaitStep:
    """Test the wait/delay step handler."""

    @pytest.mark.asyncio
    async def test_wait_with_duration(self):
        """Wait step with duration_seconds sets resume_at correctly."""
        instance = MagicMock()
        instance.context = {}
        step = {"type": "wait", "config": {"duration_seconds": 60}}
        result = await get_step_handler("wait")(
            MagicMock(), uuid.uuid4(), instance, step
        )
        assert result.advance is False
        assert result.wait_until is not None
        assert result.wait_reason == "wait"
        # resume_at should be ~60s in the future
        now = datetime.now(UTC)
        delta = result.wait_until - now
        assert 50 < delta.total_seconds() < 70

    @pytest.mark.asyncio
    async def test_wait_with_absolute_time(self):
        """Wait step with resume_at sets exact resume time."""
        instance = MagicMock()
        instance.context = {}
        future = (datetime.now(UTC) + timedelta(hours=2)).isoformat()
        step = {"type": "wait", "config": {"resume_at": future}}
        result = await get_step_handler("wait")(
            MagicMock(), uuid.uuid4(), instance, step
        )
        assert result.advance is False
        assert result.wait_until is not None
        assert result.wait_reason == "wait"

    @pytest.mark.asyncio
    async def test_wait_without_config_aborts(self):
        """Wait step without duration_seconds or resume_at aborts."""
        instance = MagicMock()
        instance.context = {}
        step = {"type": "wait", "config": {}}
        result = await get_step_handler("wait")(
            MagicMock(), uuid.uuid4(), instance, step
        )
        assert result.abort is True
        assert "requires" in result.error


# ─── SSRF Protection (G-HTTP) ────────────────────────────────────────────────


class TestSSRFProtection:
    """Test the SSRF protection for HTTP and webhook steps."""

    def test_blocks_localhost(self):
        """SSRF blocks localhost."""
        assert _is_url_safe("http://localhost:8080/api") is False
        assert _is_url_safe("http://127.0.0.1:8080/api") is False

    def test_blocks_private_ips(self):
        """SSRF blocks private IP ranges."""
        assert _is_url_safe("http://192.168.1.1/api") is False
        assert _is_url_safe("http://10.0.0.1/api") is False
        assert _is_url_safe("http://172.16.0.1/api") is False

    def test_blocks_non_http_schemes(self):
        """SSRF blocks non-http/https schemes."""
        assert _is_url_safe("ftp://example.com/file") is False
        assert _is_url_safe("file:///etc/passwd") is False
        assert _is_url_safe("gopher://example.com") is False

    def test_allows_public_urls(self):
        """SSRF allows public HTTP/HTTPS URLs."""
        assert _is_url_safe("https://api.example.com/webhook") is True
        assert _is_url_safe("http://example.com/api") is True

    def test_blocks_metadata_endpoint(self):
        """SSRF blocks cloud metadata endpoints."""
        assert _is_url_safe("http://metadata.google.internal/computeMetadata/") is False

    def test_blocks_ipv6_loopback(self):
        """SSRF blocks IPv6 loopback."""
        assert _is_url_safe("http://[::1]:8080/api") is False

    def test_handles_invalid_url(self):
        """SSRF handles invalid URLs gracefully."""
        assert _is_url_safe("") is False
        assert _is_url_safe("not-a-url") is False


# ─── HTTP Step (G-HTTP) ──────────────────────────────────────────────────────


class TestHttpStep:
    """Test the HTTP request step handler."""

    @pytest.mark.asyncio
    async def test_http_without_url_aborts(self):
        """HTTP step without URL aborts."""
        instance = MagicMock()
        instance.context = {}
        step = {"type": "http", "config": {"method": "GET"}}
        result = await get_step_handler("http")(
            MagicMock(), uuid.uuid4(), instance, step
        )
        assert result.abort is True
        assert "url" in result.error.lower()

    @pytest.mark.asyncio
    async def test_http_with_ssrf_url_aborts(self):
        """HTTP step with SSRF-blocked URL aborts."""
        instance = MagicMock()
        instance.context = {}
        step = {"type": "http", "config": {"url": "http://127.0.0.1:8080/secret"}}
        result = await get_step_handler("http")(
            MagicMock(), uuid.uuid4(), instance, step
        )
        assert result.abort is True
        assert "ssrf" in result.error.lower()


# ─── Event Step (G-EVT) ──────────────────────────────────────────────────────


class TestEventStep:
    """Test the event publishing step handler."""

    @pytest.mark.asyncio
    async def test_event_without_name_aborts(self):
        """Event step without event_name aborts."""
        instance = MagicMock()
        instance.id = uuid.uuid4()
        instance.context = {}
        step = {"type": "event", "config": {"payload": {"key": "value"}}}
        result = await get_step_handler("event")(
            MagicMock(), uuid.uuid4(), instance, step
        )
        assert result.abort is True
        assert "event_name" in result.error

    @pytest.mark.asyncio
    async def test_event_publishes_successfully(self):
        """Event step publishes event to event bus."""
        instance = MagicMock()
        instance.id = uuid.uuid4()
        instance.context = {}
        step = {"type": "event", "config": {"event_name": "test.event", "payload": {"key": "value"}}}

        with patch("app.core.event_bus.get_event_bus") as mock_get_bus:
            mock_bus = MagicMock()
            mock_bus.publish = AsyncMock()
            mock_get_bus.return_value = mock_bus

            result = await get_step_handler("event")(
                MagicMock(), uuid.uuid4(), instance, step
            )
            assert result.advance is True
            assert result.output["event_published"] == "test.event"
            mock_bus.publish.assert_called_once()


# ─── CRM Step (G-CRM) ────────────────────────────────────────────────────────


class TestCrmStep:
    """Test the CRM action step handler."""

    @pytest.mark.asyncio
    async def test_crm_without_action_aborts(self):
        """CRM step without action aborts."""
        instance = MagicMock()
        instance.context = {}
        step = {"type": "crm", "config": {"data": {"name": "Test"}}}
        result = await get_step_handler("crm")(
            MagicMock(), uuid.uuid4(), instance, step
        )
        assert result.abort is True
        assert "action" in result.error

    @pytest.mark.asyncio
    async def test_crm_unknown_action_aborts(self):
        """CRM step with unknown action aborts."""
        instance = MagicMock()
        instance.context = {}
        step = {"type": "crm", "config": {"action": "invalid_action"}}
        result = await get_step_handler("crm")(
            MagicMock(), uuid.uuid4(), instance, step
        )
        assert result.abort is True
        assert "unknown" in result.error.lower()


# ─── StepResult ──────────────────────────────────────────────────────────────


class TestStepResult:
    """Test the StepResult class."""

    def test_default_step_result_advances(self):
        """Default StepResult advances to next step."""
        result = StepResult()
        assert result.advance is True
        assert result.next_index is None
        assert result.wait_until is None
        assert result.error is None
        assert result.abort is False
        assert result.output == {}

    def test_step_result_with_wait(self):
        """StepResult with wait_until does not advance."""
        wait_time = datetime.now(UTC) + timedelta(seconds=30)
        result = StepResult(advance=False, wait_until=wait_time, wait_reason="wait")
        assert result.advance is False
        assert result.wait_until == wait_time
        assert result.wait_reason == "wait"

    def test_step_result_with_error(self):
        """StepResult with error but no abort is retryable."""
        result = StepResult(error="Something failed")
        assert result.error == "Something failed"
        assert result.abort is False

    def test_step_result_with_abort(self):
        """StepResult with abort stops the workflow."""
        result = StepResult(error="Fatal error", abort=True)
        assert result.abort is True
        assert result.error == "Fatal error"

    def test_step_result_with_branch(self):
        """StepResult with next_index branches to specific step."""
        result = StepResult(next_index=5)
        assert result.next_index == 5
        assert result.advance is True  # Still advances, just to specific index


# ─── Workflow Engine Resume (G-RUN) ──────────────────────────────────────────


class TestWorkflowEngineResume:
    """Test the WorkflowEngine resume functionality."""

    @pytest.mark.asyncio
    async def test_resume_non_waiting_instance_returns_unchanged(self):
        """Resume on a non-waiting instance returns the instance unchanged."""
        from app.workflows.engine import WorkflowEngine

        instance = MagicMock()
        instance.status = "completed"
        instance.id = uuid.uuid4()
        instance.workflow_id = uuid.uuid4()
        instance.current_step_index = 0

        db = MagicMock()
        engine = WorkflowEngine(db, uuid.uuid4())
        result = await engine.resume(instance)
        # Should return _instance_to_dict result, not process
        assert result is not None

    @pytest.mark.asyncio
    async def test_find_resumable_workflows_query(self):
        """find_resumable_workflows queries for waiting instances with passed resume_at."""
        from app.workflows.engine import find_resumable_workflows
        from app.models.workflow import WorkflowInstance

        db = MagicMock()
        mock_result = MagicMock()
        mock_result.scalars.return_value.all.return_value = []
        db.execute = AsyncMock(return_value=mock_result)

        await find_resumable_workflows(db, uuid.uuid4())
        db.execute.assert_called_once()


# ─── Workflow Schema (G-COND) ───────────────────────────────────────────────


class TestWorkflowSchema:
    """Test the extended WorkflowStep schema."""

    def test_step_schema_accepts_new_types(self):
        """WorkflowStep schema accepts all new step types."""
        from app.schemas.workflow import WorkflowStep

        for step_type in ["wait", "http", "mail", "calendar", "dms", "search", "agent", "crm", "event", "webhook"]:
            step = WorkflowStep(name=f"Test {step_type}", type=step_type, config={})
            assert step.type == step_type

    def test_step_schema_rejects_unknown_type(self):
        """WorkflowStep schema rejects unknown step types."""
        from app.schemas.workflow import WorkflowStep
        from pydantic import ValidationError

        with pytest.raises(ValidationError):
            WorkflowStep(name="Bad", type="unknown_type", config={})

    def test_step_schema_still_accepts_legacy_types(self):
        """WorkflowStep schema still accepts legacy step types."""
        from app.schemas.workflow import WorkflowStep

        for step_type in ["action", "approval", "notification", "condition"]:
            step = WorkflowStep(name=f"Legacy {step_type}", type=step_type, config={})
            assert step.type == step_type


# ─── Workflow Model (G-RUN) ─────────────────────────────────────────────────


class TestWorkflowModelDurableFields:
    """Test the new durable/resumable fields on WorkflowInstance."""

    def test_workflow_instance_has_resume_fields(self):
        """WorkflowInstance model has all G-RUN fields."""
        from app.models.workflow import WorkflowInstance

        # Check that the model has the new columns
        assert hasattr(WorkflowInstance, "resume_at")
        assert hasattr(WorkflowInstance, "resume_reason")
        assert hasattr(WorkflowInstance, "step_state")
        assert hasattr(WorkflowInstance, "idempotency_key")
        assert hasattr(WorkflowInstance, "lock_owner")
        assert hasattr(WorkflowInstance, "lock_expires_at")
        assert hasattr(WorkflowInstance, "error_message")
        assert hasattr(WorkflowInstance, "retry_count")
        assert hasattr(WorkflowInstance, "max_retries")


# ─── G-HUMAN-DEC: Automated-Decision Guard ────────────────────────────────────


class TestDecisionGuard:
    """Test the automated-decision guard (G-HUMAN-DEC)."""

    def test_no_metadata_allows_everything(self):
        """No AI use case metadata means no guard — action allowed."""
        from app.workflows.decision_guard import requires_human_review

        assert requires_human_review(None, "send_email") is False
        assert requires_human_review({}, "delete_entity") is False

    def test_auto_execute_false_requires_review(self):
        """auto_execute=False always requires review."""
        from app.workflows.decision_guard import requires_human_review

        metadata = {"auto_execute": False}
        assert requires_human_review(metadata, "any_action") is True

    def test_requires_approval_true_requires_review(self):
        """requires_approval=True always requires review."""
        from app.workflows.decision_guard import requires_human_review

        metadata = {"requires_approval": True}
        assert requires_human_review(metadata, "any_action") is True

    def test_medium_risk_requires_review(self):
        """Medium or higher risk level always requires review."""
        from app.workflows.decision_guard import requires_human_review

        assert requires_human_review({"risk_level": "medium"}, "any_action") is True
        assert requires_human_review({"risk_level": "high"}, "any_action") is True
        assert requires_human_review({"risk_level": "critical"}, "any_action") is True

    def test_low_risk_allows_normal_actions(self):
        """Low risk allows normal actions."""
        from app.workflows.decision_guard import requires_human_review

        assert requires_human_review({"risk_level": "low"}, "noop") is False
        assert requires_human_review({"risk_level": "none"}, "noop") is False

    def test_low_risk_blocks_high_risk_actions(self):
        """Low risk still blocks inherently high-risk actions."""
        from app.workflows.decision_guard import requires_human_review

        assert requires_human_review({"risk_level": "low"}, "send_email") is True
        assert requires_human_review({"risk_level": "low"}, "delete_entity") is True
        assert requires_human_review({"risk_level": "low"}, "execute_payment") is True

    def test_reviewed_actions_list_checked(self):
        """Specific actions in reviewed_actions list require review."""
        from app.workflows.decision_guard import requires_human_review

        metadata = {"risk_level": "none", "reviewed_actions": ["custom_action"]}
        assert requires_human_review(metadata, "custom_action") is True
        assert requires_human_review(metadata, "other_action") is False

    @pytest.mark.asyncio
    async def test_check_decision_guard_allows_low_risk(self):
        """check_decision_guard allows low-risk actions without metadata."""
        from app.workflows.decision_guard import check_decision_guard

        result = await check_decision_guard(
            db=MagicMock(),
            tenant_id=uuid.uuid4(),
            instance_id=uuid.uuid4(),
            step_config={},
            action="noop",
            ai_use_case_metadata=None,
        )
        assert result["allowed"] is True
        assert result["requires_approval"] is False

    @pytest.mark.asyncio
    async def test_check_decision_guard_blocks_high_risk(self):
        """check_decision_guard blocks high-risk actions."""
        from app.workflows.decision_guard import check_decision_guard

        result = await check_decision_guard(
            db=MagicMock(),
            tenant_id=uuid.uuid4(),
            instance_id=uuid.uuid4(),
            step_config={"ai_use_case_metadata": {"risk_level": "high"}},
            action="send_email",
            ai_use_case_metadata=None,
        )
        assert result["allowed"] is False
        assert result["requires_approval"] is True
        assert "send_email" in result["reason"]


# ─── G-WORK: Workflow Workstream ─────────────────────────────────────────────


class TestWorkflowWorkstream:
    """Test the workflow workstream integration (G-WORK)."""

    @pytest.mark.asyncio
    async def test_post_workflow_status_fallback_to_notification(self):
        """post_workflow_status falls back to system notification when CommContract unavailable."""
        from app.workflows.workstream import post_workflow_status

        with patch("app.core.notifications.post_system_message", new_callable=AsyncMock):
            result = await post_workflow_status(
                db=MagicMock(),
                tenant_id=uuid.uuid4(),
                instance_id=uuid.uuid4(),
                workflow_name="Test Workflow",
                status="in_progress",
                user_id=uuid.uuid4(),
            )
            # Should return None (fallback) but not crash
            assert result is None

    @pytest.mark.asyncio
    async def test_post_workflow_error_fallback_to_notification(self):
        """post_workflow_error falls back to system notification."""
        from app.workflows.workstream import post_workflow_error

        with patch("app.core.notifications.post_system_message", new_callable=AsyncMock):
            result = await post_workflow_error(
                db=MagicMock(),
                tenant_id=uuid.uuid4(),
                instance_id=uuid.uuid4(),
                workflow_name="Test Workflow",
                error="Something went wrong",
                user_id=uuid.uuid4(),
            )
            assert result is None

    @pytest.mark.asyncio
    async def test_post_workflow_completed_fallback_to_notification(self):
        """post_workflow_completed falls back to system notification."""
        from app.workflows.workstream import post_workflow_completed

        with patch("app.core.notifications.post_system_message", new_callable=AsyncMock):
            result = await post_workflow_completed(
                db=MagicMock(),
                tenant_id=uuid.uuid4(),
                instance_id=uuid.uuid4(),
                workflow_name="Test Workflow",
                result={"output": "done"},
                user_id=uuid.uuid4(),
            )
            assert result is None

    @pytest.mark.asyncio
    async def test_post_workflow_handoff_fallback_to_notification(self):
        """post_workflow_handoff falls back to system notification."""
        from app.workflows.workstream import post_workflow_handoff

        with patch("app.core.notifications.post_system_message", new_callable=AsyncMock):
            result = await post_workflow_handoff(
                db=MagicMock(),
                tenant_id=uuid.uuid4(),
                instance_id=uuid.uuid4(),
                workflow_name="Test Workflow",
                handoff_type="review_needed",
                assignee_id=uuid.uuid4(),
                description="Please review",
                user_id=uuid.uuid4(),
            )
            assert result is None