1	/*
     2	Copyright 2013 The Perkeep Authors
     3	
     4	Licensed under the Apache License, Version 2.0 (the "License");
     5	you may not use this file except in compliance with the License.
     6	You may obtain a copy of the License at
     7	
     8	     http://www.apache.org/licenses/LICENSE-2.0
     9	
    10	Unless required by applicable law or agreed to in writing, software
    11	distributed under the License is distributed on an "AS IS" BASIS,
    12	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13	See the License for the specific language governing permissions and
    14	limitations under the License.
    15	*/
    16	
    17	// Package kvtest tests sorted.KeyValue implementations.
    18	package kvtest // import "perkeep.org/pkg/sorted/kvtest"
    19	
    20	import (
    21		"reflect"
    22		"testing"
    23		"time"
    24	
    25		"perkeep.org/pkg/sorted"
    26		"perkeep.org/pkg/test"
    27	)
    28	
    29	func TestSorted(t *testing.T, kv sorted.KeyValue) {
    30		defer test.TLog(t)()
    31		if !isEmpty(t, kv) {
    32			t.Fatal("kv for test is expected to be initially empty")
    33		}
    34		set := func(k, v string) {
    35			if err := kv.Set(k, v); err != nil {
    36				t.Fatalf("Error setting %q to %q: %v", k, v, err)
    37			}
    38		}
    39		set("foo", "bar")
    40		if isEmpty(t, kv) {
    41			t.Fatalf("iterator reports the kv is empty after adding foo=bar; iterator must be broken")
    42		}
    43		if v, err := kv.Get("foo"); err != nil || v != "bar" {
    44			t.Errorf("get(foo) = %q, %v; want bar", v, err)
    45		}
    46		if v, err := kv.Get("NOT_EXIST"); err != sorted.ErrNotFound {
    47			t.Errorf("get(NOT_EXIST) = %q, %v; want error sorted.ErrNotFound", v, err)
    48		}
    49		for i := 0; i < 2; i++ {
    50			if err := kv.Delete("foo"); err != nil {
    51				t.Errorf("Delete(foo) (on loop %d/2) returned error %v", i+1, err)
    52			}
    53		}
    54		set("a", "av")
    55		set("b", "bv")
    56		set("c", "cv")
    57		testEnumerate(t, kv, "", "", "av", "bv", "cv")
    58		testEnumerate(t, kv, "a", "", "av", "bv", "cv")
    59		testEnumerate(t, kv, "b", "", "bv", "cv")
    60		testEnumerate(t, kv, "a", "c", "av", "bv")
    61		testEnumerate(t, kv, "a", "b", "av")
    62		testEnumerate(t, kv, "a", "a")
    63		testEnumerate(t, kv, "d", "")
    64		testEnumerate(t, kv, "d", "e")
    65	
    66		// Verify that < comparison works identically for all DBs (because it is affected by collation rules)
    67		// http://postgresql.1045698.n5.nabble.com/String-comparison-and-the-SQL-standard-td5740721.html
    68		set("foo|abc", "foo|abcv")
    69		testEnumerate(t, kv, "foo|", "", "foo|abcv")
    70		testEnumerate(t, kv, "foo|", "foo}", "foo|abcv")
    71	
    72		// Verify that the value isn't being used instead of the key in the range comparison.
    73		set("y", "x:foo")
    74		testEnumerate(t, kv, "x:", "x~")
    75	
    76		testInsertLarge(t, kv)
    77		testInsertTooLarge(t, kv)
    78	
    79		// TODO: test batch commits
    80	
    81		// Deleting a non-existent item in a batch should not be an error
    82		testDeleteNotFoundBatch(t, kv)
    83		testDeletePartialNotFoundBatch(t, kv)
    84	
    85		if txReader, ok := kv.(sorted.TransactionalReader); ok {
    86			testReadTransaction(t, txReader)
    87		}
    88	}
    89	
    90	// Do not ever insert that key, as it used for testing deletion of non existing entries
    91	const (
    92		notExistKey  = "I do not exist"
    93		butIExistKey = "But I do exist"
    94	)
    95	
    96	func testDeleteNotFoundBatch(t *testing.T, kv sorted.KeyValue) {
    97		b := kv.BeginBatch()
    98		b.Delete(notExistKey)
    99		if err := kv.CommitBatch(b); err != nil {
   100			t.Fatalf("Batch deletion of non existing key returned an error: %v", err)
   101		}
   102	}
   103	
   104	func testDeletePartialNotFoundBatch(t *testing.T, kv sorted.KeyValue) {
   105		if err := kv.Set(butIExistKey, "whatever"); err != nil {
   106			t.Fatal(err)
   107		}
   108		b := kv.BeginBatch()
   109		b.Delete(notExistKey)
   110		b.Delete(butIExistKey)
   111		if err := kv.CommitBatch(b); err != nil {
   112			t.Fatalf("Batch deletion with one non existing key returned an error: %v", err)
   113		}
   114		if val, err := kv.Get(butIExistKey); err != sorted.ErrNotFound || val != "" {
   115			t.Fatalf("Key %q should have been batch deleted", butIExistKey)
   116		}
   117	}
   118	
   119	func testInsertLarge(t *testing.T, kv sorted.KeyValue) {
   120		largeKey := make([]byte, sorted.MaxKeySize-1)
   121		// setting all the bytes because postgres whines about an invalid byte sequence
   122		// otherwise
   123		for k := range largeKey {
   124			largeKey[k] = 'A'
   125		}
   126		largeKey[sorted.MaxKeySize-2] = 'B'
   127		largeValue := make([]byte, sorted.MaxValueSize-1)
   128		for k := range largeValue {
   129			largeValue[k] = 'A'
   130		}
   131		largeValue[sorted.MaxValueSize-2] = 'B'
   132	
   133		// insert with large key
   134		if err := kv.Set(string(largeKey), "whatever"); err != nil {
   135			t.Fatalf("Insertion of large key failed: %v", err)
   136		}
   137	
   138		// and verify we can get it back, i.e. that the key hasn't been truncated.
   139		it := kv.Find(string(largeKey), "")
   140		if !it.Next() || it.Key() != string(largeKey) || it.Value() != "whatever" {
   141			it.Close()
   142			t.Fatalf("Find(largeKey) = %q, %q; want %q, %q", it.Key(), it.Value(), largeKey, "whatever")
   143		}
   144		it.Close()
   145	
   146		// insert with large value
   147		if err := kv.Set("whatever", string(largeValue)); err != nil {
   148			t.Fatalf("Insertion of large value failed: %v", err)
   149		}
   150		// and verify we can get it back, i.e. that the value hasn't been truncated.
   151		if v, err := kv.Get("whatever"); err != nil || v != string(largeValue) {
   152			t.Fatalf("get(\"whatever\") = %q, %v; want %q", v, err, largeValue)
   153		}
   154	
   155		// insert with large key and large value
   156		if err := kv.Set(string(largeKey), string(largeValue)); err != nil {
   157			t.Fatalf("Insertion of large key and value failed: %v", err)
   158		}
   159		// and verify we can get them back
   160		it = kv.Find(string(largeKey), "")
   161		defer it.Close()
   162		if !it.Next() || it.Key() != string(largeKey) || it.Value() != string(largeValue) {
   163			t.Fatalf("Find(largeKey) = %q, %q; want %q, %q", it.Key(), it.Value(), largeKey, largeValue)
   164		}
   165	}
   166	
   167	func testInsertTooLarge(t *testing.T, kv sorted.KeyValue) {
   168		largeKey := make([]byte, sorted.MaxKeySize+1)
   169		largeValue := make([]byte, sorted.MaxValueSize+1)
   170		err := kv.Set(string(largeKey), "whatever")
   171		if err != nil {
   172			t.Fatalf("Insertion of too large a key should have skipped with some logging, but err was %v", err)
   173		}
   174		_, err = kv.Get(string(largeKey))
   175		if err == nil {
   176			t.Fatal("large key should not have been inserted")
   177		}
   178		err = kv.Set("testInsertTooLarge", string(largeValue))
   179		if err != nil {
   180			t.Fatalf("Insertion of too large a value should have skipped with some logging, but err was %v", err)
   181		}
   182		_, err = kv.Get("testInsertTooLarge")
   183		if err == nil {
   184			t.Fatal("large value should not have been inserted")
   185		}
   186	}
   187	
   188	func testEnumerate(t *testing.T, kv sorted.KeyValue, start, end string, want ...string) {
   189		var got []string
   190		it := kv.Find(start, end)
   191		for it.Next() {
   192			key, val := it.Key(), it.Value()
   193			keyb, valb := it.KeyBytes(), it.ValueBytes()
   194			if key != string(keyb) {
   195				t.Errorf("Key and KeyBytes disagree: %q vs %q", key, keyb)
   196			}
   197			if val != string(valb) {
   198				t.Errorf("Value and ValueBytes disagree: %q vs %q", val, valb)
   199			}
   200			if key+"v" != val {
   201				t.Errorf("iterator returned unexpected pair for test: %q, %q", key, val)
   202			}
   203			got = append(got, val)
   204		}
   205		err := it.Close()
   206		if err != nil {
   207			t.Errorf("for enumerate of (%q, %q), Close error: %v", start, end, err)
   208		}
   209		if !reflect.DeepEqual(got, want) {
   210			t.Errorf("for enumerate of (%q, %q), got: %q; want %q", start, end, got, want)
   211		}
   212	}
   213	
   214	func isEmpty(t *testing.T, kv sorted.KeyValue) bool {
   215		it := kv.Find("", "")
   216		hasRow := it.Next()
   217		if err := it.Close(); err != nil {
   218			t.Fatalf("Error closing iterator while testing for emptiness: %v", err)
   219		}
   220		return !hasRow
   221	}
   222	
   223	func testReadTransaction(t *testing.T, kv sorted.TransactionalReader) {
   224		set := func(k, v string) {
   225			if err := kv.Set(k, v); err != nil {
   226				t.Fatalf("Error setting %q to %q: %v", k, v, err)
   227			}
   228		}
   229		set("raceKey", "orig")
   230		tx := kv.BeginReadTx()
   231	
   232		// We want to be sure the transaction is always closed before exiting,
   233		// but we can't just defer tx.Close(), because on implementations that
   234		// implement transactions with simple locks, the last call to set()
   235		// below cannot run until the read transaction is closed. Furthermore,
   236		// we need to make sure set() completes before returning, because if the
   237		// caller closes the database connection before set() runs, it will
   238		// panic.
   239		//
   240		// On the happy path, the sequence of events looks like:
   241		//
   242		// 1. Explicitly close the transaction.
   243		// 2. Wait for set() to complete.
   244		// 3. Return.
   245		//
   246		// ...but we use the boolean and defer statement below to ensure cleanup
   247		// on errors.
   248		txClosed := false
   249		defer func() {
   250			if !txClosed {
   251				tx.Close()
   252			}
   253		}()
   254	
   255		get := func(k string) string {
   256			v, err := tx.Get(k)
   257			if err != nil {
   258				t.Fatalf("Error getting %q: %v", k, err)
   259			}
   260			return v
   261		}
   262		if get("raceKey") != "orig" {
   263			t.Fatalf("Read saw the wrong initial value")
   264		}
   265	
   266		done := make(chan struct{}, 1)
   267		go func() {
   268			set("raceKey", "new")
   269			done <- struct{}{}
   270		}()
   271	
   272		time.Sleep(time.Second / 5)
   273		if get("raceKey") != "orig" {
   274			t.Fatalf("Read transaction saw an update that happened after it started")
   275		}
   276	
   277		tx.Close()
   278		txClosed = true
   279		<-done
   280	}