test(email): контактный адрес, Reply-To и тестовая отправка письма
Покрытие валидации/сохранения contact_email, простановки Reply-To в письмах регистрации/приглашений/саммари при включённом и выключенном контактном адресе, и эндпоинта тестовой отправки (успех, дефолтный получатель, сбой транспорта без утечки логина/пароля).
This commit is contained in:
@@ -24,6 +24,7 @@ from models.team import Team
|
||||
from models.user import User
|
||||
from services import ai_levels
|
||||
from services.conference_ids import generate_number, generate_slug
|
||||
from services.email import EmailSendError
|
||||
|
||||
FUTURE = datetime.now(UTC) + timedelta(days=3)
|
||||
|
||||
@@ -85,6 +86,7 @@ async def test_all_admin_endpoints_forbidden_for_non_admin(
|
||||
("PATCH", f"/api/v1/admin/users/{other_user.id}", {"role": "admin"}),
|
||||
("GET", "/api/v1/admin/settings", None),
|
||||
("PUT", "/api/v1/admin/settings", {}),
|
||||
("POST", "/api/v1/admin/settings/test-email", {}),
|
||||
]
|
||||
for method, path, body in requests:
|
||||
response = await client.request(method, path, json=body, headers=headers)
|
||||
@@ -606,3 +608,153 @@ async def test_put_settings_unavailable_ai_level_returns_400(
|
||||
headers=_auth_headers(admin),
|
||||
)
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
async def test_put_settings_contact_email_enable_and_persist(
|
||||
client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setattr(admin_module, "transcription_queue_served", lambda: False)
|
||||
admin = await _make_user(db_session, role="admin")
|
||||
await db_session.commit()
|
||||
|
||||
response = await client.put(
|
||||
"/api/v1/admin/settings",
|
||||
json={"contact_email_enabled": True, "contact_email": "Contact@VidConf.RU"},
|
||||
headers=_auth_headers(admin),
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
body = response.json()
|
||||
assert body["contact_email_enabled"] is True
|
||||
assert body["contact_email"] == "contact@vidconf.ru"
|
||||
|
||||
reloaded = await client.get("/api/v1/admin/settings", headers=_auth_headers(admin))
|
||||
assert reloaded.json()["contact_email"] == "contact@vidconf.ru"
|
||||
|
||||
|
||||
async def test_put_settings_contact_email_invalid_returns_400(
|
||||
client: httpx.AsyncClient, db_session: AsyncSession
|
||||
) -> None:
|
||||
admin = await _make_user(db_session, role="admin")
|
||||
await db_session.commit()
|
||||
|
||||
response = await client.put(
|
||||
"/api/v1/admin/settings",
|
||||
json={"contact_email_enabled": True, "contact_email": "not an email"},
|
||||
headers=_auth_headers(admin),
|
||||
)
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
# --- Тестовое письмо ----------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_send_test_email_defaults_to_admin_email(
|
||||
client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Без явного `to` тестовое письмо уходит на email текущего администратора."""
|
||||
admin = await _make_user(db_session, role="admin")
|
||||
await db_session.commit()
|
||||
|
||||
sent: dict[str, object] = {}
|
||||
|
||||
class _FakeBackend:
|
||||
async def send(self, **kwargs: object) -> None:
|
||||
sent.update(kwargs)
|
||||
|
||||
monkeypatch.setattr(admin_module, "create_email_backend", lambda settings: _FakeBackend())
|
||||
|
||||
response = await client.post(
|
||||
"/api/v1/admin/settings/test-email", json={}, headers=_auth_headers(admin)
|
||||
)
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
body = response.json()
|
||||
assert body["success"] is True
|
||||
assert sent["to"] == admin.email
|
||||
|
||||
|
||||
async def test_send_test_email_to_explicit_recipient(
|
||||
client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
admin = await _make_user(db_session, role="admin")
|
||||
await db_session.commit()
|
||||
|
||||
sent: dict[str, object] = {}
|
||||
|
||||
class _FakeBackend:
|
||||
async def send(self, **kwargs: object) -> None:
|
||||
sent.update(kwargs)
|
||||
|
||||
monkeypatch.setattr(admin_module, "create_email_backend", lambda settings: _FakeBackend())
|
||||
|
||||
response = await client.post(
|
||||
"/api/v1/admin/settings/test-email",
|
||||
json={"to": "other@example.com"},
|
||||
headers=_auth_headers(admin),
|
||||
)
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
assert sent["to"] == "other@example.com"
|
||||
|
||||
|
||||
async def test_send_test_email_uses_contact_email_as_reply_to(
|
||||
client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
admin = await _make_user(db_session, role="admin")
|
||||
await db_session.commit()
|
||||
await client.put(
|
||||
"/api/v1/admin/settings",
|
||||
json={"contact_email_enabled": True, "contact_email": "contact@vidconf.example"},
|
||||
headers=_auth_headers(admin),
|
||||
)
|
||||
|
||||
sent: dict[str, object] = {}
|
||||
|
||||
class _FakeBackend:
|
||||
async def send(self, **kwargs: object) -> None:
|
||||
sent.update(kwargs)
|
||||
|
||||
monkeypatch.setattr(admin_module, "create_email_backend", lambda settings: _FakeBackend())
|
||||
|
||||
response = await client.post(
|
||||
"/api/v1/admin/settings/test-email", json={}, headers=_auth_headers(admin)
|
||||
)
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
assert sent["reply_to"] == "contact@vidconf.example"
|
||||
|
||||
|
||||
async def test_send_test_email_reports_transport_failure_without_leaking_secrets(
|
||||
client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Сбой транспорта — понятный текст ошибки; хост/порт видны, логин/пароль — нет."""
|
||||
monkeypatch.setenv("EMAIL_BACKEND", "smtp")
|
||||
monkeypatch.setenv("SMTP_HOST", "smtp.example.com")
|
||||
monkeypatch.setenv("SMTP_PORT", "2525")
|
||||
monkeypatch.setenv("SMTP_USERNAME", "secret-user")
|
||||
monkeypatch.setenv("SMTP_PASSWORD", "super-secret-password")
|
||||
get_settings.cache_clear()
|
||||
|
||||
class _FailingBackend:
|
||||
async def send(self, **kwargs: object) -> None:
|
||||
raise EmailSendError("временный сбой SMTP: сервер недоступен", retryable=True)
|
||||
|
||||
monkeypatch.setattr(admin_module, "create_email_backend", lambda settings: _FailingBackend())
|
||||
try:
|
||||
admin = await _make_user(db_session, role="admin")
|
||||
await db_session.commit()
|
||||
|
||||
response = await client.post(
|
||||
"/api/v1/admin/settings/test-email", json={}, headers=_auth_headers(admin)
|
||||
)
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
body = response.json()
|
||||
assert body["success"] is False
|
||||
assert "сбой SMTP" in body["message"]
|
||||
assert body["smtp_host"] == "smtp.example.com"
|
||||
assert body["smtp_port"] == 2525
|
||||
assert "secret-user" not in response.text
|
||||
assert "super-secret-password" not in response.text
|
||||
finally:
|
||||
get_settings.cache_clear()
|
||||
|
||||
Reference in New Issue
Block a user