scheduler_snapshot_outbox_integration_test.go 1.9 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768
  1. //go:build integration
  2. package repository
  3. import (
  4. "context"
  5. "testing"
  6. "time"
  7. "github.com/Wei-Shaw/sub2api/internal/config"
  8. "github.com/Wei-Shaw/sub2api/internal/service"
  9. "github.com/stretchr/testify/require"
  10. )
  11. func TestSchedulerSnapshotOutboxReplay(t *testing.T) {
  12. ctx := context.Background()
  13. rdb := testRedis(t)
  14. client := testEntClient(t)
  15. _, _ = integrationDB.ExecContext(ctx, "TRUNCATE scheduler_outbox")
  16. accountRepo := newAccountRepositoryWithSQL(client, integrationDB, nil)
  17. outboxRepo := NewSchedulerOutboxRepository(integrationDB)
  18. cache := NewSchedulerCache(rdb)
  19. cfg := &config.Config{
  20. RunMode: config.RunModeStandard,
  21. Gateway: config.GatewayConfig{
  22. Scheduling: config.GatewaySchedulingConfig{
  23. OutboxPollIntervalSeconds: 1,
  24. FullRebuildIntervalSeconds: 0,
  25. DbFallbackEnabled: true,
  26. },
  27. },
  28. }
  29. account := &service.Account{
  30. Name: "outbox-replay-" + time.Now().Format("150405.000000"),
  31. Platform: service.PlatformOpenAI,
  32. Type: service.AccountTypeAPIKey,
  33. Status: service.StatusActive,
  34. Schedulable: true,
  35. Concurrency: 3,
  36. Priority: 1,
  37. Credentials: map[string]any{},
  38. Extra: map[string]any{},
  39. }
  40. require.NoError(t, accountRepo.Create(ctx, account))
  41. require.NoError(t, cache.SetAccount(ctx, account))
  42. svc := service.NewSchedulerSnapshotService(cache, outboxRepo, accountRepo, nil, cfg)
  43. svc.Start()
  44. t.Cleanup(svc.Stop)
  45. require.NoError(t, accountRepo.UpdateLastUsed(ctx, account.ID))
  46. updated, err := accountRepo.GetByID(ctx, account.ID)
  47. require.NoError(t, err)
  48. require.NotNil(t, updated.LastUsedAt)
  49. expectedUnix := updated.LastUsedAt.Unix()
  50. require.Eventually(t, func() bool {
  51. cached, err := cache.GetAccount(ctx, account.ID)
  52. if err != nil || cached == nil || cached.LastUsedAt == nil {
  53. return false
  54. }
  55. return cached.LastUsedAt.Unix() == expectedUnix
  56. }, 5*time.Second, 100*time.Millisecond)
  57. }