test(channel-monitor): quota mode unit/integration/migration coverage

- fetcher:海外/CN coding/CN payg 分派、账号缺失、凭据失效标记、
  TTL 命中与失败不缓存、UsageInfo→tiers 全窗口归一、状态推导矩阵
- quota mode:RunCheck 三分派(quota 单行 / quota_probe 仅挂主行 /
  probe 不变)、配额失败不翻探活状态、校验矩阵、关联账号复核
  (quota 报错 probe 自动解绑)、quota→probe 切换强制重填 key、
  Duplicate 空明文重加密
- settings:channel_monitor_show_quota 缺省关闭 + 仅 "true" 开启
- 迁移内容断言(仿 176 grok 迁移测试)
- repo 集成测试:check_mode/account_id 往返、quota JSONB 回读、
  探活旧行 NULL 兼容(testcontainers PG 实跑迁移 226)
- 修复 Fetch 在 nil receiver 上的 panic(缓存查询先于原 nil 守卫)
This commit is contained in:
Randark
2026-08-18 04:51:02 +00:00
parent 41344c20ff
commit 7ab6d3db66
7 changed files with 1083 additions and 5 deletions
@@ -0,0 +1,155 @@
//go:build integration
package repository
import (
"context"
"testing"
"time"
"github.com/Wei-Shaw/sub2api/internal/domain"
"github.com/Wei-Shaw/sub2api/internal/service"
"github.com/stretchr/testify/require"
)
// 配额模式 repo 层集成测试:
// - Create/GetByID/Update 的 check_mode/account_id 往返
// - InsertHistoryBatch → ListHistory / ListLatestForMonitorIDs 的 quota JSONB 回读
// (裸 SQL 列 + scanMonitorQuota),以及探活模式旧行 quota=NULL 的兼容
//
// 注意 channelMonitorRepository 的 GetByID/裸 SQL 走全局 client(不识别 tx ctx),
// 因此本文件用 integrationEntClient 直连 + t.Cleanup 显式清理,不走 testEntTx 回滚。
func TestChannelMonitorQuotaModeRoundTrip(t *testing.T) {
ctx := context.Background()
repo := NewChannelMonitorRepository(integrationEntClient, integrationDB)
account := mustCreateAccount(t, integrationEntClient, &service.Account{
Name: "quota-linked-kimi", Platform: domain.PlatformKimi, Type: service.AccountTypeAPIKey,
Credentials: map[string]any{"api_key": "sk-kimi", "account_mode": service.AccountModeCoding},
})
t.Cleanup(func() {
_ = integrationEntClient.Account.DeleteOneID(account.ID).Exec(ctx)
})
created := &service.ChannelMonitor{
Name: "kimi-quota-roundtrip",
Provider: service.MonitorProviderKimi,
APIMode: service.MonitorAPIModeChatCompletions,
Endpoint: "",
APIKey: "encrypted-empty",
PrimaryModel: "quota",
Enabled: true,
IntervalSeconds: 60,
CheckMode: service.MonitorCheckModeQuota,
AccountID: &account.ID,
BodyOverrideMode: service.MonitorBodyOverrideModeOff,
}
require.NoError(t, repo.Create(ctx, created))
t.Cleanup(func() {
_ = repo.Delete(ctx, created.ID)
})
loaded, err := repo.GetByID(ctx, created.ID)
require.NoError(t, err)
require.Equal(t, service.MonitorCheckModeQuota, loaded.CheckMode)
require.NotNil(t, loaded.AccountID)
require.Equal(t, account.ID, *loaded.AccountID)
// Update:切换模式并清空关联账号(probe 化)。
loaded.CheckMode = service.MonitorCheckModeProbe
loaded.AccountID = nil
loaded.Endpoint = "https://api.moonshot.cn"
require.NoError(t, repo.Update(ctx, loaded))
reloaded, err := repo.GetByID(ctx, created.ID)
require.NoError(t, err)
require.Equal(t, service.MonitorCheckModeProbe, reloaded.CheckMode)
require.Nil(t, reloaded.AccountID)
// 重新绑定账号。
reloaded.CheckMode = service.MonitorCheckModeQuotaProbe
reloaded.AccountID = &account.ID
require.NoError(t, repo.Update(ctx, reloaded))
final, err := repo.GetByID(ctx, created.ID)
require.NoError(t, err)
require.Equal(t, service.MonitorCheckModeQuotaProbe, final.CheckMode)
require.NotNil(t, final.AccountID)
}
func TestChannelMonitorHistoryQuotaRoundTrip(t *testing.T) {
ctx := context.Background()
repo := NewChannelMonitorRepository(integrationEntClient, integrationDB)
monitor := &service.ChannelMonitor{
Name: "quota-history-roundtrip",
Provider: service.MonitorProviderOpenAI,
APIMode: service.MonitorAPIModeChatCompletions,
Endpoint: "https://api.openai.com",
APIKey: "encrypted",
PrimaryModel: "gpt-test",
ExtraModels: []string{"gpt-extra"},
Enabled: true,
IntervalSeconds: 60,
BodyOverrideMode: service.MonitorBodyOverrideModeOff,
}
require.NoError(t, repo.Create(ctx, monitor))
t.Cleanup(func() {
_ = repo.Delete(ctx, monitor.ID) // histories 级联删除
})
now := time.Now().UTC()
rows := []*service.ChannelMonitorHistoryRow{
{
MonitorID: monitor.ID, Model: "gpt-test", Status: service.MonitorStatusOperational,
Message: "ok", CheckedAt: now,
Quota: &domain.MonitorQuotaSnapshot{
Source: "usage",
Success: true,
PlanLevel: "PRO",
Tiers: []domain.MonitorQuotaTier{
{Window: "5h", UsedPercent: 42.5, Used: 17, Limit: 40, ResetAt: now.Add(time.Hour).Format(time.RFC3339)},
},
FetchedAt: now,
},
},
{
// 探活模式旧行:无 quota(NULL 兼容)。
MonitorID: monitor.ID, Model: "gpt-extra", Status: service.MonitorStatusOperational,
Message: "ok", CheckedAt: now,
},
}
require.NoError(t, repo.InsertHistoryBatch(ctx, rows))
history, err := repo.ListHistory(ctx, monitor.ID, "", 10)
require.NoError(t, err)
require.Len(t, history, 2)
byModel := map[string]*service.ChannelMonitorHistoryEntry{}
for _, entry := range history {
byModel[entry.Model] = entry
}
withQuota := byModel["gpt-test"]
require.NotNil(t, withQuota.Quota)
require.True(t, withQuota.Quota.Success)
require.Equal(t, "usage", withQuota.Quota.Source)
require.Equal(t, "PRO", withQuota.Quota.PlanLevel)
require.Len(t, withQuota.Quota.Tiers, 1)
require.Equal(t, "5h", withQuota.Quota.Tiers[0].Window)
require.InDelta(t, 42.5, withQuota.Quota.Tiers[0].UsedPercent, 0.001)
require.Nil(t, byModel["gpt-extra"].Quota, "probe rows must read back as NULL quota")
// 用户视图聚合:主模型最近一行带快照。
latest, err := repo.ListLatestForMonitorIDs(ctx, []int64{monitor.ID})
require.NoError(t, err)
primaryRows := latest[monitor.ID]
require.NotEmpty(t, primaryRows)
var primaryQuota *domain.MonitorQuotaSnapshot
for _, row := range primaryRows {
if row.Model == "gpt-test" {
primaryQuota = row.Quota
}
}
require.NotNil(t, primaryQuota, "ListLatestForMonitorIDs must surface quota for the primary model")
require.True(t, primaryQuota.Success)
}
@@ -100,6 +100,11 @@ func (f *ChannelMonitorQuotaFetcher) LoadAccount(ctx context.Context, id int64)
// Fetch 抓取账号的最新配额快照。永不返回 error:失败降级为
// Success=false 快照(Error 带摘要),保证检测历史的时间线连续。
func (f *ChannelMonitorQuotaFetcher) Fetch(ctx context.Context, accountID int64) *domain.MonitorQuotaSnapshot {
if f == nil {
// fail-closed:fetcher 未注入(存量测试构造)时不 panic,降级为错误快照。
return quotaErrorSnapshot("usage", "quota fetcher is not configured", time.Now())
}
now := time.Now()
if cached, ok := f.cachedSnapshot(accountID, now); ok {
@@ -171,11 +176,11 @@ func (f *ChannelMonitorQuotaFetcher) fetchUsage(ctx context.Context, accountID i
}
}
snapshot := &domain.MonitorQuotaSnapshot{
Source: "usage",
Success: true,
PlanLevel: usage.SubscriptionTier,
Tiers: usageQuotaTiers(usage),
FetchedAt: now,
Source: "usage",
Success: true,
PlanLevel: usage.SubscriptionTier,
Tiers: usageQuotaTiers(usage),
FetchedAt: now,
}
if snapshot.PlanLevel == "" {
snapshot.PlanLevel = usage.SubscriptionTierRaw
@@ -0,0 +1,392 @@
//go:build unit
package service
import (
"context"
"errors"
"testing"
"time"
"github.com/Wei-Shaw/sub2api/internal/domain"
"github.com/Wei-Shaw/sub2api/internal/pkg/xai"
"github.com/stretchr/testify/require"
)
// --- fetcher 依赖 stub ---
type stubMonitorUsageSource struct {
usage *UsageInfo
err error
calls int
lastCtx context.Context
}
func (s *stubMonitorUsageSource) GetUsage(ctx context.Context, accountID int64, force ...bool) (*UsageInfo, error) {
s.calls++
s.lastCtx = ctx
return s.usage, s.err
}
type stubMonitorCNQuotaSource struct {
result *CNProviderQuotaProbeResult
err error
calls int
}
func (s *stubMonitorCNQuotaSource) QueryUsage(ctx context.Context, accountID int64) (*CNProviderQuotaProbeResult, error) {
s.calls++
return s.result, s.err
}
type stubMonitorCNBalanceSource struct {
result *CNProviderBalanceResult
err error
calls int
}
func (s *stubMonitorCNBalanceSource) QueryBalance(ctx context.Context, accountID int64) (*CNProviderBalanceResult, error) {
s.calls++
return s.result, s.err
}
type stubMonitorAccountSource struct {
accounts map[int64]*Account
err error
calls int
}
func (s *stubMonitorAccountSource) GetByID(ctx context.Context, id int64) (*Account, error) {
s.calls++
if s.err != nil {
return nil, s.err
}
return s.accounts[id], nil
}
func newQuotaFetcherTestSetup(t *testing.T) (*ChannelMonitorQuotaFetcher, *stubMonitorUsageSource, *stubMonitorCNQuotaSource, *stubMonitorCNBalanceSource, *stubMonitorAccountSource) {
t.Helper()
usage := &stubMonitorUsageSource{}
cnQuota := &stubMonitorCNQuotaSource{}
cnBalance := &stubMonitorCNBalanceSource{}
accounts := &stubMonitorAccountSource{accounts: make(map[int64]*Account)}
fetcher := &ChannelMonitorQuotaFetcher{
usage: usage,
cnQuota: cnQuota,
cnBalance: cnBalance,
accounts: accounts,
cache: make(map[int64]monitorQuotaCacheEntry),
}
return fetcher, usage, cnQuota, cnBalance, accounts
}
// --- 分派 ---
func TestQuotaFetcher_OverseasAccountUsesUsageService(t *testing.T) {
fetcher, usage, _, cnQuota, accounts := newQuotaFetcherTestSetup(t)
accounts.accounts[7] = &Account{ID: 7, Platform: domain.PlatformAnthropic}
resets := time.Now().Add(2 * time.Hour).UTC()
usage.usage = &UsageInfo{
FiveHour: &UsageProgress{Utilization: 42.5, UsedRequests: 17, LimitRequests: 40, ResetsAt: &resets},
SevenDay: &UsageProgress{Utilization: 10},
SubscriptionTier: "PRO",
}
snapshot := fetcher.Fetch(context.Background(), 7)
require.True(t, snapshot.Success)
require.Equal(t, "usage", snapshot.Source)
require.Equal(t, "PRO", snapshot.PlanLevel)
require.False(t, snapshot.CredentialInvalid)
require.Empty(t, snapshot.Error)
require.Len(t, snapshot.Tiers, 2)
fiveHour := snapshot.Tiers[0]
require.Equal(t, "5h", fiveHour.Window)
require.Empty(t, fiveHour.Label)
require.InDelta(t, 42.5, fiveHour.UsedPercent, 0.001)
require.Equal(t, float64(17), fiveHour.Used)
require.Equal(t, float64(40), fiveHour.Limit)
require.NotEmpty(t, fiveHour.ResetAt)
require.Equal(t, "7d", snapshot.Tiers[1].Window)
require.Equal(t, 1, usage.calls)
require.Equal(t, 0, cnQuota.calls)
}
func TestQuotaFetcher_CodingPlanAccountUsesCNQuota(t *testing.T) {
fetcher, _, cnQuota, cnBalance, accounts := newQuotaFetcherTestSetup(t)
accounts.accounts[9] = &Account{
ID: 9,
Platform: domain.PlatformKimi,
Credentials: map[string]any{"account_mode": AccountModeCoding},
}
cnQuota.result = &CNProviderQuotaProbeResult{
Success: true,
CredentialValid: true,
PlanLevel: "",
Tiers: []CNQuotaTier{
{Window: "5h", UsedPercent: 33.3, ResetAt: "2026-08-18T06:00:00Z"},
{Window: "weekly", UsedPercent: 12},
},
}
snapshot := fetcher.Fetch(context.Background(), 9)
require.True(t, snapshot.Success)
require.Equal(t, "cn_quota", snapshot.Source)
require.Len(t, snapshot.Tiers, 2)
require.Equal(t, "5h", snapshot.Tiers[0].Window)
require.InDelta(t, 33.3, snapshot.Tiers[0].UsedPercent, 0.001)
require.Equal(t, "weekly", snapshot.Tiers[1].Window)
require.Equal(t, 1, cnQuota.calls)
require.Equal(t, 0, cnBalance.calls)
}
func TestQuotaFetcher_PayGAccountUsesCNBalance(t *testing.T) {
fetcher, _, _, cnBalance, accounts := newQuotaFetcherTestSetup(t)
accounts.accounts[11] = &Account{
ID: 11,
Platform: domain.PlatformDeepseek,
Credentials: map[string]any{"account_mode": AccountModePayG},
}
cnBalance.result = &CNProviderBalanceResult{
Success: true,
Balance: 12.34,
Currency: "CNY",
Balances: []CNProviderBalanceEntry{
{Currency: "CNY", Balance: 12.34},
{Currency: "USD", Balance: 1.5},
},
}
snapshot := fetcher.Fetch(context.Background(), 11)
require.True(t, snapshot.Success)
require.Equal(t, "cn_balance", snapshot.Source)
require.NotNil(t, snapshot.Balance)
require.InDelta(t, 12.34, *snapshot.Balance, 0.001)
require.Equal(t, "CNY", snapshot.Currency)
require.Len(t, snapshot.Balances, 2)
require.Equal(t, "USD", snapshot.Balances[1].Currency)
require.Empty(t, snapshot.Error)
}
// --- 失败路径(Fetch 永不返回 error) ---
func TestQuotaFetcher_AccountMissingYieldsLinkedAccountSnapshot(t *testing.T) {
fetcher, usage, _, _, accounts := newQuotaFetcherTestSetup(t)
accounts.err = errors.New("not found")
snapshot := fetcher.Fetch(context.Background(), 404)
require.False(t, snapshot.Success)
require.Equal(t, "linked account not found", snapshot.Error)
require.Equal(t, 0, usage.calls) // 未走到数据源
}
func TestQuotaFetcher_UsageAuthErrorMarksCredentialInvalid(t *testing.T) {
fetcher, usage, _, _, accounts := newQuotaFetcherTestSetup(t)
accounts.accounts[3] = &Account{ID: 3, Platform: domain.PlatformOpenAI}
usage.err = errors.New("API returned 401: unauthorized")
snapshot := fetcher.Fetch(context.Background(), 3)
require.False(t, snapshot.Success)
require.True(t, snapshot.CredentialInvalid)
require.Contains(t, snapshot.Error, "401")
}
func TestQuotaFetcher_CNQuotaCredentialInvalidFlagPropagates(t *testing.T) {
fetcher, _, cnQuota, _, accounts := newQuotaFetcherTestSetup(t)
accounts.accounts[5] = &Account{
ID: 5,
Platform: domain.PlatformZhipu,
Credentials: map[string]any{"account_mode": AccountModeCoding},
}
cnQuota.result = &CNProviderQuotaProbeResult{Success: false, CredentialValid: false, Error: "api key expired"}
snapshot := fetcher.Fetch(context.Background(), 5)
require.False(t, snapshot.Success)
require.True(t, snapshot.CredentialInvalid)
require.Equal(t, "api key expired", snapshot.Error)
}
func TestQuotaFetcher_CNBalanceHTTP403MarksCredentialInvalid(t *testing.T) {
fetcher, _, _, cnBalance, accounts := newQuotaFetcherTestSetup(t)
accounts.accounts[6] = &Account{ID: 6, Platform: domain.PlatformKimi}
cnBalance.result = &CNProviderBalanceResult{Success: false, StatusCode: 403, Error: "forbidden"}
snapshot := fetcher.Fetch(context.Background(), 6)
require.False(t, snapshot.Success)
require.True(t, snapshot.CredentialInvalid)
}
func TestQuotaFetcher_NilDependenciesProduceErrorSnapshots(t *testing.T) {
// fetcher 本体为 nil:直接降级为错误快照,不 panic。
var nilFetcher *ChannelMonitorQuotaFetcher
snapshot := nilFetcher.Fetch(context.Background(), 1)
require.False(t, snapshot.Success)
require.Equal(t, "quota fetcher is not configured", snapshot.Error)
// 数据源缺失:账号能加载,但对应服务未注入。
fetcher, _, _, _, accounts := newQuotaFetcherTestSetup(t)
fetcher.usage = nil
accounts.accounts[2] = &Account{ID: 2, Platform: domain.PlatformOpenAI}
snapshot = fetcher.Fetch(context.Background(), 2)
require.False(t, snapshot.Success)
require.Contains(t, snapshot.Error, "not configured")
}
// --- TTL 缓存 ---
func TestQuotaFetcher_CachesSuccessSnapshotPerAccount(t *testing.T) {
fetcher, usage, _, _, accounts := newQuotaFetcherTestSetup(t)
accounts.accounts[8] = &Account{ID: 8, Platform: domain.PlatformOpenAI}
usage.usage = &UsageInfo{FiveHour: &UsageProgress{Utilization: 10}}
for i := 0; i < 3; i++ {
snapshot := fetcher.Fetch(context.Background(), 8)
require.True(t, snapshot.Success)
}
require.Equal(t, 1, usage.calls, "success snapshots should be served from cache")
// 缓存过期后重新拉取。
fetcher.mu.Lock()
entry := fetcher.cache[8]
entry.expiry = time.Now().Add(-time.Second)
fetcher.cache[8] = entry
fetcher.mu.Unlock()
_ = fetcher.Fetch(context.Background(), 8)
require.Equal(t, 2, usage.calls)
}
func TestQuotaFetcher_DoesNotCacheFailures(t *testing.T) {
fetcher, usage, _, _, accounts := newQuotaFetcherTestSetup(t)
accounts.accounts[4] = &Account{ID: 4, Platform: domain.PlatformOpenAI}
usage.err = errors.New("boom")
_ = fetcher.Fetch(context.Background(), 4)
_ = fetcher.Fetch(context.Background(), 4)
require.Equal(t, 2, usage.calls, "failed snapshots must not be cached")
}
// --- UsageInfo → tiers 归一 ---
func TestUsageQuotaTiers_MapsAllWindowKinds(t *testing.T) {
limit := int64(1000)
remaining := int64(400)
resetUnix := int64(1777283883)
usage := &UsageInfo{
FiveHour: &UsageProgress{Utilization: 50},
SevenDay: &UsageProgress{Utilization: 60},
SevenDaySonnet: &UsageProgress{Utilization: 70},
SevenDayFable: &UsageProgress{Utilization: 80},
ThirtyDay: &UsageProgress{Utilization: 20},
GeminiSharedDaily: &UsageProgress{Utilization: 11},
GeminiProDaily: &UsageProgress{Utilization: 22},
GeminiFlashDaily: &UsageProgress{Utilization: 33},
GrokRequestQuota: &xai.QuotaWindow{Limit: &limit, Remaining: &remaining, ResetUnix: &resetUnix},
GrokTokenQuota: &xai.QuotaWindow{Limit: &limit, Remaining: &remaining, ResetAt: "2026-08-19T00:00:00Z"},
AntigravityQuota: map[string]*AntigravityModelQuota{
"gemini-3-pro": {Utilization: 45},
"gemini-3-flash": {Utilization: 55},
},
}
tiers := usageQuotaTiers(usage)
// 5h/7d/7d-sonnet/7d-fable/30d + gemini×3 + grok×2 + antigravity×2
require.Len(t, tiers, 12)
byKey := make(map[string]domain.MonitorQuotaTier, len(tiers))
for _, tier := range tiers {
key := tier.Window
if tier.Label != "" {
key = tier.Window + "/" + tier.Label
}
byKey[key] = tier
}
require.Contains(t, byKey, "5h")
require.Contains(t, byKey, "7d")
require.Contains(t, byKey, "7d-sonnet")
require.Contains(t, byKey, "7d-fable")
require.Contains(t, byKey, "30d")
require.Contains(t, byKey, "daily/shared")
require.Contains(t, byKey, "daily/pro")
require.Contains(t, byKey, "daily/flash")
require.Contains(t, byKey, "daily/requests")
require.Contains(t, byKey, "daily/tokens")
require.Contains(t, byKey, "total/gemini-3-pro")
require.Contains(t, byKey, "total/gemini-3-flash")
// grok requests 窗口:used = limit - remaining,百分比 60%。
requests := byKey["daily/requests"]
require.Equal(t, float64(600), requests.Used)
require.Equal(t, float64(1000), requests.Limit)
require.InDelta(t, 60.0, requests.UsedPercent, 0.001)
require.NotEmpty(t, requests.ResetAt, "ResetUnix should fall back to RFC3339")
tokens := byKey["daily/tokens"]
require.Equal(t, "2026-08-19T00:00:00Z", tokens.ResetAt)
}
func TestUsageQuotaTiers_NilAndEmptyInputs(t *testing.T) {
require.Nil(t, usageQuotaTiers(nil))
require.Nil(t, usageQuotaTiers(&UsageInfo{}))
// Grok 窗口 limit<=0 时跳过,避免除零。
var zero int64
tiers := usageQuotaTiers(&UsageInfo{
GrokRequestQuota: &xai.QuotaWindow{Limit: &zero, Remaining: &zero},
})
require.Nil(t, tiers)
}
// --- 状态推导 ---
func TestDeriveQuotaCheckResult_StatusMatrix(t *testing.T) {
now := time.Now()
healthy := &domain.MonitorQuotaSnapshot{Success: true, Tiers: []domain.MonitorQuotaTier{{Window: "5h", UsedPercent: 40}}}
res := deriveQuotaCheckResult(healthy, "quota", now)
require.Equal(t, MonitorStatusOperational, res.Status)
require.Equal(t, "quota", res.Model)
require.Empty(t, res.Message)
highUsage := &domain.MonitorQuotaSnapshot{Success: true, Tiers: []domain.MonitorQuotaTier{
{Window: "5h", UsedPercent: 30},
{Window: "daily", Label: "pro", UsedPercent: 95},
}}
res = deriveQuotaCheckResult(highUsage, "quota", now)
require.Equal(t, MonitorStatusDegraded, res.Status)
require.Contains(t, res.Message, "pro/daily")
require.Contains(t, res.Message, "95.0%")
balance := -0.5
depleted := &domain.MonitorQuotaSnapshot{Success: true, Balance: &balance, Currency: "CNY"}
res = deriveQuotaCheckResult(depleted, "quota", now)
require.Equal(t, MonitorStatusDegraded, res.Status)
require.Contains(t, res.Message, "balance depleted")
invalid := &domain.MonitorQuotaSnapshot{Success: false, CredentialInvalid: true, Error: "401 unauthorized"}
res = deriveQuotaCheckResult(invalid, "quota", now)
require.Equal(t, MonitorStatusFailed, res.Status)
unlinked := &domain.MonitorQuotaSnapshot{Success: false, Error: "linked account not found"}
res = deriveQuotaCheckResult(unlinked, "quota", now)
require.Equal(t, MonitorStatusDegraded, res.Status)
other := &domain.MonitorQuotaSnapshot{Success: false, Error: "connection refused"}
res = deriveQuotaCheckResult(other, "quota", now)
require.Equal(t, MonitorStatusError, res.Status)
require.Equal(t, "connection refused", res.Message)
res = deriveQuotaCheckResult(nil, "quota", now)
require.Equal(t, MonitorStatusError, res.Status)
}
@@ -0,0 +1,463 @@
//go:build unit
package service
import (
"context"
"testing"
"time"
"github.com/Wei-Shaw/sub2api/internal/domain"
"github.com/stretchr/testify/require"
)
// --- repo / fetcher 装配 ---
// quotaModeRepoStub 记录 RunCheck 落库行为(历史行 + MarkChecked)。
type quotaModeRepoStub struct {
ChannelMonitorRepository
monitor *ChannelMonitor
history []*ChannelMonitorHistoryRow
markedIDs []int64
updated []*ChannelMonitor
}
func (r *quotaModeRepoStub) GetByID(_ context.Context, id int64) (*ChannelMonitor, error) {
if r.monitor == nil || r.monitor.ID != id {
return nil, ErrChannelMonitorNotFound
}
clone := *r.monitor
return &clone, nil
}
func (r *quotaModeRepoStub) InsertHistoryBatch(_ context.Context, rows []*ChannelMonitorHistoryRow) error {
r.history = append(r.history, rows...)
return nil
}
func (r *quotaModeRepoStub) MarkChecked(_ context.Context, id int64, _ time.Time) error {
r.markedIDs = append(r.markedIDs, id)
return nil
}
func (r *quotaModeRepoStub) Update(_ context.Context, m *ChannelMonitor) error {
clone := *m
r.updated = append(r.updated, &clone)
return nil
}
// newQuotaModeService 构造启用 V1 探活的 service(复用 retirement/duplicate 测试的 stub)。
func newQuotaModeService(repo *quotaModeRepoStub) *ChannelMonitorService {
svc := NewChannelMonitorService(repo, &duplicateChannelMonitorEncryptor{})
svc.SetRuntimeReader(channelMonitorRuntimeStub{rt: ChannelMonitorRuntime{
Enabled: true,
Mode: ChannelMonitorModeV1,
}})
return svc
}
func newQuotaModeFetcher(accounts map[int64]*Account, usage *stubMonitorUsageSource) *ChannelMonitorQuotaFetcher {
if accounts == nil {
accounts = make(map[int64]*Account)
}
if usage == nil {
usage = &stubMonitorUsageSource{}
}
return &ChannelMonitorQuotaFetcher{
usage: usage,
accounts: &stubMonitorAccountSource{accounts: accounts},
cache: make(map[int64]monitorQuotaCacheEntry),
}
}
// --- RunCheck 分派 ---
func TestRunCheck_QuotaModeProducesSingleQuotaResult(t *testing.T) {
repo := &quotaModeRepoStub{monitor: &ChannelMonitor{
ID: 1,
Name: "kimi-quota",
Provider: MonitorProviderKimi,
APIMode: MonitorAPIModeChatCompletions,
PrimaryModel: "quota",
Enabled: true,
IntervalSeconds: 60,
CheckMode: MonitorCheckModeQuota,
AccountID: int64Ptr(9),
}}
svc := newQuotaModeService(repo)
fetcher := newQuotaModeFetcher(map[int64]*Account{
9: {ID: 9, Platform: domain.PlatformKimi, Credentials: map[string]any{"account_mode": AccountModeCoding}},
}, nil)
fetcher.cnQuota = &stubMonitorCNQuotaSource{result: &CNProviderQuotaProbeResult{
Success: true,
CredentialValid: true,
Tiers: []CNQuotaTier{{Window: "5h", UsedPercent: 30}},
}}
svc.SetQuotaFetcher(fetcher)
results, err := svc.RunCheck(context.Background(), 1)
require.NoError(t, err)
require.Len(t, results, 1)
res := results[0]
require.Equal(t, "quota", res.Model)
require.Equal(t, MonitorStatusOperational, res.Status)
require.Nil(t, res.LatencyMs)
require.Nil(t, res.PingLatencyMs)
require.NotNil(t, res.Quota)
require.True(t, res.Quota.Success)
require.Equal(t, "cn_quota", res.Quota.Source)
// 历史行携带配额快照,并推进 last_checked_at。
require.Len(t, repo.history, 1)
require.Equal(t, "quota", repo.history[0].Model)
require.NotNil(t, repo.history[0].Quota)
require.Equal(t, []int64{1}, repo.markedIDs)
}
func TestRunCheck_QuotaModeUnlinkedAccountDegrades(t *testing.T) {
repo := &quotaModeRepoStub{monitor: &ChannelMonitor{
ID: 2,
Provider: MonitorProviderDeepseek,
APIMode: MonitorAPIModeChatCompletions,
Endpoint: "",
PrimaryModel: "quota",
Enabled: true,
IntervalSeconds: 60,
CheckMode: MonitorCheckModeQuota,
AccountID: nil, // FK ON DELETE SET NULL 后的形态
}}
svc := newQuotaModeService(repo)
svc.SetQuotaFetcher(newQuotaModeFetcher(nil, nil))
results, err := svc.RunCheck(context.Background(), 2)
require.NoError(t, err)
require.Len(t, results, 1)
require.Equal(t, MonitorStatusDegraded, results[0].Status)
require.Contains(t, results[0].Message, "linked account not found")
require.False(t, results[0].Quota.Success)
}
func TestRunCheck_QuotaModeNilFetcherFailsClosed(t *testing.T) {
repo := &quotaModeRepoStub{monitor: &ChannelMonitor{
ID: 3,
Provider: MonitorProviderZhipu,
APIMode: MonitorAPIModeChatCompletions,
PrimaryModel: "quota",
Enabled: true,
IntervalSeconds: 60,
CheckMode: MonitorCheckModeQuota,
AccountID: int64Ptr(5),
}}
svc := newQuotaModeService(repo) // 不注入 fetcher
results, err := svc.RunCheck(context.Background(), 3)
require.NoError(t, err)
require.Len(t, results, 1)
require.Equal(t, MonitorStatusError, results[0].Status)
require.Contains(t, results[0].Message, "not configured")
}
func TestRunCheck_QuotaProbeAttachesSnapshotToPrimaryRowOnly(t *testing.T) {
h := &openAICaptureHandler{}
endpoint := setupFakeOpenAI(t, h)
repo := &quotaModeRepoStub{monitor: &ChannelMonitor{
ID: 4,
Provider: MonitorProviderOpenAI,
APIMode: MonitorAPIModeChatCompletions,
Endpoint: endpoint,
APIKey: "OLD:sk-openai",
PrimaryModel: "gpt-test",
ExtraModels: []string{"gpt-extra"},
Enabled: true,
IntervalSeconds: 60,
CheckMode: MonitorCheckModeQuotaProbe,
AccountID: int64Ptr(12),
}}
svc := newQuotaModeService(repo)
usage := &stubMonitorUsageSource{usage: &UsageInfo{
FiveHour: &UsageProgress{Utilization: 20},
}}
svc.SetQuotaFetcher(newQuotaModeFetcher(map[int64]*Account{
12: {ID: 12, Platform: domain.PlatformOpenAI},
}, usage))
results, err := svc.RunCheck(context.Background(), 4)
require.NoError(t, err)
require.Len(t, results, 2)
// 探活状态为准,配额只挂主模型行。
require.Equal(t, MonitorStatusOperational, results[0].Status)
require.NotNil(t, results[0].Quota)
require.True(t, results[0].Quota.Success)
require.Equal(t, "usage", results[0].Quota.Source)
require.Nil(t, results[1].Quota, "extra model rows must not carry quota")
// 历史落库时同样只有主模型行带快照。
require.Len(t, repo.history, 2)
require.NotNil(t, repo.history[0].Quota)
require.Equal(t, "gpt-test", repo.history[0].Model)
require.Nil(t, repo.history[1].Quota)
}
func TestRunCheck_QuotaProbeQuotaFailureKeepsProbeStatus(t *testing.T) {
h := &openAICaptureHandler{}
endpoint := setupFakeOpenAI(t, h)
repo := &quotaModeRepoStub{monitor: &ChannelMonitor{
ID: 5,
Provider: MonitorProviderOpenAI,
APIMode: MonitorAPIModeChatCompletions,
Endpoint: endpoint,
APIKey: "OLD:sk-openai",
PrimaryModel: "gpt-test",
Enabled: true,
IntervalSeconds: 60,
CheckMode: MonitorCheckModeQuotaProbe,
AccountID: nil, // 配额侧失效
}}
svc := newQuotaModeService(repo)
svc.SetQuotaFetcher(newQuotaModeFetcher(nil, nil))
results, err := svc.RunCheck(context.Background(), 5)
require.NoError(t, err)
require.Len(t, results, 1)
require.Equal(t, MonitorStatusOperational, results[0].Status, "quota failure must not flip probe status")
require.False(t, results[0].Quota.Success)
}
// --- attachQuotaSnapshot 细节 ---
func TestAttachQuotaSnapshot_NoteOnlyWhenProbeMessageEmpty(t *testing.T) {
results := []*CheckResult{
{Model: "primary", Status: MonitorStatusOperational, Message: "challenge passed"},
{Model: "extra"},
}
failed := &domain.MonitorQuotaSnapshot{Success: false, Error: "boom"}
attachQuotaSnapshot(results, failed)
require.Equal(t, "challenge passed", results[0].Message, "existing probe message wins")
require.Equal(t, failed, results[0].Quota)
require.Nil(t, results[1].Quota)
quiet := []*CheckResult{{Model: "primary", Status: MonitorStatusOperational}}
attachQuotaSnapshot(quiet, failed)
require.Contains(t, quiet[0].Message, "quota fetch failed: boom")
attachQuotaSnapshot(nil, failed) // 空结果不 panic
attachQuotaSnapshot(results, nil) // 空快照不动结果
}
// --- 校验矩阵 ---
func TestValidateCreateParams_CheckModeMatrix(t *testing.T) {
accountID := int64(9)
cases := []struct {
name string
params ChannelMonitorCreateParams
wantErr error
}{
{
name: "probe requires endpoint",
params: ChannelMonitorCreateParams{
Provider: MonitorProviderOpenAI, CheckMode: MonitorCheckModeProbe,
APIKey: "sk", IntervalSeconds: 60, PrimaryModel: "gpt-5",
},
wantErr: ErrChannelMonitorInvalidEndpoint,
},
{
name: "probe requires api key",
params: ChannelMonitorCreateParams{
Provider: MonitorProviderOpenAI, CheckMode: MonitorCheckModeProbe,
Endpoint: "https://api.openai.com", IntervalSeconds: 60, PrimaryModel: "gpt-5",
},
wantErr: ErrChannelMonitorMissingAPIKey,
},
{
name: "quota drops endpoint and api key requirements",
params: ChannelMonitorCreateParams{
Provider: MonitorProviderAntigravity, CheckMode: MonitorCheckModeQuota,
IntervalSeconds: 60, AccountID: &accountID,
},
wantErr: nil, // primary_model 默认 "quota"
},
{
name: "quota requires account",
params: ChannelMonitorCreateParams{
Provider: MonitorProviderKimi, CheckMode: MonitorCheckModeQuota,
IntervalSeconds: 60, PrimaryModel: "quota",
},
wantErr: ErrChannelMonitorAccountRequired,
},
{
name: "quota_probe requires endpoint and api key too",
params: ChannelMonitorCreateParams{
Provider: MonitorProviderKimi, CheckMode: MonitorCheckModeQuotaProbe,
IntervalSeconds: 60, AccountID: &accountID, PrimaryModel: "kimi-k2",
},
wantErr: ErrChannelMonitorInvalidEndpoint,
},
{
name: "antigravity probe unsupported",
params: ChannelMonitorCreateParams{
Provider: MonitorProviderAntigravity, CheckMode: MonitorCheckModeProbe,
Endpoint: "https://example.com", APIKey: "k",
IntervalSeconds: 60, AccountID: &accountID, PrimaryModel: "gemini-3-pro",
},
wantErr: ErrChannelMonitorInvalidCheckMode,
},
{
name: "antigravity quota_probe unsupported",
params: ChannelMonitorCreateParams{
Provider: MonitorProviderAntigravity, CheckMode: MonitorCheckModeQuotaProbe,
Endpoint: "https://example.com", APIKey: "k",
IntervalSeconds: 60, AccountID: &accountID, PrimaryModel: "gemini-3-pro",
},
wantErr: ErrChannelMonitorInvalidCheckMode,
},
{
name: "unknown mode rejected",
params: ChannelMonitorCreateParams{
Provider: MonitorProviderOpenAI, CheckMode: "auto",
Endpoint: "https://api.openai.com", APIKey: "sk",
IntervalSeconds: 60, PrimaryModel: "gpt-5",
},
wantErr: ErrChannelMonitorInvalidCheckMode,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
err := validateCreateParams(tc.params)
if tc.wantErr == nil {
require.NoError(t, err)
} else {
require.ErrorIs(t, err, tc.wantErr)
}
})
}
}
func TestNormalizeMonitorPrimaryModel_QuotaDefault(t *testing.T) {
require.Equal(t, "quota", normalizeMonitorPrimaryModel(MonitorProviderKimi, MonitorCheckModeQuota, ""))
require.Equal(t, "quota", normalizeMonitorPrimaryModel(MonitorProviderAntigravity, MonitorCheckModeQuota, " "))
// 探活模式沿用原语义:grok 默认模型,其余必填(空串报错在 validateCreateParams)。
require.Equal(t, MonitorDefaultGrokModel, normalizeMonitorPrimaryModel(MonitorProviderGrok, MonitorCheckModeProbe, ""))
require.Equal(t, "kimi-k2", normalizeMonitorPrimaryModel(MonitorProviderKimi, MonitorCheckModeQuotaProbe, "kimi-k2"))
}
func TestProviderProbeCapabilityMatrix(t *testing.T) {
require.False(t, providerSupportsProbe(MonitorProviderAntigravity))
for _, p := range []string{
MonitorProviderOpenAI, MonitorProviderAnthropic, MonitorProviderGemini,
MonitorProviderGrok, MonitorProviderKimi, MonitorProviderZhipu, MonitorProviderDeepseek,
} {
require.True(t, providerSupportsProbe(p), p)
}
for _, p := range []string{
MonitorProviderOpenAI, MonitorProviderAnthropic, MonitorProviderGemini,
MonitorProviderGrok, MonitorProviderAntigravity,
MonitorProviderKimi, MonitorProviderZhipu, MonitorProviderDeepseek,
} {
require.NoError(t, validateProvider(p), p)
}
}
// --- 关联账号校验 ---
func TestValidateLinkedAccount_Matrix(t *testing.T) {
svc := NewChannelMonitorService(nil, nil)
fetcher := newQuotaModeFetcher(map[int64]*Account{
1: {ID: 1, Platform: domain.PlatformKimi},
}, nil)
svc.SetQuotaFetcher(fetcher)
require.NoError(t, svc.validateLinkedAccount(context.Background(), MonitorProviderKimi, nil))
require.NoError(t, svc.validateLinkedAccount(context.Background(), MonitorProviderKimi, int64Ptr(0)))
require.NoError(t, svc.validateLinkedAccount(context.Background(), MonitorProviderKimi, int64Ptr(1)))
require.ErrorIs(t, svc.validateLinkedAccount(context.Background(), MonitorProviderZhipu, int64Ptr(1)), ErrChannelMonitorProviderIncompatible)
require.ErrorIs(t, svc.validateLinkedAccount(context.Background(), MonitorProviderKimi, int64Ptr(404)), ErrChannelMonitorAccountRequired)
noFetcher := NewChannelMonitorService(nil, nil)
require.ErrorIs(t, noFetcher.validateLinkedAccount(context.Background(), MonitorProviderKimi, int64Ptr(1)), ErrChannelMonitorAccountRequired)
}
func TestRevalidateLinkedAccount_QuotaErrorsProbeUnbinds(t *testing.T) {
fetcher := newQuotaModeFetcher(nil, nil) // 账号一律加载失败
svc := NewChannelMonitorService(nil, nil)
svc.SetQuotaFetcher(fetcher)
quota := &ChannelMonitor{Provider: MonitorProviderKimi, CheckMode: MonitorCheckModeQuota, AccountID: int64Ptr(9)}
require.ErrorIs(t, svc.revalidateLinkedAccount(context.Background(), quota), ErrChannelMonitorAccountRequired)
require.NotNil(t, quota.AccountID)
probe := &ChannelMonitor{Provider: MonitorProviderKimi, CheckMode: MonitorCheckModeProbe, AccountID: int64Ptr(9)}
require.NoError(t, svc.revalidateLinkedAccount(context.Background(), probe))
require.Nil(t, probe.AccountID, "probe mode should silently unbind stale account")
quotaNoAccount := &ChannelMonitor{Provider: MonitorProviderKimi, CheckMode: MonitorCheckModeQuota}
require.ErrorIs(t, svc.revalidateLinkedAccount(context.Background(), quotaNoAccount), ErrChannelMonitorAccountRequired)
}
func TestRevalidateLinkedAccount_PlatformMismatch(t *testing.T) {
svc := NewChannelMonitorService(nil, nil)
svc.SetQuotaFetcher(newQuotaModeFetcher(map[int64]*Account{
2: {ID: 2, Platform: domain.PlatformDeepseek},
}, nil))
quota := &ChannelMonitor{Provider: MonitorProviderKimi, CheckMode: MonitorCheckModeQuota, AccountID: int64Ptr(2)}
require.ErrorIs(t, svc.revalidateLinkedAccount(context.Background(), quota), ErrChannelMonitorProviderIncompatible)
probe := &ChannelMonitor{Provider: MonitorProviderKimi, CheckMode: MonitorCheckModeProbe, AccountID: int64Ptr(2)}
require.NoError(t, svc.revalidateLinkedAccount(context.Background(), probe))
require.Nil(t, probe.AccountID)
}
// --- quota → probe 切换的 key 管控(validateProbeAPIKey) ---
func TestValidateProbeAPIKey_QuotaToProbeRequiresFreshKey(t *testing.T) {
svc := NewChannelMonitorService(nil, &duplicateChannelMonitorEncryptor{})
quota := &ChannelMonitor{Provider: MonitorProviderKimi, CheckMode: MonitorCheckModeQuota, APIKey: "NEW:"}
require.NoError(t, svc.validateProbeAPIKey(quota, "")) // quota 模式不管 key
quota.CheckMode = MonitorCheckModeProbe
// 存量密文解出空明文(quota 监控存的加密空串)→ 必须重填 key。
require.ErrorIs(t, svc.validateProbeAPIKey(quota, ""), ErrChannelMonitorMissingAPIKey)
// 提供新明文 key → 放行。
require.NoError(t, svc.validateProbeAPIKey(quota, "sk-fresh"))
// 密文解出非空明文 → 放行。
require.NoError(t, svc.validateProbeAPIKey(
&ChannelMonitor{Provider: MonitorProviderKimi, CheckMode: MonitorCheckModeProbe, APIKey: "OLD:sk-live"}, ""))
}
// --- Duplicate:quota 模式空明文重加密 ---
func TestDuplicateChannelMonitorQuotaModeReencryptsEmptyKey(t *testing.T) {
accountID := int64(9)
source := &ChannelMonitor{
ID: 42,
Name: "kimi-quota",
Provider: MonitorProviderKimi,
APIMode: MonitorAPIModeChatCompletions,
Endpoint: "",
APIKey: "OLD:", // 解密为空串(quota 监控的加密空 key)
PrimaryModel: "quota",
Enabled: true,
IntervalSeconds: 60,
CheckMode: MonitorCheckModeQuota,
AccountID: &accountID,
}
repo := &duplicateChannelMonitorRepoStub{source: source}
service := NewChannelMonitorService(repo, &duplicateChannelMonitorEncryptor{})
dup, err := service.Duplicate(context.Background(), 42, 7, "admin:7", "op-1")
require.NoError(t, err)
require.Equal(t, MonitorCheckModeQuota, dup.CheckMode)
require.NotNil(t, dup.AccountID)
require.Equal(t, accountID, *dup.AccountID)
require.Empty(t, dup.APIKey, "plaintext stays empty for quota monitors")
require.Len(t, repo.created, 1)
require.Equal(t, "NEW:", repo.created[0].APIKey, "empty key must be re-encrypted, not dropped")
}
@@ -9,6 +9,7 @@ func TestApplyMonitorUpdate_ProviderOnlySwitchToGrokUsesDefaultModel(t *testing.
existing := &ChannelMonitor{
Provider: MonitorProviderOpenAI,
APIMode: MonitorAPIModeResponses,
Endpoint: "https://api.openai.com/v1",
PrimaryModel: "gpt-5",
IntervalSeconds: 60,
}
@@ -31,6 +32,7 @@ func TestApplyMonitorUpdate_SwitchToGrokPreservesExplicitModel(t *testing.T) {
existing := &ChannelMonitor{
Provider: MonitorProviderOpenAI,
APIMode: MonitorAPIModeChatCompletions,
Endpoint: "https://api.openai.com/v1",
PrimaryModel: "gpt-5",
IntervalSeconds: 60,
}
@@ -52,6 +54,7 @@ func TestApplyMonitorUpdate_SameGrokProviderDoesNotResetExistingModel(t *testing
existing := &ChannelMonitor{
Provider: MonitorProviderGrok,
APIMode: MonitorAPIModeChatCompletions,
Endpoint: "https://api.x.ai",
PrimaryModel: "grok-4.3",
IntervalSeconds: 60,
}
@@ -116,6 +116,29 @@ func TestSettingService_ChannelMonitorHideThroughputDefaultsToPrivate(t *testing
}
}
func TestSettingService_ChannelMonitorShowQuotaFailsClosed(t *testing.T) {
// 缺省(迁移插入 'false' / 老库无行)一律不展示。
missingRuntime := NewSettingService(&settingPublicRepoStub{values: map[string]string{}}, &config.Config{}).GetChannelMonitorRuntime(context.Background())
require.False(t, missingRuntime.ShowQuota)
missingPublic, err := NewSettingService(&settingPublicRepoStub{values: map[string]string{}}, &config.Config{}).
GetPublicSettings(context.Background())
require.NoError(t, err)
require.False(t, missingPublic.ChannelMonitorShowQuota)
// 仅字面 "true" 视为开启;其余值(含异常值)fail-closed。
runtime := NewSettingService(&settingPublicRepoStub{values: map[string]string{
SettingKeyChannelMonitorShowQuota: "true",
}}, &config.Config{}).GetChannelMonitorRuntime(context.Background())
require.True(t, runtime.ShowQuota)
for _, value := range []string{"false", "TRUE", "1", "yes", "on", "garbage"} {
rt := NewSettingService(&settingPublicRepoStub{values: map[string]string{
SettingKeyChannelMonitorShowQuota: value,
}}, &config.Config{}).GetChannelMonitorRuntime(context.Background())
require.False(t, rt.ShowQuota, "value=%q", value)
}
}
func TestSettingService_GetPublicSettings_ExposesForceEmailOnThirdPartySignup(t *testing.T) {
repo := &settingPublicRepoStub{
values: map[string]string{
@@ -0,0 +1,37 @@
package migrations
import (
"strings"
"testing"
"github.com/stretchr/testify/require"
)
func TestChannelMonitorQuotaModeMigration(t *testing.T) {
content, err := FS.ReadFile("226_channel_monitor_quota_mode.sql")
require.NoError(t, err)
sql := strings.Join(strings.Fields(string(content)), " ")
// provider CHECK 两张表扩到 8 平台,且带幂等守卫(仿 176 grok 迁移)。
require.Contains(t, sql, "channel_monitors_provider_check")
require.Contains(t, sql, "channel_monitor_request_templates_provider_check")
require.Contains(t, sql, "CHECK (provider IN ('openai', 'anthropic', 'gemini', 'grok', 'antigravity', 'kimi', 'zhipu', 'deepseek'))")
require.Contains(t, sql, "position('kimi' IN monitor_constraint_def) = 0")
require.Contains(t, sql, "position('kimi' IN template_constraint_def) = 0")
// check_mode 三态,默认 probe。
require.Contains(t, sql, "ADD COLUMN IF NOT EXISTS check_mode VARCHAR(32) NOT NULL DEFAULT 'probe'")
require.Contains(t, sql, "CHECK (check_mode IN ('probe', 'quota', 'quota_probe'))")
// account_id 关联账号,账号删除置空(监控保留,运行时报「账号未关联」)。
require.Contains(t, sql, "ADD COLUMN IF NOT EXISTS account_id BIGINT REFERENCES accounts(id) ON DELETE SET NULL")
require.Contains(t, sql, "CREATE INDEX IF NOT EXISTS idx_channel_monitors_account_id ON channel_monitors(account_id)")
// 历史表配额快照列。
require.Contains(t, sql, "ADD COLUMN IF NOT EXISTS quota JSONB")
// 公开设置默认关闭。
require.Contains(t, sql, "VALUES ('channel_monitor_show_quota', 'false')")
require.Contains(t, sql, "ON CONFLICT (key) DO NOTHING")
}