From 38bdf7f6b8d83743131dcc89c29c3c596476a052 Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Fri, 21 Dec 2018 11:28:30 +0100 Subject: [PATCH] swarm/storage/localstore: add more subscriptions tests --- .../storage/localstore/subscriptions_test.go | 301 ++++++++++++++++++ 1 file changed, 301 insertions(+) diff --git a/swarm/storage/localstore/subscriptions_test.go b/swarm/storage/localstore/subscriptions_test.go index 066715033f..bf67a11eab 100644 --- a/swarm/storage/localstore/subscriptions_test.go +++ b/swarm/storage/localstore/subscriptions_test.go @@ -113,3 +113,304 @@ func TestSubscribePush(t *testing.T) { } } } + +// TestSubscribePush_multiple uploads chunks before and after +// multiple push syncing subscriptions are created and +// validates if all chunks are received in the right order. +func TestSubscribePush_multiple(t *testing.T) { + t.Parallel() + + db, cleanupFunc := newTestDB(t, nil) + defer cleanupFunc() + + uploader := db.NewPutter(ModePutUpload) + + chunks := make([]storage.Chunk, 0) + + uploadRandomChunks := func(count int) { + for i := 0; i < count; i++ { + chunk := generateRandomChunk() + + err := uploader.Put(chunk) + if err != nil { + t.Fatal(err) + } + + chunks = append(chunks, chunk) + } + } + + // prepopulate database with some chunks + // before the subscription + uploadRandomChunks(10) + + // set a timeout on subscription + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + // collect all errors from validating chunks, even nil ones + // to validate the number of chunks received by the subscription + errChan := make(chan error) + + subsCount := 10 + + // start a number of subscriptions + // that all of them will write every chunk error to errChan + for j := 0; j < subsCount; j++ { + sub, err := db.SubscribePush(ctx) + if err != nil { + t.Fatal(err) + } + defer sub.Stop() + + // receive and validate chunks from the subscription + go func(j int) { + var i int // chunk index + for { + select { + case got := <-sub.Chunks: + want := chunks[i] + var err error + if !bytes.Equal(got.Data(), want.Data()) { + err = fmt.Errorf("got chunk %v on subscription %v data %x, want %x", i, j, got.Data(), want.Data()) + } + if !bytes.Equal(got.Address(), want.Address()) { + err = fmt.Errorf("got chunk %v on subscription %v address %s, want %s", i, j, got.Address().Hex(), want.Address().Hex()) + } + i++ + // send one and only one error per received chunk + errChan <- err + case <-ctx.Done(): + return + } + } + }(j) + } + + // upload some chunks just after subscribe + uploadRandomChunks(5) + + time.Sleep(500 * time.Millisecond) + + // upload some chunks after some short time + uploadRandomChunks(3) + + // number of chunks received by all subscriptions + totalChunks := len(chunks) * subsCount + for i := 0; i < totalChunks; i++ { + select { + case err := <-errChan: + if err != nil { + t.Error(err) + } + case <-ctx.Done(): + t.Error(ctx.Err()) + } + } +} + +// TestSubscribePull uploads some chunks before and after +// pull syncing subscription is created and validates if +// all chunks are received in the right order +// for expected proximity order bins. +func TestSubscribePull(t *testing.T) { + t.Parallel() + + db, cleanupFunc := newTestDB(t, nil) + defer cleanupFunc() + + uploader := db.NewPutter(ModePutUpload) + + chunks := make(map[uint8][]storage.Chunk) + var uploadedChunksCount int + + uploadRandomChunks := func(count int) { + for i := 0; i < count; i++ { + chunk := generateRandomChunk() + + err := uploader.Put(chunk) + if err != nil { + t.Fatal(err) + } + + bin := db.po(chunk.Address()) + if _, ok := chunks[bin]; !ok { + chunks[bin] = make([]storage.Chunk, 0) + } + + chunks[bin] = append(chunks[bin], chunk) + uploadedChunksCount++ + } + } + + // prepopulate database with some chunks + // before the subscription + uploadRandomChunks(10) + + // set a timeout on subscription + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + // collect all errors from validating chunks, even nil ones + // to validate the number of chunks received by the subscription + errChan := make(chan error) + + var maxBin uint8 = 32 + + for bin := uint8(0); bin < maxBin; bin++ { + sub, err := db.SubscribePull(ctx, bin) + if err != nil { + t.Fatal(err) + } + defer sub.Stop() + + // receive and validate chunks from the subscription + go func(bin uint8) { + var i int // chunk index + for { + select { + case got := <-sub.Chunks: + want := chunks[bin][i] + var err error + if !bytes.Equal(got.Data(), want.Data()) { + err = fmt.Errorf("got chunk %v in bin %v data %x, want %x", i, bin, got.Data(), want.Data()) + } + if !bytes.Equal(got.Address(), want.Address()) { + err = fmt.Errorf("got chunk %v in bin %v address %s, want %s", i, bin, got.Address().Hex(), want.Address().Hex()) + } + i++ + // send one and only one error per received chunk + errChan <- err + case <-ctx.Done(): + return + } + } + }(bin) + } + + // upload some chunks just after subscribe + uploadRandomChunks(5) + + time.Sleep(500 * time.Millisecond) + + // upload some chunks after some short time + uploadRandomChunks(3) + + for i := 0; i < uploadedChunksCount; i++ { + select { + case err := <-errChan: + if err != nil { + t.Error(err) + } + case <-ctx.Done(): + t.Error(ctx.Err()) + } + } +} + +// TestSubscribePull_multiple uploads chunks before and after +// multiple pull syncing subscriptions are created and +// validates if all chunks are received in the right order +// for expected proximity order bins. +func TestSubscribePull_multiple(t *testing.T) { + t.Parallel() + + db, cleanupFunc := newTestDB(t, nil) + defer cleanupFunc() + + uploader := db.NewPutter(ModePutUpload) + + chunks := make(map[uint8][]storage.Chunk) + var uploadedChunksCount int + + uploadRandomChunks := func(count int) { + for i := 0; i < count; i++ { + chunk := generateRandomChunk() + + err := uploader.Put(chunk) + if err != nil { + t.Fatal(err) + } + + bin := db.po(chunk.Address()) + if _, ok := chunks[bin]; !ok { + chunks[bin] = make([]storage.Chunk, 0) + } + + chunks[bin] = append(chunks[bin], chunk) + uploadedChunksCount++ + } + } + + // prepopulate database with some chunks + // before the subscription + uploadRandomChunks(10) + + // set a timeout on subscription + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + // collect all errors from validating chunks, even nil ones + // to validate the number of chunks received by the subscription + errChan := make(chan error) + + var maxBin uint8 = 32 + + subsCount := 10 + + // start a number of subscriptions + // that all of them will write every chunk error to errChan + for j := 0; j < subsCount; j++ { + for bin := uint8(0); bin < maxBin; bin++ { + sub, err := db.SubscribePull(ctx, bin) + if err != nil { + t.Fatal(err) + } + defer sub.Stop() + + // receive and validate chunks from the subscription + go func(bin uint8, j int) { + var i int // chunk index + for { + select { + case got := <-sub.Chunks: + want := chunks[bin][i] + var err error + if !bytes.Equal(got.Data(), want.Data()) { + err = fmt.Errorf("got chunk %v in bin %v on subscription %v data %x, want %x", i, bin, j, got.Data(), want.Data()) + } + if !bytes.Equal(got.Address(), want.Address()) { + err = fmt.Errorf("got chunk %v in bin %v on subscription %v address %s, want %s", i, bin, j, got.Address().Hex(), want.Address().Hex()) + } + i++ + // send one and only one error per received chunk + errChan <- err + case <-ctx.Done(): + return + } + } + }(bin, j) + } + } + + // upload some chunks just after subscribe + uploadRandomChunks(5) + + time.Sleep(500 * time.Millisecond) + + // upload some chunks after some short time + uploadRandomChunks(3) + + totalChunks := uploadedChunksCount * subsCount + + for i := 0; i < totalChunks; i++ { + select { + case err := <-errChan: + if err != nil { + t.Error(err) + } + case <-ctx.Done(): + t.Error(ctx.Err()) + } + } +}