Flaky unit tests fix

blockchain/subscription_test.go

@@ -168,7 +168,7 @@ func TestSubscription_NilEventAfterClosingSubscription(t *testing.T) {
 
 	receivedEvtCount := 0
 
-	worker := func(sub *subscription, expectedBlockCount uint8) {
+	worker := func(sub *subscription, expectedBlockCount int) {
 		defer wg.Done()
 
 		startTime := time.Now().UTC()
@@ -178,7 +178,9 @@ func TestSubscription_NilEventAfterClosingSubscription(t *testing.T) {
 			evt := sub.GetEvent()
 			if evt != nil {
 				receivedEvtCount++
-			} else {
+			}
+
+			if expectedBlockCount <= receivedEvtCount {
 				return
 			}
 

consensus/polybft/block_builder_test.go

@@ -113,7 +113,7 @@ func TestBlockBuilder_BuildBlockTxOneFailedTxAndOneTakesTooMuchGas(t *testing.T)
 	}
 
 	bb := NewBlockBuilder(&BlockBuilderParams{
-		BlockTime: time.Millisecond * 100,
+		BlockTime: time.Millisecond * 200,
 		Parent:    parentHeader,
 		Coinbase:  types.ZeroAddress,
 		Executor:  executor,

jsonrpc/throttling_test.go

@@ -48,7 +48,7 @@ func TestThrottling(t *testing.T) {
 	}
 
 	go func() {
-		time.Sleep(time.Millisecond * 150)
+		time.Sleep(time.Millisecond * 50)
 
 		defer wg.Done()
 

network/server_test.go

@@ -697,8 +697,6 @@ func TestRunDial(t *testing.T) {
 }
 
 func TestSubscribe(t *testing.T) {
-	t.Parallel()
-
 	setupServer := func(t *testing.T, shouldCloseAfterTest bool) *Server {
 		t.Helper()
 
@@ -764,8 +762,6 @@ func TestSubscribe(t *testing.T) {
 	}
 
 	t.Run("should call callback", func(t *testing.T) {
-		t.Parallel()
-
 		server := setupServer(t, true)
 
 		ctx, cancel := context.WithCancel(context.Background())
@@ -784,8 +780,6 @@ func TestSubscribe(t *testing.T) {
 	})
 
 	t.Run("should not call callback after context is closed", func(t *testing.T) {
-		t.Parallel()
-
 		server := setupServer(t, true)
 
 		ctx, cancel := context.WithCancel(context.Background())
@@ -803,8 +797,6 @@ func TestSubscribe(t *testing.T) {
 	})
 
 	t.Run("should not call callback after server closed", func(t *testing.T) {
-		t.Parallel()
-
 		server := setupServer(t, false)
 
 		ctx, cancel := context.WithCancel(context.Background())

syncer/client_test.go

@@ -211,6 +211,7 @@ func TestStatusPubSub(t *testing.T) {
 
 	assert.NoError(t, err)
 
+	time.Sleep(3 * time.Second)
 	// close channel and wait for events
 	close(client.closeCh)
 

txpool/event_subscription_test.go

@@ -117,7 +117,7 @@ func TestEventSubscription_ProcessedEvents(t *testing.T) {
 				eventStore: &eventQueue{
 					events: make([]*proto.TxPoolEvent, 0),
 				},
-				notifyCh: make(chan struct{}),
+				notifyCh: make(chan struct{}, 10),
 			}
 			go subscription.runLoop()
 
@@ -141,10 +141,10 @@ func TestEventSubscription_ProcessedEvents(t *testing.T) {
 			}
 
 			wg.Wait()
-			eventWaitCtx, eventWaitFn := context.WithTimeout(context.Background(), 10*time.Second)
+			eventWaitCtx, eventWaitFn := context.WithTimeout(context.Background(), 30*time.Second)
 			defer eventWaitFn()
 			if _, err := tests.RetryUntilTimeout(eventWaitCtx, func() (interface{}, bool) {
-				return nil, atomic.LoadInt64(&processed) < int64(testCase.expectedProcessed)
+				return nil, atomic.LoadInt64(&processed) != int64(testCase.expectedProcessed)
 			}); err != nil {
 				t.Fatalf("Unable to wait for events to be processed, %v", err)
 			}

txpool/txpool_test.go

@@ -3451,7 +3451,7 @@ func TestBatchTx_SingleAccount(t *testing.T) {
 	for {
 		select {
 		case ev = <-subscription.subscriptionChannel:
-		case <-time.After(time.Second * 3):
+		case <-time.After(time.Second * 10):
 			t.Fatal(fmt.Sprintf("timeout. processed: %d/%d and %d/%d. Added: %d",
 				enqueuedCount, defaultMaxAccountEnqueued, promotedCount, defaultMaxAccountEnqueued,
 				atomic.LoadUint64(&counter)))