hanzokv/go
1
0
Fork 0
mirror of https://github.com/hanzokv/go.git synced 2026-09-24 00:31:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(fmt): proper formatting (#3670)

Browse Source 
* all: proper formatting

* .golangci.yml: add linter for format check

* ran `golangci-lint fmt`

---------

Co-authored-by: Nedyalko Dyakov <[email protected]>
This commit is contained in:
iliya
2026-01-13 14:48:02 +02:00
committed by GitHub
co-authored by Nedyalko Dyakov
parent 7a57d6374e
commit 2b8d05ed37
19 changed files with 187 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.golangci.yml
+2
View File
@@ -26,6 +26,8 @@ linters:
- builtin$
- examples$
formatters:
enable:
- gofmt
exclusions:
generated: lax
paths:
auth/reauth_credentials_listener.go
+1 -1
View File
@@ -44,4 +44,4 @@ func NewReAuthCredentialsListener(reAuth func(credentials Credentials) error, on
}
// Ensure ReAuthCredentialsListener implements the CredentialsListener interface.
var _ CredentialsListener = (*ReAuthCredentialsListener)(nil)
var _ CredentialsListener = (*ReAuthCredentialsListener)(nil)
command_digest_test.go
-1
View File
@@ -115,4 +115,3 @@ func (c *mockConn) Read(p []byte) (n int, err error) {
c.pos += n
return n, nil
}
digest_test.go
-1
View File
@@ -262,4 +262,3 @@ func TestDelExArgsWithDigest(t *testing.T) {
t.Errorf("Key should not exist after successful delete")
}
}
example/cluster-mget/main.go
+30 -30
View File
@@ -49,43 +49,43 @@ func main() {
}
/*
// Retrieve all keys using MGET
fmt.Println("\n=== Retrieving keys with MGET ===")
result, err := rdb.MGet(ctx, keys...).Result()
if err != nil {
panic(fmt.Sprintf("Failed to execute MGET: %v", err))
}
// Retrieve all keys using MGET
fmt.Println("\n=== Retrieving keys with MGET ===")
result, err := rdb.MGet(ctx, keys...).Result()
if err != nil {
panic(fmt.Sprintf("Failed to execute MGET: %v", err))
}
*/
/*
// Validate the results
fmt.Println("\n=== Validating MGET results ===")
allValid := true
for i, val := range result {
expectedValue := values[i]
actualValue, ok := val.(string)
// Validate the results
fmt.Println("\n=== Validating MGET results ===")
allValid := true
for i, val := range result {
expectedValue := values[i]
actualValue, ok := val.(string)
if !ok {
fmt.Printf("✗ %s: expected string, got %T\n", keys[i], val)
allValid = false
continue
if !ok {
fmt.Printf("✗ %s: expected string, got %T\n", keys[i], val)
allValid = false
continue
}
if actualValue != expectedValue {
fmt.Printf("✗ %s: expected '%s', got '%s'\n", keys[i], expectedValue, actualValue)
allValid = false
} else {
fmt.Printf("✓ %s: %s\n", keys[i], actualValue)
}
}
if actualValue != expectedValue {
fmt.Printf("✗ %s: expected '%s', got '%s'\n", keys[i], expectedValue, actualValue)
allValid = false
// Print summary
fmt.Println("\n=== Summary ===")
if allValid {
fmt.Println("✓ All values retrieved successfully and match expected values!")
} else {
fmt.Printf("✓ %s: %s\n", keys[i], actualValue)
fmt.Println("✗ Some values did not match expected values")
}
}
// Print summary
fmt.Println("\n=== Summary ===")
if allValid {
fmt.Println("✓ All values retrieved successfully and match expected values!")
} else {
fmt.Println("✗ Some values did not match expected values")
}
*/
// Clean up - delete the keys
@@ -97,7 +97,7 @@ func main() {
}
fmt.Println("✓ Cleanup complete")
err := rdb.Set(ctx, "{tag}exists", "asdf",0).Err()
err := rdb.Set(ctx, "{tag}exists", "asdf", 0).Err()
if err != nil {
panic(err)
}
example/digest-optimistic-locking/main.go
-1
View File
@@ -242,4 +242,3 @@ func clientSideDigestExample(ctx context.Context, rdb *redis.Client) {
fmt.Println("✓ Binary digest matches!")
}
}
extra/redisotel/metrics.go
+1 -1
View File
@@ -330,7 +330,7 @@ func milliseconds(d time.Duration) float64 {
func statusAttr(err error) attribute.KeyValue {
if err != nil {
if err == redis.Nil {
if err == redis.Nil {
return attribute.String("status", "nil")
}
return attribute.String("status", "error")
internal/auth/streaming/manager_test.go
+26 -27
View File
@@ -13,13 +13,13 @@ import (
func TestManager_Listener_ReturnsNewListener(t *testing.T) {
// Create a mock pool
mockPool := &mockPooler{}
// Create manager
manager := NewManager(mockPool, time.Second)
// Create a mock connection
conn := &pool.Conn{}
// Mock functions
reAuth := func(cn *pool.Conn, creds auth.Credentials) error {
return nil
@@ -27,35 +27,35 @@ func TestManager_Listener_ReturnsNewListener(t *testing.T) {
onErr := func(cn *pool.Conn, err error) {
}
// Get listener - this should create a new one
listener, err := manager.Listener(conn, reAuth, onErr)
// Verify no error
if err != nil {
t.Fatalf("Expected no error, got: %v", err)
}
// Verify listener is not nil (this was the bug!)
if listener == nil {
t.Fatal("Expected listener to be non-nil, but got nil")
}
// Verify it's the correct type
if _, ok := listener.(*ConnReAuthCredentialsListener); !ok {
t.Fatalf("Expected listener to be *ConnReAuthCredentialsListener, got %T", listener)
}
// Get the same listener again - should return the existing one
listener2, err := manager.Listener(conn, reAuth, onErr)
if err != nil {
t.Fatalf("Expected no error on second call, got: %v", err)
}
if listener2 == nil {
t.Fatal("Expected listener2 to be non-nil")
}
// Should be the same instance
if listener != listener2 {
t.Error("Expected to get the same listener instance on second call")
@@ -66,17 +66,17 @@ func TestManager_Listener_ReturnsNewListener(t *testing.T) {
func TestManager_Listener_NilConn(t *testing.T) {
mockPool := &mockPooler{}
manager := NewManager(mockPool, time.Second)
listener, err := manager.Listener(nil, nil, nil)
if err == nil {
t.Fatal("Expected error when conn is nil, got nil")
}
if listener != nil {
t.Error("Expected listener to be nil when error occurs")
}
expectedErr := "poolCn cannot be nil"
if err.Error() != expectedErr {
t.Errorf("Expected error message %q, got %q", expectedErr, err.Error())
@@ -86,17 +86,16 @@ func TestManager_Listener_NilConn(t *testing.T) {
// Mock pooler for testing
type mockPooler struct{}
func (m *mockPooler) NewConn(ctx context.Context) (*pool.Conn, error) { return nil, nil }
func (m *mockPooler) CloseConn(*pool.Conn) error { return nil }
func (m *mockPooler) Get(ctx context.Context) (*pool.Conn, error) { return nil, nil }
func (m *mockPooler) Put(ctx context.Context, conn *pool.Conn) {}
func (m *mockPooler) Remove(ctx context.Context, conn *pool.Conn, reason error) {}
func (m *mockPooler) NewConn(ctx context.Context) (*pool.Conn, error) { return nil, nil }
func (m *mockPooler) CloseConn(*pool.Conn) error { return nil }
func (m *mockPooler) Get(ctx context.Context) (*pool.Conn, error) { return nil, nil }
func (m *mockPooler) Put(ctx context.Context, conn *pool.Conn) {}
func (m *mockPooler) Remove(ctx context.Context, conn *pool.Conn, reason error) {}
func (m *mockPooler) RemoveWithoutTurn(ctx context.Context, conn *pool.Conn, reason error) {}
func (m *mockPooler) Len() int { return 0 }
func (m *mockPooler) IdleLen() int { return 0 }
func (m *mockPooler) Stats() *pool.Stats { return &pool.Stats{} }
func (m *mockPooler) Size() int { return 10 }
func (m *mockPooler) AddPoolHook(hook pool.PoolHook) {}
func (m *mockPooler) RemovePoolHook(hook pool.PoolHook) {}
func (m *mockPooler) Close() error { return nil }
func (m *mockPooler) Len() int { return 0 }
func (m *mockPooler) IdleLen() int { return 0 }
func (m *mockPooler) Stats() *pool.Stats { return &pool.Stats{} }
func (m *mockPooler) Size() int { return 10 }
func (m *mockPooler) AddPoolHook(hook pool.PoolHook) {}
func (m *mockPooler) RemovePoolHook(hook pool.PoolHook) {}
func (m *mockPooler) Close() error { return nil }
internal/auth/streaming/pool_hook_state_test.go
+35 -36
View File
@@ -14,48 +14,48 @@ import (
func TestReAuthOnlyWhenIdle(t *testing.T) {
// Create a connection
cn := pool.NewConn(nil)
// Initialize to IDLE state
cn.GetStateMachine().Transition(pool.StateInitializing)
cn.GetStateMachine().Transition(pool.StateIdle)
// Simulate connection being acquired (IDLE → IN_USE)
if !cn.CompareAndSwapUsed(false, true) {
t.Fatal("Failed to acquire connection")
}
// Verify state is IN_USE
if state := cn.GetStateMachine().GetState(); state != pool.StateInUse {
t.Errorf("Expected state IN_USE, got %s", state)
}
// Try to transition to UNUSABLE (for reauth) - should fail
_, err := cn.GetStateMachine().TryTransition([]pool.ConnState{pool.StateIdle}, pool.StateUnusable)
if err == nil {
t.Error("Expected error when trying to transition IN_USE → UNUSABLE, but got none")
}
// Verify state is still IN_USE
if state := cn.GetStateMachine().GetState(); state != pool.StateInUse {
t.Errorf("Expected state to remain IN_USE, got %s", state)
}
// Release connection (IN_USE → IDLE)
if !cn.CompareAndSwapUsed(true, false) {
t.Fatal("Failed to release connection")
}
// Verify state is IDLE
if state := cn.GetStateMachine().GetState(); state != pool.StateIdle {
t.Errorf("Expected state IDLE, got %s", state)
}
// Now try to transition to UNUSABLE - should succeed
_, err = cn.GetStateMachine().TryTransition([]pool.ConnState{pool.StateIdle}, pool.StateUnusable)
if err != nil {
t.Errorf("Failed to transition IDLE → UNUSABLE: %v", err)
}
// Verify state is UNUSABLE
if state := cn.GetStateMachine().GetState(); state != pool.StateUnusable {
t.Errorf("Expected state UNUSABLE, got %s", state)
@@ -67,30 +67,30 @@ func TestReAuthOnlyWhenIdle(t *testing.T) {
func TestReAuthWaitsForConnectionToBeIdle(t *testing.T) {
// Create a connection
cn := pool.NewConn(nil)
// Initialize to IDLE state
cn.GetStateMachine().Transition(pool.StateInitializing)
cn.GetStateMachine().Transition(pool.StateIdle)
// Simulate connection being acquired (IDLE → IN_USE)
if !cn.CompareAndSwapUsed(false, true) {
t.Fatal("Failed to acquire connection")
}
// Track re-auth attempts
var reAuthAttempts atomic.Int32
var reAuthSucceeded atomic.Bool
// Start a goroutine that tries to acquire the connection for re-auth
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
// Try to acquire for re-auth with timeout
timeout := time.After(2 * time.Second)
acquired := false
for !acquired {
select {
case <-timeout:
@@ -110,38 +110,38 @@ func TestReAuthWaitsForConnectionToBeIdle(t *testing.T) {
}
}
}
// Release the connection
cn.GetStateMachine().Transition(pool.StateIdle)
}()
// Keep connection IN_USE for 500ms
time.Sleep(500 * time.Millisecond)
// Verify re-auth hasn't succeeded yet (connection is still IN_USE)
if reAuthSucceeded.Load() {
t.Error("Re-auth succeeded while connection was IN_USE")
}
// Verify there were multiple attempts
attempts := reAuthAttempts.Load()
if attempts < 2 {
t.Errorf("Expected multiple re-auth attempts, got %d", attempts)
}
// Release connection (IN_USE → IDLE)
if !cn.CompareAndSwapUsed(true, false) {
t.Fatal("Failed to release connection")
}
// Wait for re-auth to complete
wg.Wait()
// Verify re-auth succeeded after connection became IDLE
if !reAuthSucceeded.Load() {
t.Error("Re-auth did not succeed after connection became IDLE")
}
// Verify final state is IDLE
if state := cn.GetStateMachine().GetState(); state != pool.StateIdle {
t.Errorf("Expected final state IDLE, got %s", state)
@@ -153,15 +153,15 @@ func TestReAuthWaitsForConnectionToBeIdle(t *testing.T) {
func TestConcurrentReAuthAndUsage(t *testing.T) {
// Create a connection
cn := pool.NewConn(nil)
// Initialize to IDLE state
cn.GetStateMachine().Transition(pool.StateInitializing)
cn.GetStateMachine().Transition(pool.StateIdle)
var wg sync.WaitGroup
var usageCount atomic.Int32
var reAuthCount atomic.Int32
// Goroutine 1: Simulate normal usage (acquire/release)
wg.Add(1)
go func() {
@@ -178,7 +178,7 @@ func TestConcurrentReAuthAndUsage(t *testing.T) {
time.Sleep(1 * time.Millisecond)
}
}()
// Goroutine 2: Simulate re-auth attempts
wg.Add(1)
go func() {
@@ -196,9 +196,9 @@ func TestConcurrentReAuthAndUsage(t *testing.T) {
time.Sleep(2 * time.Millisecond)
}
}()
wg.Wait()
// Verify both operations happened
if usageCount.Load() == 0 {
t.Error("No successful usage operations")
@@ -206,9 +206,9 @@ func TestConcurrentReAuthAndUsage(t *testing.T) {
if reAuthCount.Load() == 0 {
t.Error("No successful re-auth operations")
}
t.Logf("Usage operations: %d, Re-auth operations: %d", usageCount.Load(), reAuthCount.Load())
// Verify final state is IDLE
if state := cn.GetStateMachine().GetState(); state != pool.StateIdle {
t.Errorf("Expected final state IDLE, got %s", state)
@@ -219,23 +219,22 @@ func TestConcurrentReAuthAndUsage(t *testing.T) {
func TestReAuthRespectsClosed(t *testing.T) {
// Create a connection
cn := pool.NewConn(nil)
// Initialize to IDLE state
cn.GetStateMachine().Transition(pool.StateInitializing)
cn.GetStateMachine().Transition(pool.StateIdle)
// Close the connection
cn.GetStateMachine().Transition(pool.StateClosed)
// Try to transition to UNUSABLE - should fail
_, err := cn.GetStateMachine().TryTransition([]pool.ConnState{pool.StateIdle}, pool.StateUnusable)
if err == nil {
t.Error("Expected error when trying to transition CLOSED → UNUSABLE, but got none")
}
// Verify state is still CLOSED
if state := cn.GetStateMachine().GetState(); state != pool.StateClosed {
t.Errorf("Expected state to remain CLOSED, got %s", state)
}
}
internal/pool/conn_relaxed_timeout_test.go
+1 -1
View File
@@ -78,7 +78,7 @@ func TestRelaxedTimeoutCounterRaceCondition(t *testing.T) {
if count := cn.relaxedCounter.Load(); count != 0 {
t.Errorf("Expected relaxed counter to be 0 after concurrent clearing, got %d", count)
}
// Verify timeouts are actually cleared
if timeout := cn.relaxedReadTimeoutNs.Load(); timeout != 0 {
t.Errorf("Expected relaxed read timeout to be cleared, got %d", timeout)
internal/pool/conn_state.go
+7 -7
View File
@@ -13,11 +13,12 @@ import (
// States are designed to be lightweight and fast to check.
//
// State Transitions:
// CREATED → INITIALIZING → IDLE ⇄ IN_USE
// ↓
// UNUSABLE (handoff/reauth)
// ↓
// IDLE/CLOSED
//
// CREATED → INITIALIZING → IDLE ⇄ IN_USE
// ↓
// UNUSABLE (handoff/reauth)
// ↓
// IDLE/CLOSED
type ConnState uint32
const (
@@ -120,7 +121,7 @@ type ConnStateMachine struct {
// FIFO queue for waiters - only locked during waiter add/remove/notify
mu sync.Mutex
waiters *list.List // List of *waiter
waiters *list.List // List of *waiter
waiterCount atomic.Int32 // Fast lock-free check for waiters (avoids mutex in hot path)
}
@@ -340,4 +341,3 @@ func (sm *ConnStateMachine) notifyWaiters() {
}
}
}
internal/pool/conn_state_alloc_test.go
-1
View File
@@ -166,4 +166,3 @@ func TestAllTryTransitionUsePredefinedSlices(t *testing.T) {
t.Errorf("Expected <= 2 allocations with predefined slices, got %.2f", allocs)
}
}
internal/pool/conn_state_test.go
+5 -12
View File
@@ -33,11 +33,11 @@ func TestConnStateMachine_Transition(t *testing.T) {
func TestConnStateMachine_TryTransition(t *testing.T) {
tests := []struct {
name string
initialState ConnState
validStates []ConnState
targetState ConnState
expectError bool
name string
initialState ConnState
validStates []ConnState
targetState ConnState
expectError bool
}{
{
name: "valid transition from CREATED to INITIALIZING",
@@ -229,8 +229,6 @@ func TestConnStateMachine_ConcurrentAccess(t *testing.T) {
t.Logf("Successful transitions: %d out of %d attempts", successCount.Load(), numGoroutines*numIterations)
}
func TestConnStateMachine_StateString(t *testing.T) {
tests := []struct {
state ConnState
@@ -541,8 +539,6 @@ func BenchmarkConnStateMachine_TryTransition(b *testing.B) {
}
}
func TestConnStateMachine_IdleInUseTransitions(t *testing.T) {
sm := NewConnStateMachine()
@@ -640,7 +636,6 @@ func TestConn_UsedMethods(t *testing.T) {
}
}
func TestConnStateMachine_UnusableState(t *testing.T) {
sm := NewConnStateMachine()
@@ -738,5 +733,3 @@ func TestConn_UsableUnusable(t *testing.T) {
t.Errorf("expected state IDLE after SetUsable(true), got %s", state)
}
}
internal/pool/double_freeturn_simple_test.go
+14 -15
View File
@@ -26,15 +26,15 @@ import (
// - With the fix: Only PoolSize succeed
func TestDoubleFreeTurnSimple(t *testing.T) {
ctx := context.Background()
var dialCount atomic.Int32
dialBComplete := make(chan struct{})
requestBGotConn := make(chan struct{})
requestBCalledPut := make(chan struct{})
controlledDialer := func(ctx context.Context) (net.Conn, error) {
count := dialCount.Add(1)
if count == 1 {
// Dial A: takes 150ms
time.Sleep(150 * time.Millisecond)
@@ -48,10 +48,10 @@ func TestDoubleFreeTurnSimple(t *testing.T) {
// Other dials: fast
time.Sleep(10 * time.Millisecond)
}
return newDummyConn(), nil
}
testPool := pool.NewConnPool(&pool.Options{
Dialer: controlledDialer,
PoolSize: 2, // Only 2 concurrent operations allowed
@@ -60,38 +60,38 @@ func TestDoubleFreeTurnSimple(t *testing.T) {
PoolTimeout: 1 * time.Second,
})
defer testPool.Close()
// Request A: Short timeout (100ms), will timeout before dial completes (150ms)
go func() {
shortCtx, cancel := context.WithTimeout(ctx, 100*time.Millisecond)
defer cancel()
_, err := testPool.Get(shortCtx)
if err != nil {
t.Logf("Request A: Timed out as expected: %v", err)
}
}()
// Wait for Request A to start
time.Sleep(20 * time.Millisecond)
// Request B: Long timeout, will receive connection from Request A's dial
requestBDone := make(chan struct{})
go func() {
defer close(requestBDone)
longCtx, cancel := context.WithTimeout(ctx, 1*time.Second)
defer cancel()
cn, err := testPool.Get(longCtx)
if err != nil {
t.Errorf("Request B: Should have received connection but got error: %v", err)
return
}
t.Logf("Request B: Got connection from Request A's dial")
close(requestBGotConn)
// Wait for dial B to complete
<-dialBComplete
@@ -110,7 +110,7 @@ func TestDoubleFreeTurnSimple(t *testing.T) {
testPool.Put(ctx, cn)
t.Logf("Request B: Put() called")
}()
// Wait for Request B to get the connection
<-requestBGotConn
@@ -155,4 +155,3 @@ func TestDoubleFreeTurnSimple(t *testing.T) {
t.Logf("Unexpected QueueLen: %d (expected 1 with fix, 0 with bug)", queueLen)
}
}
internal/pool/double_freeturn_test.go
+7 -8
View File
@@ -47,7 +47,7 @@ func TestDoubleFreeTurnBug(t *testing.T) {
opt := &Options{
Dialer: slowDialer,
PoolSize: 10, // Small pool to make bug easier to trigger
PoolSize: 10, // Small pool to make bug easier to trigger
MaxConcurrentDials: 10,
MinIdleConns: 0,
PoolTimeout: 100 * time.Millisecond,
@@ -68,14 +68,14 @@ func TestDoubleFreeTurnBug(t *testing.T) {
// 8. Put() calls freeTurn() - SECOND FREE (BUG!)
var wg sync.WaitGroup
// Request A: Short timeout, will timeout before dial completes
wg.Add(1)
go func() {
defer wg.Done()
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()
cn, err := connPool.Get(ctx)
if err != nil {
// Expected to timeout
@@ -87,17 +87,17 @@ func TestDoubleFreeTurnBug(t *testing.T) {
putCount.Add(1)
}
}()
// Wait a bit for Request A to start dialing
time.Sleep(10 * time.Millisecond)
// Request B: Long timeout, will receive the connection from putIdleConn
wg.Add(1)
go func() {
defer wg.Done()
ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
defer cancel()
cn, err := connPool.Get(ctx)
if err != nil {
t.Errorf("Request B should have succeeded but got error: %v", err)
@@ -109,7 +109,7 @@ func TestDoubleFreeTurnBug(t *testing.T) {
putCount.Add(1)
}
}()
wg.Wait()
// Check results
@@ -226,4 +226,3 @@ func TestDoubleFreeTurnHighConcurrency(t *testing.T) {
t.Logf("✓ High concurrency test completed")
t.Logf("Note: This test exercises the putIdleConn delivery path where the bug occurs")
}
internal/routing/aggregator_test.go
+50 -51
View File
@@ -8,78 +8,78 @@ import (
func TestAggLogicalAndAggregator(t *testing.T) {
t.Run("all true values", func(t *testing.T) {
agg := NewResponseAggregator(RespAggLogicalAnd, "")
err := agg.Add(true, nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
err = agg.Add(int64(1), nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
err = agg.Add(1, nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
result, err := agg.Result()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if result != true {
t.Errorf("expected true, got %v", result)
}
})
t.Run("one false value", func(t *testing.T) {
agg := NewResponseAggregator(RespAggLogicalAnd, "")
err := agg.Add(true, nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
err = agg.Add(false, nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
err = agg.Add(true, nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
result, err := agg.Result()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if result != false {
t.Errorf("expected false, got %v", result)
}
})
t.Run("no results", func(t *testing.T) {
agg := NewResponseAggregator(RespAggLogicalAnd, "")
_, err := agg.Result()
if err != ErrAndAggregation {
t.Errorf("expected ErrAndAggregation, got %v", err)
}
})
t.Run("with error", func(t *testing.T) {
agg := NewResponseAggregator(RespAggLogicalAnd, "")
testErr := errors.New("test error")
err := agg.Add(nil, testErr)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
_, err = agg.Result()
if err != testErr {
t.Errorf("expected test error, got %v", err)
@@ -90,78 +90,78 @@ func TestAggLogicalAndAggregator(t *testing.T) {
func TestAggLogicalOrAggregator(t *testing.T) {
t.Run("all false values", func(t *testing.T) {
agg := NewResponseAggregator(RespAggLogicalOr, "")
err := agg.Add(false, nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
err = agg.Add(int64(0), nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
err = agg.Add(0, nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
result, err := agg.Result()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if result != false {
t.Errorf("expected false, got %v", result)
}
})
t.Run("one true value", func(t *testing.T) {
agg := NewResponseAggregator(RespAggLogicalOr, "")
err := agg.Add(false, nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
err = agg.Add(true, nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
err = agg.Add(false, nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
result, err := agg.Result()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if result != true {
t.Errorf("expected true, got %v", result)
}
})
t.Run("no results", func(t *testing.T) {
agg := NewResponseAggregator(RespAggLogicalOr, "")
_, err := agg.Result()
if err != ErrOrAggregation {
t.Errorf("expected ErrOrAggregation, got %v", err)
}
})
t.Run("with error", func(t *testing.T) {
agg := NewResponseAggregator(RespAggLogicalOr, "")
testErr := errors.New("test error")
err := agg.Add(nil, testErr)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
_, err = agg.Result()
if err != testErr {
t.Errorf("expected test error, got %v", err)
@@ -172,47 +172,47 @@ func TestAggLogicalOrAggregator(t *testing.T) {
func TestAggLogicalAndBatchAdd(t *testing.T) {
t.Run("batch add all true", func(t *testing.T) {
agg := NewResponseAggregator(RespAggLogicalAnd, "")
results := map[string]AggregatorResErr{
"key1": {Result: true, Err: nil},
"key2": {Result: int64(1), Err: nil},
"key3": {Result: 1, Err: nil},
}
err := agg.BatchAdd(results)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
result, err := agg.Result()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if result != true {
t.Errorf("expected true, got %v", result)
}
})
t.Run("batch add with false", func(t *testing.T) {
agg := NewResponseAggregator(RespAggLogicalAnd, "")
results := map[string]AggregatorResErr{
"key1": {Result: true, Err: nil},
"key2": {Result: false, Err: nil},
"key3": {Result: true, Err: nil},
}
err := agg.BatchAdd(results)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
result, err := agg.Result()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if result != false {
t.Errorf("expected false, got %v", result)
}
@@ -222,50 +222,49 @@ func TestAggLogicalAndBatchAdd(t *testing.T) {
func TestAggLogicalOrBatchAdd(t *testing.T) {
t.Run("batch add all false", func(t *testing.T) {
agg := NewResponseAggregator(RespAggLogicalOr, "")
results := map[string]AggregatorResErr{
"key1": {Result: false, Err: nil},
"key2": {Result: int64(0), Err: nil},
"key3": {Result: 0, Err: nil},
}
err := agg.BatchAdd(results)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
result, err := agg.Result()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if result != false {
t.Errorf("expected false, got %v", result)
}
})
t.Run("batch add with true", func(t *testing.T) {
agg := NewResponseAggregator(RespAggLogicalOr, "")
results := map[string]AggregatorResErr{
"key1": {Result: false, Err: nil},
"key2": {Result: true, Err: nil},
"key3": {Result: false, Err: nil},
}
err := agg.BatchAdd(results)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
result, err := agg.Result()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if result != true {
t.Errorf("expected true, got %v", result)
}
})
}
internal/semaphore.go
+1 -1
View File
@@ -190,4 +190,4 @@ func (s *FIFOSemaphore) Close() {
// Len returns the current number of acquired tokens.
func (s *FIFOSemaphore) Len() int32 {
return s.max - int32(len(s.tokens))
}
}
maintnotifications/e2e/notiftracker_test.go
+1
View File
@@ -81,6 +81,7 @@ func (tnh *TrackingNotificationsHook) Clear() {
tnh.migratedCount.Store(0)
tnh.failingOverCount.Store(0)
}
// wait for notification in prehook
func (tnh *TrackingNotificationsHook) FindOrWaitForNotification(notificationType string, timeout time.Duration) (notification []interface{}, found bool) {
if notification, found := tnh.FindNotification(notificationType); found {
push/processor_unit_test.go
+6 -6
View File
@@ -157,9 +157,9 @@ func TestProcessPendingNotificationsNilReader(t *testing.T) {
// TestWillHandleNotificationInClient tests the notification filtering logic
func TestWillHandleNotificationInClient(t *testing.T) {
testCases := []struct {
name string
name string
notificationType string
shouldHandle bool
shouldHandle bool
}{
// Pub/Sub notifications (should be handled in client)
{"message", "message", true},
@@ -243,7 +243,7 @@ func TestProcessorErrorHandlingUnit(t *testing.T) {
// TestProcessorConcurrentAccess tests concurrent access to processor
func TestProcessorConcurrentAccess(t *testing.T) {
processor := NewProcessor()
t.Run("ConcurrentRegisterAndGet", func(t *testing.T) {
done := make(chan bool, 2)
@@ -284,10 +284,10 @@ func TestProcessorInterfaceCompliance(t *testing.T) {
// UnitTestHandler is a test implementation of NotificationHandler
type UnitTestHandler struct {
name string
name string
lastNotification []interface{}
errorToReturn error
callCount int
errorToReturn error
callCount int
}
func (h *UnitTestHandler) HandlePushNotification(ctx context.Context, handlerCtx NotificationHandlerContext, notification []interface{}) error {