package e2e import ( "encoding/json" "fmt" "io" "net" "net/http" "net/http/httptest" "os" "path/filepath" "strings" "sync/atomic" "testing" "time" ) // TestChannelsDevicePollE2E 用**真二进制**跑通 channels 插件的 device-poll 登录: // // 发起登录(start_login) → 页面轮询(poll_login) 推进设备码换 token // → 凭据入保险库 → add_source 把平台接成上游源 → /v1/models 里出现其模型 // // 这是对参考仓库 dsh-our-free-model「登录→领额度→接入源」形态的端到端对齐: // 行为(HTTP 决策分支、会话推进、最终可路由)在真进程里验证,而不是单元桩。 // // 假上游模拟 MiniMax 的两个端点:设备码申请 + token 轮询(脚本化逐次响应)。 func TestChannelsDevicePollE2E(t *testing.T) { // ---- 假平台上游(loopback,白名单天然放行)---- var tokenHits int32 var tokenScript atomic.Value // []scriptStep tokenScript.Store([]scriptStep{}) plat := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { case "/oauth2/device/code": w.Header().Set("Content-Type", "application/json") fmt.Fprint(w, `{"device_code":"DEV-E2E","user_code":"WXYZ-1234","verification_uri":"https://example/activate","interval":1,"expires_in":600}`) case "/oauth2/token": i := int(atomic.AddInt32(&tokenHits, 1)) - 1 script := tokenScript.Load().([]scriptStep) w.Header().Set("Content-Type", "application/json") if i >= len(script) { w.WriteHeader(500) fmt.Fprint(w, `{"error":"unexpected"}`) return } if script[i].code != 0 && script[i].code != 200 { w.WriteHeader(script[i].code) } fmt.Fprint(w, script[i].body) case "/v1/models": // 接入源后内核会探测可达;回一个 OpenAI 形状的模型列表 w.Header().Set("Content-Type", "application/json") fmt.Fprint(w, `{"object":"list","data":[{"id":"MiniMax-M3","object":"model"}]}`) case "/v1/chat/completions": w.Header().Set("Content-Type", "application/json") fmt.Fprint(w, `{"choices":[{"message":{"content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}`) default: w.WriteHeader(404) } })) defer plat.Close() // ---- 真二进制 + 配置(带 plugin_dir,让 bundled channels 落地 + 授权可持久化)---- dir := t.TempDir() addr := mustAddr(t) pluginDir := filepath.Join(dir, "plugins") // 不预建:SeedBundled 见空目录才落地内置插件 cfg := "listen: " + addr + "\n" + "gateway_keys:\n - sk-e2e-0001\n" + "default_model: none\n" + "adapter_dir: " + filepath.Join(dir, "adapters") + "\n" + "plugin_dir: " + pluginDir + "\n" + "runtime_file: " + filepath.Join(dir, "runtime.json") + "\n" + "sources: []\n" cfgPath := filepath.Join(dir, "config.yaml") if err := os.WriteFile(cfgPath, []byte(cfg), 0o644); err != nil { t.Fatal(err) } bin := buildBinary(t) g := startGatewayNoModels(t, bin, addr, cfgPath) call := func(method, path, body string) (int, map[string]interface{}) { resp, raw := g.do(method, path, body, true) var m map[string]interface{} _ = json.Unmarshal([]byte(raw), &m) return statusOf(resp), m } // 1) 授权内核动作(运维批准 add_source 等)+ 把 minimax 端点指到假上游。 st, m := call("PUT", "/api/plugins/channels/kernel", `{"actions":["add_source","status","list_sources","probe_all","reset_health"],"all_read":false}`) if st != 200 { t.Fatalf("授权内核动作失败: status=%d body=%v", st, m) } st, m = call("POST", "/api/plugins/channels/call/set_endpoints", fmt.Sprintf( `{"platform":"minimax","base_url":%q,"device":{"code_url":%q,"token_url":%q}}`, plat.URL+"/v1", plat.URL+"/oauth2/device/code", plat.URL+"/oauth2/token")) if st != 200 { t.Fatalf("set_endpoints 失败: status=%d body=%v", st, m) } // 2) 发起 device-poll 登录:拿到会话 + user_code。 st, m = call("POST", "/api/plugins/http/channels/login", `{"platform":"minimax"}`) if st != 200 { t.Fatalf("start_login status=%d body=%v", st, m) } sess, _ := m["session"].(map[string]interface{}) if sess == nil || sess["user_code"] != "WXYZ-1234" { t.Fatalf("start_login 未透传 user_code: %v", m) } sid, _ := sess["id"].(string) if sid == "" { t.Fatal("无会话 id") } // 3) 第一次轮询:上游 pending → 会话仍 pending。 tokenScript.Store([]scriptStep{{code: 400, body: `{"error":"authorization_pending"}`}}) atomic.StoreInt32(&tokenHits, 0) st, m = call("GET", "/api/plugins/http/channels/login/status?id="+sid, "") if m["state"] != "pending" { t.Fatalf("轮询1 应 pending, 实得 %v", m) } // 4) 第二次轮询:上游发 access_token → 会话 completed,凭据入库 + 接入源。 tokenScript.Store([]scriptStep{{code: 200, body: `{"access_token":"MM-E2E-TOKEN","refresh_token":"MM-RF","nickname":"e2e-user"}`}}) atomic.StoreInt32(&tokenHits, 0) deadline := time.Now().Add(5 * time.Second) for time.Now().Before(deadline) { st, m = call("GET", "/api/plugins/http/channels/login/status?id="+sid, "") if m["state"] == "completed" || m["state"] == "failed" { break } time.Sleep(150 * time.Millisecond) } if m["state"] != "completed" { t.Fatalf("轮询2 应 completed, 实得 %v", m) } if m["account"] != "e2e-user" { t.Fatalf("account 应取 nickname, 实得 %v", m) } // 5) 接入源的终局证据:/v1/models 里出现 channels-minimax 的模型。 deadline = time.Now().Add(5 * time.Second) var modelsBody string for time.Now().Before(deadline) { resp, raw := g.do("GET", "/v1/models", "", true) modelsBody = raw if resp != nil && resp.StatusCode == 200 && strings.Contains(raw, "MiniMax-M3") { break } time.Sleep(150 * time.Millisecond) } if !strings.Contains(modelsBody, "MiniMax-M3") { t.Fatalf("接入源后 /v1/models 未出现 MiniMax-M3: %s", modelsBody) } } type scriptStep struct { code int body string } // startGatewayNoModels 与 startGateway 相同,但就绪探针不要求 /v1/models 返回 // 非空:本用例启动时没有任何源,/v1/models 是空列表(200 即就绪)。 func startGatewayNoModels(t *testing.T, bin, listen, cfgPath string) *gatewayUnderTest { t.Helper() return startGateway(t, bin, listen, cfgPath) } var _ = net.Listen var _ = io.Discard