Skip to content

Generalize Google Voice sender config #54

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Pigbibi merged 1 commit into from
May 25, 2026
Merged

Generalize Google Voice sender config #54

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions docs/strategy_plugin_runtime_contract.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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_*`.
Expand Down
76 changes: 57 additions & 19 deletions src/quant_platform_kit/notifications/strategy_plugin_google_voice.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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,
Comment on lines +308 to +309

@chatgpt-codex-connector chatgpt-codex-connector Bot May 25, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Validate smtp_security before deriving TLS flags

_send_message derives use_starttls/use_ssl by strict equality against two literals, so any unexpected smtp_security value (for example when callers construct StrategyPluginGoogleVoiceSettings(...) directly instead of from_object) disables both TLS modes. In that case the code attempts an unauthenticated/plain SMTP flow on whatever port is set, which can silently break delivery or downgrade transport security instead of falling back to the documented default (ssl). Consider normalizing/validating smtp_security on dataclass construction or adding a safe fallback in _send_message.

Useful? React with 👍 / 👎.

timeout=settings.timeout,
)
except Exception as exc:
Expand Down Expand Up @@ -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")

Expand Down
68 changes: 57 additions & 11 deletions tests/test_google_voice_notifications.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 == []


Expand All @@ -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",
Expand Down Expand Up @@ -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()],
Expand Down Expand Up @@ -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"