test: Add tests for cancelled context

These tests are failing and are disabled as a result.

Author: armsnyder

commands_test.go

@@ -4843,6 +4843,22 @@ var _ = Describe("Commands", func() {
 			Expect(err).To(Equal(redis.Nil))
 		})
 
+		// https://github.com/redis/go-redis/issues/2276
+		XDescribe("canceled context", func() {
+			It("should unblock XREAD", func() {
+				ctx2, cancel := context.WithCancel(ctx)
+				errChan := make(chan error, 1)
+				go func() {
+					errChan <- client.XRead(ctx2, &redis.XReadArgs{
+						Streams: []string{"stream", "$"},
+					}).Err()
+				}()
+				Consistently(errChan).ShouldNot(Receive())
+				cancel()
+				Eventually(errChan).Should(Receive(MatchError(context.Canceled)))
+			})
+		})
+
 		Describe("group", func() {
 			BeforeEach(func() {
 				err := client.XGroupCreate(ctx, "stream", "group", "0").Err()
@@ -5023,6 +5039,24 @@ var _ = Describe("Commands", func() {
 				Expect(err).NotTo(HaveOccurred())
 				Expect(n).To(Equal(int64(2)))
 			})
+
+			// https://github.com/redis/go-redis/issues/2276
+			XDescribe("canceled context", func() {
+				It("should unblock XReadGroup", func() {
+					ctx2, cancel := context.WithCancel(ctx)
+					errChan := make(chan error, 1)
+					go func() {
+						errChan <- client.XReadGroup(ctx2, &redis.XReadGroupArgs{
+							Group:    "group",
+							Consumer: "consumer",
+							Streams:  []string{"stream", ">"},
+						}).Err()
+					}()
+					Consistently(errChan).ShouldNot(Receive())
+					cancel()
+					Eventually(errChan).Should(Receive(MatchError(context.Canceled)))
+				})
+			})
 		})
 
 		Describe("xinfo", func() {

pubsub_test.go

@@ -1,6 +1,7 @@
 package redis_test
 
 import (
+	"context"
 	"io"
 	"net"
 	"sync"
@@ -567,4 +568,31 @@ var _ = Describe("PubSub", func() {
 		Expect(msg.Channel).To(Equal("mychannel"))
 		Expect(msg.Payload).To(Equal(text))
 	})
+
+	// https://github.com/redis/go-redis/issues/2276
+	XDescribeTable("canceled context", func(receiveFn func(ctx context.Context, pubsub *redis.PubSub) error) {
+		pubsub := client.Subscribe(ctx, "mychannel")
+		defer pubsub.Close()
+		ctx2, cancel := context.WithCancel(ctx)
+		errChan := make(chan error, 1)
+		go func() {
+			errChan <- receiveFn(ctx2, pubsub)
+		}()
+		Consistently(errChan).ShouldNot(Receive())
+		cancel()
+		Eventually(errChan).Should(Receive(MatchError(context.Canceled)))
+	},
+		Entry("Receive", func(ctx context.Context, pubsub *redis.PubSub) error {
+			_, err := pubsub.Receive(ctx)
+			return err
+		}),
+		Entry("ReceiveMessage", func(ctx context.Context, pubsub *redis.PubSub) error {
+			_, err := pubsub.ReceiveMessage(ctx)
+			return err
+		}),
+		Entry("ReceiveTimeout", func(ctx context.Context, pubsub *redis.PubSub) error {
+			_, err := pubsub.ReceiveTimeout(ctx, time.Hour)
+			return err
+		}),
+	)
 })