mirror of
https://github.com/ZhuLinsen/daily_stock_analysis
synced 2026-09-20 10:53:33 +08:00
303f4e1c18
* feat: GitHub Actions STOCK_LIST 指数入口与 CLI unsupported 明确拒绝 * feat: 指数注册表新增国证粮食与中证钢铁，补充指数自选股配置文档 - seed 与 bundled 指数清单 31 项扩展至 33 项（sz399365、csi930606） - 中英 full-guide 新增指数自选股配置小节，README/DEPLOY 同步提示 - 相关确定性测试断言 31→33 并锁定新 canonical * fix: market-only 分类跳过测试补齐交易日历 mock，消除周末失败 * fix: 测试类级默认关闭 GITHUB_ACTIONS，修复 CI runner 环境注入导致的分类分支误入 * fix: 入口分类按模式边界跳过不消费个股列表的模式，文档收窄指数入口适用范围 - 新增 _skips_stock_entry guard：--backtest/--market-review/--serve-only/--webui-only/--portfolio/--schedule/config.schedule_enabled 整体跳过 --stocks 与 Actions STOCK_LIST 的分类与索引刷新 - --schedule --stocks 的忽略快照警告与分类解耦（传 args.stocks or None） - 恢复 --stocks + --portfolio 同框提示日志 - 测试：review 反例四格 + Actions portfolio 格 + portfolio 覆盖到达 pipeline.run 入参断言 - 文档：指数分类与整批拒绝收窄为 --stocks 与 GitHub Actions 两入口
2796 lines
120 KiB
Python
2796 lines
120 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Regression tests for scheduled mode stock selection behavior."""
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import socket
|
||
import tempfile
|
||
import unittest
|
||
from datetime import date, datetime, timezone
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
from tests.litellm_stub import ensure_litellm_stub
|
||
|
||
ensure_litellm_stub()
|
||
|
||
_ENV_BEFORE_MAIN_IMPORT = dict(os.environ)
|
||
import main
|
||
from src.brokers.futu.portfolio import FutuPortfolioError
|
||
from src.config import Config
|
||
from src.services.stock_list_parser import ParseStatus, parse_analysis_target
|
||
|
||
_MAIN_IMPORT_ENV_ADDITIONS = frozenset(set(os.environ) - set(_ENV_BEFORE_MAIN_IMPORT))
|
||
_MAIN_IMPORT_ENV_OVERRIDES = {
|
||
key: value
|
||
for key, value in _ENV_BEFORE_MAIN_IMPORT.items()
|
||
if os.environ.get(key) != value
|
||
}
|
||
|
||
|
||
def _api_app_stub_modules():
|
||
"""sys.modules entries so ``start_api_server`` can ``from api.app import app``
|
||
without importing the real (heavy) app tree in these isolated unit tests.
|
||
|
||
``start_api_server`` imports the ASGI app object in the calling thread so the
|
||
import stays out of the uvicorn startup probe window; these control-flow tests
|
||
stub it the same way they already stub uvicorn.
|
||
"""
|
||
import types
|
||
|
||
api_pkg = types.ModuleType("api")
|
||
api_app_mod = types.ModuleType("api.app")
|
||
api_app_mod.app = SimpleNamespace()
|
||
api_pkg.app = api_app_mod
|
||
return {"api": api_pkg, "api.app": api_app_mod}
|
||
|
||
|
||
class _DummyConfig(SimpleNamespace):
|
||
def validate(self):
|
||
return []
|
||
|
||
|
||
class MainScheduleModeTestCase(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
self.temp_dir = tempfile.TemporaryDirectory()
|
||
self.env_path = Path(self.temp_dir.name) / ".env"
|
||
self.env_path.write_text("STOCK_LIST=600519\n", encoding="utf-8")
|
||
self.original_cwd = os.getcwd()
|
||
os.chdir(self.temp_dir.name)
|
||
self.env_patch = patch.dict(os.environ, {"ENV_FILE": str(self.env_path)}, clear=False)
|
||
self.env_patch.start()
|
||
# CI runner 自动注入 GITHUB_ACTIONS=true，会让无显式环境的 main.main()
|
||
# 测试误入 Actions STOCK_LIST 分类分支（其 _DummyConfig 不含
|
||
# stock_list 属性）。类级默认关闭，Actions 入口相关测试在用例内
|
||
# 显式 patch 为 true 覆盖。
|
||
self.actions_env_patch = patch.dict(
|
||
os.environ, {"GITHUB_ACTIONS": "false"}, clear=False
|
||
)
|
||
self.actions_env_patch.start()
|
||
Config.reset_instance()
|
||
root_logger = logging.getLogger()
|
||
self._original_root_handlers = list(root_logger.handlers)
|
||
self._original_root_level = root_logger.level
|
||
|
||
def tearDown(self) -> None:
|
||
root_logger = logging.getLogger()
|
||
current_handlers = list(root_logger.handlers)
|
||
for handler in current_handlers:
|
||
if handler not in self._original_root_handlers:
|
||
root_logger.removeHandler(handler)
|
||
try:
|
||
handler.close()
|
||
except Exception:
|
||
pass
|
||
root_logger.setLevel(self._original_root_level)
|
||
os.chdir(self.original_cwd)
|
||
Config.reset_instance()
|
||
self.env_patch.stop()
|
||
self.actions_env_patch.stop()
|
||
for key in _MAIN_IMPORT_ENV_ADDITIONS:
|
||
os.environ.pop(key, None)
|
||
for key, value in _MAIN_IMPORT_ENV_OVERRIDES.items():
|
||
os.environ[key] = value
|
||
self.temp_dir.cleanup()
|
||
|
||
def _make_args(self, **overrides):
|
||
defaults = {
|
||
"debug": False,
|
||
"stocks": None,
|
||
"portfolio": None,
|
||
"webui": False,
|
||
"webui_only": False,
|
||
"serve": False,
|
||
"serve_only": False,
|
||
"host": None,
|
||
"port": None,
|
||
"backtest": False,
|
||
"market_review": False,
|
||
"schedule": False,
|
||
"no_run_immediately": False,
|
||
"no_notify": False,
|
||
"check_notify": False,
|
||
"no_market_review": False,
|
||
"dry_run": False,
|
||
"workers": 1,
|
||
"force_run": False,
|
||
"single_notify": False,
|
||
"no_context_snapshot": False,
|
||
}
|
||
defaults.update(overrides)
|
||
return SimpleNamespace(**defaults)
|
||
|
||
def _make_config(self, **overrides):
|
||
defaults = {
|
||
"log_dir": self.temp_dir.name,
|
||
"webui_enabled": False,
|
||
"webui_host": "127.0.0.1",
|
||
"webui_port": 8000,
|
||
"dingtalk_stream_enabled": False,
|
||
"feishu_stream_enabled": False,
|
||
"schedule_enabled": False,
|
||
"schedule_time": "18:00",
|
||
"schedule_run_immediately": True,
|
||
"run_immediately": True,
|
||
"agent_event_monitor_enabled": False,
|
||
"agent_event_alert_rules_json": "",
|
||
"agent_event_monitor_interval_minutes": 5,
|
||
"daily_market_context_enabled": True,
|
||
"market_review_enabled": False,
|
||
}
|
||
defaults.update(overrides)
|
||
return _DummyConfig(**defaults)
|
||
|
||
def test_daily_market_context_target_date_routes_jp_kr_calendars(self) -> None:
|
||
current_time = datetime(2026, 5, 7, 0, 30, tzinfo=timezone.utc)
|
||
calls = []
|
||
|
||
def resolve_effective_date(market, *, current_time=None):
|
||
calls.append((market, current_time))
|
||
return date(2026, 5, 7)
|
||
|
||
with patch(
|
||
"src.core.trading_calendar.get_effective_trading_date",
|
||
side_effect=resolve_effective_date,
|
||
):
|
||
self.assertEqual(
|
||
main._resolve_daily_market_context_target_date("jp", current_time),
|
||
date(2026, 5, 7),
|
||
)
|
||
self.assertEqual(
|
||
main._resolve_daily_market_context_target_date("kr", current_time),
|
||
date(2026, 5, 7),
|
||
)
|
||
self.assertEqual(
|
||
main._resolve_daily_market_context_target_date("jp,kr", current_time),
|
||
date(2026, 5, 7),
|
||
)
|
||
|
||
self.assertEqual(
|
||
calls,
|
||
[
|
||
("jp", current_time),
|
||
("kr", current_time),
|
||
("jp", current_time),
|
||
],
|
||
)
|
||
|
||
def test_compute_trading_day_filter_supports_comma_list_regions(self) -> None:
|
||
args = self._make_args()
|
||
config = self._make_config(
|
||
trading_day_check_enabled=True,
|
||
market_review_enabled=True,
|
||
market_review_region="jp,kr",
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
|
||
stock_codes = ["cn-stock", "jp-stock", "kr-stock", "us-stock", "none-stock"]
|
||
|
||
with patch(
|
||
"src.core.trading_calendar.get_market_for_stock",
|
||
side_effect=lambda code: {"cn-stock": "cn", "jp-stock": "jp", "kr-stock": "kr", "us-stock": "us"}.get(code),
|
||
), patch("src.core.trading_calendar.get_open_markets_today", return_value={"jp", "kr"}):
|
||
filtered_codes, effective_region, should_skip_all = main._compute_trading_day_filter(
|
||
config,
|
||
args,
|
||
stock_codes,
|
||
)
|
||
|
||
self.assertEqual(filtered_codes, ["jp-stock", "kr-stock", "none-stock"])
|
||
self.assertEqual(effective_region, "jp,kr")
|
||
self.assertFalse(should_skip_all)
|
||
|
||
def test_compute_trading_day_filter_filters_registered_indices_on_cn_holiday(self) -> None:
|
||
"""Index codes whose ``get_market_for_stock`` returns None must still
|
||
participate in CN trading-day filtering via ``parse_analysis_target``
|
||
(INDEX -> market=cn). On a CN holiday the indices are dropped, a US
|
||
stock whose market is open stays, and a market-unknown non-index code
|
||
stays (fail-open unchanged)."""
|
||
args = self._make_args()
|
||
config = self._make_config(
|
||
trading_day_check_enabled=True,
|
||
market_review_enabled=False,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
|
||
stock_codes = ["sh000016", "csi930955", "930955.CSI", "AAPL", "XYZ123"]
|
||
|
||
def fake_market(code: str):
|
||
return "us" if code == "AAPL" else None
|
||
|
||
with patch(
|
||
"src.core.trading_calendar.get_market_for_stock",
|
||
side_effect=fake_market,
|
||
), patch("src.core.trading_calendar.get_open_markets_today", return_value={"us"}):
|
||
filtered_codes, effective_region, should_skip_all = main._compute_trading_day_filter(
|
||
config,
|
||
args,
|
||
stock_codes,
|
||
)
|
||
|
||
# 已登记指数按 market=cn 参与过滤，CN 休市时被剔除；AAPL 因 US 开市保留；
|
||
# 市场未知的非指数 code（XYZ123）继续 fail-open 保留。
|
||
self.assertEqual(filtered_codes, ["AAPL", "XYZ123"])
|
||
self.assertIsNone(effective_region)
|
||
self.assertFalse(should_skip_all)
|
||
|
||
def test_public_webui_bind_warns_when_auth_is_disabled(self) -> None:
|
||
with patch("src.auth.is_auth_enabled", return_value=False), \
|
||
patch("main.logger.warning") as warning_log:
|
||
main._warn_if_public_webui_without_auth("0.0.0.0")
|
||
|
||
warning_log.assert_called_once()
|
||
self.assertIn("WEBUI_HOST=%s", warning_log.call_args.args[0])
|
||
self.assertEqual(warning_log.call_args.args[1], "0.0.0.0")
|
||
|
||
def test_loopback_webui_bind_does_not_warn_when_auth_is_disabled(self) -> None:
|
||
with patch("src.auth.is_auth_enabled", return_value=False), \
|
||
patch("main.logger.warning") as warning_log:
|
||
main._warn_if_public_webui_without_auth("127.0.0.1")
|
||
|
||
warning_log.assert_not_called()
|
||
|
||
def test_web_service_bind_uses_config_when_cli_omits_host_and_port(self) -> None:
|
||
args = self._make_args(host=None, port=None)
|
||
config = self._make_config(webui_host="127.0.0.1", webui_port=18000)
|
||
|
||
host, port = main._resolve_web_service_bind(args, config)
|
||
|
||
self.assertEqual(host, "127.0.0.1")
|
||
self.assertEqual(port, 18000)
|
||
|
||
def test_web_service_bind_keeps_explicit_cli_host_and_port(self) -> None:
|
||
args = self._make_args(host="0.0.0.0", port=8000)
|
||
config = self._make_config(webui_host="127.0.0.1", webui_port=18000)
|
||
|
||
host, port = main._resolve_web_service_bind(args, config)
|
||
|
||
self.assertEqual(host, "0.0.0.0")
|
||
self.assertEqual(port, 8000)
|
||
|
||
def test_serve_only_uses_config_bind_when_cli_omits_host_and_port(self) -> None:
|
||
args = self._make_args(serve_only=True)
|
||
config = self._make_config(webui_enabled=False, webui_host="127.0.0.1", webui_port=18000)
|
||
observed_bind = []
|
||
|
||
def fake_start_api_server(host, port, config):
|
||
observed_bind.append((host, port))
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "false"}, clear=False), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.prepare_webui_frontend_assets", return_value=True), \
|
||
patch("main.start_api_server", side_effect=fake_start_api_server), \
|
||
patch("main.start_bot_stream_clients"), \
|
||
patch("main.time.sleep", side_effect=KeyboardInterrupt):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
self.assertEqual(observed_bind, [("127.0.0.1", 18000)])
|
||
|
||
def test_serve_only_keeps_explicit_cli_bind_over_config(self) -> None:
|
||
args = self._make_args(serve_only=True, host="0.0.0.0", port=8000)
|
||
config = self._make_config(webui_enabled=False, webui_host="127.0.0.1", webui_port=18000)
|
||
observed_bind = []
|
||
|
||
def fake_start_api_server(host, port, config):
|
||
observed_bind.append((host, port))
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "false"}, clear=False), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.prepare_webui_frontend_assets", return_value=True), \
|
||
patch("main.start_api_server", side_effect=fake_start_api_server), \
|
||
patch("main.start_bot_stream_clients"), \
|
||
patch("main.time.sleep", side_effect=KeyboardInterrupt):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
self.assertEqual(observed_bind, [("0.0.0.0", 8000)])
|
||
|
||
def test_start_api_server_fails_before_thread_when_port_is_busy(self) -> None:
|
||
config = self._make_config(log_level="INFO")
|
||
|
||
class BusySocket:
|
||
def bind(self, address):
|
||
raise OSError("address already in use")
|
||
|
||
def close(self):
|
||
pass
|
||
|
||
with patch("socket.socket", return_value=BusySocket()) as socket_factory, \
|
||
patch("threading.Thread") as thread_cls:
|
||
with self.assertRaises(RuntimeError) as caught:
|
||
main.start_api_server("127.0.0.1", 8000, config)
|
||
|
||
socket_factory.assert_called_once_with(socket.AF_INET, socket.SOCK_STREAM)
|
||
self.assertIn("127.0.0.1:8000", str(caught.exception))
|
||
thread_cls.assert_not_called()
|
||
|
||
def test_start_api_server_fails_when_uvicorn_background_startup_fails(self) -> None:
|
||
config = self._make_config(log_level="INFO")
|
||
|
||
class _FakeUvicornServer:
|
||
def __init__(self, config):
|
||
self.config = config
|
||
self.started = False
|
||
|
||
def run(self) -> None:
|
||
raise RuntimeError("lifespan bootstrap failed")
|
||
|
||
class _FakeUvicornConfig:
|
||
def __init__(self, *args, **kwargs):
|
||
pass
|
||
|
||
class _FakeUvicornModule:
|
||
Config = _FakeUvicornConfig
|
||
|
||
Server = _FakeUvicornServer
|
||
|
||
class _UnusedSocket:
|
||
def bind(self, address):
|
||
pass
|
||
|
||
def close(self):
|
||
pass
|
||
|
||
with patch("socket.socket", return_value=_UnusedSocket()), \
|
||
patch.dict(
|
||
"sys.modules",
|
||
{"uvicorn": _FakeUvicornModule(), **_api_app_stub_modules()},
|
||
):
|
||
|
||
with self.assertRaises(RuntimeError) as caught:
|
||
main.start_api_server("127.0.0.1", 8000, config)
|
||
|
||
self.assertIn("lifespan bootstrap failed", str(caught.exception))
|
||
|
||
def test_start_api_server_compatible_with_uvicorn_install_signal_handlers_method(self) -> None:
|
||
config = self._make_config(log_level="INFO")
|
||
|
||
class _CompatServer:
|
||
instance = None
|
||
|
||
def __init__(self, config):
|
||
type(self).instance = self
|
||
self.config = config
|
||
self.started = False
|
||
self.install_signal_handlers = self._install_signal_handlers
|
||
|
||
def _install_signal_handlers(self) -> None:
|
||
return None
|
||
|
||
def run(self) -> None:
|
||
self.started = True
|
||
|
||
class _CompatConfig:
|
||
def __init__(self, *args, **kwargs):
|
||
if "install_signal_handlers" in kwargs:
|
||
raise TypeError("install_signal_handlers is unsupported")
|
||
|
||
class _UnusedSocket:
|
||
def bind(self, address):
|
||
pass
|
||
|
||
def close(self):
|
||
pass
|
||
|
||
with patch("socket.socket", return_value=_UnusedSocket()), \
|
||
patch.dict(
|
||
"sys.modules",
|
||
{
|
||
"uvicorn": SimpleNamespace(Config=_CompatConfig, Server=_CompatServer),
|
||
**_api_app_stub_modules(),
|
||
},
|
||
):
|
||
main.start_api_server("127.0.0.1", 8000, config)
|
||
|
||
self.assertIsNotNone(_CompatServer.instance)
|
||
self.assertTrue(callable(_CompatServer.instance.install_signal_handlers))
|
||
self.assertTrue(_CompatServer.instance.started)
|
||
|
||
def test_schedule_mode_ignores_cli_stock_snapshot(self) -> None:
|
||
args = self._make_args(schedule=True, stocks="600519,000001")
|
||
config = self._make_config(schedule_enabled=False)
|
||
scheduled_call = {}
|
||
|
||
def fake_run_with_schedule(
|
||
task,
|
||
schedule_time,
|
||
run_immediately,
|
||
background_tasks=None,
|
||
schedule_time_provider=None,
|
||
):
|
||
scheduled_call["schedule_time"] = schedule_time
|
||
scheduled_call["run_immediately"] = run_immediately
|
||
scheduled_call["background_tasks"] = background_tasks or []
|
||
scheduled_call["resolved_schedule_time"] = (
|
||
schedule_time_provider() if schedule_time_provider is not None else None
|
||
)
|
||
task()
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main._reload_runtime_config", return_value=config), \
|
||
patch("main._build_schedule_time_provider", return_value=lambda: "18:00"), \
|
||
patch("main.setup_logging"), \
|
||
patch("main.run_full_analysis") as run_full_analysis, \
|
||
patch("main.logger.warning") as warning_log, \
|
||
patch("main._refresh_stock_index_cache_for_analysis"), \
|
||
patch("src.scheduler.run_with_schedule", side_effect=fake_run_with_schedule):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
self.assertEqual(
|
||
scheduled_call,
|
||
{
|
||
"schedule_time": "18:00",
|
||
"run_immediately": True,
|
||
"background_tasks": [],
|
||
"resolved_schedule_time": "18:00",
|
||
},
|
||
)
|
||
run_full_analysis.assert_called_once_with(config, args, None)
|
||
warning_log.assert_any_call(
|
||
"定时模式下检测到 --stocks 参数；计划执行将忽略启动时股票快照，并在每次运行前重新读取最新的 STOCK_LIST。"
|
||
)
|
||
|
||
def test_standalone_run_resolves_stocks_before_run_full_analysis(self) -> None:
|
||
args = self._make_args(stocks="005930")
|
||
config = self._make_config(run_immediately=True)
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch("main._refresh_stock_index_cache_for_analysis"), \
|
||
patch("main.run_full_analysis") as run_full_analysis:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
run_full_analysis.assert_called_once()
|
||
_, _, stock_codes = run_full_analysis.call_args.args
|
||
analysis_targets = run_full_analysis.call_args.kwargs.get("analysis_targets")
|
||
self.assertEqual(stock_codes, ["005930.KS"])
|
||
self.assertEqual(len(analysis_targets), 1)
|
||
self.assertEqual(analysis_targets[0].asset_type, "stock")
|
||
|
||
def test_standalone_run_builds_structured_index_targets(self) -> None:
|
||
args = self._make_args(
|
||
stocks="sh000016,000300.CSI,930955.CSI,000016"
|
||
)
|
||
config = self._make_config(run_immediately=True)
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch("main._refresh_stock_index_cache_for_analysis"), \
|
||
patch("main.run_full_analysis") as run_full_analysis:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
_, _, stock_codes = run_full_analysis.call_args.args
|
||
analysis_targets = run_full_analysis.call_args.kwargs["analysis_targets"]
|
||
self.assertEqual(
|
||
stock_codes,
|
||
["sh000016", "sh000300", "csi930955", "000016"],
|
||
)
|
||
self.assertEqual(
|
||
[target.asset_type for target in analysis_targets],
|
||
["index", "index", "index", "stock"],
|
||
)
|
||
self.assertEqual(
|
||
[target.canonical_id for target in analysis_targets],
|
||
["sh000016", "sh000300", "csi930955", "sz000016"],
|
||
)
|
||
|
||
def test_standalone_run_refreshes_index_cache_before_parsing_stocks(self) -> None:
|
||
"""``main.main()`` must refresh the stock-index registry (best-effort)
|
||
BEFORE parsing ``--stocks`` so a first run with a stale local registry
|
||
resolves a newly-registered alias to an index target in the same run."""
|
||
args = self._make_args(stocks="930955.CSI")
|
||
config = self._make_config(run_immediately=True)
|
||
calls = []
|
||
|
||
def fake_refresh(cfg):
|
||
calls.append(("refresh", cfg))
|
||
|
||
def fake_run_full_analysis(cfg, a, stock_codes, **kwargs):
|
||
calls.append(("parse", stock_codes, kwargs.get("analysis_targets")))
|
||
return 0
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch(
|
||
"main._refresh_stock_index_cache_for_analysis",
|
||
side_effect=fake_refresh,
|
||
), \
|
||
patch("main.run_full_analysis", side_effect=fake_run_full_analysis):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
# 刷新必须先于 --stocks 解析执行。
|
||
self.assertEqual(calls[0][0], "refresh")
|
||
self.assertEqual(calls[1][0], "parse")
|
||
self.assertEqual(calls[1][1], ["csi930955"])
|
||
self.assertEqual(calls[1][2][0].asset_type, ParseStatus.INDEX)
|
||
self.assertEqual(calls[1][2][0].canonical_id, "csi930955")
|
||
|
||
def test_standalone_run_rejects_unsupported_stocks_token(self) -> None:
|
||
"""`--stocks` 携带未登记 `.CSI` 目标时在入口明确报错并返回非零，
|
||
不再静默走股票路径或发起 provider 调用。"""
|
||
args = self._make_args(stocks="600519,930956.CSI")
|
||
config = self._make_config(run_immediately=True)
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch("main._refresh_stock_index_cache_for_analysis"), \
|
||
patch("main.run_full_analysis") as run_full_analysis:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 1)
|
||
run_full_analysis.assert_not_called()
|
||
|
||
def test_standalone_run_actions_stock_list_builds_index_targets(self) -> None:
|
||
"""`GITHUB_ACTIONS=true` 且无 `--stocks` 时，`main.main()` 把默认
|
||
`STOCK_LIST` 的显式指数 token 分类为 INDEX target、个股 token 保持
|
||
legacy 语义（与一次性 `--stocks` 构造等价）。"""
|
||
args = self._make_args()
|
||
config = self._make_config(run_immediately=True)
|
||
config.stock_list = ["sh000016", "600519", "930955.CSI"]
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "true"}), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch("main._refresh_stock_index_cache_for_analysis"), \
|
||
patch("main.run_full_analysis") as run_full_analysis:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
run_full_analysis.assert_called_once()
|
||
stock_codes = run_full_analysis.call_args.args[2]
|
||
analysis_targets = run_full_analysis.call_args.kwargs["analysis_targets"]
|
||
self.assertEqual(stock_codes, ["sh000016", "600519", "csi930955"])
|
||
self.assertEqual(
|
||
[t.asset_type for t in analysis_targets],
|
||
["index", "stock", "index"],
|
||
)
|
||
self.assertEqual(
|
||
[t.canonical_id for t in analysis_targets],
|
||
["sh000016", "sh600519", "csi930955"],
|
||
)
|
||
|
||
def test_standalone_run_local_default_does_not_build_targets(self) -> None:
|
||
"""非 Actions 环境的无参数默认路径不构造结构化 targets（本地/
|
||
`--schedule` 热刷新语义保持不变）。"""
|
||
args = self._make_args()
|
||
config = self._make_config(run_immediately=True)
|
||
config.stock_list = ["600519"]
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": ""}), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch("main._refresh_stock_index_cache_for_analysis"), \
|
||
patch("main.run_full_analysis") as run_full_analysis:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
run_full_analysis.assert_called_once()
|
||
self.assertIsNone(run_full_analysis.call_args.args[2])
|
||
self.assertIsNone(run_full_analysis.call_args.kwargs.get("analysis_targets"))
|
||
|
||
def test_actions_backtest_with_bad_stock_list_reaches_backtest_service(self) -> None:
|
||
"""Review 反例：`GITHUB_ACTIONS=true` + `--backtest` 时不消费个股列表，
|
||
含未登记 `.CSI` 的 STOCK_LIST 不得整批拒绝，必须进入回测分支。"""
|
||
args = self._make_args(backtest=True)
|
||
config = self._make_config(run_immediately=True)
|
||
config.stock_list = ["930956.CSI"]
|
||
stats = {
|
||
"processed": 1,
|
||
"saved": 1,
|
||
"completed": 1,
|
||
"insufficient": 0,
|
||
"errors": 0,
|
||
}
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "true"}), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch("main._refresh_stock_index_cache_for_analysis") as refresh, \
|
||
patch("main._classify_stock_list_tokens") as classify, \
|
||
patch(
|
||
"src.services.backtest_service.BacktestService",
|
||
) as backtest_class, \
|
||
patch("main.logger.error") as error_log:
|
||
backtest_service = backtest_class.return_value
|
||
backtest_service.run_backtest.return_value = stats
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
refresh.assert_not_called()
|
||
classify.assert_not_called()
|
||
error_log.assert_not_called()
|
||
backtest_class.assert_called_once_with()
|
||
backtest_service.run_backtest.assert_called_once_with(
|
||
code=None,
|
||
force=False,
|
||
eval_window_days=None,
|
||
)
|
||
|
||
def test_actions_backtest_stock_list_classification_skipped_when_no_backtest(self) -> None:
|
||
"""对照组：`GITHUB_ACTIONS=true` 无模式参数的坏 STOCK_LIST 仍整批拒绝
|
||
（分类契约未回归）。"""
|
||
args = self._make_args()
|
||
config = self._make_config(run_immediately=True)
|
||
config.stock_list = ["930956.CSI"]
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "true"}), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch("main._refresh_stock_index_cache_for_analysis"), \
|
||
patch("main.run_full_analysis") as run_full_analysis, \
|
||
patch("main.logger.error") as error_log:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 1)
|
||
run_full_analysis.assert_not_called()
|
||
error_log.assert_any_call(
|
||
"%s 包含不支持的目标 %r：%s；本轮不执行任何分析。",
|
||
"GitHub Actions STOCK_LIST",
|
||
"930956.CSI",
|
||
unittest.mock.ANY,
|
||
)
|
||
|
||
def test_portfolio_futu_with_bad_stocks_token_reaches_run_full_analysis(self) -> None:
|
||
"""Review 反例：`--portfolio futu --stocks 930956.CSI` 同框时分类整体跳过，
|
||
坏 token 不拦截；run_full_analysis 内 portfolio 覆盖 `--stocks`。"""
|
||
args = self._make_args(portfolio="futu", stocks="930956.CSI")
|
||
config = self._make_config(run_immediately=True)
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "false"}, clear=False), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch("main._refresh_stock_index_cache_for_analysis") as refresh, \
|
||
patch("main._classify_stock_list_tokens") as classify, \
|
||
patch(
|
||
"src.brokers.futu.portfolio.load_futu_stock_codes",
|
||
return_value=["AAPL"],
|
||
), \
|
||
patch(
|
||
"main._compute_trading_day_filter",
|
||
return_value=(["AAPL"], "us", False),
|
||
), \
|
||
patch("main._run_analysis_with_runtime_scheduler_lock") as run_with_lock:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
refresh.assert_not_called()
|
||
classify.assert_not_called()
|
||
run_with_lock.assert_called_once_with(config, args, None, None)
|
||
|
||
def test_portfolio_futu_with_bad_stocks_token_runs_portfolio_codes(self) -> None:
|
||
"""对照组：`--portfolio futu --stocks 600519` 时 run_full_analysis 内
|
||
portfolio 覆盖 `--stocks`，覆盖语义不回归。走真实运行时锁路径
|
||
（threading.Lock，无进程/磁盘副作用），验证 loader 被调用且进入
|
||
run_full_analysis 的代码为 portfolio 持仓。"""
|
||
args = self._make_args(portfolio="futu", stocks="600519")
|
||
config = self._make_config(run_immediately=True)
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "false"}, clear=False), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch("main._refresh_stock_index_cache_for_analysis"), \
|
||
patch(
|
||
"src.brokers.futu.portfolio.load_futu_stock_codes",
|
||
return_value=["AAPL", "HK00700"],
|
||
) as loader, \
|
||
patch(
|
||
"main._compute_trading_day_filter",
|
||
return_value=(["AAPL", "HK00700"], "us,hk", False),
|
||
), \
|
||
patch("src.core.pipeline.StockAnalysisPipeline"):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
loader.assert_called_once_with()
|
||
|
||
def test_actions_portfolio_futu_with_bad_stock_list_reaches_run_full_analysis(self) -> None:
|
||
"""Review 反例补格：`GITHUB_ACTIONS=true` + `--portfolio futu` 时 Actions
|
||
分支的 STOCK_LIST 分类同样整体跳过，坏 watchlist 不拦截 portfolio 覆盖。"""
|
||
args = self._make_args(portfolio="futu")
|
||
config = self._make_config(run_immediately=True)
|
||
config.stock_list = ["930956.CSI"]
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "true"}), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch("main._refresh_stock_index_cache_for_analysis") as refresh, \
|
||
patch("main._classify_stock_list_tokens") as classify, \
|
||
patch(
|
||
"src.brokers.futu.portfolio.load_futu_stock_codes",
|
||
return_value=["AAPL"],
|
||
), \
|
||
patch(
|
||
"main._compute_trading_day_filter",
|
||
return_value=(["AAPL"], "us", False),
|
||
), \
|
||
patch("main._run_analysis_with_runtime_scheduler_lock") as run_with_lock:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
refresh.assert_not_called()
|
||
classify.assert_not_called()
|
||
run_with_lock.assert_called_once_with(config, args, None, None)
|
||
|
||
def test_portfolio_futu_with_bad_stocks_token_runs_pipeline_with_portfolio_codes(self) -> None:
|
||
"""断言到达真实风险层：portfolio 覆盖后 pipeline.run 收到的 stock_codes
|
||
来自 Futu loader，与 `--stocks` 无关。"""
|
||
args = self._make_args(portfolio="futu", stocks="930956.CSI")
|
||
config = self._make_config(run_immediately=True)
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "false"}, clear=False), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch("main._refresh_stock_index_cache_for_analysis"), \
|
||
patch(
|
||
"src.brokers.futu.portfolio.load_futu_stock_codes",
|
||
return_value=["AAPL", "HK00700"],
|
||
), \
|
||
patch(
|
||
"main._compute_trading_day_filter",
|
||
return_value=(["AAPL", "HK00700"], "us,hk", False),
|
||
), \
|
||
patch("src.core.pipeline.StockAnalysisPipeline") as pipeline_class:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
pipeline_run_kwargs = pipeline_class.return_value.run.call_args.kwargs
|
||
self.assertEqual(pipeline_run_kwargs["stock_codes"], ["AAPL", "HK00700"])
|
||
self.assertIsNone(pipeline_run_kwargs["analysis_targets"])
|
||
|
||
def test_schedule_mode_with_bad_stocks_token_reaches_scheduler(self) -> None:
|
||
"""Review 反例：`--schedule --stocks 930956.CSI` 不因坏 token 分类退出，
|
||
进入 scheduler，且保留既有"忽略启动快照"警告。"""
|
||
args = self._make_args(schedule=True, stocks="930956.CSI")
|
||
config = self._make_config(schedule_enabled=False)
|
||
scheduled_call = {}
|
||
|
||
def fake_run_with_schedule(
|
||
task,
|
||
schedule_time,
|
||
run_immediately,
|
||
background_tasks=None,
|
||
schedule_time_provider=None,
|
||
):
|
||
scheduled_call["schedule_time"] = schedule_time
|
||
scheduled_call["run_immediately"] = run_immediately
|
||
scheduled_call["background_tasks"] = background_tasks or []
|
||
scheduled_call["resolved_schedule_time"] = (
|
||
schedule_time_provider() if schedule_time_provider is not None else None
|
||
)
|
||
task()
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "false"}, clear=False), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main._reload_runtime_config", return_value=config), \
|
||
patch("main._build_schedule_time_provider", return_value=lambda: "18:00"), \
|
||
patch("main.setup_logging"), \
|
||
patch("main.run_full_analysis") as run_full_analysis, \
|
||
patch("main.logger.warning") as warning_log, \
|
||
patch("main._refresh_stock_index_cache_for_analysis") as refresh, \
|
||
patch("main._classify_stock_list_tokens") as classify, \
|
||
patch("src.scheduler.run_with_schedule", side_effect=fake_run_with_schedule):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
refresh.assert_not_called()
|
||
classify.assert_not_called()
|
||
self.assertEqual(
|
||
scheduled_call,
|
||
{
|
||
"schedule_time": "18:00",
|
||
"run_immediately": True,
|
||
"background_tasks": [],
|
||
"resolved_schedule_time": "18:00",
|
||
},
|
||
)
|
||
run_full_analysis.assert_called_once_with(config, args, None)
|
||
warning_log.assert_any_call(
|
||
"定时模式下检测到 --stocks 参数；计划执行将忽略启动时股票快照，并在每次运行前重新读取最新的 STOCK_LIST。"
|
||
)
|
||
|
||
def test_serve_only_with_bad_stocks_token_enters_service_loop(self) -> None:
|
||
"""Review 反例：`--serve-only --stocks 930956.CSI` 不因坏 token 分类退出，
|
||
进入仅服务模式循环（time.sleep 以 KeyboardInterrupt 收尾）。"""
|
||
args = self._make_args(serve_only=True, stocks="930956.CSI")
|
||
config = self._make_config(webui_enabled=False)
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "false"}, clear=False), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.prepare_webui_frontend_assets", return_value=True), \
|
||
patch("main.start_api_server"), \
|
||
patch("main.start_bot_stream_clients"), \
|
||
patch("main._refresh_stock_index_cache_for_analysis") as refresh, \
|
||
patch("main._classify_stock_list_tokens") as classify, \
|
||
patch("main.time.sleep", side_effect=KeyboardInterrupt):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
refresh.assert_not_called()
|
||
classify.assert_not_called()
|
||
|
||
def test_standalone_run_actions_stock_list_rejects_unsupported(self) -> None:
|
||
"""Actions STOCK_LIST 携带未登记 `.CSI` token 时与 `--stocks` 同样
|
||
在入口明确拒绝并返回非零，不会静默丢弃该 token 后继续运行。"""
|
||
args = self._make_args()
|
||
config = self._make_config(run_immediately=True)
|
||
config.stock_list = ["sh000016", "930956.CSI"]
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "true"}), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch("main._refresh_stock_index_cache_for_analysis"), \
|
||
patch("main.run_full_analysis") as run_full_analysis:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 1)
|
||
run_full_analysis.assert_not_called()
|
||
|
||
def test_standalone_run_actions_market_only_skips_token_classification(self) -> None:
|
||
"""Actions 的 market-only 模式（`--market-review`）不经个股列表分析，
|
||
STOCK_LIST 分类入口必须整体跳过（含 unsupported 拒绝），保持既有
|
||
大盘复盘行为。"""
|
||
args = self._make_args(market_review=True)
|
||
config = self._make_config(run_immediately=True)
|
||
config.stock_list = ["930956.CSI"]
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "true"}), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch("main._refresh_stock_index_cache_for_analysis") as refresh, \
|
||
patch("main.run_full_analysis") as run_full_analysis, \
|
||
patch("main._run_market_review_with_shared_lock") as run_review, \
|
||
patch("src.core.trading_calendar.get_open_markets_today", return_value={"cn"}), \
|
||
patch("src.core.trading_calendar.compute_effective_region", return_value="cn"):
|
||
run_review.return_value = object()
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
refresh.assert_not_called()
|
||
run_full_analysis.assert_not_called()
|
||
run_review.assert_called_once()
|
||
|
||
def test_standalone_run_returns_nonzero_when_startup_analysis_reports_failure(self) -> None:
|
||
args = self._make_args()
|
||
config = self._make_config(run_immediately=True)
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch.object(main, "_LAST_ANALYSIS_FAILURE_REASON", "no_report"), \
|
||
patch("main._run_analysis_with_runtime_scheduler_lock", return_value=False) as run_with_lock:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 1)
|
||
run_with_lock.assert_called_once_with(config, args, None, None)
|
||
|
||
def test_standalone_futu_portfolio_failure_returns_nonzero(self) -> None:
|
||
args = self._make_args(portfolio="futu")
|
||
config = self._make_config(run_immediately=True)
|
||
error = FutuPortfolioError("OpenD unavailable")
|
||
|
||
with (
|
||
patch("main.parse_arguments", return_value=args),
|
||
patch("main.get_config", return_value=config),
|
||
patch("main.setup_logging"),
|
||
patch("main._refresh_stock_index_cache_for_analysis"),
|
||
patch(
|
||
"src.brokers.futu.portfolio.load_futu_stock_codes",
|
||
side_effect=error,
|
||
) as loader,
|
||
):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 1)
|
||
loader.assert_called_once_with()
|
||
|
||
def test_standalone_futu_portfolio_success_returns_zero(self) -> None:
|
||
args = self._make_args(portfolio="futu")
|
||
config = self._make_config(run_immediately=True)
|
||
|
||
with (
|
||
patch("main.parse_arguments", return_value=args),
|
||
patch("main.get_config", return_value=config),
|
||
patch("main.setup_logging"),
|
||
patch("main._refresh_stock_index_cache_for_analysis"),
|
||
patch(
|
||
"src.brokers.futu.portfolio.load_futu_stock_codes",
|
||
return_value=["AAPL"],
|
||
) as loader,
|
||
patch(
|
||
"main._compute_trading_day_filter",
|
||
return_value=([], "", True),
|
||
),
|
||
):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
loader.assert_called_once_with()
|
||
|
||
def test_standalone_futu_downstream_failure_keeps_existing_exit_semantics(self) -> None:
|
||
args = self._make_args(portfolio="futu")
|
||
config = self._make_config(run_immediately=True)
|
||
|
||
with (
|
||
patch("main.parse_arguments", return_value=args),
|
||
patch("main.get_config", return_value=config),
|
||
patch("main.setup_logging"),
|
||
patch("main._refresh_stock_index_cache_for_analysis"),
|
||
patch(
|
||
"src.brokers.futu.portfolio.load_futu_stock_codes",
|
||
return_value=["AAPL"],
|
||
) as loader,
|
||
patch(
|
||
"main._compute_trading_day_filter",
|
||
side_effect=RuntimeError("calendar unavailable"),
|
||
),
|
||
):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
loader.assert_called_once_with()
|
||
|
||
def test_schedule_mode_reload_uses_latest_runtime_config(self) -> None:
|
||
args = self._make_args(schedule=True)
|
||
startup_config = self._make_config(schedule_enabled=True, schedule_time="18:00")
|
||
runtime_config = self._make_config(schedule_enabled=True, schedule_time="09:30")
|
||
scheduled_call = {}
|
||
|
||
def fake_run_with_schedule(
|
||
task,
|
||
schedule_time,
|
||
run_immediately,
|
||
background_tasks=None,
|
||
schedule_time_provider=None,
|
||
):
|
||
scheduled_call["schedule_time"] = schedule_time
|
||
scheduled_call["resolved_schedule_time"] = (
|
||
schedule_time_provider() if schedule_time_provider is not None else None
|
||
)
|
||
task()
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=startup_config), \
|
||
patch("main._reload_runtime_config", return_value=runtime_config), \
|
||
patch("main._build_schedule_time_provider", return_value=lambda: "09:30"), \
|
||
patch("main.setup_logging"), \
|
||
patch("main.run_full_analysis") as run_full_analysis, \
|
||
patch("src.scheduler.run_with_schedule", side_effect=fake_run_with_schedule):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
self.assertEqual(
|
||
scheduled_call,
|
||
{"schedule_time": "18:00", "resolved_schedule_time": "09:30"},
|
||
)
|
||
run_full_analysis.assert_called_once_with(runtime_config, args, None)
|
||
|
||
def test_schedule_mode_raises_task_failure_when_analysis_returns_false(self) -> None:
|
||
args = self._make_args(schedule=True)
|
||
runtime_config = self._make_config(schedule_enabled=True, schedule_time="09:30")
|
||
scheduled_call = {}
|
||
|
||
def fake_run_with_schedule(
|
||
task,
|
||
schedule_time,
|
||
run_immediately,
|
||
background_tasks=None,
|
||
schedule_time_provider=None,
|
||
):
|
||
scheduled_call["task"] = task
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=self._make_config(schedule_enabled=True, schedule_time="18:00")), \
|
||
patch("main._reload_runtime_config", return_value=runtime_config), \
|
||
patch("main._build_schedule_time_provider", return_value=lambda: "09:30"), \
|
||
patch("main.setup_logging"), \
|
||
patch("main.run_full_analysis", return_value=False) as run_full_analysis, \
|
||
patch.object(main, "_LAST_ANALYSIS_FAILURE_REASON", "no_report"), \
|
||
patch("src.scheduler.run_with_schedule", side_effect=fake_run_with_schedule):
|
||
exit_code = main.main()
|
||
with self.assertRaisesRegex(RuntimeError, "scheduled analysis reported failure: no_report"):
|
||
scheduled_call["task"]()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
run_full_analysis.assert_called_once_with(runtime_config, args, None)
|
||
|
||
def test_schedule_mode_registers_event_monitor_background_task(self) -> None:
|
||
args = self._make_args(schedule=True)
|
||
config = self._make_config(
|
||
schedule_enabled=False,
|
||
agent_event_monitor_enabled=True,
|
||
agent_event_monitor_interval_minutes=7,
|
||
)
|
||
worker = MagicMock()
|
||
worker.run_once.return_value = {"triggered": 2}
|
||
scheduled_call = {}
|
||
|
||
def fake_run_with_schedule(
|
||
task,
|
||
schedule_time,
|
||
run_immediately,
|
||
background_tasks=None,
|
||
schedule_time_provider=None,
|
||
):
|
||
scheduled_call["schedule_time"] = schedule_time
|
||
scheduled_call["run_immediately"] = run_immediately
|
||
scheduled_call["background_tasks"] = background_tasks or []
|
||
scheduled_call["resolved_schedule_time"] = (
|
||
schedule_time_provider() if schedule_time_provider is not None else None
|
||
)
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main._reload_runtime_config", return_value=config) as reload_config, \
|
||
patch("main._build_schedule_time_provider", return_value=lambda: "18:00"), \
|
||
patch("main.setup_logging"), \
|
||
patch("main.run_full_analysis") as run_full_analysis, \
|
||
patch("src.services.alert_worker.AlertWorker", return_value=worker) as worker_cls, \
|
||
patch("src.scheduler.run_with_schedule", side_effect=fake_run_with_schedule):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
worker_cls.assert_called_once()
|
||
self.assertIs(worker_cls.call_args.kwargs["config_provider"], reload_config)
|
||
run_full_analysis.assert_not_called()
|
||
self.assertEqual(scheduled_call["schedule_time"], "18:00")
|
||
self.assertEqual(scheduled_call["run_immediately"], True)
|
||
self.assertEqual(scheduled_call["resolved_schedule_time"], "18:00")
|
||
self.assertEqual(len(scheduled_call["background_tasks"]), 1)
|
||
background_task = scheduled_call["background_tasks"][0]
|
||
self.assertEqual(background_task["name"], "agent_event_monitor")
|
||
self.assertEqual(background_task["interval_seconds"], 7 * 60)
|
||
self.assertEqual(background_task["run_immediately"], True)
|
||
|
||
with patch("main.logger.info") as info_log:
|
||
background_task["task"]()
|
||
|
||
worker.run_once.assert_called_once_with()
|
||
info_log.assert_any_call("[EventMonitor] 本轮触发 %d 条提醒", 2)
|
||
|
||
def test_schedule_mode_registers_event_monitor_worker_without_legacy_rules(self) -> None:
|
||
args = self._make_args(schedule=True)
|
||
config = self._make_config(
|
||
schedule_enabled=False,
|
||
agent_event_monitor_enabled=True,
|
||
agent_event_alert_rules_json="",
|
||
)
|
||
worker = MagicMock()
|
||
worker.run_once.return_value = {"triggered": 0}
|
||
scheduled_call = {}
|
||
|
||
def fake_run_with_schedule(
|
||
task,
|
||
schedule_time,
|
||
run_immediately,
|
||
background_tasks=None,
|
||
schedule_time_provider=None,
|
||
):
|
||
scheduled_call["background_tasks"] = background_tasks or []
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main._reload_runtime_config", return_value=config), \
|
||
patch("main._build_schedule_time_provider", return_value=lambda: "18:00"), \
|
||
patch("main.setup_logging"), \
|
||
patch("main.run_full_analysis") as run_full_analysis, \
|
||
patch("src.services.alert_worker.AlertWorker", return_value=worker) as worker_cls, \
|
||
patch("src.scheduler.run_with_schedule", side_effect=fake_run_with_schedule):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
worker_cls.assert_called_once()
|
||
run_full_analysis.assert_not_called()
|
||
self.assertEqual(len(scheduled_call["background_tasks"]), 1)
|
||
self.assertEqual(scheduled_call["background_tasks"][0]["name"], "agent_event_monitor")
|
||
|
||
def test_check_notify_returns_before_other_modes(self) -> None:
|
||
args = self._make_args(check_notify=True, serve=True, schedule=True, market_review=True)
|
||
config = self._make_config(webui_enabled=False)
|
||
diagnostic_result = SimpleNamespace(ok=True)
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch("main.start_api_server") as start_api_server, \
|
||
patch("main.run_full_analysis") as run_full_analysis, \
|
||
patch(
|
||
"src.services.notification_diagnostics.run_notification_diagnostics",
|
||
return_value=diagnostic_result,
|
||
) as run_diagnostics, \
|
||
patch(
|
||
"src.services.notification_diagnostics.format_notification_diagnostics",
|
||
return_value="通知配置诊断",
|
||
), \
|
||
patch("builtins.print") as print_output:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
run_diagnostics.assert_called_once_with(config)
|
||
print_output.assert_called_once_with("通知配置诊断")
|
||
start_api_server.assert_not_called()
|
||
run_full_analysis.assert_not_called()
|
||
|
||
def test_serve_mode_exits_when_api_server_start_fails(self) -> None:
|
||
args = self._make_args(serve_only=True, host="127.0.0.1", port=8000)
|
||
config = self._make_config(webui_enabled=False)
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "false"}, clear=False), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.prepare_webui_frontend_assets", return_value=True), \
|
||
patch("main.start_api_server", side_effect=RuntimeError("port busy")), \
|
||
patch("main.start_bot_stream_clients") as start_bots, \
|
||
patch("main.logger.error") as error_log:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 1)
|
||
start_bots.assert_not_called()
|
||
error_log.assert_called_once()
|
||
|
||
def test_webui_only_maps_to_serve_only_and_exits_when_api_server_start_fails(self) -> None:
|
||
args = self._make_args(webui_only=True, host="127.0.0.1", port=8000)
|
||
config = self._make_config(webui_enabled=False)
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "false"}, clear=False), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.prepare_webui_frontend_assets", return_value=True), \
|
||
patch("main.start_api_server", side_effect=RuntimeError("port busy")), \
|
||
patch("main.start_bot_stream_clients") as start_bots, \
|
||
patch("main.run_full_analysis") as run_full_analysis, \
|
||
patch("main.logger.error") as error_log:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 1)
|
||
start_bots.assert_not_called()
|
||
run_full_analysis.assert_not_called()
|
||
error_log.assert_called_once()
|
||
|
||
def test_serve_mode_continues_single_analysis_when_api_server_start_fails(self) -> None:
|
||
args = self._make_args(serve=True, host="127.0.0.1", port=8000)
|
||
config = self._make_config(webui_enabled=False, run_immediately=True)
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "false"}, clear=False), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.prepare_webui_frontend_assets", return_value=True), \
|
||
patch("main.start_api_server", side_effect=RuntimeError("port busy")), \
|
||
patch("main.start_bot_stream_clients") as start_bots, \
|
||
patch("main.run_full_analysis") as run_full_analysis, \
|
||
patch("main._run_analysis_with_runtime_scheduler_lock") as run_with_lock, \
|
||
patch("main.logger.error") as error_log:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
start_bots.assert_not_called()
|
||
run_with_lock.assert_called_once_with(config, args, None, None)
|
||
run_full_analysis.assert_not_called()
|
||
error_log.assert_called_once()
|
||
|
||
def test_serve_schedule_mode_continues_scheduler_when_api_server_start_fails(self) -> None:
|
||
args = self._make_args(serve=True, schedule=True, host="127.0.0.1", port=8000)
|
||
config = self._make_config(webui_enabled=False, schedule_enabled=False)
|
||
scheduled_call = {}
|
||
|
||
def fake_run_with_schedule(
|
||
task,
|
||
schedule_time,
|
||
run_immediately,
|
||
background_tasks=None,
|
||
schedule_time_provider=None,
|
||
):
|
||
scheduled_call["schedule_time"] = schedule_time
|
||
scheduled_call["run_immediately"] = run_immediately
|
||
scheduled_call["background_tasks"] = background_tasks or []
|
||
task()
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "false"}, clear=False), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main._reload_runtime_config", return_value=config), \
|
||
patch("main._build_schedule_time_provider", return_value=lambda: "18:00"), \
|
||
patch("main.prepare_webui_frontend_assets", return_value=True), \
|
||
patch("main.start_api_server", side_effect=RuntimeError("port busy")), \
|
||
patch("main.start_bot_stream_clients") as start_bots, \
|
||
patch("main.run_full_analysis") as run_full_analysis, \
|
||
patch("src.scheduler.run_with_schedule", side_effect=fake_run_with_schedule), \
|
||
patch("main.logger.error") as error_log:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
start_bots.assert_not_called()
|
||
run_full_analysis.assert_called_once_with(config, args, None)
|
||
self.assertEqual(scheduled_call["schedule_time"], "18:00")
|
||
self.assertEqual(scheduled_call["run_immediately"], True)
|
||
self.assertEqual(scheduled_call["background_tasks"], [])
|
||
error_log.assert_called_once()
|
||
|
||
def test_serve_with_enabled_schedule_uses_api_runtime_scheduler(self) -> None:
|
||
from src.services.runtime_scheduler import (
|
||
CLI_SCHEDULER_OWNER_ENV,
|
||
RUNTIME_SCHEDULER_ARGS_ENV,
|
||
RUNTIME_SCHEDULER_RUN_IMMEDIATELY_ENV,
|
||
)
|
||
|
||
args = self._make_args(
|
||
serve=True,
|
||
schedule=False,
|
||
host="127.0.0.1",
|
||
port=8000,
|
||
no_notify=True,
|
||
no_market_review=True,
|
||
dry_run=True,
|
||
force_run=True,
|
||
single_notify=True,
|
||
no_context_snapshot=True,
|
||
workers=4,
|
||
)
|
||
config = self._make_config(webui_enabled=False, schedule_enabled=True)
|
||
marker_seen_by_server = []
|
||
run_immediately_seen_by_server = []
|
||
runtime_args_seen_by_server = []
|
||
|
||
def fake_start_api_server(host, port, config):
|
||
marker_seen_by_server.append(os.getenv(CLI_SCHEDULER_OWNER_ENV))
|
||
run_immediately_seen_by_server.append(os.getenv(RUNTIME_SCHEDULER_RUN_IMMEDIATELY_ENV))
|
||
runtime_args_seen_by_server.append(json.loads(os.getenv(RUNTIME_SCHEDULER_ARGS_ENV, "{}")))
|
||
|
||
with patch.dict(
|
||
os.environ,
|
||
{"GITHUB_ACTIONS": "false", CLI_SCHEDULER_OWNER_ENV: "true"},
|
||
clear=False,
|
||
), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.prepare_webui_frontend_assets", return_value=True), \
|
||
patch("main.start_api_server", side_effect=fake_start_api_server), \
|
||
patch("main.start_bot_stream_clients") as start_bots, \
|
||
patch("main.time.sleep", side_effect=KeyboardInterrupt), \
|
||
patch("src.scheduler.run_with_schedule") as run_with_schedule:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
self.assertEqual(marker_seen_by_server, [None])
|
||
self.assertEqual(run_immediately_seen_by_server, ["true"])
|
||
self.assertEqual(runtime_args_seen_by_server, [{
|
||
"no_notify": True,
|
||
"no_market_review": True,
|
||
"dry_run": True,
|
||
"force_run": True,
|
||
"single_notify": True,
|
||
"no_context_snapshot": True,
|
||
"workers": 4,
|
||
}])
|
||
start_bots.assert_called_once_with(config)
|
||
run_with_schedule.assert_not_called()
|
||
|
||
def test_serve_mode_uses_shared_analysis_lock_for_immediate_run_full_analysis(self) -> None:
|
||
args = self._make_args(
|
||
serve=True,
|
||
schedule=False,
|
||
portfolio="futu",
|
||
host="127.0.0.1",
|
||
port=8000,
|
||
)
|
||
config = self._make_config(webui_enabled=False, run_immediately=True)
|
||
|
||
with (
|
||
patch.dict(os.environ, {"GITHUB_ACTIONS": "false"}, clear=False),
|
||
patch("main.parse_arguments", return_value=args),
|
||
patch("main.get_config", return_value=config),
|
||
patch("main.prepare_webui_frontend_assets", return_value=True),
|
||
patch("main.start_api_server"),
|
||
patch("main.start_bot_stream_clients") as start_bots,
|
||
patch("main.time.sleep", side_effect=KeyboardInterrupt),
|
||
patch("main.run_full_analysis") as run_full_analysis,
|
||
patch("main._run_analysis_with_runtime_scheduler_lock") as run_with_lock,
|
||
):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
self.assertEqual(run_with_lock.call_count, 1)
|
||
run_with_lock.assert_called_once_with(config, args, None, None)
|
||
run_full_analysis.assert_not_called()
|
||
start_bots.assert_called_once_with(config)
|
||
|
||
def test_serve_mode_keeps_running_after_futu_portfolio_load_failure(self) -> None:
|
||
args = self._make_args(
|
||
serve=True,
|
||
schedule=False,
|
||
portfolio="futu",
|
||
host="127.0.0.1",
|
||
port=8000,
|
||
)
|
||
config = self._make_config(webui_enabled=False, run_immediately=True)
|
||
error = FutuPortfolioError("OpenD unavailable")
|
||
|
||
with (
|
||
patch.dict(os.environ, {"GITHUB_ACTIONS": "false"}, clear=False),
|
||
patch("main.parse_arguments", return_value=args),
|
||
patch("main.get_config", return_value=config),
|
||
patch("main.prepare_webui_frontend_assets", return_value=True),
|
||
patch("main.start_api_server"),
|
||
patch("main.start_bot_stream_clients") as start_bots,
|
||
patch("main.time.sleep", side_effect=KeyboardInterrupt),
|
||
patch(
|
||
"main._run_analysis_with_runtime_scheduler_lock",
|
||
side_effect=error,
|
||
) as run_with_lock,
|
||
patch("main.logger.exception") as exception_log,
|
||
):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
run_with_lock.assert_called_once_with(config, args, None, None)
|
||
start_bots.assert_called_once_with(config)
|
||
exception_log.assert_any_call(
|
||
"Futu 持仓导入失败，Web/API 服务继续运行: %s",
|
||
error,
|
||
)
|
||
|
||
def test_serve_schedule_flag_enables_api_runtime_scheduler(self) -> None:
|
||
from src.services.runtime_scheduler import (
|
||
CLI_SCHEDULER_OWNER_ENV,
|
||
RUNTIME_SCHEDULER_ARGS_ENV,
|
||
RUNTIME_SCHEDULER_FORCE_ENABLED_ENV,
|
||
RUNTIME_SCHEDULER_RUN_IMMEDIATELY_ENV,
|
||
)
|
||
|
||
args = self._make_args(
|
||
serve=True,
|
||
schedule=True,
|
||
host="127.0.0.1",
|
||
port=8000,
|
||
no_notify=True,
|
||
no_market_review=True,
|
||
dry_run=True,
|
||
force_run=True,
|
||
single_notify=True,
|
||
no_context_snapshot=True,
|
||
workers=4,
|
||
)
|
||
config = self._make_config(webui_enabled=False, schedule_enabled=False)
|
||
marker_seen_by_server = []
|
||
force_enabled_seen_by_server = []
|
||
run_immediately_seen_by_server = []
|
||
runtime_args_seen_by_server = []
|
||
|
||
def fake_start_api_server(host, port, config):
|
||
marker_seen_by_server.append(os.getenv(CLI_SCHEDULER_OWNER_ENV))
|
||
force_enabled_seen_by_server.append(os.getenv(RUNTIME_SCHEDULER_FORCE_ENABLED_ENV))
|
||
run_immediately_seen_by_server.append(os.getenv(RUNTIME_SCHEDULER_RUN_IMMEDIATELY_ENV))
|
||
runtime_args_seen_by_server.append(json.loads(os.getenv(RUNTIME_SCHEDULER_ARGS_ENV, "{}")))
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "false"}, clear=False), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.prepare_webui_frontend_assets", return_value=True), \
|
||
patch("main.start_api_server", side_effect=fake_start_api_server), \
|
||
patch("main.start_bot_stream_clients"), \
|
||
patch("main.time.sleep", side_effect=KeyboardInterrupt), \
|
||
patch("src.scheduler.run_with_schedule") as run_with_schedule:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
self.assertEqual(marker_seen_by_server, [None])
|
||
self.assertEqual(force_enabled_seen_by_server, ["true"])
|
||
self.assertEqual(run_immediately_seen_by_server, ["true"])
|
||
self.assertEqual(runtime_args_seen_by_server, [{
|
||
"no_notify": True,
|
||
"no_market_review": True,
|
||
"dry_run": True,
|
||
"force_run": True,
|
||
"single_notify": True,
|
||
"no_context_snapshot": True,
|
||
"workers": 4,
|
||
}])
|
||
self.assertFalse(config.schedule_enabled)
|
||
run_with_schedule.assert_not_called()
|
||
|
||
def test_serve_schedule_flag_passes_no_run_immediately_to_runtime_scheduler(self) -> None:
|
||
from src.services.runtime_scheduler import RUNTIME_SCHEDULER_RUN_IMMEDIATELY_ENV
|
||
|
||
args = self._make_args(
|
||
serve=True,
|
||
schedule=True,
|
||
no_run_immediately=True,
|
||
host="127.0.0.1",
|
||
port=8000,
|
||
)
|
||
config = self._make_config(webui_enabled=False, schedule_enabled=False)
|
||
run_immediately_seen_by_server = []
|
||
|
||
def fake_start_api_server(host, port, config):
|
||
run_immediately_seen_by_server.append(os.getenv(RUNTIME_SCHEDULER_RUN_IMMEDIATELY_ENV))
|
||
|
||
with patch.dict(os.environ, {"GITHUB_ACTIONS": "false"}, clear=False), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.prepare_webui_frontend_assets", return_value=True), \
|
||
patch("main.start_api_server", side_effect=fake_start_api_server), \
|
||
patch("main.start_bot_stream_clients"), \
|
||
patch("main.time.sleep", side_effect=KeyboardInterrupt), \
|
||
patch("src.scheduler.run_with_schedule") as run_with_schedule:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
self.assertEqual(run_immediately_seen_by_server, ["false"])
|
||
run_with_schedule.assert_not_called()
|
||
|
||
def test_serve_only_restores_persisted_scheduler_without_running_immediately(self) -> None:
|
||
from src.services.runtime_scheduler import (
|
||
CLI_SCHEDULER_OWNER_ENV,
|
||
RUNTIME_SCHEDULER_RUN_IMMEDIATELY_ENV,
|
||
RUNTIME_SCHEDULER_SUPPRESS_START_ENV,
|
||
)
|
||
|
||
args = self._make_args(serve_only=True, host="127.0.0.1", port=8000)
|
||
config = self._make_config(webui_enabled=False, schedule_enabled=True)
|
||
marker_seen_by_server = []
|
||
suppress_seen_by_server = []
|
||
run_immediately_seen_by_server = []
|
||
|
||
def fake_start_api_server(host, port, config):
|
||
marker_seen_by_server.append(os.getenv(CLI_SCHEDULER_OWNER_ENV))
|
||
suppress_seen_by_server.append(os.getenv(RUNTIME_SCHEDULER_SUPPRESS_START_ENV))
|
||
run_immediately_seen_by_server.append(os.getenv(RUNTIME_SCHEDULER_RUN_IMMEDIATELY_ENV))
|
||
|
||
with patch.dict(
|
||
os.environ,
|
||
{"GITHUB_ACTIONS": "false", CLI_SCHEDULER_OWNER_ENV: "true"},
|
||
clear=False,
|
||
), \
|
||
patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.prepare_webui_frontend_assets", return_value=True), \
|
||
patch("main.start_api_server", side_effect=fake_start_api_server), \
|
||
patch("main.start_bot_stream_clients") as start_bots, \
|
||
patch("main.time.sleep", side_effect=KeyboardInterrupt), \
|
||
patch("src.scheduler.run_with_schedule") as run_with_schedule:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
self.assertEqual(marker_seen_by_server, [None])
|
||
self.assertEqual(suppress_seen_by_server, [None])
|
||
self.assertEqual(run_immediately_seen_by_server, ["false"])
|
||
start_bots.assert_called_once_with(config)
|
||
run_with_schedule.assert_not_called()
|
||
|
||
def test_reload_runtime_config_preserves_process_env_overrides(self) -> None:
|
||
self.env_path.write_text(
|
||
"OPENAI_API_KEY=stale-file\nSCHEDULE_TIME=09:30\n",
|
||
encoding="utf-8",
|
||
)
|
||
runtime_config = self._make_config(schedule_enabled=True, schedule_time="09:30")
|
||
|
||
with patch.dict(
|
||
os.environ,
|
||
{
|
||
"ENV_FILE": str(self.env_path),
|
||
"OPENAI_API_KEY": "runtime-secret",
|
||
"SCHEDULE_TIME": "18:00",
|
||
},
|
||
clear=False,
|
||
), patch.object(
|
||
main,
|
||
"_INITIAL_PROCESS_ENV",
|
||
{"OPENAI_API_KEY": "runtime-secret"},
|
||
), patch.object(
|
||
main,
|
||
"_RUNTIME_ENV_FILE_KEYS",
|
||
{"SCHEDULE_TIME"},
|
||
), patch(
|
||
"main.get_config",
|
||
return_value=runtime_config,
|
||
) as get_config_mock:
|
||
reloaded_config = main._reload_runtime_config()
|
||
self.assertEqual(os.environ["OPENAI_API_KEY"], "runtime-secret")
|
||
self.assertEqual(os.environ["SCHEDULE_TIME"], "09:30")
|
||
|
||
self.assertIs(reloaded_config, runtime_config)
|
||
get_config_mock.assert_called_once_with()
|
||
|
||
def test_reload_env_file_values_preserves_managed_env_vars_when_read_fails(self) -> None:
|
||
with patch.dict(
|
||
os.environ,
|
||
{
|
||
"ENV_FILE": str(self.env_path),
|
||
"OPENAI_API_KEY": "runtime-secret",
|
||
"SCHEDULE_TIME": "09:30",
|
||
},
|
||
clear=False,
|
||
), patch.object(
|
||
main,
|
||
"_INITIAL_PROCESS_ENV",
|
||
{},
|
||
), patch.object(
|
||
main,
|
||
"_RUNTIME_ENV_FILE_KEYS",
|
||
{"OPENAI_API_KEY", "SCHEDULE_TIME"},
|
||
), patch(
|
||
"main.dotenv_values",
|
||
side_effect=OSError("boom"),
|
||
):
|
||
main._reload_env_file_values_preserving_overrides()
|
||
|
||
self.assertEqual(os.environ["OPENAI_API_KEY"], "runtime-secret")
|
||
self.assertEqual(os.environ["SCHEDULE_TIME"], "09:30")
|
||
self.assertEqual(
|
||
main._RUNTIME_ENV_FILE_KEYS,
|
||
{"OPENAI_API_KEY", "SCHEDULE_TIME"},
|
||
)
|
||
|
||
def test_reload_runtime_config_refreshes_env_before_resetting_singleton(self) -> None:
|
||
runtime_config = self._make_config(schedule_enabled=True, schedule_time="09:30")
|
||
call_order = []
|
||
|
||
def fake_reload_env() -> None:
|
||
call_order.append("reload_env")
|
||
|
||
def fake_reset_instance() -> None:
|
||
call_order.append("reset_instance")
|
||
|
||
def fake_get_config():
|
||
call_order.append("get_config")
|
||
return runtime_config
|
||
|
||
with patch(
|
||
"main._reload_env_file_values_preserving_overrides",
|
||
side_effect=fake_reload_env,
|
||
), patch(
|
||
"main.Config.reset_instance",
|
||
side_effect=fake_reset_instance,
|
||
), patch(
|
||
"main.get_config",
|
||
side_effect=fake_get_config,
|
||
):
|
||
reloaded_config = main._reload_runtime_config()
|
||
|
||
self.assertIs(reloaded_config, runtime_config)
|
||
self.assertEqual(call_order, ["reload_env", "reset_instance", "get_config"])
|
||
|
||
def test_schedule_time_provider_propagates_config_read_failures(self) -> None:
|
||
with patch.object(
|
||
main,
|
||
"_INITIAL_PROCESS_ENV",
|
||
{},
|
||
), patch(
|
||
"src.core.config_manager.ConfigManager.read_config_map",
|
||
side_effect=RuntimeError("boom"),
|
||
):
|
||
provider = main._build_schedule_time_provider("18:00")
|
||
|
||
with self.assertRaisesRegex(RuntimeError, "boom"):
|
||
provider()
|
||
|
||
def test_schedule_time_provider_respects_process_env_precedence(self) -> None:
|
||
with patch.dict(
|
||
os.environ,
|
||
{"SCHEDULE_TIME": "18:00"},
|
||
clear=False,
|
||
), patch.object(
|
||
main,
|
||
"_INITIAL_PROCESS_ENV",
|
||
{"SCHEDULE_TIME": "18:00"},
|
||
), patch(
|
||
"src.core.config_manager.ConfigManager.read_config_map",
|
||
side_effect=AssertionError("should not read .env when process env override exists"),
|
||
):
|
||
provider = main._build_schedule_time_provider("09:30")
|
||
|
||
self.assertEqual(provider(), "18:00")
|
||
|
||
def test_schedule_time_provider_falls_back_to_system_default_on_clear(self) -> None:
|
||
"""When SCHEDULE_TIME is cleared/removed from config, provider returns '18:00'."""
|
||
with patch.dict(
|
||
os.environ,
|
||
{"SCHEDULE_TIME": "09:30"},
|
||
clear=False,
|
||
), patch.object(
|
||
main,
|
||
"_INITIAL_PROCESS_ENV",
|
||
{},
|
||
), patch(
|
||
"src.core.config_manager.ConfigManager.read_config_map",
|
||
return_value={},
|
||
):
|
||
provider = main._build_schedule_time_provider("09:30")
|
||
self.assertEqual(provider(), "18:00")
|
||
|
||
def test_schedule_time_provider_falls_back_to_system_default_on_empty(self) -> None:
|
||
"""When SCHEDULE_TIME is empty string in config, provider returns '18:00'."""
|
||
with patch.dict(
|
||
os.environ,
|
||
{"SCHEDULE_TIME": "09:30"},
|
||
clear=False,
|
||
), patch.object(
|
||
main,
|
||
"_INITIAL_PROCESS_ENV",
|
||
{},
|
||
), patch(
|
||
"src.core.config_manager.ConfigManager.read_config_map",
|
||
return_value={"SCHEDULE_TIME": " "},
|
||
):
|
||
provider = main._build_schedule_time_provider("09:30")
|
||
self.assertEqual(provider(), "18:00")
|
||
|
||
def test_single_run_keeps_cli_stock_override(self) -> None:
|
||
args = self._make_args(stocks="600519,000001")
|
||
config = self._make_config(run_immediately=True)
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch("main._refresh_stock_index_cache_for_analysis"), \
|
||
patch("main.run_full_analysis") as run_full_analysis:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
run_full_analysis.assert_called_once()
|
||
_, _, stock_codes = run_full_analysis.call_args.args
|
||
analysis_targets = run_full_analysis.call_args.kwargs.get("analysis_targets")
|
||
self.assertEqual(stock_codes, ["600519", "000001"])
|
||
self.assertEqual(len(analysis_targets), 2)
|
||
self.assertEqual(analysis_targets[0].asset_type, "stock")
|
||
self.assertEqual(analysis_targets[1].asset_type, "stock")
|
||
|
||
def test_run_full_analysis_skips_market_review_when_shared_lock_is_held(self) -> None:
|
||
from src.core.market_review_lock import (
|
||
release_market_review_lock,
|
||
try_acquire_market_review_lock,
|
||
)
|
||
|
||
args = self._make_args()
|
||
config = self._make_config(
|
||
trading_day_check_enabled=False,
|
||
market_review_enabled=True,
|
||
daily_market_context_enabled=True,
|
||
no_market_review=False,
|
||
single_stock_notify=False,
|
||
merge_email_notification=False,
|
||
analysis_delay=0,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
pipeline = MagicMock()
|
||
pipeline.run.return_value = []
|
||
events = []
|
||
pipeline_kwargs = {}
|
||
|
||
def refresh_index(config_arg):
|
||
events.append("refresh")
|
||
|
||
def build_pipeline(*args, **kwargs):
|
||
events.append("pipeline")
|
||
pipeline_kwargs.update(kwargs)
|
||
return pipeline
|
||
|
||
lock_token = try_acquire_market_review_lock(config)
|
||
self.assertIsNotNone(lock_token)
|
||
try:
|
||
with patch.object(main, "_refresh_stock_index_cache_for_analysis", side_effect=refresh_index) as refresh, \
|
||
patch("src.core.pipeline.StockAnalysisPipeline", side_effect=build_pipeline), \
|
||
patch("src.core.market_review.run_market_review") as run_market_review:
|
||
main.run_full_analysis(config, args, [])
|
||
finally:
|
||
release_market_review_lock(lock_token)
|
||
|
||
refresh.assert_called_once_with(config)
|
||
self.assertEqual(events[:2], ["refresh", "pipeline"])
|
||
self.assertTrue(pipeline_kwargs["daily_market_context_allow_generate"])
|
||
pipeline.run.assert_called_once()
|
||
run_market_review.assert_not_called()
|
||
|
||
def test_run_full_analysis_disables_generation_when_no_market_review_flag_set(self) -> None:
|
||
args = self._make_args(no_market_review=True)
|
||
config = self._make_config(
|
||
trading_day_check_enabled=False,
|
||
market_review_enabled=True,
|
||
daily_market_context_enabled=True,
|
||
single_stock_notify=False,
|
||
merge_email_notification=False,
|
||
analysis_delay=0,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
pipeline = MagicMock()
|
||
pipeline.run.return_value = []
|
||
pipeline_kwargs = {}
|
||
|
||
def build_pipeline(*args, **kwargs):
|
||
pipeline_kwargs.update(kwargs)
|
||
return pipeline
|
||
|
||
with patch.object(main, "_refresh_stock_index_cache_for_analysis") as refresh, \
|
||
patch("main._compute_trading_day_filter", return_value=([], "cn", False)), \
|
||
patch("src.core.pipeline.StockAnalysisPipeline", side_effect=build_pipeline), \
|
||
patch("main._prime_daily_market_context") as prime_context, \
|
||
patch("src.core.market_review.run_market_review") as run_market_review:
|
||
main.run_full_analysis(config, args, [])
|
||
|
||
self.assertFalse(pipeline_kwargs["daily_market_context_allow_generate"])
|
||
self.assertEqual(pipeline_kwargs["daily_market_context_enabled"], False)
|
||
prime_context.assert_not_called()
|
||
run_market_review.assert_not_called()
|
||
refresh.assert_called_once_with(config)
|
||
|
||
def test_run_full_analysis_defaults_daily_context_on_without_disabling_market_review(self) -> None:
|
||
args = self._make_args()
|
||
config = self._make_config(
|
||
trading_day_check_enabled=False,
|
||
market_review_enabled=True,
|
||
single_stock_notify=False,
|
||
merge_email_notification=False,
|
||
analysis_delay=0,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
pipeline = MagicMock()
|
||
pipeline.run.return_value = []
|
||
pipeline_kwargs = {}
|
||
|
||
def build_pipeline(*args, **kwargs):
|
||
pipeline_kwargs.update(kwargs)
|
||
return pipeline
|
||
|
||
with patch.object(main, "_refresh_stock_index_cache_for_analysis") as refresh, \
|
||
patch("main._compute_trading_day_filter", return_value=([], "cn", False)), \
|
||
patch("src.core.pipeline.StockAnalysisPipeline", side_effect=build_pipeline), \
|
||
patch("main._prime_daily_market_context", side_effect=[("", ""), ("缓存摘要", "完整复盘")]) as prime_context, \
|
||
patch("main._run_market_review_with_shared_lock", return_value=SimpleNamespace(report="大盘复盘")) as run_with_lock:
|
||
main.run_full_analysis(config, args, [])
|
||
|
||
self.assertTrue(pipeline_kwargs["daily_market_context_enabled"])
|
||
self.assertTrue(pipeline_kwargs["daily_market_context_allow_generate"])
|
||
self.assertEqual(prime_context.call_count, 2)
|
||
run_with_lock.assert_called_once()
|
||
refresh.assert_called_once_with(config)
|
||
|
||
def test_run_full_analysis_primes_daily_market_context_before_stock_analysis(self) -> None:
|
||
args = self._make_args()
|
||
target_date = date(2026, 3, 26)
|
||
reference_times = []
|
||
config = self._make_config(
|
||
trading_day_check_enabled=False,
|
||
market_review_enabled=True,
|
||
daily_market_context_enabled=True,
|
||
single_stock_notify=False,
|
||
merge_email_notification=False,
|
||
analysis_delay=0,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
pipeline = MagicMock()
|
||
pipeline.run.return_value = []
|
||
pipeline_kwargs = {}
|
||
|
||
def build_pipeline(*args, **kwargs):
|
||
pipeline_kwargs.update(kwargs)
|
||
return pipeline
|
||
|
||
def resolve_target_date(region, current_time):
|
||
self.assertEqual(region, "cn")
|
||
reference_times.append(current_time)
|
||
return target_date
|
||
|
||
with patch.object(main, "_refresh_stock_index_cache_for_analysis") as refresh, \
|
||
patch("main._compute_trading_day_filter", return_value=([], "cn", False)), \
|
||
patch("main._resolve_daily_market_context_target_date", side_effect=resolve_target_date), \
|
||
patch("src.core.pipeline.StockAnalysisPipeline", side_effect=build_pipeline), \
|
||
patch("main._prime_daily_market_context", return_value=("大盘退潮，高风险，建议观望，仓位上限30%。", "完整复盘正文")) as prime_context, \
|
||
patch("main._run_market_review_with_shared_lock") as run_with_lock, \
|
||
patch("src.core.market_review.run_market_review") as run_market_review:
|
||
main.run_full_analysis(config, args, [])
|
||
|
||
self.assertTrue(pipeline_kwargs["daily_market_context_allow_generate"])
|
||
prime_context.assert_has_calls(
|
||
[
|
||
unittest.mock.call(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=target_date,
|
||
return_full_report=False,
|
||
),
|
||
unittest.mock.call(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=target_date,
|
||
return_full_report=True,
|
||
require_current_query_match=True,
|
||
),
|
||
]
|
||
)
|
||
self.assertEqual(len(reference_times), 1)
|
||
self.assertIs(pipeline.run.call_args.kwargs["current_time"], reference_times[0])
|
||
self.assertEqual(pipeline.run.call_args.kwargs["current_time"].tzinfo, timezone.utc)
|
||
run_with_lock.assert_called_once()
|
||
self.assertFalse(run_with_lock.call_args.kwargs["merge_notification"])
|
||
self.assertTrue(run_with_lock.call_args.kwargs["send_notification"])
|
||
run_market_review.assert_not_called()
|
||
refresh.assert_called_once_with(config)
|
||
pipeline.run.assert_called_once()
|
||
|
||
def test_resolve_daily_market_context_target_date_passes_jp_kr_to_trading_calendar(self) -> None:
|
||
current_time = datetime(2026, 3, 26, 12, 0, tzinfo=timezone.utc)
|
||
target_date = date(2026, 3, 25)
|
||
|
||
with patch("src.core.trading_calendar.get_effective_trading_date", return_value=target_date) as get_date:
|
||
self.assertEqual(
|
||
main._resolve_daily_market_context_target_date("jp", current_time),
|
||
target_date,
|
||
)
|
||
self.assertEqual(
|
||
main._resolve_daily_market_context_target_date("kr", current_time),
|
||
target_date,
|
||
)
|
||
|
||
self.assertEqual(
|
||
get_date.call_args_list,
|
||
[
|
||
unittest.mock.call("jp", current_time=current_time),
|
||
unittest.mock.call("kr", current_time=current_time),
|
||
],
|
||
)
|
||
|
||
def test_run_full_analysis_does_not_reuse_single_context_for_multi_market_review(self) -> None:
|
||
args = self._make_args()
|
||
target_date = date(2026, 3, 26)
|
||
config = self._make_config(
|
||
trading_day_check_enabled=True,
|
||
market_review_region="both",
|
||
market_review_enabled=True,
|
||
daily_market_context_enabled=True,
|
||
single_stock_notify=False,
|
||
merge_email_notification=False,
|
||
analysis_delay=0,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
pipeline = MagicMock()
|
||
pipeline.run.return_value = []
|
||
pipeline_kwargs = {}
|
||
|
||
def build_pipeline(*args, **kwargs):
|
||
pipeline_kwargs.update(kwargs)
|
||
return pipeline
|
||
|
||
with patch.object(main, "_refresh_stock_index_cache_for_analysis") as refresh, \
|
||
patch("main._compute_trading_day_filter", return_value=([], "cn,us", False)), \
|
||
patch("main._resolve_daily_market_context_target_date", return_value=target_date), \
|
||
patch("src.core.pipeline.StockAnalysisPipeline", side_effect=build_pipeline), \
|
||
patch("main._prime_daily_market_context", return_value=("A股缓存摘要", "")) as prime_context, \
|
||
patch("main._run_market_review_with_shared_lock", return_value="多市场复盘") as run_with_lock, \
|
||
patch("src.core.market_review.run_market_review") as run_market_review:
|
||
main.run_full_analysis(config, args, [])
|
||
|
||
self.assertTrue(pipeline_kwargs["daily_market_context_allow_generate"])
|
||
prime_context.assert_has_calls(
|
||
[
|
||
unittest.mock.call(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn,us",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=target_date,
|
||
return_full_report=False,
|
||
),
|
||
unittest.mock.call(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn,us",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=target_date,
|
||
return_full_report=True,
|
||
require_current_query_match=True,
|
||
),
|
||
]
|
||
)
|
||
run_with_lock.assert_called_once()
|
||
self.assertEqual(run_with_lock.call_args.kwargs["override_region"], "cn,us")
|
||
run_market_review.assert_not_called()
|
||
refresh.assert_called_once_with(config)
|
||
pipeline.run.assert_called_once()
|
||
|
||
def test_prime_daily_market_context_readonly_mode_still_reuses_cached_context(self) -> None:
|
||
target_date = date(2026, 3, 26)
|
||
config = self._make_config(
|
||
trading_day_check_enabled=False,
|
||
market_review_enabled=True,
|
||
market_review_region="cn",
|
||
single_stock_notify=False,
|
||
merge_email_notification=False,
|
||
analysis_delay=0,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
pipeline = MagicMock()
|
||
pipeline._daily_market_context_service = None
|
||
pipeline.db = MagicMock()
|
||
pipeline.query_id = "prime-query"
|
||
context = SimpleNamespace(source="analysis_history", summary="历史复盘摘要")
|
||
service = MagicMock()
|
||
service.get_context.return_value = context
|
||
|
||
with patch(
|
||
"src.services.daily_market_context.DailyMarketContextService",
|
||
return_value=service,
|
||
) as service_cls:
|
||
summary, full_report = main._prime_daily_market_context(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=target_date,
|
||
return_full_report=True,
|
||
)
|
||
|
||
self.assertEqual(summary, "历史复盘摘要")
|
||
self.assertEqual(full_report, "")
|
||
service_cls.assert_called_once_with(db_manager=pipeline.db)
|
||
call_kwargs = service.get_context.call_args.kwargs
|
||
self.assertEqual(call_kwargs["region"], "cn")
|
||
self.assertFalse(call_kwargs["force_refresh"])
|
||
self.assertFalse(call_kwargs["allow_generate"])
|
||
self.assertFalse(call_kwargs["persist_market_review_history"])
|
||
self.assertEqual(call_kwargs["target_date"], target_date)
|
||
self.assertEqual(call_kwargs["current_query_id"], "prime-query")
|
||
|
||
def test_prime_daily_market_context_query_fallback_reuses_runtime_context(self) -> None:
|
||
target_date = date(2026, 3, 26)
|
||
config = self._make_config(
|
||
trading_day_check_enabled=False,
|
||
market_review_enabled=True,
|
||
market_review_region="cn",
|
||
single_stock_notify=False,
|
||
merge_email_notification=True,
|
||
analysis_delay=0,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
pipeline = MagicMock()
|
||
pipeline._daily_market_context_service = None
|
||
pipeline.db = MagicMock()
|
||
pipeline.query_id = "prime-query"
|
||
context = SimpleNamespace(
|
||
source="market_review_runtime",
|
||
summary="本轮运行时复盘摘要",
|
||
full_report="本轮运行时完整复盘",
|
||
query_id="prime-query",
|
||
)
|
||
service = MagicMock()
|
||
service.get_context.return_value = context
|
||
|
||
with patch(
|
||
"src.services.daily_market_context.DailyMarketContextService",
|
||
return_value=service,
|
||
):
|
||
summary, full_report = main._prime_daily_market_context(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=target_date,
|
||
return_full_report=True,
|
||
require_current_query_match=True,
|
||
)
|
||
|
||
self.assertEqual(summary, "本轮运行时复盘摘要")
|
||
self.assertEqual(full_report, "本轮运行时完整复盘")
|
||
self.assertTrue(service.get_context.call_args.kwargs["require_query_id_match"])
|
||
|
||
def test_run_full_analysis_generates_full_market_review_once_after_stock_analysis(self) -> None:
|
||
args = self._make_args()
|
||
target_date = date(2026, 3, 26)
|
||
config = self._make_config(
|
||
trading_day_check_enabled=False,
|
||
market_review_enabled=True,
|
||
daily_market_context_enabled=True,
|
||
single_stock_notify=False,
|
||
merge_email_notification=False,
|
||
analysis_delay=0,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
pipeline = MagicMock()
|
||
events = []
|
||
pipeline.run.side_effect = lambda **kwargs: events.append("stock-run") or []
|
||
pipeline_kwargs = {}
|
||
|
||
def build_pipeline(*args, **kwargs):
|
||
events.append("pipeline")
|
||
pipeline_kwargs.update(kwargs)
|
||
return pipeline
|
||
|
||
def run_with_lock(*args, **kwargs):
|
||
events.append("market-review")
|
||
return SimpleNamespace(report="完整复盘")
|
||
|
||
with patch.object(main, "_refresh_stock_index_cache_for_analysis") as refresh, \
|
||
patch("main._compute_trading_day_filter", return_value=([], "cn", False)), \
|
||
patch("main._resolve_daily_market_context_target_date", return_value=target_date), \
|
||
patch("src.core.pipeline.StockAnalysisPipeline", side_effect=build_pipeline), \
|
||
patch("main._prime_daily_market_context", return_value=("", "")) as prime_context, \
|
||
patch("main._run_market_review_with_shared_lock", side_effect=run_with_lock) as run_with_lock_mock, \
|
||
patch("src.core.market_review.run_market_review") as run_market_review:
|
||
main.run_full_analysis(config, args, [])
|
||
|
||
self.assertTrue(pipeline_kwargs["daily_market_context_allow_generate"])
|
||
self.assertEqual(events, ["pipeline", "stock-run", "market-review"])
|
||
query_scoped_read = unittest.mock.call(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=target_date,
|
||
return_full_report=True,
|
||
require_current_query_match=True,
|
||
)
|
||
self.assertEqual(
|
||
prime_context.call_args_list,
|
||
[
|
||
unittest.mock.call(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=target_date,
|
||
return_full_report=False,
|
||
),
|
||
query_scoped_read,
|
||
query_scoped_read,
|
||
],
|
||
)
|
||
run_with_lock_mock.assert_called_once()
|
||
run_market_review.assert_not_called()
|
||
refresh.assert_called_once_with(config)
|
||
pipeline.run.assert_called_once()
|
||
|
||
def test_run_full_analysis_reuses_runtime_market_context_after_stock_analysis(self) -> None:
|
||
args = self._make_args()
|
||
target_date = date(2026, 3, 26)
|
||
config = self._make_config(
|
||
trading_day_check_enabled=False,
|
||
market_review_enabled=True,
|
||
daily_market_context_enabled=True,
|
||
single_stock_notify=False,
|
||
merge_email_notification=False,
|
||
analysis_delay=0,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
pipeline = MagicMock()
|
||
pipeline.run.return_value = []
|
||
pipeline.notifier = MagicMock(
|
||
is_available=MagicMock(return_value=True),
|
||
send=MagicMock(return_value=True),
|
||
)
|
||
pipeline_kwargs = {}
|
||
|
||
def build_pipeline(*args, **kwargs):
|
||
pipeline_kwargs.update(kwargs)
|
||
return pipeline
|
||
|
||
runtime_context = ("本轮运行时复盘摘要", "## 本轮运行时完整复盘")
|
||
with patch.object(main, "_refresh_stock_index_cache_for_analysis") as refresh, \
|
||
patch("main._compute_trading_day_filter", return_value=([], "cn", False)), \
|
||
patch("main._resolve_daily_market_context_target_date", return_value=target_date), \
|
||
patch("src.core.pipeline.StockAnalysisPipeline", side_effect=build_pipeline), \
|
||
patch(
|
||
"main._prime_daily_market_context",
|
||
side_effect=[("", ""), ("", ""), runtime_context],
|
||
) as prime_context, \
|
||
patch("main._run_market_review_with_shared_lock") as run_with_lock, \
|
||
patch("src.core.market_review.run_market_review") as run_market_review:
|
||
main.run_full_analysis(config, args, [])
|
||
|
||
self.assertTrue(pipeline_kwargs["daily_market_context_allow_generate"])
|
||
run_with_lock.assert_not_called()
|
||
run_market_review.assert_not_called()
|
||
pipeline.notifier.send.assert_called_once()
|
||
self.assertIn("## 本轮运行时完整复盘", pipeline.notifier.send.call_args.args[0])
|
||
self.assertEqual(pipeline.notifier.send.call_args.kwargs["route_type"], "report")
|
||
query_scoped_read = unittest.mock.call(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=target_date,
|
||
return_full_report=True,
|
||
require_current_query_match=True,
|
||
)
|
||
self.assertEqual(
|
||
prime_context.call_args_list,
|
||
[
|
||
unittest.mock.call(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=target_date,
|
||
return_full_report=False,
|
||
),
|
||
query_scoped_read,
|
||
query_scoped_read,
|
||
],
|
||
)
|
||
refresh.assert_called_once_with(config)
|
||
pipeline.run.assert_called_once()
|
||
|
||
def test_run_full_analysis_saves_reused_runtime_market_context_without_notify(self) -> None:
|
||
args = self._make_args(no_notify=True)
|
||
target_date = date(2026, 3, 26)
|
||
config = self._make_config(
|
||
trading_day_check_enabled=False,
|
||
market_review_enabled=True,
|
||
daily_market_context_enabled=True,
|
||
single_stock_notify=False,
|
||
merge_email_notification=False,
|
||
analysis_delay=0,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
pipeline = MagicMock()
|
||
pipeline.run.return_value = []
|
||
pipeline.notifier = MagicMock()
|
||
pipeline.notifier.save_report_to_file.return_value = "/tmp/market_review.md"
|
||
|
||
def build_pipeline(*args, **kwargs):
|
||
return pipeline
|
||
|
||
runtime_context = ("本轮运行时复盘摘要", "## 本轮运行时完整复盘")
|
||
with (
|
||
patch.object(main, "_refresh_stock_index_cache_for_analysis") as refresh,
|
||
patch("main._compute_trading_day_filter", return_value=([], "cn", False)),
|
||
patch("main._resolve_daily_market_context_target_date", return_value=target_date),
|
||
patch("src.core.pipeline.StockAnalysisPipeline", side_effect=build_pipeline),
|
||
patch(
|
||
"main._prime_daily_market_context",
|
||
side_effect=[("", ""), ("", ""), runtime_context],
|
||
) as prime_context,
|
||
patch("main._run_market_review_with_shared_lock") as run_with_lock,
|
||
patch("src.core.market_review.run_market_review") as run_market_review,
|
||
):
|
||
main.run_full_analysis(config, args, [])
|
||
|
||
run_with_lock.assert_not_called()
|
||
run_market_review.assert_not_called()
|
||
pipeline.notifier.send.assert_not_called()
|
||
pipeline.notifier.save_report_to_file.assert_called_once()
|
||
saved_content, saved_filename = pipeline.notifier.save_report_to_file.call_args.args
|
||
self.assertTrue(saved_content.startswith("# 🎯 大盘复盘\n\n"))
|
||
self.assertIn("## 本轮运行时完整复盘", saved_content)
|
||
self.assertTrue(saved_filename.startswith("market_review_"))
|
||
self.assertTrue(saved_filename.endswith(".md"))
|
||
self.assertEqual(prime_context.call_count, 3)
|
||
refresh.assert_called_once_with(config)
|
||
pipeline.run.assert_called_once()
|
||
|
||
def test_run_full_analysis_still_runs_market_review_for_merge_disabled_with_reused_context(self) -> None:
|
||
args = self._make_args()
|
||
target_date = date(2026, 3, 26)
|
||
config = self._make_config(
|
||
trading_day_check_enabled=False,
|
||
market_review_enabled=True,
|
||
daily_market_context_enabled=True,
|
||
single_stock_notify=False,
|
||
merge_email_notification=False,
|
||
analysis_delay=0,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
pipeline = MagicMock()
|
||
pipeline.run.return_value = []
|
||
pipeline_kwargs = {}
|
||
|
||
def build_pipeline(*args, **kwargs):
|
||
pipeline_kwargs.update(kwargs)
|
||
return pipeline
|
||
|
||
with patch.object(main, "_refresh_stock_index_cache_for_analysis") as refresh, \
|
||
patch("main._compute_trading_day_filter", return_value=([], "cn", False)), \
|
||
patch("main._resolve_daily_market_context_target_date", return_value=target_date), \
|
||
patch("src.core.pipeline.StockAnalysisPipeline", side_effect=build_pipeline), \
|
||
patch(
|
||
"main._prime_daily_market_context",
|
||
return_value=("大盘退潮，高风险，建议观望。", "## 完整大盘复盘\n市场结构偏弱，建议保守。"),
|
||
) as prime_context, \
|
||
patch("main._run_market_review_with_shared_lock") as run_with_lock, \
|
||
patch("src.core.market_review.run_market_review") as run_market_review:
|
||
main.run_full_analysis(config, args, [])
|
||
|
||
self.assertTrue(pipeline_kwargs["daily_market_context_allow_generate"])
|
||
prime_context.assert_has_calls(
|
||
[
|
||
unittest.mock.call(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=target_date,
|
||
return_full_report=False,
|
||
),
|
||
unittest.mock.call(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=target_date,
|
||
return_full_report=True,
|
||
require_current_query_match=True,
|
||
),
|
||
]
|
||
)
|
||
run_with_lock.assert_called_once()
|
||
self.assertFalse(run_with_lock.call_args.kwargs["merge_notification"])
|
||
run_market_review.assert_not_called()
|
||
refresh.assert_called_once_with(config)
|
||
pipeline.run.assert_called_once()
|
||
|
||
def test_run_full_analysis_waits_for_analysis_delay_before_market_review(self) -> None:
|
||
args = self._make_args()
|
||
target_date = date(2026, 3, 26)
|
||
config = self._make_config(
|
||
trading_day_check_enabled=False,
|
||
market_review_enabled=True,
|
||
daily_market_context_enabled=True,
|
||
single_stock_notify=False,
|
||
merge_email_notification=False,
|
||
analysis_delay=2,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
pipeline = MagicMock()
|
||
events = []
|
||
pipeline.run.side_effect = lambda **kwargs: events.append("stock-run") or []
|
||
pipeline_kwargs = {}
|
||
|
||
def build_pipeline(*args, **kwargs):
|
||
events.append("pipeline")
|
||
pipeline_kwargs.update(kwargs)
|
||
return pipeline
|
||
|
||
def run_with_lock(*args, **kwargs):
|
||
events.append("market-review")
|
||
return SimpleNamespace(report="完整复盘")
|
||
|
||
with patch.object(main, "_refresh_stock_index_cache_for_analysis") as refresh, \
|
||
patch("main._compute_trading_day_filter", return_value=([], "cn", False)), \
|
||
patch("main._resolve_daily_market_context_target_date", return_value=target_date), \
|
||
patch("src.core.pipeline.StockAnalysisPipeline", side_effect=build_pipeline), \
|
||
patch("main._prime_daily_market_context", return_value=("", "")) as prime_context, \
|
||
patch("main._run_market_review_with_shared_lock", side_effect=run_with_lock) as run_with_lock_mock, \
|
||
patch("time.sleep") as sleep, \
|
||
patch("src.core.market_review.run_market_review") as run_market_review:
|
||
main.run_full_analysis(config, args, [])
|
||
|
||
self.assertTrue(pipeline_kwargs["daily_market_context_allow_generate"])
|
||
self.assertEqual(events, ["pipeline", "stock-run", "market-review"])
|
||
self.assertEqual(sleep.call_count, 1)
|
||
sleep.assert_called_once_with(2)
|
||
self.assertEqual(
|
||
run_with_lock_mock.call_args.kwargs["send_notification"],
|
||
True,
|
||
)
|
||
run_market_review.assert_not_called()
|
||
run_with_lock_mock.assert_called_once()
|
||
refresh.assert_called_once_with(config)
|
||
prime_context.assert_has_calls(
|
||
[
|
||
unittest.mock.call(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=target_date,
|
||
return_full_report=False,
|
||
),
|
||
unittest.mock.call(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=target_date,
|
||
return_full_report=True,
|
||
require_current_query_match=True,
|
||
),
|
||
unittest.mock.call(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=target_date,
|
||
return_full_report=True,
|
||
require_current_query_match=True,
|
||
),
|
||
]
|
||
)
|
||
|
||
def test_run_full_analysis_reuses_cached_market_context_as_full_report(self) -> None:
|
||
args = self._make_args()
|
||
target_date = date(2026, 3, 26)
|
||
config = self._make_config(
|
||
trading_day_check_enabled=False,
|
||
market_review_enabled=True,
|
||
daily_market_context_enabled=True,
|
||
single_stock_notify=False,
|
||
merge_email_notification=True,
|
||
analysis_delay=0,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
report_type="simple",
|
||
)
|
||
pipeline = MagicMock()
|
||
pipeline.run.return_value = []
|
||
pipeline.notifier = MagicMock(
|
||
is_available=MagicMock(return_value=True),
|
||
generate_aggregate_report=MagicMock(return_value=""),
|
||
send=MagicMock(return_value=True),
|
||
)
|
||
pipeline_kwargs = {}
|
||
|
||
def build_pipeline(*args, **kwargs):
|
||
pipeline_kwargs.update(kwargs)
|
||
return pipeline
|
||
|
||
with patch.object(main, "_refresh_stock_index_cache_for_analysis") as refresh, \
|
||
patch("main._compute_trading_day_filter", return_value=([], "cn", False)), \
|
||
patch("main._resolve_daily_market_context_target_date", return_value=target_date), \
|
||
patch("src.core.pipeline.StockAnalysisPipeline", side_effect=build_pipeline), \
|
||
patch(
|
||
"main._prime_daily_market_context",
|
||
return_value=(
|
||
"大盘退潮，高风险，建议观望。",
|
||
"## 完整大盘复盘\n市场结构偏弱，建议保守。",
|
||
),
|
||
) as prime_context, \
|
||
patch("main._run_market_review_with_shared_lock") as run_with_lock, \
|
||
patch("src.core.market_review.run_market_review") as run_market_review:
|
||
main.run_full_analysis(config, args, [])
|
||
|
||
self.assertTrue(pipeline_kwargs["daily_market_context_allow_generate"])
|
||
prime_context.assert_has_calls(
|
||
[
|
||
unittest.mock.call(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=target_date,
|
||
return_full_report=False,
|
||
),
|
||
unittest.mock.call(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=target_date,
|
||
return_full_report=True,
|
||
require_current_query_match=True,
|
||
),
|
||
]
|
||
)
|
||
run_with_lock.assert_not_called()
|
||
run_market_review.assert_not_called()
|
||
refresh.assert_called_once_with(config)
|
||
pipeline.run.assert_called_once_with(
|
||
stock_codes=[],
|
||
dry_run=False,
|
||
send_notification=True,
|
||
merge_notification=True,
|
||
current_time=unittest.mock.ANY,
|
||
analysis_targets=None,
|
||
)
|
||
notifier_message = pipeline.notifier.send.call_args.args[0]
|
||
self.assertIn("## 完整大盘复盘", notifier_message)
|
||
self.assertNotIn("大盘退潮，高风险，建议观望。", notifier_message)
|
||
|
||
def test_run_market_review_with_shared_lock_forwards_request_config(self) -> None:
|
||
config = self._make_config(
|
||
trading_day_check_enabled=False,
|
||
market_review_enabled=True,
|
||
daily_market_context_enabled=True,
|
||
single_stock_notify=False,
|
||
merge_email_notification=False,
|
||
analysis_delay=0,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
run_review = MagicMock(return_value="复盘结果")
|
||
|
||
with patch("src.core.market_review_lock.try_acquire_market_review_lock", return_value=object()) as acquire_lock, \
|
||
patch("src.core.market_review_lock.release_market_review_lock") as release_lock:
|
||
result = main._run_market_review_with_shared_lock(
|
||
config,
|
||
run_review,
|
||
send_notification=False,
|
||
)
|
||
|
||
self.assertEqual(result, "复盘结果")
|
||
acquire_lock.assert_called_once_with(config)
|
||
run_review.assert_called_once_with(config=config, send_notification=False)
|
||
release_lock.assert_called_once_with(unittest.mock.ANY)
|
||
|
||
def test_prime_daily_market_context_uses_ephemeral_service_for_multi_market_region(self) -> None:
|
||
config = self._make_config(
|
||
trading_day_check_enabled=False,
|
||
market_review_enabled=True,
|
||
single_stock_notify=False,
|
||
merge_email_notification=False,
|
||
analysis_delay=0,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
pipeline = MagicMock()
|
||
pipeline._daily_market_context_service = MagicMock()
|
||
pipeline._daily_market_context_service.get_context.return_value = SimpleNamespace(
|
||
source="analysis_history",
|
||
summary="旧A股复盘摘要",
|
||
)
|
||
pipeline.db = MagicMock()
|
||
context = SimpleNamespace(source="analysis_history", summary="多市场复盘摘要", full_report="完整复盘正文")
|
||
regional_service = MagicMock()
|
||
regional_service.get_context.return_value = context
|
||
|
||
with patch("src.services.daily_market_context.DailyMarketContextService", return_value=regional_service) as service_cls:
|
||
summary, full_report = main._prime_daily_market_context(
|
||
config,
|
||
pipeline=pipeline,
|
||
region="cn,us",
|
||
no_market_review=False,
|
||
allow_generate=False,
|
||
target_date=date(2026, 3, 26),
|
||
return_full_report=True,
|
||
)
|
||
|
||
self.assertEqual(summary, "多市场复盘摘要")
|
||
self.assertEqual(full_report, "完整复盘正文")
|
||
service_cls.assert_called_once_with(db_manager=pipeline.db)
|
||
regional_service.get_context.assert_called_once()
|
||
self.assertIsNot(
|
||
regional_service,
|
||
pipeline._daily_market_context_service,
|
||
"多市场预热必须使用独立服务避免共享缓存污染",
|
||
)
|
||
pipeline._daily_market_context_service.get_context.assert_not_called()
|
||
|
||
get_context_kwargs = regional_service.get_context.call_args.kwargs
|
||
self.assertEqual(get_context_kwargs["region"], "cn,us")
|
||
self.assertFalse(get_context_kwargs["force_refresh"])
|
||
self.assertFalse(get_context_kwargs["allow_generate"])
|
||
self.assertFalse(get_context_kwargs["persist_market_review_history"])
|
||
|
||
def test_config_enabled_schedule_marks_market_review_source_as_schedule(self) -> None:
|
||
args = self._make_args(schedule=False)
|
||
target_date = date(2026, 3, 26)
|
||
config = self._make_config(
|
||
schedule_enabled=True,
|
||
trading_day_check_enabled=False,
|
||
market_review_enabled=True,
|
||
no_market_review=False,
|
||
single_stock_notify=False,
|
||
merge_email_notification=False,
|
||
analysis_delay=0,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
pipeline = MagicMock()
|
||
pipeline.run.return_value = []
|
||
|
||
with patch.object(main, "_refresh_stock_index_cache_for_analysis"), \
|
||
patch.object(main, "_compute_trading_day_filter", return_value=(["600519"], "cn", False)), \
|
||
patch("main._resolve_daily_market_context_target_date", return_value=target_date), \
|
||
patch("src.core.pipeline.StockAnalysisPipeline", return_value=pipeline), \
|
||
patch("main._prime_daily_market_context", return_value=("", "")), \
|
||
patch("main._run_market_review_with_shared_lock", return_value="market report") as run_with_lock, \
|
||
patch("src.core.market_review.run_market_review") as run_market_review:
|
||
main.run_full_analysis(config, args, ["600519"])
|
||
|
||
pipeline.run.assert_called_once()
|
||
run_with_lock.assert_called_once()
|
||
call_args = run_with_lock.call_args
|
||
self.assertIs(call_args.args[1], run_market_review)
|
||
self.assertEqual(call_args.kwargs["trigger_source"], "schedule")
|
||
|
||
def test_run_full_analysis_keeps_targets_aligned_after_trading_day_filter(self) -> None:
|
||
args = self._make_args(dry_run=True, no_market_review=True)
|
||
config = self._make_config(
|
||
trading_day_check_enabled=True,
|
||
market_review_enabled=False,
|
||
daily_market_context_enabled=False,
|
||
single_stock_notify=False,
|
||
merge_email_notification=False,
|
||
analysis_delay=0,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
pipeline = MagicMock()
|
||
pipeline.run.return_value = []
|
||
targets = [
|
||
parse_analysis_target("sh000016"),
|
||
parse_analysis_target("AAPL"),
|
||
]
|
||
|
||
with patch.object(main, "_refresh_stock_index_cache_for_analysis"), \
|
||
patch.object(
|
||
main,
|
||
"_compute_trading_day_filter",
|
||
return_value=(["AAPL"], "us", False),
|
||
), \
|
||
patch("src.core.pipeline.StockAnalysisPipeline", return_value=pipeline), \
|
||
patch("src.core.market_review.run_market_review"):
|
||
main.run_full_analysis(
|
||
config,
|
||
args,
|
||
["sh000016", "AAPL"],
|
||
analysis_targets=targets,
|
||
)
|
||
|
||
run_kwargs = pipeline.run.call_args.kwargs
|
||
self.assertEqual(run_kwargs["stock_codes"], ["AAPL"])
|
||
self.assertEqual(run_kwargs["analysis_targets"], [targets[1]])
|
||
|
||
def test_market_review_mode_uses_shared_runtime_assembly(self) -> None:
|
||
args = self._make_args(market_review=True)
|
||
config = self._make_config(
|
||
trading_day_check_enabled=True,
|
||
market_review_region="both",
|
||
market_review_enabled=False,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
runtime_notifier = MagicMock()
|
||
runtime_analyzer = MagicMock()
|
||
runtime_search_service = MagicMock()
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch("main._run_market_review_with_shared_lock") as run_with_lock, \
|
||
patch(
|
||
"src.core.market_review_runtime.build_market_review_runtime",
|
||
return_value=(
|
||
runtime_notifier,
|
||
runtime_analyzer,
|
||
runtime_search_service,
|
||
),
|
||
) as runtime_builder, \
|
||
patch("src.core.market_review.run_market_review") as run_market_review, \
|
||
patch("src.core.trading_calendar.get_open_markets_today", return_value={"cn", "us"}), \
|
||
patch("src.core.trading_calendar.compute_effective_region", return_value="cn,us"):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
runtime_builder.assert_called_once_with(config)
|
||
run_with_lock.assert_called_once()
|
||
call_args = run_with_lock.call_args
|
||
self.assertEqual(call_args.args[0], config)
|
||
self.assertIs(call_args.args[1], run_market_review)
|
||
self.assertIs(call_args.kwargs["notifier"], runtime_notifier)
|
||
self.assertIs(call_args.kwargs["analyzer"], runtime_analyzer)
|
||
self.assertIs(call_args.kwargs["search_service"], runtime_search_service)
|
||
self.assertTrue(call_args.kwargs["send_notification"])
|
||
self.assertNotIn("merge_notification", call_args.kwargs)
|
||
self.assertEqual(call_args.kwargs["override_region"], "cn,us")
|
||
self.assertEqual(call_args.kwargs["trigger_source"], "cli")
|
||
|
||
def test_market_review_mode_returns_nonzero_when_no_report_is_generated(self) -> None:
|
||
args = self._make_args(market_review=True)
|
||
config = self._make_config(
|
||
trading_day_check_enabled=True,
|
||
market_review_region="both",
|
||
market_review_enabled=False,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch(
|
||
"src.core.market_review_runtime.build_market_review_runtime",
|
||
return_value=(MagicMock(), MagicMock(), MagicMock()),
|
||
), \
|
||
patch("main._run_market_review_with_shared_lock", return_value=None) as run_with_lock, \
|
||
patch("src.core.market_review.run_market_review"), \
|
||
patch("src.core.trading_calendar.get_open_markets_today", return_value={"cn", "us"}), \
|
||
patch("src.core.trading_calendar.compute_effective_region", return_value="cn,us"):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 1)
|
||
run_with_lock.assert_called_once()
|
||
|
||
def test_market_review_mode_respects_comma_list_market_review_region(self) -> None:
|
||
args = self._make_args(market_review=True)
|
||
config = self._make_config(
|
||
trading_day_check_enabled=True,
|
||
market_review_region="jp,kr",
|
||
market_review_enabled=False,
|
||
database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"),
|
||
)
|
||
runtime_notifier = MagicMock()
|
||
runtime_analyzer = MagicMock()
|
||
runtime_search_service = MagicMock()
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch("main._run_market_review_with_shared_lock") as run_with_lock, \
|
||
patch(
|
||
"src.core.market_review_runtime.build_market_review_runtime",
|
||
return_value=(runtime_notifier, runtime_analyzer, runtime_search_service),
|
||
) as runtime_builder, \
|
||
patch("src.core.market_review.run_market_review"), \
|
||
patch("src.core.trading_calendar.get_open_markets_today", return_value={"jp", "kr"}):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
runtime_builder.assert_called_once_with(config)
|
||
call_args = run_with_lock.call_args
|
||
self.assertIs(call_args.args[0], config)
|
||
self.assertEqual(call_args.kwargs["override_region"], "jp,kr")
|
||
self.assertEqual(call_args.kwargs["trigger_source"], "cli")
|
||
|
||
def test_bootstrap_logging_persists_when_config_load_fails(self) -> None:
|
||
"""Config load failure must be logged to stderr and return exit code 1.
|
||
|
||
Bootstrap logging is stderr-only so healthy runs never write to a
|
||
hard-coded directory. The error is still captured by process runners
|
||
(e.g. GitHub Actions) that collect stderr output.
|
||
"""
|
||
import io
|
||
|
||
args = self._make_args()
|
||
|
||
capture_stream = io.StringIO()
|
||
capture_handler = logging.StreamHandler(capture_stream)
|
||
capture_handler.setLevel(logging.DEBUG)
|
||
capture_handler.setFormatter(logging.Formatter("%(message)s"))
|
||
|
||
root_logger = logging.getLogger()
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", side_effect=RuntimeError("config boom")):
|
||
root_logger.addHandler(capture_handler)
|
||
try:
|
||
exit_code = main.main()
|
||
finally:
|
||
root_logger.removeHandler(capture_handler)
|
||
capture_handler.close()
|
||
|
||
self.assertEqual(exit_code, 1)
|
||
output = capture_stream.getvalue()
|
||
self.assertIn("加载配置失败", output)
|
||
self.assertIn("config boom", output)
|
||
|
||
def test_bootstrap_logging_failure_does_not_block_startup(self) -> None:
|
||
"""Bootstrap log dir unwritable must not prevent startup (P1 regression)."""
|
||
args = self._make_args()
|
||
config = self._make_config()
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main._setup_bootstrap_logging", side_effect=OSError("read-only fs")), \
|
||
patch("main.setup_logging"), \
|
||
patch("main.run_full_analysis") as run_mock:
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
run_mock.assert_called_once()
|
||
|
||
def test_runtime_file_logging_permission_error_falls_back_to_console(self) -> None:
|
||
"""Configured file logging failures should not prevent Docker startup."""
|
||
import io
|
||
|
||
args = self._make_args()
|
||
config = self._make_config(log_dir="/app/logs")
|
||
capture_stream = io.StringIO()
|
||
capture_handler = logging.StreamHandler(capture_stream)
|
||
capture_handler.setLevel(logging.DEBUG)
|
||
capture_handler.setFormatter(logging.Formatter("%(message)s"))
|
||
|
||
root_logger = logging.getLogger()
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch(
|
||
"main.setup_logging",
|
||
side_effect=PermissionError("/app/logs/stock_analysis_20260511.log"),
|
||
), \
|
||
patch("main.run_full_analysis") as run_mock:
|
||
root_logger.addHandler(capture_handler)
|
||
try:
|
||
exit_code = main.main()
|
||
finally:
|
||
root_logger.removeHandler(capture_handler)
|
||
capture_handler.close()
|
||
|
||
self.assertEqual(exit_code, 0)
|
||
run_mock.assert_called_once()
|
||
output = capture_stream.getvalue()
|
||
self.assertIn("文件日志初始化失败，已降级为控制台日志输出", output)
|
||
self.assertIn("/app/logs", output)
|
||
self.assertIn("官方 Docker 镜像启动入口会自动修复默认挂载目录权限", output)
|
||
|
||
def test_run_full_analysis_import_failure_propagates(self) -> None:
|
||
"""P1: import failures in run_full_analysis must propagate, not be swallowed."""
|
||
args = self._make_args()
|
||
config = self._make_config()
|
||
|
||
with patch("main.parse_arguments", return_value=args), \
|
||
patch("main.get_config", return_value=config), \
|
||
patch("main.setup_logging"), \
|
||
patch.dict("sys.modules", {"src.core.pipeline": None}):
|
||
exit_code = main.main()
|
||
|
||
self.assertEqual(exit_code, 1)
|
||
|
||
def test_lazy_pipeline_triggers_env_bootstrap(self) -> None:
|
||
"""P2: lazy StockAnalysisPipeline access must call _bootstrap_environment."""
|
||
# Reset the lazy descriptor cache so __get__ fires again
|
||
main._LazyPipelineDescriptor._resolved = None
|
||
main._env_bootstrapped = False
|
||
|
||
with patch("main._bootstrap_environment", wraps=main._bootstrap_environment) as mock_boot, \
|
||
patch("src.core.pipeline.StockAnalysisPipeline", create=True, new_callable=lambda: type("FakePipeline", (), {})):
|
||
try:
|
||
_ = main.StockAnalysisPipeline
|
||
except Exception:
|
||
pass
|
||
mock_boot.assert_called()
|
||
|
||
# Cleanup: reset state
|
||
main._LazyPipelineDescriptor._resolved = None
|
||
main._env_bootstrapped = False
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|