aboutsummaryrefslogtreecommitdiffstats
path: root/swarm/network
diff options
context:
space:
mode:
Diffstat (limited to 'swarm/network')
-rw-r--r--swarm/network/simulation/events_test.go7
-rw-r--r--swarm/network/simulation/http_test.go21
-rw-r--r--swarm/network/stream/delivery_test.go34
-rw-r--r--swarm/network/stream/intervals_test.go14
-rw-r--r--swarm/network/stream/snapshot_sync_test.go23
-rw-r--r--swarm/network/stream/syncer_test.go14
6 files changed, 84 insertions, 29 deletions
diff --git a/swarm/network/simulation/events_test.go b/swarm/network/simulation/events_test.go
index 34ef24ed4..529844816 100644
--- a/swarm/network/simulation/events_test.go
+++ b/swarm/network/simulation/events_test.go
@@ -81,6 +81,7 @@ func TestPeerEventsTimeout(t *testing.T) {
events := sim.PeerEvents(ctx, sim.NodeIDs())
done := make(chan struct{})
+ errC := make(chan error)
go func() {
for e := range events {
if e.Error == context.Canceled {
@@ -90,14 +91,16 @@ func TestPeerEventsTimeout(t *testing.T) {
close(done)
return
} else {
- t.Fatal(e.Error)
+ errC <- e.Error
}
}
}()
select {
case <-time.After(time.Second):
- t.Error("no context deadline received")
+ t.Fatal("no context deadline received")
+ case err := <-errC:
+ t.Fatal(err)
case <-done:
// all good, context deadline detected
}
diff --git a/swarm/network/simulation/http_test.go b/swarm/network/simulation/http_test.go
index 775cf9219..dffd03a03 100644
--- a/swarm/network/simulation/http_test.go
+++ b/swarm/network/simulation/http_test.go
@@ -73,7 +73,8 @@ func TestSimulationWithHTTPServer(t *testing.T) {
//this time the timeout should be long enough so that it doesn't kick in too early
ctx, cancel2 := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel2()
- go sendRunSignal(t)
+ errC := make(chan error, 1)
+ go triggerSimulationRun(t, errC)
result = sim.Run(ctx, func(ctx context.Context, sim *Simulation) error {
log.Debug("This run waits for the run signal from `frontend`...")
//ensure with a Sleep that simulation doesn't terminate before the signal is received
@@ -83,10 +84,13 @@ func TestSimulationWithHTTPServer(t *testing.T) {
if result.Error != nil {
t.Fatal(result.Error)
}
+ if err := <-errC; err != nil {
+ t.Fatal(err)
+ }
log.Debug("Test terminated successfully")
}
-func sendRunSignal(t *testing.T) {
+func triggerSimulationRun(t *testing.T, errC chan error) {
//We need to first wait for the sim HTTP server to start running...
time.Sleep(2 * time.Second)
//then we can send the signal
@@ -94,16 +98,13 @@ func sendRunSignal(t *testing.T) {
log.Debug("Sending run signal to simulation: POST /runsim...")
resp, err := http.Post(fmt.Sprintf("http://localhost%s/runsim", DefaultHTTPSimAddr), "application/json", nil)
if err != nil {
- t.Fatalf("Request failed: %v", err)
+ errC <- fmt.Errorf("Request failed: %v", err)
+ return
}
- defer func() {
- err := resp.Body.Close()
- if err != nil {
- log.Error("Error closing response body", "err", err)
- }
- }()
log.Debug("Signal sent")
if resp.StatusCode != http.StatusOK {
- t.Fatalf("err %s", resp.Status)
+ errC <- fmt.Errorf("err %s", resp.Status)
+ return
}
+ errC <- resp.Body.Close()
}
diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go
index 04a4bb077..c65e1386d 100644
--- a/swarm/network/stream/delivery_test.go
+++ b/swarm/network/stream/delivery_test.go
@@ -19,9 +19,11 @@ package stream
import (
"bytes"
"context"
+ "errors"
"fmt"
"os"
"sync"
+ "sync/atomic"
"testing"
"time"
@@ -500,7 +502,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, chunkCount int, skipCheck bool)
log.Info("Starting simulation")
ctx := context.Background()
- result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
+ result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) (err error) {
nodeIDs := sim.UpNodeIDs()
//determine the pivot node to be the first node of the simulation
pivot := nodeIDs[0]
@@ -553,14 +555,13 @@ func testDeliveryFromNodes(t *testing.T, nodes, chunkCount int, skipCheck bool)
}
pivotFileStore := item.(*storage.FileStore)
log.Debug("Starting retrieval routine")
+ retErrC := make(chan error)
go func() {
// start the retrieval on the pivot node - this will spawn retrieve requests for missing chunks
// we must wait for the peer connections to have started before requesting
n, err := readAll(pivotFileStore, fileHash)
log.Info(fmt.Sprintf("retrieved %v", fileHash), "read", n, "err", err)
- if err != nil {
- t.Fatalf("requesting chunks action error: %v", err)
- }
+ retErrC <- err
}()
log.Debug("Watching for disconnections")
@@ -570,11 +571,19 @@ func testDeliveryFromNodes(t *testing.T, nodes, chunkCount int, skipCheck bool)
simulation.NewPeerEventsFilter().Drop(),
)
+ var disconnected atomic.Value
go func() {
for d := range disconnections {
if d.Error != nil {
log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
- t.Fatal(d.Error)
+ disconnected.Store(true)
+ }
+ }
+ }()
+ defer func() {
+ if err != nil {
+ if yes, ok := disconnected.Load().(bool); ok && yes {
+ err = errors.New("disconnect events received")
}
}
}()
@@ -595,6 +604,9 @@ func testDeliveryFromNodes(t *testing.T, nodes, chunkCount int, skipCheck bool)
if !success {
return fmt.Errorf("Test failed, chunks not available on all nodes")
}
+ if err := <-retErrC; err != nil {
+ t.Fatalf("requesting chunks: %v", err)
+ }
log.Debug("Test terminated successfully")
return nil
})
@@ -675,7 +687,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, chunkCount int, skipCheck b
}
ctx := context.Background()
- result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
+ result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) (err error) {
nodeIDs := sim.UpNodeIDs()
node := nodeIDs[len(nodeIDs)-1]
@@ -702,11 +714,19 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, chunkCount int, skipCheck b
simulation.NewPeerEventsFilter().Drop(),
)
+ var disconnected atomic.Value
go func() {
for d := range disconnections {
if d.Error != nil {
log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
- b.Fatal(d.Error)
+ disconnected.Store(true)
+ }
+ }
+ }()
+ defer func() {
+ if err != nil {
+ if yes, ok := disconnected.Load().(bool); ok && yes {
+ err = errors.New("disconnect events received")
}
}
}()
diff --git a/swarm/network/stream/intervals_test.go b/swarm/network/stream/intervals_test.go
index 7c7feeb11..121758c18 100644
--- a/swarm/network/stream/intervals_test.go
+++ b/swarm/network/stream/intervals_test.go
@@ -19,9 +19,11 @@ package stream
import (
"context"
"encoding/binary"
+ "errors"
"fmt"
"os"
"sync"
+ "sync/atomic"
"testing"
"time"
@@ -117,7 +119,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
t.Fatal(err)
}
- result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
+ result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) (err error) {
nodeIDs := sim.UpNodeIDs()
storer := nodeIDs[0]
checker := nodeIDs[1]
@@ -162,11 +164,19 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
return err
}
+ var disconnected atomic.Value
go func() {
for d := range disconnections {
if d.Error != nil {
log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
- t.Fatal(d.Error)
+ disconnected.Store(true)
+ }
+ }
+ }()
+ defer func() {
+ if err != nil {
+ if yes, ok := disconnected.Load().(bool); ok && yes {
+ err = errors.New("disconnect events received")
}
}
}()
diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go
index f86d9acca..7a883644b 100644
--- a/swarm/network/stream/snapshot_sync_test.go
+++ b/swarm/network/stream/snapshot_sync_test.go
@@ -21,6 +21,7 @@ import (
"os"
"runtime"
"sync"
+ "sync/atomic"
"testing"
"time"
@@ -213,11 +214,13 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) {
simulation.NewPeerEventsFilter().Drop(),
)
+ var disconnected atomic.Value
go func() {
for d := range disconnections {
- log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
- t.Fatal("unexpected disconnect")
- cancelSimRun()
+ if d.Error != nil {
+ log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
+ disconnected.Store(true)
+ }
}
}()
@@ -226,6 +229,9 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) {
if result.Error != nil {
t.Fatal(result.Error)
}
+ if yes, ok := disconnected.Load().(bool); ok && yes {
+ t.Fatal("disconnect events received")
+ }
log.Info("Simulation ended")
}
@@ -395,11 +401,13 @@ func testSyncingViaDirectSubscribe(t *testing.T, chunkCount int, nodeCount int)
simulation.NewPeerEventsFilter().Drop(),
)
+ var disconnected atomic.Value
go func() {
for d := range disconnections {
- log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
- t.Fatal("unexpected disconnect")
- cancelSimRun()
+ if d.Error != nil {
+ log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
+ disconnected.Store(true)
+ }
}
}()
@@ -514,6 +522,9 @@ func testSyncingViaDirectSubscribe(t *testing.T, chunkCount int, nodeCount int)
return result.Error
}
+ if yes, ok := disconnected.Load().(bool); ok && yes {
+ t.Fatal("disconnect events received")
+ }
log.Info("Simulation ended")
return nil
}
diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go
index 2fdba71ff..09152ebfe 100644
--- a/swarm/network/stream/syncer_test.go
+++ b/swarm/network/stream/syncer_test.go
@@ -18,11 +18,13 @@ package stream
import (
"context"
+ "errors"
"fmt"
"io/ioutil"
"math"
"os"
"sync"
+ "sync/atomic"
"testing"
"time"
@@ -129,7 +131,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p
if err != nil {
t.Fatal(err)
}
- result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
+ result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) (err error) {
nodeIDs := sim.UpNodeIDs()
nodeIndex := make(map[enode.ID]int)
@@ -143,11 +145,19 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p
simulation.NewPeerEventsFilter().Drop(),
)
+ var disconnected atomic.Value
go func() {
for d := range disconnections {
if d.Error != nil {
log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
- t.Fatal(d.Error)
+ disconnected.Store(true)
+ }
+ }
+ }()
+ defer func() {
+ if err != nil {
+ if yes, ok := disconnected.Load().(bool); ok && yes {
+ err = errors.New("disconnect events received")
}
}
}()
generated by cgit v1.2.3 (git 2.25.1) at 2024-08-23 03:55:49 +0800