ModelRouter/e2e/channels_flows_e2e_test.go

package e2e

import (
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"sync/atomic"
	"testing"
	"time"
)

// TestChannelsFlowsE2E 用真二进制补齐另外三条流程的端到端：
//
//	Cline（WorkOS device-poll 两步：authenticate → register 换最终凭据）
//	paste（opencode：直接 complete_login 贴 API key）
//	device-poll 授权页（qoder：本地造 PKCE，把手动授权页给用户，再凭据完成）
//
// 终局证据与 TestChannelsDevicePollE2E 同一标准：/v1/models 出现该平台模型。
func TestChannelsFlowsE2E(t *testing.T) {
	var authHits, regHits int32
	plat := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		switch r.URL.Path {
		// ---- Cline（WorkOS）----
		case "/user_management/authorize/device":
			fmt.Fprint(w, `{"device_code":"CD-E2E","user_code":"CLNE-1234","verification_uri":"https://example/activate","interval":1,"expires_in":600}`)
		case "/user_management/authenticate":
			if atomic.AddInt32(&authHits, 1) == 1 {
				w.WriteHeader(400)
				fmt.Fprint(w, `{"error":"authorization_pending"}`)
				return
			}
			fmt.Fprint(w, `{"access_token":"WO-E2E-AT","refresh_token":"WO-E2E-RT"}`)
		case "/api/v1/auth/register":
			atomic.AddInt32(&regHits, 1)
			fmt.Fprint(w, `{"success":true,"data":{"accessToken":"CLINE-E2E-AT","refreshToken":"CLINE-E2E-RT","userInfo":{"email":"dev@example.com"}}}`)
		// ---- 上游推理面（opencode/qoder/cline 的 base_url 都指到这里）----
		case "/v1/models":
			fmt.Fprint(w, `{"object":"list","data":[{"id":"big-pickle","object":"model"},{"id":"qmodel_38max","object":"model"},{"id":"stealth/space-bunny-alpha","object":"model"}]}`)
		case "/v1/chat/completions":
			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()

	dir := t.TempDir()
	addr := mustAddr(t)
	pluginDir := filepath.Join(dir, "plugins")
	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
	}

	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)
	}

	// ---------- ① Cline：WorkOS 设备码两步 ----------
	st, m = call("POST", "/api/plugins/channels/call/set_endpoints", fmt.Sprintf(
		`{"platform":"cline","base_url":%q,"device":{"code_url":%q,"token_url":%q},"register":{"url":%q}}`,
		plat.URL, plat.URL+"/user_management/authorize/device",
		plat.URL+"/user_management/authenticate", plat.URL+"/api/v1/auth/register"))
	if st != 200 {
		t.Fatalf("set_endpoints(cline) 失败: status=%d body=%v", st, m)
	}
	st, m = call("POST", "/api/plugins/http/channels/login", `{"platform":"cline"}`)
	if st != 200 {
		t.Fatalf("cline start_login status=%d body=%v", st, m)
	}
	sess, _ := m["session"].(map[string]interface{})
	if sess == nil || sess["user_code"] != "CLNE-1234" {
		t.Fatalf("cline start_login 未透传 user_code: %v", m)
	}
	sid, _ := sess["id"].(string)

	// 第一拍 pending（authorization_pending），第二拍 completed
	atomic.StoreInt32(&authHits, 0)
	st, m = call("GET", "/api/plugins/http/channels/login/status?id="+sid, "")
	if m["state"] != "pending" {
		t.Fatalf("cline 轮询1 应 pending, 实得 %v", m)
	}
	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("cline 轮询2 应 completed, 实得 %v", m)
	}
	if m["account"] != "dev@example.com" {
		t.Fatalf("cline account 应取 register userInfo.email, 实得 %v", m)
	}
	if atomic.LoadInt32(&regHits) != 1 {
		t.Fatalf("register 应恰好调用一次, 实得 %d", regHits)
	}
	// 凭据必须是 register 的最终 token，不是 WorkOS token
	var modelsBody string
	deadline = time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		resp, raw := g.do("GET", "/v1/models", "", true)
		modelsBody = raw
		if resp != nil && resp.StatusCode == 200 && strings.Contains(raw, "stealth/space-bunny-alpha") {
			break
		}
		time.Sleep(150 * time.Millisecond)
	}
	if !strings.Contains(modelsBody, "stealth/space-bunny-alpha") {
		t.Fatalf("cline 接入后 /v1/models 未出现其模型: %s", modelsBody)
	}

	// ---------- ② paste（opencode）：complete 直接贴 key ----------
	st, m = call("POST", "/api/plugins/channels/call/set_endpoints", fmt.Sprintf(
		`{"platform":"opencode","base_url":%q}`, plat.URL))
	if st != 200 {
		t.Fatalf("set_endpoints(opencode) 失败: %v", m)
	}
	st, m = call("POST", "/api/plugins/http/channels/login", `{"platform":"opencode"}`)
	if st != 200 {
		t.Fatalf("opencode start_login status=%d", st)
	}
	if sess, _ = m["session"].(map[string]interface{}); sess == nil || sess["flow"] != "paste" {
		t.Fatalf("opencode 会话 flow 应 paste: %v", m)
	}
	st, m = call("POST", "/api/plugins/http/channels/login/complete",
		`{"platform":"opencode","credential":"OC-E2E-KEY"}`)
	if st != 200 || m["ok"] != true {
		t.Fatalf("opencode complete 失败: status=%d body=%v", st, m)
	}
	deadline = time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		resp, raw := g.do("GET", "/v1/models", "", true)
		modelsBody = raw
		if resp != nil && resp.StatusCode == 200 && strings.Contains(raw, "big-pickle") {
			break
		}
		time.Sleep(150 * time.Millisecond)
	}
	if !strings.Contains(modelsBody, "big-pickle") {
		t.Fatalf("opencode 接入后 /v1/models 未出现 big-pickle: %s", modelsBody)
	}

	// ---------- ③ device-poll（qoder）：给授权页 + 凭据完成 ----------
	// qoder 曾走 manual（返回 manual_hint）；现已改为真实的 PKCE 授权页流程
	// （本地造 challenge/verifier，服务端按 nonce 找回待授权会话）。
	st, m = call("POST", "/api/plugins/channels/call/set_endpoints", fmt.Sprintf(
		`{"platform":"qoder","base_url":%q}`, plat.URL))
	if st != 200 {
		t.Fatalf("set_endpoints(qoder) 失败: %v", m)
	}
	st, m = call("POST", "/api/plugins/http/channels/login", `{"platform":"qoder"}`)
	if st != 200 {
		t.Fatalf("qoder start_login status=%d", st)
	}
	// 真流程必须给出授权页 URL（而不是「不支持自动登录」的 manual_hint），
	// 且会话声明为 device-poll。
	if hint, _ := m["manual_hint"].(string); hint != "" {
		t.Fatalf("qoder 已不再走 manual，不应返回 manual_hint: %q", hint)
	}
	if sess, _ = m["session"].(map[string]interface{}); sess == nil || sess["flow"] != "device-poll" {
		t.Fatalf("qoder 会话 flow 应 device-poll: %v", m)
	}
	if authURL, _ := m["url"].(string); !strings.Contains(authURL, "challenge=") {
		t.Fatalf("qoder 授权页 URL 应带 PKCE challenge，实得 %q", authURL)
	}
	st, m = call("POST", "/api/plugins/http/channels/login/complete",
		`{"platform":"qoder","credential":"QD-E2E-KEY"}`)
	if st != 200 || m["ok"] != true {
		t.Fatalf("qoder complete 失败: status=%d body=%v", st, m)
	}
	deadline = time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		resp, raw := g.do("GET", "/v1/models", "", true)
		modelsBody = raw
		if resp != nil && resp.StatusCode == 200 && strings.Contains(raw, "qmodel_38max") {
			break
		}
		time.Sleep(150 * time.Millisecond)
	}
	if !strings.Contains(modelsBody, "qmodel_38max") {
		t.Fatalf("qoder 接入后 /v1/models 未出现 qmodel_38max: %s", modelsBody)
	}
}