Spaces:

a3216
/

gcli2api

Running

gcli2api / internal /syncer /writer_test.go

chore: 同步到上游 1.12.0-panel + 凭证同步/独立启动器/保活

6d60378 verified 30 days ago

2.92 kB

package syncer

import (
	"strings"
	"testing"
)

// TestHFCLIWriterEnv 锁定给 hf 子进程的环境：
//
//   - no_proxy 必须被替换掉。宿主环境里常见 no_proxy=localhost,127.0.0.1,::1,[::1]，
//     而 httpx 解析带方括号的 IPv6 会直接抛
//     InvalidURL: Invalid port: ':1]' —— hf CLI 因此连启动都进不去。
//   - token 与 CA 包必须注入：前者是鉴权，后者是本机抓包代理场景下唯一能让
//     httpx 通过证书校验的办法（它不像 Go 那样吃 Windows 信任库）。
func TestHFCLIWriterEnv(t *testing.T) {
	w := &HFCLIWriter{Bucket: "u/b", Token: "hf_secret", CABundle: `C:\certs\proxy-ca.pem`}
	env := w.env()

	get := func(key string) (string, bool) {
		for _, kv := range env {
			if strings.EqualFold(strings.SplitN(kv, "=", 2)[0], key) {
				return strings.SplitN(kv, "=", 2)[1], true
			}
		}
		return "", false
	}

	np, ok := get("no_proxy")
	if !ok {
		t.Fatal("no_proxy 未被注入")
	}
	if strings.Contains(np, "[") {
		t.Errorf("no_proxy=%q 仍含方括号，httpx 会解析失败", np)
	}
	for _, kv := range env {
		if strings.Contains(strings.ToLower(kv), "no_proxy") && strings.Contains(kv, "[::1]") {
			t.Errorf("残留会打死 httpx 的 no_proxy: %s", kv)
		}
	}

	if tk, _ := get("HF_TOKEN"); tk != "hf_secret" {
		t.Errorf("HF_TOKEN=%q", tk)
	}
	for _, key := range []string{"SSL_CERT_FILE", "REQUESTS_CA_BUNDLE", "CURL_CA_BUNDLE"} {
		if v, _ := get(key); v != `C:\certs\proxy-ca.pem` {
			t.Errorf("%s=%q want the ca bundle", key, v)
		}
	}
	if v, _ := get("PYTHONIOENCODING"); !strings.EqualFold(v, "utf-8") {
		t.Errorf("PYTHONIOENCODING=%q want utf-8", v)
	}
}

// TestHFCLIWriterEnvWithoutCABundle 没配 CA 包时不应凭空注入这几个变量
// （否则会把一个空路径塞给 httpx，反而更难排查）。
func TestHFCLIWriterEnvWithoutCABundle(t *testing.T) {
	w := &HFCLIWriter{Bucket: "u/b"}
	for _, kv := range w.env() {
		if strings.HasPrefix(kv, "SSL_CERT_FILE=") {
			t.Errorf("未配置 ca_bundle 时不应注入 SSL_CERT_FILE: %s", kv)
		}
	}
}

// TestWriterSelection 写入通道的挑选规则。
func TestWriterSelection(t *testing.T) {
	// 显式 relay：没给地址必须报错，而不是悄悄退化成只读。
	if _, err := NewWriter(Config{Writer: "relay"}); err == nil {
		t.Error("writer=relay 但 relay_url 为空时应报错")
	}
	if _, err := NewWriter(Config{Writer: "relay", RelayURL: "https://x"}); err != nil {
		t.Errorf("writer=relay 正常配置不应报错: %v", err)
	}
	if _, err := NewWriter(Config{Writer: "bogus"}); err == nil {
		t.Error("未知 writer 应报错")
	}
	// hf 通道显式指定时不要求二进制存在（Available 只在 auto 时才需要）。
	w, err := NewWriter(Config{Writer: "hf", Bucket: "u/b"})
	if err != nil {
		t.Fatalf("writer=hf 不应报错: %v", err)
	}
	if w.Name() != "hf-cli" {
		t.Errorf("name=%q", w.Name())
	}
}

	package syncer

	import (
	"strings"
	"testing"
	)

	// TestHFCLIWriterEnv 锁定给 hf 子进程的环境：
	//
	// - no_proxy 必须被替换掉。宿主环境里常见 no_proxy=localhost,127.0.0.1,::1,[::1]，
	// 而 httpx 解析带方括号的 IPv6 会直接抛
	// InvalidURL: Invalid port: ':1]' —— hf CLI 因此连启动都进不去。
	// - token 与 CA 包必须注入：前者是鉴权，后者是本机抓包代理场景下唯一能让
	// httpx 通过证书校验的办法（它不像 Go 那样吃 Windows 信任库）。
	func TestHFCLIWriterEnv(t *testing.T) {
	w := &HFCLIWriter{Bucket: "u/b", Token: "hf_secret", CABundle: `C:\certs\proxy-ca.pem`}
	env := w.env()

	get := func(key string) (string, bool) {
	for _, kv := range env {
	if strings.EqualFold(strings.SplitN(kv, "=", 2)[0], key) {
	return strings.SplitN(kv, "=", 2)[1], true
	}
	}
	return "", false
	}

	np, ok := get("no_proxy")
	if !ok {
	t.Fatal("no_proxy 未被注入")
	}
	if strings.Contains(np, "[") {
	t.Errorf("no_proxy=%q 仍含方括号，httpx 会解析失败", np)
	}
	for _, kv := range env {
	if strings.Contains(strings.ToLower(kv), "no_proxy") && strings.Contains(kv, "[::1]") {
	t.Errorf("残留会打死 httpx 的 no_proxy: %s", kv)
	}
	}

	if tk, _ := get("HF_TOKEN"); tk != "hf_secret" {
	t.Errorf("HF_TOKEN=%q", tk)
	}
	for _, key := range []string{"SSL_CERT_FILE", "REQUESTS_CA_BUNDLE", "CURL_CA_BUNDLE"} {
	if v, _ := get(key); v != `C:\certs\proxy-ca.pem` {
	t.Errorf("%s=%q want the ca bundle", key, v)
	}
	}
	if v, _ := get("PYTHONIOENCODING"); !strings.EqualFold(v, "utf-8") {
	t.Errorf("PYTHONIOENCODING=%q want utf-8", v)
	}
	}

	// TestHFCLIWriterEnvWithoutCABundle 没配 CA 包时不应凭空注入这几个变量
	// （否则会把一个空路径塞给 httpx，反而更难排查）。
	func TestHFCLIWriterEnvWithoutCABundle(t *testing.T) {
	w := &HFCLIWriter{Bucket: "u/b"}
	for _, kv := range w.env() {
	if strings.HasPrefix(kv, "SSL_CERT_FILE=") {
	t.Errorf("未配置 ca_bundle 时不应注入 SSL_CERT_FILE: %s", kv)
	}
	}
	}

	// TestWriterSelection 写入通道的挑选规则。
	func TestWriterSelection(t *testing.T) {
	// 显式 relay：没给地址必须报错，而不是悄悄退化成只读。
	if _, err := NewWriter(Config{Writer: "relay"}); err == nil {
	t.Error("writer=relay 但 relay_url 为空时应报错")
	}
	if _, err := NewWriter(Config{Writer: "relay", RelayURL: "https://x"}); err != nil {
	t.Errorf("writer=relay 正常配置不应报错: %v", err)
	}
	if _, err := NewWriter(Config{Writer: "bogus"}); err == nil {
	t.Error("未知 writer 应报错")
	}
	// hf 通道显式指定时不要求二进制存在（Available 只在 auto 时才需要）。
	w, err := NewWriter(Config{Writer: "hf", Bucket: "u/b"})
	if err != nil {
	t.Fatalf("writer=hf 不应报错: %v", err)
	}
	if w.Name() != "hf-cli" {
	t.Errorf("name=%q", w.Name())
	}
	}