#!/usr/bin/env python3
"""Phase 1: Action Pipeline — test suite.
Tests:
- Submission data normalization (1.1)
- Webhook action handler (1.2)
- Email action handler (1.3)
- Log action handler (1.4)
- Redirect action handler (1.5)
- Document action handler (1.6)
- Action pipeline execution (1.7)
- Action ordering & enabled/disabled (1.8)
- Error handling & unknown action types (1.9)
Run: python -m pytest tests/test_phase1.py -v
"""
import json
import os
import sys
import sqlite3
import tempfile
import unittest
import importlib
from unittest import mock
from datetime import datetime, timezone
# Use shared test DB
import tests.test_shared_db
# Set required env vars before importing app modules
os.environ["ENCRYPTION_KEY"] = os.environ.get("ENCRYPTION_KEY", "test-key-do-not-use-in-production")
os.environ["SECRET_KEY"] = os.environ.get("SECRET_KEY", "test-secret-key")
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import app.models
from app.app import app as flask_app
from app.services.action_pipeline import (
_build_submission_data,
_handle_webhook,
_handle_email,
_handle_log,
_handle_redirect,
_handle_document,
execute_actions,
)
class TestBuildSubmissionData(unittest.TestCase):
"""Test normalized submission data building (1.1)."""
def test_static_fields_mapped(self):
site = {"id": 1, "name": "Test Site"}
submission = {
"id": 100,
"customer_name": "John Smith",
"customer_email": "john@example.com",
"customer_phone": "555-1234",
"customer_equipment": "Generator",
"customer_message": "Need help",
"submitted_at": "2026-06-14T12:00:00",
}
result = _build_submission_data(site, submission)
self.assertEqual(result["fields"]["name"], "John Smith")
self.assertEqual(result["fields"]["email"], "john@example.com")
self.assertEqual(result["fields"]["phone"], "555-1234")
self.assertEqual(result["fields"]["equipment"], "Generator")
self.assertEqual(result["fields"]["message"], "Need help")
def test_dynamic_data_merged(self):
site = {"id": 1, "name": "Test Site"}
submission = {
"id": 100,
"customer_name": "Jane",
"data": json.dumps({"company": "Acme", "budget": "5000"}),
"submitted_at": "2026-06-14T12:00:00",
}
result = _build_submission_data(site, submission)
self.assertEqual(result["fields"]["name"], "Jane")
self.assertEqual(result["fields"]["company"], "Acme")
self.assertEqual(result["fields"]["budget"], "5000")
def test_invalid_json_data_handled(self):
site = {"id": 1, "name": "Test Site"}
submission = {
"id": 100,
"customer_name": "Test",
"data": "not valid json",
"submitted_at": "2026-06-14T12:00:00",
}
result = _build_submission_data(site, submission)
self.assertEqual(result["fields"]["name"], "Test")
# Invalid JSON is silently ignored
def test_minimal_submission(self):
site = {"id": 1, "name": "Test Site"}
submission = {"id": 100, "submitted_at": "2026-06-14T12:00:00"}
result = _build_submission_data(site, submission)
self.assertEqual(result["site_id"], 1)
self.assertEqual(result["site_name"], "Test Site")
self.assertEqual(result["submission_id"], 100)
def test_field_config_passed_through(self):
site = {"id": 1, "name": "Test Site"}
submission = {"id": 100, "submitted_at": "2026-06-14T12:00:00"}
field_config = [{"key": "name", "type": "text"}]
result = _build_submission_data(site, submission, field_config)
self.assertEqual(result["field_config"], field_config)
class TestHandleWebhook(unittest.TestCase):
"""Test webhook action handler (1.2)."""
def test_webhook_success(self):
action = {
"id": 1,
"config": {
"url": "https://example.com/webhook",
"headers": {"Content-Type": "application/json"},
"timeout": 5,
}
}
submission_data = {
"site_id": 1,
"site_name": "Test",
"submission_id": 100,
"fields": {"name": "John"},
}
mock_resp = mock.MagicMock()
mock_resp.status_code = 200
mock_resp.text = "OK"
with mock.patch("requests.post", return_value=mock_resp) as mock_post:
result = _handle_webhook(action, submission_data)
self.assertTrue(result["success"])
self.assertEqual(result["status_code"], 200)
mock_post.assert_called_once()
# Verify JSON payload was sent
call_kwargs = mock_post.call_args
self.assertIn("json", call_kwargs.kwargs)
def test_webhook_failure(self):
action = {
"id": 1,
"config": {"url": "https://example.com/webhook"},
}
submission_data = {"site_id": 1, "site_name": "Test", "submission_id": 100, "fields": {}}
mock_resp = mock.MagicMock()
mock_resp.status_code = 500
mock_resp.text = "Internal Server Error"
with mock.patch("requests.post", return_value=mock_resp):
result = _handle_webhook(action, submission_data)
self.assertFalse(result["success"])
self.assertEqual(result["status_code"], 500)
def test_webhook_missing_url(self):
action = {"id": 1, "config": {}}
submission_data = {"site_id": 1, "site_name": "Test", "submission_id": 100, "fields": {}}
result = _handle_webhook(action, submission_data)
self.assertFalse(result["success"])
self.assertEqual(result["error"], "missing_url")
def test_webhook_request_exception(self):
action = {
"id": 1,
"config": {"url": "https://example.com/webhook"},
}
submission_data = {"site_id": 1, "site_name": "Test", "submission_id": 100, "fields": {}}
with mock.patch("app.services.action_pipeline._HANDLERS"):
# Test that request exceptions are handled gracefully
import requests
with mock.patch("requests.post", side_effect=requests.RequestException("Connection refused")):
result = _handle_webhook(action, submission_data)
self.assertFalse(result["success"])
def test_webhook_201_success(self):
action = {
"id": 1,
"config": {"url": "https://example.com/webhook"},
}
submission_data = {"site_id": 1, "site_name": "Test", "submission_id": 100, "fields": {}}
mock_resp = mock.MagicMock()
mock_resp.status_code = 201
mock_resp.text = "Created"
with mock.patch("requests.post", return_value=mock_resp):
result = _handle_webhook(action, submission_data)
self.assertTrue(result["success"])
class TestHandleEmail(unittest.TestCase):
"""Test email action handler (1.3)."""
def test_email_no_recipients(self):
action = {"id": 1, "config": {"recipients": []}}
submission_data = {"site_id": 1, "site_name": "Test", "submission_id": 100, "fields": {}}
result = _handle_email(action, submission_data)
self.assertFalse(result["success"])
self.assertEqual(result["error"], "no_recipients")
def test_email_default_body(self):
action = {
"id": 1,
"config": {
"recipients": ["test@example.com"],
"subject": "Test Subject",
}
}
submission_data = {
"site_id": 1,
"site_name": "Test Site",
"submission_id": 100,
"fields": {"name": "John", "email": "john@example.com"},
}
with mock.patch("os.environ.get", side_effect=lambda k, d="": {"SMTP_HOST": "smtp.example.com", "SMTP_PORT": "587", "SMTP_USER": "user", "SMTP_PASS": "pass"}.get(k, d)):
with mock.patch("smtplib.SMTP") as mock_smtp:
result = _handle_email(action, submission_data)
self.assertTrue(result["success"])
self.assertEqual(result["recipients"], 1)
def test_email_template_body(self):
action = {
"id": 1,
"config": {
"recipients": ["test@example.com"],
"subject": "New lead: {{name}}",
"body": "Name: {{name}}, Email: {{email}}",
}
}
submission_data = {
"site_id": 1,
"site_name": "Test Site",
"submission_id": 100,
"fields": {"name": "John", "email": "john@example.com"},
}
with mock.patch("os.environ.get", side_effect=lambda k, d="": {"SMTP_HOST": "smtp.example.com", "SMTP_PORT": "587", "SMTP_USER": "user", "SMTP_PASS": "pass"}.get(k, d)):
with mock.patch("smtplib.SMTP") as mock_smtp:
result = _handle_email(action, submission_data)
self.assertTrue(result["success"])
def test_email_smtp_not_configured(self):
action = {
"id": 1,
"config": {"recipients": ["test@example.com"]},
}
submission_data = {"site_id": 1, "site_name": "Test", "submission_id": 100, "fields": {}}
# Save original env and clear SMTP_HOST
orig_host = os.environ.get("SMTP_HOST")
orig_port = os.environ.get("SMTP_PORT")
os.environ.pop("SMTP_HOST", None)
os.environ.pop("SMTP_PORT", None)
try:
result = _handle_email(action, submission_data)
self.assertFalse(result["success"])
self.assertEqual(result["error"], "smtp_not_configured")
finally:
# Restore original env
if orig_host:
os.environ["SMTP_HOST"] = orig_host
if orig_port:
os.environ["SMTP_PORT"] = orig_port
class TestHandleLog(unittest.TestCase):
"""Test log action handler (1.4)."""
def test_log_default_message(self):
action = {"id": 1, "config": {}}
submission_data = {
"site_id": 1,
"site_name": "Test Site",
"submission_id": 100,
"fields": {},
}
result = _handle_log(action, submission_data)
self.assertTrue(result["success"])
self.assertIn("100", result["message"])
def test_log_custom_template(self):
action = {
"id": 1,
"config": {
"level": "warning",
"message": "New lead: {name} from {site_name}",
}
}
submission_data = {
"site_id": 1,
"site_name": "Test Site",
"submission_id": 100,
"fields": {"name": "John"},
}
result = _handle_log(action, submission_data)
self.assertTrue(result["success"])
self.assertIn("John", result["message"])
self.assertIn("Test Site", result["message"])
class TestHandleRedirect(unittest.TestCase):
"""Test redirect action handler (1.5)."""
def test_redirect_success(self):
action = {
"id": 1,
"config": {"url": "https://example.com/thanks"},
}
submission_data = {
"site_id": 1,
"site_name": "Test Site",
"submission_id": 100,
"fields": {"name": "John"},
}
result = _handle_redirect(action, submission_data)
self.assertTrue(result["success"])
self.assertEqual(result["redirect_url"], "https://example.com/thanks")
def test_redirect_variable_substitution(self):
action = {
"id": 1,
"config": {"url": "https://example.com/thanks/{name}/sub/{submission_id}"},
}
submission_data = {
"site_id": 1,
"site_name": "Test Site",
"submission_id": 100,
"fields": {"name": "John"},
}
result = _handle_redirect(action, submission_data)
self.assertTrue(result["success"])
self.assertEqual(result["redirect_url"], "https://example.com/thanks/John/sub/100")
def test_redirect_missing_url(self):
action = {"id": 1, "config": {}}
submission_data = {
"site_id": 1,
"site_name": "Test",
"submission_id": 100,
"fields": {},
}
result = _handle_redirect(action, submission_data)
self.assertFalse(result["success"])
self.assertEqual(result["error"], "missing_url")
class TestHandleDocument(unittest.TestCase):
"""Test document action handler (1.6)."""
def test_document_missing_template_id(self):
action = {"id": 1, "config": {}}
submission_data = {
"site_id": 1,
"site_name": "Test",
"submission_id": 100,
"fields": {},
}
result = _handle_document(action, submission_data)
self.assertFalse(result["success"])
self.assertEqual(result["error"], "missing_template_id")
def test_document_template_not_found(self):
action = {"id": 1, "config": {"template_id": "nonexistent"}}
submission_data = {
"site_id": 1,
"site_name": "Test",
"submission_id": 100,
"fields": {},
}
with mock.patch("app.models.get_document_template", return_value=None):
result = _handle_document(action, submission_data)
self.assertFalse(result["success"])
self.assertEqual(result["error"], "template_not_found")
class TestExecuteActions(unittest.TestCase):
"""Test full action pipeline execution (1.7-1.9)."""
def setUp(self):
import hashlib
self.app = flask_app
self.app.config["TESTING"] = True
self.client = self.app.test_client()
with self.app.app_context():
email = "test@test.com"
email_hash = hashlib.sha256(email.encode()).hexdigest()
# Create test user via register_user (handles encryption)
# Catch duplicate email from repeated runs
try:
app.models.register_user(email, "testpassword123", "Test User")
except (ValueError, sqlite3.IntegrityError):
pass # Already registered from previous test
self.user = app.models.find_user_by_email(email)
self.user_id = self.user["id"]
# Create test site
self.site = app.models.add_site("Test Site", email, user_id=self.user_id)
self.token = self.site["token"]
def test_no_actions_returns_empty(self):
site = {"id": self.site["id"], "name": "Test Site"}
submission = {"id": 100, "submitted_at": "2026-06-14T12:00:00"}
result = execute_actions(site, submission)
self.assertEqual(result["results"], [])
self.assertIsNone(result["redirect_url"])
def test_disabled_action_skipped(self):
# Create a disabled action
app.models.create_action(
site_id=self.site["id"],
action_type="log",
config={"message": "test"},
enabled=False,
)
site = {"id": self.site["id"], "name": "Test Site"}
submission = {"id": 100, "submitted_at": "2026-06-14T12:00:00"}
result = execute_actions(site, submission)
self.assertEqual(len(result["results"]), 0)
def test_action_ordering(self):
# Create actions in specific order
app.models.create_action(
site_id=self.site["id"],
action_type="log",
config={"message": "first"},
execution_order=1,
)
app.models.create_action(
site_id=self.site["id"],
action_type="log",
config={"message": "second"},
execution_order=2,
)
site = {"id": self.site["id"], "name": "Test Site"}
submission = {"id": 100, "submitted_at": "2026-06-14T12:00:00"}
result = execute_actions(site, submission)
self.assertEqual(len(result["results"]), 2)
def test_redirect_url_captured(self):
app.models.create_action(
site_id=self.site["id"],
action_type="redirect",
config={"url": "https://example.com/thanks"},
execution_order=1,
)
site = {"id": self.site["id"], "name": "Test Site"}
submission = {"id": 100, "submitted_at": "2026-06-14T12:00:00"}
result = execute_actions(site, submission)
self.assertEqual(result["redirect_url"], "https://example.com/thanks")
def test_multiple_actions_executed(self):
app.models.create_action(
site_id=self.site["id"],
action_type="log",
config={"message": "action1"},
execution_order=1,
)
app.models.create_action(
site_id=self.site["id"],
action_type="redirect",
config={"url": "https://example.com/thanks"},
execution_order=2,
)
site = {"id": self.site["id"], "name": "Test Site"}
submission = {"id": 100, "submitted_at": "2026-06-14T12:00:00"}
result = execute_actions(site, submission)
self.assertEqual(len(result["results"]), 2)
self.assertTrue(all(r["success"] for r in result["results"]))
def test_webhook_action_integration(self):
app.models.create_action(
site_id=self.site["id"],
action_type="webhook",
config={"url": "https://example.com/webhook"},
execution_order=1,
)
site = {"id": self.site["id"], "name": "Test Site"}
submission = {"id": 100, "customer_name": "John", "submitted_at": "2026-06-14T12:00:00"}
mock_resp = mock.MagicMock()
mock_resp.status_code = 200
mock_resp.text = "OK"
with mock.patch("requests.post", return_value=mock_resp):
result = execute_actions(site, submission)
self.assertEqual(len(result["results"]), 1)
self.assertTrue(result["results"][0]["success"])
def test_error_does_not_block_subsequent_actions(self):
# First action fails, second succeeds
app.models.create_action(
site_id=self.site["id"],
action_type="webhook",
config={"url": ""}, # Will fail
execution_order=1,
)
app.models.create_action(
site_id=self.site["id"],
action_type="log",
config={"message": "still runs"},
execution_order=2,
)
site = {"id": self.site["id"], "name": "Test Site"}
submission = {"id": 100, "submitted_at": "2026-06-14T12:00:00"}
result = execute_actions(site, submission)
self.assertEqual(len(result["results"]), 2)
self.assertFalse(result["results"][0]["success"])
self.assertTrue(result["results"][1]["success"])
class TestApiIntegration(unittest.TestCase):
"""Integration test: /api/submit with action execution."""
def setUp(self):
self.app = flask_app
self.app.config["TESTING"] = True
self.client = self.app.test_client()
try:
app.models.register_user("api_test@test.com", "testpassword123", "API Test")
except ValueError:
pass # Already registered
self.user = app.models.find_user_by_email("api_test@test.com")
self.user_id = self.user["id"]
self.site = app.models.add_site("Test Site", "api_test@test.com", user_id=self.user_id)
self.token = self.site["token"]
# Add a log action
app.models.create_action(
site_id=self.site["id"],
action_type="log",
config={"message": "API submission received"},
execution_order=1,
)
def test_submission_triggers_actions(self):
resp = self.client.post(
f"/api/submit?token={self.token}",
data={"name": "John Doe"},
content_type="application/x-www-form-urlencoded",
)
self.assertEqual(resp.status_code, 201) # Created
if __name__ == "__main__":
unittest.main()