"""
Tests for Phase B: Chain Suggester Service
"""
import json
import pytest
from app.services.chain_suggester import suggest_chains, find_template_by_category
from app.services.chain_templates import (
get_chain_templates, get_chain_templates_with_config,
get_chain_template, CHAIN_TEMPLATES,
)
# All suggest_chains calls use use_fallback=True to skip LLM (fast tests)
FALLBACK = {"use_fallback": True}
class TestChainSuggester:
"""Test the chain suggestion engine (fallback mode)."""
def test_suggest_chains_basic(self):
"""Test basic suggestion generation."""
fields = [
{"name": "full_name", "type": "text", "label": "Full Name"},
{"name": "email", "type": "email", "label": "Email"},
]
result = suggest_chains("Send me an email when someone submits", fields, **FALLBACK)
# With only 3 templates, may return 0-3 matches
assert isinstance(result, list)
for s in result:
assert "name" in s
assert "description" in s
assert "steps" in s
assert len(s["steps"]) >= 1
def test_suggest_chains_onboarding(self):
"""Test onboarding suggestion matches."""
fields = [
{"name": "name", "type": "text", "label": "Name"},
{"name": "email", "type": "email", "label": "Email"},
]
result = suggest_chains("Welcome new client and onboard them", fields, form_type="onboarding", **FALLBACK)
assert len(result) >= 1
names = [s["name"].lower() for s in result]
assert any("onboarding" in n or "welcome" in n or "signup" in n for n in names)
def test_suggest_chains_notification(self):
"""Test notification suggestion."""
fields = [
{"name": "name", "type": "text"},
{"name": "phone", "type": "tel"},
]
result = suggest_chains("Notify my team on new submission", fields, form_type="notification", **FALLBACK)
assert len(result) >= 1
assert all("confidence" in s for s in result)
def test_suggest_chains_nurture(self):
"""Test nurture/follow-up suggestion."""
fields = [
{"name": "email", "type": "email"},
{"name": "name", "type": "text"},
]
result = suggest_chains("Follow up with leads", fields, form_type="marketing", **FALLBACK)
assert len(result) >= 1
def test_suggest_chains_empty_prompt(self):
"""Test empty prompt returns empty."""
result = suggest_chains("", [], **FALLBACK)
assert result == []
def test_suggest_chains_no_fields(self):
"""Test with prompt but no fields — should still match by keywords."""
result = suggest_chains("send email", [], **FALLBACK)
# Keyword matching works without fields
assert len(result) >= 1
def test_suggest_chains_step_types_valid(self):
"""Test all suggested steps use valid types."""
from app.services.chain_suggester import CHAIN_STEP_TYPES
fields = [
{"name": "name", "type": "text"},
{"name": "email", "type": "email"},
]
result = suggest_chains("Send a notification", fields, **FALLBACK)
for s in result:
for step in s.get("steps", []):
assert step["type"] in CHAIN_STEP_TYPES, \
f"Invalid step type: {step['type']}"
def test_suggest_chains_confidence_range(self):
"""Test confidence is between 0 and 1."""
fields = [
{"name": "name", "type": "text"},
{"name": "email", "type": "email"},
]
result = suggest_chains("notify team", fields, **FALLBACK)
for s in result:
assert 0 <= s.get("confidence", 0) <= 1
def test_suggest_chains_max_3(self):
"""Test returns at most 3 suggestions."""
fields = [
{"name": "name", "type": "text"},
{"name": "email", "type": "email"},
]
result = suggest_chains("send email and webhook", fields, **FALLBACK)
assert len(result) <= 3
def test_suggest_chains_requires_config(self):
"""Test requires_config lists missing pieces."""
fields = [
{"name": "name", "type": "text"},
{"name": "email", "type": "email"},
]
result = suggest_chains("send email", fields, **FALLBACK)
for s in result:
assert isinstance(s.get("requires_config", []), list)
def test_suggest_chains_source_marker(self):
"""Test fallback suggestions marked as 'template' source."""
fields = [
{"name": "name", "type": "text"},
{"name": "email", "type": "email"},
]
result = suggest_chains("send email", fields, **FALLBACK)
for s in result:
assert s.get("source") == "template"
class TestTemplateLookup:
"""Test chain template lookup."""
def test_get_all_templates(self):
"""Test getting all chain templates."""
templates = get_chain_templates()
assert len(templates) >= 3
def test_get_all_templates_with_config(self):
"""Test getting chain templates with config."""
templates = get_chain_templates_with_config()
assert len(templates) >= 3
for t in templates:
assert "chain_config" in t
assert len(t["chain_config"].get("steps", [])) >= 1
def test_find_template_by_category(self):
"""Test finding template by category."""
template = find_template_by_category("onboarding")
assert template is not None
assert template["category"] == "onboarding"
def test_find_template_not_found(self):
"""Test finding nonexistent template."""
template = find_template_by_category("nonexistent", "notfound")
assert template is None
def test_get_chain_template(self):
"""Test getting a single template by ID."""
template = get_chain_template("quote_approval")
assert template is not None
assert "config" in template
assert len(template["config"]["steps"]) >= 1
def test_templates_have_steps(self):
"""Test all templates have steps."""
templates = get_chain_templates_with_config()
for t in templates:
assert "chain_config" in t
assert len(t["chain_config"].get("steps", [])) >= 1
def test_template_categories(self):
"""Test template categories are valid."""
valid = {"onboarding", "notifications", "nurture"}
templates = get_chain_templates()
for t in templates:
assert t.get("category") in valid, f"Invalid category: {t.get('category')}"
def test_template_step_types(self):
"""Test template steps use valid types."""
templates = get_chain_templates_with_config()
for t in templates:
for step in t["chain_config"]["steps"]:
# Allow http_request (alias for webhook)
assert step["type"] in ("http_request", "email", "wait", "log"), \
f"Unexpected step type: {step['type']}"