# -*- coding: utf-8 -*- """SMTP 发送路径回归:钉住「端口 → TLS 方式」的映射。 为什么值得单独钉:587/25 上必须先 STARTTLS 再 LOGIN,否则 SMTP 授权码会明文过网; 465 则是连上即握手(隐式 TLS)。这两种映射一旦被改反,除了本文件之外没有任何地方 会报警——而它是「凭据是否明文出网」级别的差异。 本测试用假 SMTP 替身记录构造参数与调用顺序,全程不触网(也就不会因外部 SMTP 不可达而失败)。 """ import ssl from typing import Any import pytest from auth_system import mail class _FakeSMTP: """只记录行为、不做任何网络动作的 aiosmtplib.SMTP 替身。""" instances: list['_FakeSMTP'] = [] def __init__(self, **kwargs: Any) -> None: self.kwargs: dict[str, Any] = kwargs self.calls: list[Any] = [] _FakeSMTP.instances.append(self) async def connect(self) -> None: self.calls.append('connect') async def login(self, username: str, password: str) -> None: self.calls.append(('login', username, password)) async def sendmail(self, sender: str, recipients: list[str], message: str) -> None: self.calls.append(('sendmail', sender, tuple(recipients), message)) async def quit(self) -> None: self.calls.append('quit') @pytest.fixture() def fake_smtp(monkeypatch: pytest.MonkeyPatch) -> type[_FakeSMTP]: _FakeSMTP.instances.clear() monkeypatch.setattr(mail.aiosmtplib, 'SMTP', _FakeSMTP) return _FakeSMTP CFG: dict[str, Any] = { 'host': 'smtp.example.com', 'port': 587, 'sender': 'a@example.com', 'password': 'pwd', } def _step_name(call: Any) -> str: return call if isinstance(call, str) else call[0] async def test_port_587_uses_starttls_before_login(fake_smtp: type[_FakeSMTP]) -> None: """587:显式 STARTTLS,且必须握手完成之后才发 LOGIN(凭据不得先出网)。""" await mail._smtp_send(CFG, CFG['sender'], 'to@example.com', 'Raw: x') (s,) = fake_smtp.instances assert s.kwargs['port'] == 587 assert s.kwargs['use_tls'] is False assert s.kwargs['start_tls'] is True # 显式构造的 TLS 上下文——不依赖 aiosmtplib / 运行时的默认校验口径 assert isinstance(s.kwargs['tls_context'], ssl.SSLContext) assert s.kwargs['timeout'] == 15 assert [_step_name(c) for c in s.calls] == ['connect', 'login', 'sendmail', 'quit'] async def test_port_465_uses_implicit_tls(fake_smtp: type[_FakeSMTP]) -> None: """465:隐式 TLS(连上即握手),不能再叠一次 STARTTLS。""" cfg = dict(CFG, port=465) await mail._smtp_send(cfg, cfg['sender'], 'to@example.com', 'Raw: x') (s,) = fake_smtp.instances assert s.kwargs['port'] == 465 assert s.kwargs['use_tls'] is True assert s.kwargs['start_tls'] is False async def test_missing_port_falls_back_to_587(fake_smtp: type[_FakeSMTP]) -> None: """端口缺省/为空时回落到 587,并仍走 STARTTLS 分支。""" cfg = dict(CFG, port='') await mail._smtp_send(cfg, cfg['sender'], 'to@example.com', 'Raw: x') (s,) = fake_smtp.instances assert s.kwargs['port'] == 587 assert s.kwargs['start_tls'] is True async def test_quit_failure_does_not_mask_send_result( fake_smtp: type[_FakeSMTP], monkeypatch: pytest.MonkeyPatch) -> None: """关连接失败不影响「信已发出」的结论,更不得盖掉真实异常。""" async def _boom(self: _FakeSMTP) -> None: raise OSError('connection already closed') monkeypatch.setattr(_FakeSMTP, 'quit', _boom) await mail._smtp_send(CFG, CFG['sender'], 'to@example.com', 'Raw: x') # 不应抛出 async def test_send_failure_propagates_and_still_quits( fake_smtp: type[_FakeSMTP], monkeypatch: pytest.MonkeyPatch) -> None: """发信失败必须原样上抛(send_verify_code 依赖它决定「不写库不入额度」)。""" async def _boom(self: _FakeSMTP, sender: str, recipients: list[str], message: str) -> None: raise RuntimeError('smtp rejected') monkeypatch.setattr(_FakeSMTP, 'sendmail', _boom) with pytest.raises(RuntimeError, match='smtp rejected'): await mail._smtp_send(CFG, CFG['sender'], 'to@example.com', 'Raw: x') (s,) = fake_smtp.instances assert 'quit' in [_step_name(c) for c in s.calls] # finally 里仍关连接