#!/usr/bin/env python3
"""Integration test for Phase 2 email tracking."""
import os
import sys
os.environ.setdefault('RELAY_DB_PATH', '/tmp/test_integration.db')
os.environ.setdefault('APP_URL', 'https://agentforms.io')
os.environ.setdefault('SECRET_KEY', 'test-secret-key')
from app.app import create_app
from app.models import get_db
app = create_app()
errors = []
passed = 0
def check(name, condition):
global passed
if condition:
print(f" [PASS] {name}")
passed += 1
else:
print(f" [FAIL] {name}")
errors.append(name)
print("=" * 60)
print("PHASE 2 INTEGRATION TEST - Email Tracking")
print("=" * 60)
# --- 1. Schema migration ---
print("\n[1] Schema migration - columns exist in campaign_recipients")
conn = get_db()
info = conn.execute("PRAGMA table_info(campaign_recipients)").fetchall()
cols = {r["name"] for r in info}
check("tracking_token column exists", "tracking_token" in cols)
check("opened_at column exists", "opened_at" in cols)
check("clicked_at column exists", "clicked_at" in cols)
check("bounce_type column exists", "bounce_type" in cols)
check("bounce_reason column exists", "bounce_reason" in cols)
idxs = {r["name"] for r in conn.execute("PRAGMA index_list(campaign_recipients)").fetchall()}
check("tracking_token index exists", "idx_campaign_recipients_tracking_token" in idxs)
conn.close()
# --- 2. Create test user & campaign via DB directly ---
print("\n[2] Create test user and campaign")
from app.models import (
register_user, create_campaign, add_campaign_recipients,
get_campaign_recipients, get_campaign_analytics
)
try:
user = register_user("integration.test@example.com", "TestPass123!", "Integration Test")
user_id = user["id"]
check("User created", user_id > 0)
except Exception as e:
print(f" User creation error: {e}")
user_id = 1
check("User created", False)
resp = create_campaign(user_id, "Integration Test Campaign", "Test Subject", "Test body with <a href='https://example.com'>link</a>.")
check("Campaign created", resp.get("id") is not None)
campaign_id = resp.get("id")
if campaign_id:
print(f" Campaign ID: {campaign_id}")
resp = add_campaign_recipients(campaign_id, user_id, [
"alice@example.com",
"bob@example.com",
"charlie@example.com",
"dave@example.com",
"eve@example.com"
])
check("Recipients added", resp.get("added") == 5)
print(f" Added: {resp}")
# --- 3. Verify tracking tokens ---
print("\n[3] Tracking tokens generated for recipients")
conn = get_db()
recipients = conn.execute(
"SELECT id, email, tracking_token, status FROM campaign_recipients WHERE campaign_id = ?",
(campaign_id,)
).fetchall()
conn.close()
rlist = [dict(r) for r in recipients]
check("5 recipients in DB", len(rlist) == 5)
tokens = [r["tracking_token"] for r in rlist if r["tracking_token"]]
check("All have tracking tokens", len(tokens) == 5)
check("All tokens unique", len(set(tokens)) == 5)
check("Token length reasonable (>16 chars)", all(len(t) > 16 for t in tokens))
token_alice = [r for r in rlist if r["email"] == "alice@example.com"][0]["tracking_token"]
token_bob = [r for r in rlist if r["email"] == "bob@example.com"][0]["tracking_token"]
# --- 4. Open tracking ---
print("\n[4] Open tracking - simulate email clients fetching pixel")
with app.test_client() as c:
resp = c.get(f'/tracking/open/{token_alice}')
check("Open pixel returns 200", resp.status_code == 200)
check("Open pixel is GIF", resp.content_type == "image/gif")
resp = c.get(f'/tracking/open/{token_bob}')
check("Bob open pixel 200", resp.status_code == 200)
resp = c.get(f'/tracking/open/{token_alice}')
check("Duplicate open returns 200", resp.status_code == 200)
resp = c.get('/tracking/open/nonexistent_token_12345')
check("Invalid token returns 200 (no leak)", resp.status_code == 200)
# --- 5. Verify opens in DB ---
print("\n[5] Open data persisted correctly")
conn = get_db()
alice = dict(conn.execute(
"SELECT email, opened_at FROM campaign_recipients WHERE email = 'alice@example.com' AND campaign_id = ?",
(campaign_id,)
).fetchone())
bob = dict(conn.execute(
"SELECT email, opened_at FROM campaign_recipients WHERE email = 'bob@example.com' AND campaign_id = ?",
(campaign_id,)
).fetchone())
charlie = dict(conn.execute(
"SELECT email, opened_at FROM campaign_recipients WHERE email = 'charlie@example.com' AND campaign_id = ?",
(campaign_id,)
).fetchone())
conn.close()
check("Alice has opened_at", alice.get("opened_at") is not None)
check("Bob has opened_at", bob.get("opened_at") is not None)
check("Charlie NOT opened", charlie.get("opened_at") is None)
# --- 6. Click tracking ---
print("\n[6] Click tracking - simulate clicking links in email")
with app.test_client() as c:
resp = c.get(f'/tracking/click/{token_alice}?url=https://agentforms.io/signup')
check("Click redirect 302", resp.status_code == 302)
check("Redirects to destination", 'agentforms.io/signup' in resp.headers.get('Location', ''))
resp = c.get(f'/tracking/click/{token_bob}?url=https://agentforms.io/pricing')
check("Bob click redirect", resp.status_code == 302)
# --- 7. Verify clicks in DB ---
print("\n[7] Click data persisted correctly")
conn = get_db()
alice_row = dict(conn.execute(
"SELECT email, clicked_at FROM campaign_recipients WHERE email = 'alice@example.com' AND campaign_id = ?",
(campaign_id,)
).fetchone())
charlie_row = dict(conn.execute(
"SELECT email, clicked_at FROM campaign_recipients WHERE email = 'charlie@example.com' AND campaign_id = ?",
(campaign_id,)
).fetchone())
conn.close()
check("Alice has clicked_at", alice_row.get("clicked_at") is not None)
check("Charlie NOT clicked", charlie_row.get("clicked_at") is None)
# --- 8. Bounce tracking ---
print("\n[8] Bounce tracking - simulate SMTP bounce notification")
with app.test_client() as c:
resp = c.post('/tracking/bounce', json={
"email": "dave@example.com",
"bounce_type": "hard",
"reason": "User unknown"
})
bounce_data = resp.get_json()
check("Bounce returns 200", resp.status_code == 200)
check("Bounce success", bounce_data.get("success") is True)
# --- 9. Verify bounce in DB ---
print("\n[9] Bounce data persisted correctly")
conn = get_db()
dave = dict(conn.execute(
"SELECT email, status, bounce_type, bounce_reason FROM campaign_recipients WHERE email = 'dave@example.com' AND campaign_id = ?",
(campaign_id,)
).fetchone())
conn.close()
check("Dave status=bounced", dave.get("status") == "bounced")
check("Dave bounce_type=hard", dave.get("bounce_type") == "hard")
check("Dave has bounce_reason", dave.get("bounce_reason") is not None)
# --- 10. Analytics API ---
print("\n[10] Analytics API - verify aggregated metrics")
analytics = get_campaign_analytics(campaign_id, user_id)
check("Analytics returns data", analytics is not None)
if analytics:
check("Has total_recipients", "total_recipients" in analytics)
check("Has unique_opens", "unique_opens" in analytics)
check("Has unique_clicks", "unique_clicks" in analytics)
check("Has unique_bounces", "unique_bounces" in analytics)
check("Has open_rate", "open_rate" in analytics)
check("Has click_rate", "click_rate" in analytics)
check("Has bounce_rate", "bounce_rate" in analytics)
print(f" Analytics result: {analytics}")
check("Total recipients = 5", analytics.get("total_recipients") == 5)
check("Unique opens = 2", analytics.get("unique_opens") == 2)
check("Unique clicks = 2", analytics.get("unique_clicks") == 2)
check("Unique bounces = 1", analytics.get("unique_bounces") == 1)
check("Open rate = 40pct", analytics.get("open_rate") == 0.4)
check("Bounce rate = 20pct", analytics.get("bounce_rate") == 0.2)
# --- 11. Email injection ---
print("\n[11] Email HTML injection - verify tracking injected correctly")
from app.services.campaign_emails import _inject_tracking
test_html = '<html><body><p><a href="https://example.com">Link</a> <a href="mailto:x@example.com">Email</a></p></body></html>'
result = _inject_tracking(test_html, "testtoken", "https://agentforms.io")
check("URL rewritten to tracking", "tracking/click/testtoken?url=" in result)
check("Mailto preserved", "mailto:x@example.com" in result)
check("Tracking pixel injected", "tracking/open/testtoken" in result)
check("Pixel is 1x1", 'width="1" height="1"' in result)
check("Pixel is hidden", "display:none" in result)
# --- Summary ---
total = passed + len(errors)
print("\n" + "=" * 60)
print(f"RESULTS: {passed} passed, {len(errors)} failed out of {total} checks")
if errors:
print(f" FAILED: {' | '.join(errors)}")
else:
print(" ALL CHECKS PASSED")
print("=" * 60)
sys.exit(1 if errors else 0)