diff --git a/swarm/network/stream/snapshot_retrieval_test.go b/swarm/network/stream/snapshot_retrieval_test.go index 6498f599dc..3344b80c7f 100644 --- a/swarm/network/stream/snapshot_retrieval_test.go +++ b/swarm/network/stream/snapshot_retrieval_test.go @@ -104,8 +104,46 @@ func TestRetrieval(t *testing.T) { } } -/* +var retrievalSimServiceMap = map[string]simulation.ServiceFunc{ + "streamer": retrievalStreamerFunc, +} +func retrievalStreamerFunc(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) + 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: 3 * time.Second, + }) + + fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams()) + bucket.Store(bucketKeyFileStore, fileStore) + + cleanup = func() { + os.RemoveAll(datadir) + netStore.Close() + r.Close() + } + + 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 @@ -114,43 +152,7 @@ simulation's `action` function. 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 - } - 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: 3 * time.Second, - }) - - fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams()) - bucket.Store(bucketKeyFileStore, fileStore) - - cleanup = func() { - os.RemoveAll(datadir) - netStore.Close() - r.Close() - } - - return r, cleanup, nil - - }, - }) + 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{ - "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 - - }, - }) + sim := simulation.New(retrievalSimServiceMap) defer sim.Close() conf := &synctestConfig{} diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go index 3d59f6b8a2..f6886824b0 100644 --- a/swarm/network/stream/snapshot_sync_test.go +++ b/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, }) diff --git a/swarm/network/stream/visualized_snapshot_sync_sim_test.go b/swarm/network/stream/visualized_snapshot_sync_sim_test.go index 5fa80ed6c1..886afb03f8 100644 --- a/swarm/network/stream/visualized_snapshot_sync_sim_test.go +++ b/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") - 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) - } + sim := simulation.New(serviceMap).WithServer(":8888") + return nodeCount, chunkCount, sim +} +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)