From 136297022940b814a950777ca30dcc241e75f065 Mon Sep 17 00:00:00 2001 From: chengcheng Date: Tue, 21 Jul 2026 00:29:31 +0800 Subject: [PATCH] =?UTF-8?q?test(billing):=20=E8=A6=86=E7=9B=96=E4=BC=B0?= =?UTF-8?q?=E4=BB=B7=E5=8F=AA=E8=AF=BB=E4=B8=8E=E8=AF=B7=E6=B1=82=E9=87=8D?= =?UTF-8?q?=E6=94=BE=E5=A5=91=E7=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PostgreSQL HTTP 集成测试验证估价新增字段且不创建任务、冻结余额或写流水，并验证非流式重放只执行一次、异请求冲突及流式重复冲突。 --- .../httpapi/core_flow_integration_test.go | 98 +++++++++++++++++++ 1 file changed, 98 insertions(+) diff --git a/apps/api/internal/httpapi/core_flow_integration_test.go b/apps/api/internal/httpapi/core_flow_integration_test.go index a400451..351db89 100644 --- a/apps/api/internal/httpapi/core_flow_integration_test.go +++ b/apps/api/internal/httpapi/core_flow_integration_test.go @@ -48,6 +48,7 @@ func TestCoreLocalFlow(t *testing.T) { DatabaseURL: databaseURL, IdentityMode: "hybrid", JWTSecret: "test-secret", + BillingEngineMode: "enforce", TaskProgressCallbackEnabled: true, TaskProgressCallbackURL: "http://callback.local/task-progress", CORSAllowedOrigin: "*", @@ -814,6 +815,7 @@ LIMIT 1`).Scan(&gptImageModelTypesRaw); err != nil { "currency": "resource", "rules": []map[string]any{ {"ruleKey": "text_input", "displayName": "Text Input", "resourceType": "text_input", "unit": "1k_tokens", "basePrice": 1}, + {"ruleKey": "text_cached_input", "displayName": "Text Cached Input", "resourceType": "text_cached_input", "unit": "1k_tokens", "basePrice": 0.5}, {"ruleKey": "text_output", "displayName": "Text Output", "resourceType": "text_output", "unit": "1k_tokens", "basePrice": 2}, {"ruleKey": "image", "displayName": "Image", "resourceType": "image", "unit": "image", "basePrice": 7}, {"ruleKey": "image_edit", "displayName": "Image Edit", "resourceType": "image_edit", "unit": "image", "basePrice": 11}, @@ -845,6 +847,101 @@ WHERE gateway_user_id = $1::uuid AND currency = 'resource'`, smokeGatewayUserID).Scan(&walletBalanceBefore); err != nil { t.Fatalf("read wallet balance before pricing task: %v", err) } + var estimateSideEffectsBefore struct { + Tasks int + Transactions int + Frozen string + } + if err := testPool.QueryRow(ctx, ` +SELECT (SELECT count(*) FROM gateway_tasks), + (SELECT count(*) FROM gateway_wallet_transactions WHERE gateway_user_id=$1::uuid), + (SELECT frozen_balance::text FROM gateway_wallet_accounts WHERE gateway_user_id=$1::uuid AND currency='resource')`, + smokeGatewayUserID, + ).Scan(&estimateSideEffectsBefore.Tasks, &estimateSideEffectsBefore.Transactions, &estimateSideEffectsBefore.Frozen); err != nil { + t.Fatalf("read estimate side effects before request: %v", err) + } + var pricingEstimate struct { + TotalAmount float64 `json:"totalAmount"` + ReservationAmount float64 `json:"reservationAmount"` + CandidateCount int `json:"candidateCount"` + PricingVersion string `json:"pricingVersion"` + RequestFingerprint string `json:"requestFingerprint"` + } + doJSON(t, server.URL, http.MethodPost, "/api/v1/pricing/estimate", apiKeyResponse.Secret, map[string]any{ + "kind": "chat.completions", "model": pricingModel, "max_completion_tokens": 32, + "messages": []map[string]any{{"role": "user", "content": "estimate only"}}, + }, http.StatusOK, &pricingEstimate) + if pricingEstimate.TotalAmount <= 0 || pricingEstimate.ReservationAmount < pricingEstimate.TotalAmount || + pricingEstimate.CandidateCount != 1 || pricingEstimate.PricingVersion != "effective-pricing-v2" || pricingEstimate.RequestFingerprint == "" { + t.Fatalf("unexpected effective pricing estimate: %+v", pricingEstimate) + } + var estimateSideEffectsAfter struct { + Tasks int + Transactions int + Frozen string + } + if err := testPool.QueryRow(ctx, ` +SELECT (SELECT count(*) FROM gateway_tasks), + (SELECT count(*) FROM gateway_wallet_transactions WHERE gateway_user_id=$1::uuid), + (SELECT frozen_balance::text FROM gateway_wallet_accounts WHERE gateway_user_id=$1::uuid AND currency='resource')`, + smokeGatewayUserID, + ).Scan(&estimateSideEffectsAfter.Tasks, &estimateSideEffectsAfter.Transactions, &estimateSideEffectsAfter.Frozen); err != nil { + t.Fatalf("read estimate side effects after request: %v", err) + } + if estimateSideEffectsAfter != estimateSideEffectsBefore { + t.Fatalf("pricing estimate must be read-only, before=%+v after=%+v", estimateSideEffectsBefore, estimateSideEffectsAfter) + } + + idempotencyPayload := map[string]any{ + "model": pricingModel, "runMode": "simulation", "simulation": true, "simulationDurationMs": 5, + "messages": []map[string]any{{"role": "user", "content": "idempotent replay"}}, + } + idempotencyHeaders := map[string]string{"Idempotency-Key": "chat-replay-" + suffixText} + var idempotentFirst map[string]any + firstReplayHeaders := doJSONWithHeaders(t, server.URL, http.MethodPost, "/api/v1/chat/completions", apiKeyResponse.Secret, idempotencyPayload, idempotencyHeaders, http.StatusOK, &idempotentFirst) + var idempotentSecond map[string]any + secondReplayHeaders := doJSONWithHeaders(t, server.URL, http.MethodPost, "/api/v1/chat/completions", apiKeyResponse.Secret, idempotencyPayload, idempotencyHeaders, http.StatusOK, &idempotentSecond) + idempotentTaskID := firstReplayHeaders.Get("X-Gateway-Task-Id") + if idempotentTaskID == "" || secondReplayHeaders.Get("X-Gateway-Task-Id") != idempotentTaskID || secondReplayHeaders.Get("Idempotent-Replayed") != "true" { + t.Fatalf("non-stream replay headers first=%v second=%v", firstReplayHeaders, secondReplayHeaders) + } + var idempotentAttempts int + if err := testPool.QueryRow(ctx, `SELECT count(*) FROM gateway_task_attempts WHERE task_id=$1::uuid`, idempotentTaskID).Scan(&idempotentAttempts); err != nil { + t.Fatalf("count idempotent task attempts: %v", err) + } + if idempotentAttempts != 1 { + t.Fatalf("idempotent request called upstream %d times, want 1", idempotentAttempts) + } + idempotencyPayload["messages"] = []map[string]any{{"role": "user", "content": "different request"}} + doJSONWithHeaders(t, server.URL, http.MethodPost, "/api/v1/chat/completions", apiKeyResponse.Secret, idempotencyPayload, idempotencyHeaders, http.StatusConflict, nil) + + streamPayload := map[string]any{ + "model": pricingModel, "runMode": "simulation", "simulation": true, "simulationDurationMs": 5, "stream": true, + "messages": []map[string]any{{"role": "user", "content": "stream replay"}}, + } + streamRaw, err := json.Marshal(streamPayload) + if err != nil { + t.Fatal(err) + } + streamKey := "stream-replay-" + suffixText + streamRequest, err := http.NewRequest(http.MethodPost, server.URL+"/api/v1/chat/completions", bytes.NewReader(streamRaw)) + if err != nil { + t.Fatal(err) + } + streamRequest.Header.Set("Authorization", "Bearer "+apiKeyResponse.Secret) + streamRequest.Header.Set("Content-Type", "application/json") + streamRequest.Header.Set("Idempotency-Key", streamKey) + streamResponse, err := http.DefaultClient.Do(streamRequest) + if err != nil { + t.Fatalf("execute first idempotent stream: %v", err) + } + _, _ = io.ReadAll(streamResponse.Body) + _ = streamResponse.Body.Close() + if streamResponse.StatusCode != http.StatusOK || streamResponse.Header.Get("X-Gateway-Task-Id") == "" { + t.Fatalf("first idempotent stream status=%d headers=%v", streamResponse.StatusCode, streamResponse.Header) + } + doJSONWithHeaders(t, server.URL, http.MethodPost, "/api/v1/chat/completions", apiKeyResponse.Secret, streamPayload, + map[string]string{"Idempotency-Key": streamKey}, http.StatusConflict, nil) var pricingTask struct { Task struct { ID string `json:"id"` @@ -1588,6 +1685,7 @@ WHERE m.platform_id = $1::uuid DatabaseURL: databaseURL, IdentityMode: "hybrid", JWTSecret: "test-secret", + BillingEngineMode: "enforce", TaskProgressCallbackEnabled: true, TaskProgressCallbackURL: "http://callback.local/task-progress", CORSAllowedOrigin: "*",