#!/usr/bin/env python3
"""Tests for the Action Chain Engine (Phase A).
Tests cover: chain execution, conditional branching, wait triggers,
parallel execution, error handling, validation, and templates.
"""
import os
import sys
import pytest
from unittest.mock import patch, MagicMock
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# ─── Fixtures ─────────────────────────────────────────────────────────────────
@pytest.fixture
def sample_submission():
return {
"id": 1,
"name": "John Doe",
"email": "john@example.com",
"phone": "555-1234",
"company": "Test Company",
"message": "Test submission",
"budget": 5000,
"priority": "high",
}
@pytest.fixture
def sample_site():
return {
"id": 1,
"user_id": 1,
"name": "Test Site",
}
@pytest.fixture
def simple_chain(sample_site):
"""Simple linear chain: log → http_request → log."""
return {
"id": "test_chain",
"config": {
"entry_step": "step_1",
"max_steps": 10,
"steps": [
{
"id": "step_1",
"type": "log",
"config": {"message": "Starting workflow", "level": "info"},
"on_success": "step_2",
},
{
"id": "step_2",
"type": "log",
"config": {"message": "HTTP request sent", "level": "info"},
"on_success": "step_3",
},
{
"id": "step_3",
"type": "log",
"config": {"message": "Workflow complete", "level": "info"},
},
],
},
}
@pytest.fixture
def branching_chain(sample_site):
"""Chain with conditional branching based on value."""
return {
"id": "branching_chain",
"config": {
"entry_step": "send_notification",
"max_steps": 10,
"steps": [
{
"id": "send_notification",
"type": "log",
"config": {"message": "Notification sent", "level": "info"},
"on_success": "check_priority",
},
{
"id": "check_priority",
"type": "condition",
"config": {
"condition": {"field": "priority", "operator": "equals", "value": "high"}
},
"on_success": "notify_manager",
"on_failure": "end_normal",
},
{
"id": "notify_manager",
"type": "log",
"config": {"message": "Manager notified — high priority", "level": "warning"},
},
{
"id": "end_normal",
"type": "log",
"config": {"message": "Normal processing complete", "level": "info"},
},
],
},
}
@pytest.fixture
def parallel_chain(sample_site):
"""Chain with parallel execution."""
return {
"id": "parallel_chain",
"config": {
"entry_step": "parallel_notifications",
"max_steps": 10,
"steps": [
{
"id": "parallel_notifications",
"type": "parallel",
"config": {
"steps": [
{
"id": "notify_slack",
"type": "log",
"config": {"message": "Slack notification sent", "level": "info"},
},
{
"id": "notify_discord",
"type": "log",
"config": {"message": "Discord notification sent", "level": "info"},
},
]
},
"on_success": "final_log",
},
{
"id": "final_log",
"type": "log",
"config": {"message": "All notifications complete", "level": "info"},
},
],
},
}
# ─── Chain Execution Tests ────────────────────────────────────────────────────
class TestChainExecution:
"""Test the chain engine execution flow."""
def test_execute_simple_chain(self, simple_chain, sample_submission, sample_site):
"""Test basic linear chain execution."""
from app.services.chain_engine import execute_chain
result = execute_chain(simple_chain, sample_site, sample_submission)
assert result["success"] is True
assert result["steps_executed"] == 3
assert result["steps_failed"] == 0
def test_chain_with_missing_step(self, sample_submission, sample_site):
"""Test chain with invalid entry step."""
from app.services.chain_engine import execute_chain
invalid_chain = {
"id": "invalid",
"config": {
"entry_step": "nonexistent",
"max_steps": 10,
"steps": [],
},
}
result = execute_chain(invalid_chain, sample_site, sample_submission)
# Empty steps should return success with 0 executed
assert result["steps_executed"] == 0
def test_chain_max_steps(self, sample_submission, sample_site):
"""Test chain stops at max_steps."""
from app.services.chain_engine import execute_chain
looping_chain = {
"id": "looping",
"config": {
"entry_step": "step_1",
"max_steps": 3,
"steps": [
{
"id": "step_1",
"type": "log",
"config": {"message": "Loop", "level": "info"},
"on_success": "step_1", # Infinite loop
},
],
},
}
result = execute_chain(looping_chain, sample_site, sample_submission)
assert result["steps_executed"] == 3
def test_chain_with_error_handling(self, sample_submission, sample_site):
"""Test chain error handling."""
from app.services.chain_engine import execute_chain
error_chain = {
"id": "error_chain",
"config": {
"entry_step": "failing_step",
"max_steps": 10,
"steps": [
{
"id": "failing_step",
"type": "http_request",
"config": {
"method": "GET",
"url": "https://this-domain-does-not-exist-12345.com",
"timeout": 2,
},
"on_failure": "handle_error",
},
{
"id": "handle_error",
"type": "log",
"config": {"message": "Error handled gracefully", "level": "error"},
},
],
},
}
result = execute_chain(error_chain, sample_site, sample_submission)
assert result["steps_executed"] >= 1
# ─── Conditional Branching Tests ───────────────────────────────────────────────
class TestConditionalBranching:
"""Test conditional step branching."""
def test_condition_true_branch(self, branching_chain, sample_submission, sample_site):
"""Test condition evaluates to true."""
from app.services.chain_engine import execute_chain
result = execute_chain(branching_chain, sample_site, sample_submission)
# Should route to notify_manager (high priority)
assert result["success"] is True
def test_condition_false_branch(self, branching_chain, sample_site):
"""Test condition evaluates to false."""
from app.services.chain_engine import execute_chain
submission = {"id": 2, "priority": "low", "name": "Normal User"}
result = execute_chain(branching_chain, sample_site, submission)
assert result["success"] is True
# ─── Wait Trigger Tests ───────────────────────────────────────────────────────
class TestWaitTriggers:
"""Test wait step functionality."""
def test_wait_parsing(self):
"""Test wait duration parsing."""
from app.services.chain_engine import parse_wait_duration
assert parse_wait_duration("30s") == 30
assert parse_wait_duration("5m") == 300
assert parse_wait_duration("1h") == 3600
assert parse_wait_duration("24h") == 86400
assert parse_wait_duration("2d") == 172800
assert parse_wait_duration("1w") == 604800
assert parse_wait_duration("1d12h") == 129600
def test_wait_handler_signature(self):
"""Test wait handler exists and has correct signature."""
from app.services.chain_engine import STEP_HANDLERS
assert "wait" in STEP_HANDLERS
# ─── Parallel Execution Tests ─────────────────────────────────────────────────
class TestParallelExecution:
"""Test parallel step execution."""
def test_parallel_steps(self, parallel_chain, sample_submission, sample_site):
"""Test parallel execution of multiple steps."""
from app.services.chain_engine import execute_chain
result = execute_chain(parallel_chain, sample_site, sample_submission)
assert result["success"] is True
assert result["steps_executed"] >= 2
# ─── Validation Tests ─────────────────────────────────────────────────────────
class TestChainValidation:
"""Test chain validation."""
def test_valid_chain(self):
"""Test valid chain passes validation."""
from app.services.chain_engine import validate_chain
valid = {
"entry_step": "step_1",
"steps": [
{"id": "step_1", "type": "log", "config": {"message": "test"}}
],
}
result = validate_chain(valid)
assert result["valid"] is True
def test_invalid_no_entry_step(self):
"""Test chain without entry_step and no steps."""
from app.services.chain_engine import validate_chain
invalid = {
"entry_step": "nonexistent",
"steps": [
{"id": "step_1", "type": "log", "config": {}}
],
}
result = validate_chain(invalid)
assert result["valid"] is False
assert any("not found" in e.lower() for e in result["errors"])
def test_invalid_duplicate_ids(self):
"""Test chain with duplicate step IDs."""
from app.services.chain_engine import validate_chain
invalid = {
"entry_step": "step_1",
"steps": [
{"id": "step_1", "type": "log", "config": {}},
{"id": "step_1", "type": "log", "config": {}},
],
}
result = validate_chain(invalid)
assert result["valid"] is False
assert any("duplicate" in e.lower() for e in result["errors"])
def test_invalid_unknown_step_type(self):
"""Test chain with unknown step type."""
from app.services.chain_engine import validate_chain
invalid = {
"entry_step": "step_1",
"steps": [
{"id": "step_1", "type": "nonexistent_type", "config": {}}
],
}
result = validate_chain(invalid)
assert result["valid"] is False
# ─── Template Tests ───────────────────────────────────────────────────────────
class TestChainTemplates:
"""Test chain templates."""
def test_list_templates(self):
"""Test listing all templates."""
from app.services.chain_templates import get_chain_templates
templates = get_chain_templates()
assert len(templates) >= 1
template_ids = [t["id"] for t in templates]
assert "quote_approval" in template_ids
def test_get_template(self):
"""Test getting a specific template."""
from app.services.chain_templates import get_chain_template
template = get_chain_template("quote_approval")
assert template is not None
assert "config" in template
assert "steps" in template["config"]
def test_template_validation(self):
"""Test that templates pass validation."""
from app.services.chain_templates import get_chain_templates
from app.services.chain_engine import validate_chain
templates = get_chain_templates()
for template_info in templates:
from app.services.chain_templates import get_chain_template
template = get_chain_template(template_info["id"])
result = validate_chain(template["config"])
assert result["valid"], f"Template {template_info['id']} failed: {result['errors']}"
# ─── Integration Tests ────────────────────────────────────────────────────────
class TestActionPipelineIntegration:
"""Test chain integration with action pipeline."""
def test_chain_action_registered(self):
"""Test that chain action is registered in pipeline."""
from app.services.action_pipeline import _HANDLERS
assert "chain" in _HANDLERS