Files
openteam/server/internal/channel/channel_test.go
T
SakurasanandClaude 0637ce0a51 M5: 渠道体系(负载均衡+并发控制+健康检查+故障转移)
- channel.Candidates 按模型绑定取候选 + Pick 加权随机负载均衡
- TryAcquire 每渠道并发信号量, 满载溢出到其他候选
- HealthMonitor 后台定时探测, 连续失败进 cooldown, 恢复放回(可配 interval/threshold)
- doProxy 遍历候选故障转移: 网络错误/429/5xx/超时且未写出响应头时安全重试;
  400 等业务错误透传, 流式写出首字节后放弃重试
- 单测覆盖候选过滤/加权/并发/健康状态机
- E2E: 杀上游自动切换、cooldown、恢复、并发溢出 10/10

Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-15 15:57:48 +08:00

132 lines
3.6 KiB
Go

package channel
import (
"context"
"os"
"testing"
"time"
"github.com/openteam/server/internal/pkg/crypto"
"github.com/openteam/server/internal/store"
"gorm.io/gorm"
)
func newTestDB(t *testing.T) *gorm.DB {
t.Helper()
dsn := "/tmp/channel_test_" + time.Now().Format("150405.000000000") + ".db"
db, err := store.Open("sqlite", dsn)
if err != nil {
t.Fatalf("open db: %v", err)
}
t.Cleanup(func() {
sqlDB, _ := db.DB()
sqlDB.Close()
os.Remove(dsn)
})
return db
}
func seedChannel(t *testing.T, db *gorm.DB, name string, weight, priority, maxConc int) uint64 {
t.Helper()
enc := crypto.NewEncryptor("test-master-key-1234567890")
key, _ := enc.Encrypt("upstream-key")
ch := store.Channel{
Name: name, Provider: store.ChannelProviderOpenAI, BaseURL: "http://localhost:9000",
APIKeyEnc: key, Weight: weight, Priority: priority, MaxConcurrency: maxConc,
HealthStatus: store.ChannelHealthHealthy, Enabled: true,
}
if err := db.Create(&ch).Error; err != nil {
t.Fatalf("create channel: %v", err)
}
return ch.ID
}
func TestCandidatesFiltersUnhealthy(t *testing.T) {
db := newTestDB(t)
seedChannel(t, db, "a", 1, 0, 4)
seedChannel(t, db, "b", 5, 1, 4)
// 禁用渠道不入候选
bad := store.Channel{Name: "bad", Provider: store.ChannelProviderOpenAI, BaseURL: "http://x", APIKeyEnc: "x", HealthStatus: store.ChannelHealthCooldown, Enabled: true}
db.Create(&bad)
s := NewService(db, crypto.NewEncryptor("test-master-key-1234567890"))
cands := s.Candidates("")
if len(cands) != 2 {
t.Fatalf("candidates = %d, want 2", len(cands))
}
if cands[0].Name != "a" {
t.Fatalf("first by priority should be a, got %s", cands[0].Name)
}
}
func TestPickWeighted(t *testing.T) {
db := newTestDB(t)
seedChannel(t, db, "only", 1, 0, 4)
s := NewService(db, crypto.NewEncryptor("test-master-key-1234567890"))
cands := s.Candidates("")
chosen := s.Pick(cands)
if chosen == nil || chosen.Name != "only" {
t.Fatalf("pick single: %+v", chosen)
}
// 空候选返回 nil
if p := s.Pick(nil); p != nil {
t.Fatal("expected nil for empty candidates")
}
}
func TestTryAcquire(t *testing.T) {
db := newTestDB(t)
id := seedChannel(t, db, "c", 1, 0, 1)
var ch store.Channel
db.First(&ch, id)
s := NewService(db, crypto.NewEncryptor("test-master-key-1234567890"))
release, ok := s.TryAcquire(&ch)
if !ok {
t.Fatal("first acquire should succeed")
}
if _, ok := s.TryAcquire(&ch); ok {
t.Fatal("second acquire should fail (capacity 1)")
}
release()
if _, ok := s.TryAcquire(&ch); !ok {
t.Fatal("acquire after release should succeed")
}
}
func TestHealthCooldownAndRecovery(t *testing.T) {
db := newTestDB(t)
id := seedChannel(t, db, "h", 1, 0, 4)
var ch store.Channel
db.First(&ch, id)
m := NewHealthMonitor(db, crypto.NewEncryptor("test-master-key-1234567890"), HealthConfig{
Interval: time.Minute, FailThreshold: 2, Timeout: time.Second,
})
// 两次失败 → cooldown
m.record(&ch, false)
if ch.HealthStatus != store.ChannelHealthHealthy {
t.Fatal("one failure should stay healthy")
}
m.record(&ch, false)
if ch.HealthStatus != store.ChannelHealthCooldown {
t.Fatalf("two failures should cooldown, got %s", ch.HealthStatus)
}
// 恢复
m.record(&ch, true)
if ch.HealthStatus != store.ChannelHealthHealthy {
t.Fatalf("success should recover, got %s", ch.HealthStatus)
}
}
func TestHealthStartStops(t *testing.T) {
db := newTestDB(t)
seedChannel(t, db, "probe", 1, 0, 4)
m := NewHealthMonitor(db, crypto.NewEncryptor("test-master-key-1234567890"), HealthConfig{
Interval: time.Hour, FailThreshold: 2, Timeout: time.Second,
})
ctx, cancel := context.WithCancel(context.Background())
m.Start(ctx)
cancel() // 不应 panic
}