88b8f023f4
Some checks failed
Build and Deploy Backend / build-and-deploy (push) Failing after 1m36s
95 lines
2.5 KiB
Python
95 lines
2.5 KiB
Python
"""
|
|
短信发送测试脚本
|
|
用法: python test_sms.py
|
|
"""
|
|
import os
|
|
import sys
|
|
import django
|
|
|
|
# 初始化 Django
|
|
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
django.setup()
|
|
|
|
from utils.sms import get_sms_client, generate_code, send_sms_code, verify_sms_code
|
|
|
|
|
|
def test_direct_send():
|
|
"""测试1: 直接调用 SDK 发送"""
|
|
print('=' * 50)
|
|
print('测试1: 直接调用阿里云 SMS SDK')
|
|
print('=' * 50)
|
|
|
|
client = get_sms_client()
|
|
if not client.client:
|
|
print('[FAIL] SMS SDK 未初始化，检查 alibabacloud_dysmsapi20170525 是否安装')
|
|
return False
|
|
|
|
code = generate_code()
|
|
phone = '13725102796'
|
|
print(f'手机号: {phone}')
|
|
print(f'验证码: {code}')
|
|
print(f'签名: {client.sign_name}')
|
|
print(f'模板: {client.template_code}')
|
|
print('发送中...')
|
|
|
|
ok, err = client.send_code(phone, code)
|
|
if ok:
|
|
print(f'[OK] 发送成功! 请查收短信，验证码: {code}')
|
|
else:
|
|
print(f'[FAIL] 发送失败: {err}')
|
|
return ok
|
|
|
|
|
|
def test_full_flow():
|
|
"""测试2: 完整流程（发送 + 校验）"""
|
|
print()
|
|
print('=' * 50)
|
|
print('测试2: 完整流程（发送 + 存库 + 校验）')
|
|
print('=' * 50)
|
|
|
|
phone = '13725102796'
|
|
|
|
# 发送
|
|
print(f'[1/3] 发送验证码到 {phone}...')
|
|
ok, err = send_sms_code(phone)
|
|
if not ok:
|
|
print(f'[FAIL] 发送失败: {err}')
|
|
return False
|
|
print('[OK] 发送成功')
|
|
|
|
# 从数据库读取验证码
|
|
from apps.users.models import SmsCode
|
|
record = SmsCode.objects.filter(phone=phone, is_used=False).order_by('-created_at').first()
|
|
if not record:
|
|
print('[FAIL] 数据库中未找到验证码记录')
|
|
return False
|
|
print(f'[2/3] 数据库记录: code={record.code}, expire_at={record.expire_at}')
|
|
|
|
# 校验
|
|
valid, err = verify_sms_code(phone, record.code)
|
|
if valid:
|
|
print('[OK] 验证码校验通过')
|
|
else:
|
|
print(f'[FAIL] 校验失败: {err}')
|
|
return valid
|
|
|
|
|
|
if __name__ == '__main__':
|
|
print('阿里云短信服务测试')
|
|
print(f'AK: {os.environ.get("ALIYUN_ACCESS_KEY_ID", "使用默认值")}')
|
|
print()
|
|
|
|
# 测试1: 只测 SDK 连通性
|
|
ok = test_direct_send()
|
|
|
|
if ok:
|
|
# 测试2: 需要数据库连通
|
|
try:
|
|
test_full_flow()
|
|
except Exception as e:
|
|
print(f'\n[SKIP] 完整流程测试跳过（数据库连接问题）: {e}')
|
|
|
|
print()
|
|
print('测试完成')
|