swarm/network/stream: non-existing hash visualization sim

swarm/network/stream/snapshot_retrieval_test.go

@@ -104,21 +104,14 @@ func TestRetrieval(t *testing.T) {
 	}
 }
 
-/*
+var retrievalSimServiceMap = map[string]simulation.ServiceFunc{
+	"streamer": retrievalStreamerFunc,
+}
 
-The test loads a snapshot file to construct the swarm network,
+func retrievalStreamerFunc(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
-assuming that the snapshot file identifies a healthy
+	n := ctx.Config.Node()
-kademlia network. Nevertheless a health check runs in the
+	addr := network.NewAddr(n)
-simulation's `action` function.
+	store, datadir, err := createTestLocalStorageForID(n.ID(), addr)
-
-The snapshot should have 'streamer' in its service list.
-*/
-func runFileRetrievalTest(nodeCount int) error {
-	sim := simulation.New(map[string]simulation.ServiceFunc{
-		"streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
-			node := ctx.Config.Node()
-			addr := network.NewAddr(node)
-			store, datadir, err := createTestLocalStorageForID(node.ID(), addr)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -148,9 +141,18 @@ func runFileRetrievalTest(nodeCount int) error {
 	}
 
 	return r, cleanup, nil
+}
 
-		},
+/*
-	})
+The test loads a snapshot file to construct the swarm network,
+assuming that the snapshot file identifies a healthy
+kademlia network. Nevertheless a health check runs in the
+simulation's `action` function.
+
+The snapshot should have 'streamer' in its service list.
+*/
+func runFileRetrievalTest(nodeCount int) error {
+	sim := simulation.New(retrievalSimServiceMap)
 	defer sim.Close()
 
 	log.Info("Initializing test config")
@@ -263,44 +265,7 @@ simulation's `action` function.
 The snapshot should have 'streamer' in its service list.
 */
 func runRetrievalTest(chunkCount int, nodeCount int) error {
-	sim := simulation.New(map[string]simulation.ServiceFunc{
+	sim := simulation.New(retrievalSimServiceMap)
-		"streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
-			node := ctx.Config.Node()
-			addr := network.NewAddr(node)
-			store, datadir, err := createTestLocalStorageForID(node.ID(), addr)
-			if err != nil {
-				return nil, nil, err
-			}
-			bucket.Store(bucketKeyStore, store)
-
-			localStore := store.(*storage.LocalStore)
-			netStore, err := storage.NewNetStore(localStore, nil)
-			if err != nil {
-				return nil, nil, err
-			}
-			kad := network.NewKademlia(addr.Over(), network.NewKadParams())
-			delivery := NewDelivery(kad, netStore)
-			netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New
-
-			r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
-				DoSync:          true,
-				SyncUpdateDelay: 0,
-			})
-
-			fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams())
-			bucketKeyFileStore = simulation.BucketKey("filestore")
-			bucket.Store(bucketKeyFileStore, fileStore)
-
-			cleanup = func() {
-				os.RemoveAll(datadir)
-				netStore.Close()
-				r.Close()
-			}
-
-			return r, cleanup, nil
-
-		},
-	})
 	defer sim.Close()
 
 	conf := &synctestConfig{}

swarm/network/stream/snapshot_sync_test.go

@@ -30,7 +30,6 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/p2p"
-	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/ethereum/go-ethereum/p2p/simulations"
 	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
@@ -149,7 +148,6 @@ var simServiceMap = map[string]simulation.ServiceFunc{
 }
 
 func streamerFunc(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
-
 	n := ctx.Config.Node()
 	addr := network.NewAddr(n)
 	store, datadir, err := createTestLocalStorageForID(n.ID(), addr)
@@ -166,7 +164,7 @@ func streamerFunc(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Servic
 	delivery := NewDelivery(kad, netStore)
 	netStore.NewNetFetcherFunc = network.NewFetcherFactory(dummyRequestFromPeers, true).New
 
-	r := NewRegistry(addr, delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
+	r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
 		DoSync:          true,
 		SyncUpdateDelay: 3 * time.Second,
 	})

swarm/network/stream/visualized_snapshot_sync_sim_test.go

@@ -39,8 +39,7 @@ func init() {
 	*/
 }
 
-func TestSnapshotSyncWithServer(t *testing.T) {
+func setupSim(serviceMap map[string]simulation.ServiceFunc) (int, int, *simulation.Simulation) {
-
 	nodeCount := *nodes
 	chunkCount := *chunks
 
@@ -49,26 +48,12 @@ func TestSnapshotSyncWithServer(t *testing.T) {
 		chunkCount = 1
 	}
 
-	sim := simulation.New(simServiceMap).WithServer(":8888")
+	sim := simulation.New(serviceMap).WithServer(":8888")
-	defer sim.Close()
+	return nodeCount, chunkCount, sim
-
-	log.Info("Initializing test config")
-
-	conf := &synctestConfig{}
-	//map of discover ID to indexes of chunks expected at that ID
-	conf.idToChunksMap = make(map[discover.NodeID][]int)
-	//map of overlay address to discover ID
-	conf.addrToIDMap = make(map[string]discover.NodeID)
-	//array where the generated chunk hashes will be stored
-	conf.hashes = make([]storage.Address, 0)
-
-	err := sim.UploadSnapshot(fmt.Sprintf("testing/snapshot_%d.json", nodeCount))
-	if err != nil {
-		panic(err)
 }
 
+func watchSim(sim *simulation.Simulation) (context.Context, context.CancelFunc) {
 	ctx, cancelSimRun := context.WithTimeout(context.Background(), 2*time.Minute)
-	defer cancelSimRun()
 
 	if _, err := sim.WaitTillHealthy(ctx, 2); err != nil {
 		panic(err)
@@ -88,6 +73,75 @@ func TestSnapshotSyncWithServer(t *testing.T) {
 		}
 	}()
 
+	return ctx, cancelSimRun
+}
+
+func TestNonExistingHashesWithServer(t *testing.T) {
+	nodeCount, _, sim := setupSim(retrievalSimServiceMap)
+	defer sim.Close()
+
+	err := sim.UploadSnapshot(fmt.Sprintf("testing/snapshot_%d.json", nodeCount))
+	if err != nil {
+		panic(err)
+	}
+
+	ctx, cancelSimRun := watchSim(sim)
+	defer cancelSimRun()
+
+	testDuration := 30 * time.Second
+	result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
+		//check on the node's FileStore (netstore)
+		id := sim.RandomUpNode().ID
+		item, ok := sim.NodeItem(id, bucketKeyFileStore)
+		if !ok {
+			t.Fatalf("No filestore")
+		}
+		fileStore := item.(*storage.FileStore)
+		fakeHash := storage.GenerateRandomChunk(1000).Address()
+		fmt.Println(fakeHash)
+		reader, _ := fileStore.Retrieve(context.TODO(), fakeHash)
+		if _, err := reader.Size(ctx, nil); err != nil {
+			fmt.Println("expected error for non-existing chunk")
+		}
+
+		time.Sleep(testDuration)
+
+		return nil
+	})
+	if result.Error != nil {
+		t.Fatal(result.Error)
+	}
+
+	evt := &simulations.Event{
+		Type: EventTypeSimTerminated,
+	}
+	sim.Net.Events().Send(evt)
+
+}
+
+func TestSnapshotSyncWithServer(t *testing.T) {
+
+	nodeCount, chunkCount, sim := setupSim(simServiceMap)
+	defer sim.Close()
+
+	log.Info("Initializing test config")
+
+	conf := &synctestConfig{}
+	//map of discover ID to indexes of chunks expected at that ID
+	conf.idToChunksMap = make(map[discover.NodeID][]int)
+	//map of overlay address to discover ID
+	conf.addrToIDMap = make(map[string]discover.NodeID)
+	//array where the generated chunk hashes will be stored
+	conf.hashes = make([]storage.Address, 0)
+
+	err := sim.UploadSnapshot(fmt.Sprintf("testing/snapshot_%d.json", nodeCount))
+	if err != nil {
+		panic(err)
+	}
+
+	ctx, cancelSimRun := watchSim(sim)
+	defer cancelSimRun()
+
 	//sim.PeerEvents(
 	offeredHashesFilter := simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeMsgRecv).Protocol("stream").MsgCode(1)
 	wantedFilter := simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeMsgRecv).Protocol("stream").MsgCode(2)