Add crisis plugin trigger rehearsal tests

notifications/telegram.py

@@ -90,6 +90,7 @@
         "strategy_name_tqqq_growth_income": "TQQQ 增长收益",
         "strategy_name_soxl_soxx_trend_income": "SOXL/SOXX 半导体趋势收益",
         "strategy_name_global_etf_rotation": "全球 ETF 轮动",
+        "strategy_name_global_etf_confidence_vol_gate": "全球 ETF 置信波动门控",
         "strategy_name_russell_1000_multi_factor_defensive": "罗素1000多因子",
         "strategy_name_tech_communication_pullback_enhancement": "科技通信回调增强",
         "strategy_name_qqq_tech_enhancement": "科技通信回调增强",
@@ -170,6 +171,7 @@
         "strategy_name_tqqq_growth_income": "TQQQ Growth Income",
         "strategy_name_soxl_soxx_trend_income": "SOXL/SOXX Semiconductor Trend Income",
         "strategy_name_global_etf_rotation": "Global ETF Rotation",
+        "strategy_name_global_etf_confidence_vol_gate": "Global ETF Confidence Vol Gate",
         "strategy_name_russell_1000_multi_factor_defensive": "Russell 1000 Multi-Factor",
         "strategy_name_tech_communication_pullback_enhancement": "Tech/Communication Pullback Enhancement",
         "strategy_name_qqq_tech_enhancement": "Tech/Communication Pullback Enhancement",

tests/test_notifications.py

@@ -61,6 +61,8 @@ def test_supported_strategy_profiles_have_translated_names(self):
         zh_name = build_strategy_display_name(build_translator("zh"))
         en_name = build_strategy_display_name(build_translator("en"))
 
+        self.assertEqual(zh_name("global_etf_confidence_vol_gate"), "全球 ETF 置信波动门控")
+        self.assertEqual(en_name("global_etf_confidence_vol_gate"), "Global ETF Confidence Vol Gate")
         for profile in SUPPORTED_STRATEGY_PROFILES:
             self.assertNotEqual(zh_name(profile), profile)
             self.assertNotEqual(en_name(profile), profile)

tests/test_request_handling.py

@@ -294,6 +294,78 @@ def fake_run_strategy_core(client, now_ny, *, strategy_plugin_signals=()):
         self.assertEqual(plugin_summary["canonical_route"], "no_action")
         self.assertEqual(plugin_summary["suggested_action"], "monitor")
 
+    def test_handle_schwab_rehearses_triggered_shadow_plugin_report(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            signal_path = Path(temp_dir) / "latest_signal.json"
+            signal_path.write_text(
+                json.dumps(
+                    {
+                        "strategy": "tqqq_growth_income",
+                        "plugin": "crisis_response_shadow",
+                        "mode": "shadow",
+                        "configured_mode": "shadow",
+                        "effective_mode": "shadow",
+                        "schema_version": "crisis_response_shadow.v1",
+                        "as_of": "2008-03-10",
+                        "canonical_route": "true_crisis",
+                        "suggested_action": "defend",
+                        "would_trade_if_enabled": True,
+                        "execution_controls": {
+                            "broker_order_allowed": False,
+                            "live_allocation_mutation_allowed": False,
+                            "repository_broker_write_allowed": False,
+                            "repository_allocation_mutation_allowed": False,
+                        },
+                    }
+                ),
+                encoding="utf-8",
+            )
+            mount_config = json.dumps(
+                {
+                    "strategy_plugins": [
+                        {
+                            "strategy": "tqqq_growth_income",
+                            "plugin": "crisis_response_shadow",
+                            "signal_path": str(signal_path),
+                            "enabled": True,
+                            "expected_mode": "shadow",
+                        }
+                    ]
+                }
+            )
+            module = load_module(strategy_plugin_mounts_json=mount_config)
+            observed = {}
+
+            module.get_client_from_secret = lambda *args, **kwargs: object()
+            module.is_market_open_today = lambda: True
+            module.persist_execution_report = (
+                lambda report: observed.setdefault("report", report) or "/tmp/report.json"
+            )
+
+            def fake_run_strategy_core(client, now_ny, *, strategy_plugin_signals=()):
+                observed["signals"] = strategy_plugin_signals
+                self.assertEqual(len(strategy_plugin_signals), 1)
+                signal = strategy_plugin_signals[0]
+                self.assertTrue(signal.would_trade_if_enabled)
+                self.assertEqual(signal.canonical_route, "true_crisis")
+                self.assertEqual(signal.suggested_action, "defend")
+                self.assertFalse(signal.execution_controls["broker_order_allowed"])
+                self.assertFalse(signal.execution_controls["live_allocation_mutation_allowed"])
+
+            module.run_strategy_core = fake_run_strategy_core
+
+            with module.app.test_request_context("/", method="POST"):
+                body, status = module.handle_schwab()
+
+        self.assertEqual(status, 200)
+        self.assertEqual(body, "OK")
+        plugin_summary = observed["report"]["summary"]["strategy_plugins"][0]
+        self.assertEqual(plugin_summary["canonical_route"], "true_crisis")
+        self.assertEqual(plugin_summary["suggested_action"], "defend")
+        self.assertTrue(plugin_summary["would_trade_if_enabled"])
+        self.assertFalse(plugin_summary["execution_controls"]["broker_order_allowed"])
+        self.assertFalse(plugin_summary["execution_controls"]["live_allocation_mutation_allowed"])
+
     def test_strategy_plugin_notification_line_uses_i18n(self):
         module = load_module(notify_lang="zh")
         signal = types.SimpleNamespace(
@@ -311,6 +383,21 @@ def test_strategy_plugin_notification_line_uses_i18n(self):
         self.assertIn("路由：不操作", lines[0])
         self.assertIn("建议：仅观察", lines[0])
 
+    def test_strategy_plugin_notification_line_renders_triggered_shadow_signal(self):
+        module = load_module(notify_lang="zh")
+        signal = types.SimpleNamespace(
+            plugin="crisis_response_shadow",
+            effective_mode="shadow",
+            canonical_route="true_crisis",
+            suggested_action="defend",
+        )
+
+        lines = module.build_strategy_plugin_notification_lines((signal,))
+
+        self.assertEqual(len(lines), 1)
+        self.assertIn("路由：真危机", lines[0])
+        self.assertIn("建议：防守", lines[0])
+
     def test_handle_schwab_reports_plugin_config_error_without_blocking_strategy(self):
         mount_config = json.dumps(
             {

tests/test_runtime_config_support.py

@@ -34,6 +34,22 @@
 
 
 SAMPLE_STRATEGY_PROFILE = "tqqq_growth_income"
+BASE_SCHWAB_PROFILES = frozenset(
+    {
+        SAMPLE_STRATEGY_PROFILE,
+        "global_etf_rotation",
+        "mega_cap_leader_rotation_top50_balanced",
+        "russell_1000_multi_factor_defensive",
+        "soxl_soxx_trend_income",
+        "tech_communication_pullback_enhancement",
+    }
+)
+OPTIONAL_SCHWAB_PROFILES = frozenset({"global_etf_confidence_vol_gate"})
+
+
+def expected_schwab_profiles(actual_profiles) -> frozenset[str]:
+    actual = frozenset(actual_profiles)
+    return BASE_SCHWAB_PROFILES | (OPTIONAL_SCHWAB_PROFILES & actual)
 
 
 class RuntimeConfigSupportTests(unittest.TestCase):
@@ -71,34 +87,12 @@ def test_rejects_unknown_strategy_profile(self):
                 load_platform_runtime_settings()
 
     def test_platform_supported_profiles_are_filtered_by_registry(self):
-        self.assertEqual(
-            get_supported_profiles_for_platform(SCHWAB_PLATFORM),
-            frozenset(
-                {
-                    SAMPLE_STRATEGY_PROFILE,
-                    "global_etf_rotation",
-                    "mega_cap_leader_rotation_top50_balanced",
-                    "russell_1000_multi_factor_defensive",
-                    "soxl_soxx_trend_income",
-                    "tech_communication_pullback_enhancement",
-                }
-            ),
-        )
+        profiles = get_supported_profiles_for_platform(SCHWAB_PLATFORM)
+        self.assertEqual(profiles, expected_schwab_profiles(profiles))
 
     def test_platform_eligible_profiles_are_exposed_by_capability_matrix(self):
-        self.assertEqual(
-            get_eligible_profiles_for_platform(SCHWAB_PLATFORM),
-            frozenset(
-                {
-                    SAMPLE_STRATEGY_PROFILE,
-                    "global_etf_rotation",
-                    "mega_cap_leader_rotation_top50_balanced",
-                    "russell_1000_multi_factor_defensive",
-                    "soxl_soxx_trend_income",
-                    "tech_communication_pullback_enhancement",
-                }
-            ),
-        )
+        profiles = get_eligible_profiles_for_platform(SCHWAB_PLATFORM)
+        self.assertEqual(profiles, expected_schwab_profiles(profiles))
 
     def test_rejects_human_readable_alias(self):
         with patch.dict(os.environ, {"STRATEGY_PROFILE": "qqq_tqqq_growth_income"}, clear=True):
@@ -188,17 +182,7 @@ def test_platform_profile_status_matrix_matches_current_schwab_rollout(self):
         rows = get_platform_profile_status_matrix()
         by_profile = {row["canonical_profile"]: row for row in rows}
 
-        self.assertEqual(
-            set(by_profile),
-            {
-                "tqqq_growth_income",
-                "global_etf_rotation",
-                "mega_cap_leader_rotation_top50_balanced",
-                "russell_1000_multi_factor_defensive",
-                "soxl_soxx_trend_income",
-                "tech_communication_pullback_enhancement",
-            },
-        )
+        self.assertEqual(set(by_profile), expected_schwab_profiles(by_profile))
         self.assertEqual(
             by_profile["tqqq_growth_income"],
             {
@@ -216,6 +200,13 @@ def test_platform_profile_status_matrix_matches_current_schwab_rollout(self):
         )
         self.assertTrue(by_profile["global_etf_rotation"]["eligible"])
         self.assertTrue(by_profile["global_etf_rotation"]["enabled"])
+        if "global_etf_confidence_vol_gate" in by_profile:
+            self.assertEqual(
+                by_profile["global_etf_confidence_vol_gate"]["display_name"],
+                "Global ETF Confidence Vol Gate",
+            )
+            self.assertTrue(by_profile["global_etf_confidence_vol_gate"]["eligible"])
+            self.assertTrue(by_profile["global_etf_confidence_vol_gate"]["enabled"])
         self.assertEqual(
             by_profile["russell_1000_multi_factor_defensive"]["display_name"],
             "Russell 1000 Multi-Factor",