115 lines
3.1 KiB
Python
115 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
系统测试脚本
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
from pathlib import Path
|
|
|
|
# 添加项目路径
|
|
project_root = Path(__file__).parent / "gx_gp_monitor"
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
def test_imports():
|
|
"""测试模块导入"""
|
|
print("测试模块导入...")
|
|
|
|
try:
|
|
from core.config_manager import load_config, get_config
|
|
print("✓ core.config_manager 导入成功")
|
|
|
|
from core.logger import get_logger
|
|
print("✓ core.logger 导入成功")
|
|
|
|
from core.models import Announcement, AnnouncementType
|
|
print("✓ core.models 导入成功")
|
|
|
|
from core.database import get_database_manager
|
|
print("✓ core.database 导入成功")
|
|
|
|
from crawler.spider import GXGPSpider
|
|
print("✓ crawler.spider 导入成功")
|
|
|
|
from filters.filters import AnnouncementFilter
|
|
print("✓ filters.filters 导入成功")
|
|
|
|
from storage.postgresql import PostgreSQLStorage
|
|
print("✓ storage.postgresql 导入成功")
|
|
|
|
from notification.wechat import WeChatService
|
|
print("✓ notification.wechat 导入成功")
|
|
|
|
return True
|
|
except ImportError as e:
|
|
print(f"✗ 导入失败: {e}")
|
|
return False
|
|
|
|
def test_config():
|
|
"""测试配置加载"""
|
|
print("\n测试配置加载...")
|
|
|
|
try:
|
|
from core.config_manager import load_config
|
|
|
|
config = load_config()
|
|
print("✓ 配置文件加载成功")
|
|
print(f" - 调试模式: {config.debug}")
|
|
print(f" - 日志级别: {config.log_level.value}")
|
|
print(f" - 数据库启用: {config.database.enabled}")
|
|
print(f" - 企业微信启用: {config.wechat_app.enabled}")
|
|
print(f" - 公告来源数量: {len(config.sources)}")
|
|
|
|
return True
|
|
except Exception as e:
|
|
print(f"✗ 配置加载失败: {e}")
|
|
return False
|
|
|
|
def test_database_connection():
|
|
"""测试数据库连接"""
|
|
print("\n测试数据库连接...")
|
|
|
|
try:
|
|
from core.config_manager import get_config
|
|
from core.database import init_database
|
|
|
|
config = get_config()
|
|
if not config.database.enabled:
|
|
print("⚠ 数据库功能已禁用，跳过连接测试")
|
|
return True
|
|
|
|
# 尝试初始化数据库
|
|
init_database()
|
|
print("✓ 数据库连接成功")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"✗ 数据库连接失败: {e}")
|
|
return False
|
|
|
|
def main():
|
|
"""主测试函数"""
|
|
print("=== 广西政府采购网公告监控系统测试 ===")
|
|
|
|
# 测试导入
|
|
if not test_imports():
|
|
print("\n❌ 模块导入测试失败")
|
|
return False
|
|
|
|
# 测试配置
|
|
if not test_config():
|
|
print("\n❌ 配置加载测试失败")
|
|
return False
|
|
|
|
# 测试数据库
|
|
if not test_database_connection():
|
|
print("\n❌ 数据库连接测试失败")
|
|
return False
|
|
|
|
print("\n✅ 所有测试通过！系统准备就绪。")
|
|
return True
|
|
|
|
if __name__ == "__main__":
|
|
success = main()
|
|
sys.exit(0 if success else 1)
|