proxy_expiry_integration_test.go 4.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140
  1. //go:build integration
  2. package repository
  3. import (
  4. "context"
  5. "encoding/json"
  6. "testing"
  7. "time"
  8. dbent "github.com/Wei-Shaw/sub2api/ent"
  9. "github.com/Wei-Shaw/sub2api/internal/service"
  10. "github.com/stretchr/testify/suite"
  11. )
  12. type ProxyExpirySuite struct {
  13. suite.Suite
  14. ctx context.Context
  15. tx *dbent.Tx
  16. repo *proxyRepository
  17. }
  18. func (s *ProxyExpirySuite) SetupTest() {
  19. s.ctx = context.Background()
  20. s.tx = testEntTx(s.T())
  21. s.repo = newProxyRepositoryWithSQL(s.tx.Client(), s.tx)
  22. }
  23. func TestProxyExpirySuite(t *testing.T) { suite.Run(t, new(ProxyExpirySuite)) }
  24. func (s *ProxyExpirySuite) mkProxy(name, mode string, expiresAt *time.Time, backupID *int64) int64 {
  25. p := &service.Proxy{Name: name, Protocol: "http", Host: "127.0.0.1", Port: 8080,
  26. Status: service.StatusActive, FallbackMode: mode, ExpiryWarnDays: 7,
  27. ExpiresAt: expiresAt, BackupProxyID: backupID}
  28. s.Require().NoError(s.repo.Create(s.ctx, p))
  29. return p.ID
  30. }
  31. func (s *ProxyExpirySuite) mkAccountWithProxy(proxyID int64) int64 {
  32. var id int64
  33. err := scanSingleRow(s.ctx, s.tx, `
  34. INSERT INTO accounts (name, platform, type, credentials, extra, status, proxy_id, created_at, updated_at)
  35. VALUES ($1,'claude','api','{}','{}','active',$2,NOW(),NOW()) RETURNING id`,
  36. []any{"acc-" + time.Now().Format("150405.000000"), proxyID}, &id)
  37. s.Require().NoError(err)
  38. return id
  39. }
  40. func (s *ProxyExpirySuite) accountProxyID(id int64) *int64 {
  41. var pid *int64
  42. err := scanSingleRow(s.ctx, s.tx, `SELECT proxy_id FROM accounts WHERE id=$1`, []any{id}, &pid)
  43. s.Require().NoError(err)
  44. return pid
  45. }
  46. func (s *ProxyExpirySuite) TestSweep_DirectMode() {
  47. past := time.Now().Add(-time.Hour)
  48. pid := s.mkProxy("p-direct", service.FallbackModeDirect, &past, nil)
  49. aid := s.mkAccountWithProxy(pid)
  50. changed, err := s.repo.SweepExpiredProxies(s.ctx, time.Now())
  51. s.Require().NoError(err)
  52. s.Require().GreaterOrEqual(changed, int64(1))
  53. got, _ := s.repo.GetByID(s.ctx, pid)
  54. s.Require().Equal(service.StatusExpired, got.Status)
  55. s.Require().Nil(s.accountProxyID(aid))
  56. var origin *int64
  57. err = scanSingleRow(s.ctx, s.tx, `SELECT proxy_fallback_origin_id FROM accounts WHERE id=$1`, []any{aid}, &origin)
  58. s.Require().NoError(err)
  59. s.Require().NotNil(origin)
  60. s.Require().Equal(pid, *origin)
  61. }
  62. func (s *ProxyExpirySuite) TestSweep_EnqueuesChangedAccountIDsWithoutFullRebuild() {
  63. past := time.Now().Add(-time.Hour)
  64. firstProxyID := s.mkProxy("p-bulk-first", service.FallbackModeDirect, &past, nil)
  65. secondProxyID := s.mkProxy("p-bulk-second", service.FallbackModeDirect, &past, nil)
  66. firstAccountID := s.mkAccountWithProxy(firstProxyID)
  67. secondAccountID := s.mkAccountWithProxy(secondProxyID)
  68. changed, err := s.repo.SweepExpiredProxies(s.ctx, time.Now())
  69. s.Require().NoError(err)
  70. s.Require().EqualValues(2, changed)
  71. var payloadRaw []byte
  72. err = scanSingleRow(s.ctx, s.tx, `
  73. SELECT payload
  74. FROM scheduler_outbox
  75. WHERE event_type=$1
  76. ORDER BY id DESC
  77. LIMIT 1`, []any{service.SchedulerOutboxEventAccountBulkChanged}, &payloadRaw)
  78. s.Require().NoError(err)
  79. var payload struct {
  80. AccountIDs []int64 `json:"account_ids"`
  81. }
  82. s.Require().NoError(json.Unmarshal(payloadRaw, &payload))
  83. s.Require().Equal([]int64{firstAccountID, secondAccountID}, payload.AccountIDs)
  84. var fullRebuildCount int
  85. err = scanSingleRow(s.ctx, s.tx, `
  86. SELECT COUNT(*)
  87. FROM scheduler_outbox
  88. WHERE event_type=$1`, []any{service.SchedulerOutboxEventFullRebuild}, &fullRebuildCount)
  89. s.Require().NoError(err)
  90. s.Require().Zero(fullRebuildCount)
  91. }
  92. func (s *ProxyExpirySuite) TestSweep_ProxyMode_Healthy() {
  93. future := time.Now().Add(24 * time.Hour)
  94. past := time.Now().Add(-time.Hour)
  95. backup := s.mkProxy("p-backup", service.FallbackModeNone, &future, nil)
  96. pid := s.mkProxy("p-main", service.FallbackModeProxy, &past, &backup)
  97. aid := s.mkAccountWithProxy(pid)
  98. _, err := s.repo.SweepExpiredProxies(s.ctx, time.Now())
  99. s.Require().NoError(err)
  100. s.Require().Equal(backup, *s.accountProxyID(aid))
  101. var origin *int64
  102. err = scanSingleRow(s.ctx, s.tx, `SELECT proxy_fallback_origin_id FROM accounts WHERE id=$1`, []any{aid}, &origin)
  103. s.Require().NoError(err)
  104. s.Require().NotNil(origin)
  105. s.Require().Equal(pid, *origin)
  106. }
  107. func (s *ProxyExpirySuite) TestSweep_NoneMode_KeepsAccount() {
  108. past := time.Now().Add(-time.Hour)
  109. pid := s.mkProxy("p-none", service.FallbackModeNone, &past, nil)
  110. aid := s.mkAccountWithProxy(pid)
  111. _, err := s.repo.SweepExpiredProxies(s.ctx, time.Now())
  112. s.Require().NoError(err)
  113. got, _ := s.repo.GetByID(s.ctx, pid)
  114. s.Require().Equal(service.StatusExpired, got.Status)
  115. s.Require().Equal(pid, *s.accountProxyID(aid))
  116. var origin *int64
  117. err = scanSingleRow(s.ctx, s.tx, `SELECT proxy_fallback_origin_id FROM accounts WHERE id=$1`, []any{aid}, &origin)
  118. s.Require().NoError(err)
  119. s.Require().Nil(origin)
  120. }