498 lines
16 KiB
Python
498 lines
16 KiB
Python
"""
|
||
主程序模块单元测试
|
||
测试StockAnalysisSystem主类的功能
|
||
"""
|
||
|
||
import pytest
|
||
import asyncio
|
||
from unittest.mock import Mock, patch, AsyncMock
|
||
import sys
|
||
from io import StringIO
|
||
|
||
from src.main import StockAnalysisSystem
|
||
from src.utils.exceptions import StockSystemError
|
||
|
||
|
||
class TestStockAnalysisSystem:
|
||
"""股票分析系统测试类"""
|
||
|
||
@pytest.fixture
|
||
def stock_system(self):
|
||
"""股票分析系统实例"""
|
||
return StockAnalysisSystem()
|
||
|
||
def test_initialization(self, stock_system):
|
||
"""测试系统初始化"""
|
||
assert stock_system.data_manager is None
|
||
assert stock_system.data_processor is None
|
||
assert stock_system.stock_repo is None
|
||
assert stock_system.task_scheduler is None
|
||
assert stock_system.is_initialized is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_initialize_success(self, stock_system):
|
||
"""测试系统初始化成功"""
|
||
with patch.object(stock_system, "_setup_database") as mock_setup_db:
|
||
with patch.object(stock_system, "_setup_components") as mock_setup_comp:
|
||
result = await stock_system.initialize()
|
||
|
||
assert result is True
|
||
assert stock_system.is_initialized is True
|
||
mock_setup_db.assert_called_once()
|
||
mock_setup_comp.assert_called_once()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_initialize_already_initialized(self, stock_system):
|
||
"""测试重复初始化"""
|
||
# 第一次初始化
|
||
with patch.object(stock_system, "_setup_database"):
|
||
with patch.object(stock_system, "_setup_components"):
|
||
await stock_system.initialize()
|
||
|
||
# 第二次初始化
|
||
result = await stock_system.initialize()
|
||
|
||
assert result is False # 应该返回False，因为已经初始化过
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_initialize_database_failure(self, stock_system):
|
||
"""测试数据库初始化失败"""
|
||
with patch.object(stock_system, "_setup_database", side_effect=Exception("数据库错误")):
|
||
with pytest.raises(StockSystemError):
|
||
await stock_system.initialize()
|
||
|
||
assert stock_system.is_initialized is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_initialize_components_failure(self, stock_system):
|
||
"""测试组件初始化失败"""
|
||
with patch.object(stock_system, "_setup_database"):
|
||
with patch.object(stock_system, "_setup_components", side_effect=Exception("组件错误")):
|
||
with pytest.raises(StockSystemError):
|
||
await stock_system.initialize()
|
||
|
||
assert stock_system.is_initialized is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_start_scheduler_success(self, stock_system):
|
||
"""测试启动调度器成功"""
|
||
# 先初始化系统
|
||
with patch.object(stock_system, "_setup_database"):
|
||
with patch.object(stock_system, "_setup_components"):
|
||
await stock_system.initialize()
|
||
|
||
# Mock调度器启动
|
||
with patch.object(stock_system.task_scheduler, "start", return_value=True):
|
||
result = await stock_system.start_scheduler()
|
||
|
||
assert result is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_start_scheduler_not_initialized(self, stock_system):
|
||
"""测试未初始化时启动调度器"""
|
||
result = await stock_system.start_scheduler()
|
||
|
||
assert result is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_start_scheduler_failure(self, stock_system):
|
||
"""测试启动调度器失败"""
|
||
# 先初始化系统
|
||
with patch.object(stock_system, "_setup_database"):
|
||
with patch.object(stock_system, "_setup_components"):
|
||
await stock_system.initialize()
|
||
|
||
# Mock调度器启动失败
|
||
with patch.object(stock_system.task_scheduler, "start", return_value=False):
|
||
result = await stock_system.start_scheduler()
|
||
|
||
assert result is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_stop_scheduler_success(self, stock_system):
|
||
"""测试停止调度器成功"""
|
||
# 先初始化系统
|
||
with patch.object(stock_system, "_setup_database"):
|
||
with patch.object(stock_system, "_setup_components"):
|
||
await stock_system.initialize()
|
||
|
||
# Mock调度器停止
|
||
with patch.object(stock_system.task_scheduler, "stop", return_value=True):
|
||
result = await stock_system.stop_scheduler()
|
||
|
||
assert result is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_stop_scheduler_not_initialized(self, stock_system):
|
||
"""测试未初始化时停止调度器"""
|
||
result = await stock_system.stop_scheduler()
|
||
|
||
assert result is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_system_status_success(self, stock_system):
|
||
"""测试获取系统状态成功"""
|
||
# 先初始化系统
|
||
with patch.object(stock_system, "_setup_database"):
|
||
with patch.object(stock_system, "_setup_components"):
|
||
await stock_system.initialize()
|
||
|
||
# Mock组件状态
|
||
with patch.object(stock_system.data_manager, "get_stock_basic_info", return_value=[{"code": "000001"}]):
|
||
with patch.object(stock_system.stock_repo, "get_stock_basic_info", return_value=[Mock()]):
|
||
with patch.object(stock_system.task_scheduler, "get_job_status", return_value={"test_job": {}}):
|
||
with patch.object(stock_system.task_scheduler, "is_running", True):
|
||
status = await stock_system.get_system_status()
|
||
|
||
assert "initialization" in status
|
||
assert "data_sources" in status
|
||
assert "database" in status
|
||
assert "scheduler" in status
|
||
assert "data_statistics" in status
|
||
|
||
assert status["initialization"]["status"] == "已初始化"
|
||
assert status["scheduler"]["status"] == "运行中"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_system_status_not_initialized(self, stock_system):
|
||
"""测试未初始化时获取系统状态"""
|
||
status = await stock_system.get_system_status()
|
||
|
||
assert "initialization" in status
|
||
assert status["initialization"]["status"] == "未初始化"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_initialize_all_data_success(self, stock_system):
|
||
"""测试初始化所有数据成功"""
|
||
# 先初始化系统
|
||
with patch.object(stock_system, "_setup_database"):
|
||
with patch.object(stock_system, "_setup_components"):
|
||
await stock_system.initialize()
|
||
|
||
# Mock数据初始化器
|
||
mock_initializer = Mock()
|
||
mock_initializer.initialize_all_data.return_value = {
|
||
"stock_basic": {"count": 1000, "status": "成功"},
|
||
"daily_kline": {"count": 50000, "status": "成功"},
|
||
"financial_report": {"count": 2000, "status": "成功"}
|
||
}
|
||
|
||
with patch("src.data.data_initializer.DataInitializer", return_value=mock_initializer):
|
||
result = await stock_system.initialize_all_data()
|
||
|
||
assert "stock_basic" in result
|
||
assert "daily_kline" in result
|
||
assert "financial_report" in result
|
||
assert result["stock_basic"]["count"] == 1000
|
||
assert result["daily_kline"]["count"] == 50000
|
||
assert result["financial_report"]["count"] == 2000
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_initialize_all_data_not_initialized(self, stock_system):
|
||
"""测试未初始化时初始化数据"""
|
||
result = await stock_system.initialize_all_data()
|
||
|
||
assert result == {}
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_update_daily_data_success(self, stock_system):
|
||
"""测试更新每日数据成功"""
|
||
# 先初始化系统
|
||
with patch.object(stock_system, "_setup_database"):
|
||
with patch.object(stock_system, "_setup_components"):
|
||
await stock_system.initialize()
|
||
|
||
# Mock数据更新
|
||
with patch.object(stock_system.task_scheduler, "_update_daily_kline_data", return_value=True):
|
||
result = await stock_system.update_daily_data()
|
||
|
||
assert result is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_update_daily_data_not_initialized(self, stock_system):
|
||
"""测试未初始化时更新数据"""
|
||
result = await stock_system.update_daily_data()
|
||
|
||
assert result is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_update_financial_data_success(self, stock_system):
|
||
"""测试更新财务数据成功"""
|
||
# 先初始化系统
|
||
with patch.object(stock_system, "_setup_database"):
|
||
with patch.object(stock_system, "_setup_components"):
|
||
await stock_system.initialize()
|
||
|
||
# Mock数据更新
|
||
with patch.object(stock_system.task_scheduler, "_update_financial_data", return_value=True):
|
||
result = await stock_system.update_financial_data()
|
||
|
||
assert result is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_update_stock_basic_info_success(self, stock_system):
|
||
"""测试更新股票基础信息成功"""
|
||
# 先初始化系统
|
||
with patch.object(stock_system, "_setup_database"):
|
||
with patch.object(stock_system, "_setup_components"):
|
||
await stock_system.initialize()
|
||
|
||
# Mock数据更新
|
||
with patch.object(stock_system.task_scheduler, "_update_stock_basic_info", return_value=True):
|
||
result = await stock_system.update_stock_basic_info()
|
||
|
||
assert result is True
|
||
|
||
|
||
class TestCommandLineInterface:
|
||
"""命令行接口测试类"""
|
||
|
||
@pytest.fixture
|
||
def capture_output(self):
|
||
"""捕获标准输出"""
|
||
old_stdout = sys.stdout
|
||
sys.stdout = StringIO()
|
||
yield sys.stdout
|
||
sys.stdout = old_stdout
|
||
|
||
def test_parse_arguments_init(self):
|
||
"""测试解析init命令参数"""
|
||
test_args = ["main.py", "init"]
|
||
|
||
with patch("sys.argv", test_args):
|
||
args = StockAnalysisSystem.parse_arguments()
|
||
|
||
assert args.command == "init"
|
||
assert args.force is False
|
||
|
||
def test_parse_arguments_scheduler(self):
|
||
"""测试解析scheduler命令参数"""
|
||
test_args = ["main.py", "scheduler", "--start"]
|
||
|
||
with patch("sys.argv", test_args):
|
||
args = StockAnalysisSystem.parse_arguments()
|
||
|
||
assert args.command == "scheduler"
|
||
assert args.start is True
|
||
assert args.stop is False
|
||
|
||
def test_parse_arguments_status(self):
|
||
"""测试解析status命令参数"""
|
||
test_args = ["main.py", "status"]
|
||
|
||
with patch("sys.argv", test_args):
|
||
args = StockAnalysisSystem.parse_arguments()
|
||
|
||
assert args.command == "status"
|
||
|
||
def test_parse_arguments_update(self):
|
||
"""测试解析update命令参数"""
|
||
test_args = ["main.py", "update", "--daily"]
|
||
|
||
with patch("sys.argv", test_args):
|
||
args = StockAnalysisSystem.parse_arguments()
|
||
|
||
assert args.command == "update"
|
||
assert args.daily is True
|
||
assert args.financial is False
|
||
assert args.basic is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_init_command_success(self, stock_system, capture_output):
|
||
"""测试运行init命令成功"""
|
||
with patch.object(stock_system, "initialize", return_value=True):
|
||
with patch.object(stock_system, "initialize_all_data", return_value={"stock_basic": {"count": 1000}}):
|
||
result = await stock_system.run_command("init")
|
||
|
||
assert result is True
|
||
|
||
# 检查输出
|
||
output = capture_output.getvalue()
|
||
assert "系统初始化成功" in output
|
||
assert "数据初始化完成" in output
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_init_command_failure(self, stock_system, capture_output):
|
||
"""测试运行init命令失败"""
|
||
with patch.object(stock_system, "initialize", side_effect=StockSystemError("初始化失败")):
|
||
result = await stock_system.run_command("init")
|
||
|
||
assert result is False
|
||
|
||
# 检查输出
|
||
output = capture_output.getvalue()
|
||
assert "系统初始化失败" in output
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_scheduler_start_command_success(self, stock_system, capture_output):
|
||
"""测试运行scheduler start命令成功"""
|
||
# 先初始化系统
|
||
with patch.object(stock_system, "_setup_database"):
|
||
with patch.object(stock_system, "_setup_components"):
|
||
await stock_system.initialize()
|
||
|
||
with patch.object(stock_system, "start_scheduler", return_value=True):
|
||
result = await stock_system.run_command("scheduler", start=True)
|
||
|
||
assert result is True
|
||
|
||
# 检查输出
|
||
output = capture_output.getvalue()
|
||
assert "定时任务调度器启动成功" in output
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_scheduler_stop_command_success(self, stock_system, capture_output):
|
||
"""测试运行scheduler stop命令成功"""
|
||
# 先初始化系统
|
||
with patch.object(stock_system, "_setup_database"):
|
||
with patch.object(stock_system, "_setup_components"):
|
||
await stock_system.initialize()
|
||
|
||
with patch.object(stock_system, "stop_scheduler", return_value=True):
|
||
result = await stock_system.run_command("scheduler", stop=True)
|
||
|
||
assert result is True
|
||
|
||
# 检查输出
|
||
output = capture_output.getvalue()
|
||
assert "定时任务调度器停止成功" in output
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_status_command_success(self, stock_system, capture_output):
|
||
"""测试运行status命令成功"""
|
||
# 先初始化系统
|
||
with patch.object(stock_system, "_setup_database"):
|
||
with patch.object(stock_system, "_setup_components"):
|
||
await stock_system.initialize()
|
||
|
||
mock_status = {
|
||
"initialization": {"status": "已初始化"},
|
||
"data_sources": {"akshare": "正常"},
|
||
"database": {"status": "正常"},
|
||
"scheduler": {"status": "运行中"},
|
||
"data_statistics": {"stock_count": 1000}
|
||
}
|
||
|
||
with patch.object(stock_system, "get_system_status", return_value=mock_status):
|
||
result = await stock_system.run_command("status")
|
||
|
||
assert result is True
|
||
|
||
# 检查输出
|
||
output = capture_output.getvalue()
|
||
assert "系统状态" in output
|
||
assert "已初始化" in output
|
||
assert "运行中" in output
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_update_daily_command_success(self, stock_system, capture_output):
|
||
"""测试运行update daily命令成功"""
|
||
# 先初始化系统
|
||
with patch.object(stock_system, "_setup_database"):
|
||
with patch.object(stock_system, "_setup_components"):
|
||
await stock_system.initialize()
|
||
|
||
with patch.object(stock_system, "update_daily_data", return_value=True):
|
||
result = await stock_system.run_command("update", daily=True)
|
||
|
||
assert result is True
|
||
|
||
# 检查输出
|
||
output = capture_output.getvalue()
|
||
assert "每日数据更新完成" in output
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_update_financial_command_success(self, stock_system, capture_output):
|
||
"""测试运行update financial命令成功"""
|
||
# 先初始化系统
|
||
with patch.object(stock_system, "_setup_database"):
|
||
with patch.object(stock_system, "_setup_components"):
|
||
await stock_system.initialize()
|
||
|
||
with patch.object(stock_system, "update_financial_data", return_value=True):
|
||
result = await stock_system.run_command("update", financial=True)
|
||
|
||
assert result is True
|
||
|
||
# 检查输出
|
||
output = capture_output.getvalue()
|
||
assert "财务数据更新完成" in output
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_update_basic_command_success(self, stock_system, capture_output):
|
||
"""测试运行update basic命令成功"""
|
||
# 先初始化系统
|
||
with patch.object(stock_system, "_setup_database"):
|
||
with patch.object(stock_system, "_setup_components"):
|
||
await stock_system.initialize()
|
||
|
||
with patch.object(stock_system, "update_stock_basic_info", return_value=True):
|
||
result = await stock_system.run_command("update", basic=True)
|
||
|
||
assert result is True
|
||
|
||
# 检查输出
|
||
output = capture_output.getvalue()
|
||
assert "股票基础信息更新完成" in output
|
||
|
||
|
||
class TestErrorHandling:
|
||
"""错误处理测试类"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_handle_exception(self, stock_system, capture_output):
|
||
"""测试异常处理"""
|
||
# 模拟异常
|
||
try:
|
||
raise StockSystemError("测试异常")
|
||
except Exception as e:
|
||
result = stock_system._handle_exception(e)
|
||
|
||
assert result is False
|
||
|
||
# 检查输出
|
||
output = capture_output.getvalue()
|
||
assert "测试异常" in output
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_handle_keyboard_interrupt(self, stock_system, capture_output):
|
||
"""测试键盘中断处理"""
|
||
# 模拟键盘中断
|
||
try:
|
||
raise KeyboardInterrupt()
|
||
except Exception as e:
|
||
result = stock_system._handle_exception(e)
|
||
|
||
assert result is True
|
||
|
||
# 检查输出
|
||
output = capture_output.getvalue()
|
||
assert "程序被用户中断" in output
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_main_function_success(self, capture_output):
|
||
"""测试主函数成功执行"""
|
||
# Mock命令行参数和系统运行
|
||
test_args = ["main.py", "status"]
|
||
|
||
with patch("sys.argv", test_args):
|
||
with patch("src.main.StockAnalysisSystem.run_command", return_value=True):
|
||
from src.main import main
|
||
|
||
result = await main()
|
||
|
||
assert result == 0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_main_function_failure(self, capture_output):
|
||
"""测试主函数执行失败"""
|
||
# Mock命令行参数和系统运行失败
|
||
test_args = ["main.py", "status"]
|
||
|
||
with patch("sys.argv", test_args):
|
||
with patch("src.main.StockAnalysisSystem.run_command", return_value=False):
|
||
from src.main import main
|
||
|
||
result = await main()
|
||
|
||
assert result == 1 |