forked from Lesin/reposync
243 lines
9.2 KiB
Python
243 lines
9.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
RepoSyncer 插件系统测试脚本
|
|
演示如何使用代码质量检测和安全扫描插件
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
import json
|
|
from pathlib import Path
|
|
|
|
# 添加项目根目录到Python路径
|
|
project_root = Path(__file__).parent
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
from src.plugins.plugin_manager import plugin_manager, PluginConfig
|
|
from src.plugins.code_quality_guard import CodeQualityGuard
|
|
from src.plugins.security_scanner import SecurityScanner
|
|
|
|
|
|
async def test_plugin_system():
|
|
"""测试插件系统"""
|
|
print("=" * 60)
|
|
print("RepoSyncer 插件系统测试")
|
|
print("=" * 60)
|
|
|
|
# 1. 初始化插件管理器
|
|
print("\n1. 初始化插件管理器...")
|
|
|
|
# 注册代码质量检测插件
|
|
quality_config = PluginConfig(
|
|
name="CodeQualityGuard",
|
|
version="1.0.0",
|
|
description="智能代码质量检测与自动修复插件",
|
|
enabled=True
|
|
)
|
|
quality_plugin = CodeQualityGuard(quality_config)
|
|
plugin_manager.register_plugin(quality_plugin)
|
|
|
|
# 注册安全扫描插件
|
|
security_config = PluginConfig(
|
|
name="SecurityScanner",
|
|
version="1.0.0",
|
|
description="代码安全漏洞扫描插件",
|
|
enabled=True
|
|
)
|
|
security_plugin = SecurityScanner(security_config)
|
|
plugin_manager.register_plugin(security_plugin)
|
|
|
|
print(f"✓ 已注册 {len(plugin_manager.get_all_plugins())} 个插件")
|
|
|
|
# 2. 显示插件信息
|
|
print("\n2. 插件信息:")
|
|
for name, plugin in plugin_manager.get_all_plugins().items():
|
|
print(f" - {name} v{plugin.version}: {plugin.description}")
|
|
print(f" 支持语言: {', '.join(plugin.get_supported_languages())}")
|
|
print(f" 状态: {'启用' if plugin.enabled else '禁用'}")
|
|
|
|
# 3. 测试代码质量检测
|
|
print("\n3. 测试代码质量检测...")
|
|
|
|
# 使用当前项目目录作为测试目标
|
|
test_repo_path = project_root
|
|
|
|
if os.path.exists(test_repo_path):
|
|
print(f" 测试目录: {test_repo_path}")
|
|
|
|
# 执行代码质量检测
|
|
quality_context = {"repo_path": str(test_repo_path)}
|
|
quality_result = await plugin_manager.execute_plugin("CodeQualityGuard", quality_context)
|
|
|
|
if quality_result.get("success"):
|
|
report = quality_result.get("report", {})
|
|
print(f" ✓ 代码质量检测完成")
|
|
print(f" 检查文件数: {report.get('total_files', 0)}")
|
|
print(f" 发现问题数: {report.get('total_issues', 0)}")
|
|
print(f" 质量评分: {report.get('quality_score', 0):.1f}/100")
|
|
print(f" 摘要: {report.get('summary', '')}")
|
|
|
|
# 显示修复建议
|
|
fix_suggestions = quality_result.get("fix_suggestions", {})
|
|
if fix_suggestions.get("priority_fixes"):
|
|
print(f" 优先修复问题: {len(fix_suggestions['priority_fixes'])} 个")
|
|
if fix_suggestions.get("security_fixes"):
|
|
print(f" 安全修复问题: {len(fix_suggestions['security_fixes'])} 个")
|
|
else:
|
|
print(f" ✗ 代码质量检测失败: {quality_result.get('error', '未知错误')}")
|
|
else:
|
|
print(f" ✗ 测试目录不存在: {test_repo_path}")
|
|
|
|
# 4. 测试安全扫描
|
|
print("\n4. 测试安全扫描...")
|
|
|
|
if os.path.exists(test_repo_path):
|
|
# 执行安全扫描
|
|
security_context = {"repo_path": str(test_repo_path)}
|
|
security_result = await plugin_manager.execute_plugin("SecurityScanner", security_context)
|
|
|
|
if security_result.get("success"):
|
|
scan_results = security_result.get("scan_results", [])
|
|
security_report = security_result.get("security_report", {})
|
|
|
|
print(f" ✓ 安全扫描完成")
|
|
print(f" 发现漏洞数: {len(scan_results)}")
|
|
print(f" 安全评分: {security_report.get('security_score', 0):.1f}/100")
|
|
print(f" 摘要: {security_report.get('summary', '')}")
|
|
|
|
# 显示风险等级统计
|
|
risk_summary = security_report.get("risk_summary", {})
|
|
if risk_summary.get("high", 0) > 0:
|
|
print(f" 高危漏洞: {risk_summary['high']} 个")
|
|
if risk_summary.get("medium", 0) > 0:
|
|
print(f" 中危漏洞: {risk_summary['medium']} 个")
|
|
if risk_summary.get("low", 0) > 0:
|
|
print(f" 低危漏洞: {risk_summary['low']} 个")
|
|
else:
|
|
print(f" ✗ 安全扫描失败: {security_result.get('error', '未知错误')}")
|
|
|
|
# 5. 测试批量执行
|
|
print("\n5. 测试批量执行插件...")
|
|
|
|
if os.path.exists(test_repo_path):
|
|
# 执行所有启用的插件
|
|
batch_context = {"repo_path": str(test_repo_path)}
|
|
batch_results = await plugin_manager.execute_all_plugins(batch_context)
|
|
|
|
print(f" ✓ 批量执行完成，执行了 {len(batch_results)} 个插件")
|
|
|
|
for plugin_name, result in batch_results.items():
|
|
success = result.get("success", False)
|
|
status = "✓" if success else "✗"
|
|
print(f" {status} {plugin_name}: {'成功' if success else '失败'}")
|
|
|
|
# 6. 显示执行历史
|
|
print("\n6. 插件执行历史:")
|
|
history = plugin_manager.get_execution_history(10)
|
|
for record in history:
|
|
plugin_name = record.get("plugin_name", "未知")
|
|
duration = record.get("duration", 0)
|
|
success = record.get("success", False)
|
|
status = "✓" if success else "✗"
|
|
print(f" {status} {plugin_name}: {duration:.2f}s")
|
|
|
|
# 7. 导出执行报告
|
|
print("\n7. 导出执行报告...")
|
|
report_file = "plugin_execution_report.json"
|
|
if plugin_manager.export_execution_report(report_file):
|
|
print(f" ✓ 执行报告已导出到: {report_file}")
|
|
else:
|
|
print(f" ✗ 导出执行报告失败")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("插件系统测试完成")
|
|
print("=" * 60)
|
|
|
|
|
|
async def test_specific_plugin():
|
|
"""测试特定插件功能"""
|
|
print("\n" + "=" * 60)
|
|
print("特定插件功能测试")
|
|
print("=" * 60)
|
|
|
|
# 创建测试代码文件
|
|
test_code = '''
|
|
import os
|
|
import sys
|
|
|
|
def insecure_function(password="123456"):
|
|
query = "SELECT * FROM users WHERE password = '" + password + "'"
|
|
cursor.execute(query)
|
|
|
|
def another_function():
|
|
print("This is a test function")
|
|
eval("print('dangerous')")
|
|
|
|
class TestClass:
|
|
def __init__(self):
|
|
self.api_key = "sk-1234567890abcdef"
|
|
|
|
def long_method(self):
|
|
# This is a very long method with many lines
|
|
for i in range(100):
|
|
print(i)
|
|
if i > 50:
|
|
print("Still going...")
|
|
if i > 75:
|
|
print("Almost done...")
|
|
if i > 90:
|
|
print("Nearly there...")
|
|
if i > 95:
|
|
print("Final stretch...")
|
|
if i > 98:
|
|
print("Done!")
|
|
'''
|
|
|
|
# 创建临时测试文件
|
|
test_file = "test_code.py"
|
|
with open(test_file, "w", encoding="utf-8") as f:
|
|
f.write(test_code)
|
|
|
|
try:
|
|
# 测试代码质量检测
|
|
print("\n测试代码质量检测...")
|
|
quality_context = {"repo_path": "."}
|
|
quality_result = await plugin_manager.execute_plugin("CodeQualityGuard", quality_context)
|
|
|
|
if quality_result.get("success"):
|
|
report = quality_result.get("report", {})
|
|
print(f" 发现 {report.get('total_issues', 0)} 个代码质量问题")
|
|
|
|
# 显示具体问题
|
|
issues_by_file = report.get("issues_by_file", {})
|
|
for file_path, issues in issues_by_file.items():
|
|
if "test_code.py" in file_path:
|
|
print(f" 文件 {file_path} 的问题:")
|
|
for issue in issues[:5]: # 只显示前5个问题
|
|
print(f" - 第{issue.line_number}行: {issue.message}")
|
|
|
|
# 测试安全扫描
|
|
print("\n测试安全扫描...")
|
|
security_context = {"repo_path": "."}
|
|
security_result = await plugin_manager.execute_plugin("SecurityScanner", security_context)
|
|
|
|
if security_result.get("success"):
|
|
scan_results = security_result.get("scan_results", [])
|
|
print(f" 发现 {len(scan_results)} 个安全漏洞")
|
|
|
|
# 显示具体漏洞
|
|
for vuln in scan_results[:5]: # 只显示前5个漏洞
|
|
print(f" - {vuln['file_path']}:{vuln['line_number']} - {vuln['vulnerability_type']} ({vuln['risk_level']})")
|
|
print(f" 描述: {vuln['description']}")
|
|
|
|
finally:
|
|
# 清理测试文件
|
|
if os.path.exists(test_file):
|
|
os.remove(test_file)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 运行测试
|
|
asyncio.run(test_plugin_system())
|
|
asyncio.run(test_specific_plugin()) |