#!/usr/bin/env python3
"""
InsightFlow Phase 8 Task 5 - 运营与增长工具测试脚本
测试内容：
1. 用户行为分析（事件追踪、用户画像、转化漏斗、留存率）
2. A/B 测试框架（实验创建、流量分配、结果分析）
3. 邮件营销自动化（模板管理、营销活动、自动化工作流）
4. 推荐系统（推荐计划、推荐码生成、团队激励）
运行方式：
cd /root/.openclaw/workspace/projects/insightflow/backend
python test_phase8_task5.py
"""
import asyncio
import sys
import os
from datetime import datetime, timedelta
# 添加 backend 目录到路径
backend_dir = os.path.dirname(os.path.abspath(__file__))
if backend_dir not in sys.path:
sys.path.insert(0, backend_dir)
from growth_manager import (
get_growth_manager, GrowthManager, AnalyticsEvent, UserProfile, Funnel, FunnelAnalysis,
Experiment, EmailTemplate, EmailCampaign, ReferralProgram, Referral, TeamIncentive,
EventType, ExperimentStatus, TrafficAllocationType, EmailTemplateType,
EmailStatus, WorkflowTriggerType, ReferralStatus
)
class TestGrowthManager:
"""测试 Growth Manager 功能"""
def __init__(self):
self.manager = GrowthManager()
self.test_tenant_id = "test_tenant_001"
self.test_user_id = "test_user_001"
self.test_results = []
def log(self, message: str, success: bool = True):
"""记录测试结果"""
status = "✅" if success else "❌"
print(f"{status} {message}")
self.test_results.append((message, success))
# ==================== 测试用户行为分析 ====================
async def test_track_event(self):
"""测试事件追踪"""
print("\n📊 测试事件追踪...")
try:
event = await self.manager.track_event(
tenant_id=self.test_tenant_id,
user_id=self.test_user_id,
event_type=EventType.PAGE_VIEW,
event_name="dashboard_view",
properties={"page": "/dashboard", "duration": 120},
session_id="session_001",
device_info={"browser": "Chrome", "os": "MacOS"},
referrer="https://google.com",
utm_params={"source": "google", "medium": "organic", "campaign": "summer"}
)
assert event.id is not None
assert event.event_type == EventType.PAGE_VIEW
assert event.event_name == "dashboard_view"
self.log(f"事件追踪成功: {event.id}")
return True
except Exception as e:
self.log(f"事件追踪失败: {e}", success=False)
return False
async def test_track_multiple_events(self):
"""测试追踪多个事件"""
print("\n📊 测试追踪多个事件...")
try:
events = [
(EventType.FEATURE_USE, "entity_extraction", {"entity_count": 5}),
(EventType.FEATURE_USE, "relation_discovery", {"relation_count": 3}),
(EventType.CONVERSION, "upgrade_click", {"plan": "pro"}),
(EventType.SIGNUP, "user_registration", {"source": "referral"}),
]
for event_type, event_name, props in events:
await self.manager.track_event(
tenant_id=self.test_tenant_id,
user_id=self.test_user_id,
event_type=event_type,
event_name=event_name,
properties=props
)
self.log(f"成功追踪 {len(events)} 个事件")
return True
except Exception as e:
self.log(f"批量事件追踪失败: {e}", success=False)
return False
def test_get_user_profile(self):
"""测试获取用户画像"""
print("\n👤 测试用户画像...")
try:
profile = self.manager.get_user_profile(self.test_tenant_id, self.test_user_id)
if profile:
assert profile.user_id == self.test_user_id
assert profile.total_events >= 0
self.log(f"用户画像获取成功: {profile.user_id}, 事件数: {profile.total_events}")
else:
self.log("用户画像不存在（首次访问）")
return True
except Exception as e:
self.log(f"获取用户画像失败: {e}", success=False)
return False
def test_get_analytics_summary(self):
"""测试获取分析汇总"""
print("\n📈 测试分析汇总...")
try:
summary = self.manager.get_user_analytics_summary(
tenant_id=self.test_tenant_id,
start_date=datetime.now() - timedelta(days=7),
end_date=datetime.now()
)
assert "unique_users" in summary
assert "total_events" in summary
assert "event_type_distribution" in summary
self.log(f"分析汇总: {summary['unique_users']} 用户, {summary['total_events']} 事件")
return True
except Exception as e:
self.log(f"获取分析汇总失败: {e}", success=False)
return False
def test_create_funnel(self):
"""测试创建转化漏斗"""
print("\n🎯 测试创建转化漏斗...")
try:
funnel = self.manager.create_funnel(
tenant_id=self.test_tenant_id,
name="用户注册转化漏斗",
description="从访问到完成注册的转化流程",
steps=[
{"name": "访问首页", "event_name": "page_view_home"},
{"name": "点击注册", "event_name": "signup_click"},
{"name": "填写信息", "event_name": "signup_form_fill"},
{"name": "完成注册", "event_name": "signup_complete"}
],
created_by="test"
)
assert funnel.id is not None
assert len(funnel.steps) == 4
self.log(f"漏斗创建成功: {funnel.id}")
return funnel.id
except Exception as e:
self.log(f"创建漏斗失败: {e}", success=False)
return None
def test_analyze_funnel(self, funnel_id: str):
"""测试分析漏斗"""
print("\n📉 测试漏斗分析...")
if not funnel_id:
self.log("跳过漏斗分析（无漏斗ID）")
return False
try:
analysis = self.manager.analyze_funnel(
funnel_id=funnel_id,
period_start=datetime.now() - timedelta(days=30),
period_end=datetime.now()
)
if analysis:
assert "step_conversions" in analysis.__dict__
self.log(f"漏斗分析完成: 总体转化率 {analysis.overall_conversion:.2%}")
return True
else:
self.log("漏斗分析返回空结果")
return False
except Exception as e:
self.log(f"漏斗分析失败: {e}", success=False)
return False
def test_calculate_retention(self):
"""测试留存率计算"""
print("\n🔄 测试留存率计算...")
try:
retention = self.manager.calculate_retention(
tenant_id=self.test_tenant_id,
cohort_date=datetime.now() - timedelta(days=7),
periods=[1, 3, 7]
)
assert "cohort_date" in retention
assert "retention" in retention
self.log(f"留存率计算完成: 同期群 {retention['cohort_size']} 用户")
return True
except Exception as e:
self.log(f"留存率计算失败: {e}", success=False)
return False
# ==================== 测试 A/B 测试框架 ====================
def test_create_experiment(self):
"""测试创建实验"""
print("\n🧪 测试创建 A/B 测试实验...")
try:
experiment = self.manager.create_experiment(
tenant_id=self.test_tenant_id,
name="首页按钮颜色测试",
description="测试不同按钮颜色对转化率的影响",
hypothesis="蓝色按钮比红色按钮有更高的点击率",
variants=[
{"id": "control", "name": "红色按钮", "is_control": True},
{"id": "variant_a", "name": "蓝色按钮", "is_control": False},
{"id": "variant_b", "name": "绿色按钮", "is_control": False}
],
traffic_allocation=TrafficAllocationType.RANDOM,
traffic_split={"control": 0.34, "variant_a": 0.33, "variant_b": 0.33},
target_audience={"conditions": []},
primary_metric="button_click_rate",
secondary_metrics=["conversion_rate", "bounce_rate"],
min_sample_size=100,
confidence_level=0.95,
created_by="test"
)
assert experiment.id is not None
assert experiment.status == ExperimentStatus.DRAFT
self.log(f"实验创建成功: {experiment.id}")
return experiment.id
except Exception as e:
self.log(f"创建实验失败: {e}", success=False)
return None
def test_list_experiments(self):
"""测试列出实验"""
print("\n📋 测试列出实验...")
try:
experiments = self.manager.list_experiments(self.test_tenant_id)
self.log(f"列出 {len(experiments)} 个实验")
return True
except Exception as e:
self.log(f"列出实验失败: {e}", success=False)
return False
def test_assign_variant(self, experiment_id: str):
"""测试分配变体"""
print("\n🎲 测试分配实验变体...")
if not experiment_id:
self.log("跳过变体分配（无实验ID）")
return False
try:
# 先启动实验
self.manager.start_experiment(experiment_id)
# 测试多个用户的变体分配
test_users = ["user_001", "user_002", "user_003", "user_004", "user_005"]
assignments = {}
for user_id in test_users:
variant_id = self.manager.assign_variant(
experiment_id=experiment_id,
user_id=user_id,
user_attributes={"user_id": user_id, "segment": "new"}
)
if variant_id:
assignments[user_id] = variant_id
self.log(f"变体分配完成: {len(assignments)} 个用户")
return True
except Exception as e:
self.log(f"变体分配失败: {e}", success=False)
return False
def test_record_experiment_metric(self, experiment_id: str):
"""测试记录实验指标"""
print("\n📊 测试记录实验指标...")
if not experiment_id:
self.log("跳过指标记录（无实验ID）")
return False
try:
# 模拟记录一些指标
test_data = [
("user_001", "control", 1),
("user_002", "variant_a", 1),
("user_003", "variant_b", 0),
("user_004", "control", 1),
("user_005", "variant_a", 1),
]
for user_id, variant_id, value in test_data:
self.manager.record_experiment_metric(
experiment_id=experiment_id,
variant_id=variant_id,
user_id=user_id,
metric_name="button_click_rate",
metric_value=value
)
self.log(f"成功记录 {len(test_data)} 条指标")
return True
except Exception as e:
self.log(f"记录指标失败: {e}", success=False)
return False
def test_analyze_experiment(self, experiment_id: str):
"""测试分析实验结果"""
print("\n📈 测试分析实验结果...")
if not experiment_id:
self.log("跳过实验分析（无实验ID）")
return False
try:
result = self.manager.analyze_experiment(experiment_id)
if "error" not in result:
self.log(f"实验分析完成: {len(result.get('variant_results', {}))} 个变体")
return True
else:
self.log(f"实验分析返回错误: {result['error']}", success=False)
return False
except Exception as e:
self.log(f"实验分析失败: {e}", success=False)
return False
# ==================== 测试邮件营销 ====================
def test_create_email_template(self):
"""测试创建邮件模板"""
print("\n📧 测试创建邮件模板...")
try:
template = self.manager.create_email_template(
tenant_id=self.test_tenant_id,
name="欢迎邮件",
template_type=EmailTemplateType.WELCOME,
subject="欢迎加入 InsightFlow！",
html_content="""
欢迎，{{user_name}}！
感谢您注册 InsightFlow。我们很高兴您能加入我们！
您的账户已创建，可以开始使用以下功能：
立即开始使用
""",
from_name="InsightFlow 团队",
from_email="welcome@insightflow.io"
)
assert template.id is not None
assert template.template_type == EmailTemplateType.WELCOME
self.log(f"邮件模板创建成功: {template.id}")
return template.id
except Exception as e:
self.log(f"创建邮件模板失败: {e}", success=False)
return None
def test_list_email_templates(self):
"""测试列出邮件模板"""
print("\n📧 测试列出邮件模板...")
try:
templates = self.manager.list_email_templates(self.test_tenant_id)
self.log(f"列出 {len(templates)} 个邮件模板")
return True
except Exception as e:
self.log(f"列出邮件模板失败: {e}", success=False)
return False
def test_render_template(self, template_id: str):
"""测试渲染邮件模板"""
print("\n🎨 测试渲染邮件模板...")
if not template_id:
self.log("跳过模板渲染（无模板ID）")
return False
try:
rendered = self.manager.render_template(
template_id=template_id,
variables={
"user_name": "张三",
"dashboard_url": "https://app.insightflow.io/dashboard"
}
)
if rendered:
assert "subject" in rendered
assert "html" in rendered
self.log(f"模板渲染成功: {rendered['subject']}")
return True
else:
self.log("模板渲染返回空结果", success=False)
return False
except Exception as e:
self.log(f"模板渲染失败: {e}", success=False)
return False
def test_create_email_campaign(self, template_id: str):
"""测试创建邮件营销活动"""
print("\n📮 测试创建邮件营销活动...")
if not template_id:
self.log("跳过创建营销活动（无模板ID）")
return None
try:
campaign = self.manager.create_email_campaign(
tenant_id=self.test_tenant_id,
name="新用户欢迎活动",
template_id=template_id,
recipient_list=[
{"user_id": "user_001", "email": "user1@example.com"},
{"user_id": "user_002", "email": "user2@example.com"},
{"user_id": "user_003", "email": "user3@example.com"}
]
)
assert campaign.id is not None
assert campaign.recipient_count == 3
self.log(f"营销活动创建成功: {campaign.id}, {campaign.recipient_count} 收件人")
return campaign.id
except Exception as e:
self.log(f"创建营销活动失败: {e}", success=False)
return None
def test_create_automation_workflow(self):
"""测试创建自动化工作流"""
print("\n🤖 测试创建自动化工作流...")
try:
workflow = self.manager.create_automation_workflow(
tenant_id=self.test_tenant_id,
name="新用户欢迎序列",
description="用户注册后自动发送欢迎邮件序列",
trigger_type=WorkflowTriggerType.USER_SIGNUP,
trigger_conditions={"event": "user_signup"},
actions=[
{"type": "send_email", "template_type": "welcome", "delay_hours": 0},
{"type": "send_email", "template_type": "onboarding", "delay_hours": 24},
{"type": "send_email", "template_type": "feature_tips", "delay_hours": 72}
]
)
assert workflow.id is not None
assert workflow.trigger_type == WorkflowTriggerType.USER_SIGNUP
self.log(f"自动化工作流创建成功: {workflow.id}")
return True
except Exception as e:
self.log(f"创建工作流失败: {e}", success=False)
return False
# ==================== 测试推荐系统 ====================
def test_create_referral_program(self):
"""测试创建推荐计划"""
print("\n🎁 测试创建推荐计划...")
try:
program = self.manager.create_referral_program(
tenant_id=self.test_tenant_id,
name="邀请好友奖励计划",
description="邀请好友注册，双方获得积分奖励",
referrer_reward_type="credit",
referrer_reward_value=100.0,
referee_reward_type="credit",
referee_reward_value=50.0,
max_referrals_per_user=10,
referral_code_length=8,
expiry_days=30
)
assert program.id is not None
assert program.referrer_reward_value == 100.0
self.log(f"推荐计划创建成功: {program.id}")
return program.id
except Exception as e:
self.log(f"创建推荐计划失败: {e}", success=False)
return None
def test_generate_referral_code(self, program_id: str):
"""测试生成推荐码"""
print("\n🔑 测试生成推荐码...")
if not program_id:
self.log("跳过生成推荐码（无计划ID）")
return None
try:
referral = self.manager.generate_referral_code(
program_id=program_id,
referrer_id="referrer_user_001"
)
if referral:
assert referral.referral_code is not None
assert len(referral.referral_code) == 8
self.log(f"推荐码生成成功: {referral.referral_code}")
return referral.referral_code
else:
self.log("生成推荐码返回空结果", success=False)
return None
except Exception as e:
self.log(f"生成推荐码失败: {e}", success=False)
return None
def test_apply_referral_code(self, referral_code: str):
"""测试应用推荐码"""
print("\n✅ 测试应用推荐码...")
if not referral_code:
self.log("跳过应用推荐码（无推荐码）")
return False
try:
success = self.manager.apply_referral_code(
referral_code=referral_code,
referee_id="new_user_001"
)
if success:
self.log(f"推荐码应用成功: {referral_code}")
return True
else:
self.log("推荐码应用失败", success=False)
return False
except Exception as e:
self.log(f"应用推荐码失败: {e}", success=False)
return False
def test_get_referral_stats(self, program_id: str):
"""测试获取推荐统计"""
print("\n📊 测试获取推荐统计...")
if not program_id:
self.log("跳过推荐统计（无计划ID）")
return False
try:
stats = self.manager.get_referral_stats(program_id)
assert "total_referrals" in stats
assert "conversion_rate" in stats
self.log(f"推荐统计: {stats['total_referrals']} 推荐, {stats['conversion_rate']:.2%} 转化率")
return True
except Exception as e:
self.log(f"获取推荐统计失败: {e}", success=False)
return False
def test_create_team_incentive(self):
"""测试创建团队激励"""
print("\n🏆 测试创建团队升级激励...")
try:
incentive = self.manager.create_team_incentive(
tenant_id=self.test_tenant_id,
name="团队升级奖励",
description="团队规模达到5人升级到 Pro 计划可获得折扣",
target_tier="pro",
min_team_size=5,
incentive_type="discount",
incentive_value=20.0, # 20% 折扣
valid_from=datetime.now(),
valid_until=datetime.now() + timedelta(days=90)
)
assert incentive.id is not None
assert incentive.incentive_value == 20.0
self.log(f"团队激励创建成功: {incentive.id}")
return True
except Exception as e:
self.log(f"创建团队激励失败: {e}", success=False)
return False
def test_check_team_incentive_eligibility(self):
"""测试检查团队激励资格"""
print("\n🔍 测试检查团队激励资格...")
try:
incentives = self.manager.check_team_incentive_eligibility(
tenant_id=self.test_tenant_id,
current_tier="free",
team_size=5
)
self.log(f"找到 {len(incentives)} 个符合条件的激励")
return True
except Exception as e:
self.log(f"检查激励资格失败: {e}", success=False)
return False
# ==================== 测试实时仪表板 ====================
def test_get_realtime_dashboard(self):
"""测试获取实时仪表板"""
print("\n📺 测试实时分析仪表板...")
try:
dashboard = self.manager.get_realtime_dashboard(self.test_tenant_id)
assert "today" in dashboard
assert "recent_events" in dashboard
assert "top_features" in dashboard
today = dashboard["today"]
self.log(f"实时仪表板: 今日 {today['active_users']} 活跃用户, {today['total_events']} 事件")
return True
except Exception as e:
self.log(f"获取实时仪表板失败: {e}", success=False)
return False
# ==================== 运行所有测试 ====================
async def run_all_tests(self):
"""运行所有测试"""
print("=" * 60)
print("🚀 InsightFlow Phase 8 Task 5 - 运营与增长工具测试")
print("=" * 60)
# 用户行为分析测试
print("\n" + "=" * 60)
print("📊 模块 1: 用户行为分析")
print("=" * 60)
await self.test_track_event()
await self.test_track_multiple_events()
self.test_get_user_profile()
self.test_get_analytics_summary()
funnel_id = self.test_create_funnel()
self.test_analyze_funnel(funnel_id)
self.test_calculate_retention()
# A/B 测试框架测试
print("\n" + "=" * 60)
print("🧪 模块 2: A/B 测试框架")
print("=" * 60)
experiment_id = self.test_create_experiment()
self.test_list_experiments()
self.test_assign_variant(experiment_id)
self.test_record_experiment_metric(experiment_id)
self.test_analyze_experiment(experiment_id)
# 邮件营销测试
print("\n" + "=" * 60)
print("📧 模块 3: 邮件营销自动化")
print("=" * 60)
template_id = self.test_create_email_template()
self.test_list_email_templates()
self.test_render_template(template_id)
campaign_id = self.test_create_email_campaign(template_id)
self.test_create_automation_workflow()
# 推荐系统测试
print("\n" + "=" * 60)
print("🎁 模块 4: 推荐系统")
print("=" * 60)
program_id = self.test_create_referral_program()
referral_code = self.test_generate_referral_code(program_id)
self.test_apply_referral_code(referral_code)
self.test_get_referral_stats(program_id)
self.test_create_team_incentive()
self.test_check_team_incentive_eligibility()
# 实时仪表板测试
print("\n" + "=" * 60)
print("📺 模块 5: 实时分析仪表板")
print("=" * 60)
self.test_get_realtime_dashboard()
# 测试总结
print("\n" + "=" * 60)
print("📋 测试总结")
print("=" * 60)
total_tests = len(self.test_results)
passed_tests = sum(1 for _, success in self.test_results if success)
failed_tests = total_tests - passed_tests
print(f"总测试数: {total_tests}")
print(f"通过: {passed_tests} ✅")
print(f"失败: {failed_tests} ❌")
print(f"通过率: {passed_tests / total_tests * 100:.1f}%" if total_tests > 0 else "N/A")
if failed_tests > 0:
print("\n失败的测试:")
for message, success in self.test_results:
if not success:
print(f" - {message}")
print("\n" + "=" * 60)
print("✨ 测试完成!")
print("=" * 60)
async def main():
"""主函数"""
tester = TestGrowthManager()
await tester.run_all_tests()
if __name__ == "__main__":
asyncio.run(main())