From 3ea3498e5878ca83be18728b977d9cf81510c7b2 Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Wed, 29 Apr 2026 12:11:10 +0200 Subject: [PATCH] add smtp module and testing --- app/api/routes/pages.py | 181 ++++++++++++++----- app/config.py | 8 + app/services/config_page.py | 16 ++ app/services/email.py | 106 ++++++++++++ app/templates/config.html | 22 +++ tests/test_config.py | 22 +++ tests/test_smtp.py | 334 ++++++++++++++++++++++++++++++++++++ 7 files changed, 643 insertions(+), 46 deletions(-) create mode 100644 app/services/email.py create mode 100644 tests/test_smtp.py diff --git a/app/api/routes/pages.py b/app/api/routes/pages.py index 2fb774e..4b474cb 100644 --- a/app/api/routes/pages.py +++ b/app/api/routes/pages.py @@ -14,6 +14,7 @@ from app.services.config_page import ( is_ticktick_oauth_ready, save_config_updates, ) +from app.services.email import EmailConfigurationError, EmailDeliveryError, is_smtp_ready, send_smtp_test_email from sqlalchemy.orm import Session templates = Jinja2Templates(directory=str(Path(__file__).resolve().parents[2] / "templates")) @@ -33,6 +34,49 @@ def _ticktick_oauth_notice(status_value: str | None) -> tuple[str | None, str | return None, None +def _smtp_test_notice(status_value: str | None) -> tuple[str | None, str | None]: + if status_value == "success": + return "SMTP test email sent successfully.", None + if status_value == "config-error": + return None, "SMTP test failed. Check required SMTP settings before sending a test email." + if status_value == "failed": + return None, "SMTP test failed. Check saved SMTP settings and server reachability." + return None, None + + +def _build_config_context( + *, + auth_db_session: Session, + settings: Settings, + current_auth: AuthenticatedSession, + config_saved: bool, + config_error: str | None, + password_change_error: str | None, + ticktick_oauth_notice: str | None, + ticktick_oauth_error: str | None, + smtp_test_notice: str | None, + smtp_test_error: str | None, +) -> dict[str, object]: + return { + "app_name": settings.app_name, + "app_env": settings.app_env, + "current_username": current_auth.user.username, + "csrf_token": current_auth.session.csrf_token, + "force_password_change": current_auth.user.force_password_change, + "password_change_error": password_change_error, + "config_error": config_error, + "config_saved": config_saved, + "config_sections": build_config_sections(auth_db_session, settings), + "ticktick_oauth_ready": is_ticktick_oauth_ready(settings), + "ticktick_redirect_uri": settings.ticktick_redirect_uri, + "ticktick_oauth_notice": ticktick_oauth_notice, + "ticktick_oauth_error": ticktick_oauth_error, + "smtp_test_ready": is_smtp_ready(settings), + "smtp_test_notice": smtp_test_notice, + "smtp_test_error": smtp_test_error, + } + + @router.get("/", response_class=HTMLResponse) def home( request: Request, @@ -66,22 +110,19 @@ def config_page( ticktick_oauth_notice, ticktick_oauth_error = _ticktick_oauth_notice( request.query_params.get("ticktick_oauth") ) - - context = { - "app_name": settings.app_name, - "app_env": settings.app_env, - "current_username": current_auth.user.username, - "csrf_token": current_auth.session.csrf_token, - "force_password_change": current_auth.user.force_password_change, - "password_change_error": None, - "config_error": None, - "config_saved": request.query_params.get("saved") == "1", - "config_sections": build_config_sections(auth_db_session, settings), - "ticktick_oauth_ready": is_ticktick_oauth_ready(settings), - "ticktick_redirect_uri": settings.ticktick_redirect_uri, - "ticktick_oauth_notice": ticktick_oauth_notice, - "ticktick_oauth_error": ticktick_oauth_error, - } + smtp_test_notice, smtp_test_error = _smtp_test_notice(request.query_params.get("smtp_test")) + context = _build_config_context( + auth_db_session=auth_db_session, + settings=settings, + current_auth=current_auth, + config_saved=request.query_params.get("saved") == "1", + config_error=None, + password_change_error=None, + ticktick_oauth_notice=ticktick_oauth_notice, + ticktick_oauth_error=ticktick_oauth_error, + smtp_test_notice=smtp_test_notice, + smtp_test_error=smtp_test_error, + ) return templates.TemplateResponse(request, "config.html", context) @@ -99,21 +140,18 @@ async def config_submit( csrf_token = form.get("csrf_token") if csrf_token != current_auth.session.csrf_token: logger.warning("Rejected config update due to CSRF validation failure") - context = { - "app_name": settings.app_name, - "app_env": settings.app_env, - "current_username": current_auth.user.username, - "csrf_token": current_auth.session.csrf_token, - "force_password_change": current_auth.user.force_password_change, - "password_change_error": None, - "config_error": "invalid config update request", - "config_saved": False, - "config_sections": build_config_sections(auth_db_session, settings), - "ticktick_oauth_ready": is_ticktick_oauth_ready(settings), - "ticktick_redirect_uri": settings.ticktick_redirect_uri, - "ticktick_oauth_notice": None, - "ticktick_oauth_error": None, - } + context = _build_config_context( + auth_db_session=auth_db_session, + settings=settings, + current_auth=current_auth, + config_saved=False, + config_error="invalid config update request", + password_change_error=None, + ticktick_oauth_notice=None, + ticktick_oauth_error=None, + smtp_test_notice=None, + smtp_test_error=None, + ) return templates.TemplateResponse( request, "config.html", @@ -126,21 +164,18 @@ async def config_submit( except ConfigSaveError: logger.warning("Rejected config update due to invalid submitted values") refreshed_settings = get_settings() - context = { - "app_name": refreshed_settings.app_name, - "app_env": refreshed_settings.app_env, - "current_username": current_auth.user.username, - "csrf_token": current_auth.session.csrf_token, - "force_password_change": current_auth.user.force_password_change, - "password_change_error": None, - "config_error": "invalid config submission", - "config_saved": False, - "config_sections": build_config_sections(auth_db_session, refreshed_settings), - "ticktick_oauth_ready": is_ticktick_oauth_ready(refreshed_settings), - "ticktick_redirect_uri": refreshed_settings.ticktick_redirect_uri, - "ticktick_oauth_notice": None, - "ticktick_oauth_error": None, - } + context = _build_config_context( + auth_db_session=auth_db_session, + settings=refreshed_settings, + current_auth=current_auth, + config_saved=False, + config_error="invalid config submission", + password_change_error=None, + ticktick_oauth_notice=None, + ticktick_oauth_error=None, + smtp_test_notice=None, + smtp_test_error=None, + ) return templates.TemplateResponse( request, "config.html", @@ -149,3 +184,57 @@ async def config_submit( ) return RedirectResponse(url="/config?saved=1", status_code=status.HTTP_303_SEE_OTHER) + + +@router.post("/config/smtp/test", response_class=HTMLResponse) +async def smtp_test_submit( + request: Request, + auth_db_session: Session = Depends(get_auth_db), + settings: Settings = Depends(get_app_settings), + current_auth: AuthenticatedSession | None = Depends(get_current_auth_session), +) -> Response: + if current_auth is None: + return RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER) + + form = await request.form() + csrf_token = form.get("csrf_token") + if csrf_token != current_auth.session.csrf_token: + logger.warning("Rejected SMTP test due to CSRF validation failure") + context = _build_config_context( + auth_db_session=auth_db_session, + settings=settings, + current_auth=current_auth, + config_saved=False, + config_error=None, + password_change_error=None, + ticktick_oauth_notice=None, + ticktick_oauth_error=None, + smtp_test_notice=None, + smtp_test_error="invalid SMTP test request", + ) + return templates.TemplateResponse( + request, + "config.html", + context, + status_code=status.HTTP_400_BAD_REQUEST, + ) + + try: + send_smtp_test_email(settings) + except EmailConfigurationError as exc: + logger.warning("SMTP test email rejected due to configuration: %s", exc) + return RedirectResponse( + url="/config?smtp_test=config-error", + status_code=status.HTTP_303_SEE_OTHER, + ) + except EmailDeliveryError as exc: + logger.warning("SMTP test email failed: %s", exc) + return RedirectResponse( + url="/config?smtp_test=failed", + status_code=status.HTTP_303_SEE_OTHER, + ) + + return RedirectResponse( + url="/config?smtp_test=success", + status_code=status.HTTP_303_SEE_OTHER, + ) diff --git a/app/config.py b/app/config.py index 1d7e0b9..929b38a 100644 --- a/app/config.py +++ b/app/config.py @@ -23,6 +23,14 @@ class Settings(BaseSettings): home_assistant_auth_token: str = "" home_assistant_timeout_seconds: float = 1.0 home_assistant_action_task_project_id: str = "" + smtp_enabled: bool = False + smtp_host: str = "" + smtp_port: int = 587 + smtp_username: str = "" + smtp_password: str = "" + smtp_from_address: str = "" + smtp_to_address: str = "" + smtp_use_starttls: bool = True poo_webhook_id: str = "" poo_sensor_entity_name: str = "sensor.test_poo_status" poo_sensor_friendly_name: str = "Poo Status" diff --git a/app/services/config_page.py b/app/services/config_page.py index 5a68621..141f75d 100644 --- a/app/services/config_page.py +++ b/app/services/config_page.py @@ -27,6 +27,14 @@ CONFIG_FIELDS: tuple[ConfigField, ...] = ( ConfigField("System", "APP_ENV", "app_env", "App Env"), ConfigField("System", "APP_DEBUG", "app_debug", "App Debug"), ConfigField("System", "APP_HOSTNAME", "app_hostname", "App Hostname"), + ConfigField("SMTP", "SMTP_ENABLED", "smtp_enabled", "SMTP Enabled"), + ConfigField("SMTP", "SMTP_HOST", "smtp_host", "SMTP Host"), + ConfigField("SMTP", "SMTP_PORT", "smtp_port", "SMTP Port"), + ConfigField("SMTP", "SMTP_USERNAME", "smtp_username", "SMTP Username"), + ConfigField("SMTP", "SMTP_PASSWORD", "smtp_password", "SMTP Password", secret=True), + ConfigField("SMTP", "SMTP_FROM_ADDRESS", "smtp_from_address", "SMTP From Address"), + ConfigField("SMTP", "SMTP_TO_ADDRESS", "smtp_to_address", "SMTP To Address"), + ConfigField("SMTP", "SMTP_USE_STARTTLS", "smtp_use_starttls", "SMTP Use STARTTLS"), ConfigField( "Authentication", "AUTH_SESSION_COOKIE_NAME", @@ -260,6 +268,14 @@ def _settings_payload(settings: Settings) -> dict[str, Any]: "home_assistant_auth_token": settings.home_assistant_auth_token, "home_assistant_timeout_seconds": settings.home_assistant_timeout_seconds, "home_assistant_action_task_project_id": settings.home_assistant_action_task_project_id, + "smtp_enabled": settings.smtp_enabled, + "smtp_host": settings.smtp_host, + "smtp_port": settings.smtp_port, + "smtp_username": settings.smtp_username, + "smtp_password": settings.smtp_password, + "smtp_from_address": settings.smtp_from_address, + "smtp_to_address": settings.smtp_to_address, + "smtp_use_starttls": settings.smtp_use_starttls, "poo_webhook_id": settings.poo_webhook_id, "poo_sensor_entity_name": settings.poo_sensor_entity_name, "poo_sensor_friendly_name": settings.poo_sensor_friendly_name, diff --git a/app/services/email.py b/app/services/email.py new file mode 100644 index 0000000..249b921 --- /dev/null +++ b/app/services/email.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +from dataclasses import dataclass +from email.message import EmailMessage +import smtplib + +from app.config import Settings + + +class EmailConfigurationError(ValueError): + """Raised when SMTP settings are incomplete or disabled.""" + + +class EmailDeliveryError(RuntimeError): + """Raised when sending email fails.""" + + +@dataclass(frozen=True, slots=True) +class SMTPConfig: + host: str + port: int + username: str + password: str + from_address: str + to_address: str + use_starttls: bool + + +def get_smtp_config(settings: Settings, *, require_enabled: bool = True) -> SMTPConfig: + if require_enabled and not settings.smtp_enabled: + raise EmailConfigurationError("SMTP is disabled") + + if not settings.smtp_host: + raise EmailConfigurationError("SMTP host is required") + + if settings.smtp_port <= 0: + raise EmailConfigurationError("SMTP port must be greater than zero") + + if not settings.smtp_from_address: + raise EmailConfigurationError("SMTP from address is required") + + if not settings.smtp_to_address: + raise EmailConfigurationError("SMTP to address is required") + + return SMTPConfig( + host=settings.smtp_host, + port=settings.smtp_port, + username=settings.smtp_username, + password=settings.smtp_password, + from_address=settings.smtp_from_address, + to_address=settings.smtp_to_address, + use_starttls=settings.smtp_use_starttls, + ) + + +def is_smtp_ready(settings: Settings) -> bool: + try: + get_smtp_config(settings, require_enabled=False) + except EmailConfigurationError: + return False + return True + + +def send_plaintext_email( + settings: Settings, + *, + subject: str, + body: str, + recipient: str | None = None, + require_enabled: bool = True, +) -> None: + smtp_config = get_smtp_config(settings, require_enabled=require_enabled) + message = EmailMessage() + message["Subject"] = subject + message["From"] = smtp_config.from_address + message["To"] = recipient or smtp_config.to_address + message.set_content(body) + + try: + with smtplib.SMTP(smtp_config.host, smtp_config.port, timeout=10) as smtp: + smtp.ehlo() + if smtp_config.use_starttls: + smtp.starttls() + smtp.ehlo() + if smtp_config.username: + smtp.login(smtp_config.username, smtp_config.password) + smtp.send_message(message) + except (OSError, smtplib.SMTPException) as exc: + error_message = _sanitize_error_message(str(exc), smtp_config.password) + raise EmailDeliveryError(error_message or "SMTP delivery failed") from exc + + +def send_smtp_test_email(settings: Settings) -> None: + send_plaintext_email( + settings, + subject="Home Automation SMTP Test", + body="This is a test email from Home Automation SMTP settings.", + require_enabled=False, + ) + + +def _sanitize_error_message(message: str, password: str) -> str: + sanitized = message + if password: + sanitized = sanitized.replace(password, "[redacted]") + return sanitized \ No newline at end of file diff --git a/app/templates/config.html b/app/templates/config.html index 6ce1b81..0fb3f70 100644 --- a/app/templates/config.html +++ b/app/templates/config.html @@ -33,6 +33,14 @@
{{ ticktick_oauth_notice }}
{% endif %} + {% if smtp_test_error %} +
{{ smtp_test_error }}
+ {% endif %} + + {% if smtp_test_notice %} +
{{ smtp_test_notice }}
+ {% endif %} +
当前用户
@@ -102,6 +110,20 @@ {% endif %}
{% endif %} + + {% if section.name == "SMTP" %} +
+
+

SMTP Test Email

+

Save the SMTP settings first, then send a simple plaintext test email to the configured recipient.

+
+ {% if smtp_test_ready %} + + {% else %} + Send SMTP Test + {% endif %} +
+ {% endif %} {% endfor %} diff --git a/tests/test_config.py b/tests/test_config.py index 598d280..1818de3 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -53,3 +53,25 @@ def test_settings_derive_development_ticktick_redirect_uri(monkeypatch) -> None: assert settings.app_base_url == "http://localhost:11001" assert settings.ticktick_redirect_uri == "http://localhost:11001/ticktick/auth/code" + + +def test_settings_support_smtp_fields(monkeypatch) -> None: + monkeypatch.setenv("SMTP_ENABLED", "true") + monkeypatch.setenv("SMTP_HOST", "smtp.example.com") + monkeypatch.setenv("SMTP_PORT", "2525") + monkeypatch.setenv("SMTP_USERNAME", "smtp-user") + monkeypatch.setenv("SMTP_PASSWORD", "smtp-password") + monkeypatch.setenv("SMTP_FROM_ADDRESS", "sender@example.com") + monkeypatch.setenv("SMTP_TO_ADDRESS", "recipient@example.com") + monkeypatch.setenv("SMTP_USE_STARTTLS", "false") + + settings = Settings(_env_file=None) + + assert settings.smtp_enabled is True + assert settings.smtp_host == "smtp.example.com" + assert settings.smtp_port == 2525 + assert settings.smtp_username == "smtp-user" + assert settings.smtp_password == "smtp-password" + assert settings.smtp_from_address == "sender@example.com" + assert settings.smtp_to_address == "recipient@example.com" + assert settings.smtp_use_starttls is False diff --git a/tests/test_smtp.py b/tests/test_smtp.py new file mode 100644 index 0000000..f004b2e --- /dev/null +++ b/tests/test_smtp.py @@ -0,0 +1,334 @@ +import re +import sqlite3 +import smtplib + +from fastapi.testclient import TestClient + +from app.config import Settings +from app.services.email import EmailDeliveryError, get_smtp_config, is_smtp_ready, send_smtp_test_email + + +def _extract_csrf_token(html: str) -> str: + match = re.search(r'name="csrf_token" value="([^"]+)"', html) + assert match is not None + return match.group(1) + + +def _login(client: TestClient) -> None: + login_page = client.get("/login") + csrf_token = _extract_csrf_token(login_page.text) + response = client.post( + "/login", + data={ + "username": "admin", + "password": "test-password", + "csrf_token": csrf_token, + }, + follow_redirects=False, + ) + assert response.status_code == 303 + + +def _smtp_settings(**overrides) -> Settings: + payload = { + "app_env": "development", + "app_hostname": "localhost:8000", + "app_database_url": "sqlite:///./data/app.db", + "location_database_url": "sqlite:///./data/locationRecorder.db", + "poo_database_url": "sqlite:///./data/pooRecorder.db", + "auth_bootstrap_username": "admin", + "auth_bootstrap_password": "secret-password", + "smtp_enabled": True, + "smtp_host": "smtp.example.com", + "smtp_port": 587, + "smtp_username": "smtp-user", + "smtp_password": "super-secret-password", + "smtp_from_address": "sender@example.com", + "smtp_to_address": "recipient@example.com", + "smtp_use_starttls": True, + } + payload.update(overrides) + return Settings(_env_file=None, **payload) + + +def test_get_smtp_config_reads_runtime_values() -> None: + settings = _smtp_settings(smtp_port=2525, smtp_use_starttls=False) + + smtp_config = get_smtp_config(settings) + + assert smtp_config.host == "smtp.example.com" + assert smtp_config.port == 2525 + assert smtp_config.username == "smtp-user" + assert smtp_config.password == "super-secret-password" + assert smtp_config.from_address == "sender@example.com" + assert smtp_config.to_address == "recipient@example.com" + assert smtp_config.use_starttls is False + + +def test_smtp_test_readiness_does_not_require_smtp_enabled() -> None: + settings = _smtp_settings(smtp_enabled=False) + + assert is_smtp_ready(settings) is True + + +def test_send_smtp_test_email_success(monkeypatch) -> None: + sent = {} + + class FakeSMTP: + def __init__(self, host, port, timeout): + sent["host"] = host + sent["port"] = port + sent["timeout"] = timeout + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return None + + def ehlo(self): + sent["ehlo"] = sent.get("ehlo", 0) + 1 + + def starttls(self): + sent["starttls"] = True + + def login(self, username, password): + sent["username"] = username + sent["password"] = password + + def send_message(self, message): + sent["subject"] = message["Subject"] + sent["from"] = message["From"] + sent["to"] = message["To"] + sent["body"] = message.get_content() + + monkeypatch.setattr("app.services.email.smtplib.SMTP", FakeSMTP) + + send_smtp_test_email(_smtp_settings()) + + assert sent["host"] == "smtp.example.com" + assert sent["port"] == 587 + assert sent["timeout"] == 10 + assert sent["starttls"] is True + assert sent["username"] == "smtp-user" + assert sent["password"] == "super-secret-password" + assert sent["subject"] == "Home Automation SMTP Test" + assert sent["from"] == "sender@example.com" + assert sent["to"] == "recipient@example.com" + assert "This is a test email" in sent["body"] + + +def test_send_smtp_test_email_does_not_require_smtp_enabled(monkeypatch) -> None: + sent = {} + + class FakeSMTP: + def __init__(self, host, port, timeout): + sent["host"] = host + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return None + + def ehlo(self): + return None + + def starttls(self): + return None + + def login(self, username, password): + return None + + def send_message(self, message): + sent["subject"] = message["Subject"] + + monkeypatch.setattr("app.services.email.smtplib.SMTP", FakeSMTP) + + send_smtp_test_email(_smtp_settings(smtp_enabled=False)) + + assert sent["host"] == "smtp.example.com" + assert sent["subject"] == "Home Automation SMTP Test" + + +def test_send_smtp_test_email_failure_sanitizes_password(monkeypatch) -> None: + class FakeSMTP: + def __init__(self, host, port, timeout): + pass + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return None + + def ehlo(self): + return None + + def starttls(self): + raise smtplib.SMTPException("authentication failed for super-secret-password") + + monkeypatch.setattr("app.services.email.smtplib.SMTP", FakeSMTP) + + try: + send_smtp_test_email(_smtp_settings()) + assert False, "expected EmailDeliveryError" + except EmailDeliveryError as exc: + assert "super-secret-password" not in str(exc) + assert "[redacted]" in str(exc) + + +def test_config_update_does_not_clear_existing_smtp_password( + client: TestClient, test_database_urls +) -> None: + _login(client) + config_page = client.get("/config") + config_csrf_token = _extract_csrf_token(config_page.text) + + response = client.post( + "/config", + data={ + "csrf_token": config_csrf_token, + "APP_NAME": "SMTP Config Test", + "APP_ENV": "development", + "APP_DEBUG": "true", + "APP_HOSTNAME": "localhost:8000", + "SMTP_ENABLED": "true", + "SMTP_HOST": "smtp.example.com", + "SMTP_PORT": "587", + "SMTP_USERNAME": "smtp-user", + "SMTP_PASSWORD": "persist-me", + "SMTP_FROM_ADDRESS": "sender@example.com", + "SMTP_TO_ADDRESS": "recipient@example.com", + "SMTP_USE_STARTTLS": "true", + "AUTH_SESSION_COOKIE_NAME": "home_automation_session", + "AUTH_SESSION_TTL_HOURS": "12", + "AUTH_COOKIE_SECURE_OVERRIDE": "false", + "POO_WEBHOOK_ID": "", + "POO_SENSOR_ENTITY_NAME": "sensor.test_poo_status", + "POO_SENSOR_FRIENDLY_NAME": "Poo Status", + "TICKTICK_CLIENT_ID": "", + "TICKTICK_CLIENT_SECRET": "", + "TICKTICK_TOKEN": "", + "HOME_ASSISTANT_BASE_URL": "", + "HOME_ASSISTANT_AUTH_TOKEN": "", + "HOME_ASSISTANT_TIMEOUT_SECONDS": "1.0", + "HOME_ASSISTANT_ACTION_TASK_PROJECT_ID": "", + }, + follow_redirects=False, + ) + assert response.status_code == 303 + + config_page = client.get("/config") + config_csrf_token = _extract_csrf_token(config_page.text) + response = client.post( + "/config", + data={ + "csrf_token": config_csrf_token, + "APP_NAME": "SMTP Config Updated", + "APP_ENV": "development", + "APP_DEBUG": "true", + "APP_HOSTNAME": "localhost:8000", + "SMTP_ENABLED": "true", + "SMTP_HOST": "smtp.example.com", + "SMTP_PORT": "587", + "SMTP_USERNAME": "smtp-user", + "SMTP_PASSWORD": "", + "SMTP_FROM_ADDRESS": "sender@example.com", + "SMTP_TO_ADDRESS": "recipient@example.com", + "SMTP_USE_STARTTLS": "true", + "AUTH_SESSION_COOKIE_NAME": "home_automation_session", + "AUTH_SESSION_TTL_HOURS": "12", + "AUTH_COOKIE_SECURE_OVERRIDE": "false", + "POO_WEBHOOK_ID": "", + "POO_SENSOR_ENTITY_NAME": "sensor.test_poo_status", + "POO_SENSOR_FRIENDLY_NAME": "Poo Status", + "TICKTICK_CLIENT_ID": "", + "TICKTICK_CLIENT_SECRET": "", + "TICKTICK_TOKEN": "", + "HOME_ASSISTANT_BASE_URL": "", + "HOME_ASSISTANT_AUTH_TOKEN": "", + "HOME_ASSISTANT_TIMEOUT_SECONDS": "1.0", + "HOME_ASSISTANT_ACTION_TASK_PROJECT_ID": "", + }, + follow_redirects=False, + ) + assert response.status_code == 303 + + conn = sqlite3.connect(test_database_urls["app_path"]) + try: + rows = dict(conn.execute("SELECT key, value FROM app_config").fetchall()) + finally: + conn.close() + + assert rows["SMTP_PASSWORD"] == "persist-me" + assert rows["APP_NAME"] == "SMTP Config Updated" + + +def test_smtp_test_endpoint_requires_authentication(client: TestClient) -> None: + response = client.post("/config/smtp/test", data={"csrf_token": "ignored"}, follow_redirects=False) + + assert response.status_code == 303 + assert response.headers["location"] == "/login" + + +def test_smtp_test_endpoint_success_and_failure_do_not_expose_password( + client: TestClient, monkeypatch +) -> None: + from app.api.routes import pages + + _login(client) + config_page = client.get("/config") + csrf_token = _extract_csrf_token(config_page.text) + + monkeypatch.setattr(pages, "send_smtp_test_email", lambda settings: None) + response = client.post("/config/smtp/test", data={"csrf_token": csrf_token}, follow_redirects=False) + assert response.status_code == 303 + assert response.headers["location"] == "/config?smtp_test=success" + + follow_up = client.get(response.headers["location"]) + assert follow_up.status_code == 200 + assert "SMTP test email sent successfully." in follow_up.text + assert "super-secret-password" not in follow_up.text + + monkeypatch.setattr( + pages, + "send_smtp_test_email", + lambda settings: (_ for _ in ()).throw(EmailDeliveryError("smtp auth failed for [redacted]")), + ) + response = client.post("/config/smtp/test", data={"csrf_token": csrf_token}, follow_redirects=False) + assert response.status_code == 303 + assert response.headers["location"] == "/config?smtp_test=failed" + + follow_up = client.get(response.headers["location"]) + assert follow_up.status_code == 200 + assert "SMTP test failed. Check saved SMTP settings and server reachability." in follow_up.text + assert "super-secret-password" not in follow_up.text + + +def test_config_page_renders_smtp_test_button_with_formaction( + client: TestClient, test_database_urls +) -> None: + _login(client) + + conn = sqlite3.connect(test_database_urls["app_path"]) + try: + conn.executemany( + "INSERT INTO app_config (key, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP) " + "ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at", + [ + ("SMTP_ENABLED", "true"), + ("SMTP_HOST", "smtp.example.com"), + ("SMTP_PORT", "587"), + ("SMTP_FROM_ADDRESS", "sender@example.com"), + ("SMTP_TO_ADDRESS", "recipient@example.com"), + ], + ) + conn.commit() + finally: + conn.close() + + response = client.get("/config") + + assert response.status_code == 200 + assert 'formaction="/config/smtp/test"' in response.text \ No newline at end of file