"""Тесты отправки писем и шаблонов.""" import dataclasses import logging import threading import unittest from unittest import mock from splitapi import mailer from splitapi.config import from_env CFG = from_env({"PUBLIC_URL": "https://example.com/trainer", "SMTP_HOST": "smtp.example.com", "SMTP_USER": "user", "SMTP_PASSWORD": "pw", "MAIL_FROM": "Тренер "}) class TemplateTests(unittest.TestCase): def test_verify_mail_has_fragment_link_not_query(self): mail = mailer.verify_mail(CFG, "u@example.com", "TOKEN123") self.assertEqual(mail.to, "u@example.com") self.assertIn("https://example.com/trainer/#verify=TOKEN123", mail.body) self.assertNotIn("?verify", mail.body) def test_reset_mail_has_fragment_link(self): mail = mailer.reset_mail(CFG, "u@example.com", "TOK") self.assertIn("https://example.com/trainer/#reset=TOK", mail.body) self.assertIn("1 час", mail.body) def test_validity_period_comes_from_the_config_not_from_a_constant(self): longer = dataclasses.replace(CFG, reset_ttl=2 * 3600, verify_ttl=3 * 24 * 3600) shorter = dataclasses.replace(CFG, reset_ttl=30 * 60) self.assertIn("2 часа", mailer.reset_mail(longer, "u@example.com", "T").body) self.assertIn("3 дня", mailer.verify_mail(longer, "u@example.com", "T").body) self.assertIn("30 минут", mailer.reset_mail(shorter, "u@example.com", "T").body) def test_human_duration_uses_correct_russian_plurals(self): cases = {3600: "1 час", 7200: "2 часа", 5 * 3600: "5 часов", 21 * 3600: "21 час", 24 * 3600: "24 часа", 86400 * 3: "3 дня", 86400 * 7: "7 дней", 60: "1 минута", 120: "2 минуты", 15 * 60: "15 минут"} for seconds, expected in cases.items(): with self.subTest(seconds=seconds): self.assertEqual(mailer.human_duration(seconds), expected) def test_notice_mails_have_no_tokens(self): for mail in (mailer.already_registered_mail(CFG, "u@example.com"), mailer.password_changed_mail(CFG, "u@example.com")): self.assertNotIn("#verify=", mail.body) self.assertNotIn("#reset=", mail.body) def test_messages_are_russian_and_have_subject(self): for mail in (mailer.verify_mail(CFG, "u@example.com", "T"), mailer.reset_mail(CFG, "u@example.com", "T")): self.assertTrue(mail.subject) self.assertRegex(mail.body, r"[А-Яа-я]") class BuildMessageTests(unittest.TestCase): def test_headers_and_utf8_body(self): msg = mailer.build_message(mailer.Mail("u@example.com", "Тема письма", "Текст на русском"), CFG.mail_from) self.assertEqual(msg["To"], "u@example.com") self.assertIn("noreply@example.com", msg["From"]) self.assertEqual(msg.get_content().strip(), "Текст на русском") def test_header_injection_is_impossible(self): for bad_to, bad_subject in [("a@example.com\r\nBcc: evil@x.org", "ok"), ("a@example.com", "x\r\nBcc: evil@x.org")]: with self.subTest(to=bad_to, subject=bad_subject), self.assertRaises(ValueError): mailer.build_message(mailer.Mail(bad_to, bad_subject, "body"), CFG.mail_from) class MemoryMailerTests(unittest.TestCase): def test_collects_messages(self): m = mailer.MemoryMailer() m.send(mailer.Mail("a@example.com", "s", "b")) self.assertEqual([x.to for x in m.sent], ["a@example.com"]) class ConsoleMailerTests(unittest.TestCase): def test_prints_message_to_log(self): with self.assertLogs("splitapi.mail", level=logging.WARNING) as logs: mailer.ConsoleMailer().send(mailer.Mail("a@example.com", "Тема", "Ссылка: https://x/#verify=ABC")) self.assertIn("#verify=ABC", "\n".join(logs.output)) class SmtpMailerTests(unittest.TestCase): def _send(self, security: str, port: int): cfg = from_env({"PUBLIC_URL": "https://example.com", "SMTP_HOST": "smtp.example.com", "SMTP_PORT": str(port), "SMTP_SECURITY": security, "SMTP_USER": "user", "SMTP_PASSWORD": "pw"}) target = "smtplib.SMTP_SSL" if security == "ssl" else "smtplib.SMTP" with mock.patch(target) as smtp_cls: server = smtp_cls.return_value.__enter__.return_value mailer.SmtpMailer(cfg).send(mailer.Mail("u@example.com", "Тема", "Тело")) return smtp_cls, server def test_starttls_logs_in_and_sends(self): smtp_cls, server = self._send("starttls", 587) self.assertEqual(smtp_cls.call_args.args[:2], ("smtp.example.com", 587)) server.starttls.assert_called_once() server.login.assert_called_once_with("user", "pw") server.send_message.assert_called_once() def test_ssl_does_not_call_starttls(self): smtp_cls, server = self._send("ssl", 465) server.starttls.assert_not_called() server.login.assert_called_once() server.send_message.assert_called_once() def test_none_security_without_credentials_skips_login(self): cfg = from_env({"PUBLIC_URL": "https://example.com", "SMTP_HOST": "localhost", "SMTP_SECURITY": "none", "SMTP_PORT": "25"}) with mock.patch("smtplib.SMTP") as smtp_cls: server = smtp_cls.return_value.__enter__.return_value mailer.SmtpMailer(cfg).send(mailer.Mail("u@example.com", "s", "b")) server.starttls.assert_not_called() server.login.assert_not_called() server.send_message.assert_called_once() def test_starttls_uses_verified_tls_context(self): _, server = self._send("starttls", 587) context = server.starttls.call_args.kwargs.get("context") self.assertIsNotNone(context) self.assertEqual(context.verify_mode.name, "CERT_REQUIRED") class AsyncMailerTests(unittest.TestCase): def test_delivers_in_background(self): inner = mailer.MemoryMailer() am = mailer.AsyncMailer(inner) try: am.send(mailer.Mail("a@example.com", "s", "b")) am.flush(timeout=5) self.assertEqual(len(inner.sent), 1) finally: am.close() def test_failure_is_logged_and_worker_survives(self): class Flaky: def __init__(self): self.calls = 0 self.done = threading.Event() def send(self, mail): self.calls += 1 if self.calls == 1: raise OSError("smtp down") self.done.set() inner = Flaky() am = mailer.AsyncMailer(inner) try: with self.assertLogs("splitapi.mail", level=logging.ERROR) as logs: am.send(mailer.Mail("a@example.com", "s", "b")) am.flush(timeout=5) self.assertIn("OSError", "\n".join(logs.output)) am.send(mailer.Mail("b@example.com", "s", "b")) self.assertTrue(inner.done.wait(5)) finally: am.close() def test_error_text_with_an_address_is_not_written_to_the_log(self): class Refusing: def send(self, mail): raise OSError("recipient refused: private.person@secret-domain.example") am = mailer.AsyncMailer(Refusing()) try: with self.assertLogs("splitapi.mail", level=logging.ERROR) as logs: am.send(mailer.Mail("a@example.com", "s", "b")) am.flush(timeout=5) text = "\n".join(logs.output) self.assertNotIn("private.person", text) self.assertNotIn("secret-domain", text) finally: am.close() def test_full_queue_raises_instead_of_silently_dropping(self): gate = threading.Event() delivered = [] class Slow: def send(self, mail): gate.wait(5) delivered.append(mail) am = mailer.AsyncMailer(Slow(), maxsize=2) rejected = 0 try: with self.assertLogs("splitapi.mail", level=logging.ERROR): for _ in range(10): try: am.send(mailer.Mail("a@example.com", "s", "b")) except mailer.MailQueueFull: rejected += 1 self.assertGreaterEqual(rejected, 7) # из 10 принято не больше 3: одно в работе, два в очереди finally: gate.set() am.flush(timeout=5) am.close() self.assertEqual(len(delivered) + rejected, 10) # каждое письмо либо доставлено, либо явно отклонено class MakeMailerTests(unittest.TestCase): def test_selects_backend(self): self.assertIsInstance(mailer.make_mailer(from_env({"PUBLIC_URL": "https://x.org", "MAIL_BACKEND": "console"})), mailer.ConsoleMailer) m = mailer.make_mailer(CFG) try: self.assertIsInstance(m, mailer.AsyncMailer) finally: m.close() if __name__ == "__main__": unittest.main()