233 lines
8.7 KiB
Python
233 lines
8.7 KiB
Python
"""Notification email rendering and enqueue trigger tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import unittest
|
|
from contextlib import contextmanager
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from .config import settings
|
|
from .routes import sessions
|
|
from .services import notifications, state_machine
|
|
from .services.persona import P1
|
|
from .store import InProcSession, TurnRecord
|
|
|
|
|
|
@contextmanager
|
|
def patched_settings(**values: Any):
|
|
previous = {key: getattr(settings, key) for key in values}
|
|
for key, value in values.items():
|
|
setattr(settings, key, value)
|
|
try:
|
|
yield
|
|
finally:
|
|
for key, value in previous.items():
|
|
setattr(settings, key, value)
|
|
|
|
|
|
def _ended_session() -> InProcSession:
|
|
state = state_machine.init_state(params=P1.openness_params())
|
|
state.stage = state_machine.Stage.EXPLORE
|
|
return InProcSession(
|
|
session_id="00000000-0000-0000-0000-00000000e111",
|
|
case_id="00000000-0000-0000-0000-00000000e111",
|
|
learner_id="00000000-0000-0000-0000-000000000111",
|
|
persona_code=P1.code,
|
|
theory_mode="humanistic",
|
|
persona=P1,
|
|
state=state,
|
|
session_no=1,
|
|
created_at=1_000.0,
|
|
ended_at=1_600.0,
|
|
ended=True,
|
|
turns=[
|
|
TurnRecord(
|
|
turn_seq=1,
|
|
speaker="counselor",
|
|
stage=state.stage.value,
|
|
text="상담자 발화",
|
|
text_masked="상담자 발화",
|
|
),
|
|
TurnRecord(
|
|
turn_seq=2,
|
|
speaker="client",
|
|
stage=state.stage.value,
|
|
text="내담자 응답",
|
|
text_masked="내담자 응답",
|
|
),
|
|
],
|
|
)
|
|
|
|
|
|
class NotificationEmailTemplateTest(unittest.TestCase):
|
|
def test_session_review_email_uses_polite_copy_and_direct_review_link(self) -> None:
|
|
rendered = notifications.render_email(
|
|
"session_review_ready",
|
|
{
|
|
"learner_label": "테스트 학습자",
|
|
"persona_name": "서연",
|
|
"session_no": 2,
|
|
"ended_at": "2026-06-29T09:00:00+09:00",
|
|
"review_url": "https://vignette.chanpaca.net/teach/session/s1/review",
|
|
"transcript": "메일에 들어가면 안 되는 축어록",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(rendered.subject, "[Vignette] 검토할 회기가 있습니다")
|
|
self.assertIn("검토할 회기가 있습니다.", rendered.html)
|
|
self.assertIn("회기 검토하기", rendered.html)
|
|
self.assertIn("https://vignette.chanpaca.net/teach/session/s1/review", rendered.html)
|
|
self.assertIn("민감한 회기 내용은 메일에 포함하지 않았습니다", rendered.html)
|
|
self.assertNotIn("메일에 들어가면 안 되는 축어록", rendered.html)
|
|
|
|
def test_account_approval_email_links_admin_users(self) -> None:
|
|
rendered = notifications.render_email(
|
|
"account_pending_approval",
|
|
{
|
|
"display_name": "신규 사용자",
|
|
"email": "new@hs.ac.kr",
|
|
"role": "learner",
|
|
"approval_url": "https://vignette.chanpaca.net/admin/users",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(rendered.subject, "[Vignette] 새 가입 승인 요청이 있습니다")
|
|
self.assertIn("새 가입 승인 요청이 있습니다.", rendered.html)
|
|
self.assertIn("가입 승인 확인하기", rendered.html)
|
|
self.assertIn("https://vignette.chanpaca.net/admin/users", rendered.html)
|
|
|
|
def test_admin_test_email_uses_distinct_test_copy(self) -> None:
|
|
rendered = notifications.render_email(
|
|
"admin_test_email",
|
|
{
|
|
"requested_by": "admin@hs.ac.kr",
|
|
"notifications_url": "https://vignette.chanpaca.net/admin",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(rendered.subject, "[Vignette] 메일 알림 테스트입니다")
|
|
self.assertIn("메일 알림 테스트입니다.", rendered.html)
|
|
self.assertIn("발송 경로 확인용 테스트", rendered.html)
|
|
self.assertIn("알림 상태 확인하기", rendered.html)
|
|
self.assertIn("https://vignette.chanpaca.net/admin", rendered.html)
|
|
|
|
def test_smtp_sender_uses_html_text_parts_and_starttls(self) -> None:
|
|
rendered = notifications.render_email(
|
|
"session_review_ready",
|
|
{
|
|
"learner_label": "테스트 학습자",
|
|
"persona_name": "서연",
|
|
"session_no": 1,
|
|
"review_url": "https://vignette.chanpaca.net/teach/session/s1/review",
|
|
},
|
|
)
|
|
|
|
with (
|
|
patched_settings(
|
|
notification_email_provider="smtp",
|
|
smtp_host="smtp.example.test",
|
|
smtp_port=587,
|
|
smtp_username="mailer",
|
|
smtp_password="secret",
|
|
smtp_from_email="no-reply@example.test",
|
|
smtp_from_name="Vignette",
|
|
smtp_starttls=True,
|
|
smtp_ssl=False,
|
|
),
|
|
patch.object(notifications.smtplib, "SMTP") as smtp_class,
|
|
):
|
|
smtp = smtp_class.return_value.__enter__.return_value
|
|
provider_id = notifications._send_rendered_email(
|
|
recipient_email="teacher@hs.ac.kr",
|
|
recipient_name="교수자",
|
|
rendered=rendered,
|
|
)
|
|
|
|
self.assertIn("@example.test>", provider_id)
|
|
smtp_class.assert_called_once_with("smtp.example.test", 587, timeout=15)
|
|
smtp.starttls.assert_called_once()
|
|
smtp.login.assert_called_once_with("mailer", "secret")
|
|
smtp.send_message.assert_called_once()
|
|
message = smtp.send_message.call_args.args[0]
|
|
self.assertEqual(message["Subject"], "[Vignette] 검토할 회기가 있습니다")
|
|
self.assertIn("teacher@hs.ac.kr", message["To"])
|
|
self.assertTrue(message.is_multipart())
|
|
|
|
|
|
class NotificationTriggerTest(unittest.IsolatedAsyncioTestCase):
|
|
async def test_failed_session_evaluation_still_enqueues_teacher_review_notification(self) -> None:
|
|
sess = _ended_session()
|
|
|
|
with (
|
|
patch.object(
|
|
sessions.evaluator,
|
|
"evaluate_session",
|
|
AsyncMock(side_effect=RuntimeError("engine offline")),
|
|
),
|
|
patch.object(
|
|
sessions.session_persistence,
|
|
"save_session_evaluation",
|
|
AsyncMock(return_value=True),
|
|
) as save_evaluation,
|
|
patch.object(
|
|
sessions.notifications,
|
|
"enqueue_session_review_ready",
|
|
AsyncMock(),
|
|
) as enqueue_review,
|
|
):
|
|
await sessions._generate_and_save_session_evaluation(sess)
|
|
|
|
save_evaluation.assert_awaited_once()
|
|
enqueue_review.assert_awaited_once_with(session_id=sess.session_id)
|
|
|
|
async def test_session_evaluation_timeout_records_actionable_error(self) -> None:
|
|
sess = _ended_session()
|
|
|
|
with (
|
|
patched_settings(session_evaluation_timeout=1.0),
|
|
patch.object(
|
|
sessions.evaluator,
|
|
"evaluate_session",
|
|
AsyncMock(side_effect=asyncio.TimeoutError()),
|
|
),
|
|
patch.object(
|
|
sessions.session_persistence,
|
|
"save_session_evaluation",
|
|
AsyncMock(return_value=True),
|
|
) as save_evaluation,
|
|
patch.object(
|
|
sessions.notifications,
|
|
"enqueue_session_review_ready",
|
|
AsyncMock(),
|
|
),
|
|
):
|
|
await sessions._generate_and_save_session_evaluation(sess)
|
|
|
|
saved_write = save_evaluation.await_args.args[0]
|
|
self.assertEqual(saved_write.status, "error")
|
|
self.assertEqual(saved_write.error, "session evaluation timeout after 1s")
|
|
|
|
async def test_scheduled_session_evaluation_task_crash_is_observed(self) -> None:
|
|
sess = _ended_session()
|
|
|
|
async def boom(_sess: InProcSession) -> None:
|
|
raise RuntimeError("background crash")
|
|
|
|
with patch.object(sessions, "_generate_and_save_session_evaluation", boom):
|
|
with self.assertLogs(sessions.logger, level="ERROR") as logs:
|
|
task = sessions._schedule_session_evaluation(sess)
|
|
self.assertIsNotNone(task)
|
|
assert task is not None
|
|
with self.assertRaises(RuntimeError):
|
|
await task
|
|
|
|
self.assertTrue(
|
|
any("session evaluation background task crashed" in line for line in logs.output),
|
|
logs.output,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|