334 lines
11 KiB
Python
334 lines
11 KiB
Python
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 |