easyai
7c5a999e32
将通用生成、Gemini、可灵、火山、健康检查与 OpenAPI 的推荐入口统一到 /api/v1，并保留历史路径作为兼容别名。同步更新代理配置、接入文档、接口清单和前缀回归测试。\n\n验证：go vet ./...；go test ./...；pnpm openapi；pnpm lint；pnpm test；pnpm build；公开 OpenAPI 71 个方法与接口清单机器比对一致。
294 lines
11 KiB
Go
294 lines
11 KiB
Go
package httpapi
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"log/slog"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/easyai/easyai-ai-gateway/apps/api/internal/config"
|
|
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
|
|
)
|
|
|
|
func TestKelingOmniCompatibleHTTPFlow(t *testing.T) {
|
|
databaseURL := strings.TrimSpace(os.Getenv("AI_GATEWAY_TEST_DATABASE_URL"))
|
|
if databaseURL == "" {
|
|
t.Skip("set AI_GATEWAY_TEST_DATABASE_URL to run the Kling-compatible HTTP integration flow")
|
|
}
|
|
|
|
var upstreamTaskSequence atomic.Int64
|
|
var upstreamPayloadMu sync.Mutex
|
|
var upstreamPayloads []map[string]any
|
|
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if r.Header.Get("Authorization") != "Bearer upstream-keling-key" {
|
|
t.Fatalf("unexpected upstream Authorization: %q", r.Header.Get("Authorization"))
|
|
}
|
|
switch {
|
|
case r.Method == http.MethodPost && r.URL.Path == "/videos/omni-video":
|
|
var payload map[string]any
|
|
if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
|
|
t.Fatalf("decode upstream request: %v", err)
|
|
}
|
|
upstreamPayloadMu.Lock()
|
|
upstreamPayloads = append(upstreamPayloads, payload)
|
|
upstreamPayloadMu.Unlock()
|
|
id := "upstream-" + strconv.FormatInt(upstreamTaskSequence.Add(1), 10)
|
|
_ = json.NewEncoder(w).Encode(map[string]any{
|
|
"code": 0,
|
|
"request_id": "submit-" + id,
|
|
"data": map[string]any{"task_id": id, "task_status": "submitted"},
|
|
})
|
|
case r.Method == http.MethodGet && strings.HasPrefix(r.URL.Path, "/videos/omni-video/upstream-"):
|
|
id := strings.TrimPrefix(r.URL.Path, "/videos/omni-video/")
|
|
_ = json.NewEncoder(w).Encode(map[string]any{
|
|
"code": 0,
|
|
"request_id": "poll-" + id,
|
|
"data": map[string]any{
|
|
"task_id": id,
|
|
"task_status": "succeed",
|
|
"created_at": time.Now().UnixMilli(),
|
|
"task_result": map[string]any{"videos": []any{map[string]any{
|
|
"id": "video-" + id,
|
|
"url": "https://example.com/" + id + ".mp4",
|
|
"watermark_url": "https://example.com/" + id + "-watermark.mp4",
|
|
"duration": "3",
|
|
}}},
|
|
},
|
|
})
|
|
default:
|
|
t.Fatalf("unexpected upstream request %s %s", r.Method, r.URL.Path)
|
|
}
|
|
}))
|
|
defer upstream.Close()
|
|
|
|
ctx := context.Background()
|
|
applyMigration(t, ctx, databaseURL)
|
|
db, err := store.Connect(ctx, databaseURL)
|
|
if err != nil {
|
|
t.Fatalf("connect store: %v", err)
|
|
}
|
|
defer db.Close()
|
|
|
|
suffix := strconv.FormatInt(time.Now().UnixNano(), 10)
|
|
platform, err := db.CreatePlatform(ctx, store.CreatePlatformInput{
|
|
Provider: "keling",
|
|
PlatformKey: "keling-compatible-test-" + suffix,
|
|
Name: "Kling Compatible Test",
|
|
BaseURL: upstream.URL,
|
|
AuthType: "APIKey",
|
|
Credentials: map[string]any{"apiKey": "upstream-keling-key"},
|
|
Config: map[string]any{
|
|
"kelingPollIntervalMs": 100,
|
|
"kelingPollTimeoutSeconds": 5,
|
|
},
|
|
Priority: 1,
|
|
Status: "enabled",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("create test platform: %v", err)
|
|
}
|
|
_, err = db.CreatePlatformModel(ctx, store.CreatePlatformModelInput{
|
|
PlatformID: platform.ID,
|
|
CanonicalModelKey: "keling:kling-video-o1",
|
|
ModelName: "kling-o1",
|
|
ProviderModelName: "kling-video-o1",
|
|
ModelAlias: "kling-o1",
|
|
ModelType: store.StringList{"omni_video", "video_generate"},
|
|
DisplayName: "Kling O1 Compatible Test",
|
|
Capabilities: map[string]any{
|
|
"omni_video": map[string]any{
|
|
"supported_modes": []any{"text_to_video", "image_reference"},
|
|
"output_resolutions": []any{"720p", "1080p"},
|
|
"aspect_ratio_allowed": []any{"16:9", "9:16", "1:1"},
|
|
"duration_options": []any{3, 4, 5, 6, 7, 8, 9, 10},
|
|
"output_audio": false,
|
|
"max_images": 7,
|
|
},
|
|
"video_generate": map[string]any{
|
|
"supported_modes": []any{"text_to_video"},
|
|
"output_resolutions": []any{"720p", "1080p"},
|
|
"aspect_ratio_allowed": []any{"16:9", "9:16", "1:1"},
|
|
"duration_options": []any{3, 4, 5, 6, 7, 8, 9, 10},
|
|
"output_audio": true,
|
|
},
|
|
},
|
|
Enabled: true,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("create test platform model: %v", err)
|
|
}
|
|
|
|
serverCtx, cancelServer := context.WithCancel(ctx)
|
|
defer cancelServer()
|
|
gateway := httptest.NewServer(NewServerWithContext(serverCtx, config.Config{
|
|
AppEnv: "test",
|
|
HTTPAddr: ":0",
|
|
DatabaseURL: databaseURL,
|
|
IdentityMode: "hybrid",
|
|
JWTSecret: "test-secret",
|
|
CORSAllowedOrigin: "*",
|
|
}, db, slog.New(slog.NewTextHandler(io.Discard, nil))))
|
|
defer gateway.Close()
|
|
|
|
firstUserToken, firstAPIKey := createKelingCompatIntegrationUser(t, ctx, db, gateway.URL, "first", suffix, true)
|
|
_ = firstUserToken
|
|
_, secondAPIKey := createKelingCompatIntegrationUser(t, ctx, db, gateway.URL, "second", suffix, false)
|
|
|
|
var created KelingCompatibleEnvelope
|
|
doJSON(t, gateway.URL, http.MethodPost, "/api/v1/videos/omni-video", firstAPIKey, map[string]any{
|
|
"model_name": "kling-video-o1",
|
|
"prompt": "A clean product reveal",
|
|
"mode": "std",
|
|
"aspect_ratio": "16:9",
|
|
"duration": "3",
|
|
"sound": "off",
|
|
"image_list": []any{map[string]any{"image_url": "https://example.com/reference.png"}},
|
|
"external_task_id": "compat-http-1",
|
|
}, http.StatusOK, &created)
|
|
createdData, _ := created.Data.(map[string]any)
|
|
if created.Code != 0 || created.RequestID == "" || strings.TrimSpace(stringFromKelingCompat(createdData["task_id"])) == "" || createdData["task_status"] != "submitted" {
|
|
t.Fatalf("unexpected compatible create response: %+v", created)
|
|
}
|
|
taskID := stringFromKelingCompat(createdData["task_id"])
|
|
|
|
var hidden KelingCompatibleEnvelope
|
|
doJSON(t, gateway.URL, http.MethodGet, "/api/v1/videos/omni-video/"+taskID, secondAPIKey, nil, http.StatusNotFound, &hidden)
|
|
if hidden.Code != 1203 {
|
|
t.Fatalf("cross-user task must be hidden: %+v", hidden)
|
|
}
|
|
|
|
completed := waitForKelingCompatTask(t, gateway.URL, firstAPIKey, taskID, 5*time.Second)
|
|
if completed.Code != 0 {
|
|
t.Fatalf("compatible task failed: %+v", completed)
|
|
}
|
|
completedData, _ := completed.Data.(map[string]any)
|
|
if completedData["task_status"] != "succeed" {
|
|
t.Fatalf("compatible task did not succeed: %+v", completedData)
|
|
}
|
|
taskResult, _ := completedData["task_result"].(map[string]any)
|
|
videos, _ := taskResult["videos"].([]any)
|
|
video, _ := videos[0].(map[string]any)
|
|
if video["id"] == "" || video["watermark_url"] == "" || video["duration"] != "3" {
|
|
t.Fatalf("compatible result lost video metadata: %+v", video)
|
|
}
|
|
|
|
var standard struct {
|
|
TaskID string `json:"taskId"`
|
|
}
|
|
doJSONWithHeaders(t, gateway.URL, http.MethodPost, "/api/v1/videos/generations", firstAPIKey, map[string]any{
|
|
"model": "kling-o1",
|
|
"prompt": "A second product reveal",
|
|
"resolution": "720p",
|
|
"aspect_ratio": "16:9",
|
|
"duration": 3,
|
|
"audio": false,
|
|
}, map[string]string{"X-Async": "true"}, http.StatusAccepted, &standard)
|
|
if standard.TaskID == "" {
|
|
t.Fatal("standard video generation did not return taskId")
|
|
}
|
|
waitForTaskStatus(t, gateway.URL, firstAPIKey, standard.TaskID, []string{"succeeded"}, 5*time.Second)
|
|
|
|
var unsupportedAudio struct {
|
|
TaskID string `json:"taskId"`
|
|
}
|
|
doJSONWithHeaders(t, gateway.URL, http.MethodPost, "/api/v1/videos/generations", firstAPIKey, map[string]any{
|
|
"model": "kling-o1",
|
|
"prompt": "An O1 request that must not silently ignore audio",
|
|
"resolution": "1080p",
|
|
"aspect_ratio": "9:16",
|
|
"duration": 5,
|
|
"audio": true,
|
|
}, map[string]string{"X-Async": "true"}, http.StatusAccepted, &unsupportedAudio)
|
|
if unsupportedAudio.TaskID == "" {
|
|
t.Fatal("unsupported O1 audio request did not return taskId")
|
|
}
|
|
waitForTaskStatus(t, gateway.URL, firstAPIKey, unsupportedAudio.TaskID, []string{"failed"}, 5*time.Second)
|
|
var failedAudioTask store.GatewayTask
|
|
doJSON(t, gateway.URL, http.MethodGet, "/api/v1/tasks/"+unsupportedAudio.TaskID, firstAPIKey, nil, http.StatusOK, &failedAudioTask)
|
|
if failedAudioTask.ErrorCode != "invalid_parameter" || !strings.Contains(failedAudioTask.ErrorMessage, "does not support generated audio") {
|
|
t.Fatalf("O1 audio request must fail visibly before upstream submission: %+v", failedAudioTask)
|
|
}
|
|
|
|
upstreamPayloadMu.Lock()
|
|
defer upstreamPayloadMu.Unlock()
|
|
if len(upstreamPayloads) != 2 {
|
|
t.Fatalf("expected two upstream submissions, got %d", len(upstreamPayloads))
|
|
}
|
|
compatiblePayload := upstreamPayloads[0]
|
|
if compatiblePayload["model_name"] != "kling-video-o1" || compatiblePayload["mode"] != "std" || compatiblePayload["sound"] != "off" || compatiblePayload["duration"] != "3" || compatiblePayload["aspect_ratio"] != "16:9" || compatiblePayload["external_task_id"] != "compat-http-1" {
|
|
t.Fatalf("unexpected compatible upstream payload: %+v", compatiblePayload)
|
|
}
|
|
}
|
|
|
|
func createKelingCompatIntegrationUser(t *testing.T, ctx context.Context, db *store.Store, baseURL string, prefix string, suffix string, fund bool) (string, string) {
|
|
t.Helper()
|
|
username := fmt.Sprintf("kling_compat_%s_%s", prefix, suffix)
|
|
password := "password123"
|
|
var registered struct {
|
|
AccessToken string `json:"accessToken"`
|
|
}
|
|
doJSON(t, baseURL, http.MethodPost, "/api/v1/auth/register", "", map[string]any{
|
|
"username": username,
|
|
"email": username + "@example.com",
|
|
"password": password,
|
|
}, http.StatusCreated, ®istered)
|
|
var apiKey struct {
|
|
Secret string `json:"secret"`
|
|
}
|
|
doJSON(t, baseURL, http.MethodPost, "/api/v1/api-keys", registered.AccessToken, map[string]any{
|
|
"name": "Kling compatible integration key",
|
|
"scopes": []string{"video"},
|
|
}, http.StatusCreated, &apiKey)
|
|
if fund {
|
|
if _, err := db.Pool().Exec(ctx, `UPDATE gateway_users SET roles = '["admin"]'::jsonb WHERE username = $1`, username); err != nil {
|
|
t.Fatalf("promote integration user: %v", err)
|
|
}
|
|
var loggedIn struct {
|
|
AccessToken string `json:"accessToken"`
|
|
}
|
|
doJSON(t, baseURL, http.MethodPost, "/api/v1/auth/login", "", map[string]any{
|
|
"account": username,
|
|
"password": password,
|
|
}, http.StatusOK, &loggedIn)
|
|
var gatewayUserID string
|
|
if err := db.Pool().QueryRow(ctx, `SELECT id::text FROM gateway_users WHERE username = $1`, username).Scan(&gatewayUserID); err != nil {
|
|
t.Fatalf("read integration user id: %v", err)
|
|
}
|
|
doJSON(t, baseURL, http.MethodPatch, "/api/admin/users/"+gatewayUserID+"/wallet", loggedIn.AccessToken, map[string]any{
|
|
"currency": "resource",
|
|
"balance": 1000,
|
|
"reason": "seed Kling compatible integration wallet",
|
|
}, http.StatusOK, nil)
|
|
}
|
|
return registered.AccessToken, apiKey.Secret
|
|
}
|
|
|
|
func waitForKelingCompatTask(t *testing.T, baseURL string, apiKey string, taskID string, timeout time.Duration) KelingCompatibleEnvelope {
|
|
t.Helper()
|
|
deadline := time.Now().Add(timeout)
|
|
for time.Now().Before(deadline) {
|
|
var response KelingCompatibleEnvelope
|
|
doJSON(t, baseURL, http.MethodGet, "/api/v1/videos/omni-video/"+taskID, apiKey, nil, http.StatusOK, &response)
|
|
data, _ := response.Data.(map[string]any)
|
|
switch data["task_status"] {
|
|
case "succeed":
|
|
return response
|
|
case "failed":
|
|
t.Fatalf("Kling-compatible task failed: %+v", response)
|
|
}
|
|
time.Sleep(50 * time.Millisecond)
|
|
}
|
|
t.Fatalf("timed out waiting for Kling-compatible task %s", taskID)
|
|
return KelingCompatibleEnvelope{}
|
|
}
|