#!/usr/bin/env python3
"""Integration test suite for AgentForms v2 API.
Uses Flask test client with an in-memory SQLite database.
Run standalone: python3 tests/test_integration.py
Requires: flask, bcrypt (from app requirements)
Covers all /api/v2/ routes: key management, form CRUD, submissions, permissions.
"""
import json
import hashlib
import os
import sys
import sqlite3
import tempfile
import unittest
# Ensure the project root is on sys.path so we can import app
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import bcrypt
# Use shared test DB — do NOT override app.models.DB_PATH
import tests.test_shared_db # noqa: F401
os.environ.setdefault("ENCRYPTION_KEY", "5c321ae59453d7bae2e05a9e19e97347411a033d9858a0456d5cc63acfcb0369")
import app.models
from app.app import app as flask_app
class TestBase(unittest.TestCase):
"""Base test class with app factory and test user setup."""
@classmethod
def setUpClass(cls):
"""Set up test user + API key in the shared DB."""
cls.app = flask_app
cls.app.config["TESTING"] = True
cls.client = cls.app.test_client()
cls.created_user_ids = []
cls.created_key_ids = []
cls.created_site_tokens = []
# Create test user with unique email per class to avoid UNIQUE constraint collisions
test_email = f"{cls.__name__.lower()}@agentforms.test"
email_hash = hashlib.sha256(test_email.encode()).hexdigest()
pwd_hash = bcrypt.hashpw(b"testpass123", bcrypt.gensalt()).decode()
with cls.app.app_context():
conn = app.models.get_db()
cursor = conn.execute(
"INSERT INTO users (email_hash, email_encrypted, password_hash, name_encrypted, tier) VALUES (?, ?, ?, ?, ?)",
(email_hash, test_email, pwd_hash, "Test User", "pro"),
)
conn.commit()
cls.user_id = cursor.lastrowid
cls.created_user_ids.append(cls.user_id)
conn.close()
# Reset rate limiter (shared Redis across test classes)
from app.services.ratelimit import limiter
limiter.flush("rate:*")
@classmethod
def tearDownClass(cls):
"""Clean up only this class's test data."""
if not hasattr(cls, 'created_user_ids') or not cls.created_user_ids:
return
with cls.app.app_context():
conn = app.models.get_db()
for uid in cls.created_user_ids:
# FK chain: submissions.site_id → sites.id; sites.user_id → users.id
# Delete in reverse FK order to avoid constraint violations
conn.execute("""DELETE FROM webhook_logs WHERE submission_id IN (
SELECT id FROM submissions WHERE site_id IN (
SELECT id FROM sites WHERE user_id = ?))""", (uid,))
conn.execute("""DELETE FROM email_send_log WHERE submission_id IN (
SELECT id FROM submissions WHERE site_id IN (
SELECT id FROM sites WHERE user_id = ?))""", (uid,))
conn.execute("""DELETE FROM documents WHERE submission_id IN (
SELECT id FROM submissions WHERE site_id IN (
SELECT id FROM sites WHERE user_id = ?))""", (uid,))
conn.execute("""DELETE FROM submissions WHERE site_id IN (
SELECT id FROM sites WHERE user_id = ?)""", (uid,))
conn.execute("""DELETE FROM site_analytics WHERE site_id IN (
SELECT id FROM sites WHERE user_id = ?)""", (uid,))
conn.execute("""DELETE FROM webhook_destinations WHERE site_id IN (
SELECT id FROM sites WHERE user_id = ?)""", (uid,))
conn.execute("""DELETE FROM ai_prompt_history WHERE site_id IN (
SELECT id FROM sites WHERE user_id = ?)""", (uid,))
conn.execute("""DELETE FROM form_versions WHERE site_id IN (
SELECT id FROM sites WHERE user_id = ?)""", (uid,))
conn.execute("""DELETE FROM form_sessions WHERE site_id IN (
SELECT id FROM sites WHERE user_id = ?)""", (uid,))
conn.execute("""DELETE FROM form_variants WHERE site_id IN (
SELECT id FROM sites WHERE user_id = ?)""", (uid,))
conn.execute("""DELETE FROM form_impressions WHERE site_id IN (
SELECT id FROM sites WHERE user_id = ?)""", (uid,))
conn.execute("""DELETE FROM form_actions WHERE site_id IN (
SELECT id FROM sites WHERE user_id = ?)""", (uid,))
conn.execute("""DELETE FROM document_templates WHERE site_id IN (
SELECT id FROM sites WHERE user_id = ?)""", (uid,))
conn.execute("""DELETE FROM documents WHERE site_id IN (
SELECT id FROM sites WHERE user_id = ?)""", (uid,))
conn.execute("""DELETE FROM invoice_schedules WHERE site_id IN (
SELECT id FROM sites WHERE user_id = ?)""", (uid,))
conn.execute("DELETE FROM monthly_usage WHERE user_id = ?", (uid,))
conn.execute("DELETE FROM email_send_log WHERE user_id = ?", (uid,))
conn.execute("DELETE FROM custom_templates WHERE user_id = ?", (uid,))
conn.execute("DELETE FROM email_settings WHERE user_id = ?", (uid,))
conn.execute("DELETE FROM campaign_recipients WHERE campaign_id IN (SELECT id FROM campaigns WHERE user_id = ?)", (uid,))
conn.execute("DELETE FROM campaign_ab_variants WHERE campaign_id IN (SELECT id FROM campaigns WHERE user_id = ?)", (uid,))
conn.execute("DELETE FROM campaign_reminders WHERE campaign_id IN (SELECT id FROM campaigns WHERE user_id = ?)", (uid,))
conn.execute("DELETE FROM campaigns WHERE user_id = ?", (uid,))
conn.execute("DELETE FROM sites WHERE user_id = ?", (uid,))
conn.execute("DELETE FROM documents WHERE user_id = ?", (uid,))
conn.execute("DELETE FROM invoice_schedules WHERE user_id = ?", (uid,))
conn.execute("DELETE FROM document_templates WHERE user_id = ?", (uid,))
conn.execute("DELETE FROM team_members WHERE user_id = ?", (uid,))
conn.execute("DELETE FROM teams WHERE owner_id = ?", (uid,))
conn.execute("DELETE FROM api_keys WHERE user_id = ?", (uid,))
conn.execute("DELETE FROM users WHERE id = ?", (uid,))
conn.commit()
conn.close()
# Reset rate limiter
from app.services.ratelimit import limiter
limiter.flush("rate:*")
def create_api_key(self, name="test key", permissions=None):
"""Helper: create an API key for the test user, return (full_key, key_id)."""
if permissions is None:
permissions = {
"read_forms": True,
"write_forms": True,
"read_submissions": True,
"delete_forms": True,
}
import secrets
raw_key = f"afk_live_{secrets.token_hex(24)}"
key_hash = bcrypt.hashpw(raw_key.encode(), bcrypt.gensalt()).decode()
key_prefix = raw_key[:12]
with self.app.app_context():
conn = app.models.get_db()
cursor = conn.execute(
"INSERT INTO api_keys (user_id, key_hash, key_prefix, name, permissions) VALUES (?, ?, ?, ?, ?)",
(self.user_id, key_hash, key_prefix, name, json.dumps(permissions)),
)
conn.commit()
key_id = cursor.lastrowid
conn.close()
return raw_key, key_id
class TestAPIKeyManagement(TestBase):
"""Test /api/v2/keys endpoints."""
def setUp(self):
self.key, self.key_id = self.create_api_key("Admin Key")
self.auth = {"Authorization": f"Bearer {self.key}"}
def test_list_keys(self):
resp = self.client.get("/api/v2/keys", headers=self.auth)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertIn("keys", data)
self.assertGreaterEqual(len(data["keys"]), 1)
def test_create_key(self):
resp = self.client.post(
"/api/v2/keys",
headers=self.auth,
json={"name": "New Key", "permissions": {"read_forms": True}},
)
self.assertEqual(resp.status_code, 201)
data = resp.get_json()
self.assertIn("full_key", data)
self.assertTrue(data["full_key"].startswith("afk_live_"))
self.assertEqual(data["name"], "New Key")
def test_create_key_duplicate_name(self):
# Create first key
self.client.post(
"/api/v2/keys",
headers=self.auth,
json={"name": "Dup Test", "permissions": {"read_forms": True}},
)
# Try duplicate
resp = self.client.post(
"/api/v2/keys",
headers=self.auth,
json={"name": "Dup Test", "permissions": {"read_forms": True}},
)
self.assertEqual(resp.status_code, 409)
def test_delete_key(self):
# Create a key to delete
key_to_del, key_id = self.create_api_key("To Delete")
auth_del = {"Authorization": f"Bearer {key_to_del}"}
resp = self.client.delete(f"/api/v2/keys/{key_id}", headers=auth_del)
self.assertEqual(resp.status_code, 200)
self.assertTrue(resp.get_json()["revoked"])
def test_delete_key_no_delete_permission(self):
# Create a key without delete_forms permission
ro_key, ro_id = self.create_api_key(
"Read Only", permissions={"read_forms": True, "write_forms": False}
)
ro_auth = {"Authorization": f"Bearer {ro_key}"}
resp = self.client.delete(f"/api/v2/keys/{ro_id}", headers=ro_auth)
self.assertEqual(resp.status_code, 403)
def test_unauthorized(self):
resp = self.client.get("/api/v2/keys")
self.assertEqual(resp.status_code, 401)
def test_invalid_key(self):
resp = self.client.get("/api/v2/keys", headers={"Authorization": "Bearer afk_live_invalid"})
self.assertEqual(resp.status_code, 401)
def _reset_sites():
"""Reset form-related tables (keeps user + API keys intact).
Needed because Free tier has a 2-site limit and multiple test classes create forms."""
pass # No longer needed — per-class cleanup handles this
class TestFormCRUD(TestBase):
"""Test /api/v2/forms endpoints."""
def setUp(self):
self.key, self.key_id = self.create_api_key("Form Key")
self.auth = {"Authorization": f"Bearer {self.key}"}
def test_create_form(self):
fields = [
{"name": "email", "label": "Email", "type": "email", "required": True},
{"name": "message", "label": "Message", "type": "textarea", "required": True},
]
resp = self.client.post(
"/api/v2/forms",
headers=self.auth,
json={"name": "Test Form", "fields": fields},
)
self.assertEqual(resp.status_code, 201)
data = resp.get_json()
self.assertEqual(data["name"], "Test Form")
self.assertIn("token", data)
self.assertEqual(data["created_by"], "agent")
self.assertEqual(len(data["fields"]), 2)
self.form_token = data["token"]
def test_create_form_empty_fields(self):
resp = self.client.post(
"/api/v2/forms",
headers=self.auth,
json={"name": "Empty", "fields": []},
)
self.assertEqual(resp.status_code, 422)
def test_list_forms(self):
# Create a form first
self.client.post(
"/api/v2/forms",
headers=self.auth,
json={"name": "Listable", "fields": [{"name": "x", "label": "X", "type": "text"}]},
)
resp = self.client.get("/api/v2/forms", headers=self.auth)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertGreaterEqual(data["total"], 1)
self.assertIn("forms", data)
def test_get_form(self):
# Create then get
create_resp = self.client.post(
"/api/v2/forms",
headers=self.auth,
json={"name": "Gettable", "fields": [{"name": "x", "label": "X", "type": "text"}]},
)
token = create_resp.get_json()["token"]
resp = self.client.get(f"/api/v2/forms/{token}", headers=self.auth)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertEqual(data["name"], "Gettable")
def test_update_fields(self):
create_resp = self.client.post(
"/api/v2/forms",
headers=self.auth,
json={"name": "Updateable", "fields": [{"name": "x", "label": "X", "type": "text"}]},
)
token = create_resp.get_json()["token"]
new_fields = [
{"name": "email", "label": "Email", "type": "email", "required": True},
{"name": "phone", "label": "Phone", "type": "tel"},
]
resp = self.client.put(
f"/api/v2/forms/{token}/fields",
headers=self.auth,
json={"fields": new_fields},
)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertTrue(data["updated"])
self.assertEqual(data["field_count"], 2)
def test_delete_form(self):
create_resp = self.client.post(
"/api/v2/forms",
headers=self.auth,
json={"name": "Deletable", "fields": [{"name": "x", "label": "X", "type": "text"}]},
)
token = create_resp.get_json()["token"]
resp = self.client.delete(f"/api/v2/forms/{token}", headers=self.auth)
self.assertEqual(resp.status_code, 200)
self.assertTrue(resp.get_json()["deleted"])
def test_delete_form_no_permission(self):
ro_key, _ = self.create_api_key(
"RO", permissions={"read_forms": True, "write_forms": False, "delete_forms": False}
)
ro_auth = {"Authorization": f"Bearer {ro_key}"}
# Create a form with admin key
create_resp = self.client.post(
"/api/v2/forms",
headers=self.auth,
json={"name": "Protected", "fields": [{"name": "x", "label": "X", "type": "text"}]},
)
token = create_resp.get_json()["token"]
resp = self.client.delete(f"/api/v2/forms/{token}", headers=ro_auth)
self.assertEqual(resp.status_code, 403)
def test_delete_form_with_submissions(self):
"""Test that deleting a form with submissions works (cascade delete)."""
create_resp = self.client.post(
"/api/v2/forms",
headers=self.auth,
json={
"name": "Has Subs",
"fields": [{"name": "q", "label": "Q", "type": "text", "required": True}],
},
)
token = create_resp.get_json()["token"]
# Submit something
self.client.post(
f"/api/submit?token={token}",
content_type="application/json",
data=json.dumps({"q": "test submission"}),
)
# Delete should succeed despite FK constraint
resp = self.client.delete(f"/api/v2/forms/{token}", headers=self.auth)
self.assertEqual(resp.status_code, 200)
self.assertTrue(resp.get_json()["deleted"])
def test_form_not_found(self):
resp = self.client.get("/api/v2/forms/fr-nonexistent", headers=self.auth)
self.assertEqual(resp.status_code, 404)
class TestSubmissions(TestBase):
"""Test /api/v2/forms/{token}/submissions endpoints."""
def setUp(self):
super().setUp()
self.key, self.key_id = self.create_api_key("Sub Key")
self.auth = {"Authorization": f"Bearer {self.key}"}
# Create a form to work with
create_resp = self.client.post(
"/api/v2/forms",
headers=self.auth,
json={
"name": "Sub Test",
"fields": [
{"name": "email", "label": "Email", "type": "email", "required": True},
{"name": "custom", "label": "Custom", "type": "text"},
],
},
)
self.form_token = create_resp.get_json()["token"]
def test_list_submissions_empty(self):
resp = self.client.get(
f"/api/v2/forms/{self.form_token}/submissions",
headers=self.auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertEqual(data["total"], 0)
def test_submit_and_list(self):
# Submit via public API
sub_data = {"email": "test@example.com", "custom": "hello world"}
resp = self.client.post(
f"/api/submit?token={self.form_token}",
content_type="application/json",
data=json.dumps(sub_data),
)
self.assertEqual(resp.status_code, 201)
self.assertTrue(resp.get_json()["success"])
# List submissions via API
resp = self.client.get(
f"/api/v2/forms/{self.form_token}/submissions",
headers=self.auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertEqual(data["total"], 1)
sub = data["submissions"][0]
# PII fields (customer_email) are now encrypted at rest
self.assertIn("customer_email", sub)
self.assertIsNotNone(sub["customer_email"])
# Dynamic data should be parsed (non-PII)
self.assertIn("dynamic_data", sub)
self.assertEqual(sub["dynamic_data"]["custom"], "hello world")
def test_get_single_submission(self):
# Submit first
self.client.post(
f"/api/submit?token={self.form_token}",
content_type="application/json",
data=json.dumps({"email": "single@test.com"}),
)
# Get the submission
resp = self.client.get(
f"/api/v2/forms/{self.form_token}/submissions",
headers=self.auth,
)
sub_id = resp.get_json()["submissions"][0]["id"]
resp = self.client.get(
f"/api/v2/forms/{self.form_token}/submissions/{sub_id}",
headers=self.auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertEqual(data["id"], sub_id)
# PII fields are encrypted
self.assertIn("customer_email", data)
self.assertIsNotNone(data["customer_email"])
def test_get_submission_wrong_form(self):
# Create another form
self.client.post(
"/api/v2/forms",
headers=self.auth,
json={"name": "Other", "fields": [{"name": "x", "label": "X", "type": "text"}]},
)
resp = self.client.get(
f"/api/v2/forms/{self.form_token}/submissions/99999",
headers=self.auth,
)
self.assertEqual(resp.status_code, 404)
def test_read_submissions_no_permission(self):
ro_key, _ = self.create_api_key(
"No Subs",
permissions={"read_forms": True, "read_submissions": False},
)
ro_auth = {"Authorization": f"Bearer {ro_key}"}
resp = self.client.get(
f"/api/v2/forms/{self.form_token}/submissions",
headers=ro_auth,
)
self.assertEqual(resp.status_code, 403)
class TestPublicConfig(TestBase):
"""Test /api/v2/forms/{token}/config (no auth required)."""
def setUp(self):
super().setUp()
self.key, _ = self.create_api_key("Config Key")
self.auth = {"Authorization": f"Bearer {self.key}"}
create_resp = self.client.post(
"/api/v2/forms",
headers=self.auth,
json={
"name": "Config Test",
"fields": [
{"name": "name", "label": "Name", "type": "text", "required": True},
{"name": "size", "label": "Size", "type": "select", "options": ["S", "M", "L"]},
],
},
)
self.form_token = create_resp.get_json()["token"]
def test_public_config_no_auth(self):
resp = self.client.get(f"/api/v2/forms/{self.form_token}/config")
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertEqual(data["name"], "Config Test")
self.assertEqual(len(data["fields"]), 2)
self.assertIn("submit_endpoint", data)
def test_public_config_unknown_token(self):
resp = self.client.get("/api/v2/forms/fr-nonexistent/config")
self.assertEqual(resp.status_code, 404)
class TestFormCreationNoPermission(TestBase):
"""Test write_forms permission enforcement."""
def setUp(self):
self.key, _ = self.create_api_key(
"Read Only",
permissions={"read_forms": True, "write_forms": False, "delete_forms": False},
)
self.auth = {"Authorization": f"Bearer {self.key}"}
def test_create_form_no_write(self):
resp = self.client.post(
"/api/v2/forms",
headers=self.auth,
json={"name": "Should Fail", "fields": [{"name": "x", "label": "X", "type": "text"}]},
)
self.assertEqual(resp.status_code, 403)
def test_update_fields_no_write(self):
# Create form with admin key first
admin_key, _ = self.create_api_key(
"Admin",
permissions={"read_forms": True, "write_forms": True, "delete_forms": True},
)
admin_auth = {"Authorization": f"Bearer {admin_key}"}
create_resp = self.client.post(
"/api/v2/forms",
headers=admin_auth,
json={"name": "Protected", "fields": [{"name": "x", "label": "X", "type": "text"}]},
)
token = create_resp.get_json()["token"]
# Try to update with read-only key
resp = self.client.put(
f"/api/v2/forms/{token}/fields",
headers=self.auth,
json={"fields": [{"name": "y", "label": "Y", "type": "text"}]},
)
self.assertEqual(resp.status_code, 403)
class TestEmailPhase3(TestBase):
"""Test Phase 3: Template system, personalization, sender domain."""
def setUp(self):
self.key, self.key_id = self.create_api_key("Email Test Key", permissions={
"read_forms": True,
"write_forms": True,
"read_submissions": True,
"delete_forms": True,
"write_templates": True,
})
self.auth = {"Authorization": f"Bearer {self.key}"}
def test_template_render_default(self):
"""POST /api/v2/templates/render with default template."""
resp = self.client.post(
"/api/v2/templates/render",
headers=self.auth,
json={
"template": "default",
"context": {
"subject": "Test Subject",
"body": "<p>Test body</p>",
"recipient": {
"email": "test@example.com",
"first_name": "Alice",
},
"form": {"name": "Test Form"},
},
},
)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertEqual(data["template"], "default")
self.assertIn("html", data)
# Check personalization worked
self.assertIn("Alice", data["html"])
self.assertIn("Test body", data["html"])
def test_template_render_missing_name(self):
"""POST /api/v2/templates/render without template name returns 400."""
resp = self.client.post(
"/api/v2/templates/render",
headers=self.auth,
json={"context": {}},
)
self.assertEqual(resp.status_code, 400)
def test_list_templates(self):
"""GET /api/v2/templates returns builtin and custom templates."""
resp = self.client.get("/api/v2/templates", headers=self.auth)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertIn("builtin_templates", data)
self.assertIn("custom_templates", data)
# Should have at least 'default'
names = [t["name"] for t in data["builtin_templates"]]
self.assertIn("default", names)
def test_create_template(self):
"""POST /api/v2/templates creates a custom template."""
template_html = """<html><body>
<h1>Hello {{recipient.first_name}}</h1>
<p>{{body}}</p>
</body></html>"""
resp = self.client.post(
"/api/v2/templates",
headers=self.auth,
json={
"name": "test-template",
"content": template_html,
"description": "Test template",
},
)
self.assertEqual(resp.status_code, 201)
data = resp.get_json()
self.assertEqual(data["template"]["name"], "test-template")
self.assertEqual(data["template"]["description"], "Test template")
def test_create_template_invalid_name(self):
"""POST /api/v2/templates with invalid name returns 400."""
resp = self.client.post(
"/api/v2/templates",
headers=self.auth,
json={
"name": "123-invalid",
"content": "<p>test</p>",
},
)
self.assertEqual(resp.status_code, 400)
def test_create_template_invalid_syntax(self):
"""POST /api/v2/templates with invalid Jinja2 syntax returns 400."""
resp = self.client.post(
"/api/v2/templates",
headers=self.auth,
json={
"name": "bad-syntax",
"content": "<p>{{unclosed}",
},
)
self.assertEqual(resp.status_code, 400)
def test_get_template_content(self):
"""GET /api/v2/templates/<name>/content returns template content."""
# Create first
self.client.post(
"/api/v2/templates",
headers=self.auth,
json={
"name": "content-test",
"content": "<p>Hello {{recipient.name}}</p>",
},
)
# Get content
resp = self.client.get(
"/api/v2/templates/content-test/content",
headers=self.auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertEqual(data["type"], "custom")
self.assertIn("Hello", data["content"])
def test_update_template(self):
"""PUT /api/v2/templates/<name> updates a template."""
# Create first
self.client.post(
"/api/v2/templates",
headers=self.auth,
json={
"name": "update-test",
"content": "<p>Original</p>",
},
)
# Update
resp = self.client.put(
"/api/v2/templates/update-test",
headers=self.auth,
json={
"content": "<p>Updated</p>",
"description": "Updated description",
},
)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertEqual(data["template"]["description"], "Updated description")
def test_delete_template(self):
"""DELETE /api/v2/templates/<name> removes a template."""
# Create first
self.client.post(
"/api/v2/templates",
headers=self.auth,
json={
"name": "delete-test",
"content": "<p>Delete me</p>",
},
)
# Delete
resp = self.client.delete(
"/api/v2/templates/delete-test",
headers=self.auth,
)
self.assertEqual(resp.status_code, 200)
# Verify gone
resp2 = self.client.get(
"/api/v2/templates/delete-test",
headers=self.auth,
)
self.assertEqual(resp2.status_code, 404)
def test_sender_domain_update(self):
"""PUT /api/v2/email/settings with sender_domain."""
resp = self.client.put(
"/api/v2/email/settings",
headers=self.auth,
json={
"from_name": "Test Sender",
"from_email": "test@example.com",
"sender_domain": "example.com",
},
)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertEqual(data["settings"]["sender_domain"], "example.com")
def test_sender_domain_invalid(self):
"""PUT /api/v2/email/settings with invalid domain returns 400."""
resp = self.client.put(
"/api/v2/email/settings",
headers=self.auth,
json={
"sender_domain": "not-a-domain",
},
)
self.assertEqual(resp.status_code, 400)
def test_sender_domain_verify(self):
"""POST /api/v2/email/domain/verify returns DNS records."""
resp = self.client.post(
"/api/v2/email/domain/verify",
headers=self.auth,
json={"domain": "verify-test.com"},
)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertEqual(data["status"], "pending")
self.assertIn("dns_records", data)
self.assertIn("txt", data["dns_records"])
self.assertIn("dkim", data["dns_records"])
self.assertIn("spf", data["dns_records"])
def test_sender_domain_check(self):
"""GET /api/v2/email/domain/check returns status."""
resp = self.client.get("/api/v2/email/domain/check", headers=self.auth)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertIn("status", data)
self.assertIn(data["status"], ["not_configured", "pending", "verified"])
def test_tier_template_limits(self):
"""Free tier cannot create templates, starter can create 3."""
# Switch to free tier
with self.app.app_context():
import app.models
conn = app.models.get_db()
conn.execute("UPDATE users SET tier = 'free' WHERE id = ?", (self.user_id,))
conn.commit()
conn.close()
resp = self.client.post(
"/api/v2/templates",
headers=self.auth,
json={
"name": "free-test",
"content": "<p>test</p>",
},
)
self.assertEqual(resp.status_code, 403)
data = resp.get_json()
self.assertIn("limit", data["error"])
# Switch back to pro (unlimited-ish)
with self.app.app_context():
conn = app.models.get_db()
conn.execute("UPDATE users SET tier = 'pro' WHERE id = ?", (self.user_id,))
conn.commit()
conn.close()
if __name__ == "__main__":
# Run with verbose output
loader = unittest.TestLoader()
suite = loader.loadTestsFromModule(sys.modules[__name__])
runner = unittest.TextTestRunner(verbosity=2)
result = runner.run(suite)
# Clean up temp DB after all tests
try:
os.unlink(_TEMP_DB_PATH)
except FileNotFoundError:
pass
# Exit with proper code
sys.exit(0 if result.wasSuccessful() else 1)
class TestEmailPhase4(TestBase):
"""Test Phase 4: Reminder campaigns, A/B testing, rate limiting."""
@classmethod
def setUpClass(cls):
super().setUpClass()
with cls.app.app_context():
_reset_sites()
def setUp(self):
self.key, self.key_id = self.create_api_key("Phase4 Test Key", permissions={
"read_forms": True,
"write_forms": True,
"read_submissions": True,
"delete_forms": True,
"write_templates": True,
})
self.auth = {"Authorization": f"Bearer {self.key}"}
def test_create_reminder(self):
"""POST /api/v2/campaigns/<id>/reminders creates a reminder."""
# Create a campaign first
campaign_resp = self.client.post(
"/api/v2/campaigns",
headers=self.auth,
json={
"name": "Reminder Test",
"subject": "Test Subject",
"body": "<p>Test body</p>",
},
)
self.assertEqual(campaign_resp.status_code, 201)
campaign_id = campaign_resp.get_json()["id"]
# Create reminder
resp = self.client.post(
f"/api/v2/campaigns/{campaign_id}/reminders",
headers=self.auth,
json={
"delay_hours": 24,
"subject": "Reminder: {first_name}",
"body": "<p>Don't forget to respond</p>",
},
)
self.assertEqual(resp.status_code, 201)
data = resp.get_json()
self.assertEqual(data["reminder"]["delay_hours"], 24)
self.assertEqual(data["reminder"]["status"], "scheduled")
def test_create_reminder_invalid_delay(self):
"""POST /api/v2/campaigns/<id>/reminders with invalid delay returns 400."""
# Create a campaign first
campaign_resp = self.client.post(
"/api/v2/campaigns",
headers=self.auth,
json={
"name": "Reminder Invalid",
"subject": "Test Subject",
"body": "<p>Test body</p>",
},
)
campaign_id = campaign_resp.get_json()["id"]
# Invalid delay
resp = self.client.post(
f"/api/v2/campaigns/{campaign_id}/reminders",
headers=self.auth,
json={
"delay_hours": -1,
"subject": "Reminder",
"body": "<p>Don't forget</p>",
},
)
self.assertEqual(resp.status_code, 400)
def test_get_reminders(self):
"""GET /api/v2/campaigns/<id>/reminders returns all reminders."""
# Create a campaign
campaign_resp = self.client.post(
"/api/v2/campaigns",
headers=self.auth,
json={
"name": "Reminder List",
"subject": "Test Subject",
"body": "<p>Test body</p>",
},
)
campaign_id = campaign_resp.get_json()["id"]
# Create a reminder
self.client.post(
f"/api/v2/campaigns/{campaign_id}/reminders",
headers=self.auth,
json={
"delay_hours": 12,
"subject": "First Reminder",
"body": "<p>First</p>",
},
)
# Get reminders
resp = self.client.get(
f"/api/v2/campaigns/{campaign_id}/reminders",
headers=self.auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertEqual(len(data["reminders"]), 1)
self.assertEqual(data["reminders"][0]["delay_hours"], 12)
def test_cancel_reminder(self):
"""DELETE /api/v2/campaigns/<id>/reminders/<rid> cancels a reminder."""
# Create a campaign
campaign_resp = self.client.post(
"/api/v2/campaigns",
headers=self.auth,
json={
"name": "Cancel Reminder",
"subject": "Test Subject",
"body": "<p>Test body</p>",
},
)
campaign_id = campaign_resp.get_json()["id"]
# Create a reminder
create_resp = self.client.post(
f"/api/v2/campaigns/{campaign_id}/reminders",
headers=self.auth,
json={
"delay_hours": 24,
"subject": "To Cancel",
"body": "<p>Cancel me</p>",
},
)
reminder_id = create_resp.get_json()["reminder"]["id"]
# Cancel it
resp = self.client.delete(
f"/api/v2/campaigns/{campaign_id}/reminders/{reminder_id}",
headers=self.auth,
)
self.assertEqual(resp.status_code, 200)
# Verify cancelled
get_resp = self.client.get(
f"/api/v2/campaigns/{campaign_id}/reminders",
headers=self.auth,
)
self.assertEqual(get_resp.get_json()["reminders"][0]["status"], "cancelled")
def test_rate_limit_info(self):
"""GET /api/v2/email/rate-limit returns quota info."""
resp = self.client.get(
"/api/v2/email/rate-limit",
headers=self.auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertIn("max_per_hour", data)
self.assertIn("max_per_day", data)
self.assertIn("can_send", data)
self.assertTrue(data["can_send"])
def test_ab_test_create(self):
"""POST /api/v2/campaigns/<id>/ab-test creates A/B variants."""
# Create a campaign
campaign_resp = self.client.post(
"/api/v2/campaigns",
headers=self.auth,
json={
"name": "AB Test",
"subject": "Test Subject",
"body": "<p>Test body</p>",
},
)
campaign_id = campaign_resp.get_json()["id"]
# Create A/B test
resp = self.client.post(
f"/api/v2/campaigns/{campaign_id}/ab-test",
headers=self.auth,
json={
"variants": [
{
"label": "A",
"subject": "Subject A",
"body": "<p>Variant A</p>",
"split_percent": 50,
},
{
"label": "B",
"subject": "Subject B",
"body": "<p>Variant B</p>",
"split_percent": 50,
},
],
"test_size": 20,
"metric": "opens",
"declare_after_hours": 24,
},
)
self.assertEqual(resp.status_code, 201)
data = resp.get_json()
self.assertEqual(len(data["variants"]), 2)
def test_ab_test_info(self):
"""GET /api/v2/campaigns/<id>/ab-test returns A/B test info."""
# Create a campaign
campaign_resp = self.client.post(
"/api/v2/campaigns",
headers=self.auth,
json={
"name": "AB Info",
"subject": "Test Subject",
"body": "<p>Test body</p>",
},
)
campaign_id = campaign_resp.get_json()["id"]
# Create A/B test
self.client.post(
f"/api/v2/campaigns/{campaign_id}/ab-test",
headers=self.auth,
json={
"variants": [
{
"label": "A",
"subject": "Subject A",
"body": "<p>Variant A</p>",
"split_percent": 50,
},
{
"label": "B",
"subject": "Subject B",
"body": "<p>Variant B</p>",
"split_percent": 50,
},
],
"metric": "clicks",
"declare_after_hours": 48,
},
)
# Get info
resp = self.client.get(
f"/api/v2/campaigns/{campaign_id}/ab-test",
headers=self.auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertEqual(data["metric"], "clicks")
self.assertEqual(len(data["variants"]), 2)
def test_ab_test_invalid_variants(self):
"""POST /api/v2/campaigns/<id>/ab-test with <2 variants returns 400."""
campaign_resp = self.client.post(
"/api/v2/campaigns",
headers=self.auth,
json={
"name": "AB Invalid",
"subject": "Test Subject",
"body": "<p>Test body</p>",
},
)
campaign_id = campaign_resp.get_json()["id"]
resp = self.client.post(
f"/api/v2/campaigns/{campaign_id}/ab-test",
headers=self.auth,
json={
"variants": [
{
"label": "A",
"subject": "Subject A",
"body": "<p>Variant A</p>",
"split_percent": 100,
},
],
},
)
self.assertEqual(resp.status_code, 400)
def test_ab_test_invalid_metric(self):
"""POST /api/v2/campaigns/<id>/ab-test with invalid metric returns 400."""
campaign_resp = self.client.post(
"/api/v2/campaigns",
headers=self.auth,
json={
"name": "AB Bad Metric",
"subject": "Test Subject",
"body": "<p>Test body</p>",
},
)
campaign_id = campaign_resp.get_json()["id"]
resp = self.client.post(
f"/api/v2/campaigns/{campaign_id}/ab-test",
headers=self.auth,
json={
"variants": [
{
"label": "A",
"subject": "Subject A",
"body": "<p>Variant A</p>",
"split_percent": 50,
},
{
"label": "B",
"subject": "Subject B",
"body": "<p>Variant B</p>",
"split_percent": 50,
},
],
"metric": "invalid_metric",
},
)
self.assertEqual(resp.status_code, 400)