#!/usr/bin/env python3
"""Tests for Phase B: AI-Powered Action Chain Suggestion Routes.
Tests the API endpoints for chain templates, suggestions, and deployment.
Uses the same shared DB pattern as other test files.
"""
import json
import os
import random
import sqlite3
import sys
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
# Use shared test DB
import tests.test_shared_db # noqa: F401
# Set required env vars
os.environ.setdefault("AGENTFORMS_SECRET_KEY", "phaseb-test-secret")
os.environ.setdefault("ENCRYPTION_KEY", "5c321ae59453d7bae2e05a9e19e97347411a033d9858a0456d5cc63acfcb0369")
os.environ.setdefault("SMTP_HOST", "smtp.test.invalid")
os.environ.setdefault("SMTP_PORT", "587")
# Import shared Flask app
from app.app import app as flask_app
import app.models
flask_app.config["TESTING"] = True
flask_app.config["SESSION_COOKIE_SECURE"] = False
def _create_user_and_site(client, name="PhaseB Test Site"):
"""Create a test user + site and return (user_email, site_token, site_id)."""
email = f"phaseb{random.randint(10000, 99999)}@example.com"
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "PhaseB Tester",
}, follow_redirects=True)
# Mark email as verified
from app.models import hash_value
with flask_app.app_context():
conn = sqlite3.connect(app.models.DB_PATH)
conn.execute("UPDATE users SET email_verified = 1 WHERE email_hash = ?",
(hash_value(email.lower()),))
conn.commit()
conn.close()
# Create site
resp = client.post("/sites/new", data={
"name": name,
"owner_email": email,
}, follow_redirects=True)
assert resp.status_code == 200
# Get site info
with flask_app.app_context():
conn = sqlite3.connect(app.models.DB_PATH)
row = conn.execute(
"SELECT token, id FROM sites WHERE owner_email = ? ORDER BY created_at DESC LIMIT 1",
(email.lower(),)
).fetchone()
conn.close()
return email, row[0], row[1]
def _login(client, email, password="SecurePass123"):
"""Login and return the session cookie."""
client.post("/auth/login", data={
"email": email,
"password": password,
}, follow_redirects=True)
# ─── Templates ──────────────────────────────────────────────────────────────────
class TestTemplates:
"""Test /api/chains/templates endpoints."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def test_list_templates_unauthorized(self, client):
"""GET /api/chains/templates without login returns non-200."""
resp = client.get("/api/chains/templates")
# Flask login_required returns 302 redirect; unauthenticated API returns 401/403
assert resp.status_code in [302, 401, 403]
def test_list_templates(self, client):
"""GET /api/chains/templates returns templates when logged in."""
email, _, _ = _create_user_and_site(client)
_login(client, email)
resp = client.get("/api/chains/templates")
data = resp.get_json()
assert resp.status_code == 200
assert "templates" in data
assert len(data["templates"]) >= 5
def test_get_template(self, client):
"""GET /api/chains/templates/<id> returns specific template."""
email, _, _ = _create_user_and_site(client)
_login(client, email)
resp = client.get("/api/chains/templates/lead_nurture")
data = resp.get_json()
assert resp.status_code == 200
assert "template" in data
assert "name" in data["template"]
def test_get_template_not_found(self, client):
"""GET /api/chains/templates/<invalid> returns 404."""
email, _, _ = _create_user_and_site(client)
_login(client, email)
resp = client.get("/api/chains/templates/nonexistent")
assert resp.status_code == 404
# ─── Chain Suggestions ─────────────────────────────────────────────────────────
class TestSuggestChains:
"""Test /api/chains/suggest endpoint."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def test_suggest_unauthorized(self, client):
"""POST /api/chains/suggest without login returns 401."""
resp = client.post("/api/chains/suggest", json={"site_id": 1})
assert resp.status_code in [401, 403]
def test_suggest_basic(self, client):
"""POST /api/chains/suggest returns suggestions."""
email, _, site_id = _create_user_and_site(client)
_login(client, email)
resp = client.post("/api/chains/suggest", json={"site_id": site_id})
data = resp.get_json()
assert resp.status_code == 200
assert "suggestions" in data
assert len(data["suggestions"]) >= 1
assert data["source"] in ["template", "llm", "template_fallback"]
def test_suggest_with_prompt(self, client):
"""POST /api/chains/suggest with prompt returns relevant suggestions."""
email, _, site_id = _create_user_and_site(client, "Contact Form")
_login(client, email)
resp = client.post("/api/chains/suggest", json={
"site_id": site_id,
"prompt": "send a confirmation email and notify the team",
})
data = resp.get_json()
assert resp.status_code == 200
assert len(data["suggestions"]) >= 1
for s in data["suggestions"]:
assert "name" in s
assert "steps" in s
assert "confidence" in s
def test_suggest_requires_site_id(self, client):
"""POST /api/chains/suggest without site_id returns 400."""
email, _, _ = _create_user_and_site(client)
_login(client, email)
resp = client.post("/api/chains/suggest", json={})
assert resp.status_code == 400
def test_suggest_site_not_found(self, client):
"""POST /api/chains/suggest with invalid site_id returns 404."""
email, _, _ = _create_user_and_site(client)
_login(client, email)
resp = client.post("/api/chains/suggest", json={"site_id": 99999})
assert resp.status_code == 404
def test_suggest_forbidden(self, client):
"""POST /api/chains/suggest for another user's site returns 403."""
email1, _, site1_id = _create_user_and_site(client, "Site 1")
_login(client, email1)
# Create second user with separate client (sites owned by logged-in user)
client2 = flask_app.test_client()
email2 = f"phaseb2{random.randint(10000, 99999)}@example.com"
client2.post("/auth/register", data={
"email": email2,
"password": "***",
"name": "Other User",
}, follow_redirects=True)
from app.models import hash_value
with flask_app.app_context():
conn = sqlite3.connect(app.models.DB_PATH)
conn.execute("UPDATE users SET email_verified = 1 WHERE email_hash = ?",
(hash_value(email2.lower()),))
conn.commit()
conn.close()
_login(client2, email2)
_, _, other_site_id = _create_user_and_site(client2, "Other Site")
# Try user1 to suggest on user2's site
resp = client.post("/api/chains/suggest", json={"site_id": other_site_id})
assert resp.status_code == 403
class TestDeployChain:
"""Test /api/chains/deploy endpoint."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def test_deploy_requires_site_id(self, client):
"""POST /api/chains/deploy without site_id returns 400."""
email, _, _ = _create_user_and_site(client)
_login(client, email)
resp = client.post("/api/chains/deploy", json={})
assert resp.status_code == 400
def test_deploy_requires_chain_config(self, client):
"""POST /api/chains/deploy without chain_config returns 400."""
email, _, site_id = _create_user_and_site(client)
_login(client, email)
resp = client.post("/api/chains/deploy", json={"site_id": site_id})
assert resp.status_code == 400
def test_deploy_invalid_chain(self, client):
"""POST /api/chains/deploy with invalid chain returns error."""
email, _, site_id = _create_user_and_site(client)
_login(client, email)
resp = client.post("/api/chains/deploy", json={
"site_id": site_id,
"chain_config": {
"name": "Test",
"steps": [{"type": "invalid_step"}]
}
})
# Either 400 (validation) or 403 (tier check) is acceptable
assert resp.status_code in [400, 403]
def test_deploy_free_tier_no_chains(self, client):
"""POST /api/chains/deploy on free tier returns 403."""
email, _, site_id = _create_user_and_site(client)
_login(client, email)
resp = client.post("/api/chains/deploy", json={
"site_id": site_id,
"chain_config": {
"name": "Test Chain",
"steps": [{"type": "log", "message": "hello"}]
}
})
# Free tier should be rejected (chains not available)
assert resp.status_code == 403
def test_deploy_site_not_found(self, client):
"""POST /api/chains/deploy with invalid site returns 404."""
email, _, _ = _create_user_and_site(client)
_login(client, email)
resp = client.post("/api/chains/deploy", json={
"site_id": 99999,
"chain_config": {"name": "Test", "steps": []}
})
assert resp.status_code == 404
# ─── LLM Client ────────────────────────────────────────────────────────────────
class TestLLMClient:
"""Test the LLM client module."""
def test_is_llm_available(self):
"""is_llm_available returns bool."""
from app.services.llm_client import is_llm_available
result = is_llm_available()
assert isinstance(result, bool)
def test_suggest_chains_llm_fallback(self):
"""suggest_chains_llm falls back to template when LLM unavailable."""
from app.services.llm_client import suggest_chains_llm
result = suggest_chains_llm(
prompt="test form",
fields=[{"name": "email", "type": "email"}],
form_type="form"
)
assert "suggestions" in result
assert result.get("source") == "template_fallback"