ModelRouter/e2e/adapter_takeover_test.go

package e2e

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

// 本文件端到端验证 R2/R5 的核心主张：**对外服务的 wire 格式由适配器决定，
// 内核只调度**；适配器不声明时内核的 OpenAI 兜底仍然可用。
//
// 与其他测试的分工：
//   - internal/gateway/messages_test.go 在**进程内**验证同一件事（起 core+gateway，
//     不打真二进制）。它跑得快，但绕过了一些只在真实进程里发生的事：Lua 适配器的
//     磁盘 seed、配置加载、systemd 式启动路径、真 HTTP 栈。
//   - 本文件打**真编译出的二进制**，用一个只说 Anthropic Messages 的假上游，
//     证明整条链路（真进程 → 真 Lua 适配器 → 真上游 → 适配器序列化 → 客户端）
//     确实成立，而不是只在进程内装配下成立。
//
// 这是用户明确问到的"适配器接手对外服务的部分"的验收点。

// anthropicMockUpstream 是只懂 Anthropic Messages 协议的假上游。
//
// 它会对**不符形状的请求主动报错**，而不是宽容接受 —— 宽容的 mock 会让
// "内核把 OpenAI 形状发给了 Anthropic 上游"这种错误测试通过。
func anthropicMockUpstream(t *testing.T) *httptest.Server {
	t.Helper()
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// Anthropic 的请求路径必须是 <base>/messages
		if !strings.HasSuffix(r.URL.Path, "/messages") {
			t.Errorf("请求打到了非 Anthropic 路径: %s", r.URL.Path)
			w.WriteHeader(http.StatusNotFound)
			return
		}
		body, _ := io.ReadAll(r.Body)
		var probe map[string]interface{}
		if err := json.Unmarshal(body, &probe); err != nil {
			t.Errorf("上游收到非 JSON 请求体: %s", body)
			w.WriteHeader(http.StatusBadRequest)
			return
		}
		// Anthropic 形状的硬性要求：messages 是数组；不能有 OpenAI 的 choices 概念
		if _, ok := probe["messages"]; !ok {
			t.Errorf("上游请求体缺 messages: %s", body)
		}
		if _, ok := probe["max_tokens"]; !ok {
			t.Errorf("Anthropic 要求 max_tokens，请求体缺此项: %s", body)
		}
		// 反向断言：OpenAI 兜底字段不该出现在 Anthropic 上游请求里
		for _, k := range []string{"choices", "chat_completions"} {
			if _, bad := probe[k]; bad {
				t.Errorf("Anthropic 上游请求体混入了 OpenAI 字段 %q: %s", k, body)
			}
		}

		if stream, _ := probe["stream"].(bool); stream {
			w.Header().Set("Content-Type", "text/event-stream")
			fl, _ := w.(http.Flusher)
			frames := []string{
				`event: message_start` + "\n" + `data: {"type":"message_start","message":{"id":"msg_e2e","type":"message","role":"assistant","model":"claude-e2e","content":[],"stop_reason":null,"usage":{"input_tokens":3,"output_tokens":0}}}`,
				`event: content_block_start` + "\n" + `data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`,
				`event: content_block_delta` + "\n" + `data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"pong"}}`,
				`event: content_block_stop` + "\n" + `data: {"type":"content_block_stop","index":0}`,
				`event: message_delta` + "\n" + `data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":1}}`,
				`event: message_stop` + "\n" + `data: {"type":"message_stop"}`,
			}
			for _, f := range frames {
				fmt.Fprint(w, f+"\n\n")
				if fl != nil {
					fl.Flush()
				}
			}
			return
		}

		w.Header().Set("Content-Type", "application/json")
		fmt.Fprint(w, `{"id":"msg_e2e","type":"message","role":"assistant","model":"claude-e2e",`+
			`"content":[{"type":"text","text":"pong"}],"stop_reason":"end_turn",`+
			`"usage":{"input_tokens":3,"output_tokens":1}}`)
	}))
	t.Cleanup(srv.Close)
	return srv
}

// writeAnthropicConfig 写一份把源挂到 anthropic 适配器上的配置。
func writeAnthropicConfig(t *testing.T, dir, listenAddr, upstreamURL string) string {
	t.Helper()
	cfg := "listen: " + listenAddr + "\n" +
		"gateway_keys:\n  - sk-e2e-0001\n" +
		"default_model: claude-e2e\n" +
		"adapter_dir: " + filepath.Join(dir, "adapters") + "\n" +
		"runtime_file: " + filepath.Join(dir, "runtime.json") + "\n" +
		"sources:\n" +
		"  - name: anth\n" +
		"    base_url: " + upstreamURL + "\n" +
		"    adapter: anthropic\n" +
		"    api_key: test-key\n" +
		"    max_concurrent: 32\n" +
		"    models:\n" +
		"      - id: claude-e2e\n" +
		"        kind: chat\n" +
		"        priority: 100\n"
	path := filepath.Join(dir, "config.yaml")
	if err := os.WriteFile(path, []byte(cfg), 0o644); err != nil {
		t.Fatalf("write config: %v", err)
	}
	return path
}

// TestAdapterTakeoverNonStream：非流式，Anthropic 客户端形状进、出。
func TestAdapterTakeoverNonStream(t *testing.T) {
	up := anthropicMockUpstream(t)
	dir := t.TempDir()
	addr := mustAddr(t)
	cfg := writeAnthropicConfig(t, dir, addr, up.URL)
	bin := buildBinary(t)
	g := startGateway(t, bin, addr, cfg)

	resp, body := g.do("POST", "/v1/messages",
		`{"model":"claude-e2e","max_tokens":64,"messages":[{"role":"user","content":"hi"}]}`, true)
	if resp == nil || resp.StatusCode != 200 {
		t.Fatalf("status=%v body=%q", statusOf(resp), body)
	}
	// 出向必须是 Anthropic Messages 形状，不能是 OpenAI choices 形状
	if strings.Contains(body, `"choices"`) {
		t.Errorf("Anthropic 响应里混入了 OpenAI choices 外壳: %s", body)
	}
	var out map[string]interface{}
	if err := json.Unmarshal([]byte(body), &out); err != nil {
		t.Fatalf("响应不是 JSON: %v: %s", err, body)
	}
	if out["type"] != "message" {
		t.Errorf(`响应 type = %v, 期望 "message"（Anthropic 形状）: %s`, out["type"], body)
	}
	// 文本内容必须能取出来 —— 这是"内容确实穿过适配器回来了"的证据
	if !strings.Contains(body, "pong") {
		t.Errorf("响应里没有上游返回的文本: %s", body)
	}
	if _, ok := out["usage"]; !ok {
		t.Errorf("Anthropic 响应应带 usage: %s", body)
	}
}

// TestAdapterTakeoverStream：流式也要是 Anthropic SSE 序列，
// 且**不得**出现 OpenAI 的 [DONE] 终止标记（R2 的"不混合"）。
func TestAdapterTakeoverStream(t *testing.T) {
	up := anthropicMockUpstream(t)
	dir := t.TempDir()
	addr := mustAddr(t)
	cfg := writeAnthropicConfig(t, dir, addr, up.URL)
	bin := buildBinary(t)
	g := startGateway(t, bin, addr, cfg)

	resp, body := g.do("POST", "/v1/messages",
		`{"model":"claude-e2e","max_tokens":64,"stream":true,"messages":[{"role":"user","content":"hi"}]}`, true)
	if resp == nil || resp.StatusCode != 200 {
		t.Fatalf("stream status=%v body=%q", statusOf(resp), body)
	}
	if !strings.Contains(body, "message_start") {
		t.Errorf("Anthropic 流应以 message_start 开头: %s", body)
	}
	if !strings.Contains(body, "message_stop") {
		t.Errorf("Anthropic 流应以 message_stop 收尾: %s", body)
	}
	if strings.Contains(body, "[DONE]") {
		t.Errorf("Anthropic 流里出现了 OpenAI 的 [DONE]: %s", body)
	}
	if strings.Contains(body, "chat.completion.chunk") {
		t.Errorf("流里混入 OpenAI chunk 形状: %s", body)
	}
	if !strings.Contains(body, "pong") {
		t.Errorf("流里没有上游文本: %s", body)
	}
}

// TestAdapterTakeoverStress 是对外服务路径的真实压力测试：
// 并发打真二进制，混合流式/非流式，验证无 5xx、上游命中数正确、
// 且每个响应都是完整合法的 Anthropic 形状（适配器在并发下不串味）。
func TestAdapterTakeoverStress(t *testing.T) {
	up := anthropicMockUpstream(t)
	dir := t.TempDir()
	addr := mustAddr(t)
	cfg := writeAnthropicConfig(t, dir, addr, up.URL)
	bin := buildBinary(t)
	g := startGateway(t, bin, addr, cfg)

	// max_concurrent 配成 32，让 96 个请求基本都能真正流过适配器 —— 这个测试
	// 要压的是**并发下序列化是否串味**，不是内核的容量拒绝策略。真的撞上上限
	// 时收到的 429 也是合法背压（见下方断言），不算失败。
	const (
		workers   = 12
		perWorker = 8
	)

	var (
		wg      sync.WaitGroup
		mu      sync.Mutex
		ok      int
		shed    int
		failed  []string
		streams int
	)
	start := make(chan struct{})

	for w := 0; w < workers; w++ {
		wg.Add(1)
		go func(w int) {
			defer wg.Done()
			<-start
			for i := 0; i < perWorker; i++ {
				isStream := (w+i)%2 == 0
				var payload string
				if isStream {
					payload = `{"model":"claude-e2e","max_tokens":32,"stream":true,"messages":[{"role":"user","content":"hi"}]}`
				} else {
					payload = `{"model":"claude-e2e","max_tokens":32,"messages":[{"role":"user","content":"hi"}]}`
				}
				resp, body := g.do("POST", "/v1/messages", payload, true)
				mu.Lock()
				switch {
				case resp == nil:
					failed = append(failed, "nil response")
				// 429 = 内核的容量背压（provider busy）。它是**正确**行为，
				// 不是缺陷：并发超限时就该如实拒绝，而不是排队或伪造响应。
				// 只要求它是结构良好的错误，不要求 200。
				case resp.StatusCode == http.StatusTooManyRequests:
					if !strings.Contains(body, "busy") && !strings.Contains(body, "error") {
						failed = append(failed, "429 不是结构化错误: "+body)
					} else {
						shed++
					}
				// 5xx 才是真失败 —— 那是内核或适配器崩了。
				case resp.StatusCode >= 500:
					failed = append(failed, fmt.Sprintf("5xx status=%d body=%.120s", resp.StatusCode, body))
				case resp.StatusCode != 200:
					failed = append(failed, fmt.Sprintf("status=%d body=%.120s", resp.StatusCode, body))
				case !strings.Contains(body, "pong"):
					// 200 但内容不对：说明并发下适配器把响应串味了
					failed = append(failed, "missing upstream text: "+body)
				case isStream && !strings.Contains(body, "message_stop"):
					failed = append(failed, "stream missing message_stop")
				case !isStream && strings.Contains(body, `"choices"`):
					failed = append(failed, "openai shell leaked: "+body)
				default:
					ok++
					if isStream {
						streams++
					}
				}
				mu.Unlock()
			}
		}(w)
	}
	close(start)
	wg.Wait()

	if len(failed) > 0 {
		t.Fatalf("%d/%d 请求失败，前 5 例:\n%s", len(failed), workers*perWorker,
			strings.Join(failed[:min(5, len(failed))], "\n"))
	}
	// 至少大部分请求必须真正流过 200 —— 全被背压拒掉说明并发上限配得太低，
	// 那这个压力测试就没压到适配器并发。
	if ok < workers*perWorker/2 {
		t.Fatalf("只有 %d/%d 请求真正成功，压力未触及适配器并发路径",
			ok, workers*perWorker)
	}
	t.Logf("成功 %d（其中流式 %d），背压 429 %d —— 无 5xx、无 OpenAI 形状泄漏",
		ok, streams, shed)
}

func min(a, b int) int {
	if a < b {
		return a
	}
	return b
}

var _ = time.Second