gateway_multiplatform_test.go 113 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685268626872688268926902691269226932694269526962697269826992700270127022703270427052706270727082709271027112712271327142715271627172718271927202721272227232724272527262727272827292730273127322733273427352736273727382739274027412742274327442745274627472748274927502751275227532754275527562757275827592760276127622763276427652766276727682769277027712772277327742775277627772778277927802781278227832784278527862787278827892790279127922793279427952796279727982799280028012802280328042805280628072808280928102811281228132814281528162817281828192820282128222823282428252826282728282829283028312832283328342835283628372838283928402841284228432844284528462847284828492850285128522853285428552856285728582859286028612862286328642865286628672868286928702871287228732874287528762877287828792880288128822883288428852886288728882889289028912892289328942895289628972898289929002901290229032904290529062907290829092910291129122913291429152916291729182919292029212922292329242925292629272928292929302931293229332934293529362937293829392940294129422943294429452946294729482949295029512952295329542955295629572958295929602961296229632964296529662967296829692970297129722973297429752976297729782979298029812982298329842985298629872988298929902991299229932994299529962997299829993000300130023003300430053006300730083009301030113012301330143015301630173018301930203021302230233024302530263027302830293030303130323033303430353036303730383039304030413042304330443045304630473048304930503051305230533054305530563057305830593060306130623063306430653066306730683069307030713072307330743075307630773078307930803081308230833084308530863087308830893090309130923093309430953096309730983099310031013102310331043105310631073108310931103111311231133114311531163117311831193120312131223123312431253126312731283129313031313132313331343135313631373138313931403141314231433144314531463147314831493150315131523153315431553156315731583159316031613162316331643165316631673168316931703171317231733174317531763177317831793180318131823183318431853186318731883189319031913192319331943195319631973198319932003201320232033204320532063207320832093210321132123213321432153216321732183219322032213222322332243225322632273228322932303231323232333234323532363237323832393240324132423243324432453246324732483249325032513252325332543255325632573258325932603261326232633264326532663267326832693270327132723273327432753276327732783279328032813282328332843285328632873288328932903291329232933294329532963297329832993300330133023303330433053306330733083309331033113312331333143315331633173318331933203321332233233324332533263327332833293330333133323333333433353336333733383339334033413342334333443345334633473348334933503351335233533354335533563357335833593360336133623363336433653366336733683369337033713372337333743375337633773378337933803381338233833384338533863387338833893390339133923393339433953396339733983399340034013402340334043405340634073408340934103411341234133414341534163417341834193420342134223423342434253426342734283429343034313432343334343435343634373438343934403441344234433444344534463447344834493450345134523453345434553456345734583459346034613462346334643465346634673468346934703471347234733474347534763477347834793480348134823483348434853486348734883489349034913492349334943495349634973498349935003501350235033504350535063507350835093510351135123513351435153516351735183519352035213522352335243525352635273528352935303531353235333534353535363537353835393540354135423543354435453546354735483549355035513552355335543555355635573558355935603561356235633564356535663567356835693570357135723573357435753576357735783579358035813582358335843585358635873588
  1. //go:build unit
  2. package service
  3. import (
  4. "context"
  5. "errors"
  6. "testing"
  7. "time"
  8. "github.com/Wei-Shaw/sub2api/internal/config"
  9. "github.com/Wei-Shaw/sub2api/internal/pkg/ctxkey"
  10. "github.com/Wei-Shaw/sub2api/internal/pkg/pagination"
  11. "github.com/stretchr/testify/require"
  12. )
  13. // testConfig 返回一个用于测试的默认配置
  14. func testConfig() *config.Config {
  15. return &config.Config{RunMode: config.RunModeStandard}
  16. }
  17. // mockAccountRepoForPlatform 单平台测试用的 mock
  18. type mockAccountRepoForPlatform struct {
  19. accounts []Account
  20. accountsByID map[int64]*Account
  21. listPlatformFunc func(ctx context.Context, platform string) ([]Account, error)
  22. getByIDCalls int
  23. }
  24. func (m *mockAccountRepoForPlatform) GetByID(ctx context.Context, id int64) (*Account, error) {
  25. m.getByIDCalls++
  26. if acc, ok := m.accountsByID[id]; ok {
  27. return acc, nil
  28. }
  29. return nil, errors.New("account not found")
  30. }
  31. func (m *mockAccountRepoForPlatform) GetByIDs(ctx context.Context, ids []int64) ([]*Account, error) {
  32. var result []*Account
  33. for _, id := range ids {
  34. if acc, ok := m.accountsByID[id]; ok {
  35. result = append(result, acc)
  36. }
  37. }
  38. return result, nil
  39. }
  40. func (m *mockAccountRepoForPlatform) ExistsByID(ctx context.Context, id int64) (bool, error) {
  41. if m.accountsByID == nil {
  42. return false, nil
  43. }
  44. _, ok := m.accountsByID[id]
  45. return ok, nil
  46. }
  47. func (m *mockAccountRepoForPlatform) ListSchedulableByPlatform(ctx context.Context, platform string) ([]Account, error) {
  48. if m.listPlatformFunc != nil {
  49. return m.listPlatformFunc(ctx, platform)
  50. }
  51. var result []Account
  52. for _, acc := range m.accounts {
  53. if acc.Platform == platform && acc.IsSchedulable() {
  54. result = append(result, acc)
  55. }
  56. }
  57. return result, nil
  58. }
  59. func (m *mockAccountRepoForPlatform) ListSchedulableByGroupIDAndPlatform(ctx context.Context, groupID int64, platform string) ([]Account, error) {
  60. return m.ListSchedulableByPlatform(ctx, platform)
  61. }
  62. // Stub methods to implement AccountRepository interface
  63. func (m *mockAccountRepoForPlatform) Create(ctx context.Context, account *Account) error {
  64. return nil
  65. }
  66. func (m *mockAccountRepoForPlatform) GetByCRSAccountID(ctx context.Context, crsAccountID string) (*Account, error) {
  67. return nil, nil
  68. }
  69. func (m *mockAccountRepoForPlatform) FindByExtraField(ctx context.Context, key string, value any) ([]Account, error) {
  70. return nil, nil
  71. }
  72. func (m *mockAccountRepoForPlatform) ListCRSAccountIDs(ctx context.Context) (map[string]int64, error) {
  73. return nil, nil
  74. }
  75. func (m *mockAccountRepoForPlatform) Update(ctx context.Context, account *Account) error {
  76. return nil
  77. }
  78. func (m *mockAccountRepoForPlatform) Delete(ctx context.Context, id int64) error { return nil }
  79. func (m *mockAccountRepoForPlatform) List(ctx context.Context, params pagination.PaginationParams) ([]Account, *pagination.PaginationResult, error) {
  80. return nil, nil, nil
  81. }
  82. func (m *mockAccountRepoForPlatform) ListWithFilters(ctx context.Context, params pagination.PaginationParams, platform, accountType, status, search string, groupID int64, privacyMode string) ([]Account, *pagination.PaginationResult, error) {
  83. return nil, nil, nil
  84. }
  85. func (m *mockAccountRepoForPlatform) ListAllWithFilters(ctx context.Context, platform, accountType, status, search string, groupID int64, privacyMode string) ([]Account, error) {
  86. return nil, nil
  87. }
  88. func (m *mockAccountRepoForPlatform) ListByGroup(ctx context.Context, groupID int64) ([]Account, error) {
  89. return nil, nil
  90. }
  91. func (m *mockAccountRepoForPlatform) ListActive(ctx context.Context) ([]Account, error) {
  92. return nil, nil
  93. }
  94. func (m *mockAccountRepoForPlatform) ListByPlatform(ctx context.Context, platform string) ([]Account, error) {
  95. return nil, nil
  96. }
  97. func (m *mockAccountRepoForPlatform) UpdateLastUsed(ctx context.Context, id int64) error {
  98. return nil
  99. }
  100. func (m *mockAccountRepoForPlatform) BatchUpdateLastUsed(ctx context.Context, updates map[int64]time.Time) error {
  101. return nil
  102. }
  103. func (m *mockAccountRepoForPlatform) SetError(ctx context.Context, id int64, errorMsg string) error {
  104. return nil
  105. }
  106. func (m *mockAccountRepoForPlatform) ClearError(ctx context.Context, id int64) error {
  107. return nil
  108. }
  109. func (m *mockAccountRepoForPlatform) SetSchedulable(ctx context.Context, id int64, schedulable bool) error {
  110. return nil
  111. }
  112. func (m *mockAccountRepoForPlatform) AutoPauseExpiredAccounts(ctx context.Context, now time.Time) (int64, error) {
  113. return 0, nil
  114. }
  115. func (m *mockAccountRepoForPlatform) BindGroups(ctx context.Context, accountID int64, groupIDs []int64) error {
  116. return nil
  117. }
  118. func (m *mockAccountRepoForPlatform) ListSchedulable(ctx context.Context) ([]Account, error) {
  119. return nil, nil
  120. }
  121. func (m *mockAccountRepoForPlatform) ListSchedulableByGroupID(ctx context.Context, groupID int64) ([]Account, error) {
  122. return nil, nil
  123. }
  124. func (m *mockAccountRepoForPlatform) ListSchedulableByPlatforms(ctx context.Context, platforms []string) ([]Account, error) {
  125. var result []Account
  126. platformSet := make(map[string]bool)
  127. for _, p := range platforms {
  128. platformSet[p] = true
  129. }
  130. for _, acc := range m.accounts {
  131. if platformSet[acc.Platform] && acc.IsSchedulable() {
  132. result = append(result, acc)
  133. }
  134. }
  135. return result, nil
  136. }
  137. func (m *mockAccountRepoForPlatform) ListSchedulableByGroupIDAndPlatforms(ctx context.Context, groupID int64, platforms []string) ([]Account, error) {
  138. return m.ListSchedulableByPlatforms(ctx, platforms)
  139. }
  140. func (m *mockAccountRepoForPlatform) ListSchedulableUngroupedByPlatform(ctx context.Context, platform string) ([]Account, error) {
  141. return m.ListSchedulableByPlatform(ctx, platform)
  142. }
  143. func (m *mockAccountRepoForPlatform) ListSchedulableUngroupedByPlatforms(ctx context.Context, platforms []string) ([]Account, error) {
  144. return m.ListSchedulableByPlatforms(ctx, platforms)
  145. }
  146. func (m *mockAccountRepoForPlatform) ListModelAvailabilityCandidates(_ context.Context, groupID *int64, platforms []string, includeGrouped bool) ([]Account, error) {
  147. platformSet := make(map[string]struct{}, len(platforms))
  148. for _, platform := range platforms {
  149. platformSet[platform] = struct{}{}
  150. }
  151. result := make([]Account, 0, len(m.accounts))
  152. for _, acc := range m.accounts {
  153. if _, ok := platformSet[acc.Platform]; !ok || acc.Status != StatusActive || !acc.Schedulable {
  154. continue
  155. }
  156. if groupID != nil {
  157. inGroup := false
  158. for _, accountGroup := range acc.AccountGroups {
  159. if accountGroup.GroupID == *groupID {
  160. inGroup = true
  161. break
  162. }
  163. }
  164. if !inGroup {
  165. continue
  166. }
  167. } else if !includeGrouped && (len(acc.AccountGroups) > 0 || len(acc.GroupIDs) > 0) {
  168. continue
  169. }
  170. result = append(result, acc)
  171. }
  172. return result, nil
  173. }
  174. func (m *mockAccountRepoForPlatform) SetRateLimited(ctx context.Context, id int64, resetAt time.Time) error {
  175. return nil
  176. }
  177. func (m *mockAccountRepoForPlatform) SetModelRateLimit(ctx context.Context, id int64, scope string, resetAt time.Time, reason ...string) error {
  178. return nil
  179. }
  180. func (m *mockAccountRepoForPlatform) SetOverloaded(ctx context.Context, id int64, until time.Time) error {
  181. return nil
  182. }
  183. func (m *mockAccountRepoForPlatform) SetTempUnschedulable(ctx context.Context, id int64, until time.Time, reason string) error {
  184. return nil
  185. }
  186. func (m *mockAccountRepoForPlatform) ClearTempUnschedulable(ctx context.Context, id int64) error {
  187. return nil
  188. }
  189. func (m *mockAccountRepoForPlatform) ClearRateLimit(ctx context.Context, id int64) error {
  190. return nil
  191. }
  192. func (m *mockAccountRepoForPlatform) ClearAntigravityQuotaScopes(ctx context.Context, id int64) error {
  193. return nil
  194. }
  195. func (m *mockAccountRepoForPlatform) ClearModelRateLimits(ctx context.Context, id int64) error {
  196. return nil
  197. }
  198. func (m *mockAccountRepoForPlatform) UpdateSessionWindow(ctx context.Context, id int64, start, end *time.Time, status string) error {
  199. return nil
  200. }
  201. func (m *mockAccountRepoForPlatform) UpdateSessionWindowEnd(ctx context.Context, id int64, end time.Time) error {
  202. return nil
  203. }
  204. func (m *mockAccountRepoForPlatform) UpdateExtra(ctx context.Context, id int64, updates map[string]any) error {
  205. return nil
  206. }
  207. func (m *mockAccountRepoForPlatform) BulkUpdate(ctx context.Context, ids []int64, updates AccountBulkUpdate) (int64, error) {
  208. return 0, nil
  209. }
  210. func (m *mockAccountRepoForPlatform) IncrementQuotaUsed(ctx context.Context, id int64, amount float64) error {
  211. return nil
  212. }
  213. func (m *mockAccountRepoForPlatform) ResetQuotaUsedAndClearRateLimitCooldown(ctx context.Context, id int64) error {
  214. return nil
  215. }
  216. func (m *mockAccountRepoForPlatform) RevertProxyFallback(ctx context.Context, accountID int64) error {
  217. return nil
  218. }
  219. func (m *mockAccountRepoForPlatform) ListShadowsByParent(ctx context.Context, parentID int64) ([]*Account, error) {
  220. return nil, nil
  221. }
  222. // Verify interface implementation
  223. var _ AccountRepository = (*mockAccountRepoForPlatform)(nil)
  224. // mockGatewayCacheForPlatform 单平台测试用的 cache mock
  225. type mockGatewayCacheForPlatform struct {
  226. sessionBindings map[string]int64
  227. deletedSessions map[string]int
  228. }
  229. func (m *mockGatewayCacheForPlatform) GetSessionAccountID(ctx context.Context, groupID int64, sessionHash string) (int64, error) {
  230. if id, ok := m.sessionBindings[sessionHash]; ok {
  231. return id, nil
  232. }
  233. return 0, errors.New("not found")
  234. }
  235. func (m *mockGatewayCacheForPlatform) SetSessionAccountID(ctx context.Context, groupID int64, sessionHash string, accountID int64, ttl time.Duration) error {
  236. if m.sessionBindings == nil {
  237. m.sessionBindings = make(map[string]int64)
  238. }
  239. m.sessionBindings[sessionHash] = accountID
  240. return nil
  241. }
  242. func (m *mockGatewayCacheForPlatform) RefreshSessionTTL(ctx context.Context, groupID int64, sessionHash string, ttl time.Duration) error {
  243. return nil
  244. }
  245. func (m *mockGatewayCacheForPlatform) DeleteSessionAccountID(ctx context.Context, groupID int64, sessionHash string) error {
  246. if m.sessionBindings == nil {
  247. return nil
  248. }
  249. if m.deletedSessions == nil {
  250. m.deletedSessions = make(map[string]int)
  251. }
  252. m.deletedSessions[sessionHash]++
  253. delete(m.sessionBindings, sessionHash)
  254. return nil
  255. }
  256. func (m *mockGatewayCacheForPlatform) SetGrokVideoPendingBilling(_ context.Context, _ string, _ []byte, _ time.Duration) error {
  257. return nil
  258. }
  259. func (m *mockGatewayCacheForPlatform) GetGrokVideoPendingBilling(_ context.Context, _ string) ([]byte, error) {
  260. return nil, nil
  261. }
  262. func (m *mockGatewayCacheForPlatform) ClaimGrokVideoBilled(_ context.Context, _ string, _ time.Duration) (bool, error) {
  263. return true, nil
  264. }
  265. func (m *mockGatewayCacheForPlatform) ReleaseGrokVideoBilled(_ context.Context, _ string) error {
  266. return nil
  267. }
  268. func (m *mockGatewayCacheForPlatform) SetReasoningContent(_ context.Context, _ string, _ string, _ time.Duration) error {
  269. return nil
  270. }
  271. func (m *mockGatewayCacheForPlatform) GetReasoningContent(_ context.Context, _ string) (string, error) {
  272. return "", ErrReasoningContentNotFound
  273. }
  274. type mockGroupRepoForGateway struct {
  275. groups map[int64]*Group
  276. getByIDCalls int
  277. getByIDLiteCalls int
  278. }
  279. func (m *mockGroupRepoForGateway) GetByID(ctx context.Context, id int64) (*Group, error) {
  280. m.getByIDCalls++
  281. if g, ok := m.groups[id]; ok {
  282. return g, nil
  283. }
  284. return nil, ErrGroupNotFound
  285. }
  286. func (m *mockGroupRepoForGateway) GetByIDLite(ctx context.Context, id int64) (*Group, error) {
  287. m.getByIDLiteCalls++
  288. if g, ok := m.groups[id]; ok {
  289. return g, nil
  290. }
  291. return nil, ErrGroupNotFound
  292. }
  293. func (m *mockGroupRepoForGateway) Create(ctx context.Context, group *Group) error { return nil }
  294. func (m *mockGroupRepoForGateway) Update(ctx context.Context, group *Group) error { return nil }
  295. func (m *mockGroupRepoForGateway) Delete(ctx context.Context, id int64) error { return nil }
  296. func (m *mockGroupRepoForGateway) DeleteCascade(ctx context.Context, id int64) ([]int64, error) {
  297. return nil, nil
  298. }
  299. func (m *mockGroupRepoForGateway) List(ctx context.Context, params pagination.PaginationParams) ([]Group, *pagination.PaginationResult, error) {
  300. return nil, nil, nil
  301. }
  302. func (m *mockGroupRepoForGateway) ListWithFilters(ctx context.Context, params pagination.PaginationParams, platform, status, search string, isExclusive *bool) ([]Group, *pagination.PaginationResult, error) {
  303. return nil, nil, nil
  304. }
  305. func (m *mockGroupRepoForGateway) ListActive(ctx context.Context) ([]Group, error) {
  306. return nil, nil
  307. }
  308. func (m *mockGroupRepoForGateway) ListActiveByPlatform(ctx context.Context, platform string) ([]Group, error) {
  309. return nil, nil
  310. }
  311. func (m *mockGroupRepoForGateway) ExistsByName(ctx context.Context, name string) (bool, error) {
  312. return false, nil
  313. }
  314. func (m *mockGroupRepoForGateway) GetAccountCount(ctx context.Context, groupID int64) (int64, int64, error) {
  315. return 0, 0, nil
  316. }
  317. func (m *mockGroupRepoForGateway) DeleteAccountGroupsByGroupID(ctx context.Context, groupID int64) (int64, error) {
  318. return 0, nil
  319. }
  320. func (m *mockGroupRepoForGateway) BindAccountsToGroup(ctx context.Context, groupID int64, accountIDs []int64) error {
  321. return nil
  322. }
  323. func (m *mockGroupRepoForGateway) GetAccountIDsByGroupIDs(ctx context.Context, groupIDs []int64) ([]int64, error) {
  324. return nil, nil
  325. }
  326. func (m *mockGroupRepoForGateway) UpdateSortOrders(ctx context.Context, updates []GroupSortOrderUpdate) error {
  327. return nil
  328. }
  329. func ptr[T any](v T) *T {
  330. return &v
  331. }
  332. // TestGatewayService_SelectAccountForModelWithPlatform_Anthropic 测试 anthropic 单平台选择
  333. func TestGatewayService_SelectAccountForModelWithPlatform_Anthropic(t *testing.T) {
  334. ctx := context.Background()
  335. repo := &mockAccountRepoForPlatform{
  336. accounts: []Account{
  337. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  338. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true},
  339. {ID: 3, Platform: PlatformAntigravity, Priority: 1, Status: StatusActive, Schedulable: true}, // 应被隔离
  340. },
  341. accountsByID: map[int64]*Account{},
  342. }
  343. for i := range repo.accounts {
  344. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  345. }
  346. cache := &mockGatewayCacheForPlatform{}
  347. svc := &GatewayService{
  348. accountRepo: repo,
  349. cache: cache,
  350. cfg: testConfig(),
  351. }
  352. acc, err := svc.selectAccountForModelWithPlatform(ctx, nil, "", "claude-3-5-sonnet-20241022", nil, PlatformAnthropic)
  353. require.NoError(t, err)
  354. require.NotNil(t, acc)
  355. require.Equal(t, int64(1), acc.ID, "应选择优先级最高的 anthropic 账户")
  356. require.Equal(t, PlatformAnthropic, acc.Platform, "应只返回 anthropic 平台账户")
  357. }
  358. // Scenario: account-owned Composite aliases are scheduled only to accounts that declare the exact mapping.
  359. func TestGatewayService_SelectAccountForModelWithExclusions_CompositeAliasRequiresOwningAccount(t *testing.T) {
  360. groupID := int64(77)
  361. repo := &mockAccountRepoForPlatform{
  362. accounts: []Account{
  363. {
  364. ID: 1,
  365. Platform: PlatformAnthropic,
  366. Type: AccountTypeAPIKey,
  367. Priority: 1,
  368. Status: StatusActive,
  369. Schedulable: true,
  370. AccountGroups: []AccountGroup{{GroupID: groupID}},
  371. },
  372. {
  373. ID: 2,
  374. Platform: PlatformAnthropic,
  375. Type: AccountTypeAPIKey,
  376. Priority: 2,
  377. Status: StatusActive,
  378. Schedulable: true,
  379. Credentials: map[string]any{
  380. "model_mapping": map[string]any{"reasoning-alias": "claude-opus-4-8"},
  381. },
  382. AccountGroups: []AccountGroup{{GroupID: groupID}},
  383. },
  384. },
  385. accountsByID: map[int64]*Account{},
  386. }
  387. for i := range repo.accounts {
  388. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  389. }
  390. group := &Group{ID: groupID, Platform: PlatformComposite, Status: StatusActive, Hydrated: true}
  391. svc := &GatewayService{
  392. accountRepo: repo,
  393. groupRepo: &mockGroupRepoForGateway{groups: map[int64]*Group{groupID: group}},
  394. cfg: testConfig(),
  395. }
  396. ctx := WithCompositeRouteDecision(context.Background(), CompositeRouteDecision{
  397. Matched: true,
  398. Source: CompositeRouteSourceAccount,
  399. GroupID: groupID,
  400. PublicModel: "reasoning-alias",
  401. TargetPlatform: PlatformAnthropic,
  402. UpstreamModel: "reasoning-alias",
  403. Endpoint: CompositeRouteEndpointResponses,
  404. })
  405. account, err := svc.SelectAccountForModelWithExclusions(ctx, &groupID, "", "reasoning-alias", nil)
  406. require.NoError(t, err)
  407. require.NotNil(t, account)
  408. require.Equal(t, int64(2), account.ID)
  409. }
  410. // TestGatewayService_SelectAccountForModelWithPlatform_Antigravity 测试 antigravity 单平台选择
  411. func TestGatewayService_SelectAccountForModelWithPlatform_Antigravity(t *testing.T) {
  412. ctx := context.Background()
  413. repo := &mockAccountRepoForPlatform{
  414. accounts: []Account{
  415. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true}, // 应被隔离
  416. {ID: 2, Platform: PlatformAntigravity, Priority: 1, Status: StatusActive, Schedulable: true},
  417. },
  418. accountsByID: map[int64]*Account{},
  419. }
  420. for i := range repo.accounts {
  421. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  422. }
  423. cache := &mockGatewayCacheForPlatform{}
  424. svc := &GatewayService{
  425. accountRepo: repo,
  426. cache: cache,
  427. cfg: testConfig(),
  428. }
  429. acc, err := svc.selectAccountForModelWithPlatform(ctx, nil, "", "claude-sonnet-4-5", nil, PlatformAntigravity)
  430. require.NoError(t, err)
  431. require.NotNil(t, acc)
  432. require.Equal(t, int64(2), acc.ID)
  433. require.Equal(t, PlatformAntigravity, acc.Platform, "应只返回 antigravity 平台账户")
  434. }
  435. // TestGatewayService_SelectAccountForModelWithPlatform_PriorityAndLastUsed 测试优先级和最后使用时间
  436. func TestGatewayService_SelectAccountForModelWithPlatform_PriorityAndLastUsed(t *testing.T) {
  437. ctx := context.Background()
  438. now := time.Now()
  439. repo := &mockAccountRepoForPlatform{
  440. accounts: []Account{
  441. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, LastUsedAt: ptr(now.Add(-1 * time.Hour))},
  442. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, LastUsedAt: ptr(now.Add(-2 * time.Hour))},
  443. },
  444. accountsByID: map[int64]*Account{},
  445. }
  446. for i := range repo.accounts {
  447. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  448. }
  449. cache := &mockGatewayCacheForPlatform{}
  450. svc := &GatewayService{
  451. accountRepo: repo,
  452. cache: cache,
  453. cfg: testConfig(),
  454. }
  455. acc, err := svc.selectAccountForModelWithPlatform(ctx, nil, "", "claude-3-5-sonnet-20241022", nil, PlatformAnthropic)
  456. require.NoError(t, err)
  457. require.NotNil(t, acc)
  458. require.Equal(t, int64(2), acc.ID, "同优先级应选择最久未用的账户")
  459. }
  460. func TestGatewayService_SelectAccountForModelWithPlatform_GeminiOAuthPreference(t *testing.T) {
  461. ctx := context.Background()
  462. repo := &mockAccountRepoForPlatform{
  463. accounts: []Account{
  464. {ID: 1, Platform: PlatformGemini, Priority: 1, Status: StatusActive, Schedulable: true, Type: AccountTypeAPIKey},
  465. {ID: 2, Platform: PlatformGemini, Priority: 1, Status: StatusActive, Schedulable: true, Type: AccountTypeOAuth},
  466. },
  467. accountsByID: map[int64]*Account{},
  468. }
  469. for i := range repo.accounts {
  470. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  471. }
  472. cache := &mockGatewayCacheForPlatform{}
  473. svc := &GatewayService{
  474. accountRepo: repo,
  475. cache: cache,
  476. cfg: testConfig(),
  477. }
  478. acc, err := svc.selectAccountForModelWithPlatform(ctx, nil, "", "gemini-2.5-pro", nil, PlatformGemini)
  479. require.NoError(t, err)
  480. require.NotNil(t, acc)
  481. require.Equal(t, int64(2), acc.ID, "同优先级且未使用时应优先选择OAuth账户")
  482. }
  483. // TestGatewayService_SelectAccountForModelWithPlatform_NoAvailableAccounts 测试无可用账户
  484. func TestGatewayService_SelectAccountForModelWithPlatform_NoAvailableAccounts(t *testing.T) {
  485. ctx := context.Background()
  486. repo := &mockAccountRepoForPlatform{
  487. accounts: []Account{},
  488. accountsByID: map[int64]*Account{},
  489. }
  490. cache := &mockGatewayCacheForPlatform{}
  491. svc := &GatewayService{
  492. accountRepo: repo,
  493. cache: cache,
  494. cfg: testConfig(),
  495. }
  496. acc, err := svc.selectAccountForModelWithPlatform(ctx, nil, "", "claude-3-5-sonnet-20241022", nil, PlatformAnthropic)
  497. require.Error(t, err)
  498. require.Nil(t, acc)
  499. require.ErrorIs(t, err, ErrNoAvailableAccounts)
  500. }
  501. // TestGatewayService_SelectAccountForModelWithPlatform_AllExcluded 测试所有账户被排除
  502. func TestGatewayService_SelectAccountForModelWithPlatform_AllExcluded(t *testing.T) {
  503. ctx := context.Background()
  504. repo := &mockAccountRepoForPlatform{
  505. accounts: []Account{
  506. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  507. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  508. },
  509. accountsByID: map[int64]*Account{},
  510. }
  511. for i := range repo.accounts {
  512. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  513. }
  514. cache := &mockGatewayCacheForPlatform{}
  515. svc := &GatewayService{
  516. accountRepo: repo,
  517. cache: cache,
  518. cfg: testConfig(),
  519. }
  520. excludedIDs := map[int64]struct{}{1: {}, 2: {}}
  521. acc, err := svc.selectAccountForModelWithPlatform(ctx, nil, "", "claude-3-5-sonnet-20241022", excludedIDs, PlatformAnthropic)
  522. require.Error(t, err)
  523. require.Nil(t, acc)
  524. }
  525. // TestGatewayService_SelectAccountForModelWithPlatform_Schedulability 测试账户可调度性检查
  526. func TestGatewayService_SelectAccountForModelWithPlatform_Schedulability(t *testing.T) {
  527. ctx := context.Background()
  528. now := time.Now()
  529. tests := []struct {
  530. name string
  531. accounts []Account
  532. expectedID int64
  533. }{
  534. {
  535. name: "过载账户被跳过",
  536. accounts: []Account{
  537. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, OverloadUntil: ptr(now.Add(1 * time.Hour))},
  538. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true},
  539. },
  540. expectedID: 2,
  541. },
  542. {
  543. name: "限流账户被跳过",
  544. accounts: []Account{
  545. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, RateLimitResetAt: ptr(now.Add(1 * time.Hour))},
  546. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true},
  547. },
  548. expectedID: 2,
  549. },
  550. {
  551. name: "非active账户被跳过",
  552. accounts: []Account{
  553. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: "error", Schedulable: true},
  554. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true},
  555. },
  556. expectedID: 2,
  557. },
  558. {
  559. name: "schedulable=false被跳过",
  560. accounts: []Account{
  561. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: false},
  562. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true},
  563. },
  564. expectedID: 2,
  565. },
  566. {
  567. name: "过期的过载账户可调度",
  568. accounts: []Account{
  569. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, OverloadUntil: ptr(now.Add(-1 * time.Hour))},
  570. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true},
  571. },
  572. expectedID: 1,
  573. },
  574. }
  575. for _, tt := range tests {
  576. t.Run(tt.name, func(t *testing.T) {
  577. repo := &mockAccountRepoForPlatform{
  578. accounts: tt.accounts,
  579. accountsByID: map[int64]*Account{},
  580. }
  581. for i := range repo.accounts {
  582. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  583. }
  584. cache := &mockGatewayCacheForPlatform{}
  585. svc := &GatewayService{
  586. accountRepo: repo,
  587. cache: cache,
  588. cfg: testConfig(),
  589. }
  590. acc, err := svc.selectAccountForModelWithPlatform(ctx, nil, "", "claude-3-5-sonnet-20241022", nil, PlatformAnthropic)
  591. require.NoError(t, err)
  592. require.NotNil(t, acc)
  593. require.Equal(t, tt.expectedID, acc.ID)
  594. })
  595. }
  596. }
  597. // TestGatewayService_SelectAccountForModelWithPlatform_StickySession 测试粘性会话
  598. func TestGatewayService_SelectAccountForModelWithPlatform_StickySession(t *testing.T) {
  599. ctx := context.Background()
  600. t.Run("粘性会话命中-同平台", func(t *testing.T) {
  601. repo := &mockAccountRepoForPlatform{
  602. accounts: []Account{
  603. {ID: 1, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true},
  604. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  605. },
  606. accountsByID: map[int64]*Account{},
  607. }
  608. for i := range repo.accounts {
  609. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  610. }
  611. cache := &mockGatewayCacheForPlatform{
  612. sessionBindings: map[string]int64{"session-123": 1},
  613. }
  614. svc := &GatewayService{
  615. accountRepo: repo,
  616. cache: cache,
  617. cfg: testConfig(),
  618. }
  619. acc, err := svc.selectAccountForModelWithPlatform(ctx, nil, "session-123", "claude-3-5-sonnet-20241022", nil, PlatformAnthropic)
  620. require.NoError(t, err)
  621. require.NotNil(t, acc)
  622. require.Equal(t, int64(1), acc.ID, "应返回粘性会话绑定的账户")
  623. })
  624. t.Run("粘性会话不匹配平台-降级选择", func(t *testing.T) {
  625. repo := &mockAccountRepoForPlatform{
  626. accounts: []Account{
  627. {ID: 1, Platform: PlatformAntigravity, Priority: 2, Status: StatusActive, Schedulable: true}, // 粘性会话绑定但平台不匹配
  628. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  629. },
  630. accountsByID: map[int64]*Account{},
  631. }
  632. for i := range repo.accounts {
  633. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  634. }
  635. cache := &mockGatewayCacheForPlatform{
  636. sessionBindings: map[string]int64{"session-123": 1}, // 绑定 antigravity 账户
  637. }
  638. svc := &GatewayService{
  639. accountRepo: repo,
  640. cache: cache,
  641. cfg: testConfig(),
  642. }
  643. // 请求 anthropic 平台,但粘性会话绑定的是 antigravity 账户
  644. acc, err := svc.selectAccountForModelWithPlatform(ctx, nil, "session-123", "claude-3-5-sonnet-20241022", nil, PlatformAnthropic)
  645. require.NoError(t, err)
  646. require.NotNil(t, acc)
  647. require.Equal(t, int64(2), acc.ID, "粘性会话账户平台不匹配,应降级选择同平台账户")
  648. require.Equal(t, PlatformAnthropic, acc.Platform)
  649. })
  650. t.Run("粘性会话账户被排除-降级选择", func(t *testing.T) {
  651. repo := &mockAccountRepoForPlatform{
  652. accounts: []Account{
  653. {ID: 1, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true},
  654. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  655. },
  656. accountsByID: map[int64]*Account{},
  657. }
  658. for i := range repo.accounts {
  659. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  660. }
  661. cache := &mockGatewayCacheForPlatform{
  662. sessionBindings: map[string]int64{"session-123": 1},
  663. }
  664. svc := &GatewayService{
  665. accountRepo: repo,
  666. cache: cache,
  667. cfg: testConfig(),
  668. }
  669. excludedIDs := map[int64]struct{}{1: {}}
  670. acc, err := svc.selectAccountForModelWithPlatform(ctx, nil, "session-123", "claude-3-5-sonnet-20241022", excludedIDs, PlatformAnthropic)
  671. require.NoError(t, err)
  672. require.NotNil(t, acc)
  673. require.Equal(t, int64(2), acc.ID, "粘性会话账户被排除,应选择其他账户")
  674. })
  675. t.Run("粘性会话账户不可调度-降级选择", func(t *testing.T) {
  676. repo := &mockAccountRepoForPlatform{
  677. accounts: []Account{
  678. {ID: 1, Platform: PlatformAnthropic, Priority: 2, Status: "error", Schedulable: true},
  679. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  680. },
  681. accountsByID: map[int64]*Account{},
  682. }
  683. for i := range repo.accounts {
  684. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  685. }
  686. cache := &mockGatewayCacheForPlatform{
  687. sessionBindings: map[string]int64{"session-123": 1},
  688. }
  689. svc := &GatewayService{
  690. accountRepo: repo,
  691. cache: cache,
  692. cfg: testConfig(),
  693. }
  694. acc, err := svc.selectAccountForModelWithPlatform(ctx, nil, "session-123", "claude-3-5-sonnet-20241022", nil, PlatformAnthropic)
  695. require.NoError(t, err)
  696. require.NotNil(t, acc)
  697. require.Equal(t, int64(2), acc.ID, "粘性会话账户不可调度,应选择其他账户")
  698. })
  699. }
  700. func TestGatewayService_SelectAccountForModelWithExclusions_ForcePlatform(t *testing.T) {
  701. ctx := context.Background()
  702. ctx = context.WithValue(ctx, ctxkey.ForcePlatform, PlatformAntigravity)
  703. repo := &mockAccountRepoForPlatform{
  704. accounts: []Account{
  705. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  706. {ID: 2, Platform: PlatformAntigravity, Priority: 2, Status: StatusActive, Schedulable: true},
  707. },
  708. accountsByID: map[int64]*Account{},
  709. }
  710. for i := range repo.accounts {
  711. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  712. }
  713. cache := &mockGatewayCacheForPlatform{}
  714. svc := &GatewayService{
  715. accountRepo: repo,
  716. cache: cache,
  717. cfg: testConfig(),
  718. }
  719. acc, err := svc.SelectAccountForModelWithExclusions(ctx, nil, "", "claude-sonnet-4-5", nil)
  720. require.NoError(t, err)
  721. require.NotNil(t, acc)
  722. require.Equal(t, int64(2), acc.ID)
  723. require.Equal(t, PlatformAntigravity, acc.Platform)
  724. }
  725. func TestGatewayService_SelectAccountForModelWithPlatform_RoutedStickySessionClears(t *testing.T) {
  726. ctx := context.Background()
  727. groupID := int64(10)
  728. requestedModel := "claude-3-5-sonnet-20241022"
  729. repo := &mockAccountRepoForPlatform{
  730. accounts: []Account{
  731. {ID: 1, Platform: PlatformAnthropic, Priority: 2, Status: StatusDisabled, Schedulable: true},
  732. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  733. },
  734. accountsByID: map[int64]*Account{},
  735. }
  736. for i := range repo.accounts {
  737. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  738. }
  739. cache := &mockGatewayCacheForPlatform{
  740. sessionBindings: map[string]int64{"session-123": 1},
  741. }
  742. groupRepo := &mockGroupRepoForGateway{
  743. groups: map[int64]*Group{
  744. groupID: {
  745. ID: groupID,
  746. Name: "route-group",
  747. Platform: PlatformAnthropic,
  748. Status: StatusActive,
  749. Hydrated: true,
  750. ModelRoutingEnabled: true,
  751. ModelRouting: map[string][]int64{
  752. requestedModel: {1, 2},
  753. },
  754. },
  755. },
  756. }
  757. svc := &GatewayService{
  758. accountRepo: repo,
  759. cache: cache,
  760. cfg: testConfig(),
  761. groupRepo: groupRepo,
  762. }
  763. acc, err := svc.selectAccountForModelWithPlatform(ctx, &groupID, "session-123", requestedModel, nil, PlatformAnthropic)
  764. require.NoError(t, err)
  765. require.NotNil(t, acc)
  766. require.Equal(t, int64(2), acc.ID)
  767. require.Equal(t, 1, cache.deletedSessions["session-123"])
  768. require.Equal(t, int64(2), cache.sessionBindings["session-123"])
  769. }
  770. func TestGatewayService_SelectAccountForModelWithPlatform_RoutedStickySessionHit(t *testing.T) {
  771. ctx := context.Background()
  772. groupID := int64(11)
  773. requestedModel := "claude-3-5-sonnet-20241022"
  774. repo := &mockAccountRepoForPlatform{
  775. accounts: []Account{
  776. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  777. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true},
  778. },
  779. accountsByID: map[int64]*Account{},
  780. }
  781. for i := range repo.accounts {
  782. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  783. }
  784. cache := &mockGatewayCacheForPlatform{
  785. sessionBindings: map[string]int64{"session-456": 1},
  786. }
  787. groupRepo := &mockGroupRepoForGateway{
  788. groups: map[int64]*Group{
  789. groupID: {
  790. ID: groupID,
  791. Name: "route-group-hit",
  792. Platform: PlatformAnthropic,
  793. Status: StatusActive,
  794. Hydrated: true,
  795. ModelRoutingEnabled: true,
  796. ModelRouting: map[string][]int64{
  797. requestedModel: {1, 2},
  798. },
  799. },
  800. },
  801. }
  802. svc := &GatewayService{
  803. accountRepo: repo,
  804. cache: cache,
  805. cfg: testConfig(),
  806. groupRepo: groupRepo,
  807. }
  808. acc, err := svc.selectAccountForModelWithPlatform(ctx, &groupID, "session-456", requestedModel, nil, PlatformAnthropic)
  809. require.NoError(t, err)
  810. require.NotNil(t, acc)
  811. require.Equal(t, int64(1), acc.ID)
  812. }
  813. func TestGatewayService_SelectAccountForModelWithPlatform_RoutedFallbackToNormal(t *testing.T) {
  814. ctx := context.Background()
  815. groupID := int64(12)
  816. requestedModel := "claude-3-5-sonnet-20241022"
  817. repo := &mockAccountRepoForPlatform{
  818. accounts: []Account{
  819. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  820. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true},
  821. },
  822. accountsByID: map[int64]*Account{},
  823. }
  824. for i := range repo.accounts {
  825. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  826. }
  827. cache := &mockGatewayCacheForPlatform{}
  828. groupRepo := &mockGroupRepoForGateway{
  829. groups: map[int64]*Group{
  830. groupID: {
  831. ID: groupID,
  832. Name: "route-fallback",
  833. Platform: PlatformAnthropic,
  834. Status: StatusActive,
  835. Hydrated: true,
  836. ModelRoutingEnabled: true,
  837. ModelRouting: map[string][]int64{
  838. requestedModel: {99},
  839. },
  840. },
  841. },
  842. }
  843. svc := &GatewayService{
  844. accountRepo: repo,
  845. cache: cache,
  846. cfg: testConfig(),
  847. groupRepo: groupRepo,
  848. }
  849. acc, err := svc.selectAccountForModelWithPlatform(ctx, &groupID, "", requestedModel, nil, PlatformAnthropic)
  850. require.NoError(t, err)
  851. require.NotNil(t, acc)
  852. require.Equal(t, int64(1), acc.ID)
  853. }
  854. func TestGatewayService_SelectAccountForModelWithPlatform_NoModelSupport(t *testing.T) {
  855. ctx := context.Background()
  856. repo := &mockAccountRepoForPlatform{
  857. accounts: []Account{
  858. {
  859. ID: 1,
  860. Platform: PlatformAnthropic,
  861. Priority: 1,
  862. Status: StatusActive,
  863. Schedulable: true,
  864. Credentials: map[string]any{"model_mapping": map[string]any{"claude-3-5-haiku-20241022": "claude-3-5-haiku-20241022"}},
  865. },
  866. },
  867. accountsByID: map[int64]*Account{},
  868. }
  869. for i := range repo.accounts {
  870. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  871. }
  872. cache := &mockGatewayCacheForPlatform{}
  873. svc := &GatewayService{
  874. accountRepo: repo,
  875. cache: cache,
  876. cfg: testConfig(),
  877. }
  878. acc, err := svc.selectAccountForModelWithPlatform(ctx, nil, "", "claude-3-5-sonnet-20241022", nil, PlatformAnthropic)
  879. require.Error(t, err)
  880. require.Nil(t, acc)
  881. require.Contains(t, err.Error(), "supporting model")
  882. }
  883. func TestGatewayService_SelectAccountForModelWithPlatform_GeminiPreferOAuth(t *testing.T) {
  884. ctx := context.Background()
  885. repo := &mockAccountRepoForPlatform{
  886. accounts: []Account{
  887. {ID: 1, Platform: PlatformGemini, Priority: 1, Status: StatusActive, Schedulable: true, Type: AccountTypeAPIKey},
  888. {ID: 2, Platform: PlatformGemini, Priority: 1, Status: StatusActive, Schedulable: true, Type: AccountTypeOAuth},
  889. },
  890. accountsByID: map[int64]*Account{},
  891. }
  892. for i := range repo.accounts {
  893. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  894. }
  895. cache := &mockGatewayCacheForPlatform{}
  896. svc := &GatewayService{
  897. accountRepo: repo,
  898. cache: cache,
  899. cfg: testConfig(),
  900. }
  901. acc, err := svc.selectAccountForModelWithPlatform(ctx, nil, "", "gemini-2.5-pro", nil, PlatformGemini)
  902. require.NoError(t, err)
  903. require.NotNil(t, acc)
  904. require.Equal(t, int64(2), acc.ID)
  905. }
  906. func TestGatewayService_SelectAccountForModelWithPlatform_GeminiAPIKeyModelMappingFilter(t *testing.T) {
  907. ctx := context.Background()
  908. repo := &mockAccountRepoForPlatform{
  909. accounts: []Account{
  910. {
  911. ID: 1,
  912. Platform: PlatformGemini,
  913. Type: AccountTypeAPIKey,
  914. Priority: 1,
  915. Status: StatusActive,
  916. Schedulable: true,
  917. Credentials: map[string]any{"model_mapping": map[string]any{"gemini-2.5-pro": "gemini-2.5-pro"}},
  918. },
  919. {
  920. ID: 2,
  921. Platform: PlatformGemini,
  922. Type: AccountTypeAPIKey,
  923. Priority: 2,
  924. Status: StatusActive,
  925. Schedulable: true,
  926. Credentials: map[string]any{"model_mapping": map[string]any{"gemini-2.5-flash": "gemini-2.5-flash"}},
  927. },
  928. },
  929. accountsByID: map[int64]*Account{},
  930. }
  931. for i := range repo.accounts {
  932. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  933. }
  934. cache := &mockGatewayCacheForPlatform{}
  935. svc := &GatewayService{
  936. accountRepo: repo,
  937. cache: cache,
  938. cfg: testConfig(),
  939. }
  940. acc, err := svc.selectAccountForModelWithPlatform(ctx, nil, "", "gemini-2.5-flash", nil, PlatformGemini)
  941. require.NoError(t, err)
  942. require.NotNil(t, acc)
  943. require.Equal(t, int64(2), acc.ID, "应过滤不支持请求模型的 APIKey 账号")
  944. acc, err = svc.selectAccountForModelWithPlatform(ctx, nil, "", "gemini-3-pro-preview", nil, PlatformGemini)
  945. require.Error(t, err)
  946. require.Nil(t, acc)
  947. require.Contains(t, err.Error(), "supporting model")
  948. }
  949. func TestGatewayService_SelectAccountForModelWithPlatform_StickyInGroup(t *testing.T) {
  950. ctx := context.Background()
  951. groupID := int64(50)
  952. repo := &mockAccountRepoForPlatform{
  953. accounts: []Account{
  954. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, AccountGroups: []AccountGroup{{GroupID: groupID}}},
  955. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true, AccountGroups: []AccountGroup{{GroupID: groupID}}},
  956. },
  957. accountsByID: map[int64]*Account{},
  958. }
  959. for i := range repo.accounts {
  960. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  961. }
  962. cache := &mockGatewayCacheForPlatform{
  963. sessionBindings: map[string]int64{"session-group": 1},
  964. }
  965. svc := &GatewayService{
  966. accountRepo: repo,
  967. cache: cache,
  968. cfg: testConfig(),
  969. }
  970. acc, err := svc.selectAccountForModelWithPlatform(ctx, &groupID, "session-group", "", nil, PlatformAnthropic)
  971. require.NoError(t, err)
  972. require.NotNil(t, acc)
  973. require.Equal(t, int64(1), acc.ID)
  974. }
  975. func TestGatewayService_SelectAccountForModelWithPlatform_StickyModelMismatchFallback(t *testing.T) {
  976. ctx := context.Background()
  977. repo := &mockAccountRepoForPlatform{
  978. accounts: []Account{
  979. {
  980. ID: 1,
  981. Platform: PlatformAnthropic,
  982. Priority: 1,
  983. Status: StatusActive,
  984. Schedulable: true,
  985. Credentials: map[string]any{"model_mapping": map[string]any{"claude-3-5-haiku-20241022": "claude-3-5-haiku-20241022"}},
  986. },
  987. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true},
  988. },
  989. accountsByID: map[int64]*Account{},
  990. }
  991. for i := range repo.accounts {
  992. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  993. }
  994. cache := &mockGatewayCacheForPlatform{
  995. sessionBindings: map[string]int64{"session-miss": 1},
  996. }
  997. svc := &GatewayService{
  998. accountRepo: repo,
  999. cache: cache,
  1000. cfg: testConfig(),
  1001. }
  1002. acc, err := svc.selectAccountForModelWithPlatform(ctx, nil, "session-miss", "claude-3-5-sonnet-20241022", nil, PlatformAnthropic)
  1003. require.NoError(t, err)
  1004. require.NotNil(t, acc)
  1005. require.Equal(t, int64(2), acc.ID)
  1006. }
  1007. func TestGatewayService_SelectAccountForModelWithPlatform_PreferNeverUsed(t *testing.T) {
  1008. ctx := context.Background()
  1009. lastUsed := time.Now().Add(-1 * time.Hour)
  1010. repo := &mockAccountRepoForPlatform{
  1011. accounts: []Account{
  1012. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, LastUsedAt: &lastUsed},
  1013. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  1014. },
  1015. accountsByID: map[int64]*Account{},
  1016. }
  1017. for i := range repo.accounts {
  1018. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1019. }
  1020. cache := &mockGatewayCacheForPlatform{}
  1021. svc := &GatewayService{
  1022. accountRepo: repo,
  1023. cache: cache,
  1024. cfg: testConfig(),
  1025. }
  1026. acc, err := svc.selectAccountForModelWithPlatform(ctx, nil, "", "claude-3-5-sonnet-20241022", nil, PlatformAnthropic)
  1027. require.NoError(t, err)
  1028. require.NotNil(t, acc)
  1029. require.Equal(t, int64(2), acc.ID)
  1030. }
  1031. func TestGatewayService_SelectAccountForModelWithPlatform_NoAccounts(t *testing.T) {
  1032. ctx := context.Background()
  1033. repo := &mockAccountRepoForPlatform{
  1034. accounts: []Account{},
  1035. accountsByID: map[int64]*Account{},
  1036. }
  1037. cache := &mockGatewayCacheForPlatform{}
  1038. svc := &GatewayService{
  1039. accountRepo: repo,
  1040. cache: cache,
  1041. cfg: testConfig(),
  1042. }
  1043. acc, err := svc.selectAccountForModelWithPlatform(ctx, nil, "", "", nil, PlatformAnthropic)
  1044. require.Error(t, err)
  1045. require.Nil(t, acc)
  1046. require.ErrorIs(t, err, ErrNoAvailableAccounts)
  1047. }
  1048. func TestGatewayService_isModelSupportedByAccount(t *testing.T) {
  1049. svc := &GatewayService{}
  1050. tests := []struct {
  1051. name string
  1052. account *Account
  1053. model string
  1054. expected bool
  1055. }{
  1056. {
  1057. name: "Antigravity平台-支持默认映射中的claude模型",
  1058. account: &Account{Platform: PlatformAntigravity},
  1059. model: "claude-sonnet-4-5",
  1060. expected: true,
  1061. },
  1062. {
  1063. name: "Antigravity平台-不支持非默认映射中的claude模型",
  1064. account: &Account{Platform: PlatformAntigravity},
  1065. model: "claude-3-5-sonnet-20241022",
  1066. expected: false,
  1067. },
  1068. {
  1069. name: "Antigravity平台-支持gemini模型",
  1070. account: &Account{Platform: PlatformAntigravity},
  1071. model: "gemini-2.5-flash",
  1072. expected: true,
  1073. },
  1074. {
  1075. name: "Antigravity平台-不支持gpt模型",
  1076. account: &Account{Platform: PlatformAntigravity},
  1077. model: "gpt-4",
  1078. expected: false,
  1079. },
  1080. {
  1081. name: "Anthropic平台-无映射配置-支持所有模型",
  1082. account: &Account{Platform: PlatformAnthropic},
  1083. model: "claude-3-5-sonnet-20241022",
  1084. expected: true,
  1085. },
  1086. {
  1087. name: "Anthropic平台-有映射配置-只支持配置的模型",
  1088. account: &Account{
  1089. Platform: PlatformAnthropic,
  1090. Credentials: map[string]any{"model_mapping": map[string]any{"claude-opus-4": "x"}},
  1091. },
  1092. model: "claude-3-5-sonnet-20241022",
  1093. expected: false,
  1094. },
  1095. {
  1096. name: "Anthropic平台-有映射配置-支持配置的模型",
  1097. account: &Account{
  1098. Platform: PlatformAnthropic,
  1099. Credentials: map[string]any{"model_mapping": map[string]any{"claude-3-5-sonnet-20241022": "x"}},
  1100. },
  1101. model: "claude-3-5-sonnet-20241022",
  1102. expected: true,
  1103. },
  1104. {
  1105. name: "Gemini平台-无映射配置-支持所有模型",
  1106. account: &Account{Platform: PlatformGemini, Type: AccountTypeAPIKey},
  1107. model: "gemini-2.5-flash",
  1108. expected: true,
  1109. },
  1110. {
  1111. name: "Gemini平台-有映射配置-只支持配置的模型",
  1112. account: &Account{
  1113. Platform: PlatformGemini,
  1114. Type: AccountTypeAPIKey,
  1115. Credentials: map[string]any{
  1116. "model_mapping": map[string]any{"gemini-2.5-pro": "gemini-2.5-pro"},
  1117. },
  1118. },
  1119. model: "gemini-2.5-flash",
  1120. expected: false,
  1121. },
  1122. {
  1123. name: "Gemini平台-有映射配置-支持配置的模型",
  1124. account: &Account{
  1125. Platform: PlatformGemini,
  1126. Type: AccountTypeAPIKey,
  1127. Credentials: map[string]any{
  1128. "model_mapping": map[string]any{"gemini-2.5-pro": "gemini-2.5-pro"},
  1129. },
  1130. },
  1131. model: "gemini-2.5-pro",
  1132. expected: true,
  1133. },
  1134. }
  1135. for _, tt := range tests {
  1136. t.Run(tt.name, func(t *testing.T) {
  1137. got := svc.isModelSupportedByAccount(tt.account, tt.model)
  1138. require.Equal(t, tt.expected, got)
  1139. })
  1140. }
  1141. }
  1142. // TestGatewayService_selectAccountWithMixedScheduling 测试混合调度
  1143. func TestGatewayService_selectAccountWithMixedScheduling(t *testing.T) {
  1144. ctx := context.Background()
  1145. t.Run("混合调度-Gemini优先选择OAuth账户", func(t *testing.T) {
  1146. repo := &mockAccountRepoForPlatform{
  1147. accounts: []Account{
  1148. {ID: 1, Platform: PlatformGemini, Priority: 1, Status: StatusActive, Schedulable: true, Type: AccountTypeAPIKey},
  1149. {ID: 2, Platform: PlatformGemini, Priority: 1, Status: StatusActive, Schedulable: true, Type: AccountTypeOAuth},
  1150. },
  1151. accountsByID: map[int64]*Account{},
  1152. }
  1153. for i := range repo.accounts {
  1154. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1155. }
  1156. cache := &mockGatewayCacheForPlatform{}
  1157. svc := &GatewayService{
  1158. accountRepo: repo,
  1159. cache: cache,
  1160. cfg: testConfig(),
  1161. }
  1162. acc, err := svc.selectAccountWithMixedScheduling(ctx, nil, "", "gemini-2.5-pro", nil, PlatformGemini)
  1163. require.NoError(t, err)
  1164. require.NotNil(t, acc)
  1165. require.Equal(t, int64(2), acc.ID, "同优先级且未使用时应优先选择OAuth账户")
  1166. })
  1167. t.Run("混合调度-包含启用mixed_scheduling的antigravity账户", func(t *testing.T) {
  1168. repo := &mockAccountRepoForPlatform{
  1169. accounts: []Account{
  1170. {ID: 1, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true},
  1171. {ID: 2, Platform: PlatformAntigravity, Priority: 1, Status: StatusActive, Schedulable: true, Extra: map[string]any{"mixed_scheduling": true}},
  1172. },
  1173. accountsByID: map[int64]*Account{},
  1174. }
  1175. for i := range repo.accounts {
  1176. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1177. }
  1178. cache := &mockGatewayCacheForPlatform{}
  1179. svc := &GatewayService{
  1180. accountRepo: repo,
  1181. cache: cache,
  1182. cfg: testConfig(),
  1183. }
  1184. acc, err := svc.selectAccountWithMixedScheduling(ctx, nil, "", "claude-sonnet-4-5", nil, PlatformAnthropic)
  1185. require.NoError(t, err)
  1186. require.NotNil(t, acc)
  1187. require.Equal(t, int64(2), acc.ID, "应选择优先级最高的账户(包含启用混合调度的antigravity)")
  1188. })
  1189. t.Run("混合调度-Gemini家族限流后跳过Antigravity账户", func(t *testing.T) {
  1190. resetAt := time.Now().Add(10 * time.Minute).Format(time.RFC3339)
  1191. repo := &mockAccountRepoForPlatform{
  1192. accounts: []Account{
  1193. {
  1194. ID: 1,
  1195. Platform: PlatformAntigravity,
  1196. Priority: 1,
  1197. Status: StatusActive,
  1198. Schedulable: true,
  1199. Extra: map[string]any{
  1200. "mixed_scheduling": true,
  1201. modelRateLimitsKey: map[string]any{
  1202. antigravityGeminiModelRateLimitKey: map[string]any{
  1203. "rate_limit_reset_at": resetAt,
  1204. },
  1205. },
  1206. },
  1207. },
  1208. {
  1209. ID: 2,
  1210. Platform: PlatformAntigravity,
  1211. Priority: 1,
  1212. Status: StatusActive,
  1213. Schedulable: true,
  1214. Extra: map[string]any{
  1215. "mixed_scheduling": true,
  1216. modelRateLimitsKey: map[string]any{
  1217. antigravityGeminiModelRateLimitKey: map[string]any{
  1218. "rate_limit_reset_at": resetAt,
  1219. },
  1220. },
  1221. },
  1222. },
  1223. {
  1224. ID: 3,
  1225. Platform: PlatformAntigravity,
  1226. Priority: 2,
  1227. Status: StatusActive,
  1228. Schedulable: true,
  1229. Extra: map[string]any{"mixed_scheduling": true},
  1230. },
  1231. },
  1232. accountsByID: map[int64]*Account{},
  1233. }
  1234. for i := range repo.accounts {
  1235. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1236. }
  1237. svc := &GatewayService{
  1238. accountRepo: repo,
  1239. cache: &mockGatewayCacheForPlatform{},
  1240. cfg: testConfig(),
  1241. }
  1242. acc, err := svc.selectAccountWithMixedScheduling(ctx, nil, "", "gemini-3-pro-preview", nil, PlatformGemini)
  1243. require.NoError(t, err)
  1244. require.NotNil(t, acc)
  1245. require.Equal(t, int64(3), acc.ID)
  1246. })
  1247. t.Run("混合调度-Gemini家族限流不影响Claude调度", func(t *testing.T) {
  1248. resetAt := time.Now().Add(10 * time.Minute).Format(time.RFC3339)
  1249. repo := &mockAccountRepoForPlatform{
  1250. accounts: []Account{
  1251. {
  1252. ID: 1,
  1253. Platform: PlatformAntigravity,
  1254. Priority: 1,
  1255. Status: StatusActive,
  1256. Schedulable: true,
  1257. Extra: map[string]any{
  1258. "mixed_scheduling": true,
  1259. modelRateLimitsKey: map[string]any{
  1260. antigravityGeminiModelRateLimitKey: map[string]any{
  1261. "rate_limit_reset_at": resetAt,
  1262. },
  1263. },
  1264. },
  1265. },
  1266. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true},
  1267. },
  1268. accountsByID: map[int64]*Account{},
  1269. }
  1270. for i := range repo.accounts {
  1271. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1272. }
  1273. svc := &GatewayService{
  1274. accountRepo: repo,
  1275. cache: &mockGatewayCacheForPlatform{},
  1276. cfg: testConfig(),
  1277. }
  1278. acc, err := svc.selectAccountWithMixedScheduling(ctx, nil, "", "claude-sonnet-4-5", nil, PlatformAnthropic)
  1279. require.NoError(t, err)
  1280. require.NotNil(t, acc)
  1281. require.Equal(t, int64(1), acc.ID)
  1282. })
  1283. t.Run("混合调度-路由优先选择路由账号", func(t *testing.T) {
  1284. groupID := int64(30)
  1285. requestedModel := "claude-sonnet-4-5"
  1286. repo := &mockAccountRepoForPlatform{
  1287. accounts: []Account{
  1288. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  1289. {ID: 2, Platform: PlatformAntigravity, Priority: 2, Status: StatusActive, Schedulable: true, Extra: map[string]any{"mixed_scheduling": true}},
  1290. },
  1291. accountsByID: map[int64]*Account{},
  1292. }
  1293. for i := range repo.accounts {
  1294. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1295. }
  1296. cache := &mockGatewayCacheForPlatform{}
  1297. groupRepo := &mockGroupRepoForGateway{
  1298. groups: map[int64]*Group{
  1299. groupID: {
  1300. ID: groupID,
  1301. Name: "route-mixed-select",
  1302. Platform: PlatformAnthropic,
  1303. Status: StatusActive,
  1304. Hydrated: true,
  1305. ModelRoutingEnabled: true,
  1306. ModelRouting: map[string][]int64{
  1307. requestedModel: {2},
  1308. },
  1309. },
  1310. },
  1311. }
  1312. svc := &GatewayService{
  1313. accountRepo: repo,
  1314. cache: cache,
  1315. cfg: testConfig(),
  1316. groupRepo: groupRepo,
  1317. }
  1318. acc, err := svc.selectAccountWithMixedScheduling(ctx, &groupID, "", requestedModel, nil, PlatformAnthropic)
  1319. require.NoError(t, err)
  1320. require.NotNil(t, acc)
  1321. require.Equal(t, int64(2), acc.ID)
  1322. })
  1323. t.Run("混合调度-路由粘性命中", func(t *testing.T) {
  1324. groupID := int64(31)
  1325. requestedModel := "claude-sonnet-4-5"
  1326. repo := &mockAccountRepoForPlatform{
  1327. accounts: []Account{
  1328. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  1329. {ID: 2, Platform: PlatformAntigravity, Priority: 2, Status: StatusActive, Schedulable: true, Extra: map[string]any{"mixed_scheduling": true}, AccountGroups: []AccountGroup{{GroupID: groupID}}},
  1330. },
  1331. accountsByID: map[int64]*Account{},
  1332. }
  1333. for i := range repo.accounts {
  1334. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1335. }
  1336. cache := &mockGatewayCacheForPlatform{
  1337. sessionBindings: map[string]int64{"session-777": 2},
  1338. }
  1339. groupRepo := &mockGroupRepoForGateway{
  1340. groups: map[int64]*Group{
  1341. groupID: {
  1342. ID: groupID,
  1343. Name: "route-mixed-sticky",
  1344. Platform: PlatformAnthropic,
  1345. Status: StatusActive,
  1346. Hydrated: true,
  1347. ModelRoutingEnabled: true,
  1348. ModelRouting: map[string][]int64{
  1349. requestedModel: {2},
  1350. },
  1351. },
  1352. },
  1353. }
  1354. svc := &GatewayService{
  1355. accountRepo: repo,
  1356. cache: cache,
  1357. cfg: testConfig(),
  1358. groupRepo: groupRepo,
  1359. }
  1360. acc, err := svc.selectAccountWithMixedScheduling(ctx, &groupID, "session-777", requestedModel, nil, PlatformAnthropic)
  1361. require.NoError(t, err)
  1362. require.NotNil(t, acc)
  1363. require.Equal(t, int64(2), acc.ID)
  1364. })
  1365. t.Run("混合调度-路由账号缺失回退", func(t *testing.T) {
  1366. groupID := int64(32)
  1367. requestedModel := "claude-3-5-sonnet-20241022"
  1368. repo := &mockAccountRepoForPlatform{
  1369. accounts: []Account{
  1370. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  1371. {ID: 2, Platform: PlatformAntigravity, Priority: 2, Status: StatusActive, Schedulable: true, Extra: map[string]any{"mixed_scheduling": true}},
  1372. },
  1373. accountsByID: map[int64]*Account{},
  1374. }
  1375. for i := range repo.accounts {
  1376. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1377. }
  1378. cache := &mockGatewayCacheForPlatform{}
  1379. groupRepo := &mockGroupRepoForGateway{
  1380. groups: map[int64]*Group{
  1381. groupID: {
  1382. ID: groupID,
  1383. Name: "route-mixed-miss",
  1384. Platform: PlatformAnthropic,
  1385. Status: StatusActive,
  1386. Hydrated: true,
  1387. ModelRoutingEnabled: true,
  1388. ModelRouting: map[string][]int64{
  1389. requestedModel: {99},
  1390. },
  1391. },
  1392. },
  1393. }
  1394. svc := &GatewayService{
  1395. accountRepo: repo,
  1396. cache: cache,
  1397. cfg: testConfig(),
  1398. groupRepo: groupRepo,
  1399. }
  1400. acc, err := svc.selectAccountWithMixedScheduling(ctx, &groupID, "", requestedModel, nil, PlatformAnthropic)
  1401. require.NoError(t, err)
  1402. require.NotNil(t, acc)
  1403. require.Equal(t, int64(1), acc.ID)
  1404. })
  1405. t.Run("混合调度-路由账号未启用mixed_scheduling回退", func(t *testing.T) {
  1406. groupID := int64(33)
  1407. requestedModel := "claude-3-5-sonnet-20241022"
  1408. repo := &mockAccountRepoForPlatform{
  1409. accounts: []Account{
  1410. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  1411. {ID: 2, Platform: PlatformAntigravity, Priority: 2, Status: StatusActive, Schedulable: true}, // 未启用 mixed_scheduling
  1412. },
  1413. accountsByID: map[int64]*Account{},
  1414. }
  1415. for i := range repo.accounts {
  1416. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1417. }
  1418. cache := &mockGatewayCacheForPlatform{}
  1419. groupRepo := &mockGroupRepoForGateway{
  1420. groups: map[int64]*Group{
  1421. groupID: {
  1422. ID: groupID,
  1423. Name: "route-mixed-disabled",
  1424. Platform: PlatformAnthropic,
  1425. Status: StatusActive,
  1426. Hydrated: true,
  1427. ModelRoutingEnabled: true,
  1428. ModelRouting: map[string][]int64{
  1429. requestedModel: {2},
  1430. },
  1431. },
  1432. },
  1433. }
  1434. svc := &GatewayService{
  1435. accountRepo: repo,
  1436. cache: cache,
  1437. cfg: testConfig(),
  1438. groupRepo: groupRepo,
  1439. }
  1440. acc, err := svc.selectAccountWithMixedScheduling(ctx, &groupID, "", requestedModel, nil, PlatformAnthropic)
  1441. require.NoError(t, err)
  1442. require.NotNil(t, acc)
  1443. require.Equal(t, int64(1), acc.ID)
  1444. })
  1445. t.Run("混合调度-路由过滤覆盖", func(t *testing.T) {
  1446. groupID := int64(35)
  1447. requestedModel := "claude-3-5-sonnet-20241022"
  1448. resetAt := time.Now().Add(10 * time.Minute)
  1449. repo := &mockAccountRepoForPlatform{
  1450. accounts: []Account{
  1451. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  1452. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: false},
  1453. {ID: 3, Platform: PlatformAntigravity, Priority: 1, Status: StatusActive, Schedulable: true},
  1454. {
  1455. ID: 4,
  1456. Platform: PlatformAnthropic,
  1457. Priority: 1,
  1458. Status: StatusActive,
  1459. Schedulable: true,
  1460. Extra: map[string]any{
  1461. "model_rate_limits": map[string]any{
  1462. "claude-3-5-sonnet-20241022": map[string]any{
  1463. "rate_limit_reset_at": resetAt.Format(time.RFC3339),
  1464. },
  1465. },
  1466. },
  1467. },
  1468. {
  1469. ID: 5,
  1470. Platform: PlatformAnthropic,
  1471. Priority: 1,
  1472. Status: StatusActive,
  1473. Schedulable: true,
  1474. Credentials: map[string]any{"model_mapping": map[string]any{"claude-3-5-haiku-20241022": "claude-3-5-haiku-20241022"}},
  1475. },
  1476. {ID: 6, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true},
  1477. {ID: 7, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  1478. },
  1479. accountsByID: map[int64]*Account{},
  1480. }
  1481. for i := range repo.accounts {
  1482. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1483. }
  1484. cache := &mockGatewayCacheForPlatform{}
  1485. groupRepo := &mockGroupRepoForGateway{
  1486. groups: map[int64]*Group{
  1487. groupID: {
  1488. ID: groupID,
  1489. Name: "route-mixed-filter",
  1490. Platform: PlatformAnthropic,
  1491. Status: StatusActive,
  1492. Hydrated: true,
  1493. ModelRoutingEnabled: true,
  1494. ModelRouting: map[string][]int64{
  1495. requestedModel: {1, 2, 3, 4, 5, 6, 7},
  1496. },
  1497. },
  1498. },
  1499. }
  1500. svc := &GatewayService{
  1501. accountRepo: repo,
  1502. cache: cache,
  1503. cfg: testConfig(),
  1504. groupRepo: groupRepo,
  1505. }
  1506. excluded := map[int64]struct{}{1: {}}
  1507. acc, err := svc.selectAccountWithMixedScheduling(ctx, &groupID, "", requestedModel, excluded, PlatformAnthropic)
  1508. require.NoError(t, err)
  1509. require.NotNil(t, acc)
  1510. require.Equal(t, int64(7), acc.ID)
  1511. })
  1512. t.Run("混合调度-粘性命中分组账号", func(t *testing.T) {
  1513. groupID := int64(34)
  1514. repo := &mockAccountRepoForPlatform{
  1515. accounts: []Account{
  1516. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, AccountGroups: []AccountGroup{{GroupID: groupID}}},
  1517. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true, AccountGroups: []AccountGroup{{GroupID: groupID}}},
  1518. },
  1519. accountsByID: map[int64]*Account{},
  1520. }
  1521. for i := range repo.accounts {
  1522. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1523. }
  1524. cache := &mockGatewayCacheForPlatform{
  1525. sessionBindings: map[string]int64{"session-group": 1},
  1526. }
  1527. groupRepo := &mockGroupRepoForGateway{
  1528. groups: map[int64]*Group{
  1529. groupID: {
  1530. ID: groupID,
  1531. Platform: PlatformAnthropic,
  1532. Status: StatusActive,
  1533. Hydrated: true,
  1534. },
  1535. },
  1536. }
  1537. svc := &GatewayService{
  1538. accountRepo: repo,
  1539. cache: cache,
  1540. cfg: testConfig(),
  1541. groupRepo: groupRepo,
  1542. }
  1543. acc, err := svc.selectAccountWithMixedScheduling(ctx, &groupID, "session-group", "claude-3-5-sonnet-20241022", nil, PlatformAnthropic)
  1544. require.NoError(t, err)
  1545. require.NotNil(t, acc)
  1546. require.Equal(t, int64(1), acc.ID)
  1547. })
  1548. t.Run("混合调度-过滤未启用mixed_scheduling的antigravity账户", func(t *testing.T) {
  1549. repo := &mockAccountRepoForPlatform{
  1550. accounts: []Account{
  1551. {ID: 1, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true},
  1552. {ID: 2, Platform: PlatformAntigravity, Priority: 1, Status: StatusActive, Schedulable: true}, // 未启用 mixed_scheduling
  1553. },
  1554. accountsByID: map[int64]*Account{},
  1555. }
  1556. for i := range repo.accounts {
  1557. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1558. }
  1559. cache := &mockGatewayCacheForPlatform{}
  1560. svc := &GatewayService{
  1561. accountRepo: repo,
  1562. cache: cache,
  1563. cfg: testConfig(),
  1564. }
  1565. acc, err := svc.selectAccountWithMixedScheduling(ctx, nil, "", "claude-3-5-sonnet-20241022", nil, PlatformAnthropic)
  1566. require.NoError(t, err)
  1567. require.NotNil(t, acc)
  1568. require.Equal(t, int64(1), acc.ID, "未启用mixed_scheduling的antigravity账户应被过滤")
  1569. require.Equal(t, PlatformAnthropic, acc.Platform)
  1570. })
  1571. t.Run("混合调度-粘性会话命中启用mixed_scheduling的antigravity账户", func(t *testing.T) {
  1572. repo := &mockAccountRepoForPlatform{
  1573. accounts: []Account{
  1574. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  1575. {ID: 2, Platform: PlatformAntigravity, Priority: 2, Status: StatusActive, Schedulable: true, Extra: map[string]any{"mixed_scheduling": true}},
  1576. },
  1577. accountsByID: map[int64]*Account{},
  1578. }
  1579. for i := range repo.accounts {
  1580. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1581. }
  1582. cache := &mockGatewayCacheForPlatform{
  1583. sessionBindings: map[string]int64{"session-123": 2},
  1584. }
  1585. svc := &GatewayService{
  1586. accountRepo: repo,
  1587. cache: cache,
  1588. cfg: testConfig(),
  1589. }
  1590. acc, err := svc.selectAccountWithMixedScheduling(ctx, nil, "session-123", "claude-sonnet-4-5", nil, PlatformAnthropic)
  1591. require.NoError(t, err)
  1592. require.NotNil(t, acc)
  1593. require.Equal(t, int64(2), acc.ID, "应返回粘性会话绑定的启用mixed_scheduling的antigravity账户")
  1594. })
  1595. t.Run("混合调度-粘性会话命中未启用mixed_scheduling的antigravity账户-降级选择", func(t *testing.T) {
  1596. repo := &mockAccountRepoForPlatform{
  1597. accounts: []Account{
  1598. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  1599. {ID: 2, Platform: PlatformAntigravity, Priority: 2, Status: StatusActive, Schedulable: true}, // 未启用 mixed_scheduling
  1600. },
  1601. accountsByID: map[int64]*Account{},
  1602. }
  1603. for i := range repo.accounts {
  1604. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1605. }
  1606. cache := &mockGatewayCacheForPlatform{
  1607. sessionBindings: map[string]int64{"session-123": 2},
  1608. }
  1609. svc := &GatewayService{
  1610. accountRepo: repo,
  1611. cache: cache,
  1612. cfg: testConfig(),
  1613. }
  1614. acc, err := svc.selectAccountWithMixedScheduling(ctx, nil, "session-123", "claude-3-5-sonnet-20241022", nil, PlatformAnthropic)
  1615. require.NoError(t, err)
  1616. require.NotNil(t, acc)
  1617. require.Equal(t, int64(1), acc.ID, "粘性会话绑定的账户未启用mixed_scheduling,应降级选择anthropic账户")
  1618. })
  1619. t.Run("混合调度-粘性会话不可调度-清理并回退", func(t *testing.T) {
  1620. repo := &mockAccountRepoForPlatform{
  1621. accounts: []Account{
  1622. {ID: 1, Platform: PlatformAntigravity, Priority: 1, Status: StatusDisabled, Schedulable: true, Extra: map[string]any{"mixed_scheduling": true}},
  1623. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true},
  1624. },
  1625. accountsByID: map[int64]*Account{},
  1626. }
  1627. for i := range repo.accounts {
  1628. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1629. }
  1630. cache := &mockGatewayCacheForPlatform{
  1631. sessionBindings: map[string]int64{"session-123": 1},
  1632. }
  1633. svc := &GatewayService{
  1634. accountRepo: repo,
  1635. cache: cache,
  1636. cfg: testConfig(),
  1637. }
  1638. acc, err := svc.selectAccountWithMixedScheduling(ctx, nil, "session-123", "claude-3-5-sonnet-20241022", nil, PlatformAnthropic)
  1639. require.NoError(t, err)
  1640. require.NotNil(t, acc)
  1641. require.Equal(t, int64(2), acc.ID)
  1642. require.Equal(t, 1, cache.deletedSessions["session-123"])
  1643. require.Equal(t, int64(2), cache.sessionBindings["session-123"])
  1644. })
  1645. t.Run("混合调度-路由粘性不可调度-清理并回退", func(t *testing.T) {
  1646. groupID := int64(12)
  1647. requestedModel := "claude-3-5-sonnet-20241022"
  1648. repo := &mockAccountRepoForPlatform{
  1649. accounts: []Account{
  1650. {ID: 1, Platform: PlatformAntigravity, Priority: 1, Status: StatusDisabled, Schedulable: true, Extra: map[string]any{"mixed_scheduling": true}},
  1651. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true},
  1652. },
  1653. accountsByID: map[int64]*Account{},
  1654. }
  1655. for i := range repo.accounts {
  1656. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1657. }
  1658. cache := &mockGatewayCacheForPlatform{
  1659. sessionBindings: map[string]int64{"session-123": 1},
  1660. }
  1661. groupRepo := &mockGroupRepoForGateway{
  1662. groups: map[int64]*Group{
  1663. groupID: {
  1664. ID: groupID,
  1665. Name: "route-mixed",
  1666. Platform: PlatformAnthropic,
  1667. Status: StatusActive,
  1668. Hydrated: true,
  1669. ModelRoutingEnabled: true,
  1670. ModelRouting: map[string][]int64{
  1671. requestedModel: {1, 2},
  1672. },
  1673. },
  1674. },
  1675. }
  1676. svc := &GatewayService{
  1677. accountRepo: repo,
  1678. cache: cache,
  1679. cfg: testConfig(),
  1680. groupRepo: groupRepo,
  1681. }
  1682. acc, err := svc.selectAccountWithMixedScheduling(ctx, &groupID, "session-123", requestedModel, nil, PlatformAnthropic)
  1683. require.NoError(t, err)
  1684. require.NotNil(t, acc)
  1685. require.Equal(t, int64(2), acc.ID)
  1686. require.Equal(t, 1, cache.deletedSessions["session-123"])
  1687. require.Equal(t, int64(2), cache.sessionBindings["session-123"])
  1688. })
  1689. t.Run("混合调度-仅有启用mixed_scheduling的antigravity账户", func(t *testing.T) {
  1690. repo := &mockAccountRepoForPlatform{
  1691. accounts: []Account{
  1692. {ID: 1, Platform: PlatformAntigravity, Priority: 1, Status: StatusActive, Schedulable: true, Extra: map[string]any{"mixed_scheduling": true}},
  1693. },
  1694. accountsByID: map[int64]*Account{},
  1695. }
  1696. for i := range repo.accounts {
  1697. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1698. }
  1699. cache := &mockGatewayCacheForPlatform{}
  1700. svc := &GatewayService{
  1701. accountRepo: repo,
  1702. cache: cache,
  1703. cfg: testConfig(),
  1704. }
  1705. acc, err := svc.selectAccountWithMixedScheduling(ctx, nil, "", "claude-sonnet-4-5", nil, PlatformAnthropic)
  1706. require.NoError(t, err)
  1707. require.NotNil(t, acc)
  1708. require.Equal(t, int64(1), acc.ID)
  1709. require.Equal(t, PlatformAntigravity, acc.Platform)
  1710. })
  1711. t.Run("混合调度-无可用账户", func(t *testing.T) {
  1712. repo := &mockAccountRepoForPlatform{
  1713. accounts: []Account{
  1714. {ID: 1, Platform: PlatformAntigravity, Priority: 1, Status: StatusActive, Schedulable: true}, // 未启用 mixed_scheduling
  1715. },
  1716. accountsByID: map[int64]*Account{},
  1717. }
  1718. for i := range repo.accounts {
  1719. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1720. }
  1721. cache := &mockGatewayCacheForPlatform{}
  1722. svc := &GatewayService{
  1723. accountRepo: repo,
  1724. cache: cache,
  1725. cfg: testConfig(),
  1726. }
  1727. acc, err := svc.selectAccountWithMixedScheduling(ctx, nil, "", "claude-3-5-sonnet-20241022", nil, PlatformAnthropic)
  1728. require.Error(t, err)
  1729. require.Nil(t, acc)
  1730. require.ErrorIs(t, err, ErrNoAvailableAccounts)
  1731. })
  1732. t.Run("混合调度-不支持模型返回错误", func(t *testing.T) {
  1733. repo := &mockAccountRepoForPlatform{
  1734. accounts: []Account{
  1735. {
  1736. ID: 1,
  1737. Platform: PlatformAnthropic,
  1738. Priority: 1,
  1739. Status: StatusActive,
  1740. Schedulable: true,
  1741. Credentials: map[string]any{"model_mapping": map[string]any{"claude-3-5-haiku-20241022": "claude-3-5-haiku-20241022"}},
  1742. },
  1743. },
  1744. accountsByID: map[int64]*Account{},
  1745. }
  1746. for i := range repo.accounts {
  1747. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1748. }
  1749. cache := &mockGatewayCacheForPlatform{}
  1750. svc := &GatewayService{
  1751. accountRepo: repo,
  1752. cache: cache,
  1753. cfg: testConfig(),
  1754. }
  1755. acc, err := svc.selectAccountWithMixedScheduling(ctx, nil, "", "claude-3-5-sonnet-20241022", nil, PlatformAnthropic)
  1756. require.Error(t, err)
  1757. require.Nil(t, acc)
  1758. require.Contains(t, err.Error(), "supporting model")
  1759. })
  1760. t.Run("混合调度-优先未使用账号", func(t *testing.T) {
  1761. lastUsed := time.Now().Add(-2 * time.Hour)
  1762. repo := &mockAccountRepoForPlatform{
  1763. accounts: []Account{
  1764. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, LastUsedAt: &lastUsed},
  1765. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  1766. },
  1767. accountsByID: map[int64]*Account{},
  1768. }
  1769. for i := range repo.accounts {
  1770. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1771. }
  1772. cache := &mockGatewayCacheForPlatform{}
  1773. svc := &GatewayService{
  1774. accountRepo: repo,
  1775. cache: cache,
  1776. cfg: testConfig(),
  1777. }
  1778. acc, err := svc.selectAccountWithMixedScheduling(ctx, nil, "", "claude-3-5-sonnet-20241022", nil, PlatformAnthropic)
  1779. require.NoError(t, err)
  1780. require.NotNil(t, acc)
  1781. require.Equal(t, int64(2), acc.ID)
  1782. })
  1783. }
  1784. // TestAccount_IsMixedSchedulingEnabled 测试混合调度开关检查
  1785. func TestAccount_IsMixedSchedulingEnabled(t *testing.T) {
  1786. tests := []struct {
  1787. name string
  1788. account Account
  1789. expected bool
  1790. }{
  1791. {
  1792. name: "非antigravity平台-返回false",
  1793. account: Account{Platform: PlatformAnthropic},
  1794. expected: false,
  1795. },
  1796. {
  1797. name: "antigravity平台-无extra-返回false",
  1798. account: Account{Platform: PlatformAntigravity},
  1799. expected: false,
  1800. },
  1801. {
  1802. name: "antigravity平台-extra无mixed_scheduling-返回false",
  1803. account: Account{Platform: PlatformAntigravity, Extra: map[string]any{}},
  1804. expected: false,
  1805. },
  1806. {
  1807. name: "antigravity平台-mixed_scheduling=false-返回false",
  1808. account: Account{Platform: PlatformAntigravity, Extra: map[string]any{"mixed_scheduling": false}},
  1809. expected: false,
  1810. },
  1811. {
  1812. name: "antigravity平台-mixed_scheduling=true-返回true",
  1813. account: Account{Platform: PlatformAntigravity, Extra: map[string]any{"mixed_scheduling": true}},
  1814. expected: true,
  1815. },
  1816. {
  1817. name: "antigravity平台-mixed_scheduling非bool类型-返回false",
  1818. account: Account{Platform: PlatformAntigravity, Extra: map[string]any{"mixed_scheduling": "true"}},
  1819. expected: false,
  1820. },
  1821. }
  1822. for _, tt := range tests {
  1823. t.Run(tt.name, func(t *testing.T) {
  1824. got := tt.account.IsMixedSchedulingEnabled()
  1825. require.Equal(t, tt.expected, got)
  1826. })
  1827. }
  1828. }
  1829. // mockConcurrencyService for testing
  1830. type mockConcurrencyService struct {
  1831. accountLoads map[int64]*AccountLoadInfo
  1832. accountWaitCounts map[int64]int
  1833. acquireResults map[int64]bool
  1834. }
  1835. func (m *mockConcurrencyService) GetAccountsLoadBatch(ctx context.Context, accounts []AccountWithConcurrency) (map[int64]*AccountLoadInfo, error) {
  1836. if m.accountLoads == nil {
  1837. return map[int64]*AccountLoadInfo{}, nil
  1838. }
  1839. result := make(map[int64]*AccountLoadInfo)
  1840. for _, acc := range accounts {
  1841. if load, ok := m.accountLoads[acc.ID]; ok {
  1842. result[acc.ID] = load
  1843. } else {
  1844. result[acc.ID] = &AccountLoadInfo{
  1845. AccountID: acc.ID,
  1846. CurrentConcurrency: 0,
  1847. WaitingCount: 0,
  1848. LoadRate: 0,
  1849. }
  1850. }
  1851. }
  1852. return result, nil
  1853. }
  1854. func (m *mockConcurrencyService) GetAccountWaitingCount(ctx context.Context, accountID int64) (int, error) {
  1855. if m.accountWaitCounts == nil {
  1856. return 0, nil
  1857. }
  1858. return m.accountWaitCounts[accountID], nil
  1859. }
  1860. type mockConcurrencyCache struct {
  1861. acquireAccountCalls int
  1862. loadBatchCalls int
  1863. acquireResults map[int64]bool
  1864. loadBatchErr error
  1865. loadMap map[int64]*AccountLoadInfo
  1866. waitCounts map[int64]int
  1867. skipDefaultLoad bool
  1868. }
  1869. func (m *mockConcurrencyCache) AcquireAccountSlot(ctx context.Context, accountID int64, maxConcurrency int, requestID string) (bool, error) {
  1870. m.acquireAccountCalls++
  1871. if m.acquireResults != nil {
  1872. if result, ok := m.acquireResults[accountID]; ok {
  1873. return result, nil
  1874. }
  1875. }
  1876. return true, nil
  1877. }
  1878. func (m *mockConcurrencyCache) ReleaseAccountSlot(ctx context.Context, accountID int64, requestID string) error {
  1879. return nil
  1880. }
  1881. func (m *mockConcurrencyCache) GetAccountConcurrency(ctx context.Context, accountID int64) (int, error) {
  1882. return 0, nil
  1883. }
  1884. func (m *mockConcurrencyCache) GetAccountConcurrencyBatch(ctx context.Context, accountIDs []int64) (map[int64]int, error) {
  1885. result := make(map[int64]int, len(accountIDs))
  1886. for _, accountID := range accountIDs {
  1887. result[accountID] = 0
  1888. }
  1889. return result, nil
  1890. }
  1891. func (m *mockConcurrencyCache) IncrementAccountWaitCount(ctx context.Context, accountID int64, maxWait int) (bool, error) {
  1892. return true, nil
  1893. }
  1894. func (m *mockConcurrencyCache) DecrementAccountWaitCount(ctx context.Context, accountID int64) error {
  1895. return nil
  1896. }
  1897. func (m *mockConcurrencyCache) GetAccountWaitingCount(ctx context.Context, accountID int64) (int, error) {
  1898. if m.waitCounts != nil {
  1899. if count, ok := m.waitCounts[accountID]; ok {
  1900. return count, nil
  1901. }
  1902. }
  1903. return 0, nil
  1904. }
  1905. func (m *mockConcurrencyCache) AcquireUserSlot(ctx context.Context, userID int64, maxConcurrency int, requestID string) (bool, error) {
  1906. return true, nil
  1907. }
  1908. func (m *mockConcurrencyCache) ReleaseUserSlot(ctx context.Context, userID int64, requestID string) error {
  1909. return nil
  1910. }
  1911. func (m *mockConcurrencyCache) GetUserConcurrency(ctx context.Context, userID int64) (int, error) {
  1912. return 0, nil
  1913. }
  1914. func (m *mockConcurrencyCache) IncrementWaitCount(ctx context.Context, userID int64, maxWait int) (bool, error) {
  1915. return true, nil
  1916. }
  1917. func (m *mockConcurrencyCache) DecrementWaitCount(ctx context.Context, userID int64) error {
  1918. return nil
  1919. }
  1920. func (m *mockConcurrencyCache) GetAccountsLoadBatch(ctx context.Context, accounts []AccountWithConcurrency) (map[int64]*AccountLoadInfo, error) {
  1921. m.loadBatchCalls++
  1922. if m.loadBatchErr != nil {
  1923. return nil, m.loadBatchErr
  1924. }
  1925. result := make(map[int64]*AccountLoadInfo, len(accounts))
  1926. if m.skipDefaultLoad && m.loadMap != nil {
  1927. for _, acc := range accounts {
  1928. if load, ok := m.loadMap[acc.ID]; ok {
  1929. result[acc.ID] = load
  1930. }
  1931. }
  1932. return result, nil
  1933. }
  1934. for _, acc := range accounts {
  1935. if m.loadMap != nil {
  1936. if load, ok := m.loadMap[acc.ID]; ok {
  1937. result[acc.ID] = load
  1938. continue
  1939. }
  1940. }
  1941. result[acc.ID] = &AccountLoadInfo{
  1942. AccountID: acc.ID,
  1943. CurrentConcurrency: 0,
  1944. WaitingCount: 0,
  1945. LoadRate: 0,
  1946. }
  1947. }
  1948. return result, nil
  1949. }
  1950. func (m *mockConcurrencyCache) CleanupExpiredAccountSlots(ctx context.Context, accountID int64) error {
  1951. return nil
  1952. }
  1953. func (m *mockConcurrencyCache) CleanupExpiredAccountSlotKeys(ctx context.Context) error {
  1954. return nil
  1955. }
  1956. func (m *mockConcurrencyCache) CleanupStaleProcessSlots(ctx context.Context, activeRequestPrefix string) error {
  1957. return nil
  1958. }
  1959. func (m *mockConcurrencyCache) GetUsersLoadBatch(ctx context.Context, users []UserWithConcurrency) (map[int64]*UserLoadInfo, error) {
  1960. result := make(map[int64]*UserLoadInfo, len(users))
  1961. for _, user := range users {
  1962. result[user.ID] = &UserLoadInfo{
  1963. UserID: user.ID,
  1964. CurrentConcurrency: 0,
  1965. WaitingCount: 0,
  1966. LoadRate: 0,
  1967. }
  1968. }
  1969. return result, nil
  1970. }
  1971. // TestGatewayService_SelectAccountWithLoadAwareness tests load-aware account selection
  1972. func TestGatewayService_SelectAccountWithLoadAwareness(t *testing.T) {
  1973. ctx := context.Background()
  1974. t.Run("禁用负载批量查询-降级到传统选择", func(t *testing.T) {
  1975. repo := &mockAccountRepoForPlatform{
  1976. accounts: []Account{
  1977. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  1978. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true, Concurrency: 5},
  1979. },
  1980. accountsByID: map[int64]*Account{},
  1981. }
  1982. for i := range repo.accounts {
  1983. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  1984. }
  1985. cache := &mockGatewayCacheForPlatform{}
  1986. cfg := testConfig()
  1987. cfg.Gateway.Scheduling.LoadBatchEnabled = false
  1988. svc := &GatewayService{
  1989. accountRepo: repo,
  1990. cache: cache,
  1991. cfg: cfg,
  1992. concurrencyService: nil, // No concurrency service
  1993. }
  1994. result, err := svc.SelectAccountWithLoadAwareness(ctx, nil, "", "claude-3-5-sonnet-20241022", nil, "", int64(0))
  1995. require.NoError(t, err)
  1996. require.NotNil(t, result)
  1997. require.NotNil(t, result.Account)
  1998. require.Equal(t, int64(1), result.Account.ID, "应选择优先级最高的账号")
  1999. })
  2000. t.Run("模型路由-无ConcurrencyService也生效", func(t *testing.T) {
  2001. groupID := int64(1)
  2002. sessionHash := "sticky"
  2003. repo := &mockAccountRepoForPlatform{
  2004. accounts: []Account{
  2005. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5, AccountGroups: []AccountGroup{{GroupID: groupID}}},
  2006. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5, AccountGroups: []AccountGroup{{GroupID: groupID}}},
  2007. },
  2008. accountsByID: map[int64]*Account{},
  2009. }
  2010. for i := range repo.accounts {
  2011. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2012. }
  2013. cache := &mockGatewayCacheForPlatform{
  2014. sessionBindings: map[string]int64{sessionHash: 1},
  2015. }
  2016. groupRepo := &mockGroupRepoForGateway{
  2017. groups: map[int64]*Group{
  2018. groupID: {
  2019. ID: groupID,
  2020. Platform: PlatformAnthropic,
  2021. Status: StatusActive,
  2022. Hydrated: true,
  2023. ModelRoutingEnabled: true,
  2024. ModelRouting: map[string][]int64{
  2025. "claude-a": {1},
  2026. "claude-b": {2},
  2027. },
  2028. },
  2029. },
  2030. }
  2031. cfg := testConfig()
  2032. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2033. svc := &GatewayService{
  2034. accountRepo: repo,
  2035. groupRepo: groupRepo,
  2036. cache: cache,
  2037. cfg: cfg,
  2038. concurrencyService: nil, // legacy path
  2039. }
  2040. result, err := svc.SelectAccountWithLoadAwareness(ctx, &groupID, sessionHash, "claude-b", nil, "", int64(0))
  2041. require.NoError(t, err)
  2042. require.NotNil(t, result)
  2043. require.NotNil(t, result.Account)
  2044. require.Equal(t, int64(2), result.Account.ID, "切换到 claude-b 时应按模型路由切换账号")
  2045. require.Equal(t, int64(2), cache.sessionBindings[sessionHash], "粘性绑定应更新为路由选择的账号")
  2046. })
  2047. t.Run("无ConcurrencyService-降级到传统选择", func(t *testing.T) {
  2048. repo := &mockAccountRepoForPlatform{
  2049. accounts: []Account{
  2050. {ID: 1, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2051. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2052. },
  2053. accountsByID: map[int64]*Account{},
  2054. }
  2055. for i := range repo.accounts {
  2056. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2057. }
  2058. cache := &mockGatewayCacheForPlatform{}
  2059. cfg := testConfig()
  2060. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2061. svc := &GatewayService{
  2062. accountRepo: repo,
  2063. cache: cache,
  2064. cfg: cfg,
  2065. concurrencyService: nil,
  2066. }
  2067. result, err := svc.SelectAccountWithLoadAwareness(ctx, nil, "", "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2068. require.NoError(t, err)
  2069. require.NotNil(t, result)
  2070. require.NotNil(t, result.Account)
  2071. require.Equal(t, int64(2), result.Account.ID, "应选择优先级最高的账号")
  2072. })
  2073. t.Run("排除账号-不选择被排除的账号", func(t *testing.T) {
  2074. repo := &mockAccountRepoForPlatform{
  2075. accounts: []Account{
  2076. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2077. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2078. },
  2079. accountsByID: map[int64]*Account{},
  2080. }
  2081. for i := range repo.accounts {
  2082. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2083. }
  2084. cache := &mockGatewayCacheForPlatform{}
  2085. cfg := testConfig()
  2086. cfg.Gateway.Scheduling.LoadBatchEnabled = false
  2087. svc := &GatewayService{
  2088. accountRepo: repo,
  2089. cache: cache,
  2090. cfg: cfg,
  2091. concurrencyService: nil,
  2092. }
  2093. excludedIDs := map[int64]struct{}{1: {}}
  2094. result, err := svc.SelectAccountWithLoadAwareness(ctx, nil, "", "claude-3-5-sonnet-20241022", excludedIDs, "", int64(0))
  2095. require.NoError(t, err)
  2096. require.NotNil(t, result)
  2097. require.NotNil(t, result.Account)
  2098. require.Equal(t, int64(2), result.Account.ID, "不应选择被排除的账号")
  2099. })
  2100. t.Run("粘性命中-不调用GetByID", func(t *testing.T) {
  2101. repo := &mockAccountRepoForPlatform{
  2102. accounts: []Account{
  2103. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2104. },
  2105. accountsByID: map[int64]*Account{},
  2106. }
  2107. for i := range repo.accounts {
  2108. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2109. }
  2110. cache := &mockGatewayCacheForPlatform{
  2111. sessionBindings: map[string]int64{"sticky": 1},
  2112. }
  2113. cfg := testConfig()
  2114. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2115. concurrencyCache := &mockConcurrencyCache{}
  2116. svc := &GatewayService{
  2117. accountRepo: repo,
  2118. cache: cache,
  2119. cfg: cfg,
  2120. concurrencyService: NewConcurrencyService(concurrencyCache),
  2121. }
  2122. result, err := svc.SelectAccountWithLoadAwareness(ctx, nil, "sticky", "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2123. require.NoError(t, err)
  2124. require.NotNil(t, result)
  2125. require.NotNil(t, result.Account)
  2126. require.Equal(t, int64(1), result.Account.ID)
  2127. require.Equal(t, 0, repo.getByIDCalls, "粘性命中不应调用GetByID")
  2128. require.Equal(t, 0, concurrencyCache.loadBatchCalls, "粘性命中应在负载批量查询前返回")
  2129. })
  2130. t.Run("粘性账号不在候选集-回退负载感知选择", func(t *testing.T) {
  2131. repo := &mockAccountRepoForPlatform{
  2132. accounts: []Account{
  2133. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2134. },
  2135. accountsByID: map[int64]*Account{},
  2136. }
  2137. for i := range repo.accounts {
  2138. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2139. }
  2140. cache := &mockGatewayCacheForPlatform{
  2141. sessionBindings: map[string]int64{"sticky": 1},
  2142. }
  2143. cfg := testConfig()
  2144. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2145. concurrencyCache := &mockConcurrencyCache{}
  2146. svc := &GatewayService{
  2147. accountRepo: repo,
  2148. cache: cache,
  2149. cfg: cfg,
  2150. concurrencyService: NewConcurrencyService(concurrencyCache),
  2151. }
  2152. result, err := svc.SelectAccountWithLoadAwareness(ctx, nil, "sticky", "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2153. require.NoError(t, err)
  2154. require.NotNil(t, result)
  2155. require.NotNil(t, result.Account)
  2156. require.Equal(t, int64(2), result.Account.ID, "粘性账号不在候选集时应回退到可用账号")
  2157. require.Equal(t, 0, repo.getByIDCalls, "粘性账号缺失不应回退到GetByID")
  2158. require.Equal(t, 1, concurrencyCache.loadBatchCalls, "应继续进行负载批量查询")
  2159. })
  2160. t.Run("粘性账号禁用-清理会话并回退选择", func(t *testing.T) {
  2161. testCtx := context.WithValue(ctx, ctxkey.ForcePlatform, PlatformAnthropic)
  2162. repo := &mockAccountRepoForPlatform{
  2163. accounts: []Account{
  2164. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: false, Concurrency: 5},
  2165. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2166. },
  2167. accountsByID: map[int64]*Account{},
  2168. }
  2169. for i := range repo.accounts {
  2170. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2171. }
  2172. repo.listPlatformFunc = func(ctx context.Context, platform string) ([]Account, error) {
  2173. return repo.accounts, nil
  2174. }
  2175. cache := &mockGatewayCacheForPlatform{
  2176. sessionBindings: map[string]int64{"sticky": 1},
  2177. }
  2178. cfg := testConfig()
  2179. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2180. concurrencyCache := &mockConcurrencyCache{}
  2181. svc := &GatewayService{
  2182. accountRepo: repo,
  2183. cache: cache,
  2184. cfg: cfg,
  2185. concurrencyService: NewConcurrencyService(concurrencyCache),
  2186. }
  2187. result, err := svc.SelectAccountWithLoadAwareness(testCtx, nil, "sticky", "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2188. require.NoError(t, err)
  2189. require.NotNil(t, result)
  2190. require.NotNil(t, result.Account)
  2191. require.Equal(t, int64(2), result.Account.ID, "粘性账号禁用时应回退到可用账号")
  2192. updatedID, ok := cache.sessionBindings["sticky"]
  2193. require.True(t, ok, "粘性会话应更新绑定")
  2194. require.Equal(t, int64(2), updatedID, "粘性会话应绑定到新账号")
  2195. })
  2196. t.Run("无可用账号-返回错误", func(t *testing.T) {
  2197. repo := &mockAccountRepoForPlatform{
  2198. accounts: []Account{},
  2199. accountsByID: map[int64]*Account{},
  2200. }
  2201. cache := &mockGatewayCacheForPlatform{}
  2202. cfg := testConfig()
  2203. cfg.Gateway.Scheduling.LoadBatchEnabled = false
  2204. svc := &GatewayService{
  2205. accountRepo: repo,
  2206. cache: cache,
  2207. cfg: cfg,
  2208. concurrencyService: nil,
  2209. }
  2210. result, err := svc.SelectAccountWithLoadAwareness(ctx, nil, "", "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2211. require.Error(t, err)
  2212. require.Nil(t, result)
  2213. require.ErrorIs(t, err, ErrNoAvailableAccounts)
  2214. })
  2215. t.Run("过滤不可调度账号-限流账号被跳过", func(t *testing.T) {
  2216. now := time.Now()
  2217. resetAt := now.Add(10 * time.Minute)
  2218. repo := &mockAccountRepoForPlatform{
  2219. accounts: []Account{
  2220. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5, RateLimitResetAt: &resetAt},
  2221. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2222. },
  2223. accountsByID: map[int64]*Account{},
  2224. }
  2225. for i := range repo.accounts {
  2226. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2227. }
  2228. cache := &mockGatewayCacheForPlatform{}
  2229. cfg := testConfig()
  2230. cfg.Gateway.Scheduling.LoadBatchEnabled = false
  2231. svc := &GatewayService{
  2232. accountRepo: repo,
  2233. cache: cache,
  2234. cfg: cfg,
  2235. concurrencyService: nil,
  2236. }
  2237. result, err := svc.SelectAccountWithLoadAwareness(ctx, nil, "", "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2238. require.NoError(t, err)
  2239. require.NotNil(t, result)
  2240. require.NotNil(t, result.Account)
  2241. require.Equal(t, int64(2), result.Account.ID, "应跳过限流账号,选择可用账号")
  2242. })
  2243. t.Run("过滤不可调度账号-过载账号被跳过", func(t *testing.T) {
  2244. now := time.Now()
  2245. overloadUntil := now.Add(10 * time.Minute)
  2246. repo := &mockAccountRepoForPlatform{
  2247. accounts: []Account{
  2248. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5, OverloadUntil: &overloadUntil},
  2249. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2250. },
  2251. accountsByID: map[int64]*Account{},
  2252. }
  2253. for i := range repo.accounts {
  2254. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2255. }
  2256. cache := &mockGatewayCacheForPlatform{}
  2257. cfg := testConfig()
  2258. cfg.Gateway.Scheduling.LoadBatchEnabled = false
  2259. svc := &GatewayService{
  2260. accountRepo: repo,
  2261. cache: cache,
  2262. cfg: cfg,
  2263. concurrencyService: nil,
  2264. }
  2265. result, err := svc.SelectAccountWithLoadAwareness(ctx, nil, "", "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2266. require.NoError(t, err)
  2267. require.NotNil(t, result)
  2268. require.NotNil(t, result.Account)
  2269. require.Equal(t, int64(2), result.Account.ID, "应跳过过载账号,选择可用账号")
  2270. })
  2271. t.Run("粘性账号槽位满-返回粘性等待计划", func(t *testing.T) {
  2272. repo := &mockAccountRepoForPlatform{
  2273. accounts: []Account{
  2274. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2275. },
  2276. accountsByID: map[int64]*Account{},
  2277. }
  2278. for i := range repo.accounts {
  2279. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2280. }
  2281. cache := &mockGatewayCacheForPlatform{
  2282. sessionBindings: map[string]int64{"sticky": 1},
  2283. }
  2284. cfg := testConfig()
  2285. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2286. cfg.Gateway.Scheduling.StickySessionMaxWaiting = 1
  2287. concurrencyCache := &mockConcurrencyCache{
  2288. acquireResults: map[int64]bool{1: false},
  2289. waitCounts: map[int64]int{1: 0},
  2290. }
  2291. svc := &GatewayService{
  2292. accountRepo: repo,
  2293. cache: cache,
  2294. cfg: cfg,
  2295. concurrencyService: NewConcurrencyService(concurrencyCache),
  2296. }
  2297. result, err := svc.SelectAccountWithLoadAwareness(ctx, nil, "sticky", "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2298. require.NoError(t, err)
  2299. require.NotNil(t, result)
  2300. require.NotNil(t, result.WaitPlan)
  2301. require.Equal(t, int64(1), result.Account.ID)
  2302. require.Equal(t, 0, concurrencyCache.loadBatchCalls)
  2303. })
  2304. t.Run("负载批量查询失败-降级旧顺序选择", func(t *testing.T) {
  2305. repo := &mockAccountRepoForPlatform{
  2306. accounts: []Account{
  2307. {ID: 1, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2308. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2309. },
  2310. accountsByID: map[int64]*Account{},
  2311. }
  2312. for i := range repo.accounts {
  2313. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2314. }
  2315. cache := &mockGatewayCacheForPlatform{}
  2316. cfg := testConfig()
  2317. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2318. concurrencyCache := &mockConcurrencyCache{
  2319. loadBatchErr: errors.New("load batch failed"),
  2320. }
  2321. svc := &GatewayService{
  2322. accountRepo: repo,
  2323. cache: cache,
  2324. cfg: cfg,
  2325. concurrencyService: NewConcurrencyService(concurrencyCache),
  2326. }
  2327. result, err := svc.SelectAccountWithLoadAwareness(ctx, nil, "legacy", "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2328. require.NoError(t, err)
  2329. require.NotNil(t, result)
  2330. require.NotNil(t, result.Account)
  2331. require.Equal(t, int64(2), result.Account.ID)
  2332. require.Equal(t, int64(2), cache.sessionBindings["legacy"])
  2333. })
  2334. t.Run("模型路由-粘性账号等待计划", func(t *testing.T) {
  2335. groupID := int64(20)
  2336. sessionHash := "route-sticky"
  2337. repo := &mockAccountRepoForPlatform{
  2338. accounts: []Account{
  2339. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2340. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2341. },
  2342. accountsByID: map[int64]*Account{},
  2343. }
  2344. for i := range repo.accounts {
  2345. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2346. }
  2347. cache := &mockGatewayCacheForPlatform{
  2348. sessionBindings: map[string]int64{sessionHash: 1},
  2349. }
  2350. groupRepo := &mockGroupRepoForGateway{
  2351. groups: map[int64]*Group{
  2352. groupID: {
  2353. ID: groupID,
  2354. Platform: PlatformAnthropic,
  2355. Status: StatusActive,
  2356. Hydrated: true,
  2357. ModelRoutingEnabled: true,
  2358. ModelRouting: map[string][]int64{
  2359. "claude-3-5-sonnet-20241022": {1, 2},
  2360. },
  2361. },
  2362. },
  2363. }
  2364. cfg := testConfig()
  2365. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2366. cfg.Gateway.Scheduling.StickySessionMaxWaiting = 1
  2367. concurrencyCache := &mockConcurrencyCache{
  2368. acquireResults: map[int64]bool{1: false},
  2369. waitCounts: map[int64]int{1: 0},
  2370. }
  2371. svc := &GatewayService{
  2372. accountRepo: repo,
  2373. groupRepo: groupRepo,
  2374. cache: cache,
  2375. cfg: cfg,
  2376. concurrencyService: NewConcurrencyService(concurrencyCache),
  2377. }
  2378. result, err := svc.SelectAccountWithLoadAwareness(ctx, &groupID, sessionHash, "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2379. require.NoError(t, err)
  2380. require.NotNil(t, result)
  2381. require.NotNil(t, result.WaitPlan)
  2382. require.Equal(t, int64(1), result.Account.ID)
  2383. })
  2384. t.Run("模型路由-粘性账号命中", func(t *testing.T) {
  2385. groupID := int64(20)
  2386. sessionHash := "route-hit"
  2387. repo := &mockAccountRepoForPlatform{
  2388. accounts: []Account{
  2389. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2390. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2391. },
  2392. accountsByID: map[int64]*Account{},
  2393. }
  2394. for i := range repo.accounts {
  2395. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2396. }
  2397. cache := &mockGatewayCacheForPlatform{
  2398. sessionBindings: map[string]int64{sessionHash: 1},
  2399. }
  2400. groupRepo := &mockGroupRepoForGateway{
  2401. groups: map[int64]*Group{
  2402. groupID: {
  2403. ID: groupID,
  2404. Platform: PlatformAnthropic,
  2405. Status: StatusActive,
  2406. Hydrated: true,
  2407. ModelRoutingEnabled: true,
  2408. ModelRouting: map[string][]int64{
  2409. "claude-3-5-sonnet-20241022": {1, 2},
  2410. },
  2411. },
  2412. },
  2413. }
  2414. cfg := testConfig()
  2415. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2416. concurrencyCache := &mockConcurrencyCache{}
  2417. svc := &GatewayService{
  2418. accountRepo: repo,
  2419. groupRepo: groupRepo,
  2420. cache: cache,
  2421. cfg: cfg,
  2422. concurrencyService: NewConcurrencyService(concurrencyCache),
  2423. }
  2424. result, err := svc.SelectAccountWithLoadAwareness(ctx, &groupID, sessionHash, "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2425. require.NoError(t, err)
  2426. require.NotNil(t, result)
  2427. require.NotNil(t, result.Account)
  2428. require.Equal(t, int64(1), result.Account.ID)
  2429. require.Equal(t, 0, concurrencyCache.loadBatchCalls)
  2430. })
  2431. t.Run("模型路由-粘性账号缺失-清理并回退", func(t *testing.T) {
  2432. groupID := int64(22)
  2433. sessionHash := "route-missing"
  2434. repo := &mockAccountRepoForPlatform{
  2435. accounts: []Account{
  2436. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2437. },
  2438. accountsByID: map[int64]*Account{},
  2439. }
  2440. for i := range repo.accounts {
  2441. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2442. }
  2443. cache := &mockGatewayCacheForPlatform{
  2444. sessionBindings: map[string]int64{sessionHash: 1},
  2445. }
  2446. groupRepo := &mockGroupRepoForGateway{
  2447. groups: map[int64]*Group{
  2448. groupID: {
  2449. ID: groupID,
  2450. Platform: PlatformAnthropic,
  2451. Status: StatusActive,
  2452. Hydrated: true,
  2453. ModelRoutingEnabled: true,
  2454. ModelRouting: map[string][]int64{
  2455. "claude-3-5-sonnet-20241022": {1, 2},
  2456. },
  2457. },
  2458. },
  2459. }
  2460. cfg := testConfig()
  2461. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2462. concurrencyCache := &mockConcurrencyCache{}
  2463. svc := &GatewayService{
  2464. accountRepo: repo,
  2465. groupRepo: groupRepo,
  2466. cache: cache,
  2467. cfg: cfg,
  2468. concurrencyService: NewConcurrencyService(concurrencyCache),
  2469. }
  2470. result, err := svc.SelectAccountWithLoadAwareness(ctx, &groupID, sessionHash, "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2471. require.NoError(t, err)
  2472. require.NotNil(t, result)
  2473. require.NotNil(t, result.Account)
  2474. require.Equal(t, int64(2), result.Account.ID)
  2475. require.Equal(t, 1, cache.deletedSessions[sessionHash])
  2476. require.Equal(t, int64(2), cache.sessionBindings[sessionHash])
  2477. })
  2478. t.Run("模型路由-按负载选择账号", func(t *testing.T) {
  2479. groupID := int64(21)
  2480. repo := &mockAccountRepoForPlatform{
  2481. accounts: []Account{
  2482. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2483. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2484. },
  2485. accountsByID: map[int64]*Account{},
  2486. }
  2487. for i := range repo.accounts {
  2488. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2489. }
  2490. cache := &mockGatewayCacheForPlatform{}
  2491. groupRepo := &mockGroupRepoForGateway{
  2492. groups: map[int64]*Group{
  2493. groupID: {
  2494. ID: groupID,
  2495. Platform: PlatformAnthropic,
  2496. Status: StatusActive,
  2497. Hydrated: true,
  2498. ModelRoutingEnabled: true,
  2499. ModelRouting: map[string][]int64{
  2500. "claude-3-5-sonnet-20241022": {1, 2},
  2501. },
  2502. },
  2503. },
  2504. }
  2505. cfg := testConfig()
  2506. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2507. concurrencyCache := &mockConcurrencyCache{
  2508. loadMap: map[int64]*AccountLoadInfo{
  2509. 1: {AccountID: 1, LoadRate: 80},
  2510. 2: {AccountID: 2, LoadRate: 20},
  2511. },
  2512. }
  2513. svc := &GatewayService{
  2514. accountRepo: repo,
  2515. groupRepo: groupRepo,
  2516. cache: cache,
  2517. cfg: cfg,
  2518. concurrencyService: NewConcurrencyService(concurrencyCache),
  2519. }
  2520. result, err := svc.SelectAccountWithLoadAwareness(ctx, &groupID, "route", "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2521. require.NoError(t, err)
  2522. require.NotNil(t, result)
  2523. require.NotNil(t, result.Account)
  2524. require.Equal(t, int64(2), result.Account.ID)
  2525. require.Equal(t, int64(2), cache.sessionBindings["route"])
  2526. })
  2527. t.Run("模型路由-路由账号全满返回等待计划", func(t *testing.T) {
  2528. groupID := int64(23)
  2529. repo := &mockAccountRepoForPlatform{
  2530. accounts: []Account{
  2531. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2532. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2533. },
  2534. accountsByID: map[int64]*Account{},
  2535. }
  2536. for i := range repo.accounts {
  2537. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2538. }
  2539. cache := &mockGatewayCacheForPlatform{}
  2540. groupRepo := &mockGroupRepoForGateway{
  2541. groups: map[int64]*Group{
  2542. groupID: {
  2543. ID: groupID,
  2544. Platform: PlatformAnthropic,
  2545. Status: StatusActive,
  2546. Hydrated: true,
  2547. ModelRoutingEnabled: true,
  2548. ModelRouting: map[string][]int64{
  2549. "claude-3-5-sonnet-20241022": {1, 2},
  2550. },
  2551. },
  2552. },
  2553. }
  2554. cfg := testConfig()
  2555. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2556. concurrencyCache := &mockConcurrencyCache{
  2557. acquireResults: map[int64]bool{1: false, 2: false},
  2558. loadMap: map[int64]*AccountLoadInfo{
  2559. 1: {AccountID: 1, LoadRate: 10},
  2560. 2: {AccountID: 2, LoadRate: 20},
  2561. },
  2562. }
  2563. svc := &GatewayService{
  2564. accountRepo: repo,
  2565. groupRepo: groupRepo,
  2566. cache: cache,
  2567. cfg: cfg,
  2568. concurrencyService: NewConcurrencyService(concurrencyCache),
  2569. }
  2570. result, err := svc.SelectAccountWithLoadAwareness(ctx, &groupID, "route-full", "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2571. require.NoError(t, err)
  2572. require.NotNil(t, result)
  2573. require.NotNil(t, result.WaitPlan)
  2574. require.Equal(t, int64(1), result.Account.ID)
  2575. })
  2576. t.Run("模型路由-路由账号全满-回退普通选择", func(t *testing.T) {
  2577. groupID := int64(22)
  2578. repo := &mockAccountRepoForPlatform{
  2579. accounts: []Account{
  2580. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2581. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2582. {ID: 3, Platform: PlatformAnthropic, Priority: 0, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2583. },
  2584. accountsByID: map[int64]*Account{},
  2585. }
  2586. for i := range repo.accounts {
  2587. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2588. }
  2589. cache := &mockGatewayCacheForPlatform{}
  2590. groupRepo := &mockGroupRepoForGateway{
  2591. groups: map[int64]*Group{
  2592. groupID: {
  2593. ID: groupID,
  2594. Platform: PlatformAnthropic,
  2595. Status: StatusActive,
  2596. Hydrated: true,
  2597. ModelRoutingEnabled: true,
  2598. ModelRouting: map[string][]int64{
  2599. "claude-3-5-sonnet-20241022": {1, 2},
  2600. },
  2601. },
  2602. },
  2603. }
  2604. cfg := testConfig()
  2605. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2606. concurrencyCache := &mockConcurrencyCache{
  2607. loadMap: map[int64]*AccountLoadInfo{
  2608. 1: {AccountID: 1, LoadRate: 100},
  2609. 2: {AccountID: 2, LoadRate: 100},
  2610. 3: {AccountID: 3, LoadRate: 0},
  2611. },
  2612. }
  2613. svc := &GatewayService{
  2614. accountRepo: repo,
  2615. groupRepo: groupRepo,
  2616. cache: cache,
  2617. cfg: cfg,
  2618. concurrencyService: NewConcurrencyService(concurrencyCache),
  2619. }
  2620. result, err := svc.SelectAccountWithLoadAwareness(ctx, &groupID, "fallback", "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2621. require.NoError(t, err)
  2622. require.NotNil(t, result)
  2623. require.NotNil(t, result.Account)
  2624. require.Equal(t, int64(3), result.Account.ID)
  2625. require.Equal(t, int64(3), cache.sessionBindings["fallback"])
  2626. })
  2627. t.Run("负载批量失败且无法获取-兜底等待", func(t *testing.T) {
  2628. repo := &mockAccountRepoForPlatform{
  2629. accounts: []Account{
  2630. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2631. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2632. },
  2633. accountsByID: map[int64]*Account{},
  2634. }
  2635. for i := range repo.accounts {
  2636. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2637. }
  2638. cache := &mockGatewayCacheForPlatform{}
  2639. cfg := testConfig()
  2640. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2641. concurrencyCache := &mockConcurrencyCache{
  2642. loadBatchErr: errors.New("load batch failed"),
  2643. acquireResults: map[int64]bool{1: false, 2: false},
  2644. }
  2645. svc := &GatewayService{
  2646. accountRepo: repo,
  2647. cache: cache,
  2648. cfg: cfg,
  2649. concurrencyService: NewConcurrencyService(concurrencyCache),
  2650. }
  2651. result, err := svc.SelectAccountWithLoadAwareness(ctx, nil, "", "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2652. require.NoError(t, err)
  2653. require.NotNil(t, result)
  2654. require.NotNil(t, result.WaitPlan)
  2655. require.Equal(t, int64(1), result.Account.ID)
  2656. })
  2657. t.Run("Gemini负载排序-优先OAuth", func(t *testing.T) {
  2658. groupID := int64(24)
  2659. repo := &mockAccountRepoForPlatform{
  2660. accounts: []Account{
  2661. {ID: 1, Platform: PlatformGemini, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5, Type: AccountTypeAPIKey},
  2662. {ID: 2, Platform: PlatformGemini, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5, Type: AccountTypeOAuth},
  2663. },
  2664. accountsByID: map[int64]*Account{},
  2665. }
  2666. for i := range repo.accounts {
  2667. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2668. }
  2669. cache := &mockGatewayCacheForPlatform{}
  2670. groupRepo := &mockGroupRepoForGateway{
  2671. groups: map[int64]*Group{
  2672. groupID: {
  2673. ID: groupID,
  2674. Platform: PlatformGemini,
  2675. Status: StatusActive,
  2676. Hydrated: true,
  2677. },
  2678. },
  2679. }
  2680. cfg := testConfig()
  2681. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2682. concurrencyCache := &mockConcurrencyCache{
  2683. loadMap: map[int64]*AccountLoadInfo{
  2684. 1: {AccountID: 1, LoadRate: 10},
  2685. 2: {AccountID: 2, LoadRate: 10},
  2686. },
  2687. }
  2688. svc := &GatewayService{
  2689. accountRepo: repo,
  2690. groupRepo: groupRepo,
  2691. cache: cache,
  2692. cfg: cfg,
  2693. concurrencyService: NewConcurrencyService(concurrencyCache),
  2694. }
  2695. result, err := svc.SelectAccountWithLoadAwareness(ctx, &groupID, "gemini", "gemini-2.5-pro", nil, "", int64(0))
  2696. require.NoError(t, err)
  2697. require.NotNil(t, result)
  2698. require.NotNil(t, result.Account)
  2699. require.Equal(t, int64(2), result.Account.ID)
  2700. })
  2701. t.Run("模型路由-过滤路径覆盖", func(t *testing.T) {
  2702. groupID := int64(70)
  2703. now := time.Now().Add(10 * time.Minute)
  2704. repo := &mockAccountRepoForPlatform{
  2705. accounts: []Account{
  2706. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2707. {ID: 3, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: false, Concurrency: 5},
  2708. {ID: 4, Platform: PlatformAntigravity, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2709. {
  2710. ID: 5,
  2711. Platform: PlatformAnthropic,
  2712. Priority: 1,
  2713. Status: StatusActive,
  2714. Schedulable: true,
  2715. Concurrency: 5,
  2716. Extra: map[string]any{
  2717. "model_rate_limits": map[string]any{
  2718. "claude-3-5-sonnet-20241022": map[string]any{
  2719. "rate_limit_reset_at": now.Format(time.RFC3339),
  2720. },
  2721. },
  2722. },
  2723. },
  2724. {
  2725. ID: 6,
  2726. Platform: PlatformAnthropic,
  2727. Priority: 1,
  2728. Status: StatusActive,
  2729. Schedulable: true,
  2730. Concurrency: 5,
  2731. Credentials: map[string]any{"model_mapping": map[string]any{"claude-3-5-haiku-20241022": "claude-3-5-haiku-20241022"}},
  2732. },
  2733. {ID: 7, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2734. },
  2735. accountsByID: map[int64]*Account{},
  2736. }
  2737. for i := range repo.accounts {
  2738. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2739. }
  2740. cache := &mockGatewayCacheForPlatform{}
  2741. groupRepo := &mockGroupRepoForGateway{
  2742. groups: map[int64]*Group{
  2743. groupID: {
  2744. ID: groupID,
  2745. Platform: PlatformAnthropic,
  2746. Status: StatusActive,
  2747. Hydrated: true,
  2748. ModelRoutingEnabled: true,
  2749. ModelRouting: map[string][]int64{
  2750. "claude-3-5-sonnet-20241022": {1, 2, 3, 4, 5, 6},
  2751. },
  2752. },
  2753. },
  2754. }
  2755. cfg := testConfig()
  2756. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2757. concurrencyCache := &mockConcurrencyCache{}
  2758. svc := &GatewayService{
  2759. accountRepo: repo,
  2760. groupRepo: groupRepo,
  2761. cache: cache,
  2762. cfg: cfg,
  2763. concurrencyService: NewConcurrencyService(concurrencyCache),
  2764. }
  2765. excluded := map[int64]struct{}{1: {}}
  2766. result, err := svc.SelectAccountWithLoadAwareness(ctx, &groupID, "", "claude-3-5-sonnet-20241022", excluded, "", int64(0))
  2767. require.NoError(t, err)
  2768. require.NotNil(t, result)
  2769. require.NotNil(t, result.Account)
  2770. require.Equal(t, int64(7), result.Account.ID)
  2771. })
  2772. t.Run("ClaudeCode限制-回退分组", func(t *testing.T) {
  2773. groupID := int64(60)
  2774. fallbackID := int64(61)
  2775. repo := &mockAccountRepoForPlatform{
  2776. accounts: []Account{
  2777. {ID: 1, Platform: PlatformGemini, Priority: 1, Status: StatusActive, Schedulable: true},
  2778. },
  2779. accountsByID: map[int64]*Account{},
  2780. }
  2781. for i := range repo.accounts {
  2782. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2783. }
  2784. groupRepo := &mockGroupRepoForGateway{
  2785. groups: map[int64]*Group{
  2786. groupID: {
  2787. ID: groupID,
  2788. Platform: PlatformAnthropic,
  2789. Status: StatusActive,
  2790. Hydrated: true,
  2791. ClaudeCodeOnly: true,
  2792. FallbackGroupID: func() *int64 {
  2793. v := fallbackID
  2794. return &v
  2795. }(),
  2796. },
  2797. fallbackID: {
  2798. ID: fallbackID,
  2799. Platform: PlatformGemini,
  2800. Status: StatusActive,
  2801. Hydrated: true,
  2802. },
  2803. },
  2804. }
  2805. cfg := testConfig()
  2806. cfg.Gateway.Scheduling.LoadBatchEnabled = false
  2807. svc := &GatewayService{
  2808. accountRepo: repo,
  2809. groupRepo: groupRepo,
  2810. cache: &mockGatewayCacheForPlatform{},
  2811. cfg: cfg,
  2812. concurrencyService: nil,
  2813. }
  2814. result, err := svc.SelectAccountWithLoadAwareness(ctx, &groupID, "", "gemini-2.5-pro", nil, "", int64(0))
  2815. require.NoError(t, err)
  2816. require.NotNil(t, result)
  2817. require.NotNil(t, result.Account)
  2818. require.Equal(t, int64(1), result.Account.ID)
  2819. })
  2820. t.Run("ClaudeCode限制-无降级返回错误", func(t *testing.T) {
  2821. groupID := int64(62)
  2822. groupRepo := &mockGroupRepoForGateway{
  2823. groups: map[int64]*Group{
  2824. groupID: {
  2825. ID: groupID,
  2826. Platform: PlatformAnthropic,
  2827. Status: StatusActive,
  2828. Hydrated: true,
  2829. ClaudeCodeOnly: true,
  2830. },
  2831. },
  2832. }
  2833. cfg := testConfig()
  2834. cfg.Gateway.Scheduling.LoadBatchEnabled = false
  2835. svc := &GatewayService{
  2836. accountRepo: &mockAccountRepoForPlatform{},
  2837. groupRepo: groupRepo,
  2838. cache: &mockGatewayCacheForPlatform{},
  2839. cfg: cfg,
  2840. concurrencyService: nil,
  2841. }
  2842. result, err := svc.SelectAccountWithLoadAwareness(ctx, &groupID, "", "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2843. require.Error(t, err)
  2844. require.Nil(t, result)
  2845. require.ErrorIs(t, err, ErrClaudeCodeOnly)
  2846. })
  2847. t.Run("负载可用但无法获取槽位-兜底等待", func(t *testing.T) {
  2848. repo := &mockAccountRepoForPlatform{
  2849. accounts: []Account{
  2850. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2851. {ID: 2, Platform: PlatformAnthropic, Priority: 2, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2852. },
  2853. accountsByID: map[int64]*Account{},
  2854. }
  2855. for i := range repo.accounts {
  2856. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2857. }
  2858. cache := &mockGatewayCacheForPlatform{}
  2859. cfg := testConfig()
  2860. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2861. concurrencyCache := &mockConcurrencyCache{
  2862. acquireResults: map[int64]bool{1: false, 2: false},
  2863. loadMap: map[int64]*AccountLoadInfo{
  2864. 1: {AccountID: 1, LoadRate: 10},
  2865. 2: {AccountID: 2, LoadRate: 20},
  2866. },
  2867. }
  2868. svc := &GatewayService{
  2869. accountRepo: repo,
  2870. cache: cache,
  2871. cfg: cfg,
  2872. concurrencyService: NewConcurrencyService(concurrencyCache),
  2873. }
  2874. result, err := svc.SelectAccountWithLoadAwareness(ctx, nil, "wait", "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2875. require.NoError(t, err)
  2876. require.NotNil(t, result)
  2877. require.NotNil(t, result.WaitPlan)
  2878. require.Equal(t, int64(1), result.Account.ID)
  2879. })
  2880. t.Run("负载信息缺失-使用默认负载", func(t *testing.T) {
  2881. repo := &mockAccountRepoForPlatform{
  2882. accounts: []Account{
  2883. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2884. {ID: 2, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true, Concurrency: 5},
  2885. },
  2886. accountsByID: map[int64]*Account{},
  2887. }
  2888. for i := range repo.accounts {
  2889. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2890. }
  2891. cache := &mockGatewayCacheForPlatform{}
  2892. cfg := testConfig()
  2893. cfg.Gateway.Scheduling.LoadBatchEnabled = true
  2894. concurrencyCache := &mockConcurrencyCache{
  2895. loadMap: map[int64]*AccountLoadInfo{
  2896. 1: {AccountID: 1, LoadRate: 50},
  2897. },
  2898. skipDefaultLoad: true,
  2899. }
  2900. svc := &GatewayService{
  2901. accountRepo: repo,
  2902. cache: cache,
  2903. cfg: cfg,
  2904. concurrencyService: NewConcurrencyService(concurrencyCache),
  2905. }
  2906. result, err := svc.SelectAccountWithLoadAwareness(ctx, nil, "missing-load", "claude-3-5-sonnet-20241022", nil, "", int64(0))
  2907. require.NoError(t, err)
  2908. require.NotNil(t, result)
  2909. require.NotNil(t, result.Account)
  2910. require.Equal(t, int64(2), result.Account.ID)
  2911. })
  2912. }
  2913. func TestGatewayService_GroupResolution_ReusesContextGroup(t *testing.T) {
  2914. ctx := context.Background()
  2915. groupID := int64(42)
  2916. group := &Group{
  2917. ID: groupID,
  2918. Platform: PlatformAnthropic,
  2919. Status: StatusActive,
  2920. Hydrated: true,
  2921. }
  2922. ctx = context.WithValue(ctx, ctxkey.Group, group)
  2923. repo := &mockAccountRepoForPlatform{
  2924. accounts: []Account{
  2925. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  2926. },
  2927. accountsByID: map[int64]*Account{},
  2928. }
  2929. for i := range repo.accounts {
  2930. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2931. }
  2932. groupRepo := &mockGroupRepoForGateway{
  2933. groups: map[int64]*Group{groupID: group},
  2934. }
  2935. svc := &GatewayService{
  2936. accountRepo: repo,
  2937. groupRepo: groupRepo,
  2938. cfg: testConfig(),
  2939. }
  2940. account, err := svc.SelectAccountForModelWithExclusions(ctx, &groupID, "", "claude-3-5-sonnet-20241022", nil)
  2941. require.NoError(t, err)
  2942. require.NotNil(t, account)
  2943. require.Equal(t, 1, groupRepo.getByIDCalls) // +1 for require_privacy_set check
  2944. require.Equal(t, 0, groupRepo.getByIDLiteCalls)
  2945. }
  2946. func TestGatewayService_GroupResolution_IgnoresInvalidContextGroup(t *testing.T) {
  2947. ctx := context.Background()
  2948. groupID := int64(42)
  2949. ctxGroup := &Group{
  2950. ID: groupID,
  2951. Platform: PlatformAnthropic,
  2952. Status: StatusActive,
  2953. }
  2954. ctx = context.WithValue(ctx, ctxkey.Group, ctxGroup)
  2955. repo := &mockAccountRepoForPlatform{
  2956. accounts: []Account{
  2957. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  2958. },
  2959. accountsByID: map[int64]*Account{},
  2960. }
  2961. for i := range repo.accounts {
  2962. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  2963. }
  2964. group := &Group{
  2965. ID: groupID,
  2966. Platform: PlatformAnthropic,
  2967. Status: StatusActive,
  2968. Hydrated: true,
  2969. }
  2970. groupRepo := &mockGroupRepoForGateway{
  2971. groups: map[int64]*Group{groupID: group},
  2972. }
  2973. svc := &GatewayService{
  2974. accountRepo: repo,
  2975. groupRepo: groupRepo,
  2976. cfg: testConfig(),
  2977. }
  2978. account, err := svc.SelectAccountForModelWithExclusions(ctx, &groupID, "", "claude-3-5-sonnet-20241022", nil)
  2979. require.NoError(t, err)
  2980. require.NotNil(t, account)
  2981. require.Equal(t, 1, groupRepo.getByIDCalls) // +1 for require_privacy_set check
  2982. require.Equal(t, 1, groupRepo.getByIDLiteCalls)
  2983. }
  2984. func TestGatewayService_GroupContext_OverwritesInvalidContextGroup(t *testing.T) {
  2985. groupID := int64(42)
  2986. invalidGroup := &Group{
  2987. ID: groupID,
  2988. Platform: PlatformAnthropic,
  2989. Status: StatusActive,
  2990. }
  2991. hydratedGroup := &Group{
  2992. ID: groupID,
  2993. Platform: PlatformAnthropic,
  2994. Status: StatusActive,
  2995. Hydrated: true,
  2996. }
  2997. ctx := context.WithValue(context.Background(), ctxkey.Group, invalidGroup)
  2998. svc := &GatewayService{}
  2999. ctx = svc.withGroupContext(ctx, hydratedGroup)
  3000. got, ok := ctx.Value(ctxkey.Group).(*Group)
  3001. require.True(t, ok)
  3002. require.Same(t, hydratedGroup, got)
  3003. }
  3004. func TestGatewayService_GroupResolution_FallbackUsesLiteOnce(t *testing.T) {
  3005. ctx := context.Background()
  3006. groupID := int64(10)
  3007. fallbackID := int64(11)
  3008. group := &Group{
  3009. ID: groupID,
  3010. Platform: PlatformAnthropic,
  3011. Status: StatusActive,
  3012. ClaudeCodeOnly: true,
  3013. FallbackGroupID: &fallbackID,
  3014. Hydrated: true,
  3015. }
  3016. fallbackGroup := &Group{
  3017. ID: fallbackID,
  3018. Platform: PlatformAnthropic,
  3019. Status: StatusActive,
  3020. Hydrated: true,
  3021. }
  3022. ctx = context.WithValue(ctx, ctxkey.Group, group)
  3023. repo := &mockAccountRepoForPlatform{
  3024. accounts: []Account{
  3025. {ID: 1, Platform: PlatformAnthropic, Priority: 1, Status: StatusActive, Schedulable: true},
  3026. },
  3027. accountsByID: map[int64]*Account{},
  3028. }
  3029. for i := range repo.accounts {
  3030. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  3031. }
  3032. groupRepo := &mockGroupRepoForGateway{
  3033. groups: map[int64]*Group{fallbackID: fallbackGroup},
  3034. }
  3035. svc := &GatewayService{
  3036. accountRepo: repo,
  3037. groupRepo: groupRepo,
  3038. cfg: testConfig(),
  3039. }
  3040. account, err := svc.SelectAccountForModelWithExclusions(ctx, &groupID, "", "claude-3-5-sonnet-20241022", nil)
  3041. require.NoError(t, err)
  3042. require.NotNil(t, account)
  3043. require.Equal(t, 1, groupRepo.getByIDCalls) // +1 for require_privacy_set check
  3044. require.Equal(t, 1, groupRepo.getByIDLiteCalls)
  3045. }
  3046. func TestGatewayService_ResolveGatewayGroup_DetectsFallbackCycle(t *testing.T) {
  3047. ctx := context.Background()
  3048. groupID := int64(10)
  3049. fallbackID := int64(11)
  3050. group := &Group{
  3051. ID: groupID,
  3052. Platform: PlatformAnthropic,
  3053. Status: StatusActive,
  3054. ClaudeCodeOnly: true,
  3055. FallbackGroupID: &fallbackID,
  3056. }
  3057. fallbackGroup := &Group{
  3058. ID: fallbackID,
  3059. Platform: PlatformAnthropic,
  3060. Status: StatusActive,
  3061. ClaudeCodeOnly: true,
  3062. FallbackGroupID: &groupID,
  3063. }
  3064. groupRepo := &mockGroupRepoForGateway{
  3065. groups: map[int64]*Group{
  3066. groupID: group,
  3067. fallbackID: fallbackGroup,
  3068. },
  3069. }
  3070. svc := &GatewayService{
  3071. groupRepo: groupRepo,
  3072. }
  3073. gotGroup, gotID, err := svc.resolveGatewayGroup(ctx, &groupID)
  3074. require.Error(t, err)
  3075. require.Nil(t, gotGroup)
  3076. require.Nil(t, gotID)
  3077. require.Contains(t, err.Error(), "fallback group cycle")
  3078. }
  3079. func TestModelRoutingAppliesToPlatform(t *testing.T) {
  3080. for _, tc := range []struct {
  3081. name string
  3082. targetPlatform string
  3083. groupPlatform string
  3084. want bool
  3085. }{
  3086. {"anthropic group", PlatformAnthropic, PlatformAnthropic, true},
  3087. {"openai group", PlatformOpenAI, PlatformOpenAI, true},
  3088. {"composite group resolved to anthropic", PlatformAnthropic, PlatformComposite, true},
  3089. {"composite group resolved to openai", PlatformOpenAI, PlatformComposite, true},
  3090. {"target platform outside the allowed set", PlatformGemini, PlatformGemini, false},
  3091. {"composite group resolved outside the allowed set", PlatformGemini, PlatformComposite, false},
  3092. {"group platform does not match target", PlatformOpenAI, PlatformAnthropic, false},
  3093. {"empty target platform", "", PlatformOpenAI, false},
  3094. } {
  3095. t.Run(tc.name, func(t *testing.T) {
  3096. require.Equal(t, tc.want, modelRoutingAppliesToPlatform(tc.targetPlatform, tc.groupPlatform))
  3097. })
  3098. }
  3099. }
  3100. // Model routing used to be read only for requests resolved to Anthropic, so an
  3101. // OpenAI group's rules were stored and displayed but never honored. Routing an
  3102. // OpenAI request must pick the routed account instead of the group-wide winner.
  3103. func TestGatewayService_SelectAccountForModelWithPlatform_RoutedOpenAIGroup(t *testing.T) {
  3104. ctx := context.Background()
  3105. groupID := int64(13)
  3106. requestedModel := "gpt-6-astra"
  3107. repo := &mockAccountRepoForPlatform{
  3108. accounts: []Account{
  3109. // Priority 1 would win any group-wide selection, but it is not routed.
  3110. {ID: 1, Platform: PlatformOpenAI, Priority: 1, Status: StatusActive, Schedulable: true},
  3111. {ID: 2, Platform: PlatformOpenAI, Priority: 5, Status: StatusActive, Schedulable: true},
  3112. },
  3113. accountsByID: map[int64]*Account{},
  3114. }
  3115. for i := range repo.accounts {
  3116. repo.accountsByID[repo.accounts[i].ID] = &repo.accounts[i]
  3117. }
  3118. groupRepo := &mockGroupRepoForGateway{
  3119. groups: map[int64]*Group{
  3120. groupID: {
  3121. ID: groupID,
  3122. Name: "openai-route-group",
  3123. Platform: PlatformOpenAI,
  3124. Status: StatusActive,
  3125. Hydrated: true,
  3126. ModelRoutingEnabled: true,
  3127. ModelRouting: map[string][]int64{
  3128. requestedModel: {2},
  3129. },
  3130. },
  3131. },
  3132. }
  3133. svc := &GatewayService{
  3134. accountRepo: repo,
  3135. cache: &mockGatewayCacheForPlatform{},
  3136. cfg: testConfig(),
  3137. groupRepo: groupRepo,
  3138. }
  3139. acc, err := svc.selectAccountForModelWithPlatform(ctx, &groupID, "", requestedModel, nil, PlatformOpenAI)
  3140. require.NoError(t, err)
  3141. require.NotNil(t, acc)
  3142. require.Equal(t, int64(2), acc.ID, "routed account must win over the higher-priority unrouted one")
  3143. }