test(pay): 补 settle 并发/跨包全链 e2e/reconcile 装配/alipay 验签健壮性/money 零小数位
- gateway: TestSettleConcurrentSameOrder 真并发钉住 MarkAttemptPaid 行锁不变量 (文件型 sqlite,规避 in-memory cache=shared 的 SQLITE_LOCKED_SHAREDCACHE)。 - gateway: e2e_fullchain_test.go 补下单→回调→settle→webhook 实际 HTTP 投递→ delivered 整链(httptest server + 真实 store.WebhookStore/webhook.Notifier)。 - reconcile: main.go 装配抽到 reconcile.Assemble(+Runner.TaskNames 访问器), 补 assembly_test.go 钉住 7 个后台任务全部注册 + crypto 缺渠道/interval=0 分支。 - alipay: 补验签健壮性(未知多余字段/字段乱序/中文unicode/空值字段),意外定位 vendor Encoder 对空值字段的真实语义与官方文档描述不同,已记录在测试注释。 - money: 补 TestZeroDecimalCurrencyJPYNotSupported,记录 JPY/KRW 当前未注册进 exponents(接入会先踩 ErrUnknownCurrency,不是乘除法坑)。
This commit is contained in:
wangjia
@@ -0,0 +1,44 @@
|
||||
package reconcile
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"github.com/wangjia/pay/config"
|
||||
"github.com/wangjia/pay/internal/accounts"
|
||||
"github.com/wangjia/pay/internal/gateway"
|
||||
"github.com/wangjia/pay/internal/provider"
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
)
|
||||
|
||||
// Assemble builds the full P6 background-reconcile Runner: order-expire /
|
||||
// usage-refresh / refund-apply-sweep / refund-stuck-alert (local, DB-only,
|
||||
// AddLocal — run synchronously once at startup by main.go's RunOnceLocal) +
|
||||
// sync-pending / paid-spotcheck (network, Add — first tick handles it) +
|
||||
// crypto warm(直跑)+ crypto-orphan-scan(AddCryptoJobs,network,只在 crypto 渠道
|
||||
// 已注册时挂载).
|
||||
//
|
||||
// 抽出这个函数纯粹是为了让"预期任务集合是否都注册上"能被直接单测(main.go 原先把
|
||||
// 这段写死在 main() 里,main() 本身因为要拉真实 DB/HTTP server 不适合单测——见
|
||||
// assembly_test.go)。main.go 对应 `if config.C.Reconcile.Enabled` 分支现在只是薄薄
|
||||
// 一层:调这里 + RunOnceLocal/Start/日志,行为与抽取前逐行一致,未改任何调度语义。
|
||||
func Assemble(orderStore *store.OrderStore, refundStore *store.RefundStore, usage *UsageSource,
|
||||
gw *gateway.Gateway, pReg *provider.Registry, acctReg *accounts.Registry,
|
||||
orphanStore *store.OrphanStore, rc config.ReconcileConfig) *Runner {
|
||||
runner := NewRunner()
|
||||
// 纯本地(DB-only)任务用 AddLocal:会被启动预热 RunOnceLocal 同步跑一遍。
|
||||
runner.AddLocal("order-expire", time.Duration(rc.ExpireEverySec)*time.Second,
|
||||
OrderExpirerTask(orderStore, time.Duration(rc.OrderTTLMin)*time.Minute, time.Now))
|
||||
runner.AddLocal("usage-refresh", time.Duration(rc.UsageEverySec)*time.Second,
|
||||
RefreshUsageTask(usage))
|
||||
runner.AddLocal("refund-apply-sweep", time.Duration(rc.RefundApplyEverySec)*time.Second,
|
||||
RefundApplyTask(orderStore, refundStore, time.Duration(rc.RefundApplyLookbackMin)*time.Minute, time.Now, 200))
|
||||
runner.AddLocal("refund-stuck-alert", time.Duration(rc.RefundApplyEverySec)*time.Second,
|
||||
RefundStuckAlertTask(refundStore, time.Duration(rc.RefundStuckWarnMin)*time.Minute, time.Now))
|
||||
// 网络型(出网 HTTP,10-15s 超时)任务用 Add:启动预热不跑,交各自 ticker 首跳。
|
||||
runner.Add("sync-pending", time.Duration(rc.SyncEverySec)*time.Second,
|
||||
SyncPendingTask(gw, 100))
|
||||
runner.Add("paid-spotcheck", time.Duration(rc.SpotCheckEverySec)*time.Second,
|
||||
PaidSpotCheckTask(orderStore, pReg, time.Duration(rc.SpotCheckWindowMin)*time.Minute, time.Now))
|
||||
AddCryptoJobs(runner, pReg, acctReg, orderStore, orphanStore, rc) // crypto 预留冷启动 Warm(直跑,不受影响)+ 孤儿扫描(网络型,交 ticker 首跳)
|
||||
return runner
|
||||
}
|
||||
@@ -0,0 +1,153 @@
|
||||
package reconcile_test
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/wangjia/pay/config"
|
||||
"github.com/wangjia/pay/internal/accounts"
|
||||
"github.com/wangjia/pay/internal/gateway"
|
||||
"github.com/wangjia/pay/internal/model"
|
||||
"github.com/wangjia/pay/internal/provider"
|
||||
"github.com/wangjia/pay/internal/provider/crypto"
|
||||
"github.com/wangjia/pay/internal/provider/fake"
|
||||
"github.com/wangjia/pay/internal/reconcile"
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
)
|
||||
|
||||
// TestAssembleRegistersAllExpectedTasks 钉住 main.go 里那一长串 reconcile 任务(P6
|
||||
// order-expire/usage-refresh/refund-apply-sweep/refund-stuck-alert/sync-pending/
|
||||
// paid-spotcheck + P8 Task6 crypto-orphan-scan)确实全部被注册 —— main() 原先把这段
|
||||
// 装配写死在函数体内,没有任何测试保证"漏挂一个任务"能被发现;抽成
|
||||
// reconcile.Assemble(见 internal/reconcile/assembly.go)后可以直接断言注册出的任务名
|
||||
// 集合,main.go 现在只是薄薄一层调用(逐行照抄原实现,未改调度语义)。
|
||||
//
|
||||
// pReg 同时注册 fake + crypto 两个渠道:crypto-orphan-scan 只在 crypto 渠道已注册时
|
||||
// 才由 AddCryptoJobs 挂载(见其函数注释),不带 crypto 就测不到这第 7 个任务。
|
||||
func TestAssembleRegistersAllExpectedTasks(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
orders := store.NewOrderStore(db)
|
||||
refunds := store.NewRefundStore(db)
|
||||
subs := store.NewSubscriptionStore(db)
|
||||
chargebacks := store.NewChargebackStore(db)
|
||||
orphans := store.NewOrphanStore(db)
|
||||
|
||||
preg := provider.NewRegistry()
|
||||
fp := fake.New()
|
||||
preg.Register(fp)
|
||||
acctReg := accounts.New([]config.AccountConfig{
|
||||
{AccountID: "fake-a1", Channel: "fake", Region: "global", Enabled: true},
|
||||
{AccountID: "crypto-a1", Channel: "crypto", Region: "global", Enabled: true, CredentialEnvPrefix: "assembletest"},
|
||||
})
|
||||
preg.Register(crypto.New(acctReg))
|
||||
picker := accounts.NewRouter(acctReg, nil, nil)
|
||||
|
||||
usage := reconcile.NewUsageSource(orders, nil)
|
||||
gw := gateway.New(orders, refunds, preg, picker, stubResolver{}, nopEnq{}, "global", subs, chargebacks)
|
||||
|
||||
rc := config.ReconcileConfig{
|
||||
Enabled: true, OrderTTLMin: 30, ExpireEverySec: 60, SyncEverySec: 60, UsageEverySec: 60,
|
||||
SpotCheckEverySec: 60, SpotCheckWindowMin: 60, OrphanEverySec: 60, OrphanWindowMin: 60,
|
||||
RefundApplyEverySec: 60, RefundStuckWarnMin: 60, RefundApplyLookbackMin: 60,
|
||||
}
|
||||
runner := reconcile.Assemble(orders, refunds, usage, gw, preg, acctReg, orphans, rc)
|
||||
|
||||
want := []string{
|
||||
"order-expire", "usage-refresh", "refund-apply-sweep", "refund-stuck-alert",
|
||||
"sync-pending", "paid-spotcheck", "crypto-orphan-scan",
|
||||
}
|
||||
got := runner.TaskNames()
|
||||
if len(got) != len(want) {
|
||||
t.Fatalf("task names = %v (%d), want %d: %v", got, len(got), len(want), want)
|
||||
}
|
||||
seen := make(map[string]bool, len(got))
|
||||
for _, n := range got {
|
||||
seen[n] = true
|
||||
}
|
||||
for _, w := range want {
|
||||
if !seen[w] {
|
||||
t.Fatalf("缺任务 %q,got %v", w, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestAssembleSkipsCryptoOrphanScanWithoutCryptoChannel crypto 渠道未启用时(providerbuild
|
||||
// 按账户配置决定是否注册,见 internal/providerbuild),AddCryptoJobs 安全跳过——不应
|
||||
// 注册 crypto-orphan-scan,其余 6 个本地/网络任务照常注册,总数少一个。
|
||||
func TestAssembleSkipsCryptoOrphanScanWithoutCryptoChannel(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
orders := store.NewOrderStore(db)
|
||||
refunds := store.NewRefundStore(db)
|
||||
subs := store.NewSubscriptionStore(db)
|
||||
chargebacks := store.NewChargebackStore(db)
|
||||
orphans := store.NewOrphanStore(db)
|
||||
|
||||
preg := provider.NewRegistry() // 只 fake,没有 crypto
|
||||
fp := fake.New()
|
||||
preg.Register(fp)
|
||||
acctReg := accounts.New([]config.AccountConfig{
|
||||
{AccountID: "fake-a1", Channel: "fake", Region: "global", Enabled: true},
|
||||
})
|
||||
picker := accounts.NewRouter(acctReg, nil, nil)
|
||||
|
||||
usage := reconcile.NewUsageSource(orders, nil)
|
||||
gw := gateway.New(orders, refunds, preg, picker, stubResolver{}, nopEnq{}, "global", subs, chargebacks)
|
||||
|
||||
rc := config.ReconcileConfig{
|
||||
Enabled: true, OrderTTLMin: 30, ExpireEverySec: 60, SyncEverySec: 60, UsageEverySec: 60,
|
||||
SpotCheckEverySec: 60, SpotCheckWindowMin: 60, OrphanEverySec: 60, OrphanWindowMin: 60,
|
||||
RefundApplyEverySec: 60, RefundStuckWarnMin: 60, RefundApplyLookbackMin: 60,
|
||||
}
|
||||
runner := reconcile.Assemble(orders, refunds, usage, gw, preg, acctReg, orphans, rc)
|
||||
|
||||
got := runner.TaskNames()
|
||||
if len(got) != 6 {
|
||||
t.Fatalf("无 crypto 渠道时应只注册 6 个任务(缺 crypto-orphan-scan), got %v", got)
|
||||
}
|
||||
for _, n := range got {
|
||||
if n == "crypto-orphan-scan" {
|
||||
t.Fatalf("未启用 crypto 渠道时不应注册 crypto-orphan-scan, got %v", got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestAssembleZeroIntervalSkipsThatTaskOnly 覆盖 Runner.add 的 interval<=0 守卫在装配层
|
||||
// 的联动:某一项周期配成 0(operator 配置疏漏,如 usage_every_sec: 0)只会让那一个任务
|
||||
// 不注册,其余任务不受影响(呼应 runner_test.go 的
|
||||
// TestRunnerZeroIntervalSkipsRegistrationAndStartDoesNotPanic,但那里是直接摆弄 Runner,
|
||||
// 这里钉住 Assemble 真按 config 字段逐个转发 interval,没有哪个任务共用错了字段)。
|
||||
func TestAssembleZeroIntervalSkipsThatTaskOnly(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
orders := store.NewOrderStore(db)
|
||||
refunds := store.NewRefundStore(db)
|
||||
subs := store.NewSubscriptionStore(db)
|
||||
chargebacks := store.NewChargebackStore(db)
|
||||
orphans := store.NewOrphanStore(db)
|
||||
|
||||
preg := provider.NewRegistry()
|
||||
fp := fake.New()
|
||||
preg.Register(fp)
|
||||
acctReg := accounts.New([]config.AccountConfig{{AccountID: "fake-a1", Channel: "fake", Region: "global", Enabled: true}})
|
||||
picker := accounts.NewRouter(acctReg, nil, nil)
|
||||
|
||||
usage := reconcile.NewUsageSource(orders, nil)
|
||||
gw := gateway.New(orders, refunds, preg, picker, stubResolver{}, nopEnq{}, "global", subs, chargebacks)
|
||||
|
||||
rc := config.ReconcileConfig{
|
||||
Enabled: true, OrderTTLMin: 30, ExpireEverySec: 60, SyncEverySec: 60,
|
||||
UsageEverySec: 0, // 疏漏:只这一项配成 0
|
||||
SpotCheckEverySec: 60, SpotCheckWindowMin: 60, OrphanEverySec: 60, OrphanWindowMin: 60,
|
||||
RefundApplyEverySec: 60, RefundStuckWarnMin: 60, RefundApplyLookbackMin: 60,
|
||||
}
|
||||
runner := reconcile.Assemble(orders, refunds, usage, gw, preg, acctReg, orphans, rc)
|
||||
|
||||
got := runner.TaskNames()
|
||||
for _, n := range got {
|
||||
if n == "usage-refresh" {
|
||||
t.Fatalf("usage_every_sec=0 时 usage-refresh 不应注册, got %v", got)
|
||||
}
|
||||
}
|
||||
// 其余 5 个(order-expire/refund-apply-sweep/refund-stuck-alert/sync-pending/paid-spotcheck)照常。
|
||||
if len(got) != 5 {
|
||||
t.Fatalf("只 usage-refresh 应被跳过,其余应全注册, got %v", got)
|
||||
}
|
||||
}
|
||||
@@ -86,6 +86,16 @@ func (r *Runner) RunOnceLocal(ctx context.Context) {
|
||||
}
|
||||
}
|
||||
|
||||
// TaskNames 返回已注册任务名(注册序)。供装配期测试断言"预期任务集合是否都注册上"
|
||||
// (见 assembly.go::Assemble 与 assembly_test.go),不用于运行期逻辑。
|
||||
func (r *Runner) TaskNames() []string {
|
||||
names := make([]string, len(r.tasks))
|
||||
for i, t := range r.tasks {
|
||||
names[i] = t.Name
|
||||
}
|
||||
return names
|
||||
}
|
||||
|
||||
// Start 每任务一 goroutine + 独立 ticker 常驻;ctx 取消即退出。每 tick 崩溃安全。
|
||||
func (r *Runner) Start(ctx context.Context) {
|
||||
for _, t := range r.tasks {
|
||||
|
||||
Reference in New Issue
Block a user