swarm/network/stream: correctly use Simulation.Run callback

https://github.com/ethersphere/go-ethereum/issues/1198#issuecomment-461783804

swarm/network/stream/delivery_test.go

@@ -589,7 +589,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, chunkCount int, skipCheck bool)
 			return fmt.Errorf("Test failed, chunks not available on all nodes")
 		}
 		if err := <-retErrC; err != nil {
-			t.Fatalf("requesting chunks: %v", err)
+			return fmt.Errorf("requesting chunks: %v", err)
 		}
 		log.Debug("Test terminated successfully")
 		return nil
@@ -664,14 +664,14 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, chunkCount int, skipCheck b
 
 		item, ok := sim.NodeItem(node, bucketKeyFileStore)
 		if !ok {
-			b.Fatal("No filestore")
+			return errors.New("No filestore")
 		}
 		remoteFileStore := item.(*storage.FileStore)
 
 		pivotNode := nodeIDs[0]
 		item, ok = sim.NodeItem(pivotNode, bucketKeyNetStore)
 		if !ok {
-			b.Fatal("No filestore")
+			return errors.New("No filestore")
 		}
 		netStore := item.(*storage.NetStore)
 
@@ -713,12 +713,12 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, chunkCount int, skipCheck b
 				ctx := context.TODO()
 				hash, wait, err := remoteFileStore.Store(ctx, testutil.RandomReader(i, chunkSize), int64(chunkSize), false)
 				if err != nil {
-					b.Fatalf("expected no error. got %v", err)
+					return fmt.Errorf("store: %v", err)
 				}
 				// wait until all chunks stored
 				err = wait(ctx)
 				if err != nil {
-					b.Fatalf("expected no error. got %v", err)
+					return fmt.Errorf("wait store: %v", err)
 				}
 				// collect the hashes
 				hashes[i] = hash
@@ -754,10 +754,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, chunkCount int, skipCheck b
 				break Loop
 			}
 		}
-		if err != nil {
+		return err
-			b.Fatal(err)
-		}
-		return nil
 	})
 	if result.Error != nil {
 		b.Fatal(result.Error)

swarm/network/stream/intervals_test.go

@@ -118,13 +118,11 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
 
 		_, wait, err := fileStore.Store(ctx, testutil.RandomReader(1, size), int64(size), false)
 		if err != nil {
-			log.Error("Store error: %v", "err", err)
+			return fmt.Errorf("store: %v", err)
-			t.Fatal(err)
 		}
 		err = wait(ctx)
 		if err != nil {
-			log.Error("Wait error: %v", "err", err)
+			return fmt.Errorf("wait store: %v", err)
-			t.Fatal(err)
 		}
 
 		item, ok = sim.NodeItem(checker, bucketKeyRegistry)

swarm/network/stream/streamer_test.go

@@ -1290,7 +1290,7 @@ func TestGetSubscriptionsRPC(t *testing.T) {
 		select {
 		case <-allSubscriptionsDone:
 		case <-ctx.Done():
-			t.Fatal("Context timed out")
+			return errors.New("Context timed out")
 		}
 
 		lock.RLock()
@@ -1302,14 +1302,14 @@ func TestGetSubscriptionsRPC(t *testing.T) {
 			//create rpc client
 			client, err := node.Client()
 			if err != nil {
-				t.Fatalf("create node 1 rpc client fail: %v", err)
+				return fmt.Errorf("create node 1 rpc client fail: %v", err)
 			}
 
 			//ask it for subscriptions
 			pstreams := make(map[string][]string)
 			err = client.Call(&pstreams, "stream_getPeerSubscriptions")
 			if err != nil {
-				t.Fatal(err)
+				return fmt.Errorf("client call stream_getPeerSubscriptions: %v", err)
 			}
 			//length of the subscriptions can not be smaller than number of peers
 			log.Debug("node subscriptions", "node", node.String())

swarm/network/stream/syncer_test.go

@@ -142,7 +142,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p
 			id := nodeIDs[j]
 			client, err := sim.Net.GetNode(id).Client()
 			if err != nil {
-				t.Fatal(err)
+				return fmt.Errorf("node %s client: %v", id, err)
 			}
 			sid := nodeIDs[j+1]
 			client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream("SYNC", FormatSyncBinKey(1), false), NewRange(0, 0), Top)
@@ -158,7 +158,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p
 				size := chunkCount * chunkSize
 				_, wait, err := fileStore.Store(ctx, testutil.RandomReader(j, size), int64(size), false)
 				if err != nil {
-					t.Fatal(err.Error())
+					return fmt.Errorf("fileStore.Store: %v", err)
 				}
 				wait(ctx)
 			}
@@ -273,7 +273,7 @@ func TestSameVersionID(t *testing.T) {
 
 		//the peers should connect, thus getting the peer should not return nil
 		if registry.getPeer(nodes[1]) == nil {
-			t.Fatal("Expected the peer to not be nil, but it is")
+			return errors.New("Expected the peer to not be nil, but it is")
 		}
 		return nil
 	})
@@ -338,7 +338,7 @@ func TestDifferentVersionID(t *testing.T) {
 
 		//getting the other peer should fail due to the different version numbers
 		if registry.getPeer(nodes[1]) != nil {
-			t.Fatal("Expected the peer to be nil, but it is not")
+			return errors.New("Expected the peer to be nil, but it is not")
 		}
 		return nil
 	})

swarm/network/stream/visualized_snapshot_sync_sim_test.go

@@ -114,7 +114,7 @@ func TestNonExistingHashesWithServer(t *testing.T) {
 		id := sim.Net.GetRandomUpNode().ID()
 		item, ok := sim.NodeItem(id, bucketKeyFileStore)
 		if !ok {
-			t.Fatalf("No filestore")
+			return errors.New("No filestore")
 		}
 		fileStore := item.(*storage.FileStore)
 		//create a bogus hash