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