group_capacity_service_test.go 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179
  1. package service
  2. import (
  3. "context"
  4. "testing"
  5. "time"
  6. "github.com/stretchr/testify/require"
  7. )
  8. type groupCapacityAccountRepoStub struct {
  9. AccountRepository
  10. rows []GroupAccountCapacityRow
  11. requested []int64
  12. }
  13. func (s *groupCapacityAccountRepoStub) ListSchedulableCapacityByGroupIDs(_ context.Context, groupIDs []int64) ([]GroupAccountCapacityRow, error) {
  14. s.requested = append([]int64(nil), groupIDs...)
  15. return append([]GroupAccountCapacityRow(nil), s.rows...), nil
  16. }
  17. type groupCapacityGroupRepoStub struct {
  18. GroupRepository
  19. groupIDs []int64
  20. listCalls int
  21. }
  22. func (s *groupCapacityGroupRepoStub) ListActiveIDs(context.Context) ([]int64, error) {
  23. s.listCalls++
  24. return append([]int64(nil), s.groupIDs...), nil
  25. }
  26. type groupCapacityConcurrencyCacheStub struct {
  27. ConcurrencyCache
  28. counts map[int64]int
  29. requested []int64
  30. }
  31. func (s *groupCapacityConcurrencyCacheStub) GetAccountConcurrencyBatch(_ context.Context, accountIDs []int64) (map[int64]int, error) {
  32. s.requested = append([]int64(nil), accountIDs...)
  33. out := make(map[int64]int, len(accountIDs))
  34. for _, id := range accountIDs {
  35. out[id] = s.counts[id]
  36. }
  37. return out, nil
  38. }
  39. type groupCapacitySessionCacheStub struct {
  40. SessionLimitCache
  41. counts map[int64]int
  42. requested []int64
  43. idleTimeouts map[int64]time.Duration
  44. }
  45. func (s *groupCapacitySessionCacheStub) GetActiveSessionCountBatch(_ context.Context, accountIDs []int64, idleTimeouts map[int64]time.Duration) (map[int64]int, error) {
  46. s.requested = append([]int64(nil), accountIDs...)
  47. s.idleTimeouts = make(map[int64]time.Duration, len(idleTimeouts))
  48. for id, timeout := range idleTimeouts {
  49. s.idleTimeouts[id] = timeout
  50. }
  51. out := make(map[int64]int, len(accountIDs))
  52. for _, id := range accountIDs {
  53. out[id] = s.counts[id]
  54. }
  55. return out, nil
  56. }
  57. type groupCapacityRPMCacheStub struct {
  58. RPMCache
  59. counts map[int64]int
  60. requested []int64
  61. }
  62. func (s *groupCapacityRPMCacheStub) GetRPMBatch(_ context.Context, accountIDs []int64) (map[int64]int, error) {
  63. s.requested = append([]int64(nil), accountIDs...)
  64. out := make(map[int64]int, len(accountIDs))
  65. for _, id := range accountIDs {
  66. out[id] = s.counts[id]
  67. }
  68. return out, nil
  69. }
  70. func TestGetAllGroupCapacityBatchAggregatesRuntimeAndLimits(t *testing.T) {
  71. accountRepo := &groupCapacityAccountRepoStub{
  72. rows: []GroupAccountCapacityRow{
  73. {
  74. GroupID: 10,
  75. AccountID: 1,
  76. Concurrency: 2,
  77. Extra: map[string]any{
  78. "max_sessions": 3,
  79. "session_idle_timeout_minutes": 7,
  80. "base_rpm": 11,
  81. },
  82. },
  83. {
  84. GroupID: 20,
  85. AccountID: 1,
  86. Concurrency: 2,
  87. Extra: map[string]any{
  88. "max_sessions": 3,
  89. "session_idle_timeout_minutes": 7,
  90. "base_rpm": 11,
  91. },
  92. },
  93. {
  94. GroupID: 20,
  95. AccountID: 2,
  96. Concurrency: 4,
  97. Extra: map[string]any{
  98. "max_sessions": 1,
  99. "session_idle_timeout_minutes": 9,
  100. "base_rpm": 13,
  101. },
  102. },
  103. },
  104. }
  105. groupRepo := &groupCapacityGroupRepoStub{groupIDs: []int64{10, 20}}
  106. concurrencyCache := &groupCapacityConcurrencyCacheStub{counts: map[int64]int{1: 1, 2: 2}}
  107. sessionCache := &groupCapacitySessionCacheStub{counts: map[int64]int{1: 2, 2: 1}}
  108. rpmCache := &groupCapacityRPMCacheStub{counts: map[int64]int{1: 5, 2: 7}}
  109. svc := NewGroupCapacityService(
  110. accountRepo,
  111. groupRepo,
  112. NewConcurrencyService(concurrencyCache),
  113. sessionCache,
  114. rpmCache,
  115. )
  116. results, err := svc.GetAllGroupCapacity(context.Background())
  117. require.NoError(t, err)
  118. require.Equal(t, 1, groupRepo.listCalls)
  119. require.Equal(t, []int64{10, 20}, accountRepo.requested)
  120. require.Equal(t, []int64{1, 2}, concurrencyCache.requested)
  121. require.ElementsMatch(t, []int64{1, 2}, sessionCache.requested)
  122. require.ElementsMatch(t, []int64{1, 2}, rpmCache.requested)
  123. require.Equal(t, 7*time.Minute, sessionCache.idleTimeouts[1])
  124. require.Equal(t, 9*time.Minute, sessionCache.idleTimeouts[2])
  125. require.Equal(t, []GroupCapacitySummary{
  126. {
  127. GroupID: 10,
  128. ConcurrencyUsed: 1,
  129. ConcurrencyMax: 2,
  130. SessionsUsed: 2,
  131. SessionsMax: 3,
  132. RPMUsed: 5,
  133. RPMMax: 11,
  134. },
  135. {
  136. GroupID: 20,
  137. ConcurrencyUsed: 3,
  138. ConcurrencyMax: 6,
  139. SessionsUsed: 3,
  140. SessionsMax: 4,
  141. RPMUsed: 12,
  142. RPMMax: 24,
  143. },
  144. }, results)
  145. }
  146. func TestGetAllGroupCapacityBatchKeepsEmptyGroupRows(t *testing.T) {
  147. accountRepo := &groupCapacityAccountRepoStub{
  148. rows: []GroupAccountCapacityRow{
  149. {GroupID: 20, AccountID: 2, Concurrency: 4},
  150. },
  151. }
  152. groupRepo := &groupCapacityGroupRepoStub{groupIDs: []int64{10, 20}}
  153. svc := NewGroupCapacityService(accountRepo, groupRepo, nil, nil, nil)
  154. results, err := svc.GetAllGroupCapacity(context.Background())
  155. require.NoError(t, err)
  156. require.Equal(t, []GroupCapacitySummary{
  157. {GroupID: 10},
  158. {GroupID: 20, ConcurrencyMax: 4},
  159. }, results)
  160. }