"""SMS Service 单元测试""" from __future__ import annotations from unittest.mock import MagicMock, patch import pytest from packages.adapters.sms.sms_service import ( AliyunSmsService, NoopSmsService, get_sms_service, ) class TestNoopSmsService: """NoopSmsService 测试""" def test_send_verification_code_returns_true(self): svc = NoopSmsService() assert svc.send_verification_code("13800138000", "123456") is True def test_send_template_sms_returns_true(self): svc = NoopSmsService() assert svc.send_template_sms("13800138000", "SMS_123", {"code": "123456"}) is True def test_send_verification_code_empty_code(self): svc = NoopSmsService() assert svc.send_verification_code("13800138000", "") is True class TestAliyunSmsServiceInit: """AliyunSmsService 初始化测试""" def test_default_values_from_env(self, monkeypatch): monkeypatch.setenv("ALIYUN_SMS_ACCESS_KEY_ID", "env_key") monkeypatch.setenv("ALIYUN_SMS_ACCESS_KEY_SECRET", "env_secret") monkeypatch.setenv("ALIYUN_SMS_SIGN_NAME", "env_sign") monkeypatch.setenv("ALIYUN_SMS_VERIFY_TEMPLATE_ID", "env_tpl") svc = AliyunSmsService() assert svc.access_key_id == "env_key" assert svc.access_key_secret == "env_secret" assert svc.sign_name == "env_sign" assert svc.verify_template_id == "env_tpl" def test_explicit_params_override_env(self, monkeypatch): monkeypatch.setenv("ALIYUN_SMS_ACCESS_KEY_ID", "env_key") svc = AliyunSmsService(access_key_id="explicit_key") assert svc.access_key_id == "explicit_key" def test_default_sign_name(self, monkeypatch): monkeypatch.delenv("ALIYUN_SMS_SIGN_NAME", raising=False) svc = AliyunSmsService() assert svc.sign_name == "小应剪辑" def test_default_template_id(self, monkeypatch): monkeypatch.delenv("ALIYUN_SMS_VERIFY_TEMPLATE_ID", raising=False) svc = AliyunSmsService() assert svc.verify_template_id == "SMS_123456789" class TestAliyunSmsServiceSend: """发送短信测试(mock SDK)""" @pytest.fixture def svc(self): return AliyunSmsService( access_key_id="key", access_key_secret="secret", sign_name="测试签名", verify_template_id="SMS_VERIFY", ) def test_send_verification_code_delegates_to_template(self, svc): """验证码调用 send_template_sms""" with patch.object(svc, "send_template_sms", return_value=True) as mock_send: result = svc.send_verification_code("13800138000", "654321") assert result is True mock_send.assert_called_once_with("13800138000", "SMS_VERIFY", {"code": "654321"}) def test_send_template_sms_success(self, svc): """发送成功返回 True""" mock_body = MagicMock() mock_body.code = "OK" mock_body.message = "OK" mock_response = MagicMock() mock_response.body = mock_body with patch.dict("sys.modules"): # mock 整个 alibabacloud 模块 mock_client_cls = MagicMock() mock_client_cls.return_value.send_sms.return_value = mock_response mock_dysms_models = MagicMock() mock_dysms_models.SendSmsRequest = MagicMock(return_value=MagicMock()) mock_openapi_models = MagicMock() mock_openapi_models.Config = MagicMock() with patch.object(svc, "_AliyunSmsService__import_sdk", create=True): pass # 直接 patch 模块名来模拟 SDK 存在 import sys sys.modules["alibabacloud_dysmsapi20170525"] = MagicMock() sys.modules["alibabacloud_dysmsapi20170525.models"] = mock_dysms_models sys.modules["alibabacloud_dysmsapi20170525.client"] = MagicMock(Client=mock_client_cls) sys.modules["alibabacloud_tea_openapi"] = MagicMock() sys.modules["alibabacloud_tea_openapi.models"] = mock_openapi_models try: result = svc.send_template_sms("13800138000", "SMS_TPL", {"code": "123"}) assert result is True finally: for key in [ "alibabacloud_dysmsapi20170525", "alibabacloud_dysmsapi20170525.models", "alibabacloud_dysmsapi20170525.client", "alibabacloud_tea_openapi", "alibabacloud_tea_openapi.models", ]: sys.modules.pop(key, None) def test_send_template_sms_sdk_not_installed(self, svc): """SDK 未安装返回 False""" with patch.object(svc, "send_template_sms"): pass # 确保没有 SDK 时返回 False import sys saved_modules = {} for key in list(sys.modules.keys()): if "alibabacloud" in key: saved_modules[key] = sys.modules.pop(key) try: result = svc.send_template_sms("13800138000", "tpl", {}) assert result is False finally: sys.modules.update(saved_modules) class TestGetSmsService: """工厂函数测试""" def test_default_noop(self, monkeypatch): monkeypatch.delenv("SMS_PROVIDER", raising=False) svc = get_sms_service() assert isinstance(svc, NoopSmsService) def test_noop_provider(self, monkeypatch): monkeypatch.setenv("SMS_PROVIDER", "noop") svc = get_sms_service() assert isinstance(svc, NoopSmsService) def test_aliyun_provider(self, monkeypatch): monkeypatch.setenv("SMS_PROVIDER", "aliyun") svc = get_sms_service() assert isinstance(svc, AliyunSmsService) def test_case_insensitive_provider(self, monkeypatch): monkeypatch.setenv("SMS_PROVIDER", "AliYun") svc = get_sms_service() assert isinstance(svc, AliyunSmsService) def test_unknown_provider_falls_back_to_noop(self, monkeypatch): monkeypatch.setenv("SMS_PROVIDER", "unknown") svc = get_sms_service() assert isinstance(svc, NoopSmsService)