c90b4819c0
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 3m40s
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Validate - Code Quality (push) Failing after 3m21s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 1m42s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 1m42s
CI/CD Pipeline / Unit Tests (push) Failing after 5m59s
CI/CD Pipeline / Integration Tests (push) Successful in 3m26s
CI/CD Pipeline / Frontend Lint (push) Successful in 57s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 1m43s
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Successful in 15m29s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 2m4s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 5m59s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 55s
CI/CD Pipeline / Staging E2E Tests (push) Failing after 35m30s
CI/CD Pipeline / Staging API Integration Tests (push) Failing after 39m55s
158 lines
5.8 KiB
Python
Executable File
158 lines
5.8 KiB
Python
Executable File
"""SMS 短信服务单元测试."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
|
||
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):
|
||
"""发送验证码返回True."""
|
||
svc = NoopSmsService()
|
||
result = svc.send_verification_code("13800138000", "123456")
|
||
assert result is True
|
||
|
||
def test_send_template_sms_returns_true(self):
|
||
"""发送模板短信返回True."""
|
||
svc = NoopSmsService()
|
||
result = svc.send_template_sms(
|
||
"13800138000",
|
||
"SMS_123456",
|
||
{"code": "123456"},
|
||
)
|
||
assert result is True
|
||
|
||
def test_send_verification_code_empty_code(self):
|
||
"""空验证码也返回True(空实现不做校验)."""
|
||
svc = NoopSmsService()
|
||
result = svc.send_verification_code("13800138000", "")
|
||
assert result is True
|
||
|
||
def test_send_template_sms_empty_params(self):
|
||
"""空参数也返回True."""
|
||
svc = NoopSmsService()
|
||
result = svc.send_template_sms("13800138000", "TPL_001", {})
|
||
assert result is True
|
||
|
||
|
||
class TestAliyunSmsServiceInit:
|
||
"""AliyunSmsService 初始化测试."""
|
||
|
||
def test_default_config_from_env(self, monkeypatch):
|
||
"""默认从环境变量读取配置."""
|
||
monkeypatch.setenv("ALIYUN_SMS_ACCESS_KEY_ID", "test_key")
|
||
monkeypatch.setenv("ALIYUN_SMS_ACCESS_KEY_SECRET", "test_secret")
|
||
monkeypatch.setenv("ALIYUN_SMS_SIGN_NAME", "测试签名")
|
||
monkeypatch.setenv("ALIYUN_SMS_VERIFY_TEMPLATE_ID", "SMS_TEST_001")
|
||
|
||
svc = AliyunSmsService()
|
||
assert svc.access_key_id == "test_key"
|
||
assert svc.access_key_secret == "test_secret"
|
||
assert svc.sign_name == "测试签名"
|
||
assert svc.verify_template_id == "SMS_TEST_001"
|
||
|
||
def test_explicit_config_overrides_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_values_when_no_env(self, monkeypatch):
|
||
"""无环境变量时使用默认值."""
|
||
for key in [
|
||
"ALIYUN_SMS_ACCESS_KEY_ID",
|
||
"ALIYUN_SMS_ACCESS_KEY_SECRET",
|
||
"ALIYUN_SMS_SIGN_NAME",
|
||
"ALIYUN_SMS_VERIFY_TEMPLATE_ID",
|
||
]:
|
||
monkeypatch.delenv(key, raising=False)
|
||
|
||
svc = AliyunSmsService()
|
||
assert svc.access_key_id == ""
|
||
assert svc.access_key_secret == ""
|
||
assert svc.sign_name == "小应剪辑"
|
||
assert svc.verify_template_id == "SMS_123456789"
|
||
|
||
def test_send_verification_code_delegates_to_template(self):
|
||
"""send_verification_code 委托给 send_template_sms."""
|
||
svc = AliyunSmsService(
|
||
access_key_id="key",
|
||
access_key_secret="secret",
|
||
verify_template_id="SMS_VERIFY",
|
||
)
|
||
called_with = {}
|
||
|
||
def mock_template_sms(phone, template_id, params):
|
||
called_with["phone"] = phone
|
||
called_with["template_id"] = template_id
|
||
called_with["params"] = params
|
||
return True
|
||
|
||
svc.send_template_sms = mock_template_sms
|
||
result = svc.send_verification_code("13800138000", "654321")
|
||
assert result is True
|
||
assert called_with["phone"] == "13800138000"
|
||
assert called_with["template_id"] == "SMS_VERIFY"
|
||
assert called_with["params"] == {"code": "654321"}
|
||
|
||
def test_send_template_sms_import_error_returns_false(self):
|
||
"""SDK未安装时返回False(ImportError路径)."""
|
||
svc = AliyunSmsService(access_key_id="k", access_key_secret="s")
|
||
# 没有安装SDK时会返回False
|
||
# 由于测试环境可能安装了SDK,这里不强制断言具体结果
|
||
# 只验证函数不会抛异常
|
||
try:
|
||
result = svc.send_template_sms("13800138000", "TPL_001", {"code": "123"})
|
||
assert isinstance(result, bool)
|
||
except Exception as e:
|
||
# SDK可用时可能因为凭证无效而返回False,不应抛未预期的异常
|
||
pytest.fail(f"Unexpected exception: {e}")
|
||
|
||
|
||
class TestGetSmsService:
|
||
"""短信服务工厂函数测试."""
|
||
|
||
def test_default_noop(self, monkeypatch):
|
||
"""默认使用NoopSmsService."""
|
||
monkeypatch.delenv("SMS_PROVIDER", raising=False)
|
||
svc = get_sms_service()
|
||
assert isinstance(svc, NoopSmsService)
|
||
|
||
def test_noop_provider(self, monkeypatch):
|
||
"""显式指定noop provider."""
|
||
monkeypatch.setenv("SMS_PROVIDER", "noop")
|
||
svc = get_sms_service()
|
||
assert isinstance(svc, NoopSmsService)
|
||
|
||
def test_aliyun_provider(self, monkeypatch):
|
||
"""指定aliyun provider返回AliyunSmsService."""
|
||
monkeypatch.setenv("SMS_PROVIDER", "aliyun")
|
||
monkeypatch.setenv("ALIYUN_SMS_ACCESS_KEY_ID", "k")
|
||
monkeypatch.setenv("ALIYUN_SMS_ACCESS_KEY_SECRET", "s")
|
||
svc = get_sms_service()
|
||
assert isinstance(svc, AliyunSmsService)
|
||
|
||
def test_unknown_provider_falls_back_to_noop(self, monkeypatch):
|
||
"""未知provider回退到NoopSmsService."""
|
||
monkeypatch.setenv("SMS_PROVIDER", "unknown_provider_xyz")
|
||
svc = get_sms_service()
|
||
assert isinstance(svc, NoopSmsService)
|
||
|
||
def test_provider_case_insensitive(self, monkeypatch):
|
||
"""provider大小写不敏感."""
|
||
monkeypatch.setenv("SMS_PROVIDER", "ALIYUN")
|
||
monkeypatch.setenv("ALIYUN_SMS_ACCESS_KEY_ID", "k")
|
||
monkeypatch.setenv("ALIYUN_SMS_ACCESS_KEY_SECRET", "s")
|
||
svc = get_sms_service()
|
||
assert isinstance(svc, AliyunSmsService)
|