Cyrene_v2/internal/runtime/embedding/api_url_test.go

package embedding

import (
	"context"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
)

// TestNormalizeEndpointURL 钉住「回环别名 → 127.0.0.1」这张表，以及**不该被动**
// 的那些输入。全部是纯字符串断言，不连任何网络。
func TestNormalizeEndpointURL(t *testing.T) {
	t.Parallel()

	cases := []struct {
		name string
		in   string
		want string
	}{
		{
			name: "localhost 改写",
			in:   "http://localhost:11434/v1/embeddings",
			want: "http://127.0.0.1:11434/v1/embeddings",
		},
		{
			name: "localhost 大小写不敏感",
			in:   "http://LocalHost:11434/v1/embeddings",
			want: "http://127.0.0.1:11434/v1/embeddings",
		},
		{
			name: "localhost 无端口",
			in:   "http://localhost/v1/embeddings",
			want: "http://127.0.0.1/v1/embeddings",
		},
		{
			name: "https 同样改写",
			in:   "https://localhost:8443/v1/embeddings",
			want: "https://127.0.0.1:8443/v1/embeddings",
		},
		{
			// 0.0.0.0 是「监听所有接口」的语义，不是可连接的地址（拿它当连接目标会
			// 立即失败），因此这里是**防呆**：把写坏的地址掰回真正可连的回环字面量，
			// 而不是「支持用 0.0.0.0 去连」。
			name: "0.0.0.0 防呆改写",
			in:   "http://0.0.0.0:11434/v1/embeddings",
			want: "http://127.0.0.1:11434/v1/embeddings",
		},
		{
			// 空主机同样只是防呆：原文拼出来的 URL 指向不了任何服务。
			name: "空主机防呆改写",
			in:   "http:///v1/embeddings",
			want: "http://127.0.0.1/v1/embeddings",
		},
		{
			name: "显式 IPv6 字面量不改写",
			in:   "http://[::1]:11434/v1/embeddings",
			want: "http://[::1]:11434/v1/embeddings",
		},
		{
			name: "已是 127.0.0.1 原样返回",
			in:   "http://127.0.0.1:11434/v1/embeddings",
			want: "http://127.0.0.1:11434/v1/embeddings",
		},
		{
			name: "远端主机名不改写",
			in:   "https://embed.example.com/v1/embeddings",
			want: "https://embed.example.com/v1/embeddings",
		},
		{
			// 路径与查询串必须逐字不变：包括百分号编码、`+`、重复键与空值。
			name: "路径与查询串逐字不变",
			in:   "http://localhost:8080/v1/emb%65ddings?a=%2F+&b=1&b=&empty#frag",
			want: "http://127.0.0.1:8080/v1/emb%65ddings?a=%2F+&b=1&b=&empty#frag",
		},
		{
			// 未改写的主机（含显式 IPv6）必须**原串**返回，不做任何重排。
			name: "未改写时逐字返回原串",
			in:   "https://example.com:443/v1/a%2Fb?q=%7B%7D&r=1#f",
			want: "https://example.com:443/v1/a%2Fb?q=%7B%7D&r=1#f",
		},
	}

	for _, tc := range cases {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()
			got, err := normalizeEndpointURL(tc.in)
			if err != nil {
				t.Fatalf("normalizeEndpointURL(%q) error: %v", tc.in, err)
			}
			if got != tc.want {
				t.Fatalf("normalizeEndpointURL(%q) = %q, want %q", tc.in, got, tc.want)
			}
		})
	}
}

// TestNormalizeEndpointURL_NonHTTPSchemeUntouched 保证非 http(s) 的输入不被改写：
// 那类地址不是本客户端能拨的，改主机只会掩盖更该暴露的配置错误。
func TestNormalizeEndpointURL_NonHTTPSchemeUntouched(t *testing.T) {
	t.Parallel()

	const in = "file://localhost/v1/embeddings"
	got, err := normalizeEndpointURL(in)
	if err != nil {
		t.Fatalf("normalizeEndpointURL(%q) error: %v", in, err)
	}
	if got != in {
		t.Fatalf("normalizeEndpointURL(%q) = %q, want unchanged", in, got)
	}
}

// TestAPIEmbedder_NormalizesLoopbackHost 从**发出的请求**上验证规范化确实生效：
// 主机拨号目标变成 127.0.0.1，且 path + query 一字未动。用注入的 RoundTripper，
// 不连任何网络。
func TestAPIEmbedder_NormalizesLoopbackHost(t *testing.T) {
	t.Parallel()

	var gotHost, gotURI string
	rt := roundTripFunc(func(r *http.Request) (*http.Response, error) {
		gotHost = r.URL.Host
		gotURI = r.URL.RequestURI()
		return &http.Response{
			StatusCode: http.StatusOK,
			Body:       io.NopCloser(strings.NewReader(`{"data":[{"index":0,"embedding":[0.1]}]}`)),
			Header:     make(http.Header),
		}, nil
	})

	emb := NewAPI(
		config.EmbeddingConfig{APIURL: "http://localhost:11434/v1/embeddings?k=v#frag"},
		WithRoundTripper(rt),
	)
	if _, err := emb.Embed(context.Background(), []string{"hi"}); err != nil {
		t.Fatalf("Embed() error: %v", err)
	}
	if gotHost != "127.0.0.1:11434" {
		t.Errorf("dial host = %q, want 127.0.0.1:11434", gotHost)
	}
	if gotURI != "/v1/embeddings?k=v" {
		t.Errorf("request URI = %q, want /v1/embeddings?k=v", gotURI)
	}
}

// TestAPIEmbedder_KeepsExplicitIPv6Host 尊重使用者显式写下的 IPv6 字面量：
// 不改写、不重排。同样走注入的 RoundTripper，不连网络。
func TestAPIEmbedder_KeepsExplicitIPv6Host(t *testing.T) {
	t.Parallel()

	var gotHost string
	rt := roundTripFunc(func(r *http.Request) (*http.Response, error) {
		gotHost = r.URL.Host
		return &http.Response{
			StatusCode: http.StatusOK,
			Body:       io.NopCloser(strings.NewReader(`{"data":[{"index":0,"embedding":[0.1]}]}`)),
			Header:     make(http.Header),
		}, nil
	})

	emb := NewAPI(
		config.EmbeddingConfig{APIURL: "http://[::1]:11434/v1/embeddings"},
		WithRoundTripper(rt),
	)
	if _, err := emb.Embed(context.Background(), []string{"hi"}); err != nil {
		t.Fatalf("Embed() error: %v", err)
	}
	if gotHost != "[::1]:11434" {
		t.Errorf("dial host = %q, want [::1]:11434（显式 IPv6 不许被改写）", gotHost)
	}
}

// TestAPIEmbedder_LoopbackAliasReachesIPv4OnlyServer 是**真连本机**的场景验证
// （httptest 服务器只绑 127.0.0.1，正是 llama.cpp / ollama 的默认形态）：
// 端点写成 localhost / 0.0.0.0 时，请求必须仍然落到那个 IPv4 回环服务上并拿回向量。
//
// 注意它证明的是「场景通」而非「不做规范化就必挂」：2026-10-07 在本机实测，
// Go 客户端的拨号器对 localhost 会很快回落到 127.0.0.1（约 57ms），把这一条测试
// 的规范化关掉它**仍然会通过**。真正把这条测试变红的是 path/query 被破坏之类的
// 回归；「localhost 一定不去碰 ::1」这个保证由上面的字符串断言守住。
// 全程只连本进程起在 127.0.0.1 上的临时服务器，不碰任何外部服务。
func TestAPIEmbedder_LoopbackAliasReachesIPv4OnlyServer(t *testing.T) {
	t.Parallel()

	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = io.WriteString(w, `{"data":[{"index":0,"embedding":[0.1,0.2]}]}`)
	}))
	defer srv.Close()

	for _, alias := range []string{"localhost", "0.0.0.0"} {
		alias := alias
		// 刻意**不用** t.Parallel()：子测试若并行，父测试会先返回并执行
		// defer srv.Close()，子测试再跑时就只能撞上「连接被拒绝」。
		t.Run(alias, func(t *testing.T) {
			// srv.URL 形如 http://127.0.0.1:PORT —— 只把主机换成别名，端口保留，
			// 这样「服务确实只绑 IPv4」这个前提不变。
			raw := strings.Replace(srv.URL, "127.0.0.1", alias, 1) + "/v1/embeddings"
			emb := NewAPI(config.EmbeddingConfig{APIURL: raw})

			vectors, err := emb.Embed(context.Background(), []string{"hi"})
			if err != nil {
				t.Fatalf("端点写成 %s 时未落到 IPv4 回环服务：%v", alias, err)
			}
			if len(vectors) != 1 || len(vectors[0]) != 2 {
				t.Fatalf("vectors = %v, want 1 个 2 维向量", vectors)
			}
		})
	}
}

// reviewBudgetRef 是「入库前审查」单条总预算的**参照值**，只用于本文件的断言：
// 权威定义在 internal/runtime/memory/preinsert_config.go 的 defaultReviewTimeout。
// 跨包的真实断言在 memory 包（preinsert_recall_test.go）里，这里钉住同一个口径，
// 免得 embedding 侧悄悄把超时调到 ≥ 审查预算。
//
// 2026-10-13 由 20s 改为 90s：审查的召回链是 201 条 ⇒ ⌈201/8⌉ = 26 个 embedding
// 请求，整条链实测 21.4~21.6s（空闲）—— **20s 的旧值比链长还短**，于是每一条候选
// 满额的审查都必然超时（这是「92% 的审查 recall_mode 为空」的根因）。本文件钉住的
// 是同一个数，不是另一个数：embedding 侧任何一次调用的超时都必须严格小于它。
const reviewBudgetRef = 90 * time.Second

// TestDefaultRequestTimeout 钉住单次 embedding 调用的超时值，并断言它**严格小于**
// 审查总预算：否则一次卡死的 embedding 连接就能吃光整条审查的预算，排在后面的
// T1 灰区判定连执行机会都没有（2026-10-07 实测故障形态）。同时验证该超时确实
// 打在了发出去的请求上。
//
// 🔴 冷启动放宽（warmupTimeout）必须一并守住同一条界：它只发生一次，但它那一次的
// 等待**也**不能吃光整轮审查 —— 否则「一次挂死 = 整轮审查没有预算给 T1」这个形态
// 会从常态超时挪到放宽预算上原样复现（reviewer round1 的 p1）。因此这里对
// warmupTimeout 做与 DefaultRequestTimeout **同一条**断言。
func TestDefaultRequestTimeout(t *testing.T) {
	t.Parallel()

	if DefaultRequestTimeout <= 0 {
		t.Fatalf("DefaultRequestTimeout = %s, 必须为正", DefaultRequestTimeout)
	}
	if DefaultRequestTimeout >= reviewBudgetRef {
		t.Fatalf("embedding 单次超时 %s 必须严格小于审查总预算 %s",
			DefaultRequestTimeout, reviewBudgetRef)
	}
	if DefaultRequestTimeout != 5*time.Second {
		t.Fatalf("DefaultRequestTimeout = %s, want 5s（值变了要连带复核审查预算口径）",
			DefaultRequestTimeout)
	}
	// 冷启动放宽的两条界：仍在常态超时的**同一量级**（不得成倍放大），
	// 且严格小于审查总预算（一次冷启动不许吃光整轮）。
	if warmupTimeout <= 0 {
		t.Fatalf("warmupTimeout = %s, 必须为正", warmupTimeout)
	}
	if warmupTimeout > 2*DefaultRequestTimeout {
		t.Fatalf("冷启动放宽 %s 超过常态超时 %s 的两倍 —— 它必须是同一量级的一次性放宽",
			warmupTimeout, DefaultRequestTimeout)
	}
	if warmupTimeout >= reviewBudgetRef {
		t.Fatalf("冷启动放宽 %s 必须严格小于审查总预算 %s（否则一次冷启动就吃光整轮审查）",
			warmupTimeout, reviewBudgetRef)
	}

	var slack time.Duration
	rt := roundTripFunc(func(r *http.Request) (*http.Response, error) {
		dl, ok := r.Context().Deadline()
		if !ok {
			t.Error("请求上下文没有 deadline，单次超时未生效")
		} else {
			slack = time.Until(dl)
		}
		return &http.Response{
			StatusCode: http.StatusOK,
			Body:       io.NopCloser(strings.NewReader(`{"data":[{"index":0,"embedding":[0.1]}]}`)),
			Header:     make(http.Header),
		}, nil
	})

	emb := NewAPI(
		config.EmbeddingConfig{APIURL: "http://127.0.0.1:11434/v1/embeddings"},
		WithRoundTripper(rt),
	)
	// 第一次是冷启动调用（warmupTimeout），结果由它自己的测试断言；这里连发两次，
	// 只看稳态那一次打出去的 deadline。
	for i := 0; i < 2; i++ {
		if _, err := emb.Embed(context.Background(), []string{"hi"}); err != nil {
			t.Fatalf("第 %d 次 Embed() error: %v", i+1, err)
		}
	}
	if slack <= 0 || slack > DefaultRequestTimeout {
		t.Fatalf("请求 deadline 余量 = %s, want (0, %s]", slack, DefaultRequestTimeout)
	}
}