merge: 补测试批B(settle并发/全链e2e/reconcile装配/alipay健壮性/money零小数位)
This commit is contained in:
wangjia
@@ -0,0 +1,151 @@
|
||||
package gateway_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"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/fake"
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
"github.com/wangjia/pay/internal/util"
|
||||
"github.com/wangjia/pay/internal/webhook"
|
||||
)
|
||||
|
||||
// TestE2EFullChainCallbackToDeliveredWebhook 补齐「下单 → 回调 → settle → webhook
|
||||
// 实际 HTTP 投递 → 标 delivered」整链(各层此前只独立测过):
|
||||
// - 起一个 httptest.Server 当业务方回调接收器,记录收到的请求,验证 X-Pay-* 头齐全 +
|
||||
// 能用同一 secret 验签通过(出站签名互操作,与 internal/webhook/notifier_test.go 同法)。
|
||||
// - fake 渠道下单 → 模拟渠道成功回调(HandleCallback,同 settle_test.go
|
||||
// TestHandleCallbackAndSync 的回调体) → Settle 翻转 paid → outbox 入队。
|
||||
// - 用真实 store.WebhookStore + webhook.Notifier(非 spyEnqueuer)驱动一次
|
||||
// DeliverPending(即 notifier 的投递 RunOnce)→ 断言 webhook 真被投到 test server、
|
||||
// server 收到的 payload 签名正确、outbox 行标记 delivered。
|
||||
func TestE2EFullChainCallbackToDeliveredWebhook(t *testing.T) {
|
||||
const secret = "e2e-fullchain-secret"
|
||||
const bizSystem = "pangolin"
|
||||
|
||||
var gotBody []byte
|
||||
var gotHeaders http.Header
|
||||
var hits int
|
||||
biz := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
hits++
|
||||
gotBody, _ = io.ReadAll(r.Body)
|
||||
gotHeaders = r.Header.Clone()
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = w.Write([]byte("SUCCESS"))
|
||||
}))
|
||||
defer biz.Close()
|
||||
|
||||
db := model.OpenTestDB(t)
|
||||
orders := store.NewOrderStore(db)
|
||||
refunds := store.NewRefundStore(db)
|
||||
subs := store.NewSubscriptionStore(db)
|
||||
chargebacks := store.NewChargebackStore(db)
|
||||
preg := provider.NewRegistry()
|
||||
fp := fake.New()
|
||||
preg.Register(fp)
|
||||
areg := accounts.New([]config.AccountConfig{
|
||||
{AccountID: "fake-a1", Channel: "fake", Region: "global", Enabled: true, Weight: 1},
|
||||
})
|
||||
picker := accounts.NewRouter(areg, nil, nil)
|
||||
|
||||
ws := store.NewWebhookStore(db)
|
||||
// bizConfig:biz_system "pangolin" 的回调地址指向这台 test server(生产由
|
||||
// config.C.BizByName 按业务方名下发,这里直接指定同等语义)。
|
||||
bizCfg := func(system string) (config.BizSystemConfig, bool) {
|
||||
if system == bizSystem {
|
||||
return config.BizSystemConfig{CallbackURL: biz.URL, Secret: secret}, true
|
||||
}
|
||||
return config.BizSystemConfig{}, false
|
||||
}
|
||||
// orderPaid 门禁走真实订单查询(不像多数单测那样恒真),验证投递门禁与真实订单状态联动。
|
||||
orderPaid := func(outTradeNo string) (bool, error) {
|
||||
o, err := orders.GetOrder(outTradeNo)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return o.Status == model.OrderPaidV2, nil
|
||||
}
|
||||
notifier := webhook.NewNotifier(ws, bizCfg, orderPaid)
|
||||
|
||||
g := gateway.New(orders, refunds, preg, picker, stubResolver{}, notifier, "global", subs, chargebacks)
|
||||
|
||||
ctx := context.Background()
|
||||
res, err := g.CreateOrder(ctx, gateway.CreateOrderInput{
|
||||
SKU: "pro_year", Method: "fake", BizSystem: bizSystem, BizRef: "u-e2e-full",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("create order: %v", err)
|
||||
}
|
||||
ref := attemptRef(t, orders)
|
||||
|
||||
// 渠道方成功回调(同 TestHandleCallbackAndSync 的回调体)→ VerifyCallback → Settle。
|
||||
body, _ := json.Marshal(map[string]any{
|
||||
"provider_ref": ref, "status": "succeeded", "amount_minor": 29990000, "currency": "USDT",
|
||||
})
|
||||
got, err := g.HandleCallback(ctx, "fake", provider.CallbackInput{Raw: body})
|
||||
if err != nil || got != gateway.SettleProcessed {
|
||||
t.Fatalf("HandleCallback = %v, %v, want SettleProcessed", got, err)
|
||||
}
|
||||
|
||||
o, err := orders.GetOrder(res.OrderNo)
|
||||
if err != nil || o.Status != model.OrderPaidV2 {
|
||||
t.Fatalf("order = %+v, %v, want paid", o, err)
|
||||
}
|
||||
|
||||
// 入队后未投递:确认 outbox 里躺着一条待发行。
|
||||
pend, err := ws.ListUndelivered(10)
|
||||
if err != nil || len(pend) != 1 || pend[0].OutTradeNo != res.OrderNo {
|
||||
t.Fatalf("undelivered = %+v, %v, want 1 row for %s", pend, err, res.OrderNo)
|
||||
}
|
||||
|
||||
// notifier 投递一轮(RunOnce 语义)→ 真实 HTTP POST 到业务方接收器。
|
||||
sent, err := notifier.DeliverPending(10)
|
||||
if err != nil || sent != 1 {
|
||||
t.Fatalf("DeliverPending = %d, %v, want 1", sent, err)
|
||||
}
|
||||
if hits != 1 {
|
||||
t.Fatalf("业务方应恰收到 1 次 POST, got %d", hits)
|
||||
}
|
||||
|
||||
// 出站签名头齐全 + 能用同一 secret 验签通过(互操作校验,业务方即用同法验签)。
|
||||
sys := gotHeaders.Get("X-Pay-System")
|
||||
ev := gotHeaders.Get("X-Pay-Event")
|
||||
ts := gotHeaders.Get("X-Pay-Timestamp")
|
||||
nonce := gotHeaders.Get("X-Pay-Nonce")
|
||||
sign := gotHeaders.Get("X-Pay-Sign")
|
||||
if sys != bizSystem || ev != "payment.succeeded" || ts == "" || nonce == "" || sign == "" {
|
||||
t.Fatalf("X-Pay-* 头不全: system=%q event=%q ts=%q nonce=%q sign=%q", sys, ev, ts, nonce, sign)
|
||||
}
|
||||
if !util.HMACVerify(secret, sign, sys, ts, nonce, string(gotBody)) {
|
||||
t.Fatalf("业务方侧验签失败(出站签名与 HMACVerify 不互操作)")
|
||||
}
|
||||
var payload map[string]any
|
||||
if err := json.Unmarshal(gotBody, &payload); err != nil {
|
||||
t.Fatalf("payload 非法 JSON: %v, body=%s", err, gotBody)
|
||||
}
|
||||
if payload["event_type"] != "payment.succeeded" || payload["out_trade_no"] != res.OrderNo ||
|
||||
payload["product_biz_code"] != "pro_year" {
|
||||
t.Fatalf("payload = %+v", payload)
|
||||
}
|
||||
|
||||
// outbox 行已标 delivered:再投一次不应重发。
|
||||
pendAfter, err := ws.ListUndelivered(10)
|
||||
if err != nil || len(pendAfter) != 0 {
|
||||
t.Fatalf("投递后 undelivered = %+v, %v, want 0(已标 delivered)", pendAfter, err)
|
||||
}
|
||||
if sent2, _ := notifier.DeliverPending(10); sent2 != 0 {
|
||||
t.Fatalf("已投递不应重发, got %d", sent2)
|
||||
}
|
||||
if hits != 1 {
|
||||
t.Fatalf("重投检查不应产生新 POST, hits=%d", hits)
|
||||
}
|
||||
}
|
||||
@@ -3,8 +3,14 @@ package gateway_test
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/glebarez/sqlite"
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/logger"
|
||||
|
||||
"github.com/wangjia/pay/config"
|
||||
"github.com/wangjia/pay/internal/accounts"
|
||||
"github.com/wangjia/pay/internal/gateway"
|
||||
@@ -12,6 +18,7 @@ import (
|
||||
"github.com/wangjia/pay/internal/provider"
|
||||
"github.com/wangjia/pay/internal/provider/fake"
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
"github.com/wangjia/pay/internal/webhook"
|
||||
)
|
||||
|
||||
func attemptRef(t *testing.T, orders interface {
|
||||
@@ -169,6 +176,118 @@ func TestSettleTransientReadErrorIsFailed(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// openFileGatewaySettleDB 开一个 t.TempDir 下的文件型 sqlite(而非 model.OpenTestDB 的
|
||||
// in-memory cache=shared)专供并发 Settle 测试:cache=shared 的 in-memory sqlite 下,两个
|
||||
// goroutine 各自事务并发写不同表会报 "database table is locked"(SQLITE_LOCKED_SHAREDCACHE,
|
||||
// 共享缓存表级锁——不是 busy_timeout 能重试的 SQLITE_BUSY),同 store/refund_test.go
|
||||
// openFileGuardedDB 的说明。文件型连接各走独立锁路径,规避此问题;DSN 同样带
|
||||
// _txlock=immediate 保留"事务一开始即抢写锁"的守卫语义。
|
||||
func openFileGatewaySettleDB(t *testing.T) *gorm.DB {
|
||||
t.Helper()
|
||||
dsn := fmt.Sprintf("file:%s/settle.db?_txlock=immediate&_pragma=busy_timeout(5000)", t.TempDir())
|
||||
db, err := gorm.Open(sqlite.Open(dsn),
|
||||
&gorm.Config{Logger: logger.Default.LogMode(logger.Silent), TranslateError: true})
|
||||
if err != nil {
|
||||
t.Fatalf("open file settle db: %v", err)
|
||||
}
|
||||
if err := db.AutoMigrate(&model.OrderV2{}, &model.Attempt{}, &model.Account{}, &model.Refund{},
|
||||
&model.WebhookDelivery{}, &model.Product{}, &model.ProductPrice{}, &model.OrphanPayment{},
|
||||
&model.Subscription{}, &model.Chargeback{}); err != nil {
|
||||
t.Fatalf("migrate: %v", err)
|
||||
}
|
||||
if err := model.UpgradeWebhookDeliveryIndex(db); err != nil {
|
||||
t.Fatalf("upgrade webhook_deliveries uq_delivery index: %v", err)
|
||||
}
|
||||
sqlDB, _ := db.DB()
|
||||
t.Cleanup(func() { _ = sqlDB.Close() })
|
||||
return db
|
||||
}
|
||||
|
||||
// TestSettleConcurrentSameOrder 钉住 Settle 依赖的行锁不变量(MarkAttemptPaid 用条件
|
||||
// UPDATE ... WHERE status=pending 天然把并发翻转串行化,同 store/refund_test.go
|
||||
// TestCreateRefundGuardedConcurrentExactlyOneWins 的真并发模式):两个 goroutine 并发对
|
||||
// 同一订单投递同一份成功回调,必须恰好一次真正结算 —— 订单 paid 一次(不重复)、
|
||||
// webhook outbox 对该单恰好一行(不双投)、一个 goroutine 得 Settled、另一个得 Duplicate。
|
||||
//
|
||||
// 用真实 store.WebhookStore + webhook.Notifier(非 spyEnqueuer)驱动:本测试要断言的是
|
||||
// "落库的 outbox 行数",spyEnqueuer 的内存 slice/map 本身也不是并发安全的,不适合这里。
|
||||
// model.OpenTestDB 的 DSN 已带 _txlock=immediate + busy_timeout,并发事务在读阶段即串行化。
|
||||
func TestSettleConcurrentSameOrder(t *testing.T) {
|
||||
db := openFileGatewaySettleDB(t)
|
||||
orders := store.NewOrderStore(db)
|
||||
refunds := store.NewRefundStore(db)
|
||||
subs := store.NewSubscriptionStore(db)
|
||||
chargebacks := store.NewChargebackStore(db)
|
||||
preg := provider.NewRegistry()
|
||||
fp := fake.New()
|
||||
preg.Register(fp)
|
||||
areg := accounts.New([]config.AccountConfig{
|
||||
{AccountID: "fake-a1", Channel: "fake", Region: "global", Enabled: true, Weight: 1},
|
||||
})
|
||||
picker := accounts.NewRouter(areg, nil, nil)
|
||||
ws := store.NewWebhookStore(db)
|
||||
notifier := webhook.NewNotifier(ws,
|
||||
func(string) (config.BizSystemConfig, bool) { return config.BizSystemConfig{}, false },
|
||||
func(string) (bool, error) { return true, nil },
|
||||
)
|
||||
g := gateway.New(orders, refunds, preg, picker, stubResolver{}, notifier, "global", subs, chargebacks)
|
||||
|
||||
ctx := context.Background()
|
||||
res, err := g.CreateOrder(ctx, gateway.CreateOrderInput{SKU: "pro_year", Method: "fake", BizSystem: "pangolin", BizRef: "u-conc"})
|
||||
if err != nil {
|
||||
t.Fatalf("create: %v", err)
|
||||
}
|
||||
ref := attemptRef(t, orders)
|
||||
ev := &provider.PaidEvent{ProviderRef: ref, Status: provider.PaidSucceeded, PaidAmountMinor: 29990000, PaidCurrency: "USDT"}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
results := make([]gateway.SettleResult, 2)
|
||||
errs := make([]error, 2)
|
||||
for i := 0; i < 2; i++ {
|
||||
wg.Add(1)
|
||||
go func(i int) {
|
||||
defer wg.Done()
|
||||
results[i], errs[i] = g.Settle(ctx, ev)
|
||||
}(i)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
for i, gerr := range errs {
|
||||
if gerr != nil {
|
||||
t.Fatalf("goroutine %d unexpected error: %v", i, gerr)
|
||||
}
|
||||
}
|
||||
processed, duplicate := 0, 0
|
||||
for _, r := range results {
|
||||
switch r {
|
||||
case gateway.SettleProcessed:
|
||||
processed++
|
||||
case gateway.SettleDuplicate:
|
||||
duplicate++
|
||||
default:
|
||||
t.Fatalf("unexpected result %v in %v", r, results)
|
||||
}
|
||||
}
|
||||
if processed != 1 || duplicate != 1 {
|
||||
t.Fatalf("results = %v, want exactly one Processed + one Duplicate", results)
|
||||
}
|
||||
|
||||
o, err := orders.GetOrder(res.OrderNo)
|
||||
if err != nil || o.Status != model.OrderPaidV2 {
|
||||
t.Fatalf("order = %+v, %v, want paid exactly once", o, err)
|
||||
}
|
||||
|
||||
var cnt int64
|
||||
if err := db.Model(&model.WebhookDelivery{}).
|
||||
Where("out_trade_no = ? AND event_type = ?", res.OrderNo, "payment.succeeded").
|
||||
Count(&cnt).Error; err != nil {
|
||||
t.Fatalf("count outbox: %v", err)
|
||||
}
|
||||
if cnt != 1 {
|
||||
t.Fatalf("outbox rows for order = %d, want exactly 1(不双投)", cnt)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleCallbackAndSync(t *testing.T) {
|
||||
g, fp, _, orders := newGateway(t)
|
||||
ctx := context.Background()
|
||||
|
||||
Reference in New Issue
Block a user