swarm/storage, swarm/network: fix tests - all tests pass

- chunker tester split/append should wait before closing quit chan
 - synciterator now increments first
 - kademlia hive pretty print test broke on 2017 vs 2018 year in date ;)
 - append tester needs to wait for dbstore chan in other conditional branch
 - mput should enforce dbstored chan closed when chunk created for memstore
 - ...

swarm/network/kademlia_test.go

@@ -399,9 +399,10 @@ func TestPruning(t *testing.T) {
 
 func TestKademliaHiveString(t *testing.T) {
 	k := newTestKademlia("00000000").On("01000000", "00100000").Register("10000000", "10000001")
+	k.MaxProxDisplay = 8
 	h := k.String()
 	expH := "\n=========================================================================\nMon Feb 27 12:10:28 UTC 2017 KΛÐΞMLIΛ hive: queen's address: 000000\npopulation: 2 (4), MinProxBinSize: 2, MinBinSize: 1, MaxBinSize: 4\n000  0                              |  2 8100 (0) 8000 (0)\n============ DEPTH: 1 ==========================================\n001  1 4000                         |  1 4000 (0)\n002  1 2000                         |  1 2000 (0)\n003  0                              |  0\n004  0                              |  0\n005  0                              |  0\n006  0                              |  0\n007  0                              |  0\n========================================================================="
-	if expH[100:] != h[100:] {
+	if expH[104:] != h[104:] {
 		t.Fatalf("incorrect hive output. expected %v, got %v", expH, h)
 	}
 }

swarm/network/syncer.go

@@ -247,7 +247,7 @@ func (self *IncomingSwarmSyncer) TakeoverProof(s Stream, from uint64, hashes []b
 	if self.chunker != nil {
 		if from > self.sessionAt { // for live syncing currentRoot is always updated
 			//expRoot, err := self.chunker.Append(self.currentRoot, bytes.NewReader(hashes), self.retrieveC, self.storeC)
-			expRoot, err := self.chunker.Append(self.currentRoot, bytes.NewReader(hashes), self.retrieveC)
+			expRoot, _, err := self.chunker.Append(self.currentRoot, bytes.NewReader(hashes), self.retrieveC)
 			if err != nil {
 				return nil, err
 			}

swarm/storage/chunker_test.go

@@ -72,13 +72,16 @@ func (self *chunkerTester) Split(chunker Splitter, data io.Reader, size int64, c
 		}()
 	}
 
-	key, wait, err = chunker.Split(data, size, chunkC)
+	var w func()
+	key, w, err = chunker.Split(data, size, chunkC)
 	if err != nil && expectedError == nil {
 		err = fmt.Errorf("Split error: %v", err)
 	}
-
 	if chunkC != nil {
+		wait = func() {
+			w()
 			close(quitC)
+		}
 	} else {
 		wait = func() {}
 	}
@@ -102,6 +105,7 @@ func (self *chunkerTester) Append(chunker Splitter, rootKey Key, data io.Reader,
 						if !success {
 							// Requesting data
 							self.chunks[chunk.Key.String()] = chunk
+							close(chunk.dbStored)
 						} else {
 							// getting data
 							chunk.SData = stored.SData
@@ -114,14 +118,17 @@ func (self *chunkerTester) Append(chunker Splitter, rootKey Key, data io.Reader,
 			}
 		}()
 	}
-
+	var w func()
-	key, wait, err = chunker.Append(rootKey, data, chunkC)
+	key, w, err = chunker.Append(rootKey, data, chunkC)
 	if err != nil && expectedError == nil {
 		err = fmt.Errorf("Append error: %v", err)
 	}
 
 	if chunkC != nil {
+		wait = func() {
+			w()
 			close(quitC)
+		}
 	} else {
 		wait = func() {}
 	}
@@ -198,12 +205,12 @@ func testRandomData(splitter Splitter, n int, tester *chunkerTester) Key {
 
 	chunkC := make(chan *Chunk, 1000)
 
-	key, _, err := tester.Split(splitter, data, int64(n), chunkC, nil)
+	key, wait, err := tester.Split(splitter, data, int64(n), chunkC, nil)
 	if err != nil {
 		tester.t.Fatalf(err.Error())
 	}
 	tester.t.Logf(" Key = %v\n", key)
-
+	wait()
 	chunkC = make(chan *Chunk, 1000)
 	quitC := make(chan bool)
 
@@ -315,8 +322,6 @@ func TestSha3ForCorrectness(t *testing.T) {
 }
 
 func TestDataAppend(t *testing.T) {
-	t.Skip("Skip until append chunks are fixed")
-
 	sizes := []int{1, 1, 1, 4095, 4096, 4097, 1, 1, 1, 123456, 2345678, 2345678}
 	appendSizes := []int{4095, 4096, 4097, 1, 1, 1, 8191, 8192, 8193, 9000, 3000, 5000}
 

swarm/storage/common_test.go

@@ -87,13 +87,22 @@ func mput(store ChunkStore, processors int, n int, f func(i int) *Chunk) (hs []K
 					defer wg.Done()
 
 					store.Put(chunk)
+
 					<-chunk.dbStored
 				}()
 			}
 		}()
 	}
-	for i := 0; i < n; i++ {
+	fa := f
+	if _, ok := store.(*MemStore); ok {
+		fa = func(i int) *Chunk {
 			chunk := f(i)
+			close(chunk.dbStored)
+			return chunk
+		}
+	}
+	for i := 0; i < n; i++ {
+		chunk := fa(i)
 		hs = append(hs, chunk.Key)
 		c <- chunk
 	}

swarm/storage/dbstore_test.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"os"
+	"sync"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/log"
@@ -126,9 +127,16 @@ func TestIterator(t *testing.T) {
 
 	FakeChunk(getDefaultChunkSize(), chunkcount, chunks)
 
+	wg := &sync.WaitGroup{}
+	wg.Add(len(chunks))
 	for i = 0; i < len(chunks); i++ {
 		db.Put(chunks[i])
 		chunkkeys[i] = chunks[i].Key
+		j := i
+		go func() {
+			defer wg.Done()
+			<-chunks[j].dbStored
+		}()
 	}
 
 	//testSplit(m, l, 128, chunkkeys, t)
@@ -136,12 +144,12 @@ func TestIterator(t *testing.T) {
 	for i = 0; i < len(chunkkeys); i++ {
 		log.Trace(fmt.Sprintf("Chunk array pos %d/%d: '%v'", i, chunkcount, chunkkeys[i]))
 	}
-
+	wg.Wait()
 	i = 0
 	for poc = 0; poc <= 255; poc++ {
 		err := db.SyncIterator(0, uint64(chunkkeys.Len()), uint8(poc), func(k Key, n uint64) bool {
 			log.Trace(fmt.Sprintf("Got key %v number %d poc %d", k, n, uint8(poc)))
-			chunkkeys_results[n] = k
+			chunkkeys_results[n-1] = k
 			i++
 			return true
 		})