From ef80c0c4f4e7f7f58d9863190b646e13fb9cd474 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 25 May 2026 17:58:16 +0800 Subject: [PATCH] Generalize Google Voice sender config --- docs/strategy_plugin_runtime_contract.md | 12 ++- .../strategy_plugin_google_voice.py | 76 ++++++++++++++----- tests/test_google_voice_notifications.py | 68 ++++++++++++++--- 3 files changed, 124 insertions(+), 32 deletions(-) diff --git a/docs/strategy_plugin_runtime_contract.md b/docs/strategy_plugin_runtime_contract.md index 3df24a0..09cefdf 100644 --- a/docs/strategy_plugin_runtime_contract.md +++ b/docs/strategy_plugin_runtime_contract.md @@ -139,8 +139,16 @@ also surface the Google Voice prompt. The public configuration names should be channel specific: - `CRISIS_ALERT_GOOGLE_VOICE_RECIPIENTS` -- `CRISIS_ALERT_GOOGLE_VOICE_GMAIL_USER` -- `CRISIS_ALERT_GOOGLE_VOICE_GMAIL_APP_PASSWORD` +- `CRISIS_ALERT_GOOGLE_VOICE_SENDER_EMAIL` +- `CRISIS_ALERT_GOOGLE_VOICE_SENDER_PASSWORD` + +By default the transport uses Gmail SMTP (`smtp.gmail.com`, port `465`, SSL), +but the sender is not part of the Google Voice channel contract. Non-Gmail +senders can override: + +- `CRISIS_ALERT_GOOGLE_VOICE_SMTP_HOST` +- `CRISIS_ALERT_GOOGLE_VOICE_SMTP_PORT` +- `CRISIS_ALERT_GOOGLE_VOICE_SMTP_SECURITY` (`ssl`, `starttls`, or `none`) Future direct email notifications should use a separate namespace such as `CRISIS_ALERT_EMAIL_*`. 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 6276f88..41f2de4 100644 --- a/src/quant_platform_kit/notifications/strategy_plugin_google_voice.py +++ b/src/quant_platform_kit/notifications/strategy_plugin_google_voice.py @@ -18,17 +18,27 @@ from .email import parse_email_recipients, send_smtp_email -_GOOGLE_VOICE_SMTP_HOST = "smtp.gmail.com" -_GOOGLE_VOICE_SMTP_PORT = 465 -_GOOGLE_VOICE_SMTP_STARTTLS = False -_GOOGLE_VOICE_SMTP_SSL = True +_DEFAULT_GOOGLE_VOICE_SMTP_HOST = "smtp.gmail.com" +_DEFAULT_GOOGLE_VOICE_SMTP_PORT = 465 +_DEFAULT_GOOGLE_VOICE_SMTP_SECURITY = "ssl" +_SMTP_SECURITY_NONE = "none" +_SMTP_SECURITY_SSL = "ssl" +_SMTP_SECURITY_STARTTLS = "starttls" +_SMTP_SECURITY_VALUES = { + _SMTP_SECURITY_NONE, + _SMTP_SECURITY_SSL, + _SMTP_SECURITY_STARTTLS, +} @dataclass(frozen=True) class StrategyPluginGoogleVoiceSettings: recipients: tuple[str, ...] = () - gmail_user: str | None = None - gmail_app_password: str | None = field(default=None, repr=False) + sender_email: str | None = None + sender_password: str | None = field(default=None, repr=False) + smtp_host: str = _DEFAULT_GOOGLE_VOICE_SMTP_HOST + smtp_port: int = _DEFAULT_GOOGLE_VOICE_SMTP_PORT + smtp_security: str = _DEFAULT_GOOGLE_VOICE_SMTP_SECURITY timeout: float = 10.0 @classmethod @@ -39,18 +49,29 @@ def from_object(cls, value: object) -> "StrategyPluginGoogleVoiceSettings": recipients=tuple( parse_email_recipients(_get_value(value, "crisis_alert_google_voice_recipients", ())) ), - 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"), + sender_email=_first_non_empty(_get_value(value, "crisis_alert_google_voice_sender_email")), + sender_password=_get_value(value, "crisis_alert_google_voice_sender_password"), + smtp_host=_first_non_empty( + _get_value(value, "crisis_alert_google_voice_smtp_host") + ) + or _DEFAULT_GOOGLE_VOICE_SMTP_HOST, + smtp_port=_coerce_int( + _get_value(value, "crisis_alert_google_voice_smtp_port"), + _DEFAULT_GOOGLE_VOICE_SMTP_PORT, + ), + smtp_security=_coerce_smtp_security( + _get_value(value, "crisis_alert_google_voice_smtp_security") + ), ) def missing_fields(self) -> tuple[str, ...]: missing: list[str] = [] if not parse_email_recipients(self.recipients): missing.append("CRISIS_ALERT_GOOGLE_VOICE_RECIPIENTS") - 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") + if not str(self.sender_email or "").strip(): + missing.append("CRISIS_ALERT_GOOGLE_VOICE_SENDER_EMAIL") + if not str(self.sender_password or "").strip(): + missing.append("CRISIS_ALERT_GOOGLE_VOICE_SENDER_PASSWORD") return tuple(missing) @property @@ -278,14 +299,14 @@ def _send_message( sent = send_notification( subject=message.subject, body=message.body, - smtp_host=_GOOGLE_VOICE_SMTP_HOST, - smtp_port=_GOOGLE_VOICE_SMTP_PORT, - sender=settings.gmail_user, + smtp_host=settings.smtp_host, + smtp_port=settings.smtp_port, + sender=settings.sender_email, recipients=settings.recipients, - username=settings.gmail_user, - password=settings.gmail_app_password, - use_starttls=_GOOGLE_VOICE_SMTP_STARTTLS, - use_ssl=_GOOGLE_VOICE_SMTP_SSL, + username=settings.sender_email, + password=settings.sender_password, + use_starttls=settings.smtp_security == _SMTP_SECURITY_STARTTLS, + use_ssl=settings.smtp_security == _SMTP_SECURITY_SSL, timeout=settings.timeout, ) except Exception as exc: @@ -365,6 +386,23 @@ def _first_non_empty(*values: Any) -> str | None: return None +def _coerce_int(value: Any, default: int) -> int: + text = str(value or "").strip() + if not text: + return default + try: + return int(text) + except (TypeError, ValueError): + return default + + +def _coerce_smtp_security(value: Any) -> str: + security = str(value or "").strip().lower() + if security in _SMTP_SECURITY_VALUES: + return security + return _DEFAULT_GOOGLE_VOICE_SMTP_SECURITY + + 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 107ff2b..ac83267 100644 --- a/tests/test_google_voice_notifications.py +++ b/tests/test_google_voice_notifications.py @@ -87,8 +87,8 @@ def test_publish_strategy_plugin_google_voice_alerts_skips_missing_config(): assert result.skipped_count == 1 assert result.deliveries[0].reason == "missing_google_voice_config" assert "CRISIS_ALERT_GOOGLE_VOICE_RECIPIENTS" 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 "CRISIS_ALERT_GOOGLE_VOICE_SENDER_EMAIL" in result.deliveries[0].error + assert "CRISIS_ALERT_GOOGLE_VOICE_SENDER_PASSWORD" in result.deliveries[0].error assert observed == [] @@ -100,8 +100,8 @@ def test_publish_strategy_plugin_google_voice_alerts_sends_and_records_marker(tm [_alert_signal()], google_voice_settings=StrategyPluginGoogleVoiceSettings( recipients=("risk@example.com",), - gmail_user="bot@example.com", - gmail_app_password="app-password", + sender_email="bot@example.com", + sender_password="app-password", ), strategy_label="TQQQ", context_label="ibkr / paper / tqqq", @@ -129,8 +129,8 @@ def test_publish_strategy_plugin_google_voice_alerts_skips_duplicate_marker(tmp_ store = StrategyPluginGoogleVoiceAlertMarkerStore(local_dir=tmp_path) settings = StrategyPluginGoogleVoiceSettings( recipients=("risk@example.com",), - gmail_user="bot@example.com", - gmail_app_password="app-password", + sender_email="bot@example.com", + sender_password="app-password", ) first = publish_strategy_plugin_google_voice_alerts( [_alert_signal()], @@ -158,16 +158,62 @@ def test_publish_strategy_plugin_google_voice_alerts_skips_duplicate_marker(tmp_ assert second.deliveries[0].reason == "duplicate_alert" -def test_google_voice_settings_reads_google_voice_recipient_names_only(): +def test_publish_strategy_plugin_google_voice_alerts_uses_transport_overrides(): + observed = [] + + result = publish_strategy_plugin_google_voice_alerts( + [_alert_signal()], + google_voice_settings=StrategyPluginGoogleVoiceSettings( + recipients=("voice@example.com",), + sender_email="bot@example.com", + sender_password="secret", + smtp_host="smtp.example.com", + smtp_port=587, + smtp_security="starttls", + ), + strategy_label="TQQQ", + context_label="ibkr / paper / tqqq", + send_notification=lambda **kwargs: observed.append(kwargs) or True, + log_message=lambda *_args, **_kwargs: None, + ) + + assert result.sent_count == 1 + assert observed[0]["smtp_host"] == "smtp.example.com" + assert observed[0]["smtp_port"] == 587 + assert observed[0]["use_starttls"] is True + assert observed[0]["use_ssl"] is False + + +def test_google_voice_settings_reads_sender_and_default_transport_names_only(): settings = StrategyPluginGoogleVoiceSettings.from_object( SimpleNamespace( crisis_alert_google_voice_recipients="alerts@example.com; voice@example.com", - crisis_alert_google_voice_gmail_user="sender@gmail.com", - crisis_alert_google_voice_gmail_app_password="app-password", + crisis_alert_google_voice_sender_email="sender@example.com", + crisis_alert_google_voice_sender_password="app-password", ) ) - assert settings.gmail_user == "sender@gmail.com" + assert settings.sender_email == "sender@example.com" assert settings.recipients == ("alerts@example.com", "voice@example.com") - assert settings.gmail_app_password == "app-password" + assert settings.sender_password == "app-password" + assert settings.smtp_host == "smtp.gmail.com" + assert settings.smtp_port == 465 + assert settings.smtp_security == "ssl" assert settings.missing_fields() == () + + +def test_google_voice_settings_reads_optional_smtp_transport_overrides(): + settings = StrategyPluginGoogleVoiceSettings.from_object( + SimpleNamespace( + crisis_alert_google_voice_recipients="voice@example.com", + crisis_alert_google_voice_sender_email="sender@example.com", + crisis_alert_google_voice_sender_password="secret", + crisis_alert_google_voice_smtp_host="smtp.example.com", + crisis_alert_google_voice_smtp_port="587", + crisis_alert_google_voice_smtp_security="starttls", + ) + ) + + assert settings.smtp_host == "smtp.example.com" + assert settings.smtp_port == 587 + assert settings.smtp_security == "starttls"