259f2c90d0
- 修复隐式 Optional 类型注解 (RUF013) - 修复不必要的赋值后返回 (RET504) - 优化列表推导式 (PERF401) - 修复未使用的参数 (ARG002) - 清理重复导入 - 优化异常处理
703 lines
27 KiB
Python
703 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
InsightFlow Phase 8 Task 6: Developer Ecosystem Test Script
|
|
开发者生态系统测试脚本
|
|
|
|
测试功能：
|
|
1. SDK 发布与管理
|
|
2. 模板市场
|
|
3. 插件市场
|
|
4. 开发者文档与示例代码
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
from developer_ecosystem_manager import (
|
|
DeveloperEcosystemManager,
|
|
DeveloperStatus,
|
|
PluginCategory,
|
|
PluginStatus,
|
|
SDKLanguage,
|
|
TemplateCategory,
|
|
)
|
|
|
|
# Add backend directory to path
|
|
backend_dir = os.path.dirname(os.path.abspath(__file__))
|
|
if backend_dir not in sys.path:
|
|
sys.path.insert(0, backend_dir)
|
|
|
|
|
|
class TestDeveloperEcosystem:
|
|
"""开发者生态系统测试类"""
|
|
|
|
def __init__(self) -> None:
|
|
self.manager = DeveloperEcosystemManager()
|
|
self.test_results = []
|
|
self.created_ids = {
|
|
"sdk": [],
|
|
"template": [],
|
|
"plugin": [],
|
|
"developer": [],
|
|
"code_example": [],
|
|
"portal_config": [],
|
|
}
|
|
|
|
def log(self, message: str, success: bool = True) -> None:
|
|
"""记录测试结果"""
|
|
status = "✅" if success else "❌"
|
|
print(f"{status} {message}")
|
|
self.test_results.append(
|
|
{"message": message, "success": success, "timestamp": datetime.now().isoformat()},
|
|
)
|
|
|
|
def run_all_tests(self) -> None:
|
|
"""运行所有测试"""
|
|
print(" = " * 60)
|
|
print("InsightFlow Phase 8 Task 6: Developer Ecosystem Tests")
|
|
print(" = " * 60)
|
|
|
|
# SDK Tests
|
|
print("\n📦 SDK Release & Management Tests")
|
|
print("-" * 40)
|
|
self.test_sdk_create()
|
|
self.test_sdk_list()
|
|
self.test_sdk_get()
|
|
self.test_sdk_update()
|
|
self.test_sdk_publish()
|
|
self.test_sdk_version_add()
|
|
|
|
# Template Market Tests
|
|
print("\n📋 Template Market Tests")
|
|
print("-" * 40)
|
|
self.test_template_create()
|
|
self.test_template_list()
|
|
self.test_template_get()
|
|
self.test_template_approve()
|
|
self.test_template_publish()
|
|
self.test_template_review()
|
|
|
|
# Plugin Market Tests
|
|
print("\n🔌 Plugin Market Tests")
|
|
print("-" * 40)
|
|
self.test_plugin_create()
|
|
self.test_plugin_list()
|
|
self.test_plugin_get()
|
|
self.test_plugin_review()
|
|
self.test_plugin_publish()
|
|
self.test_plugin_review_add()
|
|
|
|
# Developer Profile Tests
|
|
print("\n👤 Developer Profile Tests")
|
|
print("-" * 40)
|
|
self.test_developer_profile_create()
|
|
self.test_developer_profile_get()
|
|
self.test_developer_verify()
|
|
self.test_developer_stats_update()
|
|
|
|
# Code Examples Tests
|
|
print("\n💻 Code Examples Tests")
|
|
print("-" * 40)
|
|
self.test_code_example_create()
|
|
self.test_code_example_list()
|
|
self.test_code_example_get()
|
|
|
|
# Portal Config Tests
|
|
print("\n🌐 Developer Portal Tests")
|
|
print("-" * 40)
|
|
self.test_portal_config_create()
|
|
self.test_portal_config_get()
|
|
|
|
# Revenue Tests
|
|
print("\n💰 Developer Revenue Tests")
|
|
print("-" * 40)
|
|
self.test_revenue_record()
|
|
self.test_revenue_summary()
|
|
|
|
# Print Summary
|
|
self.print_summary()
|
|
|
|
def test_sdk_create(self) -> None:
|
|
"""测试创建 SDK"""
|
|
try:
|
|
sdk = self.manager.create_sdk_release(
|
|
name="InsightFlow Python SDK",
|
|
language=SDKLanguage.PYTHON,
|
|
version="1.0.0",
|
|
description="Python SDK for InsightFlow API",
|
|
changelog="Initial release",
|
|
download_url="https://pypi.org/insightflow/1.0.0",
|
|
documentation_url="https://docs.insightflow.io/python",
|
|
repository_url="https://github.com/insightflow/python-sdk",
|
|
package_name="insightflow",
|
|
min_platform_version="1.0.0",
|
|
dependencies=[{"name": "requests", "version": ">= 2.0"}],
|
|
file_size=1024000,
|
|
checksum="abc123",
|
|
created_by="test_user",
|
|
)
|
|
self.created_ids["sdk"].append(sdk.id)
|
|
self.log(f"Created SDK: {sdk.name} ({sdk.id})")
|
|
|
|
# Create JavaScript SDK
|
|
sdk_js = self.manager.create_sdk_release(
|
|
name="InsightFlow JavaScript SDK",
|
|
language=SDKLanguage.JAVASCRIPT,
|
|
version="1.0.0",
|
|
description="JavaScript SDK for InsightFlow API",
|
|
changelog="Initial release",
|
|
download_url="https://npmjs.com/insightflow/1.0.0",
|
|
documentation_url="https://docs.insightflow.io/js",
|
|
repository_url="https://github.com/insightflow/js-sdk",
|
|
package_name="@insightflow/sdk",
|
|
min_platform_version="1.0.0",
|
|
dependencies=[{"name": "axios", "version": ">= 0.21"}],
|
|
file_size=512000,
|
|
checksum="def456",
|
|
created_by="test_user",
|
|
)
|
|
self.created_ids["sdk"].append(sdk_js.id)
|
|
self.log(f"Created SDK: {sdk_js.name} ({sdk_js.id})")
|
|
|
|
except Exception as e:
|
|
self.log(f"Failed to create SDK: {e!s}", success=False)
|
|
|
|
def test_sdk_list(self) -> None:
|
|
"""测试列出 SDK"""
|
|
try:
|
|
sdks = self.manager.list_sdk_releases()
|
|
self.log(f"Listed {len(sdks)} SDKs")
|
|
|
|
# Test filter by language
|
|
python_sdks = self.manager.list_sdk_releases(language=SDKLanguage.PYTHON)
|
|
self.log(f"Found {len(python_sdks)} Python SDKs")
|
|
|
|
# Test search
|
|
search_results = self.manager.list_sdk_releases(search="Python")
|
|
self.log(f"Search found {len(search_results)} SDKs")
|
|
|
|
except Exception as e:
|
|
self.log(f"Failed to list SDKs: {e!s}", success=False)
|
|
|
|
def test_sdk_get(self) -> None:
|
|
"""测试获取 SDK 详情"""
|
|
try:
|
|
if self.created_ids["sdk"]:
|
|
sdk = self.manager.get_sdk_release(self.created_ids["sdk"][0])
|
|
if sdk:
|
|
self.log(f"Retrieved SDK: {sdk.name}")
|
|
else:
|
|
self.log("SDK not found", success=False)
|
|
except Exception as e:
|
|
self.log(f"Failed to get SDK: {e!s}", success=False)
|
|
|
|
def test_sdk_update(self) -> None:
|
|
"""测试更新 SDK"""
|
|
try:
|
|
if self.created_ids["sdk"]:
|
|
sdk = self.manager.update_sdk_release(
|
|
self.created_ids["sdk"][0],
|
|
description="Updated description",
|
|
)
|
|
if sdk:
|
|
self.log(f"Updated SDK: {sdk.name}")
|
|
except Exception as e:
|
|
self.log(f"Failed to update SDK: {e!s}", success=False)
|
|
|
|
def test_sdk_publish(self) -> None:
|
|
"""测试发布 SDK"""
|
|
try:
|
|
if self.created_ids["sdk"]:
|
|
sdk = self.manager.publish_sdk_release(self.created_ids["sdk"][0])
|
|
if sdk:
|
|
self.log(f"Published SDK: {sdk.name} (status: {sdk.status.value})")
|
|
except Exception as e:
|
|
self.log(f"Failed to publish SDK: {e!s}", success=False)
|
|
|
|
def test_sdk_version_add(self) -> None:
|
|
"""测试添加 SDK 版本"""
|
|
try:
|
|
if self.created_ids["sdk"]:
|
|
version = self.manager.add_sdk_version(
|
|
sdk_id=self.created_ids["sdk"][0],
|
|
version="1.1.0",
|
|
is_lts=True,
|
|
release_notes="Bug fixes and improvements",
|
|
download_url="https://pypi.org/insightflow/1.1.0",
|
|
checksum="xyz789",
|
|
file_size=1100000,
|
|
)
|
|
self.log(f"Added SDK version: {version.version}")
|
|
except Exception as e:
|
|
self.log(f"Failed to add SDK version: {e!s}", success=False)
|
|
|
|
def test_template_create(self) -> None:
|
|
"""测试创建模板"""
|
|
try:
|
|
template = self.manager.create_template(
|
|
name="医疗行业实体识别模板",
|
|
description="专门针对医疗行业的实体识别模板，支持疾病、药物、症状等实体",
|
|
category=TemplateCategory.MEDICAL,
|
|
subcategory="entity_recognition",
|
|
tags=["medical", "healthcare", "ner"],
|
|
author_id="dev_001",
|
|
author_name="Medical AI Lab",
|
|
price=99.0,
|
|
currency="CNY",
|
|
preview_image_url="https://cdn.insightflow.io/templates/medical.png",
|
|
demo_url="https://demo.insightflow.io/medical",
|
|
documentation_url="https://docs.insightflow.io/templates/medical",
|
|
download_url="https://cdn.insightflow.io/templates/medical.zip",
|
|
version="1.0.0",
|
|
min_platform_version="2.0.0",
|
|
file_size=5242880,
|
|
checksum="tpl123",
|
|
)
|
|
self.created_ids["template"].append(template.id)
|
|
self.log(f"Created template: {template.name} ({template.id})")
|
|
|
|
# Create free template
|
|
template_free = self.manager.create_template(
|
|
name="通用实体识别模板",
|
|
description="适用于一般场景的实体识别模板",
|
|
category=TemplateCategory.GENERAL,
|
|
subcategory=None,
|
|
tags=["general", "ner", "basic"],
|
|
author_id="dev_002",
|
|
author_name="InsightFlow Team",
|
|
price=0.0,
|
|
currency="CNY",
|
|
)
|
|
self.created_ids["template"].append(template_free.id)
|
|
self.log(f"Created free template: {template_free.name}")
|
|
|
|
except Exception as e:
|
|
self.log(f"Failed to create template: {e!s}", success=False)
|
|
|
|
def test_template_list(self) -> None:
|
|
"""测试列出模板"""
|
|
try:
|
|
templates = self.manager.list_templates()
|
|
self.log(f"Listed {len(templates)} templates")
|
|
|
|
# Filter by category
|
|
medical_templates = self.manager.list_templates(category=TemplateCategory.MEDICAL)
|
|
self.log(f"Found {len(medical_templates)} medical templates")
|
|
|
|
# Filter by price
|
|
free_templates = self.manager.list_templates(max_price=0)
|
|
self.log(f"Found {len(free_templates)} free templates")
|
|
|
|
except Exception as e:
|
|
self.log(f"Failed to list templates: {e!s}", success=False)
|
|
|
|
def test_template_get(self) -> None:
|
|
"""测试获取模板详情"""
|
|
try:
|
|
if self.created_ids["template"]:
|
|
template = self.manager.get_template(self.created_ids["template"][0])
|
|
if template:
|
|
self.log(f"Retrieved template: {template.name}")
|
|
except Exception as e:
|
|
self.log(f"Failed to get template: {e!s}", success=False)
|
|
|
|
def test_template_approve(self) -> None:
|
|
"""测试审核通过模板"""
|
|
try:
|
|
if self.created_ids["template"]:
|
|
template = self.manager.approve_template(
|
|
self.created_ids["template"][0],
|
|
reviewed_by="admin_001",
|
|
)
|
|
if template:
|
|
self.log(f"Approved template: {template.name}")
|
|
except Exception as e:
|
|
self.log(f"Failed to approve template: {e!s}", success=False)
|
|
|
|
def test_template_publish(self) -> None:
|
|
"""测试发布模板"""
|
|
try:
|
|
if self.created_ids["template"]:
|
|
template = self.manager.publish_template(self.created_ids["template"][0])
|
|
if template:
|
|
self.log(f"Published template: {template.name}")
|
|
except Exception as e:
|
|
self.log(f"Failed to publish template: {e!s}", success=False)
|
|
|
|
def test_template_review(self) -> None:
|
|
"""测试添加模板评价"""
|
|
try:
|
|
if self.created_ids["template"]:
|
|
review = self.manager.add_template_review(
|
|
template_id=self.created_ids["template"][0],
|
|
user_id="user_001",
|
|
user_name="Test User",
|
|
rating=5,
|
|
comment="Great template! Very accurate for medical entities.",
|
|
is_verified_purchase=True,
|
|
)
|
|
self.log(f"Added template review: {review.rating} stars")
|
|
except Exception as e:
|
|
self.log(f"Failed to add template review: {e!s}", success=False)
|
|
|
|
def test_plugin_create(self) -> None:
|
|
"""测试创建插件"""
|
|
try:
|
|
plugin = self.manager.create_plugin(
|
|
name="飞书机器人集成插件",
|
|
description="将 InsightFlow 与飞书机器人集成，实现自动通知",
|
|
category=PluginCategory.INTEGRATION,
|
|
tags=["feishu", "bot", "integration", "notification"],
|
|
author_id="dev_003",
|
|
author_name="Integration Team",
|
|
price=49.0,
|
|
currency="CNY",
|
|
pricing_model="paid",
|
|
preview_image_url="https://cdn.insightflow.io/plugins/feishu.png",
|
|
demo_url="https://demo.insightflow.io/feishu",
|
|
documentation_url="https://docs.insightflow.io/plugins/feishu",
|
|
repository_url="https://github.com/insightflow/feishu-plugin",
|
|
download_url="https://cdn.insightflow.io/plugins/feishu.zip",
|
|
webhook_url="https://api.insightflow.io/webhooks/feishu",
|
|
permissions=["read:projects", "write:notifications"],
|
|
version="1.0.0",
|
|
min_platform_version="2.0.0",
|
|
file_size=1048576,
|
|
checksum="plg123",
|
|
)
|
|
self.created_ids["plugin"].append(plugin.id)
|
|
self.log(f"Created plugin: {plugin.name} ({plugin.id})")
|
|
|
|
# Create free plugin
|
|
plugin_free = self.manager.create_plugin(
|
|
name="数据导出插件",
|
|
description="支持多种格式的数据导出",
|
|
category=PluginCategory.ANALYSIS,
|
|
tags=["export", "data", "csv", "json"],
|
|
author_id="dev_004",
|
|
author_name="Data Team",
|
|
price=0.0,
|
|
currency="CNY",
|
|
pricing_model="free",
|
|
)
|
|
self.created_ids["plugin"].append(plugin_free.id)
|
|
self.log(f"Created free plugin: {plugin_free.name}")
|
|
|
|
except Exception as e:
|
|
self.log(f"Failed to create plugin: {e!s}", success=False)
|
|
|
|
def test_plugin_list(self) -> None:
|
|
"""测试列出插件"""
|
|
try:
|
|
plugins = self.manager.list_plugins()
|
|
self.log(f"Listed {len(plugins)} plugins")
|
|
|
|
# Filter by category
|
|
integration_plugins = self.manager.list_plugins(category=PluginCategory.INTEGRATION)
|
|
self.log(f"Found {len(integration_plugins)} integration plugins")
|
|
|
|
except Exception as e:
|
|
self.log(f"Failed to list plugins: {e!s}", success=False)
|
|
|
|
def test_plugin_get(self) -> None:
|
|
"""测试获取插件详情"""
|
|
try:
|
|
if self.created_ids["plugin"]:
|
|
plugin = self.manager.get_plugin(self.created_ids["plugin"][0])
|
|
if plugin:
|
|
self.log(f"Retrieved plugin: {plugin.name}")
|
|
except Exception as e:
|
|
self.log(f"Failed to get plugin: {e!s}", success=False)
|
|
|
|
def test_plugin_review(self) -> None:
|
|
"""测试审核插件"""
|
|
try:
|
|
if self.created_ids["plugin"]:
|
|
plugin = self.manager.review_plugin(
|
|
self.created_ids["plugin"][0],
|
|
reviewed_by="admin_001",
|
|
status=PluginStatus.APPROVED,
|
|
notes="Code review passed",
|
|
)
|
|
if plugin:
|
|
self.log(f"Reviewed plugin: {plugin.name} ({plugin.status.value})")
|
|
except Exception as e:
|
|
self.log(f"Failed to review plugin: {e!s}", success=False)
|
|
|
|
def test_plugin_publish(self) -> None:
|
|
"""测试发布插件"""
|
|
try:
|
|
if self.created_ids["plugin"]:
|
|
plugin = self.manager.publish_plugin(self.created_ids["plugin"][0])
|
|
if plugin:
|
|
self.log(f"Published plugin: {plugin.name}")
|
|
except Exception as e:
|
|
self.log(f"Failed to publish plugin: {e!s}", success=False)
|
|
|
|
def test_plugin_review_add(self) -> None:
|
|
"""测试添加插件评价"""
|
|
try:
|
|
if self.created_ids["plugin"]:
|
|
review = self.manager.add_plugin_review(
|
|
plugin_id=self.created_ids["plugin"][0],
|
|
user_id="user_002",
|
|
user_name="Plugin User",
|
|
rating=4,
|
|
comment="Works great with Feishu!",
|
|
is_verified_purchase=True,
|
|
)
|
|
self.log(f"Added plugin review: {review.rating} stars")
|
|
except Exception as e:
|
|
self.log(f"Failed to add plugin review: {e!s}", success=False)
|
|
|
|
def test_developer_profile_create(self) -> None:
|
|
"""测试创建开发者档案"""
|
|
try:
|
|
# Generate unique user IDs
|
|
unique_id = uuid.uuid4().hex[:8]
|
|
|
|
profile = self.manager.create_developer_profile(
|
|
user_id=f"user_dev_{unique_id}_001",
|
|
display_name="张三",
|
|
email=f"zhangsan_{unique_id}@example.com",
|
|
bio="专注于医疗AI和自然语言处理",
|
|
website="https://zhangsan.dev",
|
|
github_url="https://github.com/zhangsan",
|
|
avatar_url="https://cdn.example.com/avatars/zhangsan.png",
|
|
)
|
|
self.created_ids["developer"].append(profile.id)
|
|
self.log(f"Created developer profile: {profile.display_name} ({profile.id})")
|
|
|
|
# Create another developer
|
|
profile2 = self.manager.create_developer_profile(
|
|
user_id=f"user_dev_{unique_id}_002",
|
|
display_name="李四",
|
|
email=f"lisi_{unique_id}@example.com",
|
|
bio="全栈开发者，热爱开源",
|
|
)
|
|
self.created_ids["developer"].append(profile2.id)
|
|
self.log(f"Created developer profile: {profile2.display_name}")
|
|
|
|
except Exception as e:
|
|
self.log(f"Failed to create developer profile: {e!s}", success=False)
|
|
|
|
def test_developer_profile_get(self) -> None:
|
|
"""测试获取开发者档案"""
|
|
try:
|
|
if self.created_ids["developer"]:
|
|
profile = self.manager.get_developer_profile(self.created_ids["developer"][0])
|
|
if profile:
|
|
self.log(f"Retrieved developer profile: {profile.display_name}")
|
|
except Exception as e:
|
|
self.log(f"Failed to get developer profile: {e!s}", success=False)
|
|
|
|
def test_developer_verify(self) -> None:
|
|
"""测试验证开发者"""
|
|
try:
|
|
if self.created_ids["developer"]:
|
|
profile = self.manager.verify_developer(
|
|
self.created_ids["developer"][0],
|
|
DeveloperStatus.VERIFIED,
|
|
)
|
|
if profile:
|
|
self.log(f"Verified developer: {profile.display_name} ({profile.status.value})")
|
|
except Exception as e:
|
|
self.log(f"Failed to verify developer: {e!s}", success=False)
|
|
|
|
def test_developer_stats_update(self) -> None:
|
|
"""测试更新开发者统计"""
|
|
try:
|
|
if self.created_ids["developer"]:
|
|
self.manager.update_developer_stats(self.created_ids["developer"][0])
|
|
profile = self.manager.get_developer_profile(self.created_ids["developer"][0])
|
|
self.log(
|
|
f"Updated developer stats: {profile.plugin_count} plugins, {profile.template_count} templates",
|
|
)
|
|
except Exception as e:
|
|
self.log(f"Failed to update developer stats: {e!s}", success=False)
|
|
|
|
def test_code_example_create(self) -> None:
|
|
"""测试创建代码示例"""
|
|
try:
|
|
example = self.manager.create_code_example(
|
|
title="使用 Python SDK 创建项目",
|
|
description="演示如何使用 Python SDK 创建新项目",
|
|
language="python",
|
|
category="quickstart",
|
|
code="""from insightflow import Client
|
|
|
|
client = Client(api_key = "your_api_key")
|
|
project = client.projects.create(name = "My Project")
|
|
print(f"Created project: {project.id}")
|
|
""",
|
|
explanation="首先导入 Client 类，然后使用 API Key 初始化客户端，最后调用 create 方法创建项目。",
|
|
tags=["python", "quickstart", "projects"],
|
|
author_id="dev_001",
|
|
author_name="InsightFlow Team",
|
|
api_endpoints=["/api/v1/projects"],
|
|
)
|
|
self.created_ids["code_example"].append(example.id)
|
|
self.log(f"Created code example: {example.title}")
|
|
|
|
# Create JavaScript example
|
|
example_js = self.manager.create_code_example(
|
|
title="使用 JavaScript SDK 上传文件",
|
|
description="演示如何使用 JavaScript SDK 上传音频文件",
|
|
language="javascript",
|
|
category="upload",
|
|
code="""const { Client } = require('insightflow');
|
|
|
|
const client = new Client({ apiKey: 'your_api_key' });
|
|
const result = await client.uploads.create({
|
|
projectId: 'proj_123',
|
|
file: './meeting.mp3'
|
|
});
|
|
console.log('Upload complete:', result.id);
|
|
""",
|
|
explanation="使用 JavaScript SDK 上传文件到 InsightFlow",
|
|
tags=["javascript", "upload", "audio"],
|
|
author_id="dev_002",
|
|
author_name="JS Team",
|
|
)
|
|
self.created_ids["code_example"].append(example_js.id)
|
|
self.log(f"Created code example: {example_js.title}")
|
|
|
|
except Exception as e:
|
|
self.log(f"Failed to create code example: {e!s}", success=False)
|
|
|
|
def test_code_example_list(self) -> None:
|
|
"""测试列出代码示例"""
|
|
try:
|
|
examples = self.manager.list_code_examples()
|
|
self.log(f"Listed {len(examples)} code examples")
|
|
|
|
# Filter by language
|
|
python_examples = self.manager.list_code_examples(language="python")
|
|
self.log(f"Found {len(python_examples)} Python examples")
|
|
|
|
except Exception as e:
|
|
self.log(f"Failed to list code examples: {e!s}", success=False)
|
|
|
|
def test_code_example_get(self) -> None:
|
|
"""测试获取代码示例详情"""
|
|
try:
|
|
if self.created_ids["code_example"]:
|
|
example = self.manager.get_code_example(self.created_ids["code_example"][0])
|
|
if example:
|
|
self.log(
|
|
f"Retrieved code example: {example.title} (views: {example.view_count})",
|
|
)
|
|
except Exception as e:
|
|
self.log(f"Failed to get code example: {e!s}", success=False)
|
|
|
|
def test_portal_config_create(self) -> None:
|
|
"""测试创建开发者门户配置"""
|
|
try:
|
|
config = self.manager.create_portal_config(
|
|
name="InsightFlow Developer Portal",
|
|
description="开发者门户 - SDK、API 文档和示例代码",
|
|
theme="default",
|
|
primary_color="#1890ff",
|
|
secondary_color="#52c41a",
|
|
support_email="developers@insightflow.io",
|
|
support_url="https://support.insightflow.io",
|
|
github_url="https://github.com/insightflow",
|
|
discord_url="https://discord.gg/insightflow",
|
|
api_base_url="https://api.insightflow.io/v1",
|
|
)
|
|
self.created_ids["portal_config"].append(config.id)
|
|
self.log(f"Created portal config: {config.name}")
|
|
|
|
except Exception as e:
|
|
self.log(f"Failed to create portal config: {e!s}", success=False)
|
|
|
|
def test_portal_config_get(self) -> None:
|
|
"""测试获取开发者门户配置"""
|
|
try:
|
|
if self.created_ids["portal_config"]:
|
|
config = self.manager.get_portal_config(self.created_ids["portal_config"][0])
|
|
if config:
|
|
self.log(f"Retrieved portal config: {config.name}")
|
|
|
|
# Test active config
|
|
active_config = self.manager.get_active_portal_config()
|
|
if active_config:
|
|
self.log(f"Active portal config: {active_config.name}")
|
|
|
|
except Exception as e:
|
|
self.log(f"Failed to get portal config: {e!s}", success=False)
|
|
|
|
def test_revenue_record(self) -> None:
|
|
"""测试记录开发者收益"""
|
|
try:
|
|
if self.created_ids["developer"] and self.created_ids["plugin"]:
|
|
revenue = self.manager.record_revenue(
|
|
developer_id=self.created_ids["developer"][0],
|
|
item_type="plugin",
|
|
item_id=self.created_ids["plugin"][0],
|
|
item_name="飞书机器人集成插件",
|
|
sale_amount=49.0,
|
|
currency="CNY",
|
|
buyer_id="user_buyer_001",
|
|
transaction_id="txn_123456",
|
|
)
|
|
self.log(f"Recorded revenue: {revenue.sale_amount} {revenue.currency}")
|
|
self.log(f" - Platform fee: {revenue.platform_fee}")
|
|
self.log(f" - Developer earnings: {revenue.developer_earnings}")
|
|
except Exception as e:
|
|
self.log(f"Failed to record revenue: {e!s}", success=False)
|
|
|
|
def test_revenue_summary(self) -> None:
|
|
"""测试获取开发者收益汇总"""
|
|
try:
|
|
if self.created_ids["developer"]:
|
|
summary = self.manager.get_developer_revenue_summary(
|
|
self.created_ids["developer"][0],
|
|
)
|
|
self.log("Revenue summary for developer:")
|
|
self.log(f" - Total sales: {summary['total_sales']}")
|
|
self.log(f" - Total fees: {summary['total_fees']}")
|
|
self.log(f" - Total earnings: {summary['total_earnings']}")
|
|
self.log(f" - Transaction count: {summary['transaction_count']}")
|
|
except Exception as e:
|
|
self.log(f"Failed to get revenue summary: {e!s}", success=False)
|
|
|
|
def print_summary(self) -> None:
|
|
"""打印测试摘要"""
|
|
print("\n" + " = " * 60)
|
|
print("Test Summary")
|
|
print(" = " * 60)
|
|
|
|
total = len(self.test_results)
|
|
passed = sum(1 for r in self.test_results if r["success"])
|
|
failed = total - passed
|
|
|
|
print(f"Total tests: {total}")
|
|
print(f"Passed: {passed} ✅")
|
|
print(f"Failed: {failed} ❌")
|
|
|
|
if failed > 0:
|
|
print("\nFailed tests:")
|
|
for r in self.test_results:
|
|
if not r["success"]:
|
|
print(f" - {r['message']}")
|
|
|
|
print("\nCreated resources:")
|
|
for resource_type, ids in self.created_ids.items():
|
|
if ids:
|
|
print(f" {resource_type}: {len(ids)}")
|
|
|
|
print(" = " * 60)
|
|
|
|
|
|
def main() -> None:
|
|
"""主函数"""
|
|
test = TestDeveloperEcosystem()
|
|
test.run_all_tests()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|