452 lines
15 KiB
Python
452 lines
15 KiB
Python
#!/usr/bin/env python
|
|
"""
|
|
Test analytics functionality for email tracking
|
|
Run from backend directory with venv activated:
|
|
cd backend
|
|
source venv/bin/activate # or venv\Scripts\activate on Windows
|
|
python test_analytics.py
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
from datetime import datetime, timedelta
|
|
|
|
# Add backend directory to path
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from services.analytics_service import (
|
|
get_open_rate,
|
|
get_click_rate,
|
|
get_newsletter_metrics,
|
|
get_article_performance,
|
|
get_subscriber_activity_status,
|
|
update_subscriber_activity_statuses
|
|
)
|
|
from database import (
|
|
newsletter_sends_collection,
|
|
link_clicks_collection,
|
|
subscriber_activity_collection
|
|
)
|
|
from app import app
|
|
|
|
print("\n" + "="*80)
|
|
print("Analytics Service Tests")
|
|
print("="*80)
|
|
|
|
# Test counters
|
|
tests_passed = 0
|
|
tests_failed = 0
|
|
|
|
def test_result(test_name, passed, message=""):
|
|
"""Print test result"""
|
|
global tests_passed, tests_failed
|
|
if passed:
|
|
tests_passed += 1
|
|
print(f"✓ {test_name}")
|
|
if message:
|
|
print(f" {message}")
|
|
else:
|
|
tests_failed += 1
|
|
print(f"❌ {test_name}")
|
|
if message:
|
|
print(f" {message}")
|
|
|
|
|
|
# Setup test data
|
|
print("\n" + "-"*80)
|
|
print("Setting up test data...")
|
|
print("-"*80)
|
|
|
|
try:
|
|
# Clean up existing test data
|
|
newsletter_sends_collection.delete_many({'newsletter_id': {'$regex': '^test-analytics-'}})
|
|
link_clicks_collection.delete_many({'newsletter_id': {'$regex': '^test-analytics-'}})
|
|
subscriber_activity_collection.delete_many({'email': {'$regex': '^test-analytics-'}})
|
|
|
|
# Create test newsletter sends
|
|
test_newsletter_id = 'test-analytics-newsletter-001'
|
|
|
|
# Create 10 newsletter sends: 7 opened, 3 not opened
|
|
for i in range(10):
|
|
opened = i < 7 # First 7 are opened
|
|
doc = {
|
|
'newsletter_id': test_newsletter_id,
|
|
'subscriber_email': f'test-analytics-user{i}@example.com',
|
|
'tracking_id': f'test-pixel-{i}',
|
|
'sent_at': datetime.utcnow(),
|
|
'opened': opened,
|
|
'first_opened_at': datetime.utcnow() if opened else None,
|
|
'last_opened_at': datetime.utcnow() if opened else None,
|
|
'open_count': 1 if opened else 0,
|
|
'created_at': datetime.utcnow()
|
|
}
|
|
newsletter_sends_collection.insert_one(doc)
|
|
|
|
# Create test link clicks for an article
|
|
test_article_url = 'https://example.com/test-analytics-article'
|
|
|
|
# Create 10 link tracking records: 4 clicked, 6 not clicked
|
|
for i in range(10):
|
|
clicked = i < 4 # First 4 are clicked
|
|
doc = {
|
|
'tracking_id': f'test-link-{i}',
|
|
'newsletter_id': test_newsletter_id,
|
|
'subscriber_email': f'test-analytics-user{i}@example.com',
|
|
'article_url': test_article_url,
|
|
'article_title': 'Test Analytics Article',
|
|
'clicked': clicked,
|
|
'clicked_at': datetime.utcnow() if clicked else None,
|
|
'user_agent': 'Test Agent' if clicked else None,
|
|
'created_at': datetime.utcnow()
|
|
}
|
|
link_clicks_collection.insert_one(doc)
|
|
|
|
print("✓ Test data created")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error setting up test data: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|
|
|
|
|
|
# Test 1: Open Rate Calculation
|
|
print("\n" + "-"*80)
|
|
print("Test 1: Open Rate Calculation")
|
|
print("-"*80)
|
|
|
|
try:
|
|
open_rate = get_open_rate(test_newsletter_id)
|
|
|
|
# Expected: 7 out of 10 = 70%
|
|
is_correct = open_rate == 70.0
|
|
test_result("Calculate open rate", is_correct, f"Open rate: {open_rate}% (expected 70%)")
|
|
|
|
# Test with non-existent newsletter
|
|
open_rate_empty = get_open_rate('non-existent-newsletter')
|
|
handles_empty = open_rate_empty == 0.0
|
|
test_result("Handle non-existent newsletter", handles_empty,
|
|
f"Open rate: {open_rate_empty}% (expected 0%)")
|
|
|
|
except Exception as e:
|
|
test_result("Open rate calculation", False, f"Error: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
|
|
# Test 2: Click Rate Calculation
|
|
print("\n" + "-"*80)
|
|
print("Test 2: Click Rate Calculation")
|
|
print("-"*80)
|
|
|
|
try:
|
|
click_rate = get_click_rate(test_article_url)
|
|
|
|
# Expected: 4 out of 10 = 40%
|
|
is_correct = click_rate == 40.0
|
|
test_result("Calculate click rate", is_correct, f"Click rate: {click_rate}% (expected 40%)")
|
|
|
|
# Test with non-existent article
|
|
click_rate_empty = get_click_rate('https://example.com/non-existent')
|
|
handles_empty = click_rate_empty == 0.0
|
|
test_result("Handle non-existent article", handles_empty,
|
|
f"Click rate: {click_rate_empty}% (expected 0%)")
|
|
|
|
except Exception as e:
|
|
test_result("Click rate calculation", False, f"Error: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
|
|
# Test 3: Newsletter Metrics
|
|
print("\n" + "-"*80)
|
|
print("Test 3: Newsletter Metrics")
|
|
print("-"*80)
|
|
|
|
try:
|
|
metrics = get_newsletter_metrics(test_newsletter_id)
|
|
|
|
# Verify all expected fields
|
|
has_all_fields = all(key in metrics for key in [
|
|
'newsletter_id', 'total_sent', 'total_opened', 'open_rate',
|
|
'total_clicks', 'unique_clickers', 'click_through_rate'
|
|
])
|
|
test_result("Returns all required fields", has_all_fields)
|
|
|
|
# Verify values
|
|
correct_sent = metrics['total_sent'] == 10
|
|
test_result("Correct total_sent", correct_sent, f"Total sent: {metrics['total_sent']}")
|
|
|
|
correct_opened = metrics['total_opened'] == 7
|
|
test_result("Correct total_opened", correct_opened, f"Total opened: {metrics['total_opened']}")
|
|
|
|
correct_open_rate = metrics['open_rate'] == 70.0
|
|
test_result("Correct open_rate", correct_open_rate, f"Open rate: {metrics['open_rate']}%")
|
|
|
|
correct_clicks = metrics['total_clicks'] == 4
|
|
test_result("Correct total_clicks", correct_clicks, f"Total clicks: {metrics['total_clicks']}")
|
|
|
|
correct_unique_clickers = metrics['unique_clickers'] == 4
|
|
test_result("Correct unique_clickers", correct_unique_clickers,
|
|
f"Unique clickers: {metrics['unique_clickers']}")
|
|
|
|
correct_ctr = metrics['click_through_rate'] == 40.0
|
|
test_result("Correct click_through_rate", correct_ctr,
|
|
f"CTR: {metrics['click_through_rate']}%")
|
|
|
|
except Exception as e:
|
|
test_result("Newsletter metrics", False, f"Error: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
|
|
# Test 4: Article Performance
|
|
print("\n" + "-"*80)
|
|
print("Test 4: Article Performance")
|
|
print("-"*80)
|
|
|
|
try:
|
|
performance = get_article_performance(test_article_url)
|
|
|
|
# Verify all expected fields
|
|
has_all_fields = all(key in performance for key in [
|
|
'article_url', 'total_sent', 'total_clicks', 'click_rate',
|
|
'unique_clickers', 'newsletters'
|
|
])
|
|
test_result("Returns all required fields", has_all_fields)
|
|
|
|
# Verify values
|
|
correct_sent = performance['total_sent'] == 10
|
|
test_result("Correct total_sent", correct_sent, f"Total sent: {performance['total_sent']}")
|
|
|
|
correct_clicks = performance['total_clicks'] == 4
|
|
test_result("Correct total_clicks", correct_clicks, f"Total clicks: {performance['total_clicks']}")
|
|
|
|
correct_click_rate = performance['click_rate'] == 40.0
|
|
test_result("Correct click_rate", correct_click_rate, f"Click rate: {performance['click_rate']}%")
|
|
|
|
correct_unique = performance['unique_clickers'] == 4
|
|
test_result("Correct unique_clickers", correct_unique,
|
|
f"Unique clickers: {performance['unique_clickers']}")
|
|
|
|
has_newsletters = len(performance['newsletters']) > 0
|
|
test_result("Returns newsletter list", has_newsletters,
|
|
f"Newsletters: {performance['newsletters']}")
|
|
|
|
except Exception as e:
|
|
test_result("Article performance", False, f"Error: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
|
|
# Test 5: Activity Status Classification
|
|
print("\n" + "-"*80)
|
|
print("Test 5: Activity Status Classification")
|
|
print("-"*80)
|
|
|
|
try:
|
|
# Create test data for activity classification
|
|
now = datetime.utcnow()
|
|
|
|
# Active user (opened 10 days ago)
|
|
newsletter_sends_collection.insert_one({
|
|
'newsletter_id': 'test-analytics-activity',
|
|
'subscriber_email': 'test-analytics-active@example.com',
|
|
'tracking_id': 'test-active-pixel',
|
|
'sent_at': now - timedelta(days=10),
|
|
'opened': True,
|
|
'first_opened_at': now - timedelta(days=10),
|
|
'last_opened_at': now - timedelta(days=10),
|
|
'open_count': 1,
|
|
'created_at': now - timedelta(days=10)
|
|
})
|
|
|
|
# Inactive user (opened 45 days ago)
|
|
newsletter_sends_collection.insert_one({
|
|
'newsletter_id': 'test-analytics-activity',
|
|
'subscriber_email': 'test-analytics-inactive@example.com',
|
|
'tracking_id': 'test-inactive-pixel',
|
|
'sent_at': now - timedelta(days=45),
|
|
'opened': True,
|
|
'first_opened_at': now - timedelta(days=45),
|
|
'last_opened_at': now - timedelta(days=45),
|
|
'open_count': 1,
|
|
'created_at': now - timedelta(days=45)
|
|
})
|
|
|
|
# Dormant user (opened 90 days ago)
|
|
newsletter_sends_collection.insert_one({
|
|
'newsletter_id': 'test-analytics-activity',
|
|
'subscriber_email': 'test-analytics-dormant@example.com',
|
|
'tracking_id': 'test-dormant-pixel',
|
|
'sent_at': now - timedelta(days=90),
|
|
'opened': True,
|
|
'first_opened_at': now - timedelta(days=90),
|
|
'last_opened_at': now - timedelta(days=90),
|
|
'open_count': 1,
|
|
'created_at': now - timedelta(days=90)
|
|
})
|
|
|
|
# New user (never opened)
|
|
newsletter_sends_collection.insert_one({
|
|
'newsletter_id': 'test-analytics-activity',
|
|
'subscriber_email': 'test-analytics-new@example.com',
|
|
'tracking_id': 'test-new-pixel',
|
|
'sent_at': now - timedelta(days=5),
|
|
'opened': False,
|
|
'first_opened_at': None,
|
|
'last_opened_at': None,
|
|
'open_count': 0,
|
|
'created_at': now - timedelta(days=5)
|
|
})
|
|
|
|
# Test classifications
|
|
active_status = get_subscriber_activity_status('test-analytics-active@example.com')
|
|
is_active = active_status == 'active'
|
|
test_result("Classify active user", is_active, f"Status: {active_status}")
|
|
|
|
inactive_status = get_subscriber_activity_status('test-analytics-inactive@example.com')
|
|
is_inactive = inactive_status == 'inactive'
|
|
test_result("Classify inactive user", is_inactive, f"Status: {inactive_status}")
|
|
|
|
dormant_status = get_subscriber_activity_status('test-analytics-dormant@example.com')
|
|
is_dormant = dormant_status == 'dormant'
|
|
test_result("Classify dormant user", is_dormant, f"Status: {dormant_status}")
|
|
|
|
new_status = get_subscriber_activity_status('test-analytics-new@example.com')
|
|
is_new = new_status == 'new'
|
|
test_result("Classify new user", is_new, f"Status: {new_status}")
|
|
|
|
except Exception as e:
|
|
test_result("Activity status classification", False, f"Error: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
|
|
# Test 6: Batch Update Activity Statuses
|
|
print("\n" + "-"*80)
|
|
print("Test 6: Batch Update Activity Statuses")
|
|
print("-"*80)
|
|
|
|
try:
|
|
updated_count = update_subscriber_activity_statuses()
|
|
|
|
# Should update all test subscribers
|
|
has_updates = updated_count > 0
|
|
test_result("Updates subscriber records", has_updates,
|
|
f"Updated {updated_count} subscribers")
|
|
|
|
# Verify a record was created
|
|
activity_record = subscriber_activity_collection.find_one({
|
|
'email': 'test-analytics-active@example.com'
|
|
})
|
|
|
|
record_exists = activity_record is not None
|
|
test_result("Creates activity record", record_exists)
|
|
|
|
if activity_record:
|
|
has_required_fields = all(key in activity_record for key in [
|
|
'email', 'status', 'total_opens', 'total_clicks',
|
|
'newsletters_received', 'newsletters_opened', 'updated_at'
|
|
])
|
|
test_result("Activity record has required fields", has_required_fields)
|
|
|
|
correct_status = activity_record['status'] == 'active'
|
|
test_result("Activity record has correct status", correct_status,
|
|
f"Status: {activity_record['status']}")
|
|
|
|
except Exception as e:
|
|
test_result("Batch update activity statuses", False, f"Error: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
|
|
# Test 7: Analytics API Endpoints
|
|
print("\n" + "-"*80)
|
|
print("Test 7: Analytics API Endpoints")
|
|
print("-"*80)
|
|
|
|
try:
|
|
with app.test_client() as client:
|
|
# Test newsletter analytics endpoint
|
|
response = client.get(f'/api/analytics/newsletter/{test_newsletter_id}')
|
|
|
|
is_200 = response.status_code == 200
|
|
test_result("Newsletter endpoint returns 200", is_200, f"Status: {response.status_code}")
|
|
|
|
if is_200:
|
|
data = response.get_json()
|
|
has_data = data is not None and 'open_rate' in data
|
|
test_result("Newsletter endpoint returns data", has_data)
|
|
|
|
# Test article analytics endpoint
|
|
response = client.get(f'/api/analytics/article/{test_article_url}')
|
|
|
|
is_200 = response.status_code == 200
|
|
test_result("Article endpoint returns 200", is_200, f"Status: {response.status_code}")
|
|
|
|
if is_200:
|
|
data = response.get_json()
|
|
has_data = data is not None and 'click_rate' in data
|
|
test_result("Article endpoint returns data", has_data)
|
|
|
|
# Test subscriber analytics endpoint
|
|
response = client.get('/api/analytics/subscriber/test-analytics-active@example.com')
|
|
|
|
is_200 = response.status_code == 200
|
|
test_result("Subscriber endpoint returns 200", is_200, f"Status: {response.status_code}")
|
|
|
|
if is_200:
|
|
data = response.get_json()
|
|
has_data = data is not None and 'status' in data
|
|
test_result("Subscriber endpoint returns data", has_data)
|
|
|
|
# Test update activity endpoint
|
|
response = client.post('/api/analytics/update-activity')
|
|
|
|
is_200 = response.status_code == 200
|
|
test_result("Update activity endpoint returns 200", is_200, f"Status: {response.status_code}")
|
|
|
|
if is_200:
|
|
data = response.get_json()
|
|
has_count = data is not None and 'updated_count' in data
|
|
test_result("Update activity endpoint returns count", has_count)
|
|
|
|
except Exception as e:
|
|
test_result("Analytics API endpoints", False, f"Error: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
|
|
# Clean up test data
|
|
print("\n" + "-"*80)
|
|
print("Cleaning up test data...")
|
|
print("-"*80)
|
|
|
|
try:
|
|
newsletter_sends_collection.delete_many({'newsletter_id': {'$regex': '^test-analytics-'}})
|
|
link_clicks_collection.delete_many({'newsletter_id': {'$regex': '^test-analytics-'}})
|
|
subscriber_activity_collection.delete_many({'email': {'$regex': '^test-analytics-'}})
|
|
print("✓ Test data cleaned up")
|
|
except Exception as e:
|
|
print(f"⚠ Error cleaning up: {str(e)}")
|
|
|
|
|
|
# Summary
|
|
print("\n" + "="*80)
|
|
print("TEST SUMMARY")
|
|
print("="*80)
|
|
print(f"Total tests: {tests_passed + tests_failed}")
|
|
print(f"✓ Passed: {tests_passed}")
|
|
print(f"❌ Failed: {tests_failed}")
|
|
|
|
if tests_failed == 0:
|
|
print("\n🎉 All tests passed!")
|
|
else:
|
|
print(f"\n⚠ {tests_failed} test(s) failed")
|
|
|
|
print("="*80 + "\n")
|
|
|
|
# Exit with appropriate code
|
|
sys.exit(0 if tests_failed == 0 else 1)
|