// Copyright 2016 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package mocktikv

import (
	"math"
	"testing"

	. "github.com/pingcap/check"
	"github.com/pingcap/errors"
	"github.com/pingcap/kvproto/pkg/kvrpcpb"
)

func TestT(t *testing.T) {
	TestingT(t)
}

// testMockTiKVSuite tests MVCCStore interface.
// SetUpTest should set specific MVCCStore implementation.
type testMockTiKVSuite struct {
	store MVCCStore
}

type testMarshal struct{}

// testMVCCLevelDB is used to test MVCCLevelDB implementation.
type testMVCCLevelDB struct {
	testMockTiKVSuite
}

var (
	_ = Suite(&testMockTiKVSuite{})
	_ = Suite(&testMVCCLevelDB{})
	_ = Suite(testMarshal{})
)

func (s *testMockTiKVSuite) SetUpTest(c *C) {
	var err error
	s.store, err = NewMVCCLevelDB("")
	c.Assert(err, IsNil)
}

// PutMutations is exported for testing.
var PutMutations func(kvpairs ...string) []*kvrpcpb.Mutation = putMutations

func putMutations(kvpairs ...string) []*kvrpcpb.Mutation {
	var mutations []*kvrpcpb.Mutation
	for i := 0; i < len(kvpairs); i += 2 {
		mutations = append(mutations, &kvrpcpb.Mutation{
			Op:    kvrpcpb.Op_Put,
			Key:   []byte(kvpairs[i]),
			Value: []byte(kvpairs[i+1]),
		})
	}
	return mutations
}

func lock(key, primary string, ts uint64) *kvrpcpb.LockInfo {
	return &kvrpcpb.LockInfo{
		Key:         []byte(key),
		PrimaryLock: []byte(primary),
		LockVersion: ts,
	}
}

func (s *testMockTiKVSuite) mustGetNone(c *C, key string, ts uint64) {
	val, err := s.store.Get([]byte(key), ts, kvrpcpb.IsolationLevel_SI, nil)
	c.Assert(err, IsNil)
	c.Assert(val, IsNil)
}

func (s *testMockTiKVSuite) mustGetErr(c *C, key string, ts uint64) {
	val, err := s.store.Get([]byte(key), ts, kvrpcpb.IsolationLevel_SI, nil)
	c.Assert(err, NotNil)
	c.Assert(val, IsNil)
}

func (s *testMockTiKVSuite) mustGetOK(c *C, key string, ts uint64, expect string) {
	val, err := s.store.Get([]byte(key), ts, kvrpcpb.IsolationLevel_SI, nil)
	c.Assert(err, IsNil)
	c.Assert(string(val), Equals, expect)
}

func (s *testMockTiKVSuite) mustGetRC(c *C, key string, ts uint64, expect string) {
	val, err := s.store.Get([]byte(key), ts, kvrpcpb.IsolationLevel_RC, nil)
	c.Assert(err, IsNil)
	c.Assert(string(val), Equals, expect)
}

func (s *testMockTiKVSuite) mustPutOK(c *C, key, value string, startTS, commitTS uint64) {
	req := &kvrpcpb.PrewriteRequest{
		Mutations:    putMutations(key, value),
		PrimaryLock:  []byte(key),
		StartVersion: startTS,
	}
	errs := s.store.Prewrite(req)
	for _, err := range errs {
		c.Assert(err, IsNil)
	}
	err := s.store.Commit([][]byte{[]byte(key)}, startTS, commitTS)
	c.Assert(err, IsNil)
}

func (s *testMockTiKVSuite) mustDeleteOK(c *C, key string, startTS, commitTS uint64) {
	mutations := []*kvrpcpb.Mutation{
		{
			Op:  kvrpcpb.Op_Del,
			Key: []byte(key),
		},
	}
	req := &kvrpcpb.PrewriteRequest{
		Mutations:    mutations,
		PrimaryLock:  []byte(key),
		StartVersion: startTS,
	}
	errs := s.store.Prewrite(req)
	for _, err := range errs {
		c.Assert(err, IsNil)
	}
	err := s.store.Commit([][]byte{[]byte(key)}, startTS, commitTS)
	c.Assert(err, IsNil)
}

func (s *testMockTiKVSuite) mustScanOK(c *C, start string, limit int, ts uint64, expect ...string) {
	s.mustRangeScanOK(c, start, "", limit, ts, expect...)
}

func (s *testMockTiKVSuite) mustRangeScanOK(c *C, start, end string, limit int, ts uint64, expect ...string) {
	pairs := s.store.Scan([]byte(start), []byte(end), limit, ts, kvrpcpb.IsolationLevel_SI, nil)
	c.Assert(len(pairs)*2, Equals, len(expect))
	for i := 0; i < len(pairs); i++ {
		c.Assert(pairs[i].Err, IsNil)
		c.Assert(pairs[i].Key, BytesEquals, []byte(expect[i*2]))
		c.Assert(string(pairs[i].Value), Equals, expect[i*2+1])
	}
}

func (s *testMockTiKVSuite) mustReverseScanOK(c *C, end string, limit int, ts uint64, expect ...string) {
	s.mustRangeReverseScanOK(c, "", end, limit, ts, expect...)
}

func (s *testMockTiKVSuite) mustRangeReverseScanOK(c *C, start, end string, limit int, ts uint64, expect ...string) {
	pairs := s.store.ReverseScan([]byte(start), []byte(end), limit, ts, kvrpcpb.IsolationLevel_SI, nil)
	c.Assert(len(pairs)*2, Equals, len(expect))
	for i := 0; i < len(pairs); i++ {
		c.Assert(pairs[i].Err, IsNil)
		c.Assert(pairs[i].Key, BytesEquals, []byte(expect[i*2]))
		c.Assert(string(pairs[i].Value), Equals, expect[i*2+1])
	}
}

func MustPrewriteOK(c *C, store MVCCStore, mutations []*kvrpcpb.Mutation, primary string, startTS uint64, ttl uint64) {
	s := testMockTiKVSuite{store}
	s.mustPrewriteWithTTLOK(c, mutations, primary, startTS, ttl)
}

func (s *testMockTiKVSuite) mustPrewriteOK(c *C, mutations []*kvrpcpb.Mutation, primary string, startTS uint64) {
	s.mustPrewriteWithTTLOK(c, mutations, primary, startTS, 0)
}

func (s *testMockTiKVSuite) mustPrewriteWithTTLOK(c *C, mutations []*kvrpcpb.Mutation, primary string, startTS uint64, ttl uint64) {
	req := &kvrpcpb.PrewriteRequest{
		Mutations:    mutations,
		PrimaryLock:  []byte(primary),
		StartVersion: startTS,
		LockTtl:      ttl,
		MinCommitTs:  startTS + 1,
	}
	errs := s.store.Prewrite(req)
	for _, err := range errs {
		c.Assert(err, IsNil)
	}
}

func (s *testMockTiKVSuite) mustCommitOK(c *C, keys [][]byte, startTS, commitTS uint64) {
	err := s.store.Commit(keys, startTS, commitTS)
	c.Assert(err, IsNil)
}

func (s *testMockTiKVSuite) mustCommitErr(c *C, keys [][]byte, startTS, commitTS uint64) {
	err := s.store.Commit(keys, startTS, commitTS)
	c.Assert(err, NotNil)
}

func (s *testMockTiKVSuite) mustRollbackOK(c *C, keys [][]byte, startTS uint64) {
	err := s.store.Rollback(keys, startTS)
	c.Assert(err, IsNil)
}

func (s *testMockTiKVSuite) mustRollbackErr(c *C, keys [][]byte, startTS uint64) {
	err := s.store.Rollback(keys, startTS)
	c.Assert(err, NotNil)
}

func (s *testMockTiKVSuite) mustScanLock(c *C, maxTs uint64, expect []*kvrpcpb.LockInfo) {
	locks, err := s.store.ScanLock(nil, nil, maxTs)
	c.Assert(err, IsNil)
	c.Assert(locks, DeepEquals, expect)
}

func (s *testMockTiKVSuite) mustResolveLock(c *C, startTS, commitTS uint64) {
	c.Assert(s.store.ResolveLock(nil, nil, startTS, commitTS), IsNil)
}

func (s *testMockTiKVSuite) mustBatchResolveLock(c *C, txnInfos map[uint64]uint64) {
	c.Assert(s.store.BatchResolveLock(nil, nil, txnInfos), IsNil)
}

func (s *testMockTiKVSuite) mustGC(c *C, safePoint uint64) {
	c.Assert(s.store.GC(nil, nil, safePoint), IsNil)
}

func (s *testMockTiKVSuite) mustDeleteRange(c *C, startKey, endKey string) {
	err := s.store.DeleteRange([]byte(startKey), []byte(endKey))
	c.Assert(err, IsNil)
}

func (s *testMockTiKVSuite) TestGet(c *C) {
	s.mustGetNone(c, "x", 10)
	s.mustPutOK(c, "x", "x", 5, 10)
	s.mustGetNone(c, "x", 9)
	s.mustGetOK(c, "x", 10, "x")
	s.mustGetOK(c, "x", 11, "x")
}

func (s *testMockTiKVSuite) TestGetWithLock(c *C) {
	key := "key"
	value := "value"
	s.mustPutOK(c, key, value, 5, 10)
	mutations := []*kvrpcpb.Mutation{{
		Op:  kvrpcpb.Op_Lock,
		Key: []byte(key),
	},
	}
	// test with lock's type is lock
	s.mustPrewriteOK(c, mutations, key, 20)
	s.mustGetOK(c, key, 25, value)
	s.mustCommitOK(c, [][]byte{[]byte(key)}, 20, 30)

	// test get with lock's max ts and primary key
	s.mustPrewriteOK(c, putMutations(key, "value2", "key2", "v5"), key, 40)
	s.mustGetErr(c, key, 41)
	s.mustGetErr(c, "key2", math.MaxUint64)
	s.mustGetOK(c, key, math.MaxUint64, "value")
}

func (s *testMockTiKVSuite) TestDelete(c *C) {
	s.mustPutOK(c, "x", "x5-10", 5, 10)
	s.mustDeleteOK(c, "x", 15, 20)
	s.mustGetNone(c, "x", 5)
	s.mustGetNone(c, "x", 9)
	s.mustGetOK(c, "x", 10, "x5-10")
	s.mustGetOK(c, "x", 19, "x5-10")
	s.mustGetNone(c, "x", 20)
	s.mustGetNone(c, "x", 21)
}

func (s *testMockTiKVSuite) TestCleanupRollback(c *C) {
	s.mustPutOK(c, "secondary", "s-0", 1, 2)
	s.mustPrewriteOK(c, putMutations("primary", "p-5", "secondary", "s-5"), "primary", 5)
	s.mustGetErr(c, "secondary", 8)
	s.mustGetErr(c, "secondary", 12)
	s.mustCommitOK(c, [][]byte{[]byte("primary")}, 5, 10)
	s.mustRollbackErr(c, [][]byte{[]byte("primary")}, 5)
}

func (s *testMockTiKVSuite) TestReverseScan(c *C) {
	// ver10: A(10) - B(_) - C(10) - D(_) - E(10)
	s.mustPutOK(c, "A", "A10", 5, 10)
	s.mustPutOK(c, "C", "C10", 5, 10)
	s.mustPutOK(c, "E", "E10", 5, 10)

	checkV10 := func() {
		s.mustReverseScanOK(c, "Z", 0, 10)
		s.mustReverseScanOK(c, "Z", 1, 10, "E", "E10")
		s.mustReverseScanOK(c, "Z", 2, 10, "E", "E10", "C", "C10")
		s.mustReverseScanOK(c, "Z", 3, 10, "E", "E10", "C", "C10", "A", "A10")
		s.mustReverseScanOK(c, "Z", 4, 10, "E", "E10", "C", "C10", "A", "A10")
		s.mustReverseScanOK(c, "E\x00", 3, 10, "E", "E10", "C", "C10", "A", "A10")
		s.mustReverseScanOK(c, "C\x00", 3, 10, "C", "C10", "A", "A10")
		s.mustReverseScanOK(c, "C\x00", 4, 10, "C", "C10", "A", "A10")
		s.mustReverseScanOK(c, "B", 1, 10, "A", "A10")
		s.mustRangeReverseScanOK(c, "", "E", 5, 10, "C", "C10", "A", "A10")
		s.mustRangeReverseScanOK(c, "", "C\x00", 5, 10, "C", "C10", "A", "A10")
		s.mustRangeReverseScanOK(c, "A\x00", "C", 5, 10)
	}
	checkV10()

	// ver20: A(10) - B(20) - C(10) - D(20) - E(10)
	s.mustPutOK(c, "B", "B20", 15, 20)
	s.mustPutOK(c, "D", "D20", 15, 20)

	checkV20 := func() {
		s.mustReverseScanOK(c, "Z", 5, 20, "E", "E10", "D", "D20", "C", "C10", "B", "B20", "A", "A10")
		s.mustReverseScanOK(c, "C\x00", 5, 20, "C", "C10", "B", "B20", "A", "A10")
		s.mustReverseScanOK(c, "A\x00", 1, 20, "A", "A10")
		s.mustRangeReverseScanOK(c, "B", "D", 5, 20, "C", "C10", "B", "B20")
		s.mustRangeReverseScanOK(c, "B", "D\x00", 5, 20, "D", "D20", "C", "C10", "B", "B20")
		s.mustRangeReverseScanOK(c, "B\x00", "D\x00", 5, 20, "D", "D20", "C", "C10")
	}
	checkV10()
	checkV20()

	// ver30: A(_) - B(20) - C(10) - D(_) - E(10)
	s.mustDeleteOK(c, "A", 25, 30)
	s.mustDeleteOK(c, "D", 25, 30)

	checkV30 := func() {
		s.mustReverseScanOK(c, "Z", 5, 30, "E", "E10", "C", "C10", "B", "B20")
		s.mustReverseScanOK(c, "C", 1, 30, "B", "B20")
		s.mustReverseScanOK(c, "C\x00", 5, 30, "C", "C10", "B", "B20")
	}
	checkV10()
	checkV20()
	checkV30()

	// ver40: A(_) - B(_) - C(40) - D(40) - E(10)
	s.mustDeleteOK(c, "B", 35, 40)
	s.mustPutOK(c, "C", "C40", 35, 40)
	s.mustPutOK(c, "D", "D40", 35, 40)

	checkV40 := func() {
		s.mustReverseScanOK(c, "Z", 5, 40, "E", "E10", "D", "D40", "C", "C40")
		s.mustReverseScanOK(c, "Z", 5, 100, "E", "E10", "D", "D40", "C", "C40")
	}
	checkV10()
	checkV20()
	checkV30()
	checkV40()
}

func (s *testMockTiKVSuite) TestScan(c *C) {
	// ver10: A(10) - B(_) - C(10) - D(_) - E(10)
	s.mustPutOK(c, "A", "A10", 5, 10)
	s.mustPutOK(c, "C", "C10", 5, 10)
	s.mustPutOK(c, "E", "E10", 5, 10)

	checkV10 := func() {
		s.mustScanOK(c, "", 0, 10)
		s.mustScanOK(c, "", 1, 10, "A", "A10")
		s.mustScanOK(c, "", 2, 10, "A", "A10", "C", "C10")
		s.mustScanOK(c, "", 3, 10, "A", "A10", "C", "C10", "E", "E10")
		s.mustScanOK(c, "", 4, 10, "A", "A10", "C", "C10", "E", "E10")
		s.mustScanOK(c, "A", 3, 10, "A", "A10", "C", "C10", "E", "E10")
		s.mustScanOK(c, "A\x00", 3, 10, "C", "C10", "E", "E10")
		s.mustScanOK(c, "C", 4, 10, "C", "C10", "E", "E10")
		s.mustScanOK(c, "F", 1, 10)
		s.mustRangeScanOK(c, "", "E", 5, 10, "A", "A10", "C", "C10")
		s.mustRangeScanOK(c, "", "C\x00", 5, 10, "A", "A10", "C", "C10")
		s.mustRangeScanOK(c, "A\x00", "C", 5, 10)
	}
	checkV10()

	// ver20: A(10) - B(20) - C(10) - D(20) - E(10)
	s.mustPutOK(c, "B", "B20", 15, 20)
	s.mustPutOK(c, "D", "D20", 15, 20)

	checkV20 := func() {
		s.mustScanOK(c, "", 5, 20, "A", "A10", "B", "B20", "C", "C10", "D", "D20", "E", "E10")
		s.mustScanOK(c, "C", 5, 20, "C", "C10", "D", "D20", "E", "E10")
		s.mustScanOK(c, "D\x00", 1, 20, "E", "E10")
		s.mustRangeScanOK(c, "B", "D", 5, 20, "B", "B20", "C", "C10")
		s.mustRangeScanOK(c, "B", "D\x00", 5, 20, "B", "B20", "C", "C10", "D", "D20")
		s.mustRangeScanOK(c, "B\x00", "D\x00", 5, 20, "C", "C10", "D", "D20")
	}
	checkV10()
	checkV20()

	// ver30: A(_) - B(20) - C(10) - D(_) - E(10)
	s.mustDeleteOK(c, "A", 25, 30)
	s.mustDeleteOK(c, "D", 25, 30)

	checkV30 := func() {
		s.mustScanOK(c, "", 5, 30, "B", "B20", "C", "C10", "E", "E10")
		s.mustScanOK(c, "A", 1, 30, "B", "B20")
		s.mustScanOK(c, "C\x00", 5, 30, "E", "E10")
	}
	checkV10()
	checkV20()
	checkV30()

	// ver40: A(_) - B(_) - C(40) - D(40) - E(10)
	s.mustDeleteOK(c, "B", 35, 40)
	s.mustPutOK(c, "C", "C40", 35, 40)
	s.mustPutOK(c, "D", "D40", 35, 40)

	checkV40 := func() {
		s.mustScanOK(c, "", 5, 40, "C", "C40", "D", "D40", "E", "E10")
		s.mustScanOK(c, "", 5, 100, "C", "C40", "D", "D40", "E", "E10")
	}
	checkV10()
	checkV20()
	checkV30()
	checkV40()
}

func (s *testMockTiKVSuite) TestBatchGet(c *C) {
	s.mustPutOK(c, "k1", "v1", 1, 2)
	s.mustPutOK(c, "k2", "v2", 1, 2)
	s.mustPutOK(c, "k2", "v2", 3, 4)
	s.mustPutOK(c, "k3", "v3", 1, 2)
	batchKeys := [][]byte{[]byte("k1"), []byte("k2"), []byte("k3")}
	pairs := s.store.BatchGet(batchKeys, 5, kvrpcpb.IsolationLevel_SI, nil)
	for _, pair := range pairs {
		c.Assert(pair.Err, IsNil)
	}
	c.Assert(string(pairs[0].Value), Equals, "v1")
	c.Assert(string(pairs[1].Value), Equals, "v2")
	c.Assert(string(pairs[2].Value), Equals, "v3")
}

func (s *testMockTiKVSuite) TestScanLock(c *C) {
	s.mustPutOK(c, "k1", "v1", 1, 2)
	s.mustPrewriteOK(c, putMutations("p1", "v5", "s1", "v5"), "p1", 5)
	s.mustPrewriteOK(c, putMutations("p2", "v10", "s2", "v10"), "p2", 10)
	s.mustPrewriteOK(c, putMutations("p3", "v20", "s3", "v20"), "p3", 20)

	locks, err := s.store.ScanLock([]byte("a"), []byte("r"), 12)
	c.Assert(err, IsNil)
	c.Assert(locks, DeepEquals, []*kvrpcpb.LockInfo{
		lock("p1", "p1", 5),
		lock("p2", "p2", 10),
	})

	s.mustScanLock(c, 10, []*kvrpcpb.LockInfo{
		lock("p1", "p1", 5),
		lock("p2", "p2", 10),
		lock("s1", "p1", 5),
		lock("s2", "p2", 10),
	})
}

func (s *testMockTiKVSuite) TestScanWithResolvedLock(c *C) {
	s.mustPrewriteOK(c, putMutations("p1", "v5", "s1", "v5"), "p1", 5)
	s.mustPrewriteOK(c, putMutations("p2", "v10", "s2", "v10"), "p1", 5)

	pairs := s.store.Scan([]byte("p1"), nil, 3, 10, kvrpcpb.IsolationLevel_SI, nil)
	lock, ok := errors.Cause(pairs[0].Err).(*ErrLocked)
	c.Assert(ok, IsTrue)
	_, ok = errors.Cause(pairs[1].Err).(*ErrLocked)
	c.Assert(ok, IsTrue)

	// Mock the request after resolving lock.
	pairs = s.store.Scan([]byte("p1"), nil, 3, 10, kvrpcpb.IsolationLevel_SI, []uint64{lock.StartTS})
	for _, pair := range pairs {
		c.Assert(pair.Err, IsNil)
	}
}

func (s *testMockTiKVSuite) TestCommitConflict(c *C) {
	// txn A want set x to A
	// txn B want set x to B
	// A prewrite.
	s.mustPrewriteOK(c, putMutations("x", "A"), "x", 5)
	// B prewrite and find A's lock.
	req := &kvrpcpb.PrewriteRequest{
		Mutations:    putMutations("x", "B"),
		PrimaryLock:  []byte("x"),
		StartVersion: 10,
	}
	errs := s.store.Prewrite(req)
	c.Assert(errs[0], NotNil)
	// B find rollback A because A exist too long.
	s.mustRollbackOK(c, [][]byte{[]byte("x")}, 5)
	// if A commit here, it would find its lock removed, report error txn not found.
	s.mustCommitErr(c, [][]byte{[]byte("x")}, 5, 10)
	// B prewrite itself after it rollback A.
	s.mustPrewriteOK(c, putMutations("x", "B"), "x", 10)
	// if A commit here, it would find its lock replaced by others and commit fail.
	s.mustCommitErr(c, [][]byte{[]byte("x")}, 5, 20)
	// B commit success.
	s.mustCommitOK(c, [][]byte{[]byte("x")}, 10, 20)
	// if B commit again, it will success because the key already committed.
	s.mustCommitOK(c, [][]byte{[]byte("x")}, 10, 20)
}

func (s *testMockTiKVSuite) TestResolveLock(c *C) {
	s.mustPrewriteOK(c, putMutations("p1", "v5", "s1", "v5"), "p1", 5)
	s.mustPrewriteOK(c, putMutations("p2", "v10", "s2", "v10"), "p2", 10)
	s.mustResolveLock(c, 5, 0)
	s.mustResolveLock(c, 10, 20)
	s.mustGetNone(c, "p1", 20)
	s.mustGetNone(c, "s1", 30)
	s.mustGetOK(c, "p2", 20, "v10")
	s.mustGetOK(c, "s2", 30, "v10")
	s.mustScanLock(c, 30, nil)
}

func (s *testMockTiKVSuite) TestBatchResolveLock(c *C) {
	s.mustPrewriteOK(c, putMutations("p1", "v11", "s1", "v11"), "p1", 11)
	s.mustPrewriteOK(c, putMutations("p2", "v12", "s2", "v12"), "p2", 12)
	s.mustPrewriteOK(c, putMutations("p3", "v13"), "p3", 13)
	s.mustPrewriteOK(c, putMutations("p4", "v14", "s3", "v14", "s4", "v14"), "p4", 14)
	s.mustPrewriteOK(c, putMutations("p5", "v15", "s5", "v15"), "p5", 15)
	txnInfos := map[uint64]uint64{
		11: 0,
		12: 22,
		13: 0,
		14: 24,
	}
	s.mustBatchResolveLock(c, txnInfos)
	s.mustGetNone(c, "p1", 20)
	s.mustGetNone(c, "p3", 30)
	s.mustGetOK(c, "p2", 30, "v12")
	s.mustGetOK(c, "s4", 30, "v14")
	s.mustScanLock(c, 30, []*kvrpcpb.LockInfo{
		lock("p5", "p5", 15),
		lock("s5", "p5", 15),
	})
	txnInfos = map[uint64]uint64{
		15: 0,
	}
	s.mustBatchResolveLock(c, txnInfos)
	s.mustScanLock(c, 30, nil)
}

func (s *testMockTiKVSuite) TestGC(c *C) {
	var safePoint uint64 = 100

	// Prepare data
	s.mustPutOK(c, "k1", "v1", 1, 2)
	s.mustPutOK(c, "k1", "v2", 11, 12)

	s.mustPutOK(c, "k2", "v1", 1, 2)
	s.mustPutOK(c, "k2", "v2", 11, 12)
	s.mustPutOK(c, "k2", "v3", 101, 102)

	s.mustPutOK(c, "k3", "v1", 1, 2)
	s.mustPutOK(c, "k3", "v2", 11, 12)
	s.mustDeleteOK(c, "k3", 101, 102)

	s.mustPutOK(c, "k4", "v1", 1, 2)
	s.mustDeleteOK(c, "k4", 11, 12)

	// Check prepared data
	s.mustGetOK(c, "k1", 5, "v1")
	s.mustGetOK(c, "k1", 15, "v2")
	s.mustGetOK(c, "k2", 5, "v1")
	s.mustGetOK(c, "k2", 15, "v2")
	s.mustGetOK(c, "k2", 105, "v3")
	s.mustGetOK(c, "k3", 5, "v1")
	s.mustGetOK(c, "k3", 15, "v2")
	s.mustGetNone(c, "k3", 105)
	s.mustGetOK(c, "k4", 5, "v1")
	s.mustGetNone(c, "k4", 105)

	s.mustGC(c, safePoint)

	s.mustGetNone(c, "k1", 5)
	s.mustGetOK(c, "k1", 15, "v2")
	s.mustGetNone(c, "k2", 5)
	s.mustGetOK(c, "k2", 15, "v2")
	s.mustGetOK(c, "k2", 105, "v3")
	s.mustGetNone(c, "k3", 5)
	s.mustGetOK(c, "k3", 15, "v2")
	s.mustGetNone(c, "k3", 105)
	s.mustGetNone(c, "k4", 5)
	s.mustGetNone(c, "k4", 105)
}

func (s *testMockTiKVSuite) TestRollbackAndWriteConflict(c *C) {
	s.mustPutOK(c, "test", "test", 1, 3)
	req := &kvrpcpb.PrewriteRequest{
		Mutations:    putMutations("lock", "lock", "test", "test1"),
		PrimaryLock:  []byte("test"),
		StartVersion: 2,
		LockTtl:      2,
	}
	errs := s.store.Prewrite(req)
	s.mustWriteWriteConflict(c, errs, 1)

	s.mustPutOK(c, "test", "test2", 5, 8)

	// simulate `getTxnStatus` for txn 2.
	err := s.store.Cleanup([]byte("test"), 2, math.MaxUint64)
	c.Assert(err, IsNil)
	req = &kvrpcpb.PrewriteRequest{
		Mutations:    putMutations("test", "test3"),
		PrimaryLock:  []byte("test"),
		StartVersion: 6,
		LockTtl:      1,
	}
	errs = s.store.Prewrite(req)
	s.mustWriteWriteConflict(c, errs, 0)
}

func (s *testMockTiKVSuite) TestDeleteRange(c *C) {
	for i := 1; i <= 5; i++ {
		key := string(byte(i) + byte('0'))
		value := "v" + key
		s.mustPutOK(c, key, value, uint64(1+2*i), uint64(2+2*i))
	}

	s.mustScanOK(c, "0", 10, 20, "1", "v1", "2", "v2", "3", "v3", "4", "v4", "5", "v5")

	s.mustDeleteRange(c, "2", "4")
	s.mustScanOK(c, "0", 10, 30, "1", "v1", "4", "v4", "5", "v5")

	s.mustDeleteRange(c, "5", "5")
	s.mustScanOK(c, "0", 10, 40, "1", "v1", "4", "v4", "5", "v5")

	s.mustDeleteRange(c, "41", "42")
	s.mustScanOK(c, "0", 10, 50, "1", "v1", "4", "v4", "5", "v5")

	s.mustDeleteRange(c, "4\x00", "5\x00")
	s.mustScanOK(c, "0", 10, 60, "1", "v1", "4", "v4")

	s.mustDeleteRange(c, "0", "9")
	s.mustScanOK(c, "0", 10, 70)
}

func (s *testMockTiKVSuite) mustWriteWriteConflict(c *C, errs []error, i int) {
	c.Assert(errs[i], NotNil)
	_, ok := errs[i].(*ErrConflict)
	c.Assert(ok, IsTrue)
}

func (s *testMockTiKVSuite) TestRC(c *C) {
	s.mustPutOK(c, "key", "v1", 5, 10)
	s.mustPrewriteOK(c, putMutations("key", "v2"), "key", 15)
	s.mustGetErr(c, "key", 20)
	s.mustGetRC(c, "key", 12, "v1")
	s.mustGetRC(c, "key", 20, "v1")
}

func (s testMarshal) TestMarshalmvccLock(c *C) {
	l := mvccLock{
		startTS:     47,
		primary:     []byte{'a', 'b', 'c'},
		value:       []byte{'d', 'e'},
		op:          kvrpcpb.Op_Put,
		ttl:         444,
		minCommitTS: 666,
	}
	bin, err := l.MarshalBinary()
	c.Assert(err, IsNil)

	var l1 mvccLock
	err = l1.UnmarshalBinary(bin)
	c.Assert(err, IsNil)

	c.Assert(l.startTS, Equals, l1.startTS)
	c.Assert(l.op, Equals, l1.op)
	c.Assert(l.ttl, Equals, l1.ttl)
	c.Assert(string(l.primary), Equals, string(l1.primary))
	c.Assert(string(l.value), Equals, string(l1.value))
	c.Assert(l.minCommitTS, Equals, l1.minCommitTS)
}

func (s testMarshal) TestMarshalmvccValue(c *C) {
	v := mvccValue{
		valueType: typePut,
		startTS:   42,
		commitTS:  55,
		value:     []byte{'d', 'e'},
	}
	bin, err := v.MarshalBinary()
	c.Assert(err, IsNil)

	var v1 mvccValue
	err = v1.UnmarshalBinary(bin)
	c.Assert(err, IsNil)

	c.Assert(v.valueType, Equals, v1.valueType)
	c.Assert(v.startTS, Equals, v1.startTS)
	c.Assert(v.commitTS, Equals, v1.commitTS)
	c.Assert(string(v.value), Equals, string(v.value))
}

func (s *testMVCCLevelDB) TestErrors(c *C) {
	c.Assert((&ErrKeyAlreadyExist{}).Error(), Equals, `key already exist, key: ""`)
	c.Assert(ErrAbort("txn").Error(), Equals, "abort: txn")
	c.Assert(ErrAlreadyCommitted(0).Error(), Equals, "txn already committed")
	c.Assert((&ErrConflict{}).Error(), Equals, "write conflict")
}

func (s *testMVCCLevelDB) TestCheckTxnStatus(c *C) {
	startTS := uint64(5 << 18)
	s.mustPrewriteWithTTLOK(c, putMutations("pk", "val"), "pk", startTS, 666)

	ttl, commitTS, action, err := s.store.CheckTxnStatus([]byte("pk"), startTS, startTS+100, 666, false)
	c.Assert(err, IsNil)
	c.Assert(ttl, Equals, uint64(666))
	c.Assert(commitTS, Equals, uint64(0))
	c.Assert(action, Equals, kvrpcpb.Action_MinCommitTSPushed)

	// MaxUint64 as callerStartTS shouldn't update minCommitTS but return Action_MinCommitTSPushed.
	ttl, commitTS, action, err = s.store.CheckTxnStatus([]byte("pk"), startTS, math.MaxUint64, 666, false)
	c.Assert(err, IsNil)
	c.Assert(ttl, Equals, uint64(666))
	c.Assert(commitTS, Equals, uint64(0))
	c.Assert(action, Equals, kvrpcpb.Action_MinCommitTSPushed)
	s.mustCommitOK(c, [][]byte{[]byte("pk")}, startTS, startTS+101)

	ttl, commitTS, _, err = s.store.CheckTxnStatus([]byte("pk"), startTS, 0, 666, false)
	c.Assert(err, IsNil)
	c.Assert(ttl, Equals, uint64(0))
	c.Assert(commitTS, Equals, uint64(startTS+101))

	s.mustPrewriteWithTTLOK(c, putMutations("pk1", "val"), "pk1", startTS, 666)
	s.mustRollbackOK(c, [][]byte{[]byte("pk1")}, startTS)

	ttl, commitTS, action, err = s.store.CheckTxnStatus([]byte("pk1"), startTS, 0, 666, false)
	c.Assert(err, IsNil)
	c.Assert(ttl, Equals, uint64(0))
	c.Assert(commitTS, Equals, uint64(0))
	c.Assert(action, Equals, kvrpcpb.Action_NoAction)

	s.mustPrewriteWithTTLOK(c, putMutations("pk2", "val"), "pk2", startTS, 666)
	currentTS := uint64(777 << 18)
	ttl, commitTS, action, err = s.store.CheckTxnStatus([]byte("pk2"), startTS, 0, currentTS, false)
	c.Assert(err, IsNil)
	c.Assert(ttl, Equals, uint64(0))
	c.Assert(commitTS, Equals, uint64(0))
	c.Assert(action, Equals, kvrpcpb.Action_TTLExpireRollback)

	// Cover the TxnNotFound case.
	_, _, _, err = s.store.CheckTxnStatus([]byte("txnNotFound"), 5, 0, 666, false)
	c.Assert(err, NotNil)
	notFound, ok := errors.Cause(err).(*ErrTxnNotFound)
	c.Assert(ok, IsTrue)
	c.Assert(notFound.StartTs, Equals, uint64(5))
	c.Assert(string(notFound.PrimaryKey), Equals, "txnNotFound")

	ttl, commitTS, action, err = s.store.CheckTxnStatus([]byte("txnNotFound"), 5, 0, 666, true)
	c.Assert(err, IsNil)
	c.Assert(ttl, Equals, uint64(0))
	c.Assert(commitTS, Equals, uint64(0))
	c.Assert(action, Equals, kvrpcpb.Action_LockNotExistRollback)

	// Check the rollback tombstone blocks this prewrite which comes with a smaller startTS.
	req := &kvrpcpb.PrewriteRequest{
		Mutations:    putMutations("txnNotFound", "val"),
		PrimaryLock:  []byte("txnNotFound"),
		StartVersion: 4,
		MinCommitTs:  6,
	}
	errs := s.store.Prewrite(req)
	c.Assert(errs, NotNil)
}

func (s *testMVCCLevelDB) TestRejectCommitTS(c *C) {
	s.mustPrewriteOK(c, putMutations("x", "A"), "x", 5)
	// Push the minCommitTS
	_, _, _, err := s.store.CheckTxnStatus([]byte("x"), 5, 100, 100, false)
	c.Assert(err, IsNil)
	err = s.store.Commit([][]byte{[]byte("x")}, 5, 10)
	e, ok := errors.Cause(err).(*ErrCommitTSExpired)
	c.Assert(ok, IsTrue)
	c.Assert(e.MinCommitTs, Equals, uint64(101))
}

func (s *testMVCCLevelDB) TestMvccGetByKey(c *C) {
	s.mustPrewriteOK(c, putMutations("q1", "v5"), "p1", 5)
	debugger, ok := s.store.(MVCCDebugger)
	c.Assert(ok, IsTrue)
	mvccInfo := debugger.MvccGetByKey([]byte("q1"))
	except := &kvrpcpb.MvccInfo{
		Lock: &kvrpcpb.MvccLock{
			Type:       kvrpcpb.Op_Put,
			StartTs:    5,
			Primary:    []byte("p1"),
			ShortValue: []byte("v5"),
		},
	}
	c.Assert(mvccInfo, DeepEquals, except)
}

func (s *testMVCCLevelDB) TestTxnHeartBeat(c *C) {
	s.mustPrewriteWithTTLOK(c, putMutations("pk", "val"), "pk", 5, 666)

	// Update the ttl
	ttl, err := s.store.TxnHeartBeat([]byte("pk"), 5, 888)
	c.Assert(err, IsNil)
	c.Assert(ttl, Greater, uint64(666))

	// Advise ttl is small
	ttl, err = s.store.TxnHeartBeat([]byte("pk"), 5, 300)
	c.Assert(err, IsNil)
	c.Assert(ttl, Greater, uint64(300))

	// The lock has already been clean up
	c.Assert(s.store.Cleanup([]byte("pk"), 5, math.MaxUint64), IsNil)
	_, err = s.store.TxnHeartBeat([]byte("pk"), 5, 1000)
	c.Assert(err, NotNil)
}