yml2213
b7b68abf6b
updated_at 为 DATETIME(0)，写库会把小数秒向上取整到下一秒，退避设为 0 时刚失败的记录会被 取数条件 updated_at <= now-退避 取不到，导致 TestDispatchRetriesThenFails 必挂；退避量改为 -2s 保证连续重试可确定执行。
400 lines
12 KiB
Go
400 lines
12 KiB
Go
package smsdispatch
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
|
|
"hfb_sys/backend/internal/database"
|
|
smsprovider "hfb_sys/backend/internal/integrations/sms"
|
|
"hfb_sys/backend/internal/model"
|
|
"hfb_sys/backend/internal/modules/sms"
|
|
|
|
"go.uber.org/zap"
|
|
"gorm.io/datatypes"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
type fakeProvider struct {
|
|
sent []smsprovider.TemplateRequest
|
|
err error
|
|
}
|
|
|
|
func (p *fakeProvider) SendLoginCode(ctx context.Context, phone string, code string) error {
|
|
return nil
|
|
}
|
|
|
|
func (p *fakeProvider) SendTemplate(ctx context.Context, req smsprovider.TemplateRequest) error {
|
|
p.sent = append(p.sent, req)
|
|
return p.err
|
|
}
|
|
|
|
type fakeConfigReader struct {
|
|
value string
|
|
err error
|
|
}
|
|
|
|
func (r *fakeConfigReader) FindValue(ctx context.Context, key string) (string, error) {
|
|
if r.err != nil {
|
|
return "", r.err
|
|
}
|
|
if key != sms.ConfigKey {
|
|
return "", gorm.ErrRecordNotFound
|
|
}
|
|
return r.value, nil
|
|
}
|
|
|
|
type fakeSiteResolver struct {
|
|
provider smsprovider.Provider
|
|
signName string
|
|
templates map[string]string
|
|
err error
|
|
}
|
|
|
|
func (r *fakeSiteResolver) ResolveBusinessSender(ctx context.Context, siteID uint64) (smsprovider.Provider, string, map[string]string, error) {
|
|
return r.provider, r.signName, r.templates, r.err
|
|
}
|
|
|
|
func newDispatchTestDB(t *testing.T) *gorm.DB {
|
|
t.Helper()
|
|
db := database.NewTestDB()
|
|
if err := database.AutoMigrateIfMissing(db, &model.SmsMessage{}); err != nil {
|
|
t.Fatalf("AutoMigrate() error = %v", err)
|
|
}
|
|
return db
|
|
}
|
|
|
|
func configValue(t *testing.T, nodes map[string]sms.NodeConfig) string {
|
|
t.Helper()
|
|
raw, err := sms.MarshalConfig(sms.Config{Enabled: true, Nodes: nodes})
|
|
if err != nil {
|
|
t.Fatalf("MarshalConfig() error = %v", err)
|
|
}
|
|
return raw
|
|
}
|
|
|
|
func seedPending(t *testing.T, db *gorm.DB, row model.SmsMessage) model.SmsMessage {
|
|
t.Helper()
|
|
if row.Status == "" {
|
|
row.Status = model.SmsStatusPending
|
|
}
|
|
if row.Phone == "" {
|
|
row.Phone = "13800000000"
|
|
}
|
|
if err := db.Create(&row).Error; err != nil {
|
|
t.Fatalf("创建待发送记录失败: %v", err)
|
|
}
|
|
return row
|
|
}
|
|
|
|
func loadRow(t *testing.T, db *gorm.DB, id uint64) model.SmsMessage {
|
|
t.Helper()
|
|
var row model.SmsMessage
|
|
if err := db.First(&row, id).Error; err != nil {
|
|
t.Fatalf("查询短信记录失败: %v", err)
|
|
}
|
|
return row
|
|
}
|
|
|
|
func TestDispatchUsesMainChannelWhenNoSite(t *testing.T) {
|
|
db := newDispatchTestDB(t)
|
|
mainProvider := &fakeProvider{}
|
|
reader := &fakeConfigReader{value: configValue(t, map[string]sms.NodeConfig{
|
|
sms.NodeOrderHandoffAssist: {Enabled: true, TemplateCode: "SMS_512540945"},
|
|
})}
|
|
job := New(db, nil, zap.NewNop(), mainProvider, "大锤商行", reader, nil)
|
|
|
|
row := seedPending(t, db, model.SmsMessage{
|
|
UserID: 1,
|
|
Phone: "13800000001",
|
|
Node: sms.NodeOrderHandoffAssist,
|
|
BizType: "order",
|
|
BizID: 1001,
|
|
})
|
|
|
|
handled, err := job.dispatch(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("dispatch() error = %v", err)
|
|
}
|
|
if handled != 1 {
|
|
t.Fatalf("dispatch() handled = %d, want 1", handled)
|
|
}
|
|
if len(mainProvider.sent) != 1 {
|
|
t.Fatalf("主站通道发送次数 = %d, want 1", len(mainProvider.sent))
|
|
}
|
|
req := mainProvider.sent[0]
|
|
if req.TemplateCode != "SMS_512540945" || req.SignName != "大锤商行" || req.Phone != "13800000001" {
|
|
t.Fatalf("请求 = %+v", req)
|
|
}
|
|
|
|
got := loadRow(t, db, row.ID)
|
|
if got.Status != model.SmsStatusSent || got.TemplateCode != "SMS_512540945" || got.SignName != "大锤商行" || got.SentAt == nil {
|
|
t.Fatalf("落库记录 = %+v", got)
|
|
}
|
|
}
|
|
|
|
func TestDispatchPrefersSiteChannelWhenTemplateConfigured(t *testing.T) {
|
|
db := newDispatchTestDB(t)
|
|
mainProvider := &fakeProvider{}
|
|
siteProvider := &fakeProvider{}
|
|
siteID := uint64(9)
|
|
reader := &fakeConfigReader{value: configValue(t, map[string]sms.NodeConfig{
|
|
sms.NodeOrderHandoffAssist: {Enabled: true, TemplateCode: "SMS_MAIN"},
|
|
})}
|
|
resolver := &fakeSiteResolver{
|
|
provider: siteProvider,
|
|
signName: "分站签名",
|
|
templates: map[string]string{sms.NodeOrderHandoffAssist: "SMS_SITE"},
|
|
}
|
|
job := New(db, nil, zap.NewNop(), mainProvider, "大锤商行", reader, resolver)
|
|
|
|
row := seedPending(t, db, model.SmsMessage{
|
|
UserID: 2,
|
|
Node: sms.NodeOrderHandoffAssist,
|
|
BizType: "order",
|
|
BizID: 1002,
|
|
DistributionSiteID: &siteID,
|
|
})
|
|
|
|
if _, err := job.dispatch(context.Background()); err != nil {
|
|
t.Fatalf("dispatch() error = %v", err)
|
|
}
|
|
if len(mainProvider.sent) != 0 {
|
|
t.Fatalf("分站已配置模板时不应使用主站通道")
|
|
}
|
|
if len(siteProvider.sent) != 1 {
|
|
t.Fatalf("分站通道发送次数 = %d, want 1", len(siteProvider.sent))
|
|
}
|
|
if req := siteProvider.sent[0]; req.TemplateCode != "SMS_SITE" || req.SignName != "分站签名" {
|
|
t.Fatalf("分站请求 = %+v", req)
|
|
}
|
|
if got := loadRow(t, db, row.ID); got.TemplateCode != "SMS_SITE" || got.SignName != "分站签名" {
|
|
t.Fatalf("分站落库记录 = %+v", got)
|
|
}
|
|
}
|
|
|
|
func TestDispatchFallsBackToMainWhenSiteTemplateMissing(t *testing.T) {
|
|
db := newDispatchTestDB(t)
|
|
mainProvider := &fakeProvider{}
|
|
siteProvider := &fakeProvider{}
|
|
siteID := uint64(10)
|
|
reader := &fakeConfigReader{value: configValue(t, map[string]sms.NodeConfig{
|
|
sms.NodeListingReviewRejected: {Enabled: true, TemplateCode: "SMS_MAIN"},
|
|
})}
|
|
resolver := &fakeSiteResolver{
|
|
provider: siteProvider,
|
|
signName: "分站签名",
|
|
templates: map[string]string{sms.NodeOrderHandoffAssist: "SMS_SITE"},
|
|
}
|
|
job := New(db, nil, zap.NewNop(), mainProvider, "大锤商行", reader, resolver)
|
|
|
|
row := seedPending(t, db, model.SmsMessage{
|
|
UserID: 3,
|
|
Node: sms.NodeListingReviewRejected,
|
|
BizType: "listing",
|
|
BizID: 1003,
|
|
DistributionSiteID: &siteID,
|
|
})
|
|
|
|
if _, err := job.dispatch(context.Background()); err != nil {
|
|
t.Fatalf("dispatch() error = %v", err)
|
|
}
|
|
if len(siteProvider.sent) != 0 {
|
|
t.Fatalf("分站未配置该节点模板时不应走分站通道")
|
|
}
|
|
if len(mainProvider.sent) != 1 || mainProvider.sent[0].TemplateCode != "SMS_MAIN" {
|
|
t.Fatalf("应回退主站模板，实际 = %+v", mainProvider.sent)
|
|
}
|
|
if got := loadRow(t, db, row.ID); got.Status != model.SmsStatusSent {
|
|
t.Fatalf("status = %q, want sent", got.Status)
|
|
}
|
|
}
|
|
|
|
func TestDispatchFallsBackToMainWhenSiteResolutionFails(t *testing.T) {
|
|
db := newDispatchTestDB(t)
|
|
mainProvider := &fakeProvider{}
|
|
siteID := uint64(11)
|
|
reader := &fakeConfigReader{value: configValue(t, map[string]sms.NodeConfig{
|
|
sms.NodeOrderHandoffAssist: {Enabled: true, TemplateCode: "SMS_MAIN"},
|
|
})}
|
|
resolver := &fakeSiteResolver{err: errors.New("解密失败")}
|
|
job := New(db, nil, zap.NewNop(), mainProvider, "大锤商行", reader, resolver)
|
|
|
|
seedPending(t, db, model.SmsMessage{
|
|
UserID: 4,
|
|
Node: sms.NodeOrderHandoffAssist,
|
|
BizType: "order",
|
|
BizID: 1004,
|
|
DistributionSiteID: &siteID,
|
|
})
|
|
|
|
if _, err := job.dispatch(context.Background()); err != nil {
|
|
t.Fatalf("dispatch() error = %v", err)
|
|
}
|
|
if len(mainProvider.sent) != 1 {
|
|
t.Fatalf("分站解析失败应回退主站通道，实际发送 %d 次", len(mainProvider.sent))
|
|
}
|
|
}
|
|
|
|
func TestDispatchSkipsWhenTemplateMissing(t *testing.T) {
|
|
db := newDispatchTestDB(t)
|
|
mainProvider := &fakeProvider{}
|
|
reader := &fakeConfigReader{value: configValue(t, map[string]sms.NodeConfig{
|
|
sms.NodeOrderHandoffAssist: {Enabled: true, TemplateCode: ""},
|
|
})}
|
|
job := New(db, nil, zap.NewNop(), mainProvider, "大锤商行", reader, nil)
|
|
|
|
row := seedPending(t, db, model.SmsMessage{
|
|
UserID: 5,
|
|
Node: sms.NodeOrderHandoffAssist,
|
|
BizType: "order",
|
|
BizID: 1005,
|
|
})
|
|
|
|
if _, err := job.dispatch(context.Background()); err != nil {
|
|
t.Fatalf("dispatch() error = %v", err)
|
|
}
|
|
if len(mainProvider.sent) != 0 {
|
|
t.Fatalf("未配置模板时不应发送")
|
|
}
|
|
got := loadRow(t, db, row.ID)
|
|
if got.Status != model.SmsStatusSkipped || got.SkipReason == "" {
|
|
t.Fatalf("应标记 skipped 并写原因，实际 = %+v", got)
|
|
}
|
|
}
|
|
|
|
func TestDispatchSkipsWhenNodeDisabled(t *testing.T) {
|
|
db := newDispatchTestDB(t)
|
|
mainProvider := &fakeProvider{}
|
|
reader := &fakeConfigReader{value: configValue(t, map[string]sms.NodeConfig{
|
|
sms.NodeOrderHandoffAssist: {Enabled: false, TemplateCode: "SMS_512540945"},
|
|
})}
|
|
job := New(db, nil, zap.NewNop(), mainProvider, "大锤商行", reader, nil)
|
|
|
|
row := seedPending(t, db, model.SmsMessage{
|
|
UserID: 6,
|
|
Node: sms.NodeOrderHandoffAssist,
|
|
BizType: "order",
|
|
BizID: 1006,
|
|
})
|
|
|
|
if _, err := job.dispatch(context.Background()); err != nil {
|
|
t.Fatalf("dispatch() error = %v", err)
|
|
}
|
|
if len(mainProvider.sent) != 0 {
|
|
t.Fatalf("节点关闭时不应发送")
|
|
}
|
|
if got := loadRow(t, db, row.ID); got.Status != model.SmsStatusSkipped {
|
|
t.Fatalf("status = %q, want skipped", got.Status)
|
|
}
|
|
}
|
|
|
|
func TestDispatchRetriesThenFails(t *testing.T) {
|
|
db := newDispatchTestDB(t)
|
|
mainProvider := &fakeProvider{err: errors.New("渠道异常")}
|
|
reader := &fakeConfigReader{value: configValue(t, map[string]sms.NodeConfig{
|
|
sms.NodeOrderHandoffAssist: {Enabled: true, TemplateCode: "SMS_512540945"},
|
|
})}
|
|
job := New(db, nil, zap.NewNop(), mainProvider, "大锤商行", reader, nil)
|
|
// 退避设为负值，便于在测试内连续重试。
|
|
// updated_at 是 DATETIME(0)，写库时小数秒会向上取整到下一秒（如 .690 → 下一秒），
|
|
// 若把退避设为 0，刚失败的记录会被取数条件 updated_at <= now-退避 误判为「仍在退避中」而取不到。
|
|
job.retryDelay = -2 * time.Second
|
|
|
|
row := seedPending(t, db, model.SmsMessage{
|
|
UserID: 7,
|
|
Node: sms.NodeOrderHandoffAssist,
|
|
BizType: "order",
|
|
BizID: 1007,
|
|
})
|
|
|
|
if _, err := job.dispatch(context.Background()); err != nil {
|
|
t.Fatalf("第一次 dispatch() error = %v", err)
|
|
}
|
|
got := loadRow(t, db, row.ID)
|
|
if got.Status != model.SmsStatusPending || got.RetryCount != 1 || got.LastError == "" {
|
|
t.Fatalf("首次失败应留待重试，实际 = %+v", got)
|
|
}
|
|
|
|
// 第二次达到 maxRetry(3) 前仍是 pending，第三次用尽后置为 failed。
|
|
if _, err := job.dispatch(context.Background()); err != nil {
|
|
t.Fatalf("第二次 dispatch() error = %v", err)
|
|
}
|
|
if _, err := job.dispatch(context.Background()); err != nil {
|
|
t.Fatalf("第三次 dispatch() error = %v", err)
|
|
}
|
|
got = loadRow(t, db, row.ID)
|
|
if got.Status != model.SmsStatusFailed || got.RetryCount != job.maxRetry {
|
|
t.Fatalf("重试用尽后应标记 failed，实际 = %+v", got)
|
|
}
|
|
if len(mainProvider.sent) != job.maxRetry {
|
|
t.Fatalf("发送尝试次数 = %d, want %d", len(mainProvider.sent), job.maxRetry)
|
|
}
|
|
}
|
|
|
|
func TestDispatchPassesTemplateParams(t *testing.T) {
|
|
db := newDispatchTestDB(t)
|
|
mainProvider := &fakeProvider{}
|
|
reader := &fakeConfigReader{value: configValue(t, map[string]sms.NodeConfig{
|
|
sms.NodeOrderHandoffAssist: {Enabled: true, TemplateCode: "SMS_512540945"},
|
|
})}
|
|
job := New(db, nil, zap.NewNop(), mainProvider, "大锤商行", reader, nil)
|
|
|
|
seedPending(t, db, model.SmsMessage{
|
|
UserID: 8,
|
|
Node: sms.NodeOrderHandoffAssist,
|
|
BizType: "order",
|
|
BizID: 1008,
|
|
TemplateParams: datatypes.JSON([]byte(`{"brand_name":"大锤商行"}`)),
|
|
})
|
|
|
|
if _, err := job.dispatch(context.Background()); err != nil {
|
|
t.Fatalf("dispatch() error = %v", err)
|
|
}
|
|
if len(mainProvider.sent) != 1 {
|
|
t.Fatalf("发送次数 = %d, want 1", len(mainProvider.sent))
|
|
}
|
|
if got := mainProvider.sent[0].Params["brand_name"]; got != "大锤商行" {
|
|
t.Fatalf("params.brand_name = %q, want 大锤商行", got)
|
|
}
|
|
}
|
|
|
|
func TestDecodeParamsInvalidJSONFallsBackToNil(t *testing.T) {
|
|
if params := decodeParams(datatypes.JSON([]byte(`{oops`))); params != nil {
|
|
t.Fatalf("非法 JSON 应回落 nil, got %+v", params)
|
|
}
|
|
if params := decodeParams(datatypes.JSON([]byte(`{}`))); params != nil {
|
|
t.Fatalf("空对象应回落 nil, got %+v", params)
|
|
}
|
|
if params := decodeParams(nil); params != nil {
|
|
t.Fatalf("nil 应回落 nil, got %+v", params)
|
|
}
|
|
}
|
|
|
|
func TestDispatchIgnoresRecordsInRetryBackoff(t *testing.T) {
|
|
db := newDispatchTestDB(t)
|
|
mainProvider := &fakeProvider{}
|
|
reader := &fakeConfigReader{value: configValue(t, map[string]sms.NodeConfig{
|
|
sms.NodeOrderHandoffAssist: {Enabled: true, TemplateCode: "SMS_512540945"},
|
|
})}
|
|
job := New(db, nil, zap.NewNop(), mainProvider, "大锤商行", reader, nil)
|
|
job.retryDelay = time.Hour
|
|
|
|
seedPending(t, db, model.SmsMessage{
|
|
UserID: 12,
|
|
Node: sms.NodeOrderHandoffAssist,
|
|
BizType: "order",
|
|
BizID: 1012,
|
|
Status: model.SmsStatusPending,
|
|
RetryCount: 1,
|
|
})
|
|
|
|
if _, err := job.dispatch(context.Background()); err != nil {
|
|
t.Fatalf("dispatch() error = %v", err)
|
|
}
|
|
if len(mainProvider.sent) != 0 {
|
|
t.Fatalf("退避窗口内不应重试")
|
|
}
|
|
}
|