# -*- coding: utf-8 -*- """ =================================== A股自选股智能分析系统 - 通知服务单元测试 =================================== 职责： 1. 验证通知服务的配置检测逻辑 2. 验证通知服务的渠道检测逻辑 3. 验证通知服务的消息发送逻辑 TODO: 1. 添加发送渠道以外的测试，如： - 生成日报 2. 添加 send_to_context 的测试 """ import os import sys import unittest from unittest import mock from typing import Optional sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) # Keep this test runnable when optional LLM/runtime deps are not installed. for optional_module in ("litellm", "json_repair"): try: __import__(optional_module) except ModuleNotFoundError: sys.modules[optional_module] = mock.MagicMock() from src.config import Config from src.notification import NotificationBuilder, NotificationChannel, NotificationService from src.notification_noise import reset_notification_noise_state from src.analyzer import AnalysisResult from bot.models import BotMessage, ChatType import requests def _make_config(**overrides) -> Config: """Create a Config instance overriding only notification-related fields.""" return Config(stock_list=[], **overrides) def _make_response(status_code: int, json: Optional[dict] = None) -> requests.Response: response = requests.Response() response.status_code = status_code if json: response.json = lambda: json return response def _attach_decision_signal_summary(result: AnalysisResult) -> AnalysisResult: result.decision_signal_summary = { "action": "sell", "action_label": "卖出", "horizon": "1d", "reason": "技术面走弱", } return result def _make_feishu_message() -> BotMessage: return BotMessage( platform="feishu", message_id="msg-1", user_id="user-1", user_name="tester", chat_id="chat-1", chat_type=ChatType.GROUP, content="/a 600519", ) def _make_dingtalk_message() -> BotMessage: return BotMessage( platform="dingtalk", message_id="msg-2", user_id="user-2", user_name="tester", chat_id="dingtalk-chat", chat_type=ChatType.GROUP, content="/a 600519", raw_data={ "sessionWebhook": "https://oapi.dingtalk.com/robot/sendBySession?session=abc123", }, ) def _make_telegram_message() -> BotMessage: return BotMessage( platform="telegram", message_id="msg-3", user_id="user-3", user_name="tester", chat_id="100200300", chat_type=ChatType.PRIVATE, content="/a 600519", raw_data={"chat_id": "100200300"}, ) class TestNotificationServiceSendToMethods(unittest.TestCase): """测试通知发送服务 测试设计： 测试按照渠道的字母顺序排列，在合适位置添加新的测试方法。 如果采用长消息分批发送，必须单独测试分批发送的逻辑， e.g. test_send_to_discord_via_notification_service_with_bot_requires_chunking 1. 添加模拟配置： 使用 mock.patch 装饰器来模拟 get_config 函数， 使用 _make_config 函数添加配置，并返回 Config 实例。 2. 检查配置是否正确： 使用 assertIn 检查 NotificationChannel.xxxx 是否在 `NotificationService.get_available_channels()` 返回值中。 3. 模拟请求响应： 使用 mock.patch 装饰器来模拟 requests.post 函数， 使用 _make_response 函数模拟请求响应，并返回 Response 实例。 若使用其他函数模拟请求响应，则使用 mock.patch 装饰器来模拟该函数。 4. 使用 assertTrue 检查 send 的返回值。 5. 使用 assert_called_once 检查请求函数是否被调用一次。 测试分批发送时，使用 assertAlmostEqual(mock_post.call_count, ...) 检查请求函数被调用次数 """ def setUp(self): reset_notification_noise_state() @mock.patch("src.notification.get_config") def test_no_channels_service_unavailable_and_send_returns_false(self, mock_get_config): mock_get_config.return_value = _make_config() service = NotificationService() self.assertFalse(service.is_available()) result = service.send("test content") self.assertFalse(result) @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_astrbot_via_notification_service(self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock): cfg = _make_config(astrbot_url="https://astrbot.example") mock_get_config.return_value = cfg mock_post.return_value = _make_response(200) service = NotificationService() self.assertIn(NotificationChannel.ASTRBOT, service.get_available_channels()) ok = service.send("astrbot content") self.assertTrue(ok) mock_post.assert_called_once() @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_custom_webhook_via_notification_service( self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock ): cfg = _make_config(custom_webhook_urls=["https://example.com/webhook"]) mock_get_config.return_value = cfg mock_post.return_value = _make_response(200) service = NotificationService() self.assertIn(NotificationChannel.CUSTOM, service.get_available_channels()) ok = service.send("custom content") self.assertTrue(ok) mock_post.assert_called_once() @mock.patch("src.notification.get_config") def test_send_isolates_channel_exceptions(self, mock_get_config: mock.MagicMock): cfg = _make_config( wechat_webhook_url="https://wechat.example/hook", custom_webhook_urls=["https://example.com/webhook"], ) mock_get_config.return_value = cfg service = NotificationService() self.assertIn(NotificationChannel.WECHAT, service.get_available_channels()) self.assertIn(NotificationChannel.CUSTOM, service.get_available_channels()) with mock.patch.object(service, "send_to_wechat", side_effect=RuntimeError("boom")), \ mock.patch.object(service, "send_to_custom", return_value=True) as mock_custom: ok = service.send("content") self.assertTrue(ok) mock_custom.assert_called_once_with("content") @mock.patch("src.notification.get_config") def test_send_with_results_reports_per_channel_attempts(self, mock_get_config: mock.MagicMock): cfg = _make_config( wechat_webhook_url="https://wechat.example/hook", custom_webhook_urls=["https://example.com/webhook"], ) mock_get_config.return_value = cfg service = NotificationService() with mock.patch.object(service, "send_to_wechat", side_effect=RuntimeError("token=secret-token failed")), \ mock.patch.object(service, "send_to_custom", return_value=True): result = service.send_with_results("content") self.assertTrue(result.dispatched) self.assertTrue(result.success) self.assertEqual(result.status, "partial_failed") by_channel = {item.channel: item for item in result.channel_results} self.assertFalse(by_channel["wechat"].success) self.assertEqual(by_channel["wechat"].error_code, "exception") self.assertNotIn("secret-token", by_channel["wechat"].diagnostics) self.assertTrue(by_channel["custom"].success) @mock.patch("src.notification.get_config") def test_send_with_results_reports_route_no_channel(self, mock_get_config: mock.MagicMock): cfg = _make_config( custom_webhook_urls=["https://example.com/webhook"], notification_report_channels=["unknown-route-channel"], ) mock_get_config.return_value = cfg service = NotificationService() with mock.patch.object(service, "send_to_custom", return_value=True) as mock_custom: result = service.send_with_results("content", route_type="report") self.assertFalse(result.dispatched) self.assertFalse(result.success) self.assertEqual(result.status, "no_channel") self.assertEqual(result.channel_results, []) mock_custom.assert_not_called() @mock.patch("src.notification.get_config") def test_send_route_empty_keeps_all_configured_channels(self, mock_get_config: mock.MagicMock): cfg = _make_config( wechat_webhook_url="https://wechat.example/hook", custom_webhook_urls=["https://example.com/webhook"], ) mock_get_config.return_value = cfg service = NotificationService() with mock.patch.object(service, "send_to_wechat", return_value=True) as mock_wechat, \ mock.patch.object(service, "send_to_custom", return_value=True) as mock_custom: ok = service.send("content", route_type="report") self.assertTrue(ok) mock_wechat.assert_called_once_with("content") mock_custom.assert_called_once_with("content") @mock.patch("src.notification.get_config") def test_send_report_route_filters_static_channels(self, mock_get_config: mock.MagicMock): cfg = _make_config( wechat_webhook_url="https://wechat.example/hook", custom_webhook_urls=["https://example.com/webhook"], notification_report_channels=["custom"], ) mock_get_config.return_value = cfg service = NotificationService() with mock.patch.object(service, "send_to_wechat", return_value=True) as mock_wechat, \ mock.patch.object(service, "send_to_custom", return_value=True) as mock_custom: ok = service.send("content", route_type="report") self.assertTrue(ok) mock_wechat.assert_not_called() mock_custom.assert_called_once_with("content") @mock.patch("src.notification.get_config") def test_send_alert_and_system_error_routes_filter_independently(self, mock_get_config: mock.MagicMock): cfg = _make_config( wechat_webhook_url="https://wechat.example/hook", custom_webhook_urls=["https://example.com/webhook"], notification_alert_channels=["wechat"], notification_system_error_channels=["custom"], ) mock_get_config.return_value = cfg service = NotificationService() with mock.patch.object(service, "send_to_wechat", return_value=True) as mock_wechat, \ mock.patch.object(service, "send_to_custom", return_value=True) as mock_custom: self.assertTrue(service.send("alert", route_type="alert")) self.assertTrue(service.send("system", route_type="system_error")) mock_wechat.assert_called_once_with("alert") mock_custom.assert_called_once_with("system") @mock.patch("src.notification.get_config") def test_send_route_with_no_matching_channel_does_not_fallback(self, mock_get_config: mock.MagicMock): cfg = _make_config( custom_webhook_urls=["https://example.com/webhook"], notification_report_channels=["unknown-route-channel"], ) mock_get_config.return_value = cfg service = NotificationService() with mock.patch.object(service, "send_to_custom", return_value=True) as mock_custom: ok = service.send("content", route_type="report") self.assertFalse(ok) mock_custom.assert_not_called() @mock.patch("src.notification.get_config") def test_send_to_context_is_not_limited_by_route(self, mock_get_config: mock.MagicMock): cfg = _make_config( custom_webhook_urls=["https://example.com/webhook"], notification_report_channels=["telegram"], ) mock_get_config.return_value = cfg service = NotificationService() with mock.patch.object(service, "send_to_context", return_value=True) as mock_context, \ mock.patch.object(service, "send_to_custom", return_value=True) as mock_custom: ok = service.send("content", route_type="report") self.assertTrue(ok) mock_context.assert_called_once_with("content") mock_custom.assert_not_called() @mock.patch("src.notification.get_config") def test_feishu_context_response_skips_static_webhook(self, mock_get_config: mock.MagicMock): cfg = _make_config( feishu_webhook_url="https://open.feishu.cn/open-apis/bot/v2/hook/test-token", feishu_app_id="cli_test", feishu_app_secret="app-secret", ) mock_get_config.return_value = cfg service = NotificationService(source_message=_make_feishu_message()) with mock.patch.object(service, "_send_feishu_stream_reply", return_value=True) as mock_reply, \ mock.patch.object(service, "send_to_feishu", return_value=True) as mock_webhook: result = service.send_with_results("content", route_type="report") self.assertTrue(result.dispatched) self.assertTrue(result.success) self.assertEqual(result.status, "sent") self.assertEqual([item.channel for item in result.channel_results], ["__context__"]) mock_reply.assert_called_once_with("chat-1", "content") mock_webhook.assert_not_called() @mock.patch("src.notification.get_config") def test_feishu_context_failure_does_not_fallback_to_static_webhook(self, mock_get_config: mock.MagicMock): cfg = _make_config( feishu_webhook_url="https://open.feishu.cn/open-apis/bot/v2/hook/test-token", feishu_app_id="cli_test", feishu_app_secret="app-secret", ) mock_get_config.return_value = cfg service = NotificationService(source_message=_make_feishu_message()) with mock.patch.object(service, "_send_feishu_stream_reply", return_value=False), \ mock.patch.object(service, "send_to_feishu", return_value=True) as mock_webhook: result = service.send_with_results("content", route_type="report") self.assertTrue(result.dispatched) self.assertFalse(result.success) self.assertEqual(result.status, "all_failed") self.assertEqual([item.channel for item in result.channel_results], ["__context__"]) mock_webhook.assert_not_called() @mock.patch("src.notification.get_config") def test_dingtalk_context_response_skips_static_webhook(self, mock_get_config: mock.MagicMock): cfg = _make_config( feishu_webhook_url="https://open.feishu.cn/open-apis/bot/v2/hook/test-token", dingtalk_app_key="dingtalk-key", dingtalk_app_secret="dingtalk-secret", wechat_webhook_url="https://wechat.example/hook", ) mock_get_config.return_value = cfg service = NotificationService(source_message=_make_dingtalk_message()) with mock.patch.object(service, "_send_dingtalk_chunked", return_value=True) as mock_dingtalk, \ mock.patch.object(service, "send_to_wechat", return_value=True) as mock_wechat: result = service.send_with_results("content", route_type="report") self.assertTrue(result.dispatched) self.assertTrue(result.success) self.assertEqual([item.channel for item in result.channel_results], ["__context__"]) mock_dingtalk.assert_called_once_with("https://oapi.dingtalk.com/robot/sendBySession?session=abc123", "content", max_bytes=20000) mock_wechat.assert_not_called() @mock.patch("src.notification.get_config") def test_telegram_context_response_skips_static_webhook(self, mock_get_config: mock.MagicMock): cfg = _make_config( telegram_bot_token="TOKEN", telegram_chat_id="123456", wechat_webhook_url="https://wechat.example/hook", ) mock_get_config.return_value = cfg service = NotificationService(source_message=_make_telegram_message()) with mock.patch.object(service, "send_to_telegram", return_value=True) as mock_telegram, \ mock.patch.object(service, "send_to_wechat", return_value=True) as mock_wechat: result = service.send_with_results("content", route_type="report") self.assertTrue(result.dispatched) self.assertTrue(result.success) self.assertEqual([item.channel for item in result.channel_results], ["__context__"]) mock_telegram.assert_called_once_with("content", chat_id="100200300") mock_wechat.assert_not_called() @mock.patch("src.notification.get_config") def test_feishu_webhook_still_sends_without_source_context(self, mock_get_config: mock.MagicMock): cfg = _make_config( feishu_webhook_url="https://open.feishu.cn/open-apis/bot/v2/hook/test-token", feishu_app_id="cli_test", feishu_app_secret="app-secret", ) mock_get_config.return_value = cfg service = NotificationService() with mock.patch.object(service, "send_to_feishu", return_value=True) as mock_webhook: result = service.send_with_results("content", route_type="report") self.assertTrue(result.dispatched) self.assertTrue(result.success) self.assertEqual([item.channel for item in result.channel_results], ["feishu"]) mock_webhook.assert_called_once_with("content") @mock.patch("src.notification.get_config") def test_feishu_send_as_file_route_report_calls_send_feishu_file(self, mock_get_config): """When FEISHU_SEND_AS_FILE=true and route_type=report, use file sending.""" cfg = _make_config( feishu_webhook_url="https://feishu.example/hook", feishu_send_as_file=True, ) mock_get_config.return_value = cfg service = NotificationService() with mock.patch.object(service, "send_feishu_file", return_value=True) as mock_file, \ mock.patch.object(service, "save_report_to_file", return_value="/tmp/report.md"): result = service.send_with_results("report content", route_type="report") self.assertTrue(result.success) mock_file.assert_called_once() @mock.patch("src.notification.get_config") def test_feishu_send_as_file_route_alert_calls_send_to_feishu(self, mock_get_config): """When FEISHU_SEND_AS_FILE=true but route_type=alert, use text sending.""" cfg = _make_config( feishu_webhook_url="https://feishu.example/hook", feishu_send_as_file=True, ) mock_get_config.return_value = cfg service = NotificationService() with mock.patch.object(service, "send_to_feishu", return_value=True) as mock_text, \ mock.patch.object(service, "save_report_to_file") as mock_save: result = service.send_with_results("alert content", route_type="alert") self.assertTrue(result.success) mock_text.assert_called_once_with("alert content") mock_save.assert_not_called() @mock.patch("src.notification.get_config") def test_feishu_send_as_file_route_none_uses_text(self, mock_get_config): """When FEISHU_SEND_AS_FILE=true and route_type=None, use text (not file).""" cfg = _make_config( feishu_webhook_url="https://feishu.example/hook", feishu_send_as_file=True, ) mock_get_config.return_value = cfg service = NotificationService() with mock.patch.object(service, "send_to_feishu", return_value=True) as mock_text, \ mock.patch.object(service, "save_report_to_file") as mock_save: result = service.send_with_results("report content") self.assertTrue(result.success) mock_text.assert_called_once_with("report content") mock_save.assert_not_called() @mock.patch("src.notification.get_config") def test_feishu_send_as_file_false_uses_text_even_for_report(self, mock_get_config): """When FEISHU_SEND_AS_FILE=false (default), report still uses text.""" cfg = _make_config( feishu_webhook_url="https://feishu.example/hook", feishu_send_as_file=False, ) mock_get_config.return_value = cfg service = NotificationService() with mock.patch.object(service, "send_to_feishu", return_value=True) as mock_text, \ mock.patch.object(service, "save_report_to_file") as mock_save: result = service.send_with_results("report content", route_type="report") self.assertTrue(result.success) mock_text.assert_called_once_with("report content") mock_save.assert_not_called() @mock.patch("src.notification.get_config") def test_feishu_send_as_file_alert_does_not_leak_into_other_channels(self, mock_get_config): """FEISHU_SEND_AS_FILE only affects Feishu, not other channels.""" cfg = _make_config( feishu_webhook_url="https://feishu.example/hook", custom_webhook_urls=["https://example.com/hook"], feishu_send_as_file=True, ) mock_get_config.return_value = cfg service = NotificationService() with mock.patch.object(service, "send_feishu_file", return_value=True) as mock_file, \ mock.patch.object(service, "send_to_custom", return_value=True) as mock_custom, \ mock.patch.object(service, "save_report_to_file", return_value="/tmp/report.md"): result = service.send_with_results("content", route_type="report") self.assertTrue(result.success) mock_file.assert_called_once() mock_custom.assert_called_once_with("content") @mock.patch("src.notification.get_config") def test_feishu_send_as_file_system_error_uses_text(self, mock_get_config): """FEISHU_SEND_AS_FILE does not activate for system_error routes.""" cfg = _make_config( feishu_webhook_url="https://feishu.example/hook", feishu_send_as_file=True, ) mock_get_config.return_value = cfg service = NotificationService() with mock.patch.object(service, "send_to_feishu", return_value=True) as mock_text, \ mock.patch.object(service, "save_report_to_file") as mock_save: result = service.send_with_results("error", route_type="system_error") self.assertTrue(result.success) mock_text.assert_called_once_with("error") mock_save.assert_not_called() @mock.patch("src.notification.get_config") def test_send_dedup_suppresses_static_channels_after_success(self, mock_get_config: mock.MagicMock): cfg = _make_config( custom_webhook_urls=["https://example.com/webhook"], notification_dedup_ttl_seconds=60, ) mock_get_config.return_value = cfg service = NotificationService() with mock.patch.object(service, "send_to_custom", return_value=True) as mock_custom: self.assertTrue(service.send("content at 12:00", route_type="report", dedup_key="report:aggregate:simple:600519")) self.assertFalse(service.send("content at 12:01", route_type="report", dedup_key="report:aggregate:simple:600519")) mock_custom.assert_called_once_with("content at 12:00") @mock.patch("src.notification.get_config") def test_send_releases_noise_reservation_when_static_channels_fail(self, mock_get_config: mock.MagicMock): cfg = _make_config( custom_webhook_urls=["https://example.com/webhook"], notification_dedup_ttl_seconds=60, ) mock_get_config.return_value = cfg service = NotificationService() with mock.patch.object(service, "send_to_custom", side_effect=[False, True]) as mock_custom: self.assertFalse( service.send( "content at 12:00", route_type="report", dedup_key="report:aggregate:simple:600519", ) ) self.assertTrue( service.send( "content at 12:01", route_type="report", dedup_key="report:aggregate:simple:600519", ) ) self.assertEqual(mock_custom.call_count, 2) @mock.patch("src.notification.get_config") def test_send_to_context_is_not_limited_by_noise_controls(self, mock_get_config: mock.MagicMock): cfg = _make_config( custom_webhook_urls=["https://example.com/webhook"], notification_dedup_ttl_seconds=60, ) mock_get_config.return_value = cfg service = NotificationService() with mock.patch.object(service, "send_to_context", return_value=True) as mock_context, \ mock.patch.object(service, "send_to_custom", return_value=True) as mock_custom: self.assertTrue(service.send("content at 12:00", route_type="report", dedup_key="report:aggregate:simple:600519")) self.assertTrue(service.send("content at 12:01", route_type="report", dedup_key="report:aggregate:simple:600519")) self.assertEqual(mock_context.call_count, 2) mock_custom.assert_called_once_with("content at 12:00") @mock.patch("src.notification.get_config") def test_noise_check_failure_does_not_block_static_send(self, mock_get_config: mock.MagicMock): cfg = _make_config(custom_webhook_urls=["https://example.com/webhook"]) mock_get_config.return_value = cfg service = NotificationService() with mock.patch("src.notification_noise._evaluate_notification_noise", side_effect=RuntimeError("boom")), \ mock.patch.object(service, "send_to_custom", return_value=True) as mock_custom: ok = service.send("content", route_type="report") self.assertTrue(ok) mock_custom.assert_called_once_with("content") @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_discord_via_notification_service_with_webhook( self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock ): cfg = _make_config(discord_webhook_url="https://discord.example/webhook") mock_get_config.return_value = cfg mock_post.return_value = _make_response(204) service = NotificationService() self.assertIn(NotificationChannel.DISCORD, service.get_available_channels()) ok = service.send("discord content") self.assertTrue(ok) mock_post.assert_called_once() @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_discord_via_notification_service_with_bot( self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock ): cfg = _make_config(discord_bot_token="TOKEN", discord_main_channel_id="123") mock_get_config.return_value = cfg mock_post.return_value = _make_response(200) service = NotificationService() self.assertIn(NotificationChannel.DISCORD, service.get_available_channels()) ok = service.send("discord content") self.assertTrue(ok) mock_post.assert_called_once() @mock.patch("src.notification.get_config") @mock.patch("src.notification_sender.discord_sender.time.sleep", return_value=None) @mock.patch("requests.post") def test_send_to_discord_via_notification_service_with_bot_requires_chunking( self, mock_post: mock.MagicMock, _mock_sleep: mock.MagicMock, mock_get_config: mock.MagicMock, ): cfg = _make_config( discord_bot_token="TOKEN", discord_main_channel_id="123", discord_max_words=2000, ) mock_get_config.return_value = cfg mock_post.return_value = _make_response(200) service = NotificationService() self.assertIn(NotificationChannel.DISCORD, service.get_available_channels()) ok = service.send("A" * 6000) self.assertTrue(ok) self.assertAlmostEqual(mock_post.call_count, 4, delta=1) class TestNotificationServiceReportGeneration(unittest.TestCase): """报告生成与选路相关测试。""" def test_signal_metadata_uses_resolved_eight_state_action(self): service = NotificationService() cases = [ ("avoid", "Avoid", 90, ("Avoid", "🟡", "hold")), ("add", "Add", 50, ("Add", "🟢", "buy")), ("alert", "Alert", 85, ("Alert", "🔴", "sell")), ] for action, action_label, score, expected in cases: with self.subTest(action=action): result = AnalysisResult( code="AAPL", name="Apple", sentiment_score=score, trend_prediction="Neutral", operation_advice="Hold", report_language="en", action=action, action_label=action_label, ) self.assertEqual(service._get_signal_level(result), expected) def test_build_stock_summary_uses_resolved_eight_state_action(self): summary = NotificationBuilder.build_stock_summary( [ AnalysisResult( code="AVOID", name="Avoid Corp", sentiment_score=90, trend_prediction="Neutral", operation_advice="Avoid", report_language="en", action="avoid", action_label="Avoid", ), AnalysisResult( code="ALERT", name="Alert Corp", sentiment_score=85, trend_prediction="Neutral", operation_advice="Alert", report_language="en", action="alert", action_label="Alert", ), AnalysisResult( code="ADD", name="Add Corp", sentiment_score=50, trend_prediction="Neutral", operation_advice="Add", report_language="en", action="add", action_label="Add", ), ] ) self.assertIn("🟡 Avoid Corp(AVOID): Avoid | Score 90", summary) self.assertIn("🔴 Alert Corp(ALERT): Alert | Score 85", summary) self.assertIn("🟢 Add Corp(ADD): Add | Score 50", summary) self.assertNotIn("Buy | Score 50", summary) self.assertNotIn("Strong Buy", summary) @mock.patch("src.notification.get_config") def test_report_rows_and_summary_use_same_score_aligned_action( self, mock_get_config: mock.MagicMock ): mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="AAPL", name="Apple", sentiment_score=72, trend_prediction="Bullish", operation_advice="Hold", decision_type="hold", report_language="en", ) out = service.generate_brief_report([result], report_date="2026-07-11") self.assertIn("🟢1 🟡0 🔴0", out) self.assertIn("Buy | Score 72", out) self.assertNotIn("Hold | Score 72", out) @mock.patch("src.notification.get_config") def test_generate_aggregate_report_routes_by_report_type(self, mock_get_config: mock.MagicMock): mock_get_config.return_value = _make_config() service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", ) with mock.patch.object(service, "generate_dashboard_report", return_value="dashboard") as mock_dashboard, mock.patch.object( service, "generate_brief_report", return_value="brief" ) as mock_brief: self.assertEqual(service.generate_aggregate_report([result], "simple"), "dashboard") self.assertEqual(service.generate_aggregate_report([result], "full"), "dashboard") self.assertEqual(service.generate_aggregate_report([result], "detailed"), "dashboard") self.assertEqual(service.generate_aggregate_report([result], "brief"), "brief") self.assertEqual(mock_dashboard.call_count, 3) mock_brief.assert_called_once() @mock.patch("src.notification.get_config") def test_generate_single_stock_report_keeps_legacy_simple_format(self, mock_get_config: mock.MagicMock): mock_get_config.return_value = _make_config(report_renderer_enabled=True) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", ) with mock.patch("src.services.report_renderer.render") as mock_render: out = service.generate_single_stock_report(result) mock_render.assert_not_called() self.assertIn("贵州茅台", out) self.assertIn("600519", out) @mock.patch("src.notification.get_config") def test_generate_brief_report_shows_model_by_default(self, mock_get_config: mock.MagicMock): mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", model_used="gemini/gemini-2.5-flash", ) out = service.generate_brief_report([result], report_date="2026-02-01") self.assertIn("*分析模型: gemini/gemini-2.5-flash*", out) @mock.patch("src.notification.get_config") def test_generate_dashboard_report_shows_model_by_default(self, mock_get_config: mock.MagicMock): mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", model_used="gemini/gemini-2.5-flash", ) out = service.generate_dashboard_report([result], report_date="2026-02-01") self.assertIn("*分析模型：gemini/gemini-2.5-flash*", out) @mock.patch("src.notification.get_config") def test_generate_dashboard_report_shows_phase_decision_in_default_renderer( self, mock_get_config: mock.MagicMock ): mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="等待确认", dashboard={ "core_conclusion": {"one_sentence": "等待确认"}, "phase_decision": { "action_window": "盘中跟踪", "immediate_action": "等待确认", "watch_conditions": ["放量突破"], "next_check_time": "14:30", "confidence_reason": "数据质量可用", "data_limitations": ["quote: stale"], }, }, ) out = service.generate_dashboard_report([result], report_date="2026-02-01") self.assertIn("盘中决策护栏", out) self.assertIn("盘中跟踪", out) self.assertIn("等待确认", out) self.assertIn("放量突破", out) self.assertIn("quote: stale", out) @mock.patch("src.notification.get_config") def test_generate_dashboard_report_skips_context_only_phase_decision_default_renderer( self, mock_get_config: mock.MagicMock ): mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="等待确认", dashboard={ "core_conclusion": {"one_sentence": "等待确认"}, "phase_decision": { "phase_context": {"phase": "intraday", "market": "cn"}, "watch_conditions": [], "data_limitations": [], }, }, ) out = service.generate_dashboard_report([result], report_date="2026-02-01") self.assertNotIn("盘中决策护栏", out) @mock.patch("src.notification.get_config") def test_generate_dashboard_report_omits_decision_signal_excerpt_fallback( self, mock_get_config: mock.MagicMock ): mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = _attach_decision_signal_summary(AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", )) out = service.generate_dashboard_report([result], report_date="2026-02-01") self.assertNotIn("AI 决策信号", out) self.assertNotIn("理由: 技术面走弱", out) @mock.patch("src.notification.get_config") def test_generate_daily_report_omits_decision_signal_excerpt_fallback( self, mock_get_config: mock.MagicMock ): mock_get_config.return_value = _make_config(report_renderer_enabled=False) result = _attach_decision_signal_summary(AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", )) for summary_only in (True, False): service = NotificationService() service._report_summary_only = summary_only out = service.generate_daily_report([result], report_date="2026-02-01") self.assertNotIn("AI 决策信号", out) self.assertNotIn("理由: 技术面走弱", out) @mock.patch("src.notification.get_config") def test_generate_wechat_dashboard_omits_decision_signal_excerpt_fallback( self, mock_get_config: mock.MagicMock ): mock_get_config.return_value = _make_config(report_renderer_enabled=False) result = _attach_decision_signal_summary(AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", )) for summary_only in (True, False): service = NotificationService() service._report_summary_only = summary_only out = service.generate_wechat_dashboard([result]) self.assertNotIn("AI 决策信号", out) self.assertNotIn("理由: 技术面走弱", out) @mock.patch("src.notification.get_config") def test_generate_wechat_summary_omits_decision_signal_excerpt( self, mock_get_config: mock.MagicMock ): mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = _attach_decision_signal_summary(AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", )) out = service.generate_wechat_summary([result]) self.assertNotIn("AI 决策信号", out) self.assertNotIn("动作: 卖出", out) @mock.patch("src.notification.get_config") def test_generate_dashboard_report_omits_decision_signal_excerpt_with_renderer( self, mock_get_config: mock.MagicMock ): mock_get_config.return_value = _make_config(report_renderer_enabled=True) service = NotificationService() result = _attach_decision_signal_summary(AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", )) out = service.generate_dashboard_report([result], report_date="2026-02-01") self.assertNotIn("AI 决策信号", out) self.assertNotIn("理由: 技术面走弱", out) @mock.patch("src.notification.get_config") def test_aggregate_reports_show_compact_market_status_only(self, mock_get_config: mock.MagicMock): mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", ) result.market_phase_summary = { "phase": "intraday", "market": "cn", "trigger_source": "portfolio", "is_partial_bar": True, } result.analysis_context_pack_overview = { "data_quality": { "level": "limited", "limitations": ["quote: stale", "news: missing", "portfolio_context: hidden"], } } result.raw_response = "raw context pack and prompt should not appear" out = service.generate_brief_report([result], report_date="2026-02-01") self.assertIn("市场状态：A股 · 盘中", out) self.assertNotIn("阶段：intraday", out) self.assertNotIn("触发来源：portfolio", out) self.assertNotIn("盘中数据提示", out) self.assertNotIn("数据质量: limited", out) self.assertNotIn("限制: quote: stale", out) self.assertNotIn("限制: news: missing", out) self.assertNotIn("portfolio_context: hidden", out) self.assertNotIn("raw context pack", out) self.assertNotIn("prompt", out.lower()) @mock.patch("src.notification.get_config") def test_template_dashboard_report_uses_single_market_status_line(self, mock_get_config: mock.MagicMock): mock_get_config.return_value = _make_config(report_renderer_enabled=True) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", ) result.market_phase_summary = { "phase": "postmarket", "market": "cn", "trigger_source": "cli", } result.analysis_context_pack_overview = { "data_quality": { "level": "good", "limitations": ["technical: partial"], } } out = service.generate_dashboard_report([result], report_date="2026-02-01") self.assertIn("市场状态：A股 · 盘后", out) self.assertEqual(out.count("市场状态："), 1) self.assertNotIn("阶段：postmarket", out) self.assertNotIn("触发来源：cli", out) self.assertNotIn("数据质量: good", out) self.assertNotIn("technical: partial", out) @mock.patch("src.notification.get_config") def test_generated_reports_skip_phase_pack_excerpt_when_summary_missing(self, mock_get_config: mock.MagicMock): mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", ) out = service.generate_brief_report([result], report_date="2026-02-01") self.assertNotIn("摘要来源", out) self.assertNotIn("评估器快照", out) @mock.patch("src.notification.get_config") def test_generate_dashboard_report_collapses_unavailable_chip_structure(self, mock_get_config: mock.MagicMock): mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", dashboard={ "data_perspective": { "chip_structure": { "profit_ratio": "数据缺失，无法判断", "avg_cost": "数据缺失，无法判断", "concentration": "数据缺失，无法判断", "chip_health": "数据缺失，无法判断", } } }, ) out = service.generate_dashboard_report([result], report_date="2026-02-01") self.assertIn("**筹码**: 筹码分布未启用或数据源暂不可用，未纳入筹码判断。", out) self.assertEqual(out.count("数据缺失，无法判断"), 0) @mock.patch("src.notification.get_config") def test_generate_reports_hide_model_when_disabled(self, mock_get_config: mock.MagicMock): mock_get_config.return_value = _make_config( report_renderer_enabled=False, report_show_llm_model=False, ) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", model_used="gemini/gemini-2.5-flash", ) dashboard = service.generate_dashboard_report([result], report_date="2026-02-01") single = service.generate_single_stock_report(result) self.assertNotIn("分析模型", dashboard) self.assertNotIn("gemini/gemini-2.5-flash", dashboard) self.assertNotIn("分析模型", single) self.assertNotIn("gemini/gemini-2.5-flash", single) @mock.patch("src.notification.get_config") def test_generate_dashboard_report_localizes_english_fallback(self, mock_get_config: mock.MagicMock): mock_get_config.return_value = _make_config(report_renderer_enabled=False, report_language="en") service = NotificationService() result = AnalysisResult( code="AAPL", name="Apple", sentiment_score=78, trend_prediction="Bullish", operation_advice="Buy", analysis_summary="Momentum remains constructive.", decision_type="buy", report_language="en", dashboard={ "core_conclusion": { "one_sentence": "Favor buying on pullbacks.", "position_advice": { "no_position": "Open a starter position.", "has_position": "Hold and trail the stop.", }, }, "battle_plan": { "sniper_points": { "ideal_buy": "180-182", "stop_loss": "172", "take_profit": "195", } }, }, ) out = service.generate_dashboard_report([result], report_date="2026-03-18") self.assertIn("Decision Dashboard", out) self.assertIn("Summary", out) self.assertIn("Action Levels", out) self.assertIn("Buy", out) @mock.patch("src.notification.get_config") def test_generate_dashboard_report_localizes_english_no_dashboard_fallback( self, mock_get_config: mock.MagicMock ): mock_get_config.return_value = _make_config(report_renderer_enabled=False, report_language="en") service = NotificationService() result = AnalysisResult( code="AAPL", name="Apple", sentiment_score=61, trend_prediction="看多", operation_advice="持有", analysis_summary="Wait for confirmation.", report_language="en", buy_reason="Momentum remains constructive.", risk_warning="Watch for a failed breakout.", ma_analysis="Price remains above MA20.", volume_analysis="Volume is steady.", news_summary="Product cycle remains supportive.", ) out = service.generate_dashboard_report([result], report_date="2026-03-19") self.assertIn("Rationale", out) self.assertIn("Risk Warning", out) self.assertIn("Technicals", out) self.assertIn("Moving Averages", out) self.assertIn("Volume", out) self.assertIn("News Flow", out) self.assertNotIn("操作理由", out) self.assertNotIn("风险提示", out) self.assertNotIn("技术面", out) self.assertNotIn("消息面", out) @mock.patch("src.notification.get_config") def test_generate_single_stock_report_aligns_english_fallback_with_score(self, mock_get_config: mock.MagicMock): mock_get_config.return_value = _make_config(report_renderer_enabled=False, report_language="en") service = NotificationService() result = AnalysisResult( code="AAPL", name="Apple", sentiment_score=65, trend_prediction="Sideways", operation_advice="Hold", analysis_summary="Wait for a cleaner breakout.", report_language="en", dashboard={ "core_conclusion": {"one_sentence": "Wait for confirmation."}, "battle_plan": { "sniper_points": { "ideal_buy": "190", "stop_loss": "182", "take_profit": "205", } }, }, ) out = service.generate_single_stock_report(result) self.assertIn("Core Conclusion", out) self.assertIn("Action Levels", out) self.assertIn("Buy", out) def _make_fundamental_context(self) -> dict: return { "earnings": { "status": "ok", "data": { "financial_report": { "report_date": "2024-09-30", "revenue": 1_236_000_000_000.0, # 1.236 万亿 -> 12360.00 亿元 "net_profit_parent": 60_800_000_000.0, "operating_cash_flow": 72_500_000_000.0, "roe": 22.45, }, "dividend": { "events": [ { "event_date": "2024-06-26", "ex_dividend_date": "2024-06-26", "cash_dividend_per_share": 30.876, "is_pre_tax": True, } ], "ttm_event_count": 1, "ttm_cash_dividend_per_share": 30.876, "ttm_dividend_yield_pct": 1.85, }, }, }, "growth": { "status": "ok", "data": { "revenue_yoy": 15.23, "net_profit_yoy": 19.87, "roe": 22.45, "gross_margin": 91.55, }, }, "boards": { "status": "ok", "data": { "top": [ {"name": "白酒", "change_pct": 3.42}, {"name": "食品饮料", "change_pct": 2.10}, ], "bottom": [ {"name": "光伏设备", "change_pct": -2.65}, ], }, }, "concept_boards": { "status": "ok", "data": { "top": [ {"name": "MSCI中国", "change_pct": 1.23}, ], "bottom": [], }, }, "belong_boards": [ {"name": "白酒", "code": "BK0596", "type": "行业"}, {"name": "MSCI中国", "code": "BK0805", "type": "概念"}, ], } @mock.patch("src.notification.get_config") def test_generate_single_stock_report_appends_fundamental_blocks( self, mock_get_config: mock.MagicMock ): mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", ) result.fundamental_context = self._make_fundamental_context() out = service.generate_single_stock_report(result) # 财务摘要 self.assertIn("财务摘要", out) self.assertIn("2024-09-30", out) self.assertIn("12360.00 亿元", out) self.assertIn("22.45%", out) self.assertIn("15.23%", out) self.assertIn("91.55%", out) # 股东回报 self.assertIn("股东回报", out) self.assertIn("30.8760 元", out) self.assertIn("1.85%", out) self.assertIn("2024-06-26", out) # 关联板块（白酒带行业信号；MSCI中国 带概念信号） self.assertIn("关联板块", out) self.assertIn("白酒", out) self.assertIn("领涨", out) self.assertIn("+3.42%", out) self.assertIn("MSCI中国", out) self.assertIn("- 白酒 (行业板块 领涨 +3.42%)", out) self.assertIn("- MSCI中国 (概念板块 领涨 +1.23%)", out) self.assertIn("+1.23%", out) self.assertNotIn("| 板块 | 类型 | 板块表现 | 板块涨跌幅 |", out) @mock.patch("src.notification.get_config") def test_related_boards_uses_concept_rankings_for_concept_boards( self, mock_get_config: mock.MagicMock ): mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", ) result.fundamental_context = { "boards": {"status": "ok", "data": { "top": [{"name": "白酒", "change_pct": 2.31}], "bottom": [], }}, "concept_boards": {"status": "ok", "data": { "top": [], "bottom": [{"name": "白酒", "change_pct": -3.2}], }}, "belong_boards": [{"name": "白酒", "type": "概念"}], } out = service.generate_single_stock_report(result) self.assertIn("关联板块", out) self.assertIn("- 白酒 (概念板块 领跌 -3.20%)", out) self.assertNotIn("| 白酒 | 概念 |", out) self.assertNotIn("+2.31%", out) @mock.patch("src.notification.get_config") def test_generate_single_stock_report_skips_fundamental_blocks_when_missing( self, mock_get_config: mock.MagicMock ): mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", ) out = service.generate_single_stock_report(result) self.assertNotIn("财务摘要", out) self.assertNotIn("股东回报", out) self.assertNotIn("关联板块", out) @mock.patch("src.notification.get_config") def test_generate_single_stock_report_handles_partial_fundamental_context( self, mock_get_config: mock.MagicMock ): """Only dividend data present — render shareholder return, skip the other two.""" mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", ) result.fundamental_context = { "earnings": { "status": "partial", "data": { "dividend": { "events": [], "ttm_event_count": 0, "ttm_cash_dividend_per_share": 0.5, } }, }, } out = service.generate_single_stock_report(result) self.assertNotIn("财务摘要", out) self.assertIn("股东回报", out) self.assertIn("0.5000 元", out) self.assertNotIn("关联板块", out) @mock.patch("src.notification.get_config") def test_generate_single_stock_report_uses_currency_for_us( self, mock_get_config: mock.MagicMock ): """USD currency on financial_report yields 亿美元 suffix instead of 亿元.""" mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="AAPL", name="Apple Inc.", sentiment_score=64, trend_prediction="震荡", operation_advice="观望", analysis_summary="观望等待 AI 兑现节奏。", ) result.fundamental_context = { "earnings": { "status": "ok", "data": { "financial_report": { "report_date": "2026-03-31", "revenue": 1.11e11, "net_profit_parent": 2.95e10, "operating_cash_flow": 2.87e10, "roe": 141.47, "currency": "USD", }, "dividend": { "events": [{ "event_date": "2026-05-11", "ex_dividend_date": "2026-05-11", "cash_dividend_per_share": 0.27, "is_pre_tax": True, }], "ttm_event_count": 4, "ttm_cash_dividend_per_share": 1.05, "ttm_dividend_yield_pct": 0.36, }, }, }, "growth": {"status": "ok", "data": {"revenue_yoy": 16.60, "roe": 141.47, "gross_margin": 47.86}}, "belong_boards": [ {"name": "Technology", "type": "行业"}, {"name": "Consumer Electronics", "type": "概念"}, ], } out = service.generate_single_stock_report(result) self.assertIn("财务摘要", out) self.assertIn("亿美元", out) self.assertNotIn("12360.00 亿元", out) # Sample expected formatted values self.assertIn("1110.00 亿美元", out) self.assertIn("141.47%", out) # Dividend per share also picks up currency suffix self.assertIn("1.0500 美元", out) # Sector + industry render as belong_boards self.assertIn("Technology", out) self.assertIn("Consumer Electronics", out) @mock.patch("src.notification.get_config") def test_related_boards_drops_signal_columns_when_no_sector_data( self, mock_get_config: mock.MagicMock ): """HK/US lack 板块涨跌榜 — drop status / change_pct columns entirely.""" mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="AAPL", name="Apple Inc.", sentiment_score=64, trend_prediction="震荡", operation_advice="观望", analysis_summary="观望等待 AI 兑现节奏。", ) result.fundamental_context = { "earnings": {"status": "ok", "data": { "financial_report": { "report_date": "2026-03-31", "revenue": 1.11e11, "currency": "USD", }, }}, "growth": {"status": "ok", "data": {"revenue_yoy": 16.60}}, "belong_boards": [ {"name": "Technology", "type": "行业"}, {"name": "Consumer Electronics", "type": "概念"}, ], } out = service.generate_single_stock_report(result) self.assertIn("关联板块", out) self.assertIn("Technology / Consumer Electronics", out) self.assertIn("Technology", out) self.assertIn("Consumer Electronics", out) # When no sector ranking data is available, drop the 4-col layout. self.assertNotIn("板块表现", out) self.assertNotIn("板块涨跌幅", out) # And no table/type noise either. self.assertNotIn("| 板块 | 类型 |", out) self.assertNotIn("| -- | -- |", out) @mock.patch("src.notification.get_config") def test_related_boards_without_type_infers_concepts_and_keeps_unknowns( self, mock_get_config: mock.MagicMock ): mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", ) result.fundamental_context = { "earnings": {"status": "ok", "data": {}}, "growth": {"status": "ok", "data": {}}, "belong_boards": [ {"name": "白酒Ⅲ"}, {"name": "白酒Ⅱ"}, {"name": "食品饮料"}, {"name": "贵州板块"}, {"name": "酿酒概念"}, ], } out = service.generate_single_stock_report(result) self.assertIn("关联板块", out) self.assertIn("白酒Ⅲ / 白酒Ⅱ / 食品饮料 / 贵州板块 / 酿酒概念", out) self.assertNotIn("| 板块 | 类型 |", out) self.assertNotIn("| 白酒Ⅲ | N/A |", out) @mock.patch("src.notification.get_config") def test_related_boards_renders_each_board_signal_without_placeholder( self, mock_get_config: mock.MagicMock ): """Rows without a matching change_pct stay as plain board entries.""" mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", ) result.fundamental_context = { "earnings": {"status": "ok", "data": {}}, "growth": {"status": "ok", "data": {}}, "boards": {"status": "ok", "data": { "top": [{"name": "白酒", "change_pct": 3.42}], "bottom": [], }}, "belong_boards": [ {"name": "白酒", "code": "BK0596", "type": "行业"}, {"name": "MSCI中国", "code": "BK0805", "type": "概念"}, ], } out = service.generate_single_stock_report(result) self.assertNotIn("板块表现", out) self.assertNotIn("板块涨跌幅", out) self.assertIn("领涨", out) self.assertIn("+3.42%", out) self.assertIn("- 白酒 (行业板块 领涨 +3.42%)", out) self.assertIn("MSCI中国", out) self.assertIn("- MSCI中国", out) self.assertNotIn("- MSCI中国 (", out) self.assertNotIn("| MSCI中国 | 概念 | -- | -- |", out) @mock.patch("src.notification.get_config") def test_related_boards_ignores_matching_signal_without_change_pct( self, mock_get_config: mock.MagicMock ): mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", ) result.fundamental_context = { "earnings": {"status": "ok", "data": {}}, "growth": {"status": "ok", "data": {}}, "boards": {"status": "ok", "data": { "top": [{"name": "白酒"}], "bottom": [], }}, "belong_boards": [ {"name": "白酒", "code": "BK0596", "type": "行业"}, ], } out = service.generate_single_stock_report(result) self.assertIn("关联板块", out) self.assertIn("白酒", out) self.assertNotIn("| 白酒 | 行业 |", out) self.assertNotIn("领涨", out) self.assertNotIn("板块涨跌幅", out) @mock.patch("src.notification.get_config") def test_generate_single_stock_report_uses_currency_for_hk( self, mock_get_config: mock.MagicMock ): """HK ADRs have financialCurrency=CNY but trade/pay dividends in HKD. The financial summary must render in 元 (CNY income statement) while dividends must render in 港元 — they are NOT the same currency on yfinance HK payloads, so the renderer must read each block's own ``currency`` field rather than assuming a single global currency. """ mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="HK09988", name="阿里巴巴-W", sentiment_score=68, trend_prediction="看多", operation_advice="逢低买入", analysis_summary="云业务回正，回购持续。", ) result.fundamental_context = { "earnings": { "status": "ok", "data": { "financial_report": { "report_date": "2026-03-31", "revenue": 1.02e12, "net_profit_parent": 1.04e11, "operating_cash_flow": 3.6e10, "roe": 9.22, "currency": "CNY", }, "dividend": { "events": [{ "event_date": "2025-06-11", "ex_dividend_date": "2025-06-11", "cash_dividend_per_share": 1.95812, "is_pre_tax": True, }], "ttm_event_count": 1, "ttm_cash_dividend_per_share": 1.95812, "ttm_dividend_yield_pct": 1.75, "currency": "HKD", }, }, }, "growth": {"status": "ok", "data": {"revenue_yoy": 2.9, "roe": 9.22, "gross_margin": 39.81}}, "belong_boards": [ {"name": "Consumer Cyclical", "type": "行业"}, {"name": "Internet Retail", "type": "概念"}, ], } out = service.generate_single_stock_report(result) # Income statement still rendered in CNY (financialCurrency). self.assertIn("10200.00 亿元", out) # Dividend per share follows the dividend currency, NOT the financial currency. self.assertIn("1.9581 港元", out) self.assertNotIn("1.9581 元 ", out) self.assertIn("Consumer Cyclical", out) @mock.patch("src.notification.get_config") def test_dividend_currency_falls_back_to_financial_when_missing( self, mock_get_config: mock.MagicMock ): """A-share AkShare payload has no dividend.currency — fall back to financial_report.currency.""" mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", ) result.fundamental_context = { "earnings": { "status": "ok", "data": { "financial_report": {"report_date": "2026-03-31", "revenue": 4.5e10}, "dividend": { "events": [{"event_date": "2025-06-20", "ex_dividend_date": "2025-06-20"}], "ttm_event_count": 1, "ttm_cash_dividend_per_share": 27.6, }, }, }, } out = service.generate_single_stock_report(result) # Without explicit dividend currency, default to 元 (matches AkShare A-share semantics). self.assertIn("27.6000 元", out) @mock.patch("src.notification.get_config") def test_generate_dashboard_report_appends_fundamental_blocks( self, mock_get_config: mock.MagicMock ): mock_get_config.return_value = _make_config(report_renderer_enabled=False) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", dashboard={ "core_conclusion": {"one_sentence": "稳健持有"}, "battle_plan": { "sniper_points": { "ideal_buy": "1600", "stop_loss": "1500", "take_profit": "1800", } }, }, ) result.fundamental_context = self._make_fundamental_context() out = service.generate_dashboard_report([result], report_date="2026-05-20") self.assertIn("财务摘要", out) self.assertIn("股东回报", out) self.assertIn("关联板块", out) self.assertIn("白酒", out) self.assertIn("领涨", out) @mock.patch("src.notification.get_config") def test_history_compare_context_uses_cache(self, mock_get_config: mock.MagicMock): mock_get_config.return_value = _make_config(report_history_compare_n=3) service = NotificationService() result = AnalysisResult( code="600519", name="贵州茅台", sentiment_score=72, trend_prediction="看多", operation_advice="持有", analysis_summary="稳健", query_id="q-1", ) with mock.patch( "src.services.history_comparison_service.get_signal_changes_batch", return_value={"600519": []}, ) as mock_batch: first = service._get_history_compare_context([result]) second = service._get_history_compare_context([result]) self.assertEqual(first, {"history_by_code": {"600519": []}}) self.assertEqual(second, {"history_by_code": {"600519": []}}) mock_batch.assert_called_once() @mock.patch("src.notification.get_config") @mock.patch("smtplib.SMTP_SSL") def test_send_to_email_via_notification_service( self, mock_smtp_ssl: mock.MagicMock, mock_get_config: mock.MagicMock ): cfg = _make_config( email_sender="user@qq.com", email_password="PASS", email_receivers=["default@example.com"], ) mock_get_config.return_value = cfg service = NotificationService() self.assertIn(NotificationChannel.EMAIL, service.get_available_channels()) ok = service.send("email content") self.assertTrue(ok) mock_smtp_ssl.assert_called_once() mock_smtp_ssl.return_value.send_message.assert_called_once() @mock.patch("src.notification.get_config") @mock.patch("smtplib.SMTP_SSL") def test_send_to_email_with_stock_group_routing( self, mock_smtp_ssl: mock.MagicMock, mock_get_config: mock.MagicMock ): cfg = _make_config( email_sender="user@qq.com", email_password="PASS", email_receivers=["default@example.com"], stock_email_groups=[(["000001", "600519"], ["group@example.com"])], ) mock_get_config.return_value = cfg service = NotificationService() self.assertIn(NotificationChannel.EMAIL, service.get_available_channels()) server = mock_smtp_ssl.return_value ok = service.send("content", email_stock_codes=["000001"]) self.assertTrue(ok) mock_smtp_ssl.assert_called_once() server.send_message.assert_called_once() msg = server.send_message.call_args[0][0] self.assertIn("group@example.com", msg["To"]) @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_feishu_via_notification_service(self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock): cfg = _make_config(feishu_webhook_url="https://feishu.example") mock_get_config.return_value = cfg mock_post.return_value = _make_response(200, {"code": 0}) service = NotificationService() self.assertIn(NotificationChannel.FEISHU, service.get_available_channels()) ok = service.send("hello feishu") self.assertTrue(ok) mock_post.assert_called_once() @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_feishu_via_notification_service_requires_chunking(self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock): cfg = _make_config(feishu_webhook_url="https://feishu.example", feishu_max_bytes=2000) mock_get_config.return_value = cfg mock_post.return_value = _make_response(200, {"code": 0}) service = NotificationService() self.assertIn(NotificationChannel.FEISHU, service.get_available_channels()) ok = service.send("A" * 6000) self.assertTrue(ok) self.assertAlmostEqual(mock_post.call_count, 4, delta=1) @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_gotify_via_notification_service( self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock ): cfg = _make_config(gotify_url="https://gotify.example", gotify_token="secret-token") mock_get_config.return_value = cfg mock_post.return_value = _make_response(200) service = NotificationService() self.assertIn(NotificationChannel.GOTIFY, service.get_available_channels()) ok = service.send("gotify content") self.assertTrue(ok) mock_post.assert_called_once() self.assertEqual(mock_post.call_args.args[0], "https://gotify.example/message") self.assertEqual(mock_post.call_args.kwargs["headers"]["X-Gotify-Key"], "secret-token") self.assertEqual(mock_post.call_args.kwargs["json"]["message"], "gotify content") self.assertEqual( mock_post.call_args.kwargs["json"]["extras"]["client::display"]["contentType"], "text/markdown", ) @mock.patch("src.notification.get_config") def test_gotify_without_token_is_not_available(self, mock_get_config: mock.MagicMock): mock_get_config.return_value = _make_config(gotify_url="https://gotify.example") service = NotificationService() self.assertNotIn(NotificationChannel.GOTIFY, service.get_available_channels()) self.assertFalse(service.is_available()) @mock.patch("src.notification.get_config") def test_gotify_blank_token_is_not_available(self, mock_get_config: mock.MagicMock): mock_get_config.return_value = _make_config( gotify_url="https://gotify.example", gotify_token=" ", ) service = NotificationService() self.assertNotIn(NotificationChannel.GOTIFY, service.get_available_channels()) self.assertFalse(service.is_available()) @mock.patch("src.notification.get_config") def test_gotify_message_endpoint_is_not_available(self, mock_get_config: mock.MagicMock): mock_get_config.return_value = _make_config( gotify_url="https://gotify.example/message", gotify_token="secret-token", ) service = NotificationService() self.assertNotIn(NotificationChannel.GOTIFY, service.get_available_channels()) self.assertFalse(service.is_available()) @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_gotify_does_not_trigger_markdown_to_image( self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock ): cfg = _make_config( gotify_url="https://gotify.example", gotify_token="secret-token", markdown_to_image_channels=["gotify"], ) mock_get_config.return_value = cfg mock_post.return_value = _make_response(200) service = NotificationService() with mock.patch("src.md2img.markdown_to_image", return_value=b"png") as mock_md2img: ok = service.send("gotify content") self.assertTrue(ok) mock_md2img.assert_not_called() mock_post.assert_called_once() @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_ntfy_via_notification_service( self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock ): cfg = _make_config(ntfy_url="https://ntfy.sh/dsa-topic") mock_get_config.return_value = cfg mock_post.return_value = _make_response(200) service = NotificationService() self.assertIn(NotificationChannel.NTFY, service.get_available_channels()) ok = service.send("ntfy content") self.assertTrue(ok) mock_post.assert_called_once() self.assertEqual(mock_post.call_args.args[0], "https://ntfy.sh") self.assertEqual(mock_post.call_args.kwargs["json"]["topic"], "dsa-topic") @mock.patch("src.notification.get_config") def test_ntfy_url_without_topic_is_not_available(self, mock_get_config: mock.MagicMock): mock_get_config.return_value = _make_config(ntfy_url="https://ntfy.sh") service = NotificationService() self.assertNotIn(NotificationChannel.NTFY, service.get_available_channels()) self.assertFalse(service.is_available()) @mock.patch("src.notification.get_config") def test_ntfy_url_with_unsupported_scheme_is_not_available( self, mock_get_config: mock.MagicMock ): mock_get_config.return_value = _make_config(ntfy_url="ntfy://ntfy.sh/dsa-topic") service = NotificationService() self.assertNotIn(NotificationChannel.NTFY, service.get_available_channels()) self.assertFalse(service.is_available()) @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_ntfy_does_not_trigger_markdown_to_image( self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock ): cfg = _make_config( ntfy_url="https://ntfy.sh/dsa-topic", markdown_to_image_channels=["ntfy"], ) mock_get_config.return_value = cfg mock_post.return_value = _make_response(200) service = NotificationService() with mock.patch("src.md2img.markdown_to_image", return_value=b"png") as mock_md2img: ok = service.send("ntfy content") self.assertTrue(ok) mock_md2img.assert_not_called() mock_post.assert_called_once() @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_pushover_via_notification_service( self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock ): cfg = _make_config( pushover_user_key="USER", pushover_api_token="TOKEN", ) mock_get_config.return_value = cfg mock_post.return_value = _make_response(200, {"status": 1}) service = NotificationService() self.assertIn(NotificationChannel.PUSHOVER, service.get_available_channels()) ok = service.send("pushover content") self.assertTrue(ok) mock_post.assert_called_once() @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_pushplus_via_notification_service( self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock ): cfg = _make_config(pushplus_token="TOKEN") mock_get_config.return_value = cfg mock_post.return_value = _make_response(200, {"code": 200}) service = NotificationService() self.assertIn(NotificationChannel.PUSHPLUS, service.get_available_channels()) ok = service.send("pushplus content") self.assertTrue(ok) mock_post.assert_called_once() @mock.patch("src.notification_sender.pushplus_sender.time.sleep") @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_pushplus_via_notification_service_requires_chunking( self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock, _mock_sleep: mock.MagicMock, ): cfg = _make_config(pushplus_token="TOKEN") mock_get_config.return_value = cfg mock_post.return_value = _make_response(200, {"code": 200}) service = NotificationService() self.assertIn(NotificationChannel.PUSHPLUS, service.get_available_channels()) ok = service.send("A" * 25000) self.assertTrue(ok) self.assertGreaterEqual(mock_post.call_count, 2) @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_slack_via_notification_service_with_webhook( self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock ): cfg = _make_config(slack_webhook_url="https://hooks.slack.com/services/T/B/xxx") mock_get_config.return_value = cfg resp = mock.MagicMock() resp.status_code = 200 resp.text = "ok" mock_post.return_value = resp service = NotificationService() self.assertIn(NotificationChannel.SLACK, service.get_available_channels()) ok = service.send("slack content") self.assertTrue(ok) mock_post.assert_called_once() @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_slack_via_notification_service_with_bot( self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock ): cfg = _make_config(slack_bot_token="xoxb-test", slack_channel_id="C123") mock_get_config.return_value = cfg mock_post.return_value = _make_response(200, {"ok": True}) service = NotificationService() self.assertIn(NotificationChannel.SLACK, service.get_available_channels()) ok = service.send("slack bot content") self.assertTrue(ok) mock_post.assert_called_once() @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_serverchan3_via_notification_service( self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock ): cfg = _make_config(serverchan3_sendkey="SCTKEY") mock_get_config.return_value = cfg mock_post.return_value = _make_response(200, {"code": 0}) service = NotificationService() self.assertIn(NotificationChannel.SERVERCHAN3, service.get_available_channels()) ok = service.send("serverchan content") self.assertTrue(ok) mock_post.assert_called_once() @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_telegram_via_notification_service( self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock ): cfg = _make_config(telegram_bot_token="TOKEN", telegram_chat_id="123") mock_get_config.return_value = cfg mock_post.return_value = _make_response(200, {"ok": True}) service = NotificationService() self.assertIn(NotificationChannel.TELEGRAM, service.get_available_channels()) ok = service.send("hello telegram") self.assertTrue(ok) mock_post.assert_called_once() @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_wechat_via_notification_service(self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock): cfg = _make_config(wechat_webhook_url="https://wechat.example") mock_get_config.return_value = cfg mock_post.return_value = _make_response(200, {"errcode": 0}) service = NotificationService() self.assertIn(NotificationChannel.WECHAT, service.get_available_channels()) ok = service.send("hello wechat") self.assertTrue(ok) mock_post.assert_called_once() @mock.patch("src.notification.get_config") @mock.patch("requests.post") def test_send_to_wechat_via_notification_service_requires_chunking(self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock): cfg = _make_config(wechat_webhook_url="https://wechat.example", wechat_max_bytes=2000) mock_get_config.return_value = cfg mock_post.return_value = _make_response(200, {"errcode": 0}) service = NotificationService() self.assertIn(NotificationChannel.WECHAT, service.get_available_channels()) ok = service.send("A" * 6000) self.assertTrue(ok) self.assertAlmostEqual(mock_post.call_count, 4, delta=1) if __name__ == "__main__": unittest.main()