package leveldb_test

import (
	"bytes"
	"context"
	"io"
	"io/ioutil"
	"os"
	"testing"

	"github.com/pingcap-inc/alioth/storage/keyvalue"
	"github.com/pingcap-inc/alioth/storage/leveldb"
	. "github.com/pingcap/check"
)

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

var _ = Suite(&testLevelDBSuite{})

type testLevelDBSuite struct {
}

func (t *testLevelDBSuite) TestGoLevelDB(c *C) {
	/*tmpdir1, err := ioutil.TempDir("./", "logpattern_test")
	c.Assert(err, IsNil)
	defer os.RemoveAll(tmpdir1)

	db, err := leveldb.Open(tmpdir1, nil)
	c.Assert(err, IsNil)
	t.testLevelDB(c, db)*/

	tmpdir, err := ioutil.TempDir("./", "logpattern_test")
	c.Assert(err, IsNil)
	defer os.RemoveAll(tmpdir)

	db, err := leveldb.OpenGoLevelDB(tmpdir, nil)
	c.Assert(err, IsNil)
	t.testLevelDB(c, db)

}

func (t *testLevelDBSuite) testLevelDB(c *C, db keyvalue.DB) {
	kvs := [][]byte{
		[]byte("a"),
		[]byte("k_1"),
		[]byte("k_2"),
		[]byte("v_1"),
		[]byte("v_2"),
		[]byte("x"),
		[]byte("y_1"),
		[]byte("y_2"),
		[]byte("z"),
	}

	writer, err := db.Writer(context.Background())
	c.Assert(err, IsNil)

	for _, kv := range kvs {
		writer.Write(kv, kv)
	}
	c.Assert(writer.Close(), IsNil)

	for _, kv := range kvs {
		val, err := db.Get(context.Background(), kv, nil)
		c.Assert(err, IsNil)
		c.Assert(val, DeepEquals, val)
	}

	prefixCases := map[string][]int{
		"a": {0, 0},
		"k": {1, 2},
		"v": {3, 4},
		"y": {6, 7},
		"x": {5, 5},
		"z": {8, 8},
	}

	for prefix, tc := range prefixCases {
		it, err := db.ScanPrefix(context.Background(), []byte(prefix), nil)
		c.Assert(err, IsNil)

		for i := tc[0]; i <= tc[1]; i++ {
			k, v, err := it.Next()
			c.Assert(err, IsNil)
			c.Assert(k, DeepEquals, kvs[i])
			c.Assert(v, DeepEquals, kvs[i])
		}

		k, v, err := it.Next()
		c.Assert(err, Equals, io.EOF)
		c.Assert(k, IsNil)
		c.Assert(v, IsNil)
	}

	rangeCases := map[*keyvalue.Range][]int{
		{[]byte("a"), []byte("k")}: {0, 0},
		{[]byte("k"), []byte("v")}: {1, 2},
		{[]byte("v"), []byte("y")}: {3, 5},
		{[]byte("x"), []byte("")}:  {5, 8},
	}

	for r, tc := range rangeCases {
		it, err := db.ScanRange(context.Background(), r, nil)
		c.Assert(err, IsNil)

		for i := tc[0]; i <= tc[1]; i++ {
			k, v, err := it.Next()
			c.Assert(err, IsNil)
			c.Assert(k, DeepEquals, kvs[i])
			c.Assert(v, DeepEquals, kvs[i])
		}

		k, v, err := it.Next()
		c.Assert(err, Equals, io.EOF)
		c.Assert(k, IsNil)
		c.Assert(v, IsNil)
	}

	c.Assert(bytes.Compare([]byte("x"), []byte("v_1")) > 0, IsTrue)
	c.Assert(bytes.Compare([]byte("x"), []byte("x_1")) < 0, IsTrue)
}