api_key_cache_subscriber_test.go 1.3 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849
  1. package repository
  2. import (
  3. "context"
  4. "errors"
  5. "testing"
  6. "time"
  7. "github.com/alicebob/miniredis/v2"
  8. "github.com/redis/go-redis/v9"
  9. "github.com/stretchr/testify/require"
  10. )
  11. func TestAPIKeyCacheSubscriber_BlocksUntilContextCancellation(t *testing.T) {
  12. server := miniredis.RunT(t)
  13. client := redis.NewClient(&redis.Options{Addr: server.Addr()})
  14. defer func() { _ = client.Close() }()
  15. cache := NewAPIKeyCache(client)
  16. ctx, cancel := context.WithCancel(context.Background())
  17. received := make(chan string, 1)
  18. returned := make(chan error, 1)
  19. go func() {
  20. returned <- cache.SubscribeAuthCacheInvalidation(ctx, func(value string) { received <- value })
  21. }()
  22. var value string
  23. require.Eventually(t, func() bool {
  24. require.NoError(t, client.Publish(context.Background(), authCacheInvalidateChannel, "hash").Err())
  25. select {
  26. case value = <-received:
  27. return true
  28. default:
  29. return false
  30. }
  31. }, time.Second, 10*time.Millisecond)
  32. require.Equal(t, "hash", value)
  33. select {
  34. case err := <-returned:
  35. t.Fatalf("subscriber returned while connection was active: %v", err)
  36. default:
  37. }
  38. cancel()
  39. select {
  40. case err := <-returned:
  41. require.True(t, errors.Is(err, context.Canceled))
  42. case <-time.After(time.Second):
  43. t.Fatal("subscriber did not stop after context cancellation")
  44. }
  45. }