chore(wip): improve test coverage

internal/auth/streaming/pool_hook_coverage_test.go

@@ -0,0 +1,162 @@
+package streaming
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/redis/go-redis/v9/auth"
+	"github.com/redis/go-redis/v9/internal/pool"
+)
+
+func idleConn() *pool.Conn {
+	cn := pool.NewConn(nil)
+	cn.GetStateMachine().Transition(pool.StateInitializing)
+	cn.GetStateMachine().Transition(pool.StateIdle)
+	return cn
+}
+
+func TestReAuthPoolHook_OnGet(t *testing.T) {
+	hook := NewReAuthPoolHook(4, time.Second)
+	cn := idleConn()
+	ctx := context.Background()
+
+	// Not marked: accept.
+	if accept, err := hook.OnGet(ctx, cn, false); err != nil || !accept {
+		t.Fatalf("OnGet(unmarked) = %v, %v", accept, err)
+	}
+
+	// Marked for reauth: reject.
+	hook.MarkForReAuth(cn.GetID(), func(error) {})
+	if accept, _ := hook.OnGet(ctx, cn, false); accept {
+		t.Error("OnGet(marked) should reject")
+	}
+
+	// Scheduled reauth: reject. Reach the scheduled state through the public
+	// path: OnPut moves a marked connection to the scheduled set before
+	// re-authenticating it in the background. Block the re-auth callback so
+	// the connection stays scheduled while OnGet is probed.
+	release := make(chan struct{})
+	hook.MarkForReAuth(cn.GetID(), func(error) { <-release })
+	if pooled, removed, err := hook.OnPut(ctx, cn); !pooled || removed || err != nil {
+		t.Fatalf("OnPut(marked) = %v, %v, %v", pooled, removed, err)
+	}
+	if accept, _ := hook.OnGet(ctx, cn, false); accept {
+		t.Error("OnGet(scheduled) should reject")
+	}
+	close(release)
+}
+
+func TestReAuthPoolHook_OnPut(t *testing.T) {
+	hook := NewReAuthPoolHook(4, time.Second)
+	ctx := context.Background()
+
+	// Nil conn is a no-op.
+	if pooled, removed, err := hook.OnPut(ctx, nil); !pooled || removed || err != nil {
+		t.Fatalf("OnPut(nil) = %v, %v, %v", pooled, removed, err)
+	}
+
+	// Conn not marked: no scheduling, pooled.
+	cn := idleConn()
+	if pooled, removed, err := hook.OnPut(ctx, cn); !pooled || removed || err != nil {
+		t.Fatalf("OnPut(unmarked) = %v, %v, %v", pooled, removed, err)
+	}
+
+	// Marked conn: OnPut schedules background reauth which runs the callback.
+	done := make(chan error, 1)
+	hook.MarkForReAuth(cn.GetID(), func(err error) { done <- err })
+	if pooled, removed, err := hook.OnPut(ctx, cn); !pooled || removed || err != nil {
+		t.Fatalf("OnPut(marked) = %v, %v, %v", pooled, removed, err)
+	}
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Errorf("reauth callback err = %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for background reauth")
+	}
+}
+
+func TestReAuthPoolHook_OnRemove(t *testing.T) {
+	mockPool := &mockPooler{}
+	manager := NewManager(mockPool, time.Second)
+	hook := manager.poolHookRef
+
+	cn := idleConn()
+	// Seed a listener and reauth state for this connection.
+	_, _ = manager.Listener(cn, func(*pool.Conn, auth.Credentials) error { return nil }, func(*pool.Conn, error) {})
+	hook.MarkForReAuth(cn.GetID(), func(error) {})
+
+	hook.OnRemove(context.Background(), cn, errors.New("removed"))
+
+	hook.shouldReAuthLock.RLock()
+	_, stillMarked := hook.shouldReAuth[cn.GetID()]
+	hook.shouldReAuthLock.RUnlock()
+	if stillMarked {
+		t.Error("OnRemove should clear shouldReAuth entry")
+	}
+	if _, ok := manager.credentialsListeners.Get(cn.GetID()); ok {
+		t.Error("OnRemove should remove the credentials listener")
+	}
+}
+
+func TestManager_DelegatesAndPoolHook(t *testing.T) {
+	manager := NewManager(&mockPooler{}, time.Second)
+
+	if _, ok := manager.PoolHook().(*ReAuthPoolHook); !ok {
+		t.Fatalf("PoolHook returned %T", manager.PoolHook())
+	}
+
+	cn := idleConn()
+	manager.MarkForReAuth(cn, func(error) {})
+	manager.poolHookRef.shouldReAuthLock.RLock()
+	_, marked := manager.poolHookRef.shouldReAuth[cn.GetID()]
+	manager.poolHookRef.shouldReAuthLock.RUnlock()
+	if !marked {
+		t.Error("Manager.MarkForReAuth should delegate to the pool hook")
+	}
+
+	// RemoveListener clears the registry entry.
+	manager.credentialsListeners.Add(cn.GetID(), &ConnReAuthCredentialsListener{})
+	manager.RemoveListener(cn.GetID())
+	if _, ok := manager.credentialsListeners.Get(cn.GetID()); ok {
+		t.Error("RemoveListener should remove the entry")
+	}
+}
+
+func TestConnReAuthCredentialsListener_OnNextOnError(t *testing.T) {
+	manager := NewManager(&mockPooler{}, time.Second)
+	cn := idleConn()
+
+	var gotErr error
+	l, err := manager.Listener(cn,
+		func(*pool.Conn, auth.Credentials) error { return nil },
+		func(_ *pool.Conn, e error) { gotErr = e },
+	)
+	if err != nil {
+		t.Fatalf("Listener error: %v", err)
+	}
+	cl := l.(*ConnReAuthCredentialsListener)
+
+	// OnError delegates to onErr.
+	sentinel := errors.New("stream error")
+	cl.OnError(sentinel)
+	if gotErr != sentinel {
+		t.Errorf("OnError did not propagate; got %v", gotErr)
+	}
+
+	// OnNext on a live connection marks it for reauth.
+	cl.OnNext(nil)
+	manager.poolHookRef.shouldReAuthLock.RLock()
+	_, marked := manager.poolHookRef.shouldReAuth[cn.GetID()]
+	manager.poolHookRef.shouldReAuthLock.RUnlock()
+	if !marked {
+		t.Error("OnNext should mark the connection for reauth")
+	}
+
+	// Guard branches: nil conn / nil onErr are no-ops.
+	(&ConnReAuthCredentialsListener{}).OnNext(nil)
+	(&ConnReAuthCredentialsListener{}).OnError(sentinel)
+}

internal/helpers_test.go

@@ -0,0 +1,142 @@
+package internal
+
+import (
+	"bytes"
+	"context"
+	"log"
+	"testing"
+	"time"
+)
+
+func TestToInteger(t *testing.T) {
+	tests := []struct {
+		in   interface{}
+		want int
+	}{
+		{42, 42},
+		{int64(7), 7},
+		{"123", 123},
+		{"not-a-number", 0},
+		{3.14, 0},
+		{nil, 0},
+	}
+	for _, tt := range tests {
+		if got := ToInteger(tt.in); got != tt.want {
+			t.Errorf("ToInteger(%v) = %d, want %d", tt.in, got, tt.want)
+		}
+	}
+}
+
+func TestToFloat(t *testing.T) {
+	tests := []struct {
+		in   interface{}
+		want float64
+	}{
+		{3.5, 3.5},
+		{"2.25", 2.25},
+		{"bad", 0},
+		{42, 0},
+		{nil, 0},
+	}
+	for _, tt := range tests {
+		if got := ToFloat(tt.in); got != tt.want {
+			t.Errorf("ToFloat(%v) = %v, want %v", tt.in, got, tt.want)
+		}
+	}
+}
+
+func TestToString(t *testing.T) {
+	if got := ToString("hello"); got != "hello" {
+		t.Errorf("ToString(string) = %q, want %q", got, "hello")
+	}
+	if got := ToString(42); got != "" {
+		t.Errorf("ToString(int) = %q, want empty", got)
+	}
+	if got := ToString(nil); got != "" {
+		t.Errorf("ToString(nil) = %q, want empty", got)
+	}
+}
+
+func TestToStringSlice(t *testing.T) {
+	in := []interface{}{"a", "b", 3}
+	got := ToStringSlice(in)
+	want := []string{"a", "b", ""}
+	if len(got) != len(want) {
+		t.Fatalf("ToStringSlice len = %d, want %d", len(got), len(want))
+	}
+	for i := range want {
+		if got[i] != want[i] {
+			t.Errorf("ToStringSlice[%d] = %q, want %q", i, got[i], want[i])
+		}
+	}
+	if ToStringSlice("not-a-slice") != nil {
+		t.Errorf("ToStringSlice(non-slice) should be nil")
+	}
+}
+
+func TestReplaceSpaces(t *testing.T) {
+	if got := ReplaceSpaces("go 1 24"); got != "go-1-24" {
+		t.Errorf("ReplaceSpaces = %q, want %q", got, "go-1-24")
+	}
+	if got := ReplaceSpaces("nospaces"); got != "nospaces" {
+		t.Errorf("ReplaceSpaces = %q, want %q", got, "nospaces")
+	}
+}
+
+func TestSleep(t *testing.T) {
+	if err := Sleep(context.Background(), time.Millisecond); err != nil {
+		t.Errorf("Sleep returned error: %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if err := Sleep(ctx, time.Hour); err == nil {
+		t.Errorf("Sleep with cancelled context should return error")
+	}
+}
+
+func TestAppendArg(t *testing.T) {
+	now := time.Date(2024, 1, 2, 3, 4, 5, 0, time.UTC)
+	tests := []struct {
+		in   interface{}
+		want string
+	}{
+		{nil, "<nil>"},
+		{"str", "str"},
+		{[]byte("bytes"), "bytes"},
+		{int(-1), "-1"},
+		{int8(2), "2"},
+		{int16(3), "3"},
+		{int32(4), "4"},
+		{int64(5), "5"},
+		{uint(6), "6"},
+		{uint8(7), "7"},
+		{uint16(8), "8"},
+		{uint32(9), "9"},
+		{uint64(10), "10"},
+		{float32(1.5), "1.5"},
+		{float64(2.5), "2.5"},
+		{true, "true"},
+		{false, "false"},
+		{now, "2024-01-02T03:04:05Z"},
+		{struct{ X int }{1}, "{1}"},
+	}
+	for _, tt := range tests {
+		got := string(AppendArg(nil, tt.in))
+		if got != tt.want {
+			t.Errorf("AppendArg(%v) = %q, want %q", tt.in, got, tt.want)
+		}
+	}
+}
+
+func TestDefaultLoggerPrintf(t *testing.T) {
+	var buf bytes.Buffer
+	l := &DefaultLogger{log: log.New(&buf, "", 0)}
+	l.Printf(context.Background(), "hello %s", "world")
+	if got := buf.String(); got != "hello world\n" {
+		t.Errorf("DefaultLogger.Printf wrote %q", got)
+	}
+	if NewDefaultLogger() == nil {
+		t.Errorf("NewDefaultLogger returned nil")
+	}
+}