From 622fc009e734a50dc048b96e7d2fdded7c87343e Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 25 May 2026 16:22:30 +0800 Subject: [PATCH] Separate Google Voice alert configuration --- docs/strategy_plugin_runtime_contract.md | 18 ++++-- .../strategy_plugin_google_voice.py | 59 +++++++------------ tests/test_google_voice_notifications.py | 42 ++++++++----- 3 files changed, 64 insertions(+), 55 deletions(-) diff --git a/docs/strategy_plugin_runtime_contract.md b/docs/strategy_plugin_runtime_contract.md index d85dc73..52d3923 100644 --- a/docs/strategy_plugin_runtime_contract.md +++ b/docs/strategy_plugin_runtime_contract.md @@ -124,13 +124,23 @@ when any of the following is true: - `suggested_action` is `defend` or `blocked` - `would_trade_if_enabled` is `true` -Platforms may still choose their delivery sinks, but Google Voice escalation over -SMTP should use +Platforms may still choose their delivery sinks, but Google Voice escalation via +the Gmail-to-Google-Voice SMS gateway should use `quant_platform_kit.notifications.strategy_plugin_google_voice.publish_strategy_plugin_google_voice_alerts()`. The publisher builds the shared subject/body, prefixes platform context, returns structured sent/skipped/failed diagnostics, and can use `StrategyPluginGoogleVoiceAlertMarkerStore` to skip alert keys that were already -sent. Platforms should expose this as Google Voice notification config, not as a -generic email alert surface. +sent. + +Platforms should expose this as Google Voice notification config, not as a +generic email alert surface. The public configuration names should be channel +specific: + +- `CRISIS_ALERT_GOOGLE_VOICE_GATEWAY` +- `CRISIS_ALERT_GOOGLE_VOICE_GMAIL_USER` +- `CRISIS_ALERT_GOOGLE_VOICE_GMAIL_APP_PASSWORD` + +Future direct email notifications should use a separate namespace such as +`CRISIS_ALERT_EMAIL_*`. This keeps the Crisis Response plugin behavior consistent across IBKR, Schwab, LongBridge, Firstrade, and future platform runtimes. diff --git a/src/quant_platform_kit/notifications/strategy_plugin_google_voice.py b/src/quant_platform_kit/notifications/strategy_plugin_google_voice.py index 6de417b..4c5b32b 100644 --- a/src/quant_platform_kit/notifications/strategy_plugin_google_voice.py +++ b/src/quant_platform_kit/notifications/strategy_plugin_google_voice.py @@ -20,14 +20,13 @@ @dataclass(frozen=True) class StrategyPluginGoogleVoiceSettings: - smtp_host: str | None = None - smtp_port: int = 587 - sender: str | None = None - recipients: tuple[str, ...] = () - username: str | None = None - password: str | None = field(default=None, repr=False) - use_starttls: bool = True - use_ssl: bool = False + gateway_recipients: tuple[str, ...] = () + gmail_user: str | None = None + gmail_app_password: str | None = field(default=None, repr=False) + smtp_host: str = "smtp.gmail.com" + smtp_port: int = 465 + use_starttls: bool = False + use_ssl: bool = True timeout: float = 10.0 @classmethod @@ -35,24 +34,21 @@ def from_object(cls, value: object) -> "StrategyPluginGoogleVoiceSettings": if isinstance(value, cls): return value return cls( - smtp_host=_get_value(value, "crisis_alert_smtp_host"), - smtp_port=int(_get_value(value, "crisis_alert_smtp_port", 587) or 587), - sender=_first_non_empty(_get_value(value, "crisis_alert_smtp_from")), - recipients=tuple(parse_email_recipients(_get_value(value, "crisis_alert_google_voice_to", ()))), - username=_get_value(value, "crisis_alert_smtp_username"), - password=_get_value(value, "crisis_alert_smtp_password"), - use_starttls=_coerce_bool(_get_value(value, "crisis_alert_smtp_starttls", True), default=True), - use_ssl=_coerce_bool(_get_value(value, "crisis_alert_smtp_ssl", False), default=False), + gateway_recipients=tuple( + parse_email_recipients(_get_value(value, "crisis_alert_google_voice_gateway", ())) + ), + gmail_user=_first_non_empty(_get_value(value, "crisis_alert_google_voice_gmail_user")), + gmail_app_password=_get_value(value, "crisis_alert_google_voice_gmail_app_password"), ) def missing_fields(self) -> tuple[str, ...]: missing: list[str] = [] - if not str(self.smtp_host or "").strip(): - missing.append("CRISIS_ALERT_SMTP_HOST") - if not str(self.sender or "").strip(): - missing.append("CRISIS_ALERT_SMTP_FROM") - if not parse_email_recipients(self.recipients): - missing.append("CRISIS_ALERT_GOOGLE_VOICE_TO") + if not parse_email_recipients(self.gateway_recipients): + missing.append("CRISIS_ALERT_GOOGLE_VOICE_GATEWAY") + if not str(self.gmail_user or "").strip(): + missing.append("CRISIS_ALERT_GOOGLE_VOICE_GMAIL_USER") + if not str(self.gmail_app_password or "").strip(): + missing.append("CRISIS_ALERT_GOOGLE_VOICE_GMAIL_APP_PASSWORD") return tuple(missing) @property @@ -282,10 +278,10 @@ def _send_message( body=message.body, smtp_host=settings.smtp_host, smtp_port=settings.smtp_port, - sender=settings.sender, - recipients=settings.recipients, - username=settings.username, - password=settings.password, + sender=settings.gmail_user, + recipients=settings.gateway_recipients, + username=settings.gmail_user, + password=settings.gmail_app_password, use_starttls=settings.use_starttls, use_ssl=settings.use_ssl, timeout=settings.timeout, @@ -367,17 +363,6 @@ def _first_non_empty(*values: Any) -> str | None: return None -def _coerce_bool(value: Any, *, default: bool) -> bool: - if value is None: - return default - if isinstance(value, bool): - return value - text = str(value).strip().lower() - if not text: - return default - return text in {"1", "true", "yes", "y", "on"} - - def _fallback_alert_key(message: StrategyPluginAlertMessage) -> str: return "strategy_plugin_google_voice_alert/" + _clean_relative_key(message.subject or "unknown") diff --git a/tests/test_google_voice_notifications.py b/tests/test_google_voice_notifications.py index 0dff4a3..f1deee9 100644 --- a/tests/test_google_voice_notifications.py +++ b/tests/test_google_voice_notifications.py @@ -86,7 +86,9 @@ def test_publish_strategy_plugin_google_voice_alerts_skips_missing_config(): assert result.sent_count == 0 assert result.skipped_count == 1 assert result.deliveries[0].reason == "missing_google_voice_config" - assert "CRISIS_ALERT_SMTP_HOST" in result.deliveries[0].error + assert "CRISIS_ALERT_GOOGLE_VOICE_GATEWAY" in result.deliveries[0].error + assert "CRISIS_ALERT_GOOGLE_VOICE_GMAIL_USER" in result.deliveries[0].error + assert "CRISIS_ALERT_GOOGLE_VOICE_GMAIL_APP_PASSWORD" in result.deliveries[0].error assert observed == [] @@ -97,9 +99,9 @@ def test_publish_strategy_plugin_google_voice_alerts_sends_and_records_marker(tm result = publish_strategy_plugin_google_voice_alerts( [_alert_signal()], google_voice_settings=StrategyPluginGoogleVoiceSettings( - smtp_host="smtp.example.com", - sender="bot@example.com", - recipients=("risk@example.com",), + gateway_recipients=("risk@example.com",), + gmail_user="bot@example.com", + gmail_app_password="app-password", ), strategy_label="TQQQ", context_label="ibkr / paper / tqqq", @@ -112,15 +114,23 @@ def test_publish_strategy_plugin_google_voice_alerts_sends_and_records_marker(tm assert result.failed_count == 0 assert result.deliveries[0].alert_key assert "[ibkr / paper / tqqq]" in observed[0]["subject"] + assert observed[0]["smtp_host"] == "smtp.gmail.com" + assert observed[0]["smtp_port"] == 465 + assert observed[0]["sender"] == "bot@example.com" + assert observed[0]["recipients"] == ("risk@example.com",) + assert observed[0]["username"] == "bot@example.com" + assert observed[0]["password"] == "app-password" + assert observed[0]["use_starttls"] is False + assert observed[0]["use_ssl"] is True assert store.has_alert(result.deliveries[0].alert_key) def test_publish_strategy_plugin_google_voice_alerts_skips_duplicate_marker(tmp_path): store = StrategyPluginGoogleVoiceAlertMarkerStore(local_dir=tmp_path) settings = StrategyPluginGoogleVoiceSettings( - smtp_host="smtp.example.com", - sender="bot@example.com", - recipients=("risk@example.com",), + gateway_recipients=("risk@example.com",), + gmail_user="bot@example.com", + gmail_app_password="app-password", ) first = publish_strategy_plugin_google_voice_alerts( [_alert_signal()], @@ -148,16 +158,20 @@ def test_publish_strategy_plugin_google_voice_alerts_skips_duplicate_marker(tmp_ assert second.deliveries[0].reason == "duplicate_alert" -def test_google_voice_settings_read_google_voice_names_only(): +def test_google_voice_settings_reads_google_voice_gmail_names_only(): settings = StrategyPluginGoogleVoiceSettings.from_object( SimpleNamespace( - crisis_alert_smtp_host="smtp.gmail.com", - crisis_alert_smtp_from="sender@gmail.com", - crisis_alert_google_voice_to="gateway@txt.voice.google.com", - crisis_alert_smtp_username="sender@gmail.com", + crisis_alert_google_voice_gateway="gateway@txt.voice.google.com", + crisis_alert_google_voice_gmail_user="sender@gmail.com", + crisis_alert_google_voice_gmail_app_password="app-password", ) ) - assert settings.sender == "sender@gmail.com" - assert settings.recipients == ("gateway@txt.voice.google.com",) + assert settings.gmail_user == "sender@gmail.com" + assert settings.gateway_recipients == ("gateway@txt.voice.google.com",) + assert settings.gmail_app_password == "app-password" + assert settings.smtp_host == "smtp.gmail.com" + assert settings.smtp_port == 465 + assert settings.use_ssl is True + assert settings.use_starttls is False assert settings.missing_fields() == ()