Unify the store tests (#1952)

Add more tests for store
2020-08-19 23:41:03 +01:00
parent eee5b98d78
commit d5da9c0728
5 changed files with 571 additions and 97 deletions
--- a/store/cockroach/cockroach.go
+++ b/store/cockroach/cockroach.go
@@ -27,7 +27,7 @@ var (
 	re = regexp.MustCompile("[^a-zA-Z0-9]+")

 	statements = map[string]string{
-		"list":       "SELECT key, value, metadata, expiry FROM %s.%s;",
+		"list":       "SELECT key, value, metadata, expiry FROM %s.%s WHERE key LIKE $1 ORDER BY key DESC LIMIT $2 OFFSET $3;",
 		"read":       "SELECT key, value, metadata, expiry FROM %s.%s WHERE key = $1;",
 		"readMany":   "SELECT key, value, metadata, expiry FROM %s.%s WHERE key LIKE $1;",
 		"readOffset": "SELECT key, value, metadata, expiry FROM %s.%s WHERE key LIKE $1 ORDER BY key DESC LIMIT $2 OFFSET $3;",
@@ -213,6 +213,23 @@ func (s *sqlStore) List(opts ...store.ListOption) ([]string, error) {
 	if err := s.createDB(options.Database, options.Table); err != nil {
 		return nil, err
 	}
+	limit := sql.NullInt32{}
+	offset := 0
+	pattern := "%"
+	if options.Prefix != "" || options.Suffix != "" {
+		if options.Prefix != "" {
+			pattern = options.Prefix + pattern
+		}
+		if options.Suffix != "" {
+			pattern = pattern + options.Suffix
+		}
+	}
+	if options.Offset > 0 {
+		offset = int(options.Offset)
+	}
+	if options.Limit > 0 {
+		limit = sql.NullInt32{Int32: int32(options.Limit), Valid: true}
+	}

 	st, err := s.prepare(options.Database, options.Table, "list")
 	if err != nil {
@@ -220,7 +237,7 @@ func (s *sqlStore) List(opts ...store.ListOption) ([]string, error) {
 	}
 	defer st.Close()

-	rows, err := st.Query()
+	rows, err := st.Query(pattern, limit, offset)
 	if err != nil {
 		if err == sql.ErrNoRows {
 			return nil, nil
@@ -228,8 +245,55 @@ func (s *sqlStore) List(opts ...store.ListOption) ([]string, error) {
 		return nil, err
 	}
 	defer rows.Close()
-
 	var keys []string
+	records, err := s.rowsToRecords(rows)
+	if err != nil {
+		return nil, err
+	}
+	for _, k := range records {
+		keys = append(keys, k.Key)
+	}
+	rowErr := rows.Close()
+	if rowErr != nil {
+		// transaction rollback or something
+		return keys, rowErr
+	}
+	if err := rows.Err(); err != nil {
+		return keys, err
+	}
+	return keys, nil
+}
+
+// rowToRecord converts from sql.Row to a store.Record. If the record has expired it will issue a delete in a separate goroutine
+func (s *sqlStore) rowToRecord(row *sql.Row) (*store.Record, error) {
+	var timehelper pq.NullTime
+	record := &store.Record{}
+	metadata := make(Metadata)
+
+	if err := row.Scan(&record.Key, &record.Value, &metadata, &timehelper); err != nil {
+		if err == sql.ErrNoRows {
+			return record, store.ErrNotFound
+		}
+		return nil, err
+	}
+
+	// set the metadata
+	record.Metadata = toMetadata(&metadata)
+	if timehelper.Valid {
+		if timehelper.Time.Before(time.Now()) {
+			// record has expired
+			go s.Delete(record.Key)
+			return nil, store.ErrNotFound
+		}
+		record.Expiry = time.Until(timehelper.Time)
+
+	}
+	return record, nil
+}
+
+// rowsToRecords converts from sql.Rows to  []*store.Record. If a record has expired it will issue a delete in a separate goroutine
+func (s *sqlStore) rowsToRecords(rows *sql.Rows) ([]*store.Record, error) {
+	var records []*store.Record
 	var timehelper pq.NullTime

 	for rows.Next() {
@@ -237,7 +301,7 @@ func (s *sqlStore) List(opts ...store.ListOption) ([]string, error) {
 		metadata := make(Metadata)

 		if err := rows.Scan(&record.Key, &record.Value, &metadata, &timehelper); err != nil {
-			return keys, err
+			return records, err
 		}

 		// set the metadata
@@ -249,22 +313,13 @@ func (s *sqlStore) List(opts ...store.ListOption) ([]string, error) {
 				go s.Delete(record.Key)
 			} else {
 				record.Expiry = time.Until(timehelper.Time)
-				keys = append(keys, record.Key)
+				records = append(records, record)
 			}
 		} else {
-			keys = append(keys, record.Key)
+			records = append(records, record)
 		}
-
 	}
-	rowErr := rows.Close()
-	if rowErr != nil {
-		// transaction rollback or something
-		return keys, rowErr
-	}
-	if err := rows.Err(); err != nil {
-		return keys, err
-	}
-	return keys, nil
+	return records, nil
 }

 // Read a single key
@@ -283,9 +338,6 @@ func (s *sqlStore) Read(key string, opts ...store.ReadOption) ([]*store.Record,
 		return s.read(key, options)
 	}

-	var records []*store.Record
-	var timehelper pq.NullTime
-
 	st, err := s.prepare(options.Database, options.Table, "read")
 	if err != nil {
 		return nil, err
@@ -293,32 +345,12 @@ func (s *sqlStore) Read(key string, opts ...store.ReadOption) ([]*store.Record,
 	defer st.Close()

 	row := st.QueryRow(key)
-	record := &store.Record{}
-	metadata := make(Metadata)
-
-	if err := row.Scan(&record.Key, &record.Value, &metadata, &timehelper); err != nil {
-		if err == sql.ErrNoRows {
-			return records, store.ErrNotFound
-		}
-		return records, err
+	record, err := s.rowToRecord(row)
+	if err != nil {
+		return nil, err
 	}
-
-	// set the metadata
-	record.Metadata = toMetadata(&metadata)
-
-	if timehelper.Valid {
-		if timehelper.Time.Before(time.Now()) {
-			// record has expired
-			go s.Delete(key)
-			return records, store.ErrNotFound
-		}
-		record.Expiry = time.Until(timehelper.Time)
-		records = append(records, record)
-	} else {
-		records = append(records, record)
-	}
-
-	return records, nil
+	var records []*store.Record
+	return append(records, record), nil
 }

 // Read Many records
@@ -361,31 +393,9 @@ func (s *sqlStore) read(key string, options store.ReadOptions) ([]*store.Record,

 	defer rows.Close()

-	var records []*store.Record
-	var timehelper pq.NullTime
-
-	for rows.Next() {
-		record := &store.Record{}
-		metadata := make(Metadata)
-
-		if err := rows.Scan(&record.Key, &record.Value, &metadata, &timehelper); err != nil {
-			return records, err
-		}
-
-		// set the metadata
-		record.Metadata = toMetadata(&metadata)
-
-		if timehelper.Valid {
-			if timehelper.Time.Before(time.Now()) {
-				// record has expired
-				go s.Delete(record.Key)
-			} else {
-				record.Expiry = time.Until(timehelper.Time)
-				records = append(records, record)
-			}
-		} else {
-			records = append(records, record)
-		}
+	records, err := s.rowsToRecords(rows)
+	if err != nil {
+		return nil, err
 	}
 	rowErr := rows.Close()
 	if rowErr != nil {
--- a/store/file/file.go
+++ b/store/file/file.go
@@ -153,13 +153,18 @@ func (m *fileStore) list(db *bolt.DB, limit, offset uint) []string {

 	if limit != 0 || offset != 0 {
 		sort.Slice(allKeys, func(i, j int) bool { return allKeys[i] < allKeys[j] })
-		min := func(i, j uint) uint {
-			if i < j {
-				return i
+		end := len(allKeys)
+		if limit > 0 {
+			calcLimit := int(offset + limit)
+			if calcLimit < end {
+				end = calcLimit
 			}
-			return j
 		}
-		return allKeys[offset:min(limit, uint(len(allKeys)))]
+
+		if int(offset) >= end {
+			return nil
+		}
+		return allKeys[offset:end]
 	}

 	return allKeys
--- a/store/memory/memory.go
+++ b/store/memory/memory.go
@@ -136,13 +136,19 @@ func (m *memoryStore) list(prefix string, limit, offset uint) []string {

 	if limit != 0 || offset != 0 {
 		sort.Slice(allKeys, func(i, j int) bool { return allKeys[i] < allKeys[j] })
-		min := func(i, j uint) uint {
-			if i < j {
-				return i
+		sort.Slice(allKeys, func(i, j int) bool { return allKeys[i] < allKeys[j] })
+		end := len(allKeys)
+		if limit > 0 {
+			calcLimit := int(offset + limit)
+			if calcLimit < end {
+				end = calcLimit
 			}
-			return j
 		}
-		return allKeys[offset:min(limit, uint(len(allKeys)))]
+
+		if int(offset) >= end {
+			return nil
+		}
+		return allKeys[offset:end]
 	}

 	return allKeys
--- a/store/test/store_test.go
+++ b/store/test/store_test.go
@@ -0,0 +1,471 @@
+// Package test provides a way to run tests against all the various implementations of the Store interface.
+// It can't live in the store package itself because of circular import issues
+package test
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/unistack-org/micro/v3/store/cache"
+
+	"github.com/unistack-org/micro/v3/store/memory"
+
+	"github.com/kr/pretty"
+	"github.com/unistack-org/micro/v3/store/cockroach"
+
+	"github.com/unistack-org/micro/v3/store"
+
+	"github.com/unistack-org/micro/v3/store/file"
+
+	"github.com/davecgh/go-spew/spew"
+)
+
+func fileStoreCleanup(db string, s store.Store) {
+	s.Close()
+	dir := filepath.Join(file.DefaultDir, db+"/")
+	os.RemoveAll(dir)
+}
+
+func cockroachStoreCleanup(db string, s store.Store) {
+	keys, _ := s.List()
+	for _, k := range keys {
+		s.Delete(k)
+	}
+	s.Close()
+}
+
+func memoryCleanup(db string, s store.Store) {
+	s.Close()
+}
+
+func cacheCleanup(db string, s store.Store) {
+	s.Close()
+}
+
+func TestStoreReInit(t *testing.T) {
+	tcs := []struct {
+		name    string
+		s       store.Store
+		cleanup func(db string, s store.Store)
+	}{
+		{name: "file", s: file.NewStore(store.Table("aaa")), cleanup: fileStoreCleanup},
+		{name: "cockroach", s: cockroach.NewStore(store.Table("aaa")), cleanup: cockroachStoreCleanup},
+		{name: "memory", s: memory.NewStore(store.Table("aaa")), cleanup: memoryCleanup},
+		{name: "cache", s: cache.NewStore(memory.NewStore(store.Table("aaa"))), cleanup: cacheCleanup},
+	}
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			defer tc.cleanup(file.DefaultDatabase, tc.s)
+			tc.s.Init(store.Table("bbb"))
+			if tc.s.Options().Table != "bbb" {
+				t.Error("Init didn't reinitialise the store")
+			}
+		})
+	}
+}
+
+func TestStoreBasic(t *testing.T) {
+	tcs := []struct {
+		name    string
+		s       store.Store
+		cleanup func(db string, s store.Store)
+	}{
+		{name: "file", s: file.NewStore(), cleanup: fileStoreCleanup},
+		{name: "cockroach", s: cockroach.NewStore(), cleanup: cockroachStoreCleanup},
+		{name: "memory", s: memory.NewStore(), cleanup: memoryCleanup},
+		{name: "cache", s: cache.NewStore(memory.NewStore()), cleanup: cacheCleanup},
+	}
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			defer tc.cleanup(file.DefaultDatabase, tc.s)
+			runStoreTest(tc.s, t)
+		})
+	}
+
+}
+
+func TestStoreTable(t *testing.T) {
+	tcs := []struct {
+		name    string
+		s       store.Store
+		cleanup func(db string, s store.Store)
+	}{
+		{name: "file", s: file.NewStore(store.Table("testTable")), cleanup: fileStoreCleanup},
+		{name: "cockroach", s: cockroach.NewStore(store.Table("testTable")), cleanup: cockroachStoreCleanup},
+		{name: "memory", s: memory.NewStore(store.Table("testTable")), cleanup: memoryCleanup},
+		{name: "cache", s: cache.NewStore(memory.NewStore(store.Table("testTable"))), cleanup: cacheCleanup},
+	}
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			defer tc.cleanup(file.DefaultDatabase, tc.s)
+			runStoreTest(tc.s, t)
+		})
+	}
+}
+
+func TestStoreDatabase(t *testing.T) {
+	tcs := []struct {
+		name    string
+		s       store.Store
+		cleanup func(db string, s store.Store)
+	}{
+		{name: "file", s: file.NewStore(store.Database("testdb")), cleanup: fileStoreCleanup},
+		{name: "cockroach", s: cockroach.NewStore(store.Database("testdb")), cleanup: cockroachStoreCleanup},
+		{name: "memory", s: memory.NewStore(store.Database("testdb")), cleanup: memoryCleanup},
+		{name: "cache", s: cache.NewStore(memory.NewStore(store.Database("testdb"))), cleanup: cacheCleanup},
+	}
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			defer tc.cleanup("testdb", tc.s)
+			runStoreTest(tc.s, t)
+		})
+	}
+}
+
+func TestStoreDatabaseTable(t *testing.T) {
+	tcs := []struct {
+		name    string
+		s       store.Store
+		cleanup func(db string, s store.Store)
+	}{
+		{name: "file", s: file.NewStore(store.Database("testdb"), store.Table("testTable")), cleanup: fileStoreCleanup},
+		{name: "cockroach", s: cockroach.NewStore(store.Database("testdb"), store.Table("testTable")), cleanup: cockroachStoreCleanup},
+		{name: "memory", s: memory.NewStore(store.Database("testdb"), store.Table("testTable")), cleanup: memoryCleanup},
+		{name: "cache", s: cache.NewStore(memory.NewStore(store.Database("testdb"), store.Table("testTable"))), cleanup: cacheCleanup},
+	}
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			defer tc.cleanup("testdb", tc.s)
+			runStoreTest(tc.s, t)
+		})
+	}
+}
+
+func runStoreTest(s store.Store, t *testing.T) {
+	if len(os.Getenv("IN_TRAVIS_CI")) == 0 {
+		t.Logf("Options %s %v\n", s.String(), s.Options())
+	}
+
+	expiryTests(s, t)
+	suffixPrefixExpiryTests(s, t)
+	readTests(s, t)
+	listTests(s, t)
+
+}
+
+func readTests(s store.Store, t *testing.T) {
+	// Test Table, Suffix and WriteOptions
+	if err := s.Write(&store.Record{
+		Key:   "foofoobarbar",
+		Value: []byte("something"),
+	}, store.WriteTTL(time.Millisecond*100)); err != nil {
+		t.Error(err)
+	}
+	if err := s.Write(&store.Record{
+		Key:   "foofoo",
+		Value: []byte("something"),
+	}, store.WriteExpiry(time.Now().Add(time.Millisecond*100))); err != nil {
+		t.Error(err)
+	}
+	if err := s.Write(&store.Record{
+		Key:   "barbar",
+		Value: []byte("something"),
+		// TTL has higher precedence than expiry
+	}, store.WriteExpiry(time.Now().Add(time.Hour)), store.WriteTTL(time.Millisecond*100)); err != nil {
+		t.Error(err)
+	}
+
+	if results, err := s.Read("foo", store.ReadPrefix(), store.ReadSuffix()); err != nil {
+		t.Error(err)
+	} else {
+		if len(results) != 1 {
+			t.Errorf("Expected 1 results, got %d: %# v", len(results), spew.Sdump(results))
+		}
+	}
+
+	time.Sleep(time.Millisecond * 100)
+
+	if results, err := s.List(); err != nil {
+		t.Fatalf("List failed: %s", err)
+	} else {
+		if len(results) != 0 {
+			t.Fatalf("Expiry options were not effective, results :%v", spew.Sdump(results))
+		}
+	}
+
+	// write the following records
+	for i := 0; i < 10; i++ {
+		s.Write(&store.Record{
+			Key:   fmt.Sprintf("a%d", i),
+			Value: []byte{},
+		})
+	}
+
+	// read back a few records
+	if results, err := s.Read("a", store.ReadLimit(5), store.ReadPrefix()); err != nil {
+		t.Error(err)
+	} else {
+		if len(results) != 5 {
+			t.Fatal("Expected 5 results, got ", len(results))
+		}
+		if !strings.HasPrefix(results[0].Key, "a") {
+			t.Fatalf("Expected a prefix, got %s", results[0].Key)
+		}
+	}
+
+	// read the rest back
+	if results, err := s.Read("a", store.ReadLimit(30), store.ReadOffset(5), store.ReadPrefix()); err != nil {
+		t.Fatal(err)
+	} else {
+		if len(results) != 5 {
+			t.Fatal("Expected 5 results, got ", len(results))
+		}
+	}
+}
+
+func listTests(s store.Store, t *testing.T) {
+	for i := 0; i < 10; i++ {
+		s.Write(&store.Record{Key: fmt.Sprintf("List%d", i), Value: []byte("bar")})
+	}
+
+	recs, err := s.List(store.ListPrefix("List"))
+	if err != nil {
+		t.Fatalf("Error listing records %s", err)
+	}
+	if len(recs) != 10 {
+		t.Fatalf("Expected 10 records, received %d", len(recs))
+	}
+
+	recs, err = s.List(store.ListPrefix("List"), store.ListLimit(5))
+	if err != nil {
+		t.Fatalf("Error listing records %s", err)
+	}
+	if len(recs) != 5 {
+		t.Fatalf("Expected 5 records, received %d", len(recs))
+	}
+
+	recs, err = s.List(store.ListPrefix("List"), store.ListOffset(6))
+	if err != nil {
+		t.Fatalf("Error listing records %s", err)
+	}
+	if len(recs) != 4 {
+		t.Fatalf("Expected 4 records, received %d %+v", len(recs), recs)
+	}
+
+	recs, err = s.List(store.ListPrefix("List"), store.ListOffset(6), store.ListLimit(2))
+	if err != nil {
+		t.Fatalf("Error listing records %s", err)
+	}
+	if len(recs) != 2 {
+		t.Fatalf("Expected 2 records, received %d %+v", len(recs), recs)
+	}
+
+}
+
+func expiryTests(s store.Store, t *testing.T) {
+	// Read and Write an expiring Record
+	if err := s.Write(&store.Record{
+		Key:    "Hello",
+		Value:  []byte("World"),
+		Expiry: time.Millisecond * 150,
+	}); err != nil {
+		t.Error(err)
+	}
+
+	if r, err := s.Read("Hello"); err != nil {
+		t.Fatal(err)
+	} else {
+		if len(r) != 1 {
+			t.Error("Read returned multiple records")
+		}
+		if r[0].Key != "Hello" {
+			t.Errorf("Expected %s, got %s", "Hello", r[0].Key)
+		}
+		if string(r[0].Value) != "World" {
+			t.Errorf("Expected %s, got %s", "World", r[0].Value)
+		}
+	}
+
+	// wait for expiry
+	time.Sleep(time.Millisecond * 200)
+
+	if _, err := s.Read("Hello"); err != store.ErrNotFound {
+		t.Errorf("Expected %# v, got %# v", store.ErrNotFound, err)
+	}
+
+	// exercise the different ways to write an expiry, record.expiry
+	s.Write(&store.Record{Key: "aaa", Value: []byte("bbb"), Expiry: 1 * time.Second})
+	s.Write(&store.Record{Key: "aaaa", Value: []byte("bbb"), Expiry: 1 * time.Second})
+	s.Write(&store.Record{Key: "aaaaa", Value: []byte("bbb"), Expiry: 1 * time.Second})
+	results, err := s.Read("a", store.ReadPrefix())
+	if err != nil {
+		t.Error(err)
+	}
+	if len(results) != 3 {
+		t.Fatal("Results should have returned 3 records")
+	}
+	time.Sleep(1 * time.Second)
+	results, err = s.Read("a", store.ReadPrefix())
+	if err != nil {
+		t.Error(err)
+	}
+	if len(results) != 0 {
+		t.Fatal("Results should have returned 0 records")
+	}
+
+	// exercise the different ways to write an expiry, WriteExpiry
+	s.Write(&store.Record{Key: "bbb", Value: []byte("bbb")}, store.WriteExpiry(time.Now().Add(1*time.Second)))
+	s.Write(&store.Record{Key: "bbbb", Value: []byte("bbb")}, store.WriteExpiry(time.Now().Add(1*time.Second)))
+	s.Write(&store.Record{Key: "bbbbb", Value: []byte("bbb")}, store.WriteExpiry(time.Now().Add(1*time.Second)))
+	results, err = s.Read("b", store.ReadPrefix())
+	if err != nil {
+		t.Error(err)
+	}
+	if len(results) != 3 {
+		t.Fatalf("Results should have returned 3 records. Received %d", len(results))
+	}
+	time.Sleep(1 * time.Second)
+	results, err = s.Read("b", store.ReadPrefix())
+	if err != nil {
+		t.Error(err)
+	}
+	if len(results) != 0 {
+		t.Fatalf("Results should have returned 0 records. Received %d", len(results))
+	}
+
+	// exercise the different ways to write an expiry, WriteTTL
+	s.Write(&store.Record{Key: "ccc", Value: []byte("bbb")}, store.WriteTTL(1*time.Second))
+	s.Write(&store.Record{Key: "cccc", Value: []byte("bbb")}, store.WriteTTL(1*time.Second))
+	s.Write(&store.Record{Key: "ccccc", Value: []byte("bbb")}, store.WriteTTL(1*time.Second))
+	results, err = s.Read("c", store.ReadPrefix())
+	if err != nil {
+		t.Error(err)
+	}
+	if len(results) != 3 {
+		t.Fatalf("Results should have returned 3 records. Received %d", len(results))
+	}
+	time.Sleep(1 * time.Second)
+	results, err = s.Read("c", store.ReadPrefix())
+	if err != nil {
+		t.Error(err)
+	}
+	if len(results) != 0 {
+		t.Fatalf("Results should have returned 0 records. Received %d", len(results))
+	}
+}
+
+func suffixPrefixExpiryTests(s store.Store, t *testing.T) {
+	// Write 3 records with various expiry and get with Prefix
+	records := []*store.Record{
+		&store.Record{
+			Key:   "foo",
+			Value: []byte("foofoo"),
+		},
+		&store.Record{
+			Key:    "foobar",
+			Value:  []byte("foobarfoobar"),
+			Expiry: time.Millisecond * 100,
+		},
+	}
+
+	for _, r := range records {
+		if err := s.Write(r); err != nil {
+			t.Errorf("Couldn't write k: %s, v: %# v (%s)", r.Key, pretty.Formatter(r.Value), err)
+		}
+	}
+
+	if results, err := s.Read("foo", store.ReadPrefix()); err != nil {
+		t.Errorf("Couldn't read all \"foo\" keys, got %#v (%s)", spew.Sdump(results), err)
+	} else {
+		if len(results) != 2 {
+			t.Errorf("Expected 2 items, got %d", len(results))
+		}
+	}
+
+	// wait for the expiry
+	time.Sleep(time.Millisecond * 200)
+
+	if results, err := s.Read("foo", store.ReadPrefix()); err != nil {
+		t.Errorf("Couldn't read all \"foo\" keys, got %# v (%s)", spew.Sdump(results), err)
+	} else if len(results) != 1 {
+		t.Errorf("Expected 1 item, got %d", len(results))
+	}
+
+	if err := s.Delete("foo"); err != nil {
+		t.Errorf("Delete failed (%v)", err)
+	}
+
+	if results, err := s.Read("foo"); err != store.ErrNotFound {
+		t.Errorf("Expected read failure read all \"foo\" keys, got %# v (%s)", spew.Sdump(results), err)
+	} else {
+		if len(results) != 0 {
+			t.Errorf("Expected 0 items, got %d (%# v)", len(results), spew.Sdump(results))
+		}
+	}
+
+	// Write 3 records with various expiry and get with Suffix
+	records = []*store.Record{
+		&store.Record{
+			Key:   "foo",
+			Value: []byte("foofoo"),
+		},
+		&store.Record{
+			Key:   "barfoo",
+			Value: []byte("barfoobarfoo"),
+
+			Expiry: time.Millisecond * 100,
+		},
+		&store.Record{
+			Key:    "bazbarfoo",
+			Value:  []byte("bazbarfoobazbarfoo"),
+			Expiry: 2 * time.Millisecond * 100,
+		},
+	}
+	for _, r := range records {
+		if err := s.Write(r); err != nil {
+			t.Errorf("Couldn't write k: %s, v: %# v (%s)", r.Key, pretty.Formatter(r.Value), err)
+		}
+	}
+	if results, err := s.Read("foo", store.ReadSuffix()); err != nil {
+		t.Errorf("Couldn't read all \"foo\" keys, got %# v (%s)", spew.Sdump(results), err)
+	} else {
+		if len(results) != 3 {
+			t.Errorf("Expected 3 items, got %d", len(results))
+			//t.Logf("Table test: %v\n", spew.Sdump(results))
+		}
+
+	}
+	time.Sleep(time.Millisecond * 100)
+	if results, err := s.Read("foo", store.ReadSuffix()); err != nil {
+		t.Errorf("Couldn't read all \"foo\" keys, got %# v (%s)", spew.Sdump(results), err)
+	} else {
+		if len(results) != 2 {
+			t.Errorf("Expected 2 items, got %d", len(results))
+			//t.Logf("Table test: %v\n", spew.Sdump(results))
+		}
+
+	}
+	time.Sleep(time.Millisecond * 100)
+	if results, err := s.Read("foo", store.ReadSuffix()); err != nil {
+		t.Errorf("Couldn't read all \"foo\" keys, got %# v (%s)", spew.Sdump(results), err)
+	} else {
+		if len(results) != 1 {
+			t.Errorf("Expected 1 item, got %d", len(results))
+			//	t.Logf("Table test: %# v\n", spew.Sdump(results))
+		}
+	}
+	if err := s.Delete("foo"); err != nil {
+		t.Errorf("Delete failed (%v)", err)
+	}
+	if results, err := s.Read("foo", store.ReadSuffix()); err != nil {
+		t.Errorf("Couldn't read all \"foo\" keys, got %# v (%s)", spew.Sdump(results), err)
+	} else {
+		if len(results) != 0 {
+			t.Errorf("Expected 0 items, got %d (%# v)", len(results), spew.Sdump(results))
+		}
+	}
+}