claw/insightflow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: auto-fix code issues (cron)

Browse Source 
- 修复重复导入/字段
- 修复异常处理
- 修复PEP8格式问题
- 添加类型注解
- 修复重复函数定义 (health_check, create_webhook_endpoint, etc)
- 修复未定义名称 (SearchOperator, TenantTier, Query, Body, logger)
- 修复 workflow_manager.py 的类定义重复问题
- 添加缺失的导入
This commit is contained in:
OpenClaw Bot
2026-02-27 09:18:58 +08:00
parent 1d55ae8f1e
commit be22b763fa
39 changed files with 12535 additions and 10327 deletions
Download Patch File Download Diff File Expand all files Collapse all files

260
backend/test_phase8_task5.py
View File

@@ -13,6 +13,9 @@ InsightFlow Phase 8 Task 5 - 运营与增长工具测试脚本
python test_phase8_task5.py
"""
from growth_manager import (
GrowthManager, EventType, ExperimentStatus, TrafficAllocationType, EmailTemplateType, WorkflowTriggerType
)
import asyncio
import sys
import os
@@ -23,35 +26,28 @@ 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,
@@ -64,21 +60,21 @@ class TestGrowthManager:
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}),
@@ -86,7 +82,7 @@ class TestGrowthManager:
(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,
@@ -95,57 +91,57 @@ class TestGrowthManager:
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,
@@ -159,31 +155,31 @@ class TestGrowthManager:
],
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%}")
@@ -194,33 +190,33 @@ class TestGrowthManager:
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,
@@ -241,69 +237,69 @@ class TestGrowthManager:
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 = [
@@ -313,7 +309,7 @@ class TestGrowthManager:
("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,
@@ -322,24 +318,24 @@ class TestGrowthManager:
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
@@ -349,13 +345,13 @@ class TestGrowthManager:
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,
@@ -376,37 +372,37 @@ class TestGrowthManager:
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,
@@ -415,7 +411,7 @@ class TestGrowthManager:
"dashboard_url": "https://app.insightflow.io/dashboard"
}
)
if rendered:
assert "subject" in rendered
assert "html" in rendered
@@ -427,15 +423,15 @@ class TestGrowthManager:
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,
@@ -447,20 +443,20 @@ class TestGrowthManager:
{"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,
@@ -474,22 +470,22 @@ class TestGrowthManager:
{"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,
@@ -503,34 +499,34 @@ class TestGrowthManager:
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:
@@ -539,21 +535,21 @@ class TestGrowthManager:
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
@@ -563,31 +559,31 @@ class TestGrowthManager:
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,
@@ -600,66 +596,66 @@ class TestGrowthManager:
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()
@@ -667,68 +663,68 @@ class TestGrowthManager:
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_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)