ModelRouter/e2e/channels_device_e2e_test.go

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")
	if err := os.MkdirAll(pluginDir, 0o755); err != nil {
		t.Fatal(err)
	}
	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, _ := 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", st)
	}
	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/channels/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/channels/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/channels/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